GH-51341: [C++][FlightRPC][ODBC] Fix statement descriptor and cursor lifecycle

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

@@ -241,7 +241,7 @@ ODBCStatement::ODBCStatement(ODBCConnection& connection,
       ird_(std::make_shared<ODBCDescriptor>(spi_statement_->GetDiagnostics(), nullptr,
                                             this, false, false,
                                             connection.IsOdbc2Connection())),
-      current_ard_(built_in_apd_.get()),
+      current_ard_(built_in_ard_.get()),
       current_apd_(built_in_apd_.get()),
       row_number_(0),
       max_rows_(0),
@@ -323,6 +323,10 @@ void ODBCStatement::ExecuteDirect(const std::string& query) {
 
 bool ODBCStatement::Fetch(size_t rows, SQLULEN* row_count_ptr,
                           SQLUSMALLINT* row_status_array) {
+  if (!current_result_) {
+    throw DriverException("Invalid cursor state", "24000");
+  }
+
   if (has_reached_end_of_result_) {
     ird_->SetRowsProcessed(0);
     return false;
@@ -558,8 +562,9 @@ void ODBCStatement::SetStmtAttr(SQLINTEGER statement_attribute, SQLPOINTER value
 
   switch (statement_attribute) {
     case SQL_ATTR_APP_PARAM_DESC: {
-      ODBCDescriptor* desc = static_cast<ODBCDescriptor*>(value);
-      if (desc && current_apd_ != desc) {
+      ODBCDescriptor* desc =
+          value ? static_cast<ODBCDescriptor*>(value) : built_in_apd_.get();
+      if (current_apd_ != desc) {
         if (current_apd_ != built_in_apd_.get()) {
           current_apd_->DetachFromStatement(this, true);
         }
@@ -571,12 +576,14 @@ void ODBCStatement::SetStmtAttr(SQLINTEGER statement_attribute, SQLPOINTER value
       return;
     }
     case SQL_ATTR_APP_ROW_DESC: {
-      ODBCDescriptor* desc = static_cast<ODBCDescriptor*>(value);
-      if (desc && current_ard_ != desc) {
+      ODBCDescriptor* desc =
+          value ? static_cast<ODBCDescriptor*>(value) : built_in_ard_.get();
+      if (current_ard_ != desc) {
         if (current_ard_ != built_in_ard_.get()) {
           current_ard_->DetachFromStatement(this, false);
         }
         current_ard_ = desc;
+        current_ard_->NotifyBindingsHaveChanged();
         if (current_ard_ != built_in_ard_.get()) {
           desc->RegisterToStatement(this, false);
         }
@@ -718,6 +725,7 @@ void ODBCStatement::RevertAppDescriptor(bool isApd) {
     current_apd_ = built_in_apd_.get();
   } else {
     current_ard_ = built_in_ard_.get();
+    current_ard_->NotifyBindingsHaveChanged();
   }
 }
 
@@ -740,6 +748,10 @@ void ODBCStatement::CloseCursor(bool suppress_errors) {
 SQLRETURN ODBCStatement::GetData(SQLSMALLINT record_number, SQLSMALLINT c_type,
                                  SQLPOINTER data_ptr, SQLLEN buffer_length,
                                  SQLLEN* indicator_ptr) {
+  if (!current_result_) {
+    throw DriverException("Invalid cursor state", "24000");
+  }
+
   if (record_number == 0) {
     throw DriverException("Bookmarks are not supported", "07009");
   } else if (static_cast<size_t>(record_number) > ird_->GetRecords().size()) {
@@ -785,6 +797,7 @@ SQLRETURN ODBCStatement::GetData(SQLSMALLINT record_number, SQLSMALLINT c_type,
 
 SQLRETURN ODBCStatement::GetMoreResults() {
   // Multiple result sets are not supported by Arrow protocol.
+  CloseCursor(/*suppress_errors=*/true);
   return SQL_NO_DATA;
 }
 
@@ -811,6 +824,12 @@ void ODBCStatement::GetRowCount(SQLLEN* row_count_ptr) {
 
 void ODBCStatement::ReleaseStatement() {
   CloseCursor(true);
+  if (current_apd_ != built_in_apd_.get()) {
+    current_apd_->DetachFromStatement(this, true);
+  }
+  if (current_ard_ != built_in_ard_.get()) {
+    current_ard_->DetachFromStatement(this, false);
+  }
   connection_.DropStatement(this);
 }
 

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

@@ -35,6 +35,7 @@ set(ARROW_FLIGHT_SQL_ODBC_TEST_SRCS
     errors_test.cc
     get_functions_test.cc
     statement_attr_test.cc
+    statement_driver_api_test.cc
     statement_test.cc
     tables_test.cc
     type_info_test.cc

cpp/src/arrow/flight/sql/odbc/tests/connection_test.cc

@@ -528,6 +528,7 @@ TYPED_TEST(ConnectionTest, TestSQLSetStmtAttrDescriptor) {
 
   EXPECT_EQ(SQL_SUCCESS, SQLGetStmtAttr(this->stmt, SQL_ATTR_APP_ROW_DESC, &internal_ard,
                                         sizeof(internal_ard), 0));
+  EXPECT_NE(internal_apd, internal_ard);
 
   // Set APD descriptor to explicitly allocated handle
   EXPECT_EQ(SQL_SUCCESS, SQLSetStmtAttr(this->stmt, SQL_ATTR_APP_PARAM_DESC,
@@ -549,12 +550,31 @@ TYPED_TEST(ConnectionTest, TestSQLSetStmtAttrDescriptor) {
 
   EXPECT_EQ(ard_descriptor, value);
 
-  // Free explicitly allocated APD and ARD descriptor handles
-  ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, apd_descriptor));
+  // Assign replacement descriptors. Freeing the old descriptors must not revert these
+  // replacements, which proves switching descriptors detached the old handles. Null
+  // descriptor restoration is covered through the direct driver API because some driver
+  // managers intercept it.
+  SQLHDESC replacement_apd, replacement_ard;
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_DESC, this->conn, &replacement_apd));
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_DESC, this->conn, &replacement_ard));
+  ASSERT_EQ(SQL_SUCCESS,
+            SQLSetStmtAttr(this->stmt, SQL_ATTR_APP_PARAM_DESC, replacement_apd, 0));
+  ASSERT_EQ(SQL_SUCCESS,
+            SQLSetStmtAttr(this->stmt, SQL_ATTR_APP_ROW_DESC, replacement_ard, 0));
 
+  ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, apd_descriptor));
   ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, ard_descriptor));
+  EXPECT_EQ(SQL_SUCCESS, SQLGetStmtAttr(this->stmt, SQL_ATTR_APP_PARAM_DESC, &value,
+                                        sizeof(value), 0));
+  EXPECT_EQ(replacement_apd, value);
+  EXPECT_EQ(SQL_SUCCESS,
+            SQLGetStmtAttr(this->stmt, SQL_ATTR_APP_ROW_DESC, &value, sizeof(value), 0));
+  EXPECT_EQ(replacement_ard, value);
 
-  // Verify APD and ARD descriptors has been reverted to implicit descriptors
+  ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, replacement_apd));
+  ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, replacement_ard));
+
+  // Freeing the active explicit descriptors restores the implicit descriptors.
   value = nullptr;
 
   EXPECT_EQ(SQL_SUCCESS, SQLGetStmtAttr(this->stmt, SQL_ATTR_APP_PARAM_DESC, &value,
@@ -568,4 +588,21 @@ TYPED_TEST(ConnectionTest, TestSQLSetStmtAttrDescriptor) {
   EXPECT_EQ(internal_ard, value);
 }
 
+TYPED_TEST(ConnectionTest, TestExplicitDescriptorsOutliveStatement) {
+  SQLHSTMT statement;
+  SQLHDESC apd_descriptor, ard_descriptor;
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_STMT, this->conn, &statement));
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_DESC, this->conn, &apd_descriptor));
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_DESC, this->conn, &ard_descriptor));
+
+  ASSERT_EQ(SQL_SUCCESS,
+            SQLSetStmtAttr(statement, SQL_ATTR_APP_PARAM_DESC, apd_descriptor, 0));
+  ASSERT_EQ(SQL_SUCCESS,
+            SQLSetStmtAttr(statement, SQL_ATTR_APP_ROW_DESC, ard_descriptor, 0));
+
+  ASSERT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_STMT, statement));
+  EXPECT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, apd_descriptor));
+  EXPECT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_DESC, ard_descriptor));
+}
+
 }  // namespace arrow::flight::sql::odbc

