GH-46584 Iterate endpoint

cpp/src/arrow/flight/CMakeLists.txt

@@ -291,7 +291,9 @@ if(ARROW_TESTING)
                 STATIC_INSTALL_INTERFACE_LIBS
                 ${ARROW_FLIGHT_TESTING_STATIC_INSTALL_INTERFACE_LIBS}
                 PRIVATE_INCLUDES
-                "${Protobuf_INCLUDE_DIRS}")
+                "${Protobuf_INCLUDE_DIRS}"
+                SHARED_PRIVATE_LINK_LIBS
+                GTest::gmock)
 
   foreach(LIB_TARGET ${ARROW_FLIGHT_TESTING_LIBRARIES})
     target_compile_definitions(${LIB_TARGET} PRIVATE ARROW_FLIGHT_EXPORTING)

cpp/src/arrow/flight/sql/odbc/flight_sql/CMakeLists.txt

@@ -136,9 +136,11 @@ add_arrow_test(arrow_odbc_spi_impl_test
                accessors/time_array_accessor_test.cc
                accessors/timestamp_array_accessor_test.cc
                flight_sql_connection_test.cc
+               flight_sql_stream_chunk_buffer_test.cc
                parse_table_types_test.cc
                json_converter_test.cc
                record_batch_transformer_test.cc
                utils_test.cc
                EXTRA_LINK_LIBS
-               arrow_odbc_spi_impl)
+               arrow_odbc_spi_impl
+               arrow_flight_testing_shared)

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_auth_method.cc

