InfluxDB: 规范批量写入 Line Protocol，并声明批量能力 by Copilot · Pull Request #87 · NewLifeX/NewLife.XCode

Copilot · 2026-07-15T03:21:32Z

InfluxDB 批量写入路径存在协议细节偏差：数值类型格式不规范、时间字段会重复写入、以及换行符在 Windows 下可能使用 \r\n。本 PR 对批量写入构造逻辑做了收敛，并补齐 InfluxDB 的 BatchCapability 声明以对齐批量接口分流。

Line Protocol 值格式标准化
- 整数统一附加后缀（如 7i，UInt64 为 u）
- 布尔值固定为小写 true/false
- 字符串统一双引号并转义
- 浮点/小数按 InvariantCulture 输出，避免区域化格式污染协议
时间列去重写入
- Time/CreateTime/UpdateTime 仅作为行尾 timestamp 参与写入
- 该时间列不再同时进入 tags/fields，避免冗余与索引语义冲突
换行符协议一致性
- 批量拼接由平台相关换行改为显式 \n，保证 Line Protocol 兼容性
批量能力声明对齐
- 为 InfluxDB 显式声明 BatchCapability.Insert | BatchCapability.Upsert
- 让 Entity.Insert/Upsert 的批量分流可识别并命中 InfluxDBSession 对应路径
示例（构造结果形态）

temperature,DeviceId=1001 Count=7i,Enabled=true,Name="sensor \"A\"" 1784073600000000000

nygula · 2026-07-15T03:23:25Z

