- 增加 Oracle/达梦 BulkCopy 支持;

This commit is contained in:
2881099 2020-11-18 09:42:09 +08:00
parent 2947572f05
commit d3fd022000
10 changed files with 239 additions and 12 deletions

View File

@ -502,5 +502,14 @@
<param name="that"></param> <param name="that"></param>
<returns></returns> <returns></returns>
</member> </member>
<member name="M:Microsoft.Extensions.DependencyInjection.FreeSqlRepositoryDependencyInjection.AddFreeRepository(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Action{FreeSql.FluentDataFilter},System.Reflection.Assembly[])">
<summary>
批量注入 Repository可以参考代码自行调整
</summary>
<param name="services"></param>
<param name="globalDataFilter"></param>
<param name="assemblies"></param>
<returns></returns>
</member>
</members> </members>
</doc> </doc>

View File

@ -200,6 +200,16 @@ INTO ""TB_TOPIC_INSERT""(""CLICKS"") VALUES(900)
//var items2 = insert.AppendData(items).ExecuteInserted(); //var items2 = insert.AppendData(items).ExecuteInserted();
} }
//[Fact]
//public void ExecuteDmBulkCopy()
//{
// var items = new List<Topic>();
// for (var a = 0; a < 10; a++) items.Add(new Topic { Id = a + 1, Title = $"newtitle{a}", Clicks = a * 100, CreateTime = DateTime.Now });
// insert.AppendData(items).InsertIdentity().ExecuteDmBulkCopy();
// //Dm.DmException:¡°The fastloading dll not loading!¡±
//}
[Fact] [Fact]
public void AsTable() public void AsTable()
{ {

View File

@ -200,6 +200,26 @@ INTO ""TB_TOPIC_INSERT""(""CLICKS"") VALUES(:Clicks_9)
//var items2 = insert.AppendData(items).ExecuteInserted(); //var items2 = insert.AppendData(items).ExecuteInserted();
} }
[Fact]
public void ExecuteOracleBulkCopy()
{
var items = new List<Topic_bulkcopy>();
for (var a = 0; a < 10; a++) items.Add(new Topic_bulkcopy { Title = $"newtitle{a}", Clicks = a * 100, CreateTime = DateTime.Now });
g.oracle.Insert<Topic_bulkcopy>().AppendData(items).InsertIdentity().ExecuteOracleBulkCopy();
//insert.AppendData(items).IgnoreColumns(a => new { a.CreateTime, a.Clicks }).ExecuteSqlBulkCopy();
// System.NotSupportedException:“DataSet does not support System.Nullable<>.”
}
[Table(Name = "tb_topic_bulkcopy")]
class Topic_bulkcopy
{
public Guid Id { get; set; }
public int? Clicks { get; set; }
public TestTypeInfo Type { get; set; }
public string Title { get; set; }
public DateTime CreateTime { get; set; }
}
[Fact] [Fact]
public void AsTable() public void AsTable()
{ {

View File

@ -19,6 +19,10 @@ namespace FreeSql.Dameng.Curd
{ {
} }
internal IFreeSql InternalOrm => _orm as IFreeSql;
internal DbConnection InternalConnection => _connection;
internal DbTransaction InternalTransaction => _transaction;
public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);
public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);
public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);

View File

