Merge pull request #1431 from d4ilys/master

QuestDb增加RestAPI解决并发问题

FreeSql.Tests/FreeSql.Tests/QuestDb/Crud/QuestDbTestInsertAndUpdate.cs

@@ -110,6 +110,124 @@ namespace FreeSql.Tests.QuestDb.Crud
             Assert.True(result > 0);
         }
 
+        [Fact]
+        public async Task TestRestInsertAsync()
+        {
+            var result = await restFsql.Insert(new QuestDb_Model_Test01()
+            {
+                Primarys = Guid.NewGuid().ToString(),
+                CreateTime = DateTime.Now,
+                Activos = 100.21,
+                Id = "IdAsync",
+                IsCompra = true,
+                NameInsert = "NameInsert",
+                NameUpdate = "NameUpdate"
+            }).ExecuteAffrowsAsync();
+            Assert.True(result > 0);
+        }
+
+        [Fact]
+        public async Task TestRestInsertBatchAsync()
+        {
+            var list = new List<QuestDb_Model_Test01>()
+            {
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "1",
+                    IsCompra = true,
+                    NameInsert = "NameInsertAsync",
+                    NameUpdate = "NameUpdate"
+                },
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "2",
+                    IsCompra = true,
+                    NameInsert = "NameInsertAsync",
+                    NameUpdate = "NameUpdate"
+                },
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "3",
+                    IsCompra = true,
+                    NameInsert = "NameInsertAsync",
+                    NameUpdate = "NameUpdate"
+                },
+            };
+            var result = await restFsql.Insert(list).ExecuteAffrowsAsync();
+            Assert.True(result > 0);
+        }
+
+        [Fact]
+        public async Task TestRestInsertColumnsAsync()
+        {
+            var list = new List<QuestDb_Model_Test01>()
+            {
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "1",
+                    IsCompra = true,
+                    NameInsert = "NameInsert",
+                    NameUpdate = "NameUpdate"
+                },
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "2",
+                    IsCompra = true,
+                    NameInsert = "NameInsert",
+                    NameUpdate = "NameUpdate"
+                },
+                new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100.21,
+                    Id = "3",
+                    IsCompra = true,
+                    NameInsert = "NameInsert",
+                    NameUpdate = "NameUpdate"
+                },
+            };
+            var result = await restFsql.Insert(list).IgnoreColumns(q => q.NameInsert).ExecuteAffrowsAsync();
+            Assert.True(result > 0);
+        }
+
+        [Fact]
+        public async Task TestSqlBulkCopy()
+        {
+            var list = new List<QuestDb_Model_Test01>();
+            for (int i = 0; i < 10; i++)
+            {
+                list.Add(new QuestDb_Model_Test01()
+                {
+                    Primarys = Guid.NewGuid().ToString(),
+                    CreateTime = DateTime.Now,
+                    Activos = 100 + i,
+                    Id = "1",
+                    IsCompra = true,
+                    NameInsert = "NameInsertAsync",
+                    NameUpdate = "NameUpdate"
+                });
+            }
+            var result = await restFsql.Insert(list).ExecuteBulkCopyAsync();
+            Assert.True(result > 0);
+        }
+
+
         [Fact, Order(4)]
         public void TestNormalUpdate()
         {
@@ -209,5 +327,7 @@ WHERE (""Id"" = '{primary}')", sql);
                 .Set(q => q.UpdateTime, DateTime.Now)
                 .ExecuteAffrowsAsync();
         }
+
+
     }
 }

FreeSql.Tests/FreeSql.Tests/QuestDb/Crud/QuestDbTestUpdate.cs

@@ -14,6 +14,104 @@ namespace FreeSql.Tests.QuestDb.Crud
     [TestCaseOrderer("FreeSql.Tests.QuestDb.Utils.TestOrders", "FreeSql.Tests")]
     public class QuestDbTestUpdate
     {
-        //多线程以及questdb问题转移至 insert中测试
+        [Fact]
+        public void TestNormalRestUpdate()
+        {
+            var updateTime = DateTime.Now;
+            var updateObj = restFsql.Update<QuestDb_Model_Test01>()
+                .Set(q => q.NameUpdate, "UpdateNow")
+                //     .Set(q => q.CreateTime, DateTime.Now)   分表的时间不可以随便改
+                .Where(q => q.Id == "1");
+            var updateSql = updateObj.ToSql();
+            Debug.WriteLine(updateSql);
+            var sql =
+                $@"UPDATE ""QuestDb_Model_Test01"" SET ""NameUpdate"" = 'UpdateNow' 
+WHERE (""Id"" = '1')";
+            Debug.WriteLine(sql);
+            Assert.Equal(updateSql, sql);
+            var result = updateObj.ExecuteAffrows();
+            Assert.True(result > 0);
+        }
+
+        [Fact]
+        public void TestRestUpdateByModel()
+        {
+            var primary = Guid.NewGuid().ToString();
+            //先插入
+            restFsql.Insert(new QuestDb_Model_Test01()
+            {
+                Primarys = primary,
+                CreateTime = DateTime.Now,
+                Activos = 100.21,
+                Id = primary,
+                IsCompra = true,
+                NameInsert = "NameInsert",
+                NameUpdate = "NameUpdate"
+            }).ExecuteAffrows();
+            var updateModel = new QuestDb_Model_Test01
+            {
+                Primarys = primary,
+                Id = primary,
+                Activos = 12.65,
+            };
+            var updateObj = restFsql.Update<QuestDb_Model_Test01>().SetSourceIgnore(updateModel, o => o == null);
+            var sql = updateObj.ToSql();
+            Debug.WriteLine(sql);
+            var result = updateObj.ExecuteAffrows();
+            var resultAsync = restFsql.Update<QuestDb_Model_Test01>().SetSourceIgnore(updateModel, o => o == null)
+                .ExecuteAffrows();
+            Assert.True(result > 0);
+            Assert.True(resultAsync > 0);
+            Assert.Equal(
+                @$"UPDATE ""QuestDb_Model_Test01"" SET ""Primarys"" = '{primary}', ""NameInsert"" = 'NameDefault', ""Activos"" = 12.65 
+WHERE (""Id"" = '{primary}')", sql);
+        }
+
+        [Fact]
+        public async Task TestRestUpdateIgnoreColumnsAsync()
+        {
+            var primary = Guid.NewGuid().ToString();
+            var updateTime = DateTime.Now;
+            //先插入
+            restFsql.Insert(new QuestDb_Model_Test01()
+            {
+                Primarys = primary,
+                CreateTime = DateTime.Now,
+                Activos = 100.21,
+                Id = primary,
+                IsCompra = true,
+                NameInsert = "NameInsert",
+                NameUpdate = "NameUpdate"
+            }).ExecuteAffrows();
+            var updateModel = new QuestDb_Model_Test01
+            {
+                Id = primary,
+                Activos = 12.65,
+                IsCompra = true,
+                CreateTime = DateTime.Now
+            };
+            var updateObj = restFsql.Update<QuestDb_Model_Test01>().SetSource(updateModel)
+                .IgnoreColumns(q => new { q.Id, q.CreateTime });
+            var sql = updateObj.ToSql();
+            Debug.WriteLine(sql);
+            var result = updateObj.ExecuteAffrows();
+            var resultAsync = await restFsql.Update<QuestDb_Model_Test01>().SetSource(updateModel)
+                .IgnoreColumns(q => new { q.Id, q.CreateTime }).ExecuteAffrowsAsync();
+            Assert.True(result > 0);
+            Assert.True(resultAsync > 0);
+            Assert.Equal(
+                $@"UPDATE ""QuestDb_Model_Test01"" SET ""Primarys"" = NULL, ""NameUpdate"" = NULL, ""NameInsert"" = 'NameDefault', ""Activos"" = 12.65, ""UpdateTime"" = NULL, ""IsCompra"" = True 
+WHERE (""Id"" = '{primary}')", sql);
+        }
+
+        [Fact]
+        public async Task TestUpdateToUpdateAsync()
+        {
+            //官网demo有问题，暂时放弃此功能
+            var result = await restFsql.Select<QuestDb_Model_Test01>().Where(q => q.Id == "IdAsync" && q.NameInsert == null)
+                .ToUpdate()
+                .Set(q => q.UpdateTime, DateTime.Now)
+                .ExecuteAffrowsAsync();
+        }
     }
 }

FreeSql.Tests/FreeSql.Tests/QuestDb/QuestDbTest.cs

@@ -15,5 +15,12 @@ namespace FreeSql.Tests.QuestDb
             .UseMonitorCommand(cmd => Debug.WriteLine($"Sql：{cmd.CommandText}")) //监听SQL语句
             .UseNoneCommandParameter(true)
             .Build();
+
+        public static IFreeSql restFsql = new FreeSql.FreeSqlBuilder()
+            .UseConnectionString(FreeSql.DataType.QuestDb,
+                @"host=192.168.0.36;port=8812;username=admin;password=quest;database=qdb;ServerCompatibilityMode=NoTypeLoading;")
+            .UseMonitorCommand(cmd => Debug.WriteLine($"Sql：{cmd.CommandText}")) //监听SQL语句
+            .UseQuestDbRestAPI("192.168.0.36:9001", "admin", "ushahL(aer2r")
+            .Build();
     }
 }