比如 `using System.Data;
using System.Data.Common;
using System.Text;
using NewLife.Collections;
using NewLife.Data;
using NewLife.Log;
using XCode.InfluxDB;

namespace XCode.DataAccessLayer;

class InfluxDB : RemoteDb
{
#region 属性
///

返回数据库类型。

public override DatabaseType Type => DatabaseType.InfluxDB;

/// <summary>批量操作能力</summary>
public override BatchCapability BatchCapability => BatchCapability.Insert | BatchCapability.Upsert;

/// <summary>创建工厂</summary>
/// <returns></returns>
protected override DbProviderFactory CreateFactory() => InfluxDBFactory.Instance;

const String Server_Key = "Server";
protected override void OnSetConnectionString(ConnectionStringBuilder builder)
{
    base.OnSetConnectionString(builder);

    // 确保 Server 地址以 http:// 或 https:// 开头
    var server = builder[Server_Key];
    if (!String.IsNullOrEmpty(server) && !server.StartsWithIgnoreCase("http://", "https://"))
    {
        builder[Server_Key] = $"http://{server}";
    }
}
#endregion

#region 方法
/// <summary>创建数据库会话</summary>
/// <returns></returns>
protected override IDbSession OnCreateSession() => new InfluxDBSession(this);

/// <summary>创建元数据对象</summary>
/// <returns></returns>
protected override IMetaData OnCreateMetaData() => new InfluxDBMetaData();

public override Boolean Support(String providerName)
{
    providerName = providerName.ToLower();
    if (providerName.EqualIgnoreCase("InfluxDB", "Influx")) return true;

    return false;
}
#endregion

#region 数据库特性
protected override String ReservedWordsStr => "AND,OR,NOT,FROM,WHERE,SELECT,DELETE,DROP,SHOW,MEASUREMENT,TAG,FIELD,TIME";

/// <summary>格式化关键字</summary>
/// <param name="keyWord">关键字</param>
/// <returns></returns>
public override String FormatKeyWord(String keyWord)
{
    if (keyWord.IsNullOrEmpty()) return keyWord;
    if (keyWord.StartsWith("\"") && keyWord.EndsWith("\"")) return keyWord;
    return $"\"{keyWord}\"";
}

/// <summary>格式化数据为SQL数据</summary>
/// <param name="field">字段</param>
/// <param name="value">数值</param>
/// <returns></returns>
public override String FormatValue(IDataColumn field, Object? value)
{
    var code = System.Type.GetTypeCode(field.DataType);
    if (code == TypeCode.String)
    {
        if (value == null)
            return field.Nullable ? "null" : "\"\"";

        return "\"" + value.ToString()?.Replace("\"", "\\\"") + "\"";
    }
    else if (code == TypeCode.Boolean)
    {
        return value.ToBoolean() ? "true" : "false";
    }

    return base.FormatValue(field, value);
}

/// <summary>格式化时间为SQL字符串</summary>
/// <param name="column">字段</param>
/// <param name="dateTime">时间值</param>
/// <returns></returns>
public override String FormatDateTime(IDataColumn column, DateTime dateTime)
{
    // InfluxDB 使用纳秒时间戳
    var epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
    var nanos = (dateTime.ToUniversalTime() - epoch).Ticks * 100;
    return nanos.ToString();
}

/// <summary>长文本长度</summary>
public override Int32 LongTextLength => 65535;

internal protected override String ParamPrefix => "@";

/// <summary>系统数据库名</summary>
public override String SystemDatabaseName => "_internal";

/// <summary>字符串相加</summary>
/// <param name="left"></param>
/// <param name="right"></param>
/// <returns></returns>
public override String StringConcat(String left, String right) => $"{left} + {right}";
#endregion

}

///

InfluxDB数据库会话

internal class InfluxDBSession : RemoteDbSession
{
#region 构造函数
public InfluxDBSession(IDatabase db) : base(db) { }
#endregion

#region 基本方法 查询/执行
/// <summary>执行插入语句并返回新增行的自动编号</summary>
/// <param name="sql">SQL语句</param>
/// <param name="type">命令类型，默认SQL文本</param>
/// <param name="ps">命令参数</param>
/// <returns>新增行的自动编号</returns>
public override Int64 InsertAndGetIdentity(String sql, CommandType type = CommandType.Text, params IDataParameter[]? ps)
{
    // InfluxDB 是时序数据库，通常使用时间戳作为主键，不支持自增ID
    Execute(sql, type, ps);
    return 0;
}

public override Task<Int64> InsertAndGetIdentityAsync(String sql, CommandType type = CommandType.Text, params IDataParameter[]? ps)
{
    // InfluxDB 是时序数据库，通常使用时间戳作为主键，不支持自增ID
    ExecuteAsync(sql, type, ps).Wait();
    return Task.FromResult(0L);
}
#endregion

#region 批量操作
/// <summary>批量插入</summary>
/// <param name="table">数据表</param>
/// <param name="columns">要插入的字段</param>
/// <param name="list">实体列表</param>
/// <returns></returns>
public override Int32 Insert(IDataTable table, IDataColumn[] columns, IEnumerable<IModel> list)
{
    var sb = Pool.StringBuilder.Get();
    var db = (Database as DbBase)!;

    // InfluxDB 使用 Line Protocol 格式写入
    // 格式: measurement,tag1=value1,tag2=value2 field1=value1,field2=value2 timestamp
    foreach (var entity in list)
    {
        // measurement 名称（表名）
        sb.Append(db.FormatName(table));

        // tags（索引字段，通常是维度），时间戳字段不作为 tag
        var tags = columns.Where(c => (c.PrimaryKey || c.Master) && !IsTimeColumn(c)).ToArray();
        if (tags.Length > 0)
        {
            sb.Append(',');
            sb.Append(tags.Join(",", c =>
            {
                var value = entity[c.Name];
                return $"{db.FormatName(c)}={value}";
            }));
        }

        // fields（数据字段）
        var fields = columns.Where(c => !c.PrimaryKey && !c.Master && !IsTimeColumn(c)).ToArray();
        if (fields.Length > 0)
        {
            sb.Append(' ');
            sb.Append(fields.Join(",", c =>
            {
                var value = entity[c.Name];
                var strValue = FormatFieldValue(c, value);
                return $"{db.FormatName(c)}={strValue}";
            }));
        }

        // timestamp（纳秒级时间戳）
        var timeCol = columns.FirstOrDefault(IsTimeColumn);
        if (timeCol != null)
        {
            var time = entity[timeCol.Name];
            if (time is DateTime dt)
            {
                var epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
                var nanos = (dt.ToUniversalTime() - epoch).Ticks * 100;
                sb.Append($" {nanos}");
            }
        }

        sb.Append('\n');
    }

    var lineProtocol = sb.Return(true);
    return Execute(lineProtocol);
}

private static String FormatFieldValue(IDataColumn c, Object? value)
{
    if (value == null) return "\"\"";

    var code = System.Type.GetTypeCode(c.DataType);
    switch (code)
    {
        case TypeCode.String:
            return $"\"{value}\"";
        case TypeCode.Boolean:
            return value.ToBoolean() ? "true" : "false";
        case TypeCode.Byte:
        case TypeCode.SByte:
        case TypeCode.Int16:
        case TypeCode.UInt16:
        case TypeCode.Int32:
        case TypeCode.UInt32:
        case TypeCode.Int64:
        case TypeCode.UInt64:
            return $"{value}i";
        case TypeCode.Single:
        case TypeCode.Double:
        case TypeCode.Decimal:
            return value.ToString()!;
        case TypeCode.DateTime:
            var epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
            return ((DateTime)value - epoch).Ticks.ToString();
        default:
            return $"\"{value}\"";
    }
}

private static Boolean IsTimeColumn(IDataColumn c) => c.Name.EqualIgnoreCase("Time", "CreateTime", "UpdateTime", "Ts") || c.DataType == typeof(DateTime) && c.PrimaryKey;

/// <summary>批量插入或更新</summary>
/// <param name="table">数据表</param>
/// <param name="columns">要插入的字段</param>
/// <param name="updateColumns">主键已存在时，要更新的字段</param>
/// <param name="addColumns">主键已存在时，要累加更新的字段</param>
/// <param name="list">实体列表</param>
/// <returns></returns>
public override Int32 Upsert(IDataTable table, IDataColumn[] columns, ICollection<String>? updateColumns, ICollection<String>? addColumns, IEnumerable<IModel> list)
{
    // InfluxDB 自动处理相同 measurement + tags + timestamp 的写入，新值会覆盖旧值
    return Insert(table, columns, list);
}
#endregion

#region 架构
public override DataTable GetSchema(DbConnection? conn, String collectionName, String?[]? restrictionValues) => new DataTable();
#endregion

}

///

InfluxDB元数据

class InfluxDBMetaData : RemoteDbMetaData
{
public InfluxDBMetaData() => Types = _DataTypes;

#region 数据类型
/// <summary>数据类型映射</summary>
private static readonly Dictionary<Type, String[]> _DataTypes = new()
{
    { typeof(Byte), new String[] { "INTEGER" } },
    { typeof(Int16), new String[] { "INTEGER" } },
    { typeof(Int32), new String[] { "INTEGER" } },
    { typeof(Int64), new String[] { "INTEGER" } },
    { typeof(Single), new String[] { "FLOAT" } },
    { typeof(Double), new String[] { "FLOAT" } },
    { typeof(Decimal), new String[] { "FLOAT" } },
    { typeof(DateTime), new String[] { "TIMESTAMP" } },
    { typeof(String), new String[] { "STRING" } },
    { typeof(Boolean), new String[] { "BOOLEAN" } },
};
#endregion

#region 架构
protected override List<IDataTable> OnGetTables(String[]? names)
{
    var ss = Database.CreateSession();
    var list = new List<IDataTable>();

    var old = ss.ShowSQL;
    ss.ShowSQL = false;
    try
    {
        // InfluxDB Flux 查询获取所有 measurement
        var flux = @"

import ""influxdata/influxdb/schema""
schema.measurements(bucket: v.bucket)
";
var dt = ss.Query(flux, null);
if (dt.Rows.Count == 0) return [];

        var hs = new HashSet<String>(names ?? [], StringComparer.OrdinalIgnoreCase);

        // 所有表（measurement）
        foreach (var dr in dt)
        {
            var name = dr["_value"] + "";
            if (name.IsNullOrEmpty() || hs.Count > 0 && !hs.Contains(name)) continue;

            var table = DAL.CreateTable();
            table.TableName = name;
            table.DbType = Database.Type;

            // InfluxDB 的 schema 需要通过查询数据来推断
            // 这里简化处理，不详细查询字段信息
            #region 字段
            // 默认添加 time 字段
            var timeField = table.CreateColumn();
            timeField.ColumnName = "time";
            timeField.DataType = typeof(DateTime);
            timeField.PrimaryKey = true;
            table.Columns.Add(timeField);
            #endregion

            // 修正关系数据
            table.Fix();

            list.Add(table);
        }
    }
    finally
    {
        ss.ShowSQL = old;
    }

    return list;
}

/// <summary>快速取得所有表名</summary>
/// <returns></returns>
public override IList<String> GetTableNames()
{
    var list = new List<String>();

    var flux = @"

import ""influxdata/influxdb/schema""
schema.measurements(bucket: v.bucket)
";
var dt = base.Database.CreateSession().Query(flux, null);
if (dt.Rows.Count == 0) return list;

    foreach (var dr in dt)
    {
        var name = dr["_value"] + "";
        if (!name.IsNullOrEmpty()) list.Add(name);
    }

    return list;
}

public override String FieldClause(IDataColumn field, Boolean onlyDefine)
{
    var sb = new StringBuilder();
    sb.AppendFormat("{0} ", FormatName(field));

    String? typeName = null;
    if (Database.Type == field.Table.DbType && !field.Identity) typeName = field.RawType;
    if (String.IsNullOrEmpty(typeName)) typeName = GetFieldType(field);

    sb.Append(typeName);
    return sb.ToString();
}
#endregion

#region 反向工程
public override Boolean DatabaseExist(String? databaseName)
{
    if (databaseName.IsNullOrEmpty()) return base.DatabaseExist(databaseName);

    // InfluxDB 2.x 使用 bucket 概念
    var flux = @"buckets() |> filter(fn: (r) => r.name == """ + databaseName + @""")";
    var dt = Database.CreateSession().Query(flux, null);
    return dt != null && dt.Rows != null && dt.Rows.Count > 0;
}

public override String CreateDatabaseSQL(String dbname, String? file)
{
    // InfluxDB 2.x 不支持通过 SQL/Flux 创建 bucket，需要使用 HTTP API
    throw new NotSupportedException("InfluxDB does not support creating buckets via SQL. Use HTTP API or CLI.");
}

public override String DropDatabaseSQL(String dbname)
{
    throw new NotSupportedException("InfluxDB does not support dropping buckets via SQL. Use HTTP API or CLI.");
}

public override String CreateTableSQL(IDataTable table)
{
    // InfluxDB 不需要显式创建表（measurement），写入数据时自动创建
    return String.Empty;
}

public override String AddTableDescriptionSQL(IDataTable table) => String.Empty;

public override String AlterColumnSQL(IDataColumn field, IDataColumn? oldfield)
{
    // InfluxDB 不支持修改字段
    throw new NotSupportedException("InfluxDB does not support altering columns.");
}

public override String AddColumnDescriptionSQL(IDataColumn field) => String.Empty;
#endregion

}
`

