npgsql：如何在一個以集合為引數的命令中使用npgsql選擇多行（具有多個列值）？

我在下面定義了兩個表，supplier_balances并且supplier_balance_items（順便說一句，兩者之間存在1[supplier_balance]:N[supplier_balance_items]關系）：

CREATE TABLE IF NOT EXISTS sch_brand_payment_data_lake_proxy.supplier_balances (
/* id is here for joining purposes with items table, instead of joining with the 4 columns used for sake
   of making sure a record is deemed as unique */
  id                             bigserial NOT NULL,
  accounting_document            text      NOT NULL,
  accounting_document_type       text      NOT NULL,
  company_code                   text      NOT NULL,
  document_date_year             int4      NOT NULL,
  accounting_doc_created_by_user text,
  accounting_clerk               text,
  assignment_reference           text,
  document_reference_id          text,
  original_reference_document    text,
  payment_terms                  text,
  supplier                       text,
  supplier_name                  text,
  document_date                  timestamp,
  posting_date                   timestamp,
  net_due_date                   timestamp,
  created_on                     timestamp default NULL,
  modified_on                    timestamp default NULL,
  pushed_on                      timestamp default NULL,
  is_modified bool GENERATED ALWAYS AS (modified_on IS NOT NULL AND modified_on > created_on) STORED,
  is_pushed   bool GENERATED ALWAYS AS (pushed_on   IS NOT NULL AND pushed_on > modified_on)  STORED,
  CONSTRAINT supplier_balances_pkey   PRIMARY KEY (id),
  /* accounting_document being the field of the composite unique index -> faster querying */
  CONSTRAINT supplier_balances_unique UNIQUE (
     accounting_document,
     accounting_document_type,
     company_code,
     document_date_year)
);

/* Creating other indexes for querying of those as well */
CREATE INDEX IF NOT EXISTS supplier_balances_accounting_document_type_idx
ON sch_brand_payment_data_lake_proxy.supplier_balances (accounting_document_type);
CREATE INDEX IF NOT EXISTS supplier_balances_company_code_idx
ON sch_brand_payment_data_lake_proxy.supplier_balances (company_code);
CREATE INDEX IF NOT EXISTS supplier_balances_document_date_year_idx
ON sch_brand_payment_data_lake_proxy.supplier_balances (document_date_year);

CREATE TABLE IF NOT EXISTS sch_brand_payment_data_lake_proxy.supplier_balance_items
(
    supplier_balance_id             bigserial NOT NULL,
    posting_view_item               text      NOT NULL,
    posting_key                     text,
    amount_in_company_code_currency numeric,
    amount_in_transaction_currency  numeric,
    cash_discount_1_percent         numeric,
    cash_discount_amount            numeric,
    clearing_accounting_document    text,
    document_item_text              text,
    gl_account                      text,
    is_cleared                      bool,
    clearing_date                   timestamp,
    due_calculation_base_date       timestamp,
    /* uniqueness is basically the posting_view_item for a given supplier balance */
    CONSTRAINT supplier_balance_items_pkey PRIMARY KEY (supplier_balance_id, posting_view_item),
    /* 1(supplier balance):N(supplier balance items) */
    CONSTRAINT supplier_balance_items_fkey FOREIGN KEY (supplier_balance_id)
               REFERENCES sch_brand_payment_data_lake_proxy.supplier_balances (id)
               ON DELETE CASCADE
               ON UPDATE CASCADE
);

NULL注意：為了簡單起見，我只是填寫不能填寫的列。

INSERT INTO 
sch_brand_payment_data_lake_proxy.supplier_balances 
(accounting_document, accounting_document_type, company_code, document_date_year)
VALUES 
('A', 'B', 'C', 0),
('A', 'B', 'C', 1),
('A', 'B', 'C', 2),
('A', 'B', 'C', 3),
('A', 'B', 'C', 4),
('A', 'B', 'C', 5)
RETURNING id;

輸出：

INSERT INTO 
sch_brand_payment_data_lake_proxy.supplier_balance_items 
(supplier_balance_id, posting_view_item)
VALUES 
(1, 'A'),
(1, 'B'),
(3, 'A'),
(3, 'B'),
(2, 'A'),
(1, 'C');