Providers/FreeSql.Provider.QuestDb/Curd/QuestDbInsert.cs

@@ -1,6 +1,9 @@
 using FreeSql.Internal;
 using FreeSql.Internal.Model;
+using NetTopologySuite.Mathematics;
+using Newtonsoft.Json;
 using System;
+using System.Collections;
 using System.Collections.Generic;
 using System.Data;
 using System.Data.Common;
@@ -11,7 +14,6 @@ using System.Threading.Tasks;
 
 namespace FreeSql.QuestDb.Curd
 {
-
     class QuestDbInsert<T1> : Internal.CommonProvider.InsertProvider<T1> where T1 : class
     {
         public QuestDbInsert(IFreeSql orm, CommonUtils commonUtils, CommonExpression commonExpression)
@@ -30,9 +32,42 @@ namespace FreeSql.QuestDb.Curd
         internal Dictionary<string, bool> InternalIgnore => _ignore;
         internal void InternalClearData() => ClearData();
 
-        public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+        private int InternelExecuteAffrows()
-        public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+        {
-        public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+            //如果设置了RestAPI的Url则走HTTP
+            var sql = ToSql();
+            var execAsync = RestAPIExtension.ExecAsync(sql).GetAwaiter().GetResult();
+            var resultHash = new Hashtable();
+            try
+            {
+                resultHash = JsonConvert.DeserializeObject<Hashtable>(execAsync);
+            }
+            catch
+            {
+                if (execAsync.Contains("401"))
+                {
+                    throw new Exception("请确认QuestDb设置的RestAPI账号是否正确.");
+                }
+            }
+            var ddl = resultHash["ddl"]?.ToString();
+            return ddl?.ToLower() == "ok" ? 1 : 0;
+        }
+
+        public override int ExecuteAffrows()
+        {
+            if (string.IsNullOrWhiteSpace(RestAPIExtension.BaseUrl))
+            {
+                return base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 5000,
+                    _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+            }
+            return InternelExecuteAffrows();
+        }
+
+        public override long ExecuteIdentity() => base.SplitExecuteIdentity(
+            _batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+
+        public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(
+            _batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
 
         protected override long RawExecuteIdentity()
         {
@@ -50,7 +85,8 @@ namespace FreeSql.QuestDb.Curd
                 _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
                 try
                 {
-                    ret = _orm.Ado.ExecuteNonQuery(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params);
+                    ret = _orm.Ado.ExecuteNonQuery(_connection, _transaction, CommandType.Text, sql, _commandTimeout,
+                        _params);
                 }
                 catch (Exception ex)
                 {
@@ -62,14 +98,18 @@ namespace FreeSql.QuestDb.Curd
                     var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                     _orm.Aop.CurdAfterHandler?.Invoke(this, after);
                 }
+
                 return 0;
             }
+
             sql = string.Concat(sql, " RETURNING ", _commonUtils.QuoteSqlName(identCols.First().Value.Attribute.Name));
             before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
             _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
             try
             {
-                long.TryParse(string.Concat(_orm.Ado.ExecuteScalar(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params)), out ret);
+                long.TryParse(
+                    string.Concat(_orm.Ado.ExecuteScalar(_connection, _transaction, CommandType.Text, sql,
+                        _commandTimeout, _params)), out ret);
             }
             catch (Exception ex)
             {
@@ -81,6 +121,7 @@ namespace FreeSql.QuestDb.Curd
                 var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                 _orm.Aop.CurdAfterHandler?.Invoke(this, after);
             }
+
             return ret;
         }
 
@@ -96,9 +137,11 @@ namespace FreeSql.QuestDb.Curd
             foreach (var col in _table.Columns.Values)
             {
                 if (colidx > 0) sb.Append(", ");
-                sb.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
+                sb.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ")
+                    .Append(_commonUtils.QuoteSqlName(col.CsName));
                 ++colidx;
             }
+
             sql = sb.ToString();
             var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
             _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
@@ -106,7 +149,8 @@ namespace FreeSql.QuestDb.Curd
             Exception exception = null;
             try
             {
-                ret = _orm.Ado.Query<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction, CommandType.Text, sql, _commandTimeout, _params);
+                ret = _orm.Ado.Query<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction, CommandType.Text,
+                    sql, _commandTimeout, _params);
             }
             catch (Exception ex)
             {
@@ -118,14 +162,30 @@ namespace FreeSql.QuestDb.Curd
                 var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                 _orm.Aop.CurdAfterHandler?.Invoke(this, after);
             }
+
             return ret;
         }
 
 #if net40
 #else
-        public override Task<int> ExecuteAffrowsAsync(CancellationToken cancellationToken = default) => base.SplitExecuteAffrowsAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+        public override Task<int> ExecuteAffrowsAsync(CancellationToken cancellationToken = default)
-        public override Task<long> ExecuteIdentityAsync(CancellationToken cancellationToken = default) => base.SplitExecuteIdentityAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+        {
-        public override Task<List<T1>> ExecuteInsertedAsync(CancellationToken cancellationToken = default) => base.SplitExecuteInsertedAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000, _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+            if (string.IsNullOrWhiteSpace(RestAPIExtension.BaseUrl))
+            {
+                return base.SplitExecuteAffrowsAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000,
+                    _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+            }
+            return Task.FromResult(InternelExecuteAffrows());
+        }
+        
+
+        public override Task<long> ExecuteIdentityAsync(CancellationToken cancellationToken = default) =>
+            base.SplitExecuteIdentityAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000,
+                _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+
+        public override Task<List<T1>> ExecuteInsertedAsync(CancellationToken cancellationToken = default) =>
+            base.SplitExecuteInsertedAsync(_batchValuesLimit > 0 ? _batchValuesLimit : 5000,
+                _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
 
         async protected override Task<long> RawExecuteIdentityAsync(CancellationToken cancellationToken = default)
         {
@@ -143,7 +203,8 @@ namespace FreeSql.QuestDb.Curd
                 _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
                 try
                 {
-                    ret = await _orm.Ado.ExecuteNonQueryAsync(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params, cancellationToken);
+                    ret = await _orm.Ado.ExecuteNonQueryAsync(_connection, _transaction, CommandType.Text, sql,
+                        _commandTimeout, _params, cancellationToken);
                 }
                 catch (Exception ex)
                 {
@@ -155,14 +216,18 @@ namespace FreeSql.QuestDb.Curd
                     var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                     _orm.Aop.CurdAfterHandler?.Invoke(this, after);
                 }
+
                 return 0;
             }
+
             sql = string.Concat(sql, " RETURNING ", _commonUtils.QuoteSqlName(identCols.First().Value.Attribute.Name));
             before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
             _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
             try
             {
-                long.TryParse(string.Concat(await _orm.Ado.ExecuteScalarAsync(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params, cancellationToken)), out ret);
+                long.TryParse(
+                    string.Concat(await _orm.Ado.ExecuteScalarAsync(_connection, _transaction, CommandType.Text, sql,
+                        _commandTimeout, _params, cancellationToken)), out ret);
             }
             catch (Exception ex)
             {
@@ -174,8 +239,10 @@ namespace FreeSql.QuestDb.Curd
                 var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                 _orm.Aop.CurdAfterHandler?.Invoke(this, after);
             }
+
             return ret;
         }
