[PECOBLR-642] Add header to metadata commands in SEA mode (databricks#1034)

## Description
<!-- Provide a brief summary of the changes made and the issue they aim
to address.-->
Add header to metadata commands in SEA mode

## Testing
<!-- Describe how the changes have been tested-->
Unit + manual testing

## Additional Notes to the Reviewer
<!-- Share any additional context or insights that may help the reviewer
understand the changes better. This could include challenges faced,
limitations, or compromises made during the development process.
Also, mention any areas of the code that you would like the reviewer to
focus on specifically. -->

NO_CHANGELOG=true

---------

Signed-off-by: Vikrant Puppala <vikrant.puppala@databricks.com>

Author: vikrantpuppala

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

@@ -1077,4 +1077,9 @@ public TelemetryLogLevel getTelemetryLogLevel() {
     return TelemetryLogLevel.parse(
         getParameter(DatabricksJdbcUrlParams.TELEMETRY_LOG_LEVEL), TelemetryLogLevel.DEBUG);
   }
+
+  @Override
+  public boolean isSeaSyncMetadataEnabled() {
+    return getParameter(DatabricksJdbcUrlParams.ENABLE_SEA_SYNC_METADATA).equals("1");
+  }
 }

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

@@ -390,4 +390,10 @@ public interface IDatabricksConnectionContext {
 
   /* Returns whether metric view metadata is enabled */
   boolean getEnableMetricViewMetadata();
+
+  /**
+   * Returns whether the x-databricks-sea-can-run-fully-sync header should be enabled for
+   * synchronous metadata requests in SEA mode
+   */
+  boolean isSeaSyncMetadataEnabled();
 }

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

@@ -167,7 +167,12 @@ public enum DatabricksJdbcUrlParams {
   IGNORE_TRANSACTIONS("IgnoreTransactions", "Ignore transaction-related method calls", "0"),
   ENABLE_METRIC_VIEW_METADATA("EnableMetricViewMetadata", "Enable metric view metadata", "0"),
   ENABLE_MULTIPLE_CATALOG_SUPPORT(
-      "enableMultipleCatalogSupport", "Enable multiple catalog support", "1");
+      "enableMultipleCatalogSupport", "Enable multiple catalog support", "1"),
+  ENABLE_SEA_SYNC_METADATA(
+      "EnableSeaSyncMetadata",
+      "Enable x-databricks-sea-can-run-fully-sync header for synchronous metadata requests in SEA mode",
+      "1");
+
   private final String paramName;
   private final String defaultValue;
   private final String description;

src/main/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClient.java

