using FreeSql;
using FreeSql.PostgreSQL.Curd;
using Npgsql;
using System;
using System.Data;
using System.Linq.Expressions;
using System.Text;
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("ExecutePgCopy 是 FreeSql.Provider.PostgreSQL 特有的功能");
var dt = that.InsertIdentity().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 未实现错误,请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
#if net45
#else
async public static Task ExecutePgCopyAsync(this IInsert that) where T : class
{
var insert = that as FreeSql.PostgreSQL.Curd.PostgreSQLInsert;
if (insert == null) throw new Exception("ExecutePgCopyAsync 是 FreeSql.Provider.PostgreSQL 特有的功能");
var dt = that.InsertIdentity().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(System.Threading.CancellationToken.None, 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();
}
try
{
await binaryImportAsync(conn);
}
finally
{
if (isNotOpen)
await conn.CloseAsync();
}
}
else
{
throw new NotImplementedException("ExecutePgCopyAsync 未实现错误,请反馈给作者");
}
}
finally
{
dt.Clear();
}
}
#endif
#endregion
}