@@ -58,9 +58,9 @@ class NoOpClientAuthHandler : public arrow::flight::ClientAuthHandler {
 
   arrow::Status Authenticate(arrow::flight::ClientAuthSender* outgoing,
                              arrow::flight::ClientAuthReader* incoming) override {
-    // Write a blank string. The server should ignore this and just accept any Handshake
+    // Return OK Status. The server should ignore this and just accept any Handshake
     // request.
-    return outgoing->Write(std::string());
+    return arrow::Status::OK();
   }
 
   arrow::Status GetToken(std::string* token) override {

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_connection.cc

@@ -112,6 +112,8 @@ inline std::string GetCerts() { return ""; }
 #endif
 
 const std::set<std::string_view, CaseInsensitiveComparatorStrView> BUILT_IN_PROPERTIES = {
+    FlightSqlConnection::DRIVER,
+    FlightSqlConnection::DSN,
     FlightSqlConnection::HOST,
     FlightSqlConnection::PORT,
     FlightSqlConnection::USER,
@@ -165,15 +167,15 @@ void FlightSqlConnection::Connect(const ConnPropertyMap& properties,
     auto flight_ssl_configs = LoadFlightSslConfigs(properties);
 
     Location location = BuildLocation(properties, missing_attr, flight_ssl_configs);
-    FlightClientOptions client_options =
+    client_options_ =
         BuildFlightClientOptions(properties, missing_attr, flight_ssl_configs);
 
     const std::shared_ptr<arrow::flight::ClientMiddlewareFactory>& cookie_factory =
         arrow::flight::GetCookieFactory();
-    client_options.middleware.push_back(cookie_factory);
+    client_options_.middleware.push_back(cookie_factory);
 
     std::unique_ptr<FlightClient> flight_client;
-    ThrowIfNotOK(FlightClient::Connect(location, client_options).Value(&flight_client));
+    ThrowIfNotOK(FlightClient::Connect(location, client_options_).Value(&flight_client));
 
     PopulateMetadataSettings(properties);
     PopulateCallOptions(properties);
@@ -377,7 +379,7 @@ void FlightSqlConnection::Close() {
 
 std::shared_ptr<Statement> FlightSqlConnection::CreateStatement() {
   return std::shared_ptr<Statement>(new FlightSqlStatement(
-      diagnostics_, *sql_client_, call_options_, metadata_settings_));
+      diagnostics_, *sql_client_, client_options_, call_options_, metadata_settings_));
 }
 
 bool FlightSqlConnection::SetAttribute(Connection::AttributeId attribute,
@@ -423,7 +425,7 @@ FlightSqlConnection::FlightSqlConnection(OdbcVersion odbc_version,
                                          const std::string& driver_version)
     : diagnostics_("Apache Arrow", "Flight SQL", odbc_version),
       odbc_version_(odbc_version),
-      info_(call_options_, sql_client_, driver_version),
+      info_(client_options_, call_options_, sql_client_, driver_version),
       closed_(true) {
   attribute_[CONNECTION_DEAD] = static_cast<uint32_t>(SQL_TRUE);
   attribute_[LOGIN_TIMEOUT] = static_cast<uint32_t>(0);

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_result_set.cc

@@ -44,13 +44,14 @@ using odbcabstraction::DriverException;
 
 FlightSqlResultSet::FlightSqlResultSet(
     FlightSqlClient& flight_sql_client,
+    const arrow::flight::FlightClientOptions& client_options,
     const arrow::flight::FlightCallOptions& call_options,
     const std::shared_ptr<FlightInfo>& flight_info,
     const std::shared_ptr<RecordBatchTransformer>& transformer,
     odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings)
     : metadata_settings_(metadata_settings),
-      chunk_buffer_(flight_sql_client, call_options, flight_info,
+      chunk_buffer_(flight_sql_client, client_options, call_options, flight_info,
                     metadata_settings_.chunk_buffer_capacity_),
       transformer_(transformer),
       metadata_(transformer

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_result_set.h

@@ -63,6 +63,7 @@ class FlightSqlResultSet : public ResultSet {
   ~FlightSqlResultSet() override;
 
   FlightSqlResultSet(FlightSqlClient& flight_sql_client,
+                     const arrow::flight::FlightClientOptions& client_options,
                      const arrow::flight::FlightCallOptions& call_options,
                      const std::shared_ptr<FlightInfo>& flight_info,
                      const std::shared_ptr<RecordBatchTransformer>& transformer,

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_statement.cc

@@ -64,11 +64,12 @@ void ClosePreparedStatementIfAny(
 
 FlightSqlStatement::FlightSqlStatement(
     const odbcabstraction::Diagnostics& diagnostics, FlightSqlClient& sql_client,
-    FlightCallOptions call_options,
+    arrow::flight::FlightClientOptions client_options, FlightCallOptions call_options,
     const odbcabstraction::MetadataSettings& metadata_settings)
     : diagnostics_("Apache Arrow", diagnostics.GetDataSourceComponent(),
                    diagnostics.GetOdbcVersion()),
       sql_client_(sql_client),
+      client_options_(std::move(client_options)),
       call_options_(std::move(call_options)),
       metadata_settings_(metadata_settings) {
   attribute_[METADATA_ID] = static_cast<size_t>(SQL_FALSE);
@@ -135,8 +136,8 @@ bool FlightSqlStatement::ExecutePrepared() {
   ThrowIfNotOK(result.status());
 
   current_result_set_ = std::make_shared<FlightSqlResultSet>(
-      sql_client_, call_options_, result.ValueOrDie(), nullptr, diagnostics_,
-      metadata_settings_);
+      sql_client_, client_options_, call_options_, result.ValueOrDie(), nullptr,
+      diagnostics_, metadata_settings_);
 
   return true;
 }
@@ -148,8 +149,8 @@ bool FlightSqlStatement::Execute(const std::string& query) {
   ThrowIfNotOK(result.status());
 
   current_result_set_ = std::make_shared<FlightSqlResultSet>(
-      sql_client_, call_options_, result.ValueOrDie(), nullptr, diagnostics_,
-      metadata_settings_);
+      sql_client_, client_options_, call_options_, result.ValueOrDie(), nullptr,
+      diagnostics_, metadata_settings_);
 
   return true;
 }
@@ -170,27 +171,29 @@ std::shared_ptr<odbcabstraction::ResultSet> FlightSqlStatement::GetTables(
 
   if ((catalog_name && *catalog_name == "%") && (schema_name && schema_name->empty()) &&
       (table_name && table_name->empty())) {
-    current_result_set_ = GetTablesForSQLAllCatalogs(
-        column_names, call_options_, sql_client_, diagnostics_, metadata_settings_);
+    current_result_set_ =
+        GetTablesForSQLAllCatalogs(column_names, client_options_, call_options_,
+                                   sql_client_, diagnostics_, metadata_settings_);
   } else if ((catalog_name && catalog_name->empty()) &&
              (schema_name && *schema_name == "%") &&
              (table_name && table_name->empty())) {
-    current_result_set_ =
-        GetTablesForSQLAllDbSchemas(column_names, call_options_, sql_client_, schema_name,
-                                    diagnostics_, metadata_settings_);
+    current_result_set_ = GetTablesForSQLAllDbSchemas(
+        column_names, client_options_, call_options_, sql_client_, schema_name,
+        diagnostics_, metadata_settings_);
   } else if ((catalog_name && catalog_name->empty()) &&
              (schema_name && schema_name->empty()) &&
              (table_name && table_name->empty()) && (table_type && *table_type == "%")) {
-    current_result_set_ = GetTablesForSQLAllTableTypes(
-        column_names, call_options_, sql_client_, diagnostics_, metadata_settings_);
+    current_result_set_ =
+        GetTablesForSQLAllTableTypes(column_names, client_options_, call_options_,
+                                     sql_client_, diagnostics_, metadata_settings_);
   } else {
     if (table_type) {
       ParseTableTypes(*table_type, table_types);
     }
 
     current_result_set_ = GetTablesForGenericUse(
-        column_names, call_options_, sql_client_, catalog_name, schema_name, table_name,
-        table_types, diagnostics_, metadata_settings_);
+        column_names, client_options_, call_options_, sql_client_, catalog_name,
+        schema_name, table_name, table_types, diagnostics_, metadata_settings_);
   }
 
   return current_result_set_;
@@ -228,9 +231,9 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V2(
   auto transformer = std::make_shared<GetColumns_Transformer>(
       metadata_settings_, odbcabstraction::V_2, column_name);
 
-  current_result_set_ =
-      std::make_shared<FlightSqlResultSet>(sql_client_, call_options_, flight_info,
-                                           transformer, diagnostics_, metadata_settings_);
+  current_result_set_ = std::make_shared<FlightSqlResultSet>(
+      sql_client_, client_options_, call_options_, flight_info, transformer, diagnostics_,
+      metadata_settings_);
 
   return current_result_set_;
 }
@@ -249,9 +252,9 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V3(
   auto transformer = std::make_shared<GetColumns_Transformer>(
       metadata_settings_, odbcabstraction::V_3, column_name);
 
-  current_result_set_ =
-      std::make_shared<FlightSqlResultSet>(sql_client_, call_options_, flight_info,
-                                           transformer, diagnostics_, metadata_settings_);
+  current_result_set_ = std::make_shared<FlightSqlResultSet>(
+      sql_client_, client_options_, call_options_, flight_info, transformer, diagnostics_,
+      metadata_settings_);
 
   return current_result_set_;
 }
@@ -267,9 +270,9 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V2(int16_t data_type)
   auto transformer = std::make_shared<GetTypeInfo_Transformer>(
       metadata_settings_, odbcabstraction::V_2, data_type);
 
-  current_result_set_ =
-      std::make_shared<FlightSqlResultSet>(sql_client_, call_options_, flight_info,
-                                           transformer, diagnostics_, metadata_settings_);
+  current_result_set_ = std::make_shared<FlightSqlResultSet>(
+      sql_client_, client_options_, call_options_, flight_info, transformer, diagnostics_,
+      metadata_settings_);
 
   return current_result_set_;
 }
@@ -285,9 +288,9 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V3(int16_t data_type)
   auto transformer = std::make_shared<GetTypeInfo_Transformer>(
       metadata_settings_, odbcabstraction::V_3, data_type);
 
-  current_result_set_ =
-      std::make_shared<FlightSqlResultSet>(sql_client_, call_options_, flight_info,
-                                           transformer, diagnostics_, metadata_settings_);
+  current_result_set_ = std::make_shared<FlightSqlResultSet>(
+      sql_client_, client_options_, call_options_, flight_info, transformer, diagnostics_,
+      metadata_settings_);
 
   return current_result_set_;
 }

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_statement.h

@@ -33,6 +33,7 @@ class FlightSqlStatement : public odbcabstraction::Statement {
  private:
   odbcabstraction::Diagnostics diagnostics_;
   std::map<StatementAttributeId, Attribute> attribute_;
+  arrow::flight::FlightClientOptions client_options_;
   arrow::flight::FlightCallOptions call_options_;
   arrow::flight::sql::FlightSqlClient& sql_client_;
   std::shared_ptr<odbcabstraction::ResultSet> current_result_set_;
@@ -48,6 +49,7 @@ class FlightSqlStatement : public odbcabstraction::Statement {
  public:
   FlightSqlStatement(const odbcabstraction::Diagnostics& diagnostics,
                      arrow::flight::sql::FlightSqlClient& sql_client,
+                     arrow::flight::FlightClientOptions client_options,
                      arrow::flight::FlightCallOptions call_options,
                      const odbcabstraction::MetadataSettings& metadata_settings);
   ~FlightSqlStatement();

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_statement_get_tables.cc

@@ -72,8 +72,9 @@ void ParseTableTypes(const std::string& table_type,
 }
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllCatalogs(
-    const ColumnNames& names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings) {
   Result<std::shared_ptr<FlightInfo>> result = sql_client.GetCatalogs(call_options);
 
@@ -92,14 +93,15 @@ std::shared_ptr<ResultSet> GetTablesForSQLAllCatalogs(
                          .AddFieldOfNulls(names.remarks_column, arrow::utf8())
                          .Build();
 
-  return std::make_shared<FlightSqlResultSet>(
-      sql_client, call_options, flight_info, transformer, diagnostics, metadata_settings);
+  return std::make_shared<FlightSqlResultSet>(sql_client, client_options, call_options,
+                                              flight_info, transformer, diagnostics,
+                                              metadata_settings);
 }
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllDbSchemas(
-    const ColumnNames& names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, const std::string* schema_name,
-    odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    const std::string* schema_name, odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings) {
   Result<std::shared_ptr<FlightInfo>> result =
       sql_client.GetDbSchemas(call_options, nullptr, schema_name);
@@ -119,13 +121,15 @@ std::shared_ptr<ResultSet> GetTablesForSQLAllDbSchemas(
                          .AddFieldOfNulls(names.remarks_column, arrow::utf8())
                          .Build();
 
-  return std::make_shared<FlightSqlResultSet>(
-      sql_client, call_options, flight_info, transformer, diagnostics, metadata_settings);
+  return std::make_shared<FlightSqlResultSet>(sql_client, client_options, call_options,
+                                              flight_info, transformer, diagnostics,
+                                              metadata_settings);
 }
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllTableTypes(
-    const ColumnNames& names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings) {
   Result<std::shared_ptr<FlightInfo>> result = sql_client.GetTableTypes(call_options);
 
@@ -144,15 +148,16 @@ std::shared_ptr<ResultSet> GetTablesForSQLAllTableTypes(
                          .AddFieldOfNulls(names.remarks_column, arrow::utf8())
                          .Build();
 
-  return std::make_shared<FlightSqlResultSet>(
-      sql_client, call_options, flight_info, transformer, diagnostics, metadata_settings);
+  return std::make_shared<FlightSqlResultSet>(sql_client, client_options, call_options,
+                                              flight_info, transformer, diagnostics,
+                                              metadata_settings);
 }
 
 std::shared_ptr<ResultSet> GetTablesForGenericUse(
-    const ColumnNames& names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, const std::string* catalog_name,
-    const std::string* schema_name, const std::string* table_name,
-    const std::vector<std::string>& table_types,
+    const ColumnNames& names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    const std::string* catalog_name, const std::string* schema_name,
+    const std::string* table_name, const std::vector<std::string>& table_types,
     odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings) {
   Result<std::shared_ptr<FlightInfo>> result = sql_client.GetTables(
@@ -173,8 +178,9 @@ std::shared_ptr<ResultSet> GetTablesForGenericUse(
                          .AddFieldOfNulls(names.remarks_column, arrow::utf8())
                          .Build();
 
-  return std::make_shared<FlightSqlResultSet>(
-      sql_client, call_options, flight_info, transformer, diagnostics, metadata_settings);
+  return std::make_shared<FlightSqlResultSet>(sql_client, client_options, call_options,
+                                              flight_info, transformer, diagnostics,
+                                              metadata_settings);
 }
 
 }  // namespace flight_sql

cpp/src/arrow/flight/sql/odbc/flight_sql/flight_sql_statement_get_tables.h

@@ -30,6 +30,7 @@ namespace driver {
 namespace flight_sql {
 
 using arrow::flight::FlightCallOptions;
+using arrow::flight::FlightClientOptions;
 using arrow::flight::sql::FlightSqlClient;
 using odbcabstraction::MetadataSettings;
 using odbcabstraction::ResultSet;
@@ -46,26 +47,28 @@ void ParseTableTypes(const std::string& table_type,
                      std::vector<std::string>& table_types);
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllCatalogs(
-    const ColumnNames& column_names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& column_names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings);
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllDbSchemas(
-    const ColumnNames& column_names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, const std::string* schema_name,
-    odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& column_names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    const std::string* schema_name, odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings);
 
 std::shared_ptr<ResultSet> GetTablesForSQLAllTableTypes(
-    const ColumnNames& column_names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, odbcabstraction::Diagnostics& diagnostics,
+    const ColumnNames& column_names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings);
 
 std::shared_ptr<ResultSet> GetTablesForGenericUse(
-    const ColumnNames& column_names, FlightCallOptions& call_options,
-    FlightSqlClient& sql_client, const std::string* catalog_name,
-    const std::string* schema_name, const std::string* table_name,
-    const std::vector<std::string>& table_types,
+    const ColumnNames& column_names, FlightClientOptions& client_options,
+    FlightCallOptions& call_options, FlightSqlClient& sql_client,
+    const std::string* catalog_name, const std::string* schema_name,
+    const std::string* table_name, const std::vector<std::string>& table_types,
     odbcabstraction::Diagnostics& diagnostics,
     const odbcabstraction::MetadataSettings& metadata_settings);
 }  // namespace flight_sql