Add Thrift-native metadata support

NEXT_CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 ### Added
+- Added `EnableThriftNativeMetadata` to request and consume supported Thrift-native SEA metadata results.
 
 ### Updated
 - `DatabaseMetaData.getColumns(...)` with a `null` catalog now issues a single `SHOW COLUMNS IN ALL CATALOGS` statement (consistent with `getSchemas`/`getTables`) instead of enumerating every catalog and issuing a per-catalog `SHOW COLUMNS`. Older DBR versions that do not support the syntax transparently fall back to the previous enumerate-and-fan-out behavior.

src/main/java/com/databricks/jdbc/api/impl/DatabricksConnectionContext.java

@@ -1511,6 +1511,11 @@ public boolean isSeaSyncMetadataEnabled() {
     return getParameter(DatabricksJdbcUrlParams.ENABLE_SEA_SYNC_METADATA).equals("1");
   }
 
+  @Override
+  public boolean isThriftNativeMetadataEnabled() {
+    return getParameter(DatabricksJdbcUrlParams.ENABLE_THRIFT_NATIVE_METADATA).equals("1");
+  }
+
   @Override
   public boolean getDisableOauthRefreshToken() {
     return getParameter(DatabricksJdbcUrlParams.DISABLE_OAUTH_REFRESH_TOKEN, "1").equals("1");

src/main/java/com/databricks/jdbc/api/impl/DatabricksResultSet.java

@@ -81,6 +81,7 @@ enum ResultSetType {
   // Set to true when next() returns false for the bounded-SEA path, so that isAfterLast()
   // returns true only after the cursor has moved PAST the last row (not while ON it).
   private boolean boundedSeaExhausted = false;
+  private boolean thriftNativeMetadataResult = false;
 
   // Cached telemetry collector resolved once at construction time to avoid
   // per-row overhead in next(). The connection-to-collector mapping is stable
@@ -109,6 +110,8 @@ public DatabricksResultSet(
       throws SQLException {
     this.executionStatus = new ExecutionStatus(statementStatus);
     this.statementId = statementId;
+    this.thriftNativeMetadataResult =
+        resultManifest != null && Boolean.TRUE.equals(resultManifest.getIsNativeMetadataResult());
     if (resultData != null) {
       this.executionResult =
           ExecutionResultFactory.getResultSet(
@@ -803,6 +806,10 @@ public ResultSetMetaData getMetaData() throws SQLException {
     return resultSetMetaData;
   }
 
+  public boolean isThriftNativeMetadataResult() {
+    return thriftNativeMetadataResult;
+  }
+
   /**
    * Checks if the given type name represents a geospatial type (GEOMETRY or GEOGRAPHY).
    *

src/main/java/com/databricks/jdbc/api/internal/IDatabricksConnectionContext.java

@@ -443,6 +443,9 @@ default int getHeartbeatIntervalSeconds() {
    */
   boolean isSeaSyncMetadataEnabled();
 
+  /** Returns whether SEA metadata requests should require Thrift-native execution. */
+  boolean isThriftNativeMetadataEnabled();
+
   /** Returns whether OAuth refresh tokens should be disabled (omit offline_access by default). */
   boolean getDisableOauthRefreshToken();
 

src/main/java/com/databricks/jdbc/common/DatabricksJdbcConstants.java

@@ -107,9 +107,16 @@ public final class DatabricksJdbcConstants {
   public static final String AAD_CLIENT_ID = "databricks-sql-jdbc";
   public static final String GCP_GOOGLE_CREDENTIALS_AUTH_TYPE = "google-credentials";
   public static final String GCP_GOOGLE_ID_AUTH_TYPE = "google-id";
-  public static final String DEFAULT_HTTP_EXCEPTION_SQLSTATE = "08000";
+
+  /** SQL state used by Thrift for generic operation errors (SQLSTATE 08000). */
+  public static final String OPERATION_ERROR_SQLSTATE = "08000";
+
+  public static final String DEFAULT_HTTP_EXCEPTION_SQLSTATE = OPERATION_ERROR_SQLSTATE;
   public static final String QUERY_EXECUTION_TIMEOUT_SQLSTATE = "57KD0";
 
+  /** Standard SQL state for syntax error or access rule violation (SQLSTATE 42000). */
+  public static final String SYNTAX_OR_ACCESS_VIOLATION_SQLSTATE = "42000";
+
   /** Standard SQL state for operation cancelled (SQLSTATE HY008). */
   public static final String OPERATION_CANCELLED_SQLSTATE = "HY008";
 

src/main/java/com/databricks/jdbc/common/DatabricksJdbcUrlParams.java

@@ -215,6 +215,10 @@ public enum DatabricksJdbcUrlParams {
       "UseBoundedSeaApi",
       "Use bounded SEA API for CloudFetch: send row_offset on GetResultData, force StreamingChunkProvider, stop relying on total_chunk_count. Requires server support.",
       "0"),
+  ENABLE_THRIFT_NATIVE_METADATA(
+      "EnableThriftNativeMetadata",
+      "Request Thrift-native SEA results for catalogs, schemas, tables, columns, functions, primary keys, imported keys, and cross references",
+      "0"),
   DISABLE_OAUTH_REFRESH_TOKEN(
       "DisableOauthRefreshToken",
       "Disable requesting OAuth refresh tokens (omit offline_access unless explicitly provided)",

src/main/java/com/databricks/jdbc/common/MetadataOperationType.java

@@ -5,24 +5,30 @@
  * HTTP headers to track which metadata operation is being performed.
  */
 public enum MetadataOperationType {
-  GET_CATALOGS("GetCatalogs"),
-  GET_SCHEMAS("GetSchemas"),
-  GET_TABLES("GetTables"),
-  GET_COLUMNS("GetColumns"),
-  GET_FUNCTIONS("GetFunctions"),
-  GET_PRIMARY_KEYS("GetPrimaryKeys"),
-  GET_CROSS_REFERENCE("GetCrossReference"),
-  GET_PROCEDURES("GetProcedures"),
-  GET_PROCEDURE_COLUMNS("GetProcedureColumns");
+  GET_CATALOGS("GetCatalogs", true),
+  GET_SCHEMAS("GetSchemas", true),
+  GET_TABLES("GetTables", true),
+  GET_COLUMNS("GetColumns", true),
+  GET_FUNCTIONS("GetFunctions", true),
+  GET_PRIMARY_KEYS("GetPrimaryKeys", true),
+  GET_CROSS_REFERENCE("GetCrossReference", true),
+  GET_PROCEDURES("GetProcedures", false),
+  GET_PROCEDURE_COLUMNS("GetProcedureColumns", false);
 
   private final String headerValue;
+  private final boolean thriftNativeSupported;
 
-  MetadataOperationType(String headerValue) {
+  MetadataOperationType(String headerValue, boolean thriftNativeSupported) {
     this.headerValue = headerValue;
+    this.thriftNativeSupported = thriftNativeSupported;
   }
 
   /** Returns the header value to be sent in the HTTP request. */
   public String getHeaderValue() {
     return headerValue;
   }
+
+  public boolean isThriftNativeSupported() {
+    return thriftNativeSupported;
+  }
 }

src/main/java/com/databricks/jdbc/common/util/SqlStateClassifier.java

@@ -2,6 +2,7 @@
 
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.COMMUNICATION_LINK_FAILURE_SQLSTATE;
 import static com.databricks.jdbc.common.DatabricksJdbcConstants.SERIALIZATION_FAILURE_SQLSTATE;
+import static com.databricks.jdbc.common.DatabricksJdbcConstants.SYNTAX_OR_ACCESS_VIOLATION_SQLSTATE;
 
 /**
  * Reclassifies SQL states for known transient or mis-categorized server errors so callers can
@@ -25,9 +26,6 @@
  * regress the classifier.
  */
 public final class SqlStateClassifier {
-
-  private static final String SYNTAX_OR_ACCESS_VIOLATION_SQLSTATE = "42000";
-
   private SqlStateClassifier() {}
 
   /**

src/main/java/com/databricks/jdbc/dbclient/impl/common/CrossReferenceKeysDatabricksResultSetAdapter.java

@@ -40,23 +40,22 @@ public boolean includeRow(ResultSet resultSet, List<ResultColumn> columns) throw
     final ResultColumn parentNamespaceColumn = mapColumn(PKTABLE_SCHEM);
     final ResultColumn parentTableNameColumn = mapColumn(PKTABLE_NAME);
 
-    boolean isParentCatalogMatching =
-        resultSet
-            .getString(parentCatalogNameColumn.getResultSetColumnName())
-            .equalsIgnoreCase(targetParentCatalogName);
-    boolean isParentNamespaceMatching =
-        resultSet
-            .getString(parentNamespaceColumn.getResultSetColumnName())
-            .equalsIgnoreCase(targetParentNamespaceName);
-    boolean isParentTableMatching =
-        resultSet
-            .getString(parentTableNameColumn.getResultSetColumnName())
-            .equalsIgnoreCase(targetParentTableName);
-
-    if (!isParentTableMatching || !isParentCatalogMatching || !isParentNamespaceMatching) {
+    if (!matchesParent(
+        resultSet.getString(parentCatalogNameColumn.getResultSetColumnName()),
+        resultSet.getString(parentNamespaceColumn.getResultSetColumnName()),
+        resultSet.getString(parentTableNameColumn.getResultSetColumnName()))) {
       return false;
     }
 
     return super.includeRow(resultSet, columns);
   }
+
+  boolean matchesParent(String catalog, String schema, String table) {
+    return catalog != null
+        && schema != null
+        && table != null
+        && catalog.equalsIgnoreCase(targetParentCatalogName)
+        && schema.equalsIgnoreCase(targetParentNamespaceName)
+        && table.equalsIgnoreCase(targetParentTableName);
+  }
 }

src/main/java/com/databricks/jdbc/dbclient/impl/common/MetadataResultSetBuilder.java

@@ -502,6 +502,9 @@ public boolean shouldAllowCatalogAccess(
 
   public DatabricksResultSet getFunctionsResult(DatabricksResultSet resultSet, String catalog)
       throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getFunctionsResult(catalog, copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows = getRowsForFunctions(resultSet, FUNCTION_COLUMNS, catalog);
     return buildResultSet(
         FUNCTION_COLUMNS,
@@ -550,6 +553,9 @@ public DatabricksResultSet getProcedureColumnsResult(List<List<Object>> rows) {
   }
 
   public DatabricksResultSet getColumnsResult(DatabricksResultSet resultSet) throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getColumnsResult(copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows = getRows(resultSet, COLUMN_COLUMNS, defaultAdapter);
     return buildResultSet(
         COLUMN_COLUMNS,
@@ -560,6 +566,9 @@ public DatabricksResultSet getColumnsResult(DatabricksResultSet resultSet) throw
   }
 
   public DatabricksResultSet getCatalogsResult(DatabricksResultSet resultSet) throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getCatalogsResult(copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows = getRows(resultSet, CATALOG_COLUMNS, defaultAdapter);
     return buildResultSet(
         CATALOG_COLUMNS,
@@ -571,6 +580,9 @@ public DatabricksResultSet getCatalogsResult(DatabricksResultSet resultSet) thro
 
   public DatabricksResultSet getSchemasResult(DatabricksResultSet resultSet, String catalog)
       throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getSchemasResult(copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows =
         getRowsForSchemas(
             resultSet, SCHEMA_COLUMNS, catalog, new SchemasDatabricksResultSetAdapter());
@@ -582,8 +594,11 @@ public DatabricksResultSet getSchemasResult(DatabricksResultSet resultSet, Strin
         CommandName.LIST_SCHEMAS);
   }
 
-  public DatabricksResultSet getTablesResult(DatabricksResultSet resultSet, String[] tableTypes)
-      throws SQLException {
+  public DatabricksResultSet getTablesResult(
+      DatabricksResultSet resultSet, String catalog, String[] tableTypes) throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getTablesResult(catalog, tableTypes, copyThriftNativeMetadataRows(resultSet));
+    }
     List<String> allowedTableTypes = List.of(tableTypes);
     List<List<Object>> rows =
         getRows(resultSet, TABLE_COLUMNS, defaultAdapter).stream()
@@ -622,6 +637,9 @@ public DatabricksResultSet getTableTypesResult() {
 
   public DatabricksResultSet getPrimaryKeysResult(DatabricksResultSet resultSet)
       throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getPrimaryKeysResult(copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows = getRows(resultSet, PRIMARY_KEYS_COLUMNS, defaultAdapter);
     return buildResultSet(
         PRIMARY_KEYS_COLUMNS,
@@ -633,6 +651,9 @@ public DatabricksResultSet getPrimaryKeysResult(DatabricksResultSet resultSet)
 
   public DatabricksResultSet getImportedKeysResult(DatabricksResultSet resultSet)
       throws SQLException {
+    if (resultSet.isThriftNativeMetadataResult()) {
+      return getImportedKeys(copyThriftNativeMetadataRows(resultSet));
+    }
     List<List<Object>> rows = getRows(resultSet, IMPORTED_KEYS_COLUMNS, importedKeysAdapter);
     return buildResultSet(
         IMPORTED_KEYS_COLUMNS,
@@ -651,6 +672,20 @@ public DatabricksResultSet getCrossReferenceKeysResult(
     final CrossReferenceKeysDatabricksResultSetAdapter crossReferenceKeysResultSetAdapter =
         new CrossReferenceKeysDatabricksResultSetAdapter(
             targetParentCatalogName, targetParentNamespaceName, targetParentTableName);
+    // Cross-reference SQL narrows only the foreign side, so parent filtering remains necessary.
+    if (resultSet.isThriftNativeMetadataResult()) {
+      List<List<Object>> rows = copyThriftNativeMetadataRows(resultSet);
+      int parentCatalogIndex = CROSS_REFERENCE_COLUMNS.indexOf(PKTABLE_CAT);
+      int parentSchemaIndex = CROSS_REFERENCE_COLUMNS.indexOf(PKTABLE_SCHEM);
+      int parentTableIndex = CROSS_REFERENCE_COLUMNS.indexOf(PKTABLE_NAME);
+      rows.removeIf(
+          row ->
+              !crossReferenceKeysResultSetAdapter.matchesParent(
+                  (String) row.get(parentCatalogIndex),
+                  (String) row.get(parentSchemaIndex),
+                  (String) row.get(parentTableIndex)));
+      return getCrossRefsResult(rows);
+    }
     List<List<Object>> rows =
         getRows(resultSet, CROSS_REFERENCE_COLUMNS, crossReferenceKeysResultSetAdapter);
 
@@ -662,6 +697,21 @@ public DatabricksResultSet getCrossReferenceKeysResult(
         CommandName.GET_CROSS_REFERENCE);
   }
 
+  /** Copies native rows for Thrift normalization and JDBC metadata, not just column ordering. */
+  private List<List<Object>> copyThriftNativeMetadataRows(DatabricksResultSet resultSet)
+      throws SQLException {
+    List<List<Object>> rows = new ArrayList<>();
+    int columnCount = resultSet.getMetaData().getColumnCount();
+    while (resultSet.next()) {
+      List<Object> row = new ArrayList<>(columnCount);
+      for (int columnIndex = 1; columnIndex <= columnCount; columnIndex++) {
+        row.add(resultSet.getObject(columnIndex));
+      }
+      rows.add(row);
+    }
+    return rows;
+  }
+
   private boolean isTextType(String typeVal) {
     return (typeVal.contains(TEXT_TYPE)
         || typeVal.contains(CHAR_TYPE)
@@ -1546,7 +1596,7 @@ public DatabricksResultSet getTablesResult(
     List<List<Object>> updatedRows = new ArrayList<>();
     for (List<Object> row : rows) {
       // If the catalog is not null and the catalog does not match, skip the row
-      if (catalog != null && !row.get(0).toString().equals(catalog)) {
+      if (catalog != null && !catalog.equals(row.get(0))) {
         continue;
       }