SELECT 
    accounting_document, 
    accounting_document_type, 
    company_code, 
    document_date_year
FROM sch_brand_payment_data_lake_proxy.supplier_balances;

輸出：

SELECT 
    supplier_balance_id,
    posting_view_item
FROM sch_brand_payment_data_lake_proxy.supplier_balance_items;

輸出：

現在，如果我們想在 JOIN 中選擇多個值，我們可以在原始 SQL 中執行：

SELECT 
    id,
    accounting_document, 
    accounting_document_type, 
    company_code, 
    document_date_year, 
    posting_view_item
FROM sch_brand_payment_data_lake_proxy.supplier_balances
LEFT OUTER JOIN sch_brand_payment_data_lake_proxy.supplier_balance_items
ON supplier_balances.id = supplier_balance_items.supplier_balance_id
WHERE (accounting_document, accounting_document_type, company_code, document_date_year)
IN  (('A', 'B', 'C', 1), ('A', 'B', 'C', 2))

輸出：

https://github.com/npgsql/npgsql/issues/1199

現在，當在 C# 中使用npgsql時，重現上面的查詢是一件容易的事：

using System.Data;

using Npgsql;

var connectionStringBuilder = new NpgsqlConnectionStringBuilder
{
    Host     = "localhost",
    Port     = 5432,
    Username = "brand_payment_migration",
    Password = "secret",
    Database = "brand_payment"
};
using var connection = new NpgsqlConnection(connectionStringBuilder.ToString());
connection.Open();
using var command = connection.CreateCommand();
command.CommandText = 
"SELECT id, accounting_document, accounting_document_type, company_code, document_date_year, posting_view_item "  
"FROM sch_brand_payment_data_lake_proxy.supplier_balances "  
"LEFT OUTER JOIN sch_brand_payment_data_lake_proxy.supplier_balance_items "  
"ON supplier_balances.id = supplier_balance_items.supplier_balance_id "  
"WHERE (accounting_document, accounting_document_type, company_code, document_date_year) "  
"IN (('A', 'B', 'C', 1), ('A', 'B', 'C', 2));";

using var reader = command.ExecuteReader();
using var dataTable = new DataTable();
dataTable.Load(reader);
var cols = dataTable.Columns.Cast<DataColumn>().ToArray();
Console.WriteLine(string.Join(Environment.NewLine, cols.Select((x, i) => $"Col{i} = {x}")));
Console.WriteLine(string.Join("\t", cols.Select((_, i) => $"Col{i}")));
foreach (var dataRow in dataTable.Rows.Cast<DataRow>())
{
    Console.WriteLine(string.Join("\t", dataRow.ItemArray));
}

其中，正如預期的輸出：

Col0 = id
Col1 = accounting_document
Col2 = accounting_document_type
Col3 = company_code
Col4 = document_date_year
Col5 = posting_view_item
Col0    Col1    Col2    Col3    Col4    Col5
2       A       B       C       1       A
3       A       B       C       2       A
3       A       B       C       2       B

現在，我想要實作的是，(('A', 'B', 'C', 1), ('A', 'B', 'C', 2));我不想將原始字串傳遞給 ，而是將 aNpgSqlParameter與值集的集合一起使用（即，對于每一列））。

所以我改變了上面的 C# 片段并添加了引數

// ...
"WHERE (accounting_document, accounting_document_type, company_code, document_date_year) "  
"IN @values;";
var parameter = command.CreateParameter();
parameter.ParameterName = "@values";
parameter.NpgsqlDbType = NpgsqlDbType.Array;
parameter.NpgsqlValue = new object[,]
{
    { "A", "B", "C", 1 }, 
    { "A", "B", "C", 2 }
};
// Note: the same kind of issue arises when using tuples, i.e.
// ( "A", "B", "C", 1 )
// ( "A", "B", "C", 2 )
command.Parameters.Add(parameter);
using var reader = command.ExecuteReader();
// ...

然后我得到了這個例外：

