using FreeSql; using FreeSql.PostgreSQL.Curd; using Npgsql; using System; using System.Data; using System.Diagnostics; using System.Linq.Expressions; using System.Text; using System.Threading; using System.Threading.Tasks; public static partial class FreeSqlPostgreSQLGlobalExtensions { /// /// 特殊处理类似 string.Format 的使用方法,防止注入,以及 IS NULL 转换 /// /// /// /// public static string FormatPostgreSQL(this string that, params object[] args) => _postgresqlAdo.Addslashes(that, args); static FreeSql.PostgreSQL.PostgreSQLAdo _postgresqlAdo = new FreeSql.PostgreSQL.PostgreSQLAdo(); /// /// PostgreSQL9.5+ 特有的功能,On Conflict Do Update /// 注意:此功能会开启插入【自增列】 /// /// /// /// 默认是以主键作为重复判断,也可以指定其他列:a => a.Name | a => new{a.Name,a.Time} | a => new[]{"name","time"} /// public static OnConflictDoUpdate OnConflictDoUpdate(this IInsert that, Expression> columns = null) where T1 : class => new FreeSql.PostgreSQL.Curd.OnConflictDoUpdate(that.InsertIdentity(), columns); #region ExecutePgCopy /// /// PostgreSQL COPY 批量导入功能,封装了 NpgsqlConnection.BeginBinaryImport 方法 /// 使用 IgnoreColumns/InsertColumns 设置忽略/指定导入的列 /// 使用 WithConnection/WithTransaction 传入连接/事务对象 /// 提示:若本方法不能满足,请使用 IInsert<T>.ToDataTable 方法得到 DataTable 对象后,自行处理。 /// COPY 与 insert into t values(..),(..),(..) 性能测试参考: /// 插入180000行,52列:10,090ms 与 46,756ms,10列:4,081ms 与 9,786ms /// 插入10000行,52列:583ms 与 3,294ms,10列:167ms 与 568ms /// 插入5000行,52列:337ms 与 2,269ms,10列:93ms 与 366ms /// 插入2000行,52列:136ms 与 1,019ms,10列:39ms 与 157ms /// 插入1000行,52列:88ms 与 374ms,10列:21ms 与 102ms /// 插入500行,52列:61ms 与 209ms,10列:12ms 与 34ms /// 插入100行,52列:30ms 与 51ms,10列:4ms 与 9ms /// 插入50行,52列:25ms 与 37ms,10列:2ms 与 6ms /// /// /// public static void ExecutePgCopy(this IInsert that) where T : class { var insert = that as FreeSql.PostgreSQL.Curd.PostgreSQLInsert; if (insert == null) throw new Exception(CoreStrings.S_Features_Unique("ExecutePgCopy", "PostgreSQL")); var dt = that.ToDataTable(); if (dt.Rows.Count == 0) return; Action 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 #else async public static Task ExecutePgCopyAsync(this IInsert that, CancellationToken cancellationToken = default) where T : class { var insert = that as FreeSql.PostgreSQL.Curd.PostgreSQLInsert; if (insert == null) throw new Exception(CoreStrings.S_Features_Unique("ExecutePgCopyAsync", "PostgreSQL")); var dt = that.ToDataTable(); if (dt.Rows.Count == 0) return; Func binaryImportAsync = async 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) await writer.WriteRowAsync(cancellationToken, item.ItemArray); writer.Complete(); } copyFromCommand.Clear(); }; try { if (insert.InternalConnection == null && insert.InternalTransaction == null) { using (var conn = await insert.InternalOrm.Ado.MasterPool.GetAsync()) { await binaryImportAsync(conn.Value as NpgsqlConnection); } } else if (insert.InternalTransaction != null) { await binaryImportAsync(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; 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 }