+
         async protected override Task<List<T1>> RawExecuteInsertedAsync(CancellationToken cancellationToken = default)
         {
             var sql = this.ToSql();
@@ -188,9 +255,11 @@ namespace FreeSql.QuestDb.Curd
             foreach (var col in _table.Columns.Values)
             {
                 if (colidx > 0) sb.Append(", ");
-                sb.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
+                sb.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ")
+                    .Append(_commonUtils.QuoteSqlName(col.CsName));
                 ++colidx;
             }
+
             sql = sb.ToString();
             var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
             _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
@@ -198,7 +267,8 @@ namespace FreeSql.QuestDb.Curd
             Exception exception = null;
             try
             {
-                ret = await _orm.Ado.QueryAsync<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction, CommandType.Text, sql, _commandTimeout, _params, cancellationToken);
+                ret = await _orm.Ado.QueryAsync<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction,
+                    CommandType.Text, sql, _commandTimeout, _params, cancellationToken);
             }
             catch (Exception ex)
             {
@@ -210,6 +280,7 @@ namespace FreeSql.QuestDb.Curd
                 var after = new Aop.CurdAfterEventArgs(before, exception, ret);
                 _orm.Aop.CurdAfterHandler?.Invoke(this, after);
             }
+
             return ret;
         }
 #endif

Providers/FreeSql.Provider.QuestDb/Curd/QuestDbUpdate.cs

@@ -1,6 +1,8 @@
 using FreeSql.Internal;
 using FreeSql.Internal.Model;
+using Newtonsoft.Json;
 using System;
+using System.Collections;
 using System.Collections.Generic;
 using System.Data;
 using System.Data.Common;
@@ -11,10 +13,8 @@ using System.Threading.Tasks;
 
 namespace FreeSql.QuestDb.Curd
 {
-
     class QuestDbUpdate<T1> : Internal.CommonProvider.UpdateProvider<T1>
     {
-
         public QuestDbUpdate(IFreeSql orm, CommonUtils commonUtils, CommonExpression commonExpression, object dywhere)
             : base(orm, commonUtils, commonExpression, dywhere)
         {
@@ -25,11 +25,46 @@ namespace FreeSql.QuestDb.Curd
         internal StringBuilder InternalSbSetIncr => _setIncr;
         internal Dictionary<string, bool> InternalIgnore => _ignore;
         internal void InternalResetSource(List<T1> source) => _source = source;
-        internal string InternalWhereCaseSource(string CsName, Func<string, string> thenValue) => WhereCaseSource(CsName, thenValue);
+
+        internal string InternalWhereCaseSource(string CsName, Func<string, string> thenValue) =>
+            WhereCaseSource(CsName, thenValue);
+
         internal void InternalToSqlCaseWhenEnd(StringBuilder sb, ColumnInfo col) => ToSqlCaseWhenEnd(sb, col);
 
-        public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchRowsLimit > 0 ? _batchRowsLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+        private int InternelExecuteAffrows()
-        public override List<T1> ExecuteUpdated() => base.SplitExecuteUpdated(_batchRowsLimit > 0 ? _batchRowsLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+        {
+            //如果设置了RestAPI的Url则走HTTP
+            var sql = ToSql();
+            var execAsync = RestAPIExtension.ExecAsync(sql).GetAwaiter().GetResult();
+            var resultHash = new Hashtable();
+            try
+            {
+                resultHash = JsonConvert.DeserializeObject<Hashtable>(execAsync);
+            }
+            catch
+            {
+                if (execAsync.Contains("401"))
+                {
+                    throw new Exception("请确认QuestDb设置的RestAPI账号是否正确.");
+                }
+            }
+            var ddl = resultHash["ddl"]?.ToString();
+            var updated = Convert.ToInt32(resultHash["updated"]);
+            return ddl?.ToLower() == "ok" ? updated : 0;
+        }
+
+        public override int ExecuteAffrows()
+        {
+            if (string.IsNullOrWhiteSpace(RestAPIExtension.BaseUrl))
+            {
+                return base.SplitExecuteAffrows(_batchRowsLimit > 0 ? _batchRowsLimit : 500,
+                    _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
+            }
+            return InternelExecuteAffrows();
+        }
+
+        public override List<T1> ExecuteUpdated() => base.SplitExecuteUpdated(
+            _batchRowsLimit > 0 ? _batchRowsLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 3000);
 
         protected override List<T1> RawExecuteUpdated()
         {
@@ -48,10 +83,12 @@ namespace FreeSql.QuestDb.Curd
                     foreach (var col in _table.Columns.Values)
                     {
                         if (colidx > 0) sbret.Append(", ");
-                        sbret.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
+                        sbret.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name)))
+                            .Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
                         ++colidx;
                     }
                 }
+
                 var sql = sb.Append(sbret).ToString();
                 var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Update, sql, dbParms);
                 _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
@@ -59,7 +96,8 @@ namespace FreeSql.QuestDb.Curd
                 Exception exception = null;
                 try
                 {
-                    var rettmp = _orm.Ado.Query<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction, CommandType.Text, sql, _commandTimeout, dbParms);
+                    var rettmp = _orm.Ado.Query<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction,
+                        CommandType.Text, sql, _commandTimeout, dbParms);
                     ValidateVersionAndThrow(rettmp.Count, sql, dbParms);
                     ret.AddRange(rettmp);
                 }
@@ -87,15 +125,18 @@ namespace FreeSql.QuestDb.Curd
                 caseWhen.Append(_commonUtils.RereadColumn(pk, _commonUtils.QuoteSqlName(pk.Attribute.Name)));
                 return;
             }
+
             caseWhen.Append("(");
             var pkidx = 0;
             foreach (var pk in primarys)
             {
                 if (pkidx > 0) caseWhen.Append(" || '+' || ");
                 if (string.IsNullOrEmpty(InternalTableAlias) == false) caseWhen.Append(InternalTableAlias).Append(".");
-                caseWhen.Append(_commonUtils.RereadColumn(pk, _commonUtils.QuoteSqlName(pk.Attribute.Name))).Append("::text");
+                caseWhen.Append(_commonUtils.RereadColumn(pk, _commonUtils.QuoteSqlName(pk.Attribute.Name)))
+                    .Append("::text");
                 ++pkidx;
             }
+
             caseWhen.Append(")");
         }
 
@@ -106,6 +147,7 @@ namespace FreeSql.QuestDb.Curd
                 sb.Append(_commonUtils.FormatSql("{0}", primarys[0].GetDbValue(d)));
                 return;
             }
