tk/FreeSql
1
0
Fork 0
mirror of https://github.com/nsnail/FreeSql.git synced 2025-06-18 20:08:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

重写批量修改，去除插入和修改事务，修改插入数据的限制默认值为int.MaxValue,自增字段直接返回0

Browse Source
This commit is contained in:
chenbo
2021-11-30 15:02:21 +08:00
parent 846c180191
commit ae0f2e9d0d
14 changed files with 869 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

236
Providers/FreeSql.Provider.ClickHouse/Curd/ClickHouseInsert.cs
View File

@ -6,6 +6,7 @@ using System;
using System.Collections.Generic;
using System.Data;
using System.Data.Common;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
@ -33,9 +34,9 @@ namespace FreeSql.ClickHouse.Curd
internal Dictionary<string, bool> InternalIgnore => _ignore;
internal void InternalClearData() => ClearData();
public override int ExecuteAffrows() => base.SplitExecuteAffrows(int.MaxValue, int.MaxValue);
public override long ExecuteIdentity() => base.SplitExecuteIdentity(int.MaxValue, int.MaxValue);
public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(int.MaxValue, int.MaxValue);
public override int ExecuteAffrows() => SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchParameterLimit > 0 ? _batchParameterLimit : int.MaxValue);
public override long ExecuteIdentity() => SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchParameterLimit > 0 ? _batchParameterLimit : int.MaxValue);
public override List<T1> ExecuteInserted() => SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchParameterLimit > 0 ? _batchParameterLimit : int.MaxValue);
public override string ToSql()
@ -54,6 +55,7 @@ namespace FreeSql.ClickHouse.Curd
{
try
{
Debug.WriteLine($"开始执行时间：{DateTime.Now}");
before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, null, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
using var bulkCopyInterface = new ClickHouseBulkCopy(_orm.Ado.MasterPool.Get().Value as ClickHouseConnection)
@ -61,7 +63,8 @@ namespace FreeSql.ClickHouse.Curd
DestinationTableName = _table.DbName,
BatchSize = _source.Count
};
bulkCopyInterface.WriteToServerAsync(ToDataTable(),default).Wait();
var data=ToDataTable();
bulkCopyInterface.WriteToServerAsync(data, default).Wait();
return affrows;
}
catch (Exception ex)
@ -82,7 +85,7 @@ namespace FreeSql.ClickHouse.Curd
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
try
{
affrows = _orm.Ado.ExecuteNonQuery(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params);
affrows = _orm.Ado.ExecuteNonQuery(_connection, null, CommandType.Text, sql, _commandTimeout, _params);
}
catch (Exception ex)
{
@ -122,28 +125,10 @@ namespace FreeSql.ClickHouse.Curd
protected override long RawExecuteIdentity()
{
var sql = this.ToSql();
if (string.IsNullOrEmpty(sql)) return 0;
sql = string.Concat(sql, "; SELECT LAST_INSERT_ID();");
var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
long ret = 0;
Exception exception = null;
try
{
ret = long.TryParse(string.Concat(_orm.Ado.ExecuteScalar(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params)), out var trylng) ? trylng : 0;
}
catch (Exception ex)
{
exception = ex;
throw;
}
finally
{
var after = new Aop.CurdAfterEventArgs(before, exception, ret);
_orm.Aop.CurdAfterHandler?.Invoke(this, after);
}
var identCols = _table.Columns.Where(a => a.Value.Attribute.IsIdentity == true);
if (identCols.Any()&&_source.Count==1)
ret = (long)identCols.First().Value.GetValue(_source.First());
return ret;
}
protected override List<T1> RawExecuteInserted()
@ -168,7 +153,7 @@ namespace FreeSql.ClickHouse.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, null, CommandType.Text, sql, _commandTimeout, _params);
}
catch (Exception ex)
{
@ -185,23 +170,65 @@ namespace FreeSql.ClickHouse.Curd
#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<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);
public override Task<int> ExecuteAffrowsAsync(CancellationToken cancellationToken = default) => base.SplitExecuteAffrowsAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken);
public override Task<long> ExecuteIdentityAsync(CancellationToken cancellationToken = default) => base.SplitExecuteIdentityAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken);
public override Task<List<T1>> ExecuteInsertedAsync(CancellationToken cancellationToken = default) => base.SplitExecuteInsertedAsync(_batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, _batchValuesLimit > 0 ? _batchValuesLimit : int.MaxValue, cancellationToken);
async protected override Task<long> RawExecuteIdentityAsync(CancellationToken cancellationToken = default)
{
var sql = this.ToSql();
if (string.IsNullOrEmpty(sql)) return 0;
//var sql = this.ToSql();
//if (string.IsNullOrEmpty(sql)) return 0;
sql = string.Concat(sql, "; SELECT LAST_INSERT_ID();");
var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
//sql = string.Concat(sql, "; SELECT LAST_INSERT_ID();");
//var before = new Aop.CurdBeforeEventArgs(_table.Type, _table, Aop.CurdType.Insert, sql, _params);
//_orm.Aop.CurdBeforeHandler?.Invoke(this, before);
long ret = 0;
//Exception exception = null;
//try
//{
// ret = long.TryParse(string.Concat(await _orm.Ado.ExecuteScalarAsync(_connection, null, CommandType.Text, sql, _commandTimeout, _params, cancellationToken)), out var trylng) ? trylng : 0;
//}
//catch (Exception ex)
//{
// exception = ex;
// throw;
//}
//finally
//{
// var after = new Aop.CurdAfterEventArgs(before, exception, ret);
// _orm.Aop.CurdAfterHandler?.Invoke(this, after);
//}
return await Task.FromResult(ret);
}
protected override int SplitExecuteAffrows(int valuesLimit, int parameterLimit)
{
var ss = SplitSource(valuesLimit, parameterLimit);
var ret = 0;
if (ss.Any() == false)
{
ClearData();
return ret;
}
if (ss.Length == 1)
{
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, 1, 1));
ret = this.RawExecuteAffrows();
ClearData();
return ret;
}
var before = new Aop.TraceBeforeEventArgs("SplitExecuteAffrows", null);
_orm.Aop.TraceBeforeHandler?.Invoke(this, before);
Exception exception = null;
try
{
ret = long.TryParse(string.Concat(await _orm.Ado.ExecuteScalarAsync(_connection, _transaction, CommandType.Text, sql, _commandTimeout, _params, cancellationToken)), out var trylng) ? trylng : 0;
for (var a = 0; a < ss.Length; a++)
{
_source = ss[a];
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, a + 1, ss.Length));
ret += this.RawExecuteAffrows();
}
}
catch (Exception ex)
{
@ -210,11 +237,142 @@ namespace FreeSql.ClickHouse.Curd
}
finally
{
var after = new Aop.CurdAfterEventArgs(before, exception, ret);
_orm.Aop.CurdAfterHandler?.Invoke(this, after);
var after = new Aop.TraceAfterEventArgs(before, null, exception);
_orm.Aop.TraceAfterHandler?.Invoke(this, after);
}
ClearData();
return ret;
}
async protected override Task<int> SplitExecuteAffrowsAsync(int valuesLimit, int parameterLimit, CancellationToken cancellationToken = default)
{
var ss = SplitSource(valuesLimit, parameterLimit);
var ret = 0;
if (ss.Any() == false)
{
ClearData();
return ret;
}
if (ss.Length == 1)
{
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, 1, 1));
ret = await this.RawExecuteAffrowsAsync(cancellationToken);
ClearData();
return ret;
}
var before = new Aop.TraceBeforeEventArgs("SplitExecuteAffrowsAsync", null);
_orm.Aop.TraceBeforeHandler?.Invoke(this, before);
Exception exception = null;
try
{
for (var a = 0; a < ss.Length; a++)
{
_source = ss[a];
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, a + 1, ss.Length));
ret += await this.RawExecuteAffrowsAsync(cancellationToken);
}
}
catch (Exception ex)
{
exception = ex;
throw;
}
finally
{
var after = new Aop.TraceAfterEventArgs(before, null, exception);
_orm.Aop.TraceAfterHandler?.Invoke(this, after);
}
ClearData();
return ret;
}
async protected override Task<long> SplitExecuteIdentityAsync(int valuesLimit, int parameterLimit, CancellationToken cancellationToken = default)
{
var ss = SplitSource(valuesLimit, parameterLimit);
long ret = 0;
if (ss.Any() == false)
{
ClearData();
return ret;
}
if (ss.Length == 1)
{
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, 1, 1));
ret = await this.RawExecuteIdentityAsync(cancellationToken);
ClearData();
return ret;
}
var before = new Aop.TraceBeforeEventArgs("SplitExecuteIdentityAsync", null);
_orm.Aop.TraceBeforeHandler?.Invoke(this, before);
Exception exception = null;
try
{
for (var a = 0; a < ss.Length; a++)
{
_source = ss[a];
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, a + 1, ss.Length));
if (a < ss.Length - 1) await this.RawExecuteAffrowsAsync(cancellationToken);
else ret = await this.RawExecuteIdentityAsync(cancellationToken);
}
}
catch (Exception ex)
{
exception = ex;
throw;
}
finally
{
var after = new Aop.TraceAfterEventArgs(before, null, exception);
_orm.Aop.TraceAfterHandler?.Invoke(this, after);
}
ClearData();
return ret;
}
async protected override Task<List<T1>> SplitExecuteInsertedAsync(int valuesLimit, int parameterLimit, CancellationToken cancellationToken = default)
{
var ss = SplitSource(valuesLimit, parameterLimit);
var ret = new List<T1>();
if (ss.Any() == false)
{
ClearData();
return ret;
}
if (ss.Length == 1)
{
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, 1, 1));
ret = await this.RawExecuteInsertedAsync(cancellationToken);
ClearData();
return ret;
}
var before = new Aop.TraceBeforeEventArgs("SplitExecuteInsertedAsync", null);
_orm.Aop.TraceBeforeHandler?.Invoke(this, before);
Exception exception = null;
try
{
for (var a = 0; a < ss.Length; a++)
{
_source = ss[a];
_batchProgress?.Invoke(new BatchProgressStatus<T1>(_source, a + 1, ss.Length));
ret.AddRange(await this.RawExecuteInsertedAsync(cancellationToken));
}
}
catch (Exception ex)
{
exception = ex;
throw;
}
finally
{
var after = new Aop.TraceAfterEventArgs(before, null, exception);
_orm.Aop.TraceAfterHandler?.Invoke(this, after);
}
ClearData();
return ret;
}
async protected override Task<List<T1>> RawExecuteInsertedAsync(CancellationToken cancellationToken = default)
{
var sql = this.ToSql();
@ -237,7 +395,7 @@ namespace FreeSql.ClickHouse.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, null, CommandType.Text, sql, _commandTimeout, _params, cancellationToken);
}
catch (Exception ex)
{