GH-47710: [C++][FlightRPC] Statement attribute Support in ODBC (#47773)

### Rationale for this change
Support for getting and setting statement attributes in ODBC is added. 

### What changes are included in this PR?
- Implementation of `SQLGetStmtAttr` and  `SQLSetStmtAttr` to get and set statement attributes. 
- Tests

### Are these changes tested?
Tested on local MSVC

### Are there any user-facing changes?
No
* GitHub Issue: #47710

Lead-authored-by: Alina (Xi) Li <alina.li@improving.com>
Co-authored-by: alinalibq <alina.li@improving.com>
Co-authored-by: justing-bq <justin.gossett@improving.com>
Signed-off-by: David Li <li.davidm96@gmail.com>

Author: alinaliBQ

cpp/src/arrow/flight/sql/odbc/odbc_api.cc

@@ -923,17 +923,38 @@ SQLRETURN SQLGetStmtAttr(SQLHSTMT stmt, SQLINTEGER attribute, SQLPOINTER value_p
                    << ", attribute: " << attribute << ", value_ptr: " << value_ptr
                    << ", buffer_length: " << buffer_length << ", string_length_ptr: "
                    << static_cast<const void*>(string_length_ptr);
-  // GH-47710 TODO: Implement SQLGetStmtAttr
-  return SQL_INVALID_HANDLE;
+
+  using ODBC::ODBCStatement;
+
+  return ODBCStatement::ExecuteWithDiagnostics(stmt, SQL_ERROR, [=]() {
+    ODBCStatement* statement = reinterpret_cast<ODBCStatement*>(stmt);
+
+    bool is_unicode = true;
+
+    statement->GetStmtAttr(attribute, value_ptr, buffer_length, string_length_ptr,
+                           is_unicode);
+
+    return SQL_SUCCESS;
+  });
 }
 
 SQLRETURN SQLSetStmtAttr(SQLHSTMT stmt, SQLINTEGER attribute, SQLPOINTER value_ptr,
                          SQLINTEGER string_length) {
   ARROW_LOG(DEBUG) << "SQLSetStmtAttrW called with stmt: " << stmt
                    << ", attribute: " << attribute << ", value_ptr: " << value_ptr
                    << ", string_length: " << string_length;
-  // GH-47710 TODO: Implement SQLSetStmtAttr
-  return SQL_INVALID_HANDLE;
+
+  using ODBC::ODBCStatement;
+
+  return ODBCStatement::ExecuteWithDiagnostics(stmt, SQL_ERROR, [=]() {
+    ODBCStatement* statement = reinterpret_cast<ODBCStatement*>(stmt);
+
+    bool is_unicode = true;
+
+    statement->SetStmtAttr(attribute, value_ptr, string_length, is_unicode);
+
+    return SQL_SUCCESS;
+  });
 }
 
 SQLRETURN SQLExecDirect(SQLHSTMT stmt, SQLWCHAR* query_text, SQLINTEGER text_length) {

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

@@ -375,6 +375,14 @@ void ODBCStatement::GetStmtAttr(SQLINTEGER statement_attribute, SQLPOINTER outpu
       return;
     case SQL_ATTR_PARAM_BIND_TYPE:
       current_apd_->GetHeaderField(SQL_DESC_BIND_TYPE, output, buffer_size, str_len_ptr);
+      if (output) {
+        // Convert SQLINTEGER output to SQLULEN, since SQL_DESC_BIND_TYPE is SQLINTEGER
+        // and SQL_ATTR_PARAM_BIND_TYPE is SQLULEN
+        SQLINTEGER* output_int_ptr = reinterpret_cast<SQLINTEGER*>(output);
+        SQLINTEGER output_int = *output_int_ptr;
+        SQLULEN* typed_output = reinterpret_cast<SQLULEN*>(output);
+        *typed_output = static_cast<SQLULEN>(output_int);
+      }
       return;
     case SQL_ATTR_PARAM_OPERATION_PTR:
       current_apd_->GetHeaderField(SQL_DESC_ARRAY_STATUS_PTR, output, buffer_size,
@@ -398,6 +406,14 @@ void ODBCStatement::GetStmtAttr(SQLINTEGER statement_attribute, SQLPOINTER outpu
       return;
     case SQL_ATTR_ROW_BIND_TYPE:
       current_ard_->GetHeaderField(SQL_DESC_BIND_TYPE, output, buffer_size, str_len_ptr);
+      if (output) {
+        // Convert SQLINTEGER output to SQLULEN, since SQL_DESC_BIND_TYPE is SQLINTEGER
+        // and SQL_ATTR_ROW_BIND_TYPE is SQLULEN
+        SQLINTEGER* output_int_ptr = reinterpret_cast<SQLINTEGER*>(output);
+        SQLINTEGER output_int = *output_int_ptr;
+        SQLULEN* typed_output = reinterpret_cast<SQLULEN*>(output);
+        *typed_output = static_cast<SQLULEN>(output_int);
+      }
       return;
     case SQL_ATTR_ROW_OPERATION_PTR:
       current_ard_->GetHeaderField(SQL_DESC_ARRAY_STATUS_PTR, output, buffer_size,
@@ -627,7 +643,7 @@ void ODBCStatement::SetStmtAttr(SQLINTEGER statement_attribute, SQLPOINTER value
       CheckIfAttributeIsSetToOnlyValidValue(value, static_cast<SQLULEN>(SQL_UB_OFF));
       return;
     case SQL_ATTR_RETRIEVE_DATA:
-      CheckIfAttributeIsSetToOnlyValidValue(value, static_cast<SQLULEN>(SQL_TRUE));
+      CheckIfAttributeIsSetToOnlyValidValue(value, static_cast<SQLULEN>(SQL_RD_ON));
       return;
     case SQL_ROWSET_SIZE:
       SetAttribute(value, rowset_size_);

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

@@ -34,6 +34,7 @@ add_arrow_test(flight_sql_odbc_test
                SOURCES
                odbc_test_suite.cc
                odbc_test_suite.h
+               statement_attr_test.cc
                connection_test.cc
                # Enable Protobuf cleanup after test execution
                # GH-46889: move protobuf_test_util to a more common location

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

@@ -61,13 +61,11 @@ void ODBCRemoteTestBase::ConnectWithString(std::string connect_str) {
                              kOdbcBufferSize, &out_str_len, SQL_DRIVER_NOPROMPT))
       << GetOdbcErrorMessage(SQL_HANDLE_DBC, conn);
 
-  // GH-47710: TODO Allocate a statement using alloc handle
-  // ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_STMT, conn, &stmt));
+  ASSERT_EQ(SQL_SUCCESS, SQLAllocHandle(SQL_HANDLE_STMT, conn, &stmt));
 }
 
 void ODBCRemoteTestBase::Disconnect() {
-  // GH-47710: TODO Close statement
-  // EXPECT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_STMT, stmt));
+  EXPECT_EQ(SQL_SUCCESS, SQLFreeHandle(SQL_HANDLE_STMT, stmt));
 
   // Disconnect from ODBC
   EXPECT_EQ(SQL_SUCCESS, SQLDisconnect(conn))