Copilot

Pull request overview

本 PR 聚焦修正 InfluxDB 批量写入的 Line Protocol 构造细节，并补齐 InfluxDB 的批量能力声明，以便 XCode 批量接口在 Insert/Upsert 分流时能正确命中 InfluxDB 路径，同时提升跨平台协议一致性。

Changes:

在 InfluxDB 实现中声明 BatchCapability.Insert | BatchCapability.Upsert，使批量分流可识别 InfluxDB 的能力边界。
收敛批量写入 Line Protocol 构造逻辑：字段类型格式化（整数后缀、布尔小写、浮点 InvariantCulture）、时间列作为行尾 timestamp 且避免冗余写入、换行符强制使用 \n。
新增纯单元测试覆盖 Line Protocol 格式化与时间列去重写入行为。

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
XUnitTest.XCode/DataAccessLayer/InfluxDBLineProtocolTests.cs	新增 Line Protocol 纯单元测试，验证字段格式、时间列不重复写入与换行符一致性
XUnitTest.XCode/DataAccessLayer/BatchCapabilityTests.cs	增加 InfluxDB 的 BatchCapability 断言用例
XCode/DataAccessLayer/Database/InfluxDB.cs	实现 BatchCapability 声明，并调整批量写入 Line Protocol 构造与字段值格式化

    public override String FormatValue(IDataColumn field, Object? value)
    {
+        if (value == null)
+            return field.Nullable ? "null" : "";
+
        var code = System.Type.GetTypeCode(field.DataType);
        if (code == TypeCode.String)
        {
-            if (value == null)
-                return field.Nullable ? "null" : "\"\"";
-
            return "\"" + value.ToString()?.Replace("\"", "\\\"") + "\"";
        }
        else if (code == TypeCode.Boolean)
        {
            return value.ToBoolean() ? "true" : "false";
        }
+        else if (code is TypeCode.SByte or TypeCode.Byte or TypeCode.Int16 or TypeCode.UInt16 or TypeCode.Int32 or TypeCode.UInt32 or TypeCode.Int64)
+        {
+            return $"{value}i";
+        }
+        else if (code == TypeCode.UInt64)
+        {
+            return $"{value}u";
+        }
+        else if (code is TypeCode.Single or TypeCode.Double or TypeCode.Decimal)
+        {
+            return Convert.ToString(value, CultureInfo.InvariantCulture) ?? "0";
+        }


+            var timeCol = columns.FirstOrDefault(c =>
+            {
+                var name = c.Name ?? c.ColumnName;
+                return !name.IsNullOrEmpty() && name.EqualIgnoreCase("Time", "CreateTime", "UpdateTime") && entity[name] != null;
+            });


Initial plan

f3eee3f

Copilot AI assigned Copilot and nygula Jul 15, 2026

Copilot started work on behalf of nygula July 15, 2026 03:21 View session

Copilot AI linked an issue Jul 15, 2026 that may be closed by this pull request

InfluxDB功能优化方案 #86

Closed

fix: 规范 InfluxDB 批量写入与能力声明

8af139c

Copilot AI changed the title ~~[WIP] Optimize InfluxDB functionality for batch writing~~ InfluxDB: 规范批量写入 Line Protocol，并声明批量能力 Jul 15, 2026

Copilot finished work on behalf of nygula July 15, 2026 03:31

Copilot AI requested a review from nygula July 15, 2026 03:31

nnhy marked this pull request as ready for review July 15, 2026 03:57

Copilot AI review requested due to automatic review settings July 15, 2026 03:57

nnhy merged commit 9c5f681 into master Jul 15, 2026

Copilot started reviewing on behalf of nnhy July 15, 2026 03:58 View session

Copilot AI reviewed Jul 15, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

InfluxDB: 规范批量写入 Line Protocol，并声明批量能力#87

InfluxDB: 规范批量写入 Line Protocol，并声明批量能力#87
nnhy merged 2 commits into
masterfrom
copilot/optimize-influxdb-functionality

Copilot AI commented Jul 15, 2026 •

edited

Loading

Uh oh!

nygula commented Jul 15, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

Copilot AI commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nygula commented Jul 15, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Copilot AI commented Jul 15, 2026 •

edited

Loading