GH-46584: [C++][FlightRPC] Iterate over endpoints in ODBC driver

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

@@ -162,9 +162,11 @@ add_arrow_test(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
                util_test.cc
                EXTRA_LINK_LIBS
-               arrow_odbc_spi_impl)
+               arrow_odbc_spi_impl
+               arrow_flight_testing_shared)

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

@@ -44,8 +44,8 @@ class NoOpClientAuthHandler : public ClientAuthHandler {
   NoOpClientAuthHandler() {}
 
   Status Authenticate(ClientAuthSender* outgoing, ClientAuthReader* incoming) override {
-    // Write a blank string. The server should ignore this and just accept any Handshake
-    // request.
+    // The server should ignore this and just accept any Handshake
+    // request. Some servers do not allow authentication with no handshakes.
     return outgoing->Write(std::string());
   }
 

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

@@ -100,6 +100,8 @@ inline std::string GetCerts() { return ""; }
 #endif
 
 const std::set<std::string_view, CaseInsensitiveComparator> BUILT_IN_PROPERTIES = {
+    FlightSqlConnection::DRIVER,
+    FlightSqlConnection::DSN,
     FlightSqlConnection::HOST,
     FlightSqlConnection::PORT,
     FlightSqlConnection::USER,
@@ -153,14 +155,14 @@ 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<ClientMiddlewareFactory>& cookie_factory = 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));
 
     std::unique_ptr<FlightSqlAuthMethod> auth_method =
         FlightSqlAuthMethod::FromProperties(flight_client, properties);
@@ -364,7 +366,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,
@@ -410,7 +412,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/odbc_impl/flight_sql_result_set.cc

@@ -29,12 +29,12 @@
 namespace arrow::flight::sql::odbc {
 
 FlightSqlResultSet::FlightSqlResultSet(
-    FlightSqlClient& flight_sql_client, const FlightCallOptions& call_options,
-    const std::shared_ptr<FlightInfo>& flight_info,
+    FlightSqlClient& flight_sql_client, const FlightClientOptions& client_options,
+    const FlightCallOptions& call_options, const std::shared_ptr<FlightInfo>& flight_info,
     const std::shared_ptr<RecordBatchTransformer>& transformer, Diagnostics& diagnostics,
     const 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/odbc_impl/flight_sql_result_set.h

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

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

@@ -41,9 +41,10 @@ using util::ThrowIfNotOK;
 
 namespace {
 
-void ClosePreparedStatementIfAny(std::shared_ptr<PreparedStatement>& prepared_statement) {
+void ClosePreparedStatementIfAny(std::shared_ptr<PreparedStatement>& prepared_statement,
+                                 const FlightCallOptions& options) {
   if (prepared_statement != nullptr) {
-    ThrowIfNotOK(prepared_statement->Close());
+    ThrowIfNotOK(prepared_statement->Close(options));
     prepared_statement.reset();
   }
 }
@@ -52,11 +53,13 @@ void ClosePreparedStatementIfAny(std::shared_ptr<PreparedStatement>& prepared_st
 
 FlightSqlStatement::FlightSqlStatement(const Diagnostics& diagnostics,
                                        FlightSqlClient& sql_client,
+                                       FlightClientOptions client_options,
                                        FlightCallOptions call_options,
                                        const 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);
@@ -97,7 +100,7 @@ boost::optional<Statement::Attribute> FlightSqlStatement::GetAttribute(
 
 boost::optional<std::shared_ptr<ResultSetMetadata>> FlightSqlStatement::Prepare(
     const std::string& query) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<PreparedStatement>> result =
       sql_client_.Prepare(call_options_, query);
@@ -111,27 +114,30 @@ boost::optional<std::shared_ptr<ResultSetMetadata>> FlightSqlStatement::Prepare(
 }
 
 bool FlightSqlStatement::ExecutePrepared() {
+  // GH-47990 TODO: use DCHECK instead of assert
   assert(prepared_statement_.get() != nullptr);
 
-  Result<std::shared_ptr<FlightInfo>> result = prepared_statement_->Execute();
+  Result<std::shared_ptr<FlightInfo>> result =
+      prepared_statement_->Execute(call_options_);
+
   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;
 }
 
 bool FlightSqlStatement::Execute(const std::string& query) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<FlightInfo>> result = sql_client_.Execute(call_options_, 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;
 }
@@ -146,33 +152,35 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTables(
     const std::string* catalog_name, const std::string* schema_name,
     const std::string* table_name, const std::string* table_type,
     const ColumnNames& column_names) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   std::vector<std::string> table_types;
 
   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_;
@@ -199,7 +207,7 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTables_V3(
 std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V2(
     const std::string* catalog_name, const std::string* schema_name,
     const std::string* table_name, const std::string* column_name) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<FlightInfo>> result = sql_client_.GetTables(
       call_options_, catalog_name, schema_name, table_name, true, nullptr);
@@ -210,17 +218,17 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V2(
   auto transformer = std::make_shared<GetColumns_Transformer>(
       metadata_settings_, OdbcVersion::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_;
 }
 
 std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V3(
     const std::string* catalog_name, const std::string* schema_name,
     const std::string* table_name, const std::string* column_name) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<FlightInfo>> result = sql_client_.GetTables(
       call_options_, catalog_name, schema_name, table_name, true, nullptr);
@@ -231,15 +239,15 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetColumns_V3(
   auto transformer = std::make_shared<GetColumns_Transformer>(
       metadata_settings_, OdbcVersion::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_;
 }
 
 std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V2(int16_t data_type) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<FlightInfo>> result = sql_client_.GetXdbcTypeInfo(call_options_);
   ThrowIfNotOK(result.status());
@@ -249,15 +257,15 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V2(int16_t data_type)
   auto transformer = std::make_shared<GetTypeInfoTransformer>(
       metadata_settings_, OdbcVersion::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_;
 }
 
 std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V3(int16_t data_type) {
-  ClosePreparedStatementIfAny(prepared_statement_);
+  ClosePreparedStatementIfAny(prepared_statement_, call_options_);
 
   Result<std::shared_ptr<FlightInfo>> result = sql_client_.GetXdbcTypeInfo(call_options_);
   ThrowIfNotOK(result.status());
@@ -267,9 +275,9 @@ std::shared_ptr<ResultSet> FlightSqlStatement::GetTypeInfo_V3(int16_t data_type)
   auto transformer = std::make_shared<GetTypeInfoTransformer>(
       metadata_settings_, OdbcVersion::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/odbc_impl/flight_sql_statement.h

@@ -32,6 +32,7 @@ class FlightSqlStatement : public Statement {
  private:
   Diagnostics diagnostics_;
   std::map<StatementAttributeId, Attribute> attribute_;
+  FlightClientOptions client_options_;
   FlightCallOptions call_options_;
   FlightSqlClient& sql_client_;
   std::shared_ptr<ResultSet> current_result_set_;
@@ -46,7 +47,7 @@ class FlightSqlStatement : public Statement {
 
  public:
   FlightSqlStatement(const Diagnostics& diagnostics, FlightSqlClient& sql_client,
-                     FlightCallOptions call_options,
+                     FlightClientOptions client_options, FlightCallOptions call_options,
                      const MetadataSettings& metadata_settings);
 
   bool SetAttribute(StatementAttributeId attribute, const Attribute& value) override;