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

- 增加 IInsert.ExecutePgCopy 扩展方法执行 PostgreSQL Copy 批量导入，在 FreeSql.Provider.PostgreSQL 可用；

Browse Source
This commit is contained in:
28810 2019-12-21 16:44:31 +08:00
parent c335eab82c
commit 834bdea11f
4 changed files with 137 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
FreeSql/Internal/CommonProvider/InsertProvider.cs
View File

@ -488,6 +488,7 @@ namespace FreeSql.Internal.CommonProvider
if (col.Attribute.IsIdentity == false && _ignore.ContainsKey(col.Attribute.Name)) continue;
dt.Columns.Add(col.Attribute.Name, col.Attribute.MapType);
}
if (dt.Columns.Count == 0) return dt;
foreach (var d in _source)
{
var row = new object[dt.Columns.Count];

90
Providers/FreeSql.Provider.PostgreSQL/PostgreSQLExtensions.cs
View File

@ -1,7 +1,10 @@
using FreeSql;
using FreeSql.PostgreSQL.Curd;
using Npgsql;
using System;
using System.Data;
using System.Linq.Expressions;
using System.Text;
public static partial class FreeSqlPostgreSQLGlobalExtensions
{
@ -24,4 +27,91 @@ public static partial class FreeSqlPostgreSQLGlobalExtensions
/// <param name="columns">默认是以主键作为重复判断，也可以指定其他列：a => a.Name | a => new{a.Name,a.Time} | a => new[]{"name","time"}</param>
/// <returns></returns>
public static OnConflictDoUpdate<T1> OnConflictDoUpdate<T1>(this IInsert<T1> that, Expression<Func<T1, object>> columns = null) where T1 : class => new FreeSql.PostgreSQL.Curd.OnConflictDoUpdate<T1>(that.InsertIdentity(), columns);
/// <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>
public static void ExecutePgCopy<T>(this IInsert<T> that) where T : class
{
var insert = that as FreeSql.PostgreSQL.Curd.PostgreSQLInsert<T>;
if (insert == null) throw new Exception("ExecuteSqlBulkCopy 是 FreeSql.Provider.PostgreSQL 特有的功能");
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("ExecuteCopy 未实现错误，请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
}

5
Providers/FreeSql.Provider.SqlServer/Curd/SqlServerInsert.cs
View File

@ -4,7 +4,6 @@ using System;
using System.Collections.Generic;
using System.Data;
using System.Data.Common;
using System.Data.SqlClient;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
@ -20,8 +19,8 @@ namespace FreeSql.SqlServer.Curd
}
internal IFreeSql InternalOrm => _orm as IFreeSql;
internal SqlConnection InternalConnection => _connection as SqlConnection;
internal SqlTransaction InternalTransaction => _transaction as SqlTransaction;
internal DbConnection InternalConnection => _connection;
internal DbTransaction InternalTransaction => _transaction;
public override int ExecuteAffrows() => base.SplitExecuteAffrows(_batchValuesLimit > 0 ? _batchValuesLimit : 1000, _batchParameterLimit > 0 ? _batchParameterLimit : 2100);
public override long ExecuteIdentity() => base.SplitExecuteIdentity(_batchValuesLimit > 0 ? _batchValuesLimit : 1000, _batchParameterLimit > 0 ? _batchParameterLimit : 2100);

25
Providers/FreeSql.Provider.SqlServer/SqlServerExtensions.cs
View File

@ -78,6 +78,8 @@ public static partial class FreeSqlSqlServerGlobalExtensions
bulkCopy.WriteToServer(dt);
};
try
{
if (insert.InternalConnection == null && insert.InternalTransaction == null)
{
using (var conn = insert.InternalOrm.Ado.MasterPool.Get())
@ -93,35 +95,44 @@ public static partial class FreeSqlSqlServerGlobalExtensions
else if (insert.InternalTransaction != null)
{
using (var bulkCopy = copyOptions == SqlBulkCopyOptions.Default ?
new SqlBulkCopy(insert.InternalTransaction.Connection) :
new SqlBulkCopy(insert.InternalTransaction.Connection, copyOptions, insert.InternalTransaction))
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection) :
new SqlBulkCopy(insert.InternalTransaction.Connection as SqlConnection, copyOptions, insert.InternalTransaction as SqlTransaction))
{
writeToServer(bulkCopy);
}
}
else if (insert.InternalConnection != null)
{
var conn = insert.InternalConnection;
var conn = insert.InternalConnection as SqlConnection;
var isNotOpen = false;
if (conn.State != System.Data.ConnectionState.Open)
{
isNotOpen = true;
conn.Open();
}
try
{
using (var bulkCopy = copyOptions == SqlBulkCopyOptions.Default ?
new SqlBulkCopy(insert.InternalConnection) :
new SqlBulkCopy(insert.InternalConnection, copyOptions, null))
new SqlBulkCopy(conn) :
new SqlBulkCopy(conn, copyOptions, null))
{
writeToServer(bulkCopy);
}
if (isNotOpen)
}
finally
{
if (isNotOpen)
conn.Close();
}
}
else
{
throw new NotImplementedException("未实现错误，请反馈给作者");
throw new NotImplementedException("ExecuteSqlBulkCopy 未实现错误，请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
}