+
             sb.Append("(");
             var pkidx = 0;
             foreach (var pk in primarys)
@@ -114,6 +156,7 @@ namespace FreeSql.QuestDb.Curd
                 sb.Append(_commonUtils.FormatSql("{0}", pk.GetDbValue(d))).Append("::text");
                 ++pkidx;
             }
+
             sb.Append(")");
         }
 
@@ -125,6 +168,7 @@ namespace FreeSql.QuestDb.Curd
                 sb.Append("::text");
                 return;
             }
+
             var dbtype = _commonUtils.CodeFirst.GetDbInfo(col.Attribute.MapType)?.dbtype;
             if (dbtype == null) return;
 
@@ -133,8 +177,21 @@ namespace FreeSql.QuestDb.Curd
 
 #if net40
 #else
-        public override Task<int> ExecuteAffrowsAsync(CancellationToken cancellationToken = default) => base.SplitExecuteAffrowsAsync(_batchRowsLimit > 0 ? _batchRowsLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+        public override Task<int> ExecuteAffrowsAsync(CancellationToken cancellationToken = default)
-        public override Task<List<T1>> ExecuteUpdatedAsync(CancellationToken cancellationToken = default) => base.SplitExecuteUpdatedAsync(_batchRowsLimit > 0 ? _batchRowsLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+        {
+            if (string.IsNullOrWhiteSpace(RestAPIExtension.BaseUrl))
+            {
+                return base.SplitExecuteAffrowsAsync(_batchRowsLimit > 0 ? _batchRowsLimit : 500,
+                    _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
+            }
+         
+            return Task.FromResult(InternelExecuteAffrows());
+        }
+       
+
+        public override Task<List<T1>> ExecuteUpdatedAsync(CancellationToken cancellationToken = default) =>
+            base.SplitExecuteUpdatedAsync(_batchRowsLimit > 0 ? _batchRowsLimit : 500,
+                _batchParameterLimit > 0 ? _batchParameterLimit : 3000, cancellationToken);
 
         async protected override Task<List<T1>> RawExecuteUpdatedAsync(CancellationToken cancellationToken = default)
         {
@@ -153,10 +210,12 @@ namespace FreeSql.QuestDb.Curd
                     foreach (var col in _table.Columns.Values)
                     {
                         if (colidx > 0) sbret.Append(", ");
-                        sbret.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name))).Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
+                        sbret.Append(_commonUtils.RereadColumn(col, _commonUtils.QuoteSqlName(col.Attribute.Name)))
+                            .Append(" as ").Append(_commonUtils.QuoteSqlName(col.CsName));
                         ++colidx;
                     }
                 }
+
                 var sql = sb.Append(sbret).ToString();
                 var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Update, sql, dbParms);
                 _orm.Aop.CurdBeforeHandler?.Invoke(this, before);
@@ -164,7 +223,8 @@ namespace FreeSql.QuestDb.Curd
                 Exception exception = null;
                 try
                 {
-                    var rettmp = await _orm.Ado.QueryAsync<T1>(_table.TypeLazy ?? _table.Type, _connection, _transaction, CommandType.Text, sql, _commandTimeout, dbParms, cancellationToken);
+                    var rettmp = await _orm.Ado.QueryAsync<T1>(_table.TypeLazy ?? _table.Type, _connection,
+                        _transaction, CommandType.Text, sql, _commandTimeout, dbParms, cancellationToken);
                     ValidateVersionAndThrow(rettmp.Count, sql, dbParms);
                     ret.AddRange(rettmp);
                 }

Providers/FreeSql.Provider.QuestDb/FreeSql.Provider.QuestDb.csproj

@@ -4,29 +4,31 @@
 		<TargetFrameworks>netstandard2.0;</TargetFrameworks>
 		<GeneratePackageOnBuild>true</GeneratePackageOnBuild>
 		<Authors>YeXiangQin;Daily</Authors>
-		<Description>FreeSql适配QuestDb时序数据库访问</Description>
+		<Description>FreeSql实现QuestDb时序数据库访问</Description>
 		<PackageProjectUrl>https://github.com/2881099/FreeSql</PackageProjectUrl>
 		<RepositoryUrl>https://github.com/2881099/FreeSql</RepositoryUrl>
 		<RepositoryType>git</RepositoryType>
 		<PackageLicenseExpression>MIT</PackageLicenseExpression>
 		<PackageTags>FreeSql;ORM;QuestDb</PackageTags>
 		<PackageId>$(AssemblyName)</PackageId>
-		<PackageIcon>logo.png</PackageIcon>
+		<!--<PackageIcon>logo.png</PackageIcon>-->
 		<Title>$(AssemblyName)</Title>
 		<IsPackable>true</IsPackable>
 		<GenerateAssemblyInfo>true</GenerateAssemblyInfo>
 		<SignAssembly>true</SignAssembly>
 		<AssemblyOriginatorKeyFile>key.snk</AssemblyOriginatorKeyFile>
 		<DelaySign>false</DelaySign>
-		<Version>3.2.688-preview20230216</Version>
+		<Version>4.0.0.1</Version>
 	</PropertyGroup>
 
-	<ItemGroup>
+	<!--<ItemGroup>
 		<None Include="../../logo.png" Pack="true" PackagePath="\" />
-	</ItemGroup>
+	</ItemGroup>-->
 
 	<ItemGroup>
+		<PackageReference Include="Microsoft.Extensions.Http" Version="3.1.32" />
 		<PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
+		<PackageReference Include="CsvHelper" Version="30.0.1" />
 	</ItemGroup>
 
 	<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
@@ -41,7 +43,7 @@
 		<PackageReference Include="Npgsql.LegacyPostgis" Version="4.0.11" />
 	</ItemGroup>
 	<ItemGroup>
-	  <ProjectReference Include="..\..\FreeSql\FreeSql.csproj" />
+	  <ProjectReference Include="..\..\..\Demo\FreeSql\FreeSql\FreeSql.csproj" />
 	</ItemGroup>
 
 	<PropertyGroup Condition="'$(TargetFramework)' == 'net452' or '$(TargetFramework)' == 'net451' or '$(TargetFramework)' == 'net45'">

Providers/FreeSql.Provider.QuestDb/QuestDbContainer.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace FreeSql.Provider.QuestDb
+{
+    internal class QuestDbContainer
+    {
+        private static IServiceCollection Services;
+        public static IServiceProvider ServiceProvider { get; private set; }
+
+        internal static void Initialize(Action<IServiceCollection> service)
+        {
+            Services = new ServiceCollection();
+            service?.Invoke(Services);
+            ServiceProvider = Services.BuildServiceProvider();
+        }
+
+        internal static T GetService<T>()
+        {
+            return ServiceProvider.GetService<T>();
+        }
+    }
+}

Providers/FreeSql.Provider.QuestDb/QuestDbDbFirst.cs

@@ -717,10 +717,33 @@ namespace FreeSql.QuestDb
             return res;
         }
 