Unhandled exception. System.ArgumentOutOfRangeException: Cannot set NpgsqlDbType to just Array, Binary-Or with the element type (e.g. Array of Box is NpgsqlDbType.Array | Npg
sqlDbType.Box). (Parameter 'value')
   at Npgsql.NpgsqlParameter.set_NpgsqlDbType(NpgsqlDbType value)
   at Program.<Main>$(String[] args) in C:\Users\natalie-perret\Desktop\Personal\playground\csharp\CSharpPlayground\Program.cs:line 25

然后我嘗試使用以下方法解決該錯誤：

parameter.NpgsqlDbType = NpgsqlDbType.Array | NpgsqlDbType.Unknown;

但隨后又得到另一個例外：

Unhandled exception. System.ArgumentException: No array type could be found in the database for element .<unknown>
   at Npgsql.TypeMapping.ConnectorTypeMapper.ResolveByNpgsqlDbType(NpgsqlDbType npgsqlDbType)
   at Npgsql.NpgsqlParameter.ResolveHandler(ConnectorTypeMapper typeMapper)
   at Npgsql.NpgsqlParameterCollection.ValidateAndBind(ConnectorTypeMapper typeMapper)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior, Boolean async, CancellationToken cancellationToken)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior, Boolean async, CancellationToken cancellationToken)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior)
   at Program.<Main>$(String[] args) in C:\Users\natalie-perret\Desktop\Personal\playground\csharp\CSharpPlayground\Program.cs:line 32

似乎由于某種原因需要注冊型別，實際上如果我不指定型別：

Unhandled exception. System.NotSupportedException: The CLR type System.Object isn't natively supported by Npgsql or your PostgreSQL. To use it with a PostgreSQL composite
 you need to specify DataTypeName or to map it, please refer to the documentation.
   at Npgsql.TypeMapping.ConnectorTypeMapper.ResolveByClrType(Type type)
   at Npgsql.TypeMapping.ConnectorTypeMapper.ResolveByClrType(Type type)
   at Npgsql.NpgsqlParameter.ResolveHandler(ConnectorTypeMapper typeMapper)
   at Npgsql.NpgsqlParameter.Bind(ConnectorTypeMapper typeMapper)
   at Npgsql.NpgsqlParameterCollection.ValidateAndBind(ConnectorTypeMapper typeMapper)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior, Boolean async, CancellationToken cancellationToken)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior, Boolean async, CancellationToken cancellationToken)
   at Npgsql.NpgsqlCommand.ExecuteReader(CommandBehavior behavior)
   at Program.<Main>$(String[] args) in C:\Users\natalie-perret\Desktop\Personal\playground\csharp\CSharpPlayground\Program.cs:line 31

[編輯]

我最終得到的臨時解決方案是依賴 jsonb 支持，特別是jsonb_to_recordset 函式（請參閱PostgreSQL 檔案部分關于 json 函式）：

using System.Data;
using System.Text.Json;

using Npgsql;
using NpgsqlTypes;


var connectionStringBuilder = new NpgsqlConnectionStringBuilder
{
    Host     = "localhost",
    Port     = 5432,
    Username = "brand_payment_migration",
    Password = "secret",
    Database = "brand_payment"
};
using var connection = new NpgsqlConnection(connectionStringBuilder.ToString());
connection.Open();
using var command = connection.CreateCommand();
command.CommandText = 
"SELECT id, accounting_document, accounting_document_type, company_code, document_date_year, posting_view_item "  
"FROM sch_brand_payment_data_lake_proxy.supplier_balances "  
"LEFT OUTER JOIN sch_brand_payment_data_lake_proxy.supplier_balance_items "  
"ON supplier_balances.id = supplier_balance_items.supplier_balance_id "  
"WHERE (accounting_document, accounting_document_type, company_code, document_date_year) "  
"IN (SELECT * FROM jsonb_to_recordset(@values) "  
"AS params (accounting_document text, accounting_document_type text, company_code text, document_date_year integer));";
var parameter = command.CreateParameter();
parameter.ParameterName = "@values";
parameter.NpgsqlDbType = NpgsqlDbType.Jsonb;
parameter.NpgsqlValue = JsonSerializer.Serialize(new []
{
    new Params("A", "B", "C", 1), 
    new Params("A", "B", "C", 2)
});
command.Parameters.Add(parameter);
using var reader = command.ExecuteReader();
using var dataTable = new DataTable();
dataTable.Load(reader);
var cols = dataTable.Columns.Cast<DataColumn>().ToArray();
Console.WriteLine(string.Join(Environment.NewLine, cols.Select((x, i) => $"Col{i} = {x}")));
Console.WriteLine(string.Join("\t", cols.Select((_, i) => $"Col{i}")));
foreach (var dataRow in dataTable.Rows.Cast<DataRow>())
{
    Console.WriteLine(string.Join("\t", dataRow.ItemArray));
}


