tdengine批量插入超级表

FreeSql.Tests/FreeSql.Tests.Provider.TDengine/TDengine/TDengineCrud/TDengineBatchTest.cs

@@ -0,0 +1,62 @@
+using FreeSql.Tests.Provider.TDengine.TDengine.Tables;
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Xunit.Abstractions;
+
+namespace FreeSql.Tests.Provider.TDengine.TDengine.TDengineCrud;
+
+public class TDengineBatchTest
+{
+    private readonly ITestOutputHelper _output;
+
+    // 通过构造函数注入
+    public TDengineBatchTest(ITestOutputHelper output)
+    {
+        _output = output;
+    }
+
+    [Fact]
+    public void BatchInsertToSuperTable()
+    {
+        var fsql = g.tdengine;
+        fsql.CodeFirst.SyncStructure<PointDataEntity>();
+
+        var generateCount = 2_000; // 生成数据数量
+        var dt = DateTime.Now;
+        var data = Enumerable.Range(1, generateCount)
+            .Select(t => new PointDataEntity
+            {
+                Ts = dt,
+                PointNumber = $"001A{t:D5}",
+                PointValue = 345
+            })
+            .ToList();
+        // 生成第二批数据，时间戳加1秒，IsAlarm为true
+        var data2 = Enumerable.Range(1, generateCount)
+            .Select(t => new PointDataEntity
+            {
+                Ts = dt.AddSeconds(1),
+                PointNumber = $"001A{t:D5}",
+                IsAlarm = true,
+                PointValue = 123.45
+            })
+            .ToList();
+        data.AddRange(data2);
+
+        var insertable = fsql.Insert(data)
+            .AsTdEngineTableName(t => $"child_{t.PointNumber}") // 自定义子表名
+            .BatchOptions(5000, 5000, false); // tdengine 关闭事务
+
+        //_output.WriteLine($"集合中总数量：{data.Count}");
+        //var sql = insertable.ToSql();
+        //_output.WriteLine(sql);
+
+        var affrows = insertable.ExecuteAffrows();
+
+        Assert.NotEqual(0, affrows);
+    }
+}

FreeSql.Tests/FreeSql.Tests.Provider.TDengine/TDengine/Tables/PointData.cs

@@ -0,0 +1,26 @@
+using FreeSql.DataAnnotations;
+using FreeSql.Provider.TDengine.Attributes;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace FreeSql.Tests.Provider.TDengine.TDengine.Tables;
+
+[TDengineSuperTable(Name = "point_data")]
+public class PointDataEntity
+{
+    [Column(Name = "ts")]
+    public DateTime Ts { get; set; }
+
+    [Column(Name = "is_alarm")]
+    public bool IsAlarm { get; set; }
+
+    [Column(Name = "point_value")]
+    public double PointValue { get; set; }
+
+    [TDengineTag(Name = "point_number")]
+    public string PointNumber { get; set; }
+
+}

FreeSql/Interface/Curd/IInsert.cs

@@ -120,6 +120,13 @@ public interface IInsert<T1> where T1 : class
         /// <returns></returns>
         IInsert<T1> BatchProgress(Action<BatchProgressStatus<T1>> callback);
 
+        /// <summary>
+        /// 设置TdEngine子表名称规则
+        /// </summary>
+        /// <param name="tableNameRule"></param>
+        /// <returns></returns>
+        IInsert<T1> AsTdEngineTableName(Func<T1, string> tableNameRule);
+
         /// <summary>
         /// 设置表名规则，可用于分库/分表，参数1：默认表名；返回值：新表名；
         /// </summary>

FreeSql/Internal/CommonProvider/InsertProvider.cs

@@ -41,6 +41,8 @@ public abstract partial class InsertProvider<T1> : InsertProvider, IInsert<T1> w
         internal List<T1> _sourceOld;
         public Action<BatchProgressStatus<T1>> _batchProgress;
 
+        public Func<T1, string> _tdEngineChildTableRule;
+
         public InsertProvider(IFreeSql orm, CommonUtils commonUtils, CommonExpression commonExpression)
         {
             _orm = orm;
@@ -744,5 +746,11 @@ public DataTable ToDataTable()
             }
             return dt;
         }
+
+        public IInsert<T1> AsTdEngineTableName(Func<T1, string> tableNameRule)
+        {
+            this._tdEngineChildTableRule = tableNameRule;
+            return this;
+        }
     }
 }

Providers/FreeSql.Provider.TDengine/Curd/TDengineInsert.cs