-        public DbTableInfo GetTableByName(string name, bool ignoreCase = true) =>
+        public DbTableInfo GetTableByName(string name, bool ignoreCase = true)
-            GetTables(null, name, ignoreCase)?.FirstOrDefault();
+        {
+            var tableColumns = _orm.Ado.ExecuteDataTable($"SHOW COLUMNS FROM '{name}'");
+            List<DbColumnInfo> dbColumnInfos = new List<DbColumnInfo>();
+            var dbTableInfo = new DbTableInfo()
+            {
+                Name = name,
+                Columns = new List<DbColumnInfo>()
+            };
+            foreach (DataRow tableColumnsRow in tableColumns.Rows)
+            {
+                dbColumnInfos.Add(new DbColumnInfo()
+                {
+                    Name = tableColumnsRow["column"].ToString(),
+                    DbTypeText = tableColumnsRow["type"].ToString(),
+                    Table = dbTableInfo,
+                });
+            }
 
-        public List<DbTableInfo> GetTablesByDatabase(params string[] database) => GetTables(database, null, false);
+            dbTableInfo.Columns = dbColumnInfos;
+            return dbTableInfo;
+        }
+
+        public List<DbTableInfo> GetTablesByDatabase(params string[] database)
+        {
+            return GetTables(database, null, false);
+        }
 
         public List<DbTableInfo> GetTables(string[] database, string tablename, bool ignoreCase)
         {

Providers/FreeSql.Provider.QuestDb/QuestDbGlobalExtensions.cs

@@ -1,253 +1,340 @@
-using FreeSql;
+using CsvHelper;
+using FreeSql;
 using FreeSql.Internal.CommonProvider;
 using FreeSql.Internal.Model;
 using FreeSql.QuestDb;
 using FreeSql.QuestDb.Curd;
+using Newtonsoft.Json;
 using Npgsql;
 using System;
+using System.Collections;
 using System.Collections.Generic;
 using System.Data;
+using System.Globalization;
+using System.IO;
 using System.Linq;
 using System.Linq.Expressions;
+using System.Net.Http;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Web;
+using FreeSql.Provider.QuestDb;
+using System.Net;
 
 public static partial class QuestDbGlobalExtensions
 {
-
     /// <summary>
     /// 特殊处理类似 string.Format 的使用方法，防止注入，以及 IS NULL 转换
     /// </summary>
     /// <param name="that"></param>
     /// <param name="args"></param>
     /// <returns></returns>
-    public static string FormatPostgreSQL(this string that, params object[] args) => _postgresqlAdo.Addslashes(that, args);
+    public static string FormatPostgreSQL(this string that, params object[] args) =>
+        _postgresqlAdo.Addslashes(that, args);
+
     static QuestDbAdo _postgresqlAdo = new QuestDbAdo();
 
+
     /// <summary>
-    /// PostgreSQL9.5+ 特有的功能，On Conflict Do Update<para></para>
+    /// 逐行读取，包含空行
-    /// 注意：此功能会开启插入【自增列】
+    /// </summary>
+    /// <param name="text"></param>
+    /// <returns></returns>
+    private static List<string> SplitByLine(string text)
+    {
+        List<string> lines = new List<string>();
+        byte[] array = Encoding.UTF8.GetBytes(text);
+        using (MemoryStream stream = new MemoryStream(array))
+        {
+            using (var sr = new StreamReader(stream))
+            {
+                string line = sr.ReadLine();
+                while (line != null)
+                {
+                    lines.Add(line);
+                    line = sr.ReadLine();
+                }
+
+                ;
+            }
+        }
+
+        return lines;
+    }
+
+    /// <summary>
+    /// 批量快速插入
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    /// <param name="that"></param>
+    /// <returns></returns>
+    public static async Task<int> ExecuteBulkCopyAsync<T>(this IInsert<T> that) where T : class
+    {
+        var result = 0;
+        var fileName = $"{Guid.NewGuid()}.csv";
+        var filePath = Path.Combine(AppContext.BaseDirectory, fileName);
+        try
+        {
+            var client = QuestDbContainer.GetService<IHttpClientFactory>().CreateClient();
+            var boundary = "---------------" + DateTime.Now.Ticks.ToString("x");
+            var name = typeof(T).Name;
+            var list = new List<Hashtable>();
+            var insert = that as QuestDbInsert<T>;
+            insert.InternalOrm.DbFirst.GetTableByName(name).Columns.ForEach(d =>
+            {
+                if (d.DbTypeText == "TIMESTAMP")
+                {
+                    list.Add(new Hashtable()
+                    {
+                        { "name", d.Name },
+                        { "type", d.DbTypeText },
+                        { "pattern", "yyyy/M/dd HH:mm:ss" }
+                    });
+                }
+                else
+                {
+                    list.Add(new Hashtable()
+                    {
+                        { "name", d.Name },
+                        { "type", d.DbTypeText }
+                    });
+                }
+            });
+            var schema = JsonConvert.SerializeObject(list);
+            using (var writer = new StreamWriter(filePath))
+            using (var csv = new CsvWriter(writer, CultureInfo.CurrentCulture))
+            {
+                csv.WriteRecords(insert._source);
+            }
+
+            var httpContent = new MultipartFormDataContent(boundary);
+            if (!string.IsNullOrWhiteSpace(RestAPIExtension.authorization))
+                client.DefaultRequestHeaders.Add("Authorization", RestAPIExtension.authorization);
+            httpContent.Add(new StringContent(schema), "schema");
+            httpContent.Add(new ByteArrayContent(File.ReadAllBytes(filePath)), "data");
+            //boundary带双引号 可能导致服务器错误情况
+            httpContent.Headers.Remove("Content-Type");
+            httpContent.Headers.TryAddWithoutValidation("Content-Type",
+                "multipart/form-data; boundary=" + boundary);
+            var httpResponseMessage =
+                await client.PostAsync($"{RestAPIExtension.BaseUrl}/imp?name={name}", httpContent);
+            var readAsStringAsync = await httpResponseMessage.Content.ReadAsStringAsync();
+            var splitByLine = SplitByLine(readAsStringAsync);
+            Console.WriteLine(readAsStringAsync);
+            foreach (var s in splitByLine)
+            {
+                if (s.Contains("Rows"))
+                {
+                    var strings = s.Split('|');
+                    if (strings[1].Trim() == "Rows imported")
+                    {
+                        result = Convert.ToInt32(strings[2].Trim());
+                    }
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            Console.WriteLine(e);
+            throw;
+        }
+        finally
+        {
+            try
+            {
+                File.Delete(filePath);
+            }
+            catch
+            {
+            }
+        }
+
+        return result;
+    }
+
+    /// <summary>
+    /// 批量快速插入
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    /// <param name="that"></param>
+    /// <returns></returns>
+    public static int ExecuteBulkCopy<T>(this IInsert<T> insert) where T : class
+    {
+        return ExecuteBulkCopyAsync(insert).GetAwaiter().GetResult();
+    }
+}
+
+public static class SampleByExtension
+{
+    //是否使用该方法
+    internal static AsyncLocal<bool> IsExistence = new AsyncLocal<bool>()
+    {
+        Value = false
+    };
+
+    internal static AsyncLocal<string> SamoleByString = new AsyncLocal<string>()
+    {
+        Value = string.Empty
+    };
+
+    internal static void Initialize()
+    {
+        IsExistence.Value = false;
+        SamoleByString.Value = string.Empty;
+    }
+
+    /// <summary>
+    /// SAMPLE BY用于时间序列数据，将大型数据集汇总为同质时间块的聚合，作为SELECT语句的一部分。对缺少数据的数据集执行查询的用户可以使用FILL关键字指定填充行为
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    /// <param name="select"></param>
+    /// <param name="time">时长</param>
+    /// <param name="unit">单位</param>
+    /// <returns></returns>
+    public static ISelect<T> SampleBy<T>(this ISelect<T> select, double time, SampleUnits unit)
+    {
+        var _unit = Enum.GetName(typeof(SampleUnits), unit);
+        IsExistence.Value = true;
+        var samoleByTemple = $"{Environment.NewLine}SAMPLE BY {{0}}{{1}}{Environment.NewLine}";
+        SamoleByString.Value = string.Format(samoleByTemple, time.ToString(), _unit);
+        return select;
+    }
+}
+
+public static class LatestOnExtension
+{
+    //是否使用该方法
+    internal static AsyncLocal<bool> IsExistence = new AsyncLocal<bool>()
+    {
+        Value = false
+    };
+
+    internal static AsyncLocal<string> LatestOnString = new AsyncLocal<string>()
+    {
+        Value = string.Empty
+    };
+
+    internal static void Initialize()
+    {
+        IsExistence.Value = false;
+        LatestOnString.Value = string.Empty;
+    }
+
+    /// <summary>
+    /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
     /// </summary>
     /// <typeparam name="T1"></typeparam>
-    /// <param name="that"></param>
+    /// <typeparam name="TKey"></typeparam>
-    /// <param name="columns">默认是以主键作为重复判断，也可以指定其他列：a => a.Name | a => new{a.Name,a.Time} | a => new[]{"name","time"}</param>
+    /// <param name="select"></param>
+    /// <param name="timestamp">时间标识</param>
+    /// <param name="partition">最新项的列</param>
     /// <returns></returns>
-    public static OnConflictDoUpdate<T1> OnConflictDoUpdate<T1>(this IInsert<T1> that, Expression<Func<T1, object>> columns = null) where T1 : class => new OnConflictDoUpdate<T1>(that.InsertIdentity(), columns);
+    public static ISelect<T1> LatestOn<T1, TKey>(this ISelect<T1> select, Expression<Func<T1, DateTime?>> timestamp,
-
+        Expression<Func<T1, TKey>> partition)
-    #region ExecutePgCopy
-    /// <summary>
-    /// 批量更新（更新字段数量超过 2000 时收益大）<para></para>
-    /// 实现原理：使用 PgCopy 插入临时表，再使用 UPDATE INNER JOIN 联表更新
-    /// </summary>
-    /// <typeparam name="T"></typeparam>
-    /// <param name="that"></param>
-    /// <returns></returns>
-    public static int ExecutePgCopy<T>(this IUpdate<T> that) where T : class
     {
-        var update = that as UpdateProvider<T>;
+        Provider(timestamp, partition);
-        if (update._source.Any() != true || update._tempPrimarys.Any() == false) return 0;
+        return select;
-        var state = ExecutePgCopyState(update);
-        return UpdateProvider.ExecuteBulkUpdate(update, state, insert => insert.ExecutePgCopy());
     }
-    static NativeTuple<string, string, string, string, string[]> ExecutePgCopyState<T>(UpdateProvider<T> update) where T : class
-    {
-        if (update._source.Any() != true) return null;
-        var _table = update._table;
-        var _commonUtils = update._commonUtils;
-        var updateTableName = update._tableRule?.Invoke(_table.DbName) ?? _table.DbName;
-        var tempTableName = $"Temp_{Guid.NewGuid().ToString("N")}";
-        if (update._orm.CodeFirst.IsSyncStructureToLower) tempTableName = tempTableName.ToLower();
-        if (update._orm.CodeFirst.IsSyncStructureToUpper) tempTableName = tempTableName.ToUpper();
-        if (update._connection == null && update._orm.Ado.TransactionCurrentThread != null)
-            update.WithTransaction(update._orm.Ado.TransactionCurrentThread);
-        var sb = new StringBuilder().Append("CREATE TEMP TABLE ").Append(_commonUtils.QuoteSqlName(tempTableName)).Append(" ( ");
-        var setColumns = new List<string>();
-        var pkColumns = new List<string>();
-        foreach (var col in _table.Columns.Values)
-        {
-            if (update._tempPrimarys.Any(a => a.CsName == col.CsName)) pkColumns.Add(col.Attribute.Name);
-            else if (col.Attribute.IsIdentity == false && col.Attribute.IsVersion == false && update._ignore.ContainsKey(col.Attribute.Name) == false) setColumns.Add(col.Attribute.Name);
-            else continue;
-            sb.Append(" \r\n  ").Append(_commonUtils.QuoteSqlName(col.Attribute.Name)).Append(" ").Append(col.Attribute.DbType.Replace("NOT NULL", ""));
-            sb.Append(",");
-        }
-        var sql1 = sb.Remove(sb.Length - 1, 1).Append("\r\n) WITH (OIDS=FALSE);").ToString();
 
-        sb.Clear().Append("UPDATE ").Append(_commonUtils.QuoteSqlName(updateTableName)).Append(" a ")
+    private static void Provider<T1, TKey>(Expression<Func<T1, DateTime?>> timestamp,
-            .Append("\r\nSET \r\n  ").Append(string.Join(", \r\n  ", setColumns.Select(col => $"{_commonUtils.QuoteSqlName(col)} = b.{_commonUtils.QuoteSqlName(col)}")))
+        Expression<Func<T1, TKey>> partition)
-            .Append("\r\nFROM ").Append(_commonUtils.QuoteSqlName(tempTableName)).Append(" b ")
+    {
-            .Append("\r\nWHERE ").Append(string.Join(" AND ", pkColumns.Select(col => $"a.{_commonUtils.QuoteSqlName(col)} = b.{_commonUtils.QuoteSqlName(col)}")));
+        IsExistence.Value = true;
-        var sql2 = sb.ToString();
+        var latestOnTemple = $"{Environment.NewLine}LATEST ON {{0}} PARTITION BY {{1}} ";
-        sb.Clear();
+        var expressionVisitor = new QuestDbExpressionVisitor();
-        var sql3 = $"DROP TABLE {_commonUtils.QuoteSqlName(tempTableName)}";
+        expressionVisitor.Visit(timestamp);
-        return NativeTuple.Create(sql1, sql2, sql3, tempTableName, pkColumns.Concat(setColumns).ToArray());
+        var _timestamp = expressionVisitor.Fields();
+        expressionVisitor.Visit(partition);
+        var _partition = expressionVisitor.Fields();
+        LatestOnString.Value = string.Format(latestOnTemple, _timestamp, _partition);
     }
 
     /// <summary>
-    /// PostgreSQL COPY 批量导入功能，封装了 NpgsqlConnection.BeginBinaryImport 方法<para></para>
+    /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-    /// 使用 IgnoreColumns/InsertColumns 设置忽略/指定导入的列<para></para>
-    /// 使用 WithConnection/WithTransaction 传入连接/事务对象<para></para>
-    /// 提示：若本方法不能满足，请使用 IInsert&lt;T&gt;.ToDataTable 方法得到 DataTable 对象后，自行处理。<para></para>
-    /// COPY 与 insert into t values(..),(..),(..) 性能测试参考：<para></para>
-    /// 插入180000行，52列：10,090ms 与 46,756ms，10列：4,081ms 与 9,786ms<para></para>
-    /// 插入10000行，52列：583ms 与 3,294ms，10列：167ms 与 568ms<para></para>
-    /// 插入5000行，52列：337ms 与 2,269ms，10列：93ms 与 366ms<para></para>
-    /// 插入2000行，52列：136ms 与 1,019ms，10列：39ms 与 157ms<para></para>
-    /// 插入1000行，52列：88ms 与 374ms，10列：21ms 与 102ms<para></para>
-    /// 插入500行，52列：61ms 与 209ms，10列：12ms 与 34ms<para></para>
-    /// 插入100行，52列：30ms 与 51ms，10列：4ms 与 9ms<para></para>
-    /// 插入50行，52列：25ms 与 37ms，10列：2ms 与 6ms<para></para>
     /// </summary>
     /// <typeparam name="T"></typeparam>
-    /// <param name="that"></param>
+    /// <typeparam name="TKey"></typeparam>
-    public static void ExecutePgCopy<T>(this IInsert<T> that) where T : class
+    /// <param name="select"></param>
+    /// <param name="timestamp">时间标识</param>
+    /// <param name="partition">最新项的列</param>
+    /// <returns></returns>
+    public static ISelect<T1, T2> LatestOn<T1, T2, TKey>(this ISelect<T1, T2> select,
+        Expression<Func<T1, DateTime?>> timestamp,
+        Expression<Func<T1, TKey>> partition) where T2 : class
     {
-        var insert = that as QuestDbInsert<T>;
+        Provider(timestamp, partition);
-        if (insert == null) throw new Exception(CoreStrings.S_Features_Unique("ExecutePgCopy", "PostgreSQL"));
+        return select;
-
-        var dt = that.ToDataTable();
-        if (dt.Rows.Count == 0) return;
-
-        Action<NpgsqlConnection> binaryImport = conn =>
-        {
-            var copyFromCommand = new StringBuilder().Append("COPY ").Append(insert.InternalCommonUtils.QuoteSqlName(dt.TableName)).Append("(");
-            var colIndex = 0;
-            foreach (DataColumn col in dt.Columns)
-            {
-                if (colIndex++ > 0) copyFromCommand.Append(", ");
-                copyFromCommand.Append(insert.InternalCommonUtils.QuoteSqlName(col.ColumnName));
-            }
-            copyFromCommand.Append(") FROM STDIN BINARY");
-            using (var writer = conn.BeginBinaryImport(copyFromCommand.ToString()))
-            {
-                foreach (DataRow item in dt.Rows)
-                    writer.WriteRow(item.ItemArray);
-                writer.Complete();
-            }
-            copyFromCommand.Clear();
-        };
-
-        try
-        {
-            if (insert.InternalConnection == null && insert.InternalTransaction == null)
-            {
-                using (var conn = insert.InternalOrm.Ado.MasterPool.Get())
-                {
-                    binaryImport(conn.Value as NpgsqlConnection);
-                }
-            }
-            else if (insert.InternalTransaction != null)
-            {
-                binaryImport(insert.InternalTransaction.Connection as NpgsqlConnection);
-            }
-            else if (insert.InternalConnection != null)
-            {
-                var conn = insert.InternalConnection as NpgsqlConnection;
-                var isNotOpen = false;
-                if (conn.State != System.Data.ConnectionState.Open)
-                {
-                    isNotOpen = true;
-                    conn.Open();
-                }
-                try
-                {
-                    binaryImport(conn);
-                }
-                finally
-                {
-                    if (isNotOpen)
-                        conn.Close();
-                }
-            }
-            else
-            {
-                throw new NotImplementedException($"ExecutePgCopy {CoreStrings.S_Not_Implemented_FeedBack}");
-            }
-        }
-        finally
-        {
-            dt.Clear();
-        }
     }
 
-#if net45
+    /// <summary>
-#else
+    /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-    public static Task<int> ExecutePgCopyAsync<T>(this IUpdate<T> that, CancellationToken cancellationToken = default) where T : class
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    /// <typeparam name="TKey"></typeparam>
+    /// <param name="select"></param>
+    /// <param name="timestamp">时间标识</param>
+    /// <param name="partition">最新项的列</param>
+    /// <returns></returns>
+    public static ISelect<T1, T2, T3> LatestOn<T1, T2, T3, TKey>(this ISelect<T1, T2, T3> select,
+        Expression<Func<T1, DateTime?>> timestamp,
+        Expression<Func<T1, TKey>> partition) where T2 : class where T3 : class
     {
-        var update = that as UpdateProvider<T>;
+        Provider(timestamp, partition);
-        if (update._source.Any() != true || update._tempPrimarys.Any() == false) return Task.FromResult(0);
+        return select;
-        var state = ExecutePgCopyState(update);
-        return UpdateProvider.ExecuteBulkUpdateAsync(update, state, insert => insert.ExecutePgCopyAsync(cancellationToken));
     }
-    async public static Task ExecutePgCopyAsync<T>(this IInsert<T> that, CancellationToken cancellationToken = default) where T : class
+
+    /// <summary>
+    /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
+    /// </summary>
+    /// <typeparam name="T"></typeparam>
+    /// <typeparam name="TKey"></typeparam>
+    /// <param name="select"></param>
+    /// <param name="timestamp">时间标识</param>
+    /// <param name="partition">最新项的列</param>
+    /// <returns></returns>
+    public static ISelect<T1, T2, T3, T4> LatestOn<T1, T2, T3, T4, TKey>(this ISelect<T1, T2, T3, T4> select,
+        Expression<Func<T1, DateTime?>> timestamp,
+        Expression<Func<T1, TKey>> partition) where T2 : class where T3 : class where T4 : class
     {
-        var insert = that as QuestDbInsert<T>;
+        Provider(timestamp, partition);
-        if (insert == null) throw new Exception(CoreStrings.S_Features_Unique("ExecutePgCopyAsync", "PostgreSQL"));
+        return select;
-
+    }
-        var dt = that.ToDataTable();
+}
-        if (dt.Rows.Count == 0) return;
+
-        Func<NpgsqlConnection, Task> binaryImportAsync = async conn =>
+public static class RestAPIExtension
-        {
+{
-            var copyFromCommand = new StringBuilder().Append("COPY ").Append(insert.InternalCommonUtils.QuoteSqlName(dt.TableName)).Append("(");
+    internal static string BaseUrl = string.Empty;
-            var colIndex = 0;
+    internal static string authorization = string.Empty;
-            foreach (DataColumn col in dt.Columns)
+
-            {
+    internal static async Task<string> ExecAsync(string sql)
-                if (colIndex++ > 0) copyFromCommand.Append(", ");
+    {
-                copyFromCommand.Append(insert.InternalCommonUtils.QuoteSqlName(col.ColumnName));
+        var result = string.Empty;
-            }
+        var client = QuestDbContainer.GetService<IHttpClientFactory>().CreateClient();
-            copyFromCommand.Append(") FROM STDIN BINARY");
+        var url = $"{BaseUrl}/exec?query={HttpUtility.UrlEncode(sql)}";
-            using (var writer = conn.BeginBinaryImport(copyFromCommand.ToString()))
+        if (!string.IsNullOrWhiteSpace(authorization))
-            {
+            client.DefaultRequestHeaders.Add("Authorization", authorization);
-                foreach (DataRow item in dt.Rows)
+        var httpResponseMessage = await client.GetAsync(url);
-                    await writer.WriteRowAsync(cancellationToken, item.ItemArray);
+        result = await httpResponseMessage.Content.ReadAsStringAsync();
-                writer.Complete();
+
-            }
+        return result;
-            copyFromCommand.Clear();
+    }
-        };
+
-
+    public static FreeSqlBuilder UseQuestDbRestAPI(this FreeSqlBuilder buider, string host, string username = "",
-        try
+        string password = "")
-        {
+    {
-            if (insert.InternalConnection == null && insert.InternalTransaction == null)
+        BaseUrl = host;
-            {
+        if (BaseUrl.EndsWith("/"))
-                using (var conn = await insert.InternalOrm.Ado.MasterPool.GetAsync())
+            BaseUrl = BaseUrl.Remove(BaseUrl.Length - 1);
-                {
+
-                    await binaryImportAsync(conn.Value as NpgsqlConnection);
+        if (!BaseUrl.ToLower().StartsWith("http"))
-                }
+            BaseUrl = $"http://{BaseUrl}";
-            }
+        if (!string.IsNullOrWhiteSpace(username) && !string.IsNullOrWhiteSpace(password))
-            else if (insert.InternalTransaction != null)
+        {
-            {
+            var base64 = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{username}:{password}"));
-                await binaryImportAsync(insert.InternalTransaction.Connection as NpgsqlConnection);
+            authorization = $"Basic {base64}";
-            }
+        }
-            else if (insert.InternalConnection != null)
+        buider.UseNoneCommandParameter(true);
-            {
+        return buider;
-                var conn = insert.InternalConnection as NpgsqlConnection;
-                var isNotOpen = false;
-                if (conn.State != System.Data.ConnectionState.Open)
-                {
-                    isNotOpen = true;
-                    await conn.OpenAsync(cancellationToken);
-                }
-                try
-                {
-                    await binaryImportAsync(conn);
-                }
-                finally
-                {
-                    if (isNotOpen)
-                        await conn.CloseAsync();
-                }
-            }
-            else
-            {
-                throw new NotImplementedException($"ExecutePgCopyAsync {CoreStrings.S_Not_Implemented_FeedBack}");
-            }
-        }
-        finally
-        {
-            dt.Clear();
-        }
     }
-#endif
-    #endregion
 }

Providers/FreeSql.Provider.QuestDb/QuestDbProvider.cs

@@ -1,6 +1,7 @@
 using FreeSql.Internal;
 using FreeSql.Internal.CommonProvider;
 using FreeSql.PostgreSQL.Curd;
+using FreeSql.Provider.QuestDb;
 using FreeSql.QuestDb.Curd;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
@@ -16,6 +17,7 @@ using System.Net.NetworkInformation;
 using System.Numerics;
 using System.Reflection;
 using System.Threading;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace FreeSql.QuestDb
 {
@@ -134,6 +136,11 @@ namespace FreeSql.QuestDb
 
             Select0Provider._dicMethodDataReaderGetValue[typeof(Guid)] =
                 typeof(DbDataReader).GetMethod("GetGuid", new Type[] { typeof(int) });
+
+            QuestDbContainer.Initialize(service =>
+            {
+                service.AddHttpClient();
+            });
         }
 
         public override ISelect<T1> CreateSelectProvider<T1>(object dywhere) =>

Providers/FreeSql.Provider.QuestDb/QuestDbUtils.cs

@@ -14,10 +14,16 @@ using System.Globalization;
 using System.Linq;
 using System.Linq.Expressions;
 using System.Net;
+using System.Net.Http;
 using System.Numerics;
 using System.Text;
 using System.Threading;
 using FreeSql.QuestDb;
+using System.Web;
+using System.Threading.Tasks;
+using Newtonsoft.Json;
+using System.IO;
+using CsvHelper;
 
 namespace FreeSql.QuestDb
 {
@@ -382,141 +388,4 @@ namespace FreeSql
         /// </summary>
         M
     }
-
-    public static class SampleByExtension
-    {
-        //是否使用该方法
-        internal static AsyncLocal<bool> IsExistence = new AsyncLocal<bool>()
-        {
-            Value = false
-        };
-
-        internal static AsyncLocal<string> SamoleByString = new AsyncLocal<string>()
-        {
-            Value = string.Empty
-        };
-
-        public static void Initialize()
-        {
-            IsExistence.Value = false;
-            SamoleByString.Value = string.Empty;
-        }
-
-        /// <summary>
-        /// SAMPLE BY用于时间序列数据，将大型数据集汇总为同质时间块的聚合，作为SELECT语句的一部分。对缺少数据的数据集执行查询的用户可以使用FILL关键字指定填充行为
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <param name="select"></param>
-        /// <param name="time">时长</param>
-        /// <param name="unit">单位</param>
-        /// <returns></returns>
-        public static ISelect<T> SampleBy<T>(this ISelect<T> select, double time, SampleUnits unit)
-        {
-            var _unit = Enum.GetName(typeof(SampleUnits), unit);
-            IsExistence.Value = true;
-            var samoleByTemple = $"{Environment.NewLine}SAMPLE BY {{0}}{{1}}{Environment.NewLine}";
-            SamoleByString.Value = string.Format(samoleByTemple, time.ToString(), _unit);
-            return select;
-        }
-    }
-
-    public static class LatestOnExtension
-    {
-        //是否使用该方法
-        internal static AsyncLocal<bool> IsExistence = new AsyncLocal<bool>()
-        {
-            Value = false
-        };
-
-        internal static AsyncLocal<string> LatestOnString = new AsyncLocal<string>()
-        {
-            Value = string.Empty
-        };
-
-        public static void Initialize()
-        {
-            IsExistence.Value = false;
-            LatestOnString.Value = string.Empty;
-        }
-
-        /// <summary>
-        /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-        /// </summary>
-        /// <typeparam name="T1"></typeparam>
-        /// <typeparam name="TKey"></typeparam>
-        /// <param name="select"></param>
-        /// <param name="timestamp">时间标识</param>
-        /// <param name="partition">最新项的列</param>
-        /// <returns></returns>
-        public static ISelect<T1> LatestOn<T1, TKey>(this ISelect<T1> select, Expression<Func<T1, DateTime?>> timestamp,
-            Expression<Func<T1, TKey>> partition)
-        {
-            Provider(timestamp, partition);
-            return select;
-        }
-
-        private static void Provider<T1, TKey>(Expression<Func<T1, DateTime?>> timestamp,
-            Expression<Func<T1, TKey>> partition)
-        {
-            IsExistence.Value = true;
-            var latestOnTemple = $"{Environment.NewLine}LATEST ON {{0}} PARTITION BY {{1}} ";
-            var expressionVisitor = new QuestDbExpressionVisitor();
-            expressionVisitor.Visit(timestamp);
-            var _timestamp = expressionVisitor.Fields();
-            expressionVisitor.Visit(partition);
-            var _partition = expressionVisitor.Fields();
-            LatestOnString.Value = string.Format(latestOnTemple, _timestamp, _partition);
-        }
-
-        /// <summary>
-        /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <typeparam name="TKey"></typeparam>
-        /// <param name="select"></param>
-        /// <param name="timestamp">时间标识</param>
-        /// <param name="partition">最新项的列</param>
-        /// <returns></returns>
-        public static ISelect<T1, T2> LatestOn<T1, T2, TKey>(this ISelect<T1, T2> select,
-            Expression<Func<T1, DateTime?>> timestamp,
-            Expression<Func<T1, TKey>> partition) where T2 : class
-        {
-            Provider(timestamp, partition);
-            return select;
-        }
-
-        /// <summary>
-        /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <typeparam name="TKey"></typeparam>
-        /// <param name="select"></param>
-        /// <param name="timestamp">时间标识</param>
-        /// <param name="partition">最新项的列</param>
-        /// <returns></returns>
-        public static ISelect<T1, T2, T3> LatestOn<T1, T2, T3, TKey>(this ISelect<T1, T2, T3> select,
-            Expression<Func<T1, DateTime?>> timestamp,
-            Expression<Func<T1, TKey>> partition) where T2 : class where T3 : class
-        {
-            Provider(timestamp, partition);
-            return select;
-        }
-
-        /// <summary>
-        /// 对于多个时间序列存储在同一个表中的场景，根据时间戳检索给定键或键组合的最新项。
-        /// </summary>
-        /// <typeparam name="T"></typeparam>
-        /// <typeparam name="TKey"></typeparam>
-        /// <param name="select"></param>
-        /// <param name="timestamp">时间标识</param>
-        /// <param name="partition">最新项的列</param>
-        /// <returns></returns>
-        public static ISelect<T1, T2, T3, T4> LatestOn<T1, T2, T3, T4, TKey>(this ISelect<T1, T2, T3, T4> select,
-            Expression<Func<T1, DateTime?>> timestamp,
-            Expression<Func<T1, TKey>> partition) where T2 : class where T3 : class where T4 : class
-        {
-            Provider(timestamp, partition);
-            return select;
-        }
-    }
 }