public Params(
    string accounting_document, 
    string accounting_document_type,
    string company_code,
    int document_date_year);

輸出：

Col0 = id
Col1 = accounting_document
Col2 = accounting_document_type
Col3 = company_code
Col4 = document_date_year
Col5 = posting_view_item
Col0    Col1    Col2    Col3    Col4    Col5
2       A       B       C       1       A
3       A       B       C       2       A
3       A       B       C       2       B

但這是以在傳遞引數時添加額外的 json 序列化步驟為代價的。因此，除了構建一個非常長的字串之外，我有點困惑的是，沒有辦法直接將實際值傳遞給NpgsqlParameter.NpgsqlValue屬性，而無需額外的步驟。

[編輯 2]

添加一個DbFiddle

[編輯 3]

可以使用相同的 jsonb“技巧”來提供資料（盡管我已經在上面提到了同樣的問題）：

INSERT INTO sch_brand_payment_data_lake_proxy.supplier_balances
    (accounting_document, accounting_document_type, company_code, document_date_year)
SELECT * FROM jsonb_to_recordset(
    '[{"accounting_document":"E","accounting_document_type":"B","company_code":"C","document_date_year":1},
      {"accounting_document":"E","accounting_document_type":"B","company_code":"C","document_date_year":2}]'::jsonb)
       AS params (accounting_document text, accounting_document_type text, company_code text, document_date_year integer)
RETURNING id;

uj5u.com熱心網友回復：

[編輯 3] 這篇文章很好地總結了一些事情： https ://dev.to/forbeslindesay/postgres-unnest-cheat-sheet-for-bulk-operations-1obg

[編輯 2]

跟進我今天早些時候提出的問題 https://github.com/npgsql/npgsql/issues/4437#issuecomment-1113999994

我已經解決了@dhedey在另一個以某種方式相關的問題中提到的解決方案/解決方法：

如果它對其他人有幫助，我發現使用該UNNEST命令對這些型別的查詢有一個很好的解決方法，該命令可以采用多個陣列引數并將它們一起壓縮到列中，這些列可以與表連接以過濾到相關列。

在某些情況下，連接的使用也比 ANY/IN 模式更高效。

SELECT * FROM table WHERE (itemauthor, itemtitle) = ANY (('bob', 'hello'), ('frank', 'hi')...)

可以表示為：

 var authorsParameter = new NpgsqlParameter("@authors", NpgsqlDbType.Array | NpgsqlDbType.Varchar)
    { Value = authors.ToList() };
var titlesParameter = new NpgsqlParameter("@titles", NpgsqlDbType.Array | NpgsqlDbType.Varchar)
    { Value = titles.ToList() };

