GH-47721: [C++][FlightRPC] Return ODBC Column Attribute from result set

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

@@ -1088,8 +1088,90 @@ SQLRETURN SQLColAttribute(SQLHSTMT stmt, SQLUSMALLINT record_number,
                    << ", output_length: " << static_cast<const void*>(output_length)
                    << ", numeric_attribute_ptr: "
                    << static_cast<const void*>(numeric_attribute_ptr);
-  // GH-47721 TODO: Implement SQLColAttribute, pre-requisite requires SQLColumns
-  return SQL_INVALID_HANDLE;
+
+  using ODBC::ODBCDescriptor;
+  using ODBC::ODBCStatement;
+  return ODBCStatement::ExecuteWithDiagnostics(stmt, SQL_ERROR, [=]() {
+    ODBCStatement* statement = reinterpret_cast<ODBCStatement*>(stmt);
+    ODBCDescriptor* ird = statement->GetIRD();
+    SQLINTEGER output_length_int;
+    switch (field_identifier) {
+      // Numeric attributes
+      // internal is SQLLEN, no conversion is needed
+      case SQL_DESC_DISPLAY_SIZE:
+      case SQL_DESC_OCTET_LENGTH: {
+        ird->GetField(record_number, field_identifier, numeric_attribute_ptr,
+                      buffer_length, &output_length_int);
+        break;
+      }
+      // internal is SQLULEN, conversion is needed.
+      case SQL_COLUMN_LENGTH:  // ODBC 2.0
+      case SQL_DESC_LENGTH: {
+        SQLULEN temp;
+        ird->GetField(record_number, field_identifier, &temp, buffer_length,
+                      &output_length_int);
+        if (numeric_attribute_ptr) {
+          *numeric_attribute_ptr = static_cast<SQLLEN>(temp);
+        }
+        break;
+      }
+      // internal is SQLINTEGER, conversion is needed.
+      case SQL_DESC_AUTO_UNIQUE_VALUE:
+      case SQL_DESC_CASE_SENSITIVE:
+      case SQL_DESC_NUM_PREC_RADIX: {
+        SQLINTEGER temp;
+        ird->GetField(record_number, field_identifier, &temp, buffer_length,
+                      &output_length_int);
+        if (numeric_attribute_ptr) {
+          *numeric_attribute_ptr = static_cast<SQLLEN>(temp);
+        }
+        break;
+      }
+      // internal is SQLSMALLINT, conversion is needed.
+      case SQL_DESC_CONCISE_TYPE:
+      case SQL_DESC_COUNT:
+      case SQL_DESC_FIXED_PREC_SCALE:
+      case SQL_DESC_TYPE:
+      case SQL_DESC_NULLABLE:
+      case SQL_COLUMN_PRECISION:  // ODBC 2.0
+      case SQL_DESC_PRECISION:
+      case SQL_COLUMN_SCALE:  // ODBC 2.0
+      case SQL_DESC_SCALE:
+      case SQL_DESC_SEARCHABLE:
+      case SQL_DESC_UNNAMED:
+      case SQL_DESC_UNSIGNED:
+      case SQL_DESC_UPDATABLE: {
+        SQLSMALLINT temp;
+        ird->GetField(record_number, field_identifier, &temp, buffer_length,
+                      &output_length_int);
+        if (numeric_attribute_ptr) {
+          *numeric_attribute_ptr = static_cast<SQLLEN>(temp);
+        }
+        break;
+      }
+      // Character attributes
+      case SQL_DESC_BASE_COLUMN_NAME:
+      case SQL_DESC_BASE_TABLE_NAME:
+      case SQL_DESC_CATALOG_NAME:
+      case SQL_DESC_LABEL:
+      case SQL_DESC_LITERAL_PREFIX:
+      case SQL_DESC_LITERAL_SUFFIX:
+      case SQL_DESC_LOCAL_TYPE_NAME:
+      case SQL_DESC_NAME:
+      case SQL_DESC_SCHEMA_NAME:
+      case SQL_DESC_TABLE_NAME:
+      case SQL_DESC_TYPE_NAME:
+        ird->GetField(record_number, field_identifier, character_attribute_ptr,
+                      buffer_length, &output_length_int);
+        break;
+      default:
+        throw DriverException("Invalid descriptor field", "HY091");
+    }
+    if (output_length) {
+      *output_length = static_cast<SQLSMALLINT>(output_length_int);
+    }
+    return SQL_SUCCESS;
+  });
 }
 
 SQLRETURN SQLGetTypeInfo(SQLHSTMT stmt, SQLSMALLINT data_type) {

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

@@ -20,6 +20,8 @@
 #include "arrow/flight/sql/column_metadata.h"
 #include "arrow/flight/sql/odbc/odbc_impl/platform.h"
 #include "arrow/flight/sql/odbc/odbc_impl/util.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/key_value_metadata.h"
 
 #include <utility>
 #include "arrow/flight/sql/odbc/odbc_impl/exceptions.h"
@@ -40,12 +42,8 @@ constexpr int32_t DefaultDecimalPrecision = 38;
 constexpr int32_t DefaultLengthForVariableLengthColumns = 1024;
 
 namespace {
-std::shared_ptr<const KeyValueMetadata> empty_metadata_map(new KeyValueMetadata);
-
 inline ColumnMetadata GetMetadata(const std::shared_ptr<Field>& field) {
-  const auto& metadata_map = field->metadata();
-
-  ColumnMetadata metadata(metadata_map ? metadata_map : empty_metadata_map);
+  ColumnMetadata metadata(field->metadata());
   return metadata;
 }
 
@@ -207,10 +205,14 @@ size_t FlightSqlResultSetMetadata::GetOctetLength(int column_position) {
       .value_or(DefaultLengthForVariableLengthColumns);
 }
 
-std::string FlightSqlResultSetMetadata::GetTypeName(int column_position) {
+std::string FlightSqlResultSetMetadata::GetTypeName(int column_position,
+                                                    int16_t data_type) {
   ColumnMetadata metadata = GetMetadata(schema_->field(column_position - 1));
 
-  return metadata.GetTypeName().ValueOrElse([] { return ""; });
+  return metadata.GetTypeName().ValueOrElse([data_type] {
+    // If we get an empty type name, figure out the type name from the data_type.
+    return util::GetTypeNameFromSqlDataType(data_type);
+  });
 }
 
 Updatability FlightSqlResultSetMetadata::GetUpdatable(int column_position) {
@@ -240,19 +242,11 @@ Searchability FlightSqlResultSetMetadata::IsSearchable(int column_position) {
 bool FlightSqlResultSetMetadata::IsUnsigned(int column_position) {
   const std::shared_ptr<Field>& field = schema_->field(column_position - 1);
 
-  switch (field->type()->id()) {
-    case Type::UINT8:
-    case Type::UINT16:
-    case Type::UINT32:
-    case Type::UINT64:
-      return true;
-    default:
-      return false;
-  }
+  return arrow::is_unsigned_integer(field->type()->id());
 }
 
 bool FlightSqlResultSetMetadata::IsFixedPrecScale(int column_position) {
-  // TODO: Flight SQL column metadata does not have this, should we add to the spec?
+  // Precision for Arrow data types are modifiable by the user
   return false;
 }
 

cpp/src/arrow/flight/sql/odbc/odbc_impl/flight_sql_result_set_metadata.h

@@ -77,7 +77,7 @@ class FlightSqlResultSetMetadata : public ResultSetMetadata {
 
   size_t GetOctetLength(int column_position) override;
 
-  std::string GetTypeName(int column_position) override;
+  std::string GetTypeName(int column_position, int16_t data_type) override;
 
   Updatability GetUpdatable(int column_position) override;
 
@@ -87,6 +87,7 @@ class FlightSqlResultSetMetadata : public ResultSetMetadata {
 
   Searchability IsSearchable(int column_position) override;
 
+  /// \brief Return true if the column is unsigned or not numeric
   bool IsUnsigned(int column_position) override;
 
   bool IsFixedPrecScale(int column_position) override;

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

@@ -479,6 +479,8 @@ void ODBCDescriptor::PopulateFromResultSetMetadata(ResultSetMetadata* rsmd) {
 
   for (size_t i = 0; i < records_.size(); ++i) {
     size_t one_based_index = i + 1;
+    int16_t concise_type = rsmd->GetConciseType(one_based_index);
+
     records_[i].base_column_name = rsmd->GetBaseColumnName(one_based_index);
     records_[i].base_table_name = rsmd->GetBaseTableName(one_based_index);
     records_[i].catalog_name = rsmd->GetCatalogName(one_based_index);
@@ -489,9 +491,8 @@ void ODBCDescriptor::PopulateFromResultSetMetadata(ResultSetMetadata* rsmd) {
     records_[i].name = rsmd->GetName(one_based_index);
     records_[i].schema_name = rsmd->GetSchemaName(one_based_index);
     records_[i].table_name = rsmd->GetTableName(one_based_index);
-    records_[i].type_name = rsmd->GetTypeName(one_based_index);
-    records_[i].concise_type = GetSqlTypeForODBCVersion(
-        rsmd->GetConciseType(one_based_index), is_2x_connection_);
+    records_[i].type_name = rsmd->GetTypeName(one_based_index, concise_type);
+    records_[i].concise_type = GetSqlTypeForODBCVersion(concise_type, is_2x_connection_);
     records_[i].data_ptr = nullptr;
     records_[i].indicator_ptr = nullptr;
     records_[i].display_size = rsmd->GetColumnDisplaySize(one_based_index);

cpp/src/arrow/flight/sql/odbc/odbc_impl/spi/result_set_metadata.h

@@ -17,9 +17,8 @@
 
 #pragma once
 
-#include "arrow/flight/sql/odbc/odbc_impl/types.h"
-
 #include <string>
+#include "arrow/flight/sql/odbc/odbc_impl/types.h"
 
 namespace arrow::flight::sql::odbc {
 
@@ -143,8 +142,9 @@ class ResultSetMetadata {
 
   /// \brief It returns the data type as a string.
   /// \param column_position [in] the position of the column, starting from 1.
+  /// \param data_type [in] the data type of the column.
   /// \return the data type string.
-  virtual std::string GetTypeName(int column_position) = 0;
+  virtual std::string GetTypeName(int column_position, int16_t data_type) = 0;
 
   /// \brief It returns a numeric values indicate the updatability of the
   /// column.

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
+               columns_test.cc
                connection_test.cc
                # Enable Protobuf cleanup after test execution
                # GH-46889: move protobuf_test_util to a more common location