@@ -46,19 +46,106 @@ public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(
 
         public override string ToSql()
         {
-            //处理Insert忽略Tag
-            var ignoreColumnList = _ignoreInsertColumns.GetOrAdd(typeof(T1), s =>
-            {
-                //如果是超表不处理
-                if (!s.IsDefined(typeof(TDengineSubTableAttribute))) return new List<string>(0);
-                var tableByEntity = _commonUtils.GetTableByEntity(s);
-                var keyValuePairs = tableByEntity.Properties.Where(pair =>
-                    pair.Value.GetCustomAttribute<TDengineTagAttribute>() != null);
-                return keyValuePairs.Select(keyValuePair => keyValuePair.Value.Name).ToList();
+            if (_tdEngineChildTableRule != null)
+            {
+                return ToSTableBatchInsertSql();
+            }
+            else
+            {
+                //处理Insert忽略Tag
+                var ignoreColumnList = _ignoreInsertColumns.GetOrAdd(typeof(T1), s =>
+                {
+                    //如果是超表不处理
+                    if (!s.IsDefined(typeof(TDengineSubTableAttribute))) return new List<string>(0);
+                    var tableByEntity = _commonUtils.GetTableByEntity(s);
+                    var keyValuePairs = tableByEntity.Properties.Where(pair =>
+                        pair.Value.GetCustomAttribute<TDengineTagAttribute>() != null);
+                    return keyValuePairs.Select(keyValuePair => keyValuePair.Value.Name).ToList();
+                });
+                if (InternalIsIgnoreInto == false) return base.ToSqlValuesOrSelectUnionAll(ignoreColumn: ignoreColumnList);
+                var sql = base.ToSqlValuesOrSelectUnionAll(ignoreColumn: ignoreColumnList);
+                return $"INSERT IGNORE INTO {sql.Substring(12)}";
+            }
+        }
+
+        /// <summary>
+        /// Key 为 T1 Type，Value 为 Tuple&lt;超级表名, Tag属性数组, Value属性数组&gt;
+        /// </summary>
+        private static ConcurrentDictionary<Type, Tuple<string, ColumnInfo[], ColumnInfo[]>> _superTableColumns =
+            new ConcurrentDictionary<Type, Tuple<string, ColumnInfo[], ColumnInfo[]>>();
+
+        private string ToSTableBatchInsertSql()
+        {
+            if (_source == null || _source.Count == 0) return null;
+
+            var typePropertiesCached = _superTableColumns.GetOrAdd(typeof(T1), s =>
+            {
+                var tableInfo = _commonUtils.GetTableByEntity(s);
+                var stableName = tableInfo.DbName;
+                var tagColumnNames = tableInfo.Properties
+                    .Where(kvp => kvp.Value.GetCustomAttribute<TDengineTagAttribute>() != null)
+                    .Select(kvp => kvp.Value.Name).ToList();
+                var tagColumns = tableInfo.Columns.Values
+                    .Where(col => tagColumnNames.Contains(col.CsName))
+                    .ToArray();
+
+                var valueColumns = tableInfo.Columns.Values
+                    .Except(tagColumns)
+                    .Where(col => !_ignore.ContainsKey(col.CsName))
+                    .ToArray();              
+                return new Tuple<string, ColumnInfo[], ColumnInfo[]>(stableName, tagColumns, valueColumns);
             });
-            if (InternalIsIgnoreInto == false) return base.ToSqlValuesOrSelectUnionAll(ignoreColumn: ignoreColumnList);
-            var sql = base.ToSqlValuesOrSelectUnionAll(ignoreColumn: ignoreColumnList);
-            return $"INSERT IGNORE INTO {sql.Substring(12)}";
+
+            var tagNames = string.Join(",", typePropertiesCached.Item2.Select(x => x.Attribute.Name));
+            var valueColumnNames = string.Join(",", typePropertiesCached.Item3.Select(x => x.Attribute.Name));
+            var groups = _source.GroupBy(_tdEngineChildTableRule);
+            var sql = new StringBuilder();
+            sql.Append("INSERT INTO ");
+            foreach (var group in groups)
+            {
+                string childTableName = group.Key;
+                var batchList = group.ToList();
+                // 2. 取第一条数据（用于生成 TAGS）
+                var first = batchList.First();
+                var tagValues = typePropertiesCached.Item2
+                    .Select(p => FormatValue(p.GetDbValue(first)))
+                    .ToArray();
+                var itemValueStringList = new List<string>();
+                foreach (var item in batchList)
+                {
+                    var itemValues = typePropertiesCached.Item3.Select(p => FormatValue(p.GetDbValue(item))).ToArray();
+                    var valueString = string.Join(", ", itemValues);
+                    itemValueStringList.Add($"({valueString})");
+                }
+
+                var values = string.Join(" ", itemValueStringList);
+                // 拼 TDengine SQL
+                sql.AppendLine().Append($"{childTableName} ") //子表名，不需要使用`包裹，可以忽略大小写
+                          .Append($"USING {typePropertiesCached.Item1} ") //超级表名
+                          .Append($"({tagNames}) ") //超级表名
+                          .Append($"tags({string.Join(", ", tagValues)}) ") //子表tags
+                          .AppendLine($"({valueColumnNames})") //除Tags和Ignored外的字段，不需要使用`包裹，可以忽略大小写
+                          .Append($"values {values}");
+            }
+            return sql.ToString();
+        }
+
+        private static string FormatValue(object v)
+        {
+            if (v == null)
+                return "NULL";
+
+            if (v is string s)
+            {
+                return "'" + s.Replace("'", "''") + "'";
+            }
+
+            if (v is DateTime d)
+            {
+                return $"'{d:O}'";
+            }
+
+            return v.ToString();
         }
 
         private static ConcurrentDictionary<Type, List<string>> _ignoreInsertColumns =