var results = dbContext.Set<MyRow>()
    .FromSqlInterpolated($@"
SELECT
    t.*
FROM UNNEST({authorsParameter}, {titlesParameter}) params (author, title)
INNER JOIN table t
    ON t.author = params.author
    AND t.title = params.title
");

注意 - Varchar 可以被其他型別的引數替換為其他型別的陣列（例如 Bigint） - 查看NpgsqlDbType列舉以獲取更多詳細資訊。

然后我重寫了一些我最初發布的代碼，似乎unnestPostgreSQL 函式解決方案就像一個魅力。這是我暫時接受的答案，它看起來比 Json / JsonB 更整潔，后者需要進一步的 postgresql-json 特定映射惡作劇或提取。

不過，我還不太確定性能影響：

• unnest涉及到你映射差異
• jsonb_to_recordset需要額外的 .NET Json 序列化步驟，并且在某些情況下，需要將輸出顯式映射jsonb_to_recordset到相關列。

兩者都不是免費的。但我喜歡unnest它明確地為每一列（即更大的 .NET 型別（元組、記錄、類、結構等）的值的每個集合/集合）傳遞給要傳遞給NpgsqlParameter.NpgsqlValueDB 型別的屬性通過NpgsqlDbType列舉使用

using System.Data;

using Npgsql;
using NpgsqlTypes;


var connectionStringBuilder = new NpgsqlConnectionStringBuilder
{
    Host     = "localhost",
    Port     = 5432,
    Username = "brand_payment_migration",
    Password = "secret",
    Database = "brand_payment"
};
using var connection = new NpgsqlConnection(connectionStringBuilder.ToString());
connection.Open();

var selectStatement =
    "SELECT * FROM sch_brand_payment_data_lake_proxy.supplier_balances "  
    "WHERE (accounting_document, accounting_document_type, company_code, document_date_year) "  
    "IN (SELECT * FROM  unnest("  
    "@accounting_document_texts, "  
    "@accounting_document_types, "  
    "@company_codes, "  
    "@document_date_years"  
    "))";

var insertStatement = 
    "INSERT INTO sch_brand_payment_data_lake_proxy.supplier_balances "  
    "(accounting_document, accounting_document_type, company_code, document_date_year) "   
    "SELECT * FROM unnest("  
    "@accounting_document_texts, "  
    "@accounting_document_types, "  
    "@company_codes, "  
    "@document_date_years"   
    ") RETURNING id;";

var parameters = new (string Name, NpgsqlDbType DbType, object Value)[]
{
    ("@accounting_document_texts", NpgsqlDbType.Array | NpgsqlDbType.Text,    new[] {"G", "G", "G"}),
    ("@accounting_document_types", NpgsqlDbType.Array | NpgsqlDbType.Text,    new[] {"Y", "Y", "Y"}),
    ("@company_codes",             NpgsqlDbType.Array | NpgsqlDbType.Text,    new[] {"Z", "Z", "Z"}),
    ("@document_date_years",       NpgsqlDbType.Array | NpgsqlDbType.Integer, new[] {1, 2, 3})
};

connection.ExecuteNewCommandAndWriteResultToConsole(insertStatement, parameters);
connection.ExecuteNewCommandAndWriteResultToConsole(selectStatement, parameters);

public static class Extensions
{
    public static void AddParameter(this NpgsqlCommand command, string name, NpgsqlDbType dbType, object value)
    {
        var parameter = command.CreateParameter();
        parameter.ParameterName = name;
        parameter.NpgsqlDbType  = dbType;
        parameter.NpgsqlValue   = value;
        command.Parameters.Add(parameter);
    }

    public static NpgsqlCommand CreateCommand(this NpgsqlConnection connection, 
        string text, 
        IEnumerable<(string Name, NpgsqlDbType DbType, object Value)> parameters)
    {
        var command = connection.CreateCommand();
        command.CommandText = text;
        foreach (var (name, dbType, value) in parameters)
        {
            command.AddParameter(name, dbType, value);
        }

        return command;
    }
    public static void ExecuteAndWriteResultToConsole(this NpgsqlCommand command)
    {
        Console.WriteLine($"Executing command... {command.CommandText}");
        
        using var reader = command.ExecuteReader();
        using var dataTable = new DataTable();
        dataTable.Load(reader);
        var cols = dataTable.Columns.Cast<DataColumn>().ToArray();
        Console.WriteLine(string.Join(Environment.NewLine, cols.Select((x, i) => $"Col{i} = {x}")));
        Console.WriteLine(string.Join("\t", cols.Select((_, i) => $"Col{i}")));
        foreach (var dataRow in dataTable.Rows.Cast<DataRow>())
        {
            Console.WriteLine(string.Join("\t", dataRow.ItemArray));
        }
    }

    public static void ExecuteNewCommandAndWriteResultToConsole(this NpgsqlConnection connection, 
        string text,
        IEnumerable<(string Name, NpgsqlDbType DbType, object Value)> parameters)
    {
        using var command = connection.CreateCommand(text, parameters);
        command.ExecuteAndWriteResultToConsole();
    }
}

輸出：

Executing command... INSERT INTO sch_brand_payment_data_lake_proxy.supplier_balances (accounting_document, accounting_document_type, company_code, document_date_year) SEL
ECT * FROM unnest(@accounting_document_texts, @accounting_document_types, @company_codes, @document_date_years) RETURNING id;
Col0 = id
Col0
28
29
30
Executing command... SELECT * FROM sch_brand_payment_data_lake_proxy.supplier_balances WHERE (accounting_document, accounting_document_type, company_code, document_date_y
ear) IN (SELECT * FROM  unnest(@accounting_document_texts, @accounting_document_types, @company_codes, @document_date_years))
Col0 = id
Col1 = accounting_document
Col2 = accounting_document_type
Col3 = company_code
Col4 = document_date_year
Col5 = accounting_doc_created_by_user
Col6 = accounting_clerk
Col7 = assignment_reference
Col8 = document_reference_id
Col9 = original_reference_document
Col10 = payment_terms
Col11 = supplier
Col12 = supplier_name
Col13 = document_date
Col14 = posting_date
Col15 = net_due_date
Col16 = created_on
Col17 = modified_on
Col18 = pushed_on
Col19 = is_modified
Col20 = is_pushed
Col0    Col1    Col2    Col3    Col4    Col5    Col6    Col7    Col8    Col9    Col10   Col11   Col12   Col13   Col14   Col15   Col16   Col17   Col18   Col19   Col20
28      G       Y       Z       1                                                                                                                       False   False
29      G       Y       Z       2                                                                                                                       False   False
30      G       Y       Z       3                                                                                                                       False   False

[編輯 1]

由于@Charlieface 指出這不是合適的答案，我認為最好從 npgsql 維護者/貢獻者那里獲得答案/資訊。

因此在他們的 GitHub 存盤庫中提交了一個問題：https ://github.com/npgsql/npgsql/issues/4437

原答案：

到今天為止，除了其他東西之外，沒有辦法將元組或集合作為復合“型別”或通過位置斜杠隱式“定義”（然后可以在已經傳遞給引數值的集合中使用）屬性），npgslq 需要先前的 PostgreSQL 型別定義（但元組和嵌套集合仍然無法解決，因為維護者或至少其中一個認為不夠安全）。 https://github.com/npgsql/npgsql/issues/2154

正如例外所說，資料庫中需要相應的組合。這是因為匿名型別沒有映射到記錄。

因此，您應該創建一個型別和一個必須映射到該型別的結構。

僅供參考，有一個類似的問題#2097來跟蹤映射組合到值元組。

但這需要 npgsql 的其他一些相關開發人員，例如#2097 ，作者/主要貢獻在https://github.com/dotnet/efcore/issues/14661#issuecomment-462440199中被認為過于脆弱

請注意，在npgsql/npgsql#2097中討論之后，我們決定放棄這個想法。C# 值元組沒有名稱，因此任何到 PostgreSQL 組合的映射都將依賴于欄位定義排序，這似乎非常危險/脆弱。

我終于決定接受 jsonb 替代方案，不是一個超級粉絲，但至少它允許以相對安全的方式傳遞集合（只要傳遞 jsonb 的序列化在控制之下）。

但是我最初設想的做法不是今天可以做到的。

在寫這篇文章的程序中，我還學到了一件事：

• 有一個非常好的專用于 PostgreSQL 的 Slack 服務器：postgresteam.slack.com
• 關于在尋求 PostgreSQL 相關幫助時如何正確格式化 SQL 的一個很好的指南（盡管對作者的意見有意見）：https ://www.depesz.com/2010/05/28/what-mistakes-you-can-在 irc 上尋找幫助時避免/
• 根據作者偏好自動格式化 SQL 的粘貼箱：https ://paste.depesz.com

標籤：C＃ 。网 PostgreSQL npgsql

上一篇：PostgreSQLtimescale-如何按時間和id對行進行排序？

下一篇：使用ffmpeg把mp4與m3u8相互轉換的操作