@@ -192,7 +192,7 @@ public DatabricksResultSet executeStatement(
     ExecuteStatementResponse response;
     try {
       Request req = new Request(Request.POST, STATEMENT_PATH, apiClient.serialize(request));
-      req.withHeaders(getHeaders("executeStatement"));
+      req.withHeaders(getHeaders("executeStatement", statementType, false));
       response = apiClient.execute(req, ExecuteStatementResponse.class);
     } catch (IOException e) {
       String errorMessage = "Error while processing the execute statement request";
@@ -299,9 +299,10 @@ public DatabricksResultSet executeStatementAsync(
         session,
         parentStatement);
     DatabricksThreadContextHolder.setSessionId(session.getSessionId());
+    StatementType statementType = StatementType.SQL;
     ExecuteStatementRequest request =
         getRequest(
-            StatementType.SQL,
+            statementType,
             sql,
             ((Warehouse) computeResource).getWarehouseId(),
             session,
@@ -311,7 +312,7 @@ public DatabricksResultSet executeStatementAsync(
     ExecuteStatementResponse response;
     try {
       Request req = new Request(Request.POST, STATEMENT_PATH, apiClient.serialize(request));
-      req.withHeaders(getHeaders("executeStatement"));
+      req.withHeaders(getHeaders("executeStatement", statementType, true));
       response = apiClient.execute(req, ExecuteStatementResponse.class);
     } catch (IOException e) {
       String errorMessage = "Error while processing the execute statement async request";
@@ -335,7 +336,7 @@ public DatabricksResultSet executeStatementAsync(
         typedStatementId,
         response.getResult(),
         response.getManifest(),
-        StatementType.SQL,
+        statementType,
         session,
         parentStatement);
   }
@@ -457,18 +458,58 @@ private boolean useCloudFetchForResult(StatementType statementType) {
   }
 
   private Map<String, String> getHeaders(String method) {
+    return getHeaders(method, null, false);
+  }
+
+  private Map<String, String> getHeaders(
+      String method, StatementType statementType, boolean isAsync) {
     Map<String, String> headers = new HashMap<>(JSON_HTTP_HEADERS);
     if (connectionContext.isRequestTracingEnabled()) {
       String traceHeader = TracingUtil.getTraceHeader();
       LOGGER.debug("Tracing header for method {}: [{}]", method, traceHeader);
       headers.put(TracingUtil.TRACE_HEADER, traceHeader);
     }
 
+    // Add SEA sync metadata header when appropriate
+    if (shouldAddSeaSyncMetadataHeader(statementType, isAsync)) {
+      headers.put("x-databricks-sea-can-run-fully-sync", "true");
+      LOGGER.debug(
+          "Adding x-databricks-sea-can-run-fully-sync header for synchronous metadata request");
+    }
+
     // Overriding with URL defined headers
     headers.putAll(this.connectionContext.getCustomHeaders());
     return headers;
   }
 
+  /**
+   * Determines whether the x-databricks-sea-can-run-fully-sync header should be added to the
+   * request.
+   *
+   * <p>This header is only added when all of the following conditions are met:
+   *
+   * <ul>
+   *   <li>The EnableSeaSyncMetadata URL parameter is enabled (default: true)
+   *   <li>The statement type is METADATA
+   *   <li>The execution mode is synchronous (not async)
+   * </ul>
+   *
+   * <p>The header signals to the server that the metadata operation can be executed fully
+   * synchronously in SEA mode, enabling optimized execution paths.
+   *
+   * <p>Note: This client is only used for SEA mode, so no client type check is needed.
+   *
+   * @param statementType the type of statement being executed (e.g., METADATA, QUERY, SQL)
+   * @param isAsync true if the statement is being executed asynchronously, false for synchronous
+   *     execution
+   * @return true if the header should be added, false otherwise
+   */
+  private boolean shouldAddSeaSyncMetadataHeader(StatementType statementType, boolean isAsync) {
+    return connectionContext.isSeaSyncMetadataEnabled()
+        && statementType == StatementType.METADATA
+        && !isAsync;
+  }
+
   private ExecuteStatementRequest getRequest(
       StatementType statementType,
       String sql,

src/test/java/com/databricks/jdbc/dbclient/impl/sqlexec/DatabricksSdkClientTest.java

@@ -589,4 +589,137 @@ private StatementParameterListItem getParam(String type, String value, int ordin
         .setType(type)
         .setValue(value);
   }
+
+  @Test
+  public void testSeaSyncMetadataHeaderIsAdded() throws Exception {
+    // Test that x-databricks-sea-can-run-fully-sync header is added for SEA + METADATA + sync
+    setupClientMocks(true, false);
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(JDBC_URL, new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    // Execute a metadata request (synchronous)
+    databricksSdkClient.executeStatement(
+        "SHOW CATALOGS",
+        warehouse,
+        new HashMap<>(),
+        StatementType.METADATA,
+        connection.getSession(),
+        statement);
+
+    // Verify that the request was made with the correct header
+    verify(apiClient, atLeastOnce())
+        .execute(
+            argThat(
+                req -> {
+                  Map<String, String> headers = req.getHeaders();
+                  return headers != null
+                      && "true".equals(headers.get("x-databricks-sea-can-run-fully-sync"));
+                }),
+            eq(ExecuteStatementResponse.class));
+  }
+
+  @Test
+  public void testSeaSyncMetadataHeaderNotAddedForAsyncExecution() throws Exception {
+    // Test that header is NOT added for async execution
+    setupClientMocks(false, true);
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(JDBC_URL, new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    // Execute an async statement
+    databricksSdkClient.executeStatementAsync(
+        "SELECT * FROM table", warehouse, new HashMap<>(), connection.getSession(), statement);
+
+    // Verify that the request was made WITHOUT the header
+    verify(apiClient)
+        .execute(
+            argThat(
+                req -> {
+                  Map<String, String> headers = req.getHeaders();
+                  return headers == null
+                      || !headers.containsKey("x-databricks-sea-can-run-fully-sync");
+                }),
+            eq(ExecuteStatementResponse.class));
+  }
+
+  @Test
+  public void testSeaSyncMetadataHeaderNotAddedForQueryType() throws Exception {
+    // Test that header is NOT added for non-METADATA statement types
+    setupClientMocks(true, false);
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(JDBC_URL, new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    // Execute a regular query (not metadata)
+    databricksSdkClient.executeStatement(
+        "SELECT * FROM table",
+        warehouse,
+        new HashMap<>(),
+        StatementType.QUERY,
+        connection.getSession(),
+        statement);
+
+    // Verify that the request was made WITHOUT the header
+    verify(apiClient, atLeastOnce())
+        .execute(
+            argThat(
+                req -> {
+                  Map<String, String> headers = req.getHeaders();
+                  return headers == null
+                      || !headers.containsKey("x-databricks-sea-can-run-fully-sync");
+                }),
+            eq(ExecuteStatementResponse.class));
+  }
+
+  @Test
+  public void testSeaSyncMetadataHeaderNotAddedWhenDisabled() throws Exception {
+    // Test that header is NOT added when the URL parameter is disabled
+    setupClientMocks(true, false);
+    String urlWithDisabledFlag =
+        "jdbc:databricks://sample-host.18.azuredatabricks.net:4423/default;transportMode=http;ssl=1;AuthMech=3;httpPath=/sql/1.0/warehouses/99999999;EnableSeaSyncMetadata=0;";
+    IDatabricksConnectionContext connectionContext =
+        DatabricksConnectionContext.parse(urlWithDisabledFlag, new Properties());
+    DatabricksSdkClient databricksSdkClient =
+        new DatabricksSdkClient(connectionContext, statementExecutionService, apiClient);
+    DatabricksConnection connection =
+        new DatabricksConnection(connectionContext, databricksSdkClient);
+    connection.open();
+    DatabricksStatement statement = new DatabricksStatement(connection);
+
+    // Execute a metadata request (should NOT add header because flag is disabled)
+    databricksSdkClient.executeStatement(
+        "SHOW CATALOGS",
+        warehouse,
+        new HashMap<>(),
+        StatementType.METADATA,
+        connection.getSession(),
+        statement);
+
+    // Verify that the request was made WITHOUT the header
+    verify(apiClient, atLeastOnce())
+        .execute(
+            argThat(
+                req -> {
+                  Map<String, String> headers = req.getHeaders();
+                  return headers == null
+                      || !headers.containsKey("x-databricks-sea-can-run-fully-sync");
+                }),
+            eq(ExecuteStatementResponse.class));
+  }
 }