cpp/src/arrow/flight/sql/odbc/tests/statement_driver_api_test.cc

@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+#include "arrow/flight/sql/odbc/tests/odbc_test_suite.h"
+
+#include "arrow/flight/sql/odbc/odbc_api_internal.h"
+#include "arrow/flight/sql/odbc/odbc_impl/odbc_statement.h"
+
+#include <cstdint>
+#include <string_view>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+namespace arrow::flight::sql::odbc {
+
+class StatementDriverApiTest : public FlightSQLODBCMockTestBase {
+ protected:
+  void SetUp() override {
+    FlightSQLODBCMockTestBase::SetUp();
+
+    ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLAllocHandle(
+                               SQL_HANDLE_ENV, SQL_NULL_HANDLE, &driver_env_));
+    ASSERT_EQ(SQL_SUCCESS,
+              arrow::flight::sql::odbc::SQLSetEnvAttr(
+                  driver_env_, SQL_ATTR_ODBC_VERSION,
+                  reinterpret_cast<SQLPOINTER>(static_cast<intptr_t>(SQL_OV_ODBC3)), 0));
+    ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLAllocHandle(
+                               SQL_HANDLE_DBC, driver_env_, &driver_conn_));
+
+    const std::string connect_string = GetConnectionString();
+    std::vector<SQLWCHAR> wide_connect_string(connect_string.begin(),
+                                              connect_string.end());
+    SQLWCHAR output[kOdbcBufferSize];
+    SQLSMALLINT output_length;
+    ASSERT_EQ(SQL_SUCCESS,
+              arrow::flight::sql::odbc::SQLDriverConnect(
+                  driver_conn_, nullptr, wide_connect_string.data(),
+                  static_cast<SQLSMALLINT>(wide_connect_string.size()), output,
+                  kOdbcBufferSize, &output_length, SQL_DRIVER_NOPROMPT));
+    driver_connected_ = true;
+
+    ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLAllocHandle(
+                               SQL_HANDLE_STMT, driver_conn_, &driver_stmt_));
+  }
+
+  void TearDown() override {
+    if (driver_stmt_ != SQL_NULL_HSTMT) {
+      EXPECT_EQ(SQL_SUCCESS,
+                arrow::flight::sql::odbc::SQLFreeHandle(SQL_HANDLE_STMT, driver_stmt_));
+    }
+    if (driver_connected_) {
+      EXPECT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLDisconnect(driver_conn_));
+    }
+    if (driver_conn_ != SQL_NULL_HDBC) {
+      EXPECT_EQ(SQL_SUCCESS,
+                arrow::flight::sql::odbc::SQLFreeHandle(SQL_HANDLE_DBC, driver_conn_));
+    }
+    if (driver_env_ != SQL_NULL_HENV) {
+      EXPECT_EQ(SQL_SUCCESS,
+                arrow::flight::sql::odbc::SQLFreeHandle(SQL_HANDLE_ENV, driver_env_));
+    }
+    FlightSQLODBCMockTestBase::TearDown();
+  }
+
+  void VerifyDriverErrorState(std::string_view expected_state) {
+    ODBC::ODBCStatement* statement = ODBC::ODBCStatement::Of(driver_stmt_);
+    ASSERT_TRUE(statement->GetDiagnostics().HasRecord(0));
+    EXPECT_EQ(expected_state, statement->GetDiagnostics().GetSQLState(0));
+  }
+
+  SQLHENV driver_env_ = SQL_NULL_HENV;
+  SQLHDBC driver_conn_ = SQL_NULL_HDBC;
+  SQLHSTMT driver_stmt_ = SQL_NULL_HSTMT;
+  bool driver_connected_ = false;
+};
+
+TEST_F(StatementDriverApiTest, TestSQLFetchWithoutCursor) {
+  ASSERT_EQ(SQL_ERROR, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  VerifyDriverErrorState(kErrorState24000);
+}
+
+TEST_F(StatementDriverApiTest, TestSQLGetDataWithoutCursor) {
+  SQLINTEGER value;
+  SQLLEN indicator;
+  ASSERT_EQ(SQL_ERROR,
+            arrow::flight::sql::odbc::SQLGetData(driver_stmt_, 1, SQL_C_LONG, &value,
+                                                 sizeof(value), &indicator));
+  VerifyDriverErrorState(kErrorState24000);
+}
+
+TEST_F(StatementDriverApiTest, TestChangingAppRowDescriptorRebindsColumns) {
+  SQLINTEGER implicit_value = -1;
+  SQLLEN implicit_indicator;
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLBindCol(
+                             driver_stmt_, 1, SQL_C_LONG, &implicit_value,
+                             sizeof(implicit_value), &implicit_indicator));
+
+  ASSIGN_SQLWCHAR_ARR_AND_LEN(wsql,
+                              LR"(
+   SELECT 1 AS value
+   UNION ALL SELECT 2
+   UNION ALL SELECT 3
+   UNION ALL SELECT 4
+   UNION ALL SELECT 5;
+ )");
+  ASSERT_EQ(SQL_SUCCESS,
+            arrow::flight::sql::odbc::SQLExecDirect(driver_stmt_, wsql, wsql_len));
+
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  EXPECT_EQ(1, implicit_value);
+
+  SQLHDESC explicit_ard;
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLAllocHandle(
+                             SQL_HANDLE_DESC, driver_conn_, &explicit_ard));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLSetStmtAttr(
+                             driver_stmt_, SQL_ATTR_APP_ROW_DESC, explicit_ard, 0));
+  SQLINTEGER explicit_value = -1;
+  SQLLEN explicit_indicator;
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLBindCol(
+                             driver_stmt_, 1, SQL_C_LONG, &explicit_value,
+                             sizeof(explicit_value), &explicit_indicator));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  EXPECT_EQ(2, explicit_value);
+
+  // Restoring the implicit ARD must rebind the result set to its original buffer.
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLSetStmtAttr(
+                             driver_stmt_, SQL_ATTR_APP_ROW_DESC, SQL_NULL_HANDLE, 0));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  EXPECT_EQ(3, implicit_value);
+  EXPECT_EQ(2, explicit_value);
+  ASSERT_EQ(SQL_SUCCESS,
+            arrow::flight::sql::odbc::SQLFreeHandle(SQL_HANDLE_DESC, explicit_ard));
+
+  // Releasing an active explicit ARD must also restore and rebind the implicit ARD.
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLAllocHandle(
+                             SQL_HANDLE_DESC, driver_conn_, &explicit_ard));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLSetStmtAttr(
+                             driver_stmt_, SQL_ATTR_APP_ROW_DESC, explicit_ard, 0));
+  explicit_value = -1;
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLBindCol(
+                             driver_stmt_, 1, SQL_C_LONG, &explicit_value,
+                             sizeof(explicit_value), &explicit_indicator));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  EXPECT_EQ(4, explicit_value);
+
+  ASSERT_EQ(SQL_SUCCESS,
+            arrow::flight::sql::odbc::SQLFreeHandle(SQL_HANDLE_DESC, explicit_ard));
+  ASSERT_EQ(SQL_SUCCESS, arrow::flight::sql::odbc::SQLFetch(driver_stmt_));
+  EXPECT_EQ(5, implicit_value);
+  EXPECT_EQ(4, explicit_value);
+}
+
+}  // namespace arrow::flight::sql::odbc