@ -1,4 +1,8 @@
public static partial class FreeSqlDamengGlobalExtensions using Dm;
using FreeSql;
using System;
public static partial class FreeSqlDamengGlobalExtensions
{ {
/// <summary> /// <summary>
@ -9,4 +13,91 @@
/// <returns></returns> /// <returns></returns>
public static string FormatDameng(this string that, params object[] args) => _damengAdo.Addslashes(that, args); public static string FormatDameng(this string that, params object[] args) => _damengAdo.Addslashes(that, args);
static FreeSql.Dameng.DamengAdo _damengAdo = new FreeSql.Dameng.DamengAdo(); static FreeSql.Dameng.DamengAdo _damengAdo = new FreeSql.Dameng.DamengAdo();
#region ExecuteDmBulkCopy
/// <summary>
/// 达梦 CopyBulk 批量插入功能<para></para>
/// 使用 IgnoreColumns/InsertColumns 设置忽略/指定导入的列<para></para>
/// 使用 WithConnection/WithTransaction 传入连接/事务对象<para></para>
/// 提示:若本方法不能满足,请使用 IInsert&lt;T&gt;.ToDataTable 方法得到 DataTable 对象后,自行处理。
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="that"></param>
/// <param name="copyOptions"></param>
/// <param name="batchSize"></param>
/// <param name="bulkCopyTimeout"></param>
public static void ExecuteDmBulkCopy<T>(this IInsert<T> that, DmBulkCopyOptions copyOptions = DmBulkCopyOptions.Default, int? batchSize = null, int? bulkCopyTimeout = null) where T : class
{
var insert = that as FreeSql.Dameng.Curd.DamengInsert<T>;
if (insert == null) throw new Exception("ExecuteDmBulkCopy 是 FreeSql.Provider.Dameng 特有的功能");
var dt = that.ToDataTable();
if (dt.Rows.Count == 0) return;
Action<DmBulkCopy> writeToServer = bulkCopy =>
{
if (batchSize.HasValue) bulkCopy.BatchSize = batchSize.Value;
if (bulkCopyTimeout.HasValue) bulkCopy.BulkCopyTimeout = bulkCopyTimeout.Value;
bulkCopy.DestinationTableName = dt.TableName;
for (int i = 0; i < dt.Columns.Count; i++)
bulkCopy.ColumnMappings.Add(dt.Columns[i].ColumnName, dt.Columns[i].ColumnName);
bulkCopy.WriteToServer(dt);
};
try
{
if (insert.InternalConnection == null && insert.InternalTransaction == null)
{
using (var conn = insert.InternalOrm.Ado.MasterPool.Get())
{
using (var bulkCopy = copyOptions == DmBulkCopyOptions.Default ?
new DmBulkCopy(conn.Value as DmConnection) :
new DmBulkCopy(conn.Value as DmConnection, copyOptions, insert.InternalTransaction as DmTransaction))
{
writeToServer(bulkCopy);
}
}
}
else if (insert.InternalTransaction != null)
{
using (var bulkCopy = new DmBulkCopy(insert.InternalTransaction.Connection as DmConnection, copyOptions, insert.InternalTransaction as DmTransaction))
{
writeToServer(bulkCopy);
}
}
else if (insert.InternalConnection != null)
{
var conn = insert.InternalConnection as DmConnection;
var isNotOpen = false;
if (conn.State != System.Data.ConnectionState.Open)
{
isNotOpen = true;
conn.Open();
}
try
{
using (var bulkCopy = copyOptions == DmBulkCopyOptions.Default ?
new DmBulkCopy(conn) :
new DmBulkCopy(conn, copyOptions, null))
{
writeToServer(bulkCopy);
}
}
finally
{
if (isNotOpen)
conn.Close();
}
}
else
{
throw new NotImplementedException("ExecuteDmBulkCopy 未实现错误,请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
#endregion
} }

View File

@ -57,7 +57,7 @@ public static class FreeSqlMySqlConnectorGlobalExtensions
} }
else if (insert.InternalTransaction != null) else if (insert.InternalTransaction != null)
{ {
writeToServer(new MySqlBulkCopy(insert.InternalTransaction.Connection as MySqlConnection)); writeToServer(new MySqlBulkCopy(insert.InternalTransaction.Connection as MySqlConnection, insert.InternalTransaction as MySqlTransaction));
} }
else if (insert.InternalConnection != null) else if (insert.InternalConnection != null)
{ {
@ -116,7 +116,7 @@ public static class FreeSqlMySqlConnectorGlobalExtensions
} }
else if (insert.InternalTransaction != null) else if (insert.InternalTransaction != null)
{ {
await writeToServer(new MySqlBulkCopy(insert.InternalTransaction.Connection as MySqlConnection)); await writeToServer(new MySqlBulkCopy(insert.InternalTransaction.Connection as MySqlConnection, insert.InternalTransaction as MySqlTransaction));
} }
else if (insert.InternalConnection != null) else if (insert.InternalConnection != null)
{ {

View File

@ -20,6 +20,10 @@ namespace FreeSql.Oracle.Curd
{ {
} }
internal IFreeSql InternalOrm => _orm as IFreeSql;
internal DbConnection InternalConnection => _connection;
internal DbTransaction InternalTransaction => _transaction;
public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);
public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);
public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999); public override List<T1> ExecuteInserted() => base.SplitExecuteInserted(_batchValuesLimit > 0 ? _batchValuesLimit : 500, _batchParameterLimit > 0 ? _batchParameterLimit : 999);

View File

@ -26,11 +26,11 @@
</ItemGroup> </ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'"> <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
<PackageReference Include="Oracle.ManagedDataAccess.Core" Version="2.19.91" /> <PackageReference Include="Oracle.ManagedDataAccess.Core" Version="2.19.100" />
</ItemGroup> </ItemGroup>
<ItemGroup Condition="'$(TargetFramework)' == 'net45' or '$(TargetFramework)' == 'net40'"> <ItemGroup Condition="'$(TargetFramework)' == 'net45' or '$(TargetFramework)' == 'net40'">
<PackageReference Include="Oracle.ManagedDataAccess" Version="19.9.0" /> <PackageReference Include="Oracle.ManagedDataAccess" Version="19.10.0" />
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>

View File

@ -1,4 +1,8 @@
public static partial class FreeSqlOracleGlobalExtensions using FreeSql;
using Oracle.ManagedDataAccess.Client;
using System;
public static partial class FreeSqlOracleGlobalExtensions
{ {
/// <summary> /// <summary>
@ -9,4 +13,93 @@
/// <returns></returns> /// <returns></returns>
public static string FormatOracle(this string that, params object[] args) => _oracleAdo.Addslashes(that, args); public static string FormatOracle(this string that, params object[] args) => _oracleAdo.Addslashes(that, args);
static FreeSql.Oracle.OracleAdo _oracleAdo = new FreeSql.Oracle.OracleAdo(); static FreeSql.Oracle.OracleAdo _oracleAdo = new FreeSql.Oracle.OracleAdo();
#region ExecuteOracleBulkCopy
/// <summary>
/// Oracle CopyBulk 批量插入功能<para></para>
/// 使用 IgnoreColumns/InsertColumns 设置忽略/指定导入的列<para></para>
/// 使用 WithConnection/WithTransaction 传入连接/事务对象<para></para>
/// 提示:若本方法不能满足,请使用 IInsert&lt;T&gt;.ToDataTable 方法得到 DataTable 对象后,自行处理。
/// </summary>
/// <typeparam name="T"></typeparam>
/// <param name="that"></param>
/// <param name="copyOptions"></param>
/// <param name="batchSize"></param>
/// <param name="bulkCopyTimeout"></param>
public static void ExecuteOracleBulkCopy<T>(this IInsert<T> that, OracleBulkCopyOptions copyOptions = OracleBulkCopyOptions.Default, int? batchSize = null, int? bulkCopyTimeout = null) where T : class
{
var insert = that as FreeSql.Oracle.Curd.OracleInsert<T>;
if (insert == null) throw new Exception("ExecuteOracleBulkCopy 是 FreeSql.Provider.Oracle 特有的功能");
var dt = that.ToDataTable();
if (dt.Rows.Count == 0) return;
Action<OracleBulkCopy> writeToServer = bulkCopy =>
{
if (batchSize.HasValue) bulkCopy.BatchSize = batchSize.Value;
if (bulkCopyTimeout.HasValue) bulkCopy.BulkCopyTimeout = bulkCopyTimeout.Value;
bulkCopy.DestinationTableName = dt.TableName;
for (int i = 0; i < dt.Columns.Count; i++)
bulkCopy.ColumnMappings.Add(dt.Columns[i].ColumnName, dt.Columns[i].ColumnName);
bulkCopy.WriteToServer(dt);
};
try
{
if (insert.InternalConnection == null && insert.InternalTransaction == null)
{
using (var conn = insert.InternalOrm.Ado.MasterPool.Get())
{
using (var bulkCopy = copyOptions == OracleBulkCopyOptions.Default ?
new OracleBulkCopy(conn.Value as OracleConnection) :
new OracleBulkCopy(conn.Value as OracleConnection, copyOptions))
{
writeToServer(bulkCopy);
}
}
}
else if (insert.InternalTransaction != null)
{
using (var bulkCopy = copyOptions == OracleBulkCopyOptions.Default ?
new OracleBulkCopy(insert.InternalTransaction.Connection as OracleConnection) :
new OracleBulkCopy(insert.InternalTransaction.Connection as OracleConnection, copyOptions))
{
writeToServer(bulkCopy);
}
}
else if (insert.InternalConnection != null)
{
var conn = insert.InternalConnection as OracleConnection;
var isNotOpen = false;
if (conn.State != System.Data.ConnectionState.Open)
{
isNotOpen = true;
conn.Open();
}
try
{
using (var bulkCopy = copyOptions == OracleBulkCopyOptions.Default ?
new OracleBulkCopy(conn) :
new OracleBulkCopy(conn, copyOptions))
{
writeToServer(bulkCopy);
}
}
finally
{
if (isNotOpen)
conn.Close();
}
}
else
{
throw new NotImplementedException("ExecuteOracleBulkCopy 未实现错误,请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
#endregion
} }

View File

@ -104,9 +104,7 @@ public static partial class FreeSqlSqlServerGlobalExtensions
} }
else if (insert.InternalTransaction != null) else if (insert.InternalTransaction != null)
{ {
using (var bulkCopy = copyOptions == SqlBulkCopyOptions.Default ? using (var bulkCopy = new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection, copyOptions, insert.InternalTransaction as SqlTransaction))
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection) :
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection, copyOptions, insert.InternalTransaction as SqlTransaction))
{ {
writeToServer(bulkCopy); writeToServer(bulkCopy);
} }
@ -181,9 +179,7 @@ public static partial class FreeSqlSqlServerGlobalExtensions
} }
else if (insert.InternalTransaction != null) else if (insert.InternalTransaction != null)
{ {
using (var bulkCopy = copyOptions == SqlBulkCopyOptions.Default ? using (var bulkCopy = new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection, copyOptions, insert.InternalTransaction as SqlTransaction))
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection) :
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection, copyOptions, insert.InternalTransaction as SqlTransaction))
{ {
await writeToServerAsync(bulkCopy); await writeToServerAsync(bulkCopy);
} }