隆重介紹 Apache Arrow Flight SQL：加速資料庫存取

已發布 2022 年 2 月 16 日
作者 José Almeida, James Duong, Vinicius Fraga, Juscelino Junior, David Li, Kyle Porter, Rafael Telles

我們想介紹 Flight SQL，這是一個由 Apache Arrow 社群開發的新客戶端-伺服器協定，用於與 SQL 資料庫互動，它利用了 Arrow 記憶體內 columnar 格式和 Flight RPC 框架。

Flight SQL 旨在提供與現有 API（如 JDBC 和 ODBC）大致相似的功能，包括執行查詢、建立預備語句以及獲取有關支援的 SQL 方言、可用類型、已定義表格等的元數據。然而，透過建構於 Apache Arrow 之上，Flight SQL 使客戶端可以輕鬆地與 Arrow 原生資料庫交談，而無需轉換資料。並且透過使用 Flight，它提供了一個線路格式的有效實作，該格式開箱即用地支援加密和身份驗證等功能，同時允許進一步的優化，例如並行資料存取。

雖然它可以直接用於資料庫存取，但它並不是 JDBC/ODBC 的直接替代品。相反，Flight SQL 作為一個具體的線路協定/驅動程式實作，可以支援 JDBC/ODBC 驅動程式，並減輕資料庫的實作負擔。

Illustration of where Flight SQL sits in the stack. JDBC and ODBC drivers can wrap Flight SQL, or an Arrow-native application can directly use the Flight SQL libraries. Flight SQL in turn talks over Arrow Flight to a database exposing a Flight SQL endpoint.

動機

雖然像 JDBC 和 ODBC 這樣的標準已經為使用者服務了數十年，但對於希望使用 Apache Arrow 或一般 columnar 資料的資料庫和客戶端來說，它們仍有不足之處。在這種情況下，像 JDBC 或 PEP 249 這樣基於列的 API 需要轉置資料，而對於本身就是 columnar 的資料庫來說，這意味著資料必須轉置兩次——一次是為了在 API 中以列的形式呈現，另一次是為了讓消費者將其恢復為欄。同時，雖然像 ODBC 這樣的 API 確實提供了對結果緩衝區的批量存取，但這些資料仍然必須複製到 Arrow 陣列中，以便與更廣泛的 Arrow 生態系統一起使用，正如像 Turbodbc 這樣的專案所實作的那樣。Flight SQL 旨在消除這些中間步驟。

Flight SQL 意味著資料庫伺服器可以實作一個標準介面，該介面從一開始就圍繞 Apache Arrow 和 columnar 資料設計。就像 Arrow 提供標準的記憶體內格式一樣，Flight SQL 讓開發人員無需設計和實作全新的線路協定。如前所述，Flight 已經實作了諸如線路加密和請求驗證等功能，資料庫無需重新實作這些功能。

對於客戶端，Flight SQL 提供了對查詢結果的批量存取，而無需從另一個 API 或格式轉換資料。此外，透過將實作線路協定的工作推送到 Flight 和 Flight SQL 函式庫中，每個客戶端語言或驅動程式需要編寫的程式碼更少。並且透過在底層使用 Flight，客戶端和伺服器可以合作實作諸如並行資料存取等優化，這是 Flight 本身的最初目標之一。資料庫可以向 Flight SQL 客戶端返回多個「端點」，然後客戶端可以從所有端點並行提取資料，從而使資料庫後端能夠水平擴展。

Flight SQL 基礎知識

Flight SQL 充分利用了 Flight RPC 框架及其可擴展性，透過 Protobuf 定義了額外的請求/回應訊息。我們將簡要介紹 Flight SQL 協定，但 C++ 和 Java 已經實作了管理大部分工作的客戶端。完整的協定可以在 GitHub 上找到。

大多數請求都遵循此模式

  1. 客戶端使用定義的 Protobuf 訊息之一建構請求。
  2. 客戶端透過 GetSchema RPC 方法（取得回應的 schema）或 GetFlightInfo RPC 方法（執行請求）發送請求。
  3. 客戶端向從 GetFlightInfo 返回的端點發出請求以獲取回應。

Flight SQL 定義了查詢資料庫元數據、執行查詢或操作預備語句的方法。

元數據請求

  • CommandGetCatalogs：列出資料庫中的 catalogs。
  • CommandGetCrossReference：列出引用特定其他表格的外鍵欄。
  • CommandGetDbSchemas：列出 catalog 中的 schemas。
  • CommandGetExportedKeys：列出引用表格的外鍵。
  • CommandGetImportedKeys：列出表格的外鍵。
  • CommandGetPrimaryKeys：列出表格的主鍵。
  • CommandGetSqlInfo：取得有關資料庫本身及其支援的 SQL 方言的資訊。
  • CommandGetTables：列出 catalog/schema 中的表格。
  • CommandGetTableTypes：列出支援的表格類型（例如，表格、視圖、系統表格）。

查詢

  • CommandStatementQuery：執行一次性的 SQL 查詢。
  • CommandStatementUpdate：執行一次性的 SQL 更新查詢。

預備語句

  • ActionClosePreparedStatementRequest：關閉預備語句。
  • ActionCreatePreparedStatementRequest：建立新的預備語句。
  • CommandPreparedStatementQuery：執行預備語句。
  • CommandPreparedStatementUpdate：執行更新資料的預備語句。

例如，列出所有表格

Sequence diagram showing how to use CommandGetTables. First, the client calls the GetFlightInfo RPC method with a serialized CommandGetTables message as the argument. The server returns a FlightInfo message containing a Ticket message. The client then calls the DoGet RPC method with the Ticket as the argument, and gets back a stream of Arrow record batches containing the tables in the database.

執行查詢

Sequence diagram showing how to use CommandStatementQuery. First, the client calls the GetFlightInfo RPC method with a serialized CommandStatementQuery message as the argument. This message contains the SQL query. The server returns a FlightInfo message containing a Ticket message. The client then calls the DoGet RPC method with the Ticket as the argument, and gets back a stream of Arrow record batches containing the query results.

建立並執行預備語句以插入列

Sequence diagram showing how to use ActionCreatePreparedStatementResult. First, the client calls the DoAction RPC method with a serialized ActionCreatePreparedStatementResult message as the argument. This message contains the SQL query. The server returns a serialized ActionCreatePreparedStatementResult message containing an opaque handle for the prepared statement. The client then calls the DoPut RPC method with a CommandPreparedStatementUpdate message, containing the opaque handle, as the argument, and uploads a stream of Arrow record batches containing query parameters. The server responds with a serialized DoPutUpdateResult message containing the number of affected rows. Finally, the client calls DoAction again with ActionClosePreparedStatementRequest to clean up the prepared statement.

開始使用

請注意，雖然 Flight SQL 作為 Apache Arrow 7.0.0 的一部分發布，但它仍在開發中，詳細的文件即將推出。然而，C++ 和 Java 中已經提供了實作，它們提供了一個可以使用的低階客戶端以及一個可以實作的伺服器骨架。

對於那些感興趣的人，來源程式碼中提供了一個封裝 Apache Derby 的伺服器實作和一個封裝 SQLite 的伺服器實作。還提供了一個展示客戶端的簡單 CLI。最後，我們可以看一個執行查詢和獲取結果的簡短範例

flight::FlightCallOptions call_options;

// Execute the query, getting a FlightInfo describing how to fetch the results
std::cout << "Executing query: '" << FLAGS_query << "'" << std::endl;
ARROW_ASSIGN_OR_RAISE(std::unique_ptr<flight::FlightInfo> flight_info,
                      client->Execute(call_options, FLAGS_query));

// Fetch each partition sequentially (though this can be done in parallel)
for (const flight::FlightEndpoint& endpoint : flight_info->endpoints()) {
  // Here we assume each partition is on the same server we originally queried, but this
  // isn't true in general: the server may split the query results between multiple
  // other servers, which we would have to connect to.

  // The "ticket" in the endpoint is opaque to the client. The server uses it to
  // identify which part of the query results to return.
  ARROW_ASSIGN_OR_RAISE(auto stream, client->DoGet(call_options, endpoint.ticket));
  // Read all results into an Arrow Table, though we can iteratively process record
  // batches as they arrive as well
  std::shared_ptr<arrow::Table> table;
  ARROW_RETURN_NOT_OK(stream->ReadAll(&table));
  std::cout << "Read one partition:" << std::endl;
  std::cout << table->ToString() << std::endl;
}

完整的原始碼可在 GitHub 上取得。

下一步 & 參與方式

與現有的函式庫（如 PyODBC）相比，Arrow Flight 已經快了 20 倍（~00:21:00）。Flight SQL 將把這些效能優勢打包到一個標準介面中，供客戶端和資料庫實作。

預計將進一步完善和擴展協定。部分工作是為了使在 Flight SQL 之上實作像 JDBC 這樣的 API 成為可能；JDBC 驅動程式正在積極開發中。雖然這再次引入了資料轉換的開銷，但這意味著資料庫可以透過實作 Flight SQL 使自身可供 Arrow 原生客戶端和傳統客戶端訪問。未來的其他改進可能包括 Python 綁定、ODBC 驅動程式等等。

對於任何有興趣參與的人，無論是作為貢獻者還是採用者，請透過郵件列表聯繫，或加入 GitHub 上的討論。