Re: [PR] feat: built-in databricks unity catalog sync [incubator-xtable]

["vinishjail97 (via GitHub)" <gi...@apache.org>]

vinishjail97 commented on code in PR #802:
URL: https://github.com/apache/incubator-xtable/pull/802#discussion_r2897354648


##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogConfig.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.Map;
+
+import lombok.Value;
+
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+
+@Value
+public class DatabricksUnityCatalogConfig {
+  public static final String HOST = "externalCatalog.uc.host";
+  public static final String WAREHOUSE_ID = "externalCatalog.uc.warehouseId";
+  public static final String AUTH_TYPE = "externalCatalog.uc.authType";
+  public static final String CLIENT_ID = "externalCatalog.uc.clientId";
+  public static final String CLIENT_SECRET = "externalCatalog.uc.clientSecret";
+
+  String host;
+  String warehouseId;
+  String authType;
+  String clientId;

Review Comment:
   Lombok `@Value` generates `toString()` that includes **all fields** — `clientSecret` will be printed in any log statement or exception message that calls `.toString()` on this object, which is a credential leak.
   
   Add `@ToString.Exclude` on the `clientSecret` field:
   ```java
   @ToString.Exclude
   String clientSecret;
   ```



##########
xtable-databricks/pom.xml:
##########
@@ -0,0 +1,84 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.xtable</groupId>
+        <artifactId>xtable</artifactId>
+        <version>0.3.0-incubating</version>
+    </parent>
+
+    <artifactId>xtable-databricks</artifactId>
+    <name>XTable Databricks</name>
+
+    <properties>
+        <databricks.sdk.version>0.85.0</databricks.sdk.version>
+    </properties>
+
+    <dependencies>
+        <dependency>

Review Comment:
   `databricks.sdk.version` is declared in `<properties>` here AND in the root `pom.xml`'s `<properties>` block. The child module's declaration is redundant since the version is already managed by the parent. Remove the `<properties>` block from this module's pom.



##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,572 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.Disposition;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.ExecuteStatementRequestOnWaitTimeout;
+import com.databricks.sdk.service.sql.Format;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+
+import org.apache.xtable.catalog.CatalogUtils;
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+/**
+ * Databricks Unity Catalog implementation of {@link CatalogSyncClient}.
+ *
+ * <p>Supports external Delta table sync operations (create, drop, create-or-replace, get table and
+ * schema/database operations) using Databricks SDK APIs and SQL Statement Execution API. Schema
+ * refresh compares desired and existing schemas and triggers metadata sync when differences are
+ * detected.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+  private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
+
+  private ExternalCatalogConfig catalogConfig;
+  private DatabricksUnityCatalogConfig databricksConfig;
+  private String tableFormat;
+  private WorkspaceClient workspaceClient;
+  private StatementExecutionAPI statementExecution;
+  private TablesAPI tablesApi;
+  private SchemasAPI schemasApi;
+
+  /**
+   * For loading the instance using ServiceLoader.
+   *
+   * <p>This constructor requires {@link #init(ExternalCatalogConfig, String, Configuration)} to be
+   * called before invoking catalog operations.
+   */
+  public DatabricksUnityCatalogSyncClient() {}
+
+  public DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig,
+      String tableFormat,
+      Configuration configuration,
+      StatementExecutionAPI statementExecution,
+      TablesAPI tablesApi,
+      SchemasAPI schemasApi) {
+    this.statementExecution = statementExecution;
+    this.tablesApi = tablesApi;
+    this.schemasApi = schemasApi;
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  @Override
+  public String getCatalogId() {
+    ensureInitialized();
+    return catalogConfig.getCatalogId();
+  }
+
+  @Override
+  public String getCatalogType() {
+    return CatalogType.DATABRICKS_UC;
+  }
+
+  @Override
+  public String getStorageLocation(TableInfo table) {
+    if (table == null) {
+      return null;
+    }
+    return table.getStorageLocation();
+  }
+
+  @Override
+  public boolean hasDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    String fullName = catalog + "." + schema;
+    try {
+      SchemaInfo schemaInfo = schemasApi.get(fullName);
+      return schemaInfo != null;
+    } catch (NotFound e) {
+      return false;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get schema: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    try {
+      schemasApi.create(new CreateSchema().setCatalogName(catalog).setName(schema));
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to create database: " + schema, e);
+    }
+  }
+
+  @Override
+  public TableInfo getTable(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      return tablesApi.get(fullName);
+    } catch (NotFound e) {
+      return null;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    String location = table.getBasePath();
+    if (StringUtils.isBlank(location)) {
+      throw new CatalogSyncException("Storage location is required for external Delta tables");
+    }
+
+    String statement =
+        String.format(
+            "CREATE TABLE IF NOT EXISTS %s USING DELTA LOCATION '%s'",
+            fullName, escapeSqlString(location));
+    log.info("Databricks UC create table: {}", fullName);
+    executeStatement(statement);
+  }
+
+  @Override
+  public void refreshTable(
+      InternalTable table, TableInfo catalogTable, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    if (catalogTable == null) {
+      log.warn(
+          "Databricks UC refreshTable called with null catalog table for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    InternalSchema schema = table.getReadSchema();
+    if (schema == null || schema.getFields() == null || schema.getFields().isEmpty()) {
+      log.warn(
+          "Databricks UC refreshTable skipped due to missing schema for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    if (!schemasMatch(schema, catalogTable)) {
+      String fullName = getFullName(tableIdentifier);
+      log.info("Databricks UC refresh table metadata (MSCK REPAIR TABLE): {}", fullName);
+      executeStatement(String.format("MSCK REPAIR TABLE %s SYNC METADATA", fullName));
+    } else {
+      log.info(
+          "Databricks UC refreshTable: schema already up to date for {}", tableIdentifier.getId());
+    }
+  }
+
+  @Override
+  public void createOrReplaceTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    dropTable(table, tableIdentifier);
+    try {
+      createTable(table, tableIdentifier);
+    } catch (Exception e) {
+      log.error(
+          "Databricks UC createOrReplaceTable failed to recreate {} after drop. "
+              + "Table was dropped but not recreated.",
+          fullName,
+          e);
+      throw new CatalogSyncException(
+          "Databricks UC createOrReplaceTable failed after drop for "
+              + fullName
+              + ": table was dropped but not recreated",
+          e);
+    }
+  }
+
+  @Override
+  public void dropTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      log.info("Databricks UC drop table: {}", fullName);
+      tablesApi.delete(fullName);
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to drop table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void init(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    this.catalogConfig = catalogConfig;
+    this.tableFormat = tableFormat;
+    this.databricksConfig = DatabricksUnityCatalogConfig.from(catalogConfig);
+
+    if (databricksConfig.getHost() == null || databricksConfig.getWarehouseId() == null) {
+      throw new CatalogSyncException(
+          "Databricks UC catalog requires host and warehouseId in catalogProperties");
+    }
+    if (shouldInitializeWorkspaceClient()) {
+      this.workspaceClient = new WorkspaceClient(buildConfig(databricksConfig));
+    }
+    if (this.statementExecution == null) {
+      this.statementExecution = workspaceClient.statementExecution();
+    }
+    if (this.tablesApi == null) {
+      this.tablesApi = this.workspaceClient.tables();
+    }
+    if (this.schemasApi == null) {
+      this.schemasApi = this.workspaceClient.schemas();
+    }
+    log.info(
+        "Initialized Databricks UC sync client for catalogId={} tableFormat={}",
+        catalogConfig.getCatalogId(),
+        tableFormat);
+  }
+
+  private boolean shouldInitializeWorkspaceClient() {
+    boolean missingApis =
+        this.statementExecution == null || this.tablesApi == null || this.schemasApi == null;
+    return this.workspaceClient == null && missingApis;
+  }
+
+  private void ensureInitialized() {
+    if (this.catalogConfig == null
+        || this.databricksConfig == null
+        || this.tableFormat == null
+        || this.statementExecution == null
+        || this.tablesApi == null
+        || this.schemasApi == null) {
+      throw new CatalogSyncException(
+          "DatabricksUnityCatalogSyncClient is not initialized. "
+              + "Call init(catalogConfig, tableFormat, configuration) before use.");
+    }
+  }
+
+  private void ensureDeltaOnly() {
+    if (!Objects.equals(tableFormat, TableFormat.DELTA)) {
+      throw new CatalogSyncException(
+          "Databricks UC sync client currently supports external DELTA only");
+    }
+  }
+
+  private String getFullName(CatalogTableIdentifier tableIdentifier) {
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    String table = hierarchical.getTableName();
+    if (StringUtils.isBlank(table)) {
+      throw new CatalogSyncException("Databricks UC requires a table name");
+    }
+
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+    validateIdentifierComponent("table", table);
+    return hierarchical.getId();
+  }
+
+  private static void validateIdentifierComponent(String componentName, String value) {
+    if (!IDENTIFIER_PATTERN.matcher(value).matches()) {
+      throw new CatalogSyncException(
+          String.format(
+              "Invalid Databricks UC %s name '%s'. Only [a-zA-Z0-9_]+ is allowed.",
+              componentName, value));
+    }
+  }
+
+  private StatementResponse executeStatement(String statement) {
+    ExecuteStatementRequest request =
+        new ExecuteStatementRequest()
+            .setStatement(statement)
+            .setWarehouseId(databricksConfig.getWarehouseId())
+            .setFormat(Format.JSON_ARRAY)
+            .setDisposition(Disposition.INLINE)
+            .setWaitTimeout("30s")
+            .setOnWaitTimeout(ExecuteStatementRequestOnWaitTimeout.CANCEL);
+
+    StatementResponse response = statementExecution.executeStatement(request);
+    if (response.getStatus() != null && response.getStatus().getState() == StatementState.FAILED) {

Review Comment:
   `executeStatement` only checks `StatementState.FAILED`. When the 30s wait timeout fires, the SDK cancels the statement and returns `StatementState.CANCELLED` — not `FAILED`. This means a timed-out DDL (CREATE TABLE / DROP / MSCK REPAIR) silently returns without throwing, leaving the caller believing the operation succeeded.
   
   Change the check to cover all non-success states:
   ```java
   StatementState state = response.getStatus() == null ? null : response.getStatus().getState();
   if (state != null && state != StatementState.SUCCEEDED) {
       String errorMsg = (response.getStatus().getError() != null)
           ? response.getStatus().getError().getMessage() : null;
       throw new CatalogSyncException("Databricks UC statement " + state + ": " + statement
           + (StringUtils.isBlank(errorMsg) ? "" : " (" + errorMsg + ")"));
   }
   ```



##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,572 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.Disposition;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.ExecuteStatementRequestOnWaitTimeout;
+import com.databricks.sdk.service.sql.Format;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+
+import org.apache.xtable.catalog.CatalogUtils;
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+/**
+ * Databricks Unity Catalog implementation of {@link CatalogSyncClient}.
+ *
+ * <p>Supports external Delta table sync operations (create, drop, create-or-replace, get table and
+ * schema/database operations) using Databricks SDK APIs and SQL Statement Execution API. Schema
+ * refresh compares desired and existing schemas and triggers metadata sync when differences are
+ * detected.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+  private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
+
+  private ExternalCatalogConfig catalogConfig;
+  private DatabricksUnityCatalogConfig databricksConfig;
+  private String tableFormat;
+  private WorkspaceClient workspaceClient;
+  private StatementExecutionAPI statementExecution;
+  private TablesAPI tablesApi;
+  private SchemasAPI schemasApi;
+
+  /**
+   * For loading the instance using ServiceLoader.
+   *
+   * <p>This constructor requires {@link #init(ExternalCatalogConfig, String, Configuration)} to be
+   * called before invoking catalog operations.
+   */
+  public DatabricksUnityCatalogSyncClient() {}
+
+  public DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig,
+      String tableFormat,
+      Configuration configuration,
+      StatementExecutionAPI statementExecution,
+      TablesAPI tablesApi,
+      SchemasAPI schemasApi) {
+    this.statementExecution = statementExecution;
+    this.tablesApi = tablesApi;
+    this.schemasApi = schemasApi;
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  @Override
+  public String getCatalogId() {
+    ensureInitialized();
+    return catalogConfig.getCatalogId();
+  }
+
+  @Override
+  public String getCatalogType() {
+    return CatalogType.DATABRICKS_UC;
+  }
+
+  @Override
+  public String getStorageLocation(TableInfo table) {
+    if (table == null) {
+      return null;
+    }
+    return table.getStorageLocation();
+  }
+
+  @Override
+  public boolean hasDatabase(CatalogTableIdentifier tableIdentifier) {

Review Comment:
   The catalog+schema validation block (parse `HierarchicalTableIdentifier`, blank-check catalog, blank-check schema, call `validateIdentifierComponent` twice) is duplicated verbatim in `hasDatabase` (lines 129–141) and `createDatabase` (lines 157–169), and partially again inside `getFullName`. Consider extracting a private `validateAndGetHierarchical(CatalogTableIdentifier)` helper that returns the parsed identifier and throws on any validation failure, then have all three methods delegate to it.



##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,572 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.Disposition;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.ExecuteStatementRequestOnWaitTimeout;
+import com.databricks.sdk.service.sql.Format;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+
+import org.apache.xtable.catalog.CatalogUtils;
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+/**
+ * Databricks Unity Catalog implementation of {@link CatalogSyncClient}.
+ *
+ * <p>Supports external Delta table sync operations (create, drop, create-or-replace, get table and
+ * schema/database operations) using Databricks SDK APIs and SQL Statement Execution API. Schema
+ * refresh compares desired and existing schemas and triggers metadata sync when differences are
+ * detected.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+  private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
+
+  private ExternalCatalogConfig catalogConfig;
+  private DatabricksUnityCatalogConfig databricksConfig;
+  private String tableFormat;
+  private WorkspaceClient workspaceClient;
+  private StatementExecutionAPI statementExecution;
+  private TablesAPI tablesApi;
+  private SchemasAPI schemasApi;
+
+  /**
+   * For loading the instance using ServiceLoader.
+   *
+   * <p>This constructor requires {@link #init(ExternalCatalogConfig, String, Configuration)} to be
+   * called before invoking catalog operations.
+   */
+  public DatabricksUnityCatalogSyncClient() {}
+
+  public DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig,
+      String tableFormat,
+      Configuration configuration,
+      StatementExecutionAPI statementExecution,
+      TablesAPI tablesApi,
+      SchemasAPI schemasApi) {
+    this.statementExecution = statementExecution;
+    this.tablesApi = tablesApi;
+    this.schemasApi = schemasApi;
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  @Override
+  public String getCatalogId() {
+    ensureInitialized();
+    return catalogConfig.getCatalogId();
+  }
+
+  @Override
+  public String getCatalogType() {
+    return CatalogType.DATABRICKS_UC;
+  }
+
+  @Override
+  public String getStorageLocation(TableInfo table) {
+    if (table == null) {
+      return null;
+    }
+    return table.getStorageLocation();
+  }
+
+  @Override
+  public boolean hasDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    String fullName = catalog + "." + schema;
+    try {
+      SchemaInfo schemaInfo = schemasApi.get(fullName);
+      return schemaInfo != null;
+    } catch (NotFound e) {
+      return false;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get schema: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    try {
+      schemasApi.create(new CreateSchema().setCatalogName(catalog).setName(schema));
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to create database: " + schema, e);
+    }
+  }
+
+  @Override
+  public TableInfo getTable(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      return tablesApi.get(fullName);
+    } catch (NotFound e) {
+      return null;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    String location = table.getBasePath();
+    if (StringUtils.isBlank(location)) {
+      throw new CatalogSyncException("Storage location is required for external Delta tables");
+    }
+
+    String statement =
+        String.format(
+            "CREATE TABLE IF NOT EXISTS %s USING DELTA LOCATION '%s'",
+            fullName, escapeSqlString(location));
+    log.info("Databricks UC create table: {}", fullName);
+    executeStatement(statement);
+  }
+
+  @Override
+  public void refreshTable(
+      InternalTable table, TableInfo catalogTable, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    if (catalogTable == null) {
+      log.warn(
+          "Databricks UC refreshTable called with null catalog table for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    InternalSchema schema = table.getReadSchema();
+    if (schema == null || schema.getFields() == null || schema.getFields().isEmpty()) {
+      log.warn(
+          "Databricks UC refreshTable skipped due to missing schema for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    if (!schemasMatch(schema, catalogTable)) {
+      String fullName = getFullName(tableIdentifier);
+      log.info("Databricks UC refresh table metadata (MSCK REPAIR TABLE): {}", fullName);
+      executeStatement(String.format("MSCK REPAIR TABLE %s SYNC METADATA", fullName));
+    } else {
+      log.info(
+          "Databricks UC refreshTable: schema already up to date for {}", tableIdentifier.getId());
+    }
+  }
+
+  @Override
+  public void createOrReplaceTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    dropTable(table, tableIdentifier);
+    try {
+      createTable(table, tableIdentifier);
+    } catch (Exception e) {
+      log.error(
+          "Databricks UC createOrReplaceTable failed to recreate {} after drop. "
+              + "Table was dropped but not recreated.",
+          fullName,
+          e);
+      throw new CatalogSyncException(
+          "Databricks UC createOrReplaceTable failed after drop for "
+              + fullName
+              + ": table was dropped but not recreated",
+          e);
+    }
+  }
+
+  @Override
+  public void dropTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      log.info("Databricks UC drop table: {}", fullName);
+      tablesApi.delete(fullName);
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to drop table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void init(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    this.catalogConfig = catalogConfig;
+    this.tableFormat = tableFormat;
+    this.databricksConfig = DatabricksUnityCatalogConfig.from(catalogConfig);
+
+    if (databricksConfig.getHost() == null || databricksConfig.getWarehouseId() == null) {
+      throw new CatalogSyncException(
+          "Databricks UC catalog requires host and warehouseId in catalogProperties");
+    }
+    if (shouldInitializeWorkspaceClient()) {
+      this.workspaceClient = new WorkspaceClient(buildConfig(databricksConfig));
+    }
+    if (this.statementExecution == null) {
+      this.statementExecution = workspaceClient.statementExecution();
+    }
+    if (this.tablesApi == null) {
+      this.tablesApi = this.workspaceClient.tables();
+    }
+    if (this.schemasApi == null) {
+      this.schemasApi = this.workspaceClient.schemas();
+    }
+    log.info(
+        "Initialized Databricks UC sync client for catalogId={} tableFormat={}",
+        catalogConfig.getCatalogId(),
+        tableFormat);
+  }
+
+  private boolean shouldInitializeWorkspaceClient() {
+    boolean missingApis =
+        this.statementExecution == null || this.tablesApi == null || this.schemasApi == null;
+    return this.workspaceClient == null && missingApis;
+  }
+
+  private void ensureInitialized() {
+    if (this.catalogConfig == null
+        || this.databricksConfig == null
+        || this.tableFormat == null
+        || this.statementExecution == null
+        || this.tablesApi == null
+        || this.schemasApi == null) {
+      throw new CatalogSyncException(
+          "DatabricksUnityCatalogSyncClient is not initialized. "
+              + "Call init(catalogConfig, tableFormat, configuration) before use.");
+    }
+  }
+
+  private void ensureDeltaOnly() {
+    if (!Objects.equals(tableFormat, TableFormat.DELTA)) {
+      throw new CatalogSyncException(
+          "Databricks UC sync client currently supports external DELTA only");
+    }
+  }
+
+  private String getFullName(CatalogTableIdentifier tableIdentifier) {
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    String table = hierarchical.getTableName();
+    if (StringUtils.isBlank(table)) {
+      throw new CatalogSyncException("Databricks UC requires a table name");
+    }
+
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+    validateIdentifierComponent("table", table);
+    return hierarchical.getId();
+  }
+
+  private static void validateIdentifierComponent(String componentName, String value) {
+    if (!IDENTIFIER_PATTERN.matcher(value).matches()) {
+      throw new CatalogSyncException(
+          String.format(
+              "Invalid Databricks UC %s name '%s'. Only [a-zA-Z0-9_]+ is allowed.",
+              componentName, value));
+    }
+  }
+
+  private StatementResponse executeStatement(String statement) {
+    ExecuteStatementRequest request =
+        new ExecuteStatementRequest()
+            .setStatement(statement)
+            .setWarehouseId(databricksConfig.getWarehouseId())
+            .setFormat(Format.JSON_ARRAY)
+            .setDisposition(Disposition.INLINE)
+            .setWaitTimeout("30s")

Review Comment:
   `"30s"` is a magic string with no ability to tune for slower warehouses or large DDL operations. Define it as a constant:
   ```java
   private static final String STATEMENT_WAIT_TIMEOUT = "30s";
   ```
   Ideally, expose it as an optional config key in `DatabricksUnityCatalogConfig` so users can override it.



##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,572 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.Disposition;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.ExecuteStatementRequestOnWaitTimeout;
+import com.databricks.sdk.service.sql.Format;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+
+import org.apache.xtable.catalog.CatalogUtils;
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+/**
+ * Databricks Unity Catalog implementation of {@link CatalogSyncClient}.
+ *
+ * <p>Supports external Delta table sync operations (create, drop, create-or-replace, get table and
+ * schema/database operations) using Databricks SDK APIs and SQL Statement Execution API. Schema
+ * refresh compares desired and existing schemas and triggers metadata sync when differences are
+ * detected.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+  private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
+
+  private ExternalCatalogConfig catalogConfig;
+  private DatabricksUnityCatalogConfig databricksConfig;
+  private String tableFormat;
+  private WorkspaceClient workspaceClient;
+  private StatementExecutionAPI statementExecution;
+  private TablesAPI tablesApi;
+  private SchemasAPI schemasApi;
+
+  /**
+   * For loading the instance using ServiceLoader.
+   *
+   * <p>This constructor requires {@link #init(ExternalCatalogConfig, String, Configuration)} to be
+   * called before invoking catalog operations.
+   */
+  public DatabricksUnityCatalogSyncClient() {}
+
+  public DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig,
+      String tableFormat,
+      Configuration configuration,
+      StatementExecutionAPI statementExecution,
+      TablesAPI tablesApi,
+      SchemasAPI schemasApi) {
+    this.statementExecution = statementExecution;
+    this.tablesApi = tablesApi;
+    this.schemasApi = schemasApi;
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  @Override
+  public String getCatalogId() {
+    ensureInitialized();
+    return catalogConfig.getCatalogId();
+  }
+
+  @Override
+  public String getCatalogType() {
+    return CatalogType.DATABRICKS_UC;
+  }
+
+  @Override
+  public String getStorageLocation(TableInfo table) {
+    if (table == null) {
+      return null;
+    }
+    return table.getStorageLocation();
+  }
+
+  @Override
+  public boolean hasDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    String fullName = catalog + "." + schema;
+    try {
+      SchemaInfo schemaInfo = schemasApi.get(fullName);
+      return schemaInfo != null;
+    } catch (NotFound e) {
+      return false;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get schema: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    try {
+      schemasApi.create(new CreateSchema().setCatalogName(catalog).setName(schema));
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to create database: " + schema, e);
+    }
+  }
+
+  @Override
+  public TableInfo getTable(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      return tablesApi.get(fullName);
+    } catch (NotFound e) {
+      return null;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    String location = table.getBasePath();
+    if (StringUtils.isBlank(location)) {
+      throw new CatalogSyncException("Storage location is required for external Delta tables");
+    }
+
+    String statement =
+        String.format(
+            "CREATE TABLE IF NOT EXISTS %s USING DELTA LOCATION '%s'",
+            fullName, escapeSqlString(location));
+    log.info("Databricks UC create table: {}", fullName);
+    executeStatement(statement);
+  }
+
+  @Override
+  public void refreshTable(
+      InternalTable table, TableInfo catalogTable, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    if (catalogTable == null) {
+      log.warn(
+          "Databricks UC refreshTable called with null catalog table for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    InternalSchema schema = table.getReadSchema();
+    if (schema == null || schema.getFields() == null || schema.getFields().isEmpty()) {
+      log.warn(
+          "Databricks UC refreshTable skipped due to missing schema for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    if (!schemasMatch(schema, catalogTable)) {
+      String fullName = getFullName(tableIdentifier);
+      log.info("Databricks UC refresh table metadata (MSCK REPAIR TABLE): {}", fullName);
+      executeStatement(String.format("MSCK REPAIR TABLE %s SYNC METADATA", fullName));
+    } else {
+      log.info(
+          "Databricks UC refreshTable: schema already up to date for {}", tableIdentifier.getId());
+    }
+  }
+
+  @Override
+  public void createOrReplaceTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    dropTable(table, tableIdentifier);
+    try {
+      createTable(table, tableIdentifier);
+    } catch (Exception e) {
+      log.error(
+          "Databricks UC createOrReplaceTable failed to recreate {} after drop. "
+              + "Table was dropped but not recreated.",
+          fullName,
+          e);
+      throw new CatalogSyncException(
+          "Databricks UC createOrReplaceTable failed after drop for "
+              + fullName
+              + ": table was dropped but not recreated",
+          e);
+    }
+  }
+
+  @Override
+  public void dropTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      log.info("Databricks UC drop table: {}", fullName);
+      tablesApi.delete(fullName);
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to drop table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void init(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    this.catalogConfig = catalogConfig;
+    this.tableFormat = tableFormat;
+    this.databricksConfig = DatabricksUnityCatalogConfig.from(catalogConfig);
+
+    if (databricksConfig.getHost() == null || databricksConfig.getWarehouseId() == null) {
+      throw new CatalogSyncException(
+          "Databricks UC catalog requires host and warehouseId in catalogProperties");
+    }
+    if (shouldInitializeWorkspaceClient()) {
+      this.workspaceClient = new WorkspaceClient(buildConfig(databricksConfig));
+    }
+    if (this.statementExecution == null) {
+      this.statementExecution = workspaceClient.statementExecution();
+    }
+    if (this.tablesApi == null) {
+      this.tablesApi = this.workspaceClient.tables();
+    }
+    if (this.schemasApi == null) {
+      this.schemasApi = this.workspaceClient.schemas();
+    }
+    log.info(
+        "Initialized Databricks UC sync client for catalogId={} tableFormat={}",
+        catalogConfig.getCatalogId(),
+        tableFormat);
+  }
+
+  private boolean shouldInitializeWorkspaceClient() {
+    boolean missingApis =
+        this.statementExecution == null || this.tablesApi == null || this.schemasApi == null;
+    return this.workspaceClient == null && missingApis;
+  }
+
+  private void ensureInitialized() {
+    if (this.catalogConfig == null
+        || this.databricksConfig == null
+        || this.tableFormat == null
+        || this.statementExecution == null
+        || this.tablesApi == null
+        || this.schemasApi == null) {
+      throw new CatalogSyncException(
+          "DatabricksUnityCatalogSyncClient is not initialized. "
+              + "Call init(catalogConfig, tableFormat, configuration) before use.");
+    }
+  }
+
+  private void ensureDeltaOnly() {
+    if (!Objects.equals(tableFormat, TableFormat.DELTA)) {
+      throw new CatalogSyncException(
+          "Databricks UC sync client currently supports external DELTA only");
+    }
+  }
+
+  private String getFullName(CatalogTableIdentifier tableIdentifier) {
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    String table = hierarchical.getTableName();
+    if (StringUtils.isBlank(table)) {
+      throw new CatalogSyncException("Databricks UC requires a table name");
+    }
+
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+    validateIdentifierComponent("table", table);
+    return hierarchical.getId();
+  }
+
+  private static void validateIdentifierComponent(String componentName, String value) {
+    if (!IDENTIFIER_PATTERN.matcher(value).matches()) {
+      throw new CatalogSyncException(
+          String.format(
+              "Invalid Databricks UC %s name '%s'. Only [a-zA-Z0-9_]+ is allowed.",
+              componentName, value));
+    }
+  }
+
+  private StatementResponse executeStatement(String statement) {
+    ExecuteStatementRequest request =
+        new ExecuteStatementRequest()
+            .setStatement(statement)
+            .setWarehouseId(databricksConfig.getWarehouseId())
+            .setFormat(Format.JSON_ARRAY)
+            .setDisposition(Disposition.INLINE)
+            .setWaitTimeout("30s")
+            .setOnWaitTimeout(ExecuteStatementRequestOnWaitTimeout.CANCEL);
+
+    StatementResponse response = statementExecution.executeStatement(request);
+    if (response.getStatus() != null && response.getStatus().getState() == StatementState.FAILED) {
+      String errorMessage = null;
+      if (response.getStatus().getError() != null) {
+        errorMessage = response.getStatus().getError().getMessage();
+      }
+      if (StringUtils.isBlank(errorMessage)) {
+        throw new CatalogSyncException("Databricks UC statement failed: " + statement);
+      }
+      throw new CatalogSyncException(
+          "Databricks UC statement failed: " + statement + " (" + errorMessage + ")");
+    }
+    return response;
+  }
+
+  private DatabricksConfig buildConfig(DatabricksUnityCatalogConfig config) {
+    DatabricksConfig dbConfig = new DatabricksConfig().setHost(config.getHost());
+    if (!StringUtils.isBlank(config.getAuthType())) {
+      dbConfig.setAuthType(config.getAuthType());
+    }
+    if (!StringUtils.isBlank(config.getClientId())
+        && !StringUtils.isBlank(config.getClientSecret())) {
+      dbConfig.setClientId(config.getClientId());
+      dbConfig.setClientSecret(config.getClientSecret());
+      if (StringUtils.isBlank(config.getAuthType())) {
+        dbConfig.setAuthType("oauth-m2m");
+      }
+    }
+    return dbConfig;
+  }
+
+  private static String escapeSqlString(String value) {
+    return value.replace("'", "''");
+  }
+
+  private static boolean schemasMatch(InternalSchema desired, TableInfo existing) {
+    Map<String, ColumnSignature> desiredColumns = toColumnSignatureMap(desired);
+    Map<String, ColumnSignature> existingColumns = toColumnSignatureMap(existing);
+    if (desiredColumns.size() != existingColumns.size()) {
+      return false;
+    }
+    for (Map.Entry<String, ColumnSignature> entry : desiredColumns.entrySet()) {
+      ColumnSignature existingSignature = existingColumns.get(entry.getKey());
+      if (existingSignature == null) {
+        return false;
+      }
+      if (!entry.getValue().equals(existingSignature)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static Map<String, ColumnSignature> toColumnSignatureMap(InternalSchema schema) {
+    Map<String, ColumnSignature> result = new HashMap<>();
+    for (InternalField field : schema.getFields()) {
+      String name = normalizeName(field.getName());
+      String type = normalizeType(toSparkSqlType(field));
+      boolean nullable = field.getSchema().isNullable();
+      String comment = StringUtils.defaultString(field.getSchema().getComment());
+      result.put(name, new ColumnSignature(type, nullable, comment));
+    }
+    return result;
+  }
+
+  private static Map<String, ColumnSignature> toColumnSignatureMap(TableInfo tableInfo) {
+    Map<String, ColumnSignature> result = new HashMap<>();
+    if (tableInfo == null || tableInfo.getColumns() == null) {
+      return result;
+    }
+    for (ColumnInfo column : tableInfo.getColumns()) {
+      String name = normalizeName(column.getName());
+      String type = normalizeType(column.getTypeText());
+      boolean nullable = column.getNullable() == null || column.getNullable();
+      String comment = StringUtils.defaultString(column.getComment());
+      result.put(name, new ColumnSignature(type, nullable, comment));
+    }
+    return result;
+  }
+
+  private static final class ColumnSignature {

Review Comment:
   `ColumnSignature` manually implements constructor, `equals`, and `hashCode` (~28 lines). Annotate with Lombok `@Value` to eliminate the boilerplate — consistent with the rest of the codebase.



##########
xtable-databricks/src/test/java/org/apache/xtable/databricks/TestDatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,478 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+import com.databricks.sdk.service.sql.StatementStatus;
+
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.ThreePartHierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.schema.InternalType;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+
+@ExtendWith(MockitoExtension.class)
+public class TestDatabricksUnityCatalogSyncClient {
+
+  @Mock private StatementExecutionAPI mockStatementExecution;
+  @Mock private TablesAPI mockTablesApi;
+  @Mock private SchemasAPI mockSchemasApi;
+
+  @Test
+  void testGetCatalogIdFailsWhenNotInitialized() {
+    DatabricksUnityCatalogSyncClient client = new DatabricksUnityCatalogSyncClient();
+
+    CatalogSyncException exception = assertThrows(CatalogSyncException.class, client::getCatalogId);
+    assertTrue(exception.getMessage().contains("not initialized"));
+  }
+
+  @Test
+  void testCreateTableDelta_NoColumns() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    when(mockStatementExecution.executeStatement(any(ExecuteStatementRequest.class)))
+        .thenReturn(
+            new StatementResponse()
+                .setStatus(new StatementStatus().setState(StatementState.SUCCEEDED)));
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.createTable(table, tableIdentifier);
+
+    ArgumentCaptor<ExecuteStatementRequest> requestCaptor =
+        ArgumentCaptor.forClass(ExecuteStatementRequest.class);
+    verify(mockStatementExecution).executeStatement(requestCaptor.capture());
+    ExecuteStatementRequest request = requestCaptor.getValue();
+    assertEquals("wh-1", request.getWarehouseId());
+    assertEquals(
+        "CREATE TABLE IF NOT EXISTS main.default.people USING DELTA LOCATION 's3://bucket/path'",
+        request.getStatement());
+  }
+
+  @Test
+  void testCreateTableRejectsNonDelta() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.ICEBERG);
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    assertThrows(CatalogSyncException.class, () -> client.createTable(table, tableIdentifier));
+  }
+
+  @Test
+  void testCreateTableRejectsInvalidIdentifier() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people;DROP_TABLE");
+
+    assertThrows(CatalogSyncException.class, () -> client.createTable(table, tableIdentifier));
+  }
+
+  @Test
+  void testCreateOrReplaceTableDelta() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    when(mockStatementExecution.executeStatement(any(ExecuteStatementRequest.class)))
+        .thenReturn(
+            new StatementResponse()
+                .setStatus(new StatementStatus().setState(StatementState.SUCCEEDED)));
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.createOrReplaceTable(table, tableIdentifier);
+
+    ArgumentCaptor<ExecuteStatementRequest> requestCaptor =
+        ArgumentCaptor.forClass(ExecuteStatementRequest.class);
+    verify(mockStatementExecution).executeStatement(requestCaptor.capture());
+    verify(mockTablesApi).delete("main.default.people");
+    ExecuteStatementRequest request = requestCaptor.getValue();
+    assertEquals(
+        "CREATE TABLE IF NOT EXISTS main.default.people USING DELTA LOCATION 's3://bucket/path'",
+        request.getStatement());
+  }
+
+  @Test
+  void testCreateOrReplaceTableRejectsNonDelta() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.ICEBERG);
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    assertThrows(
+        CatalogSyncException.class, () -> client.createOrReplaceTable(table, tableIdentifier));
+  }
+
+  @Test
+  void testCreateOrReplaceTableRecreateFailureWarnsAndThrows() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    when(mockStatementExecution.executeStatement(any(ExecuteStatementRequest.class)))
+        .thenReturn(
+            new StatementResponse()
+                .setStatus(new StatementStatus().setState(StatementState.FAILED)));
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    CatalogSyncException exception =
+        assertThrows(
+            CatalogSyncException.class, () -> client.createOrReplaceTable(table, tableIdentifier));
+    assertTrue(exception.getMessage().contains("table was dropped but not recreated"));
+    verify(mockTablesApi).delete("main.default.people");
+  }
+
+  @Test
+  void testRefreshTableSchemaEvolution() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    when(mockStatementExecution.executeStatement(any(ExecuteStatementRequest.class)))
+        .thenReturn(
+            new StatementResponse()
+                .setStatus(new StatementStatus().setState(StatementState.SUCCEEDED)));
+
+    InternalSchema idSchema =
+        InternalSchema.builder().name("id").dataType(InternalType.INT).isNullable(true).build();
+    InternalSchema readSchema =
+        InternalSchema.builder()
+            .name("root")
+            .dataType(InternalType.RECORD)
+            .isNullable(true)
+            .fields(
+                java.util.Arrays.asList(
+                    InternalField.builder().name("id").schema(idSchema).build()))
+            .build();
+    InternalTable table =
+        InternalTable.builder().basePath("s3://bucket/path").readSchema(readSchema).build();
+    TableInfo catalogTable =
+        new TableInfo()
+            .setColumns(
+                java.util.Arrays.asList(
+                    new ColumnInfo().setName("name").setTypeText("string").setNullable(true)));
+
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.refreshTable(table, catalogTable, tableIdentifier);
+
+    ArgumentCaptor<ExecuteStatementRequest> requestCaptor =
+        ArgumentCaptor.forClass(ExecuteStatementRequest.class);
+    verify(mockStatementExecution).executeStatement(requestCaptor.capture());
+    assertEquals(
+        "MSCK REPAIR TABLE main.default.people SYNC METADATA",
+        requestCaptor.getValue().getStatement());
+  }
+
+  @Test
+  void testRefreshTableSchemaMatchNoop() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    InternalSchema idSchema =
+        InternalSchema.builder().name("id").dataType(InternalType.INT).isNullable(true).build();
+    InternalSchema readSchema =
+        InternalSchema.builder()
+            .name("root")
+            .dataType(InternalType.RECORD)
+            .isNullable(true)
+            .fields(
+                java.util.Arrays.asList(
+                    InternalField.builder().name("id").schema(idSchema).build()))
+            .build();
+    InternalTable table =
+        InternalTable.builder().basePath("s3://bucket/path").readSchema(readSchema).build();
+    TableInfo catalogTable =
+        new TableInfo()
+            .setColumns(
+                java.util.Arrays.asList(
+                    new ColumnInfo().setName("id").setTypeText("int").setNullable(true)));
+
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.refreshTable(table, catalogTable, tableIdentifier);
+
+    verify(mockStatementExecution, never()).executeStatement(any(ExecuteStatementRequest.class));
+  }
+
+  @Test
+  void testRefreshTableWithNullCatalogTableNoop() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    InternalSchema idSchema =
+        InternalSchema.builder().name("id").dataType(InternalType.INT).isNullable(true).build();
+    InternalSchema readSchema =
+        InternalSchema.builder()
+            .name("root")
+            .dataType(InternalType.RECORD)
+            .isNullable(true)
+            .fields(
+                java.util.Arrays.asList(
+                    InternalField.builder().name("id").schema(idSchema).build()))
+            .build();
+    InternalTable table =
+        InternalTable.builder().basePath("s3://bucket/path").readSchema(readSchema).build();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.refreshTable(table, null, tableIdentifier);
+
+    verify(mockStatementExecution, never()).executeStatement(any(ExecuteStatementRequest.class));
+  }
+
+  @Test
+  void testRefreshTableWithNullSchemaNoop() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    InternalTable table = InternalTable.builder().basePath("s3://bucket/path").build();
+    TableInfo catalogTable = new TableInfo();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.refreshTable(table, catalogTable, tableIdentifier);
+
+    verify(mockStatementExecution, never()).executeStatement(any(ExecuteStatementRequest.class));
+  }
+
+  @Test
+  void testRefreshTableWithEmptySchemaNoop() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    InternalSchema readSchema =
+        InternalSchema.builder()
+            .name("root")
+            .dataType(InternalType.RECORD)
+            .isNullable(true)
+            .fields(java.util.Collections.emptyList())
+            .build();
+    InternalTable table =
+        InternalTable.builder().basePath("s3://bucket/path").readSchema(readSchema).build();
+    TableInfo catalogTable = new TableInfo();
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.refreshTable(table, catalogTable, tableIdentifier);
+
+    verify(mockStatementExecution, never()).executeStatement(any(ExecuteStatementRequest.class));
+  }
+
+  @Test
+  void testDropTable() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    client.dropTable(InternalTable.builder().basePath("s3://bucket/path").build(), tableIdentifier);
+
+    verify(mockTablesApi).delete("main.default.people");
+  }
+
+  @Test
+  void testDropTableFailure() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+
+    doThrow(new RuntimeException("boom")).when(mockTablesApi).delete("main.default.people");
+
+    assertThrows(
+        CatalogSyncException.class,
+        () ->
+            client.dropTable(
+                InternalTable.builder().basePath("s3://bucket/path").build(), tableIdentifier));
+  }
+
+  @Test
+  void testHasDatabaseTrue() {
+    DatabricksUnityCatalogSyncClient client = createClient(TableFormat.DELTA);
+
+    when(mockSchemasApi.get("main.default")).thenReturn(new SchemaInfo());
+
+    ThreePartHierarchicalTableIdentifier tableIdentifier =
+        new ThreePartHierarchicalTableIdentifier("main", "default", "people");
+    boolean exists = client.hasDatabase(tableIdentifier);
+    assertEquals(true, exists);

Review Comment:
   Nit: `assertEquals(true, exists)` → `assertTrue(exists)`.



##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,572 @@
+/*
+ * 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.
+ */
+ 
+package org.apache.xtable.databricks;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.databricks.sdk.WorkspaceClient;
+import com.databricks.sdk.core.DatabricksConfig;
+import com.databricks.sdk.core.error.platform.NotFound;
+import com.databricks.sdk.service.catalog.ColumnInfo;
+import com.databricks.sdk.service.catalog.CreateSchema;
+import com.databricks.sdk.service.catalog.SchemaInfo;
+import com.databricks.sdk.service.catalog.SchemasAPI;
+import com.databricks.sdk.service.catalog.TableInfo;
+import com.databricks.sdk.service.catalog.TablesAPI;
+import com.databricks.sdk.service.sql.Disposition;
+import com.databricks.sdk.service.sql.ExecuteStatementRequest;
+import com.databricks.sdk.service.sql.ExecuteStatementRequestOnWaitTimeout;
+import com.databricks.sdk.service.sql.Format;
+import com.databricks.sdk.service.sql.StatementExecutionAPI;
+import com.databricks.sdk.service.sql.StatementResponse;
+import com.databricks.sdk.service.sql.StatementState;
+
+import org.apache.xtable.catalog.CatalogUtils;
+import org.apache.xtable.conversion.ExternalCatalogConfig;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.CatalogType;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.spi.sync.CatalogSyncClient;
+
+/**
+ * Databricks Unity Catalog implementation of {@link CatalogSyncClient}.
+ *
+ * <p>Supports external Delta table sync operations (create, drop, create-or-replace, get table and
+ * schema/database operations) using Databricks SDK APIs and SQL Statement Execution API. Schema
+ * refresh compares desired and existing schemas and triggers metadata sync when differences are
+ * detected.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+  private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("[a-zA-Z0-9_]+");
+
+  private ExternalCatalogConfig catalogConfig;
+  private DatabricksUnityCatalogConfig databricksConfig;
+  private String tableFormat;
+  private WorkspaceClient workspaceClient;
+  private StatementExecutionAPI statementExecution;
+  private TablesAPI tablesApi;
+  private SchemasAPI schemasApi;
+
+  /**
+   * For loading the instance using ServiceLoader.
+   *
+   * <p>This constructor requires {@link #init(ExternalCatalogConfig, String, Configuration)} to be
+   * called before invoking catalog operations.
+   */
+  public DatabricksUnityCatalogSyncClient() {}
+
+  public DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  DatabricksUnityCatalogSyncClient(
+      ExternalCatalogConfig catalogConfig,
+      String tableFormat,
+      Configuration configuration,
+      StatementExecutionAPI statementExecution,
+      TablesAPI tablesApi,
+      SchemasAPI schemasApi) {
+    this.statementExecution = statementExecution;
+    this.tablesApi = tablesApi;
+    this.schemasApi = schemasApi;
+    init(catalogConfig, tableFormat, configuration);
+  }
+
+  @Override
+  public String getCatalogId() {
+    ensureInitialized();
+    return catalogConfig.getCatalogId();
+  }
+
+  @Override
+  public String getCatalogType() {
+    return CatalogType.DATABRICKS_UC;
+  }
+
+  @Override
+  public String getStorageLocation(TableInfo table) {
+    if (table == null) {
+      return null;
+    }
+    return table.getStorageLocation();
+  }
+
+  @Override
+  public boolean hasDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    String fullName = catalog + "." + schema;
+    try {
+      SchemaInfo schemaInfo = schemasApi.get(fullName);
+      return schemaInfo != null;
+    } catch (NotFound e) {
+      return false;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get schema: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createDatabase(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+
+    try {
+      schemasApi.create(new CreateSchema().setCatalogName(catalog).setName(schema));
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to create database: " + schema, e);
+    }
+  }
+
+  @Override
+  public TableInfo getTable(CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      return tablesApi.get(fullName);
+    } catch (NotFound e) {
+      return null;
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to get table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void createTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    String location = table.getBasePath();
+    if (StringUtils.isBlank(location)) {
+      throw new CatalogSyncException("Storage location is required for external Delta tables");
+    }
+
+    String statement =
+        String.format(
+            "CREATE TABLE IF NOT EXISTS %s USING DELTA LOCATION '%s'",
+            fullName, escapeSqlString(location));
+    log.info("Databricks UC create table: {}", fullName);
+    executeStatement(statement);
+  }
+
+  @Override
+  public void refreshTable(
+      InternalTable table, TableInfo catalogTable, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    if (catalogTable == null) {
+      log.warn(
+          "Databricks UC refreshTable called with null catalog table for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    InternalSchema schema = table.getReadSchema();
+    if (schema == null || schema.getFields() == null || schema.getFields().isEmpty()) {
+      log.warn(
+          "Databricks UC refreshTable skipped due to missing schema for {}",
+          tableIdentifier.getId());
+      return;
+    }
+    if (!schemasMatch(schema, catalogTable)) {
+      String fullName = getFullName(tableIdentifier);
+      log.info("Databricks UC refresh table metadata (MSCK REPAIR TABLE): {}", fullName);
+      executeStatement(String.format("MSCK REPAIR TABLE %s SYNC METADATA", fullName));
+    } else {
+      log.info(
+          "Databricks UC refreshTable: schema already up to date for {}", tableIdentifier.getId());
+    }
+  }
+
+  @Override
+  public void createOrReplaceTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    ensureDeltaOnly();
+    String fullName = getFullName(tableIdentifier);
+    dropTable(table, tableIdentifier);
+    try {
+      createTable(table, tableIdentifier);
+    } catch (Exception e) {
+      log.error(
+          "Databricks UC createOrReplaceTable failed to recreate {} after drop. "
+              + "Table was dropped but not recreated.",
+          fullName,
+          e);
+      throw new CatalogSyncException(
+          "Databricks UC createOrReplaceTable failed after drop for "
+              + fullName
+              + ": table was dropped but not recreated",
+          e);
+    }
+  }
+
+  @Override
+  public void dropTable(InternalTable table, CatalogTableIdentifier tableIdentifier) {
+    ensureInitialized();
+    String fullName = getFullName(tableIdentifier);
+    try {
+      log.info("Databricks UC drop table: {}", fullName);
+      tablesApi.delete(fullName);
+    } catch (Exception e) {
+      throw new CatalogSyncException("Failed to drop table: " + fullName, e);
+    }
+  }
+
+  @Override
+  public void init(
+      ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+    this.catalogConfig = catalogConfig;
+    this.tableFormat = tableFormat;
+    this.databricksConfig = DatabricksUnityCatalogConfig.from(catalogConfig);
+
+    if (databricksConfig.getHost() == null || databricksConfig.getWarehouseId() == null) {
+      throw new CatalogSyncException(
+          "Databricks UC catalog requires host and warehouseId in catalogProperties");
+    }
+    if (shouldInitializeWorkspaceClient()) {
+      this.workspaceClient = new WorkspaceClient(buildConfig(databricksConfig));
+    }
+    if (this.statementExecution == null) {
+      this.statementExecution = workspaceClient.statementExecution();
+    }
+    if (this.tablesApi == null) {
+      this.tablesApi = this.workspaceClient.tables();
+    }
+    if (this.schemasApi == null) {
+      this.schemasApi = this.workspaceClient.schemas();
+    }
+    log.info(
+        "Initialized Databricks UC sync client for catalogId={} tableFormat={}",
+        catalogConfig.getCatalogId(),
+        tableFormat);
+  }
+
+  private boolean shouldInitializeWorkspaceClient() {
+    boolean missingApis =
+        this.statementExecution == null || this.tablesApi == null || this.schemasApi == null;
+    return this.workspaceClient == null && missingApis;
+  }
+
+  private void ensureInitialized() {
+    if (this.catalogConfig == null
+        || this.databricksConfig == null
+        || this.tableFormat == null
+        || this.statementExecution == null
+        || this.tablesApi == null
+        || this.schemasApi == null) {
+      throw new CatalogSyncException(
+          "DatabricksUnityCatalogSyncClient is not initialized. "
+              + "Call init(catalogConfig, tableFormat, configuration) before use.");
+    }
+  }
+
+  private void ensureDeltaOnly() {
+    if (!Objects.equals(tableFormat, TableFormat.DELTA)) {
+      throw new CatalogSyncException(
+          "Databricks UC sync client currently supports external DELTA only");
+    }
+  }
+
+  private String getFullName(CatalogTableIdentifier tableIdentifier) {
+    HierarchicalTableIdentifier hierarchical =
+        CatalogUtils.toHierarchicalTableIdentifier(tableIdentifier);
+    String catalog = hierarchical.getCatalogName();
+    if (StringUtils.isBlank(catalog)) {
+      throw new CatalogSyncException(
+          "Databricks UC requires a catalog name (expected catalog.schema.table)");
+    }
+    String schema = hierarchical.getDatabaseName();
+    if (StringUtils.isBlank(schema)) {
+      throw new CatalogSyncException("Databricks UC requires a schema name");
+    }
+    String table = hierarchical.getTableName();
+    if (StringUtils.isBlank(table)) {
+      throw new CatalogSyncException("Databricks UC requires a table name");
+    }
+
+    validateIdentifierComponent("catalog", catalog);
+    validateIdentifierComponent("schema", schema);
+    validateIdentifierComponent("table", table);
+    return hierarchical.getId();
+  }
+
+  private static void validateIdentifierComponent(String componentName, String value) {
+    if (!IDENTIFIER_PATTERN.matcher(value).matches()) {
+      throw new CatalogSyncException(
+          String.format(
+              "Invalid Databricks UC %s name '%s'. Only [a-zA-Z0-9_]+ is allowed.",
+              componentName, value));
+    }
+  }
+
+  private StatementResponse executeStatement(String statement) {
+    ExecuteStatementRequest request =
+        new ExecuteStatementRequest()
+            .setStatement(statement)
+            .setWarehouseId(databricksConfig.getWarehouseId())
+            .setFormat(Format.JSON_ARRAY)
+            .setDisposition(Disposition.INLINE)
+            .setWaitTimeout("30s")
+            .setOnWaitTimeout(ExecuteStatementRequestOnWaitTimeout.CANCEL);
+
+    StatementResponse response = statementExecution.executeStatement(request);
+    if (response.getStatus() != null && response.getStatus().getState() == StatementState.FAILED) {
+      String errorMessage = null;
+      if (response.getStatus().getError() != null) {
+        errorMessage = response.getStatus().getError().getMessage();
+      }
+      if (StringUtils.isBlank(errorMessage)) {
+        throw new CatalogSyncException("Databricks UC statement failed: " + statement);
+      }
+      throw new CatalogSyncException(
+          "Databricks UC statement failed: " + statement + " (" + errorMessage + ")");
+    }
+    return response;
+  }
+
+  private DatabricksConfig buildConfig(DatabricksUnityCatalogConfig config) {
+    DatabricksConfig dbConfig = new DatabricksConfig().setHost(config.getHost());
+    if (!StringUtils.isBlank(config.getAuthType())) {
+      dbConfig.setAuthType(config.getAuthType());
+    }
+    if (!StringUtils.isBlank(config.getClientId())
+        && !StringUtils.isBlank(config.getClientSecret())) {
+      dbConfig.setClientId(config.getClientId());
+      dbConfig.setClientSecret(config.getClientSecret());
+      if (StringUtils.isBlank(config.getAuthType())) {
+        dbConfig.setAuthType("oauth-m2m");
+      }
+    }
+    return dbConfig;
+  }
+
+  private static String escapeSqlString(String value) {
+    return value.replace("'", "''");
+  }
+
+  private static boolean schemasMatch(InternalSchema desired, TableInfo existing) {
+    Map<String, ColumnSignature> desiredColumns = toColumnSignatureMap(desired);
+    Map<String, ColumnSignature> existingColumns = toColumnSignatureMap(existing);
+    if (desiredColumns.size() != existingColumns.size()) {
+      return false;
+    }
+    for (Map.Entry<String, ColumnSignature> entry : desiredColumns.entrySet()) {
+      ColumnSignature existingSignature = existingColumns.get(entry.getKey());
+      if (existingSignature == null) {
+        return false;
+      }
+      if (!entry.getValue().equals(existingSignature)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static Map<String, ColumnSignature> toColumnSignatureMap(InternalSchema schema) {
+    Map<String, ColumnSignature> result = new HashMap<>();
+    for (InternalField field : schema.getFields()) {
+      String name = normalizeName(field.getName());
+      String type = normalizeType(toSparkSqlType(field));
+      boolean nullable = field.getSchema().isNullable();
+      String comment = StringUtils.defaultString(field.getSchema().getComment());
+      result.put(name, new ColumnSignature(type, nullable, comment));
+    }
+    return result;
+  }
+
+  private static Map<String, ColumnSignature> toColumnSignatureMap(TableInfo tableInfo) {
+    Map<String, ColumnSignature> result = new HashMap<>();
+    if (tableInfo == null || tableInfo.getColumns() == null) {
+      return result;
+    }
+    for (ColumnInfo column : tableInfo.getColumns()) {
+      String name = normalizeName(column.getName());
+      String type = normalizeType(column.getTypeText());
+      boolean nullable = column.getNullable() == null || column.getNullable();
+      String comment = StringUtils.defaultString(column.getComment());
+      result.put(name, new ColumnSignature(type, nullable, comment));
+    }
+    return result;
+  }
+
+  private static final class ColumnSignature {
+    private final String type;
+    private final boolean nullable;
+    private final String comment;
+
+    private ColumnSignature(String type, boolean nullable, String comment) {
+      this.type = type;
+      this.nullable = nullable;
+      this.comment = comment;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      ColumnSignature that = (ColumnSignature) o;
+      return nullable == that.nullable
+          && Objects.equals(type, that.type)
+          && Objects.equals(comment, that.comment);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(type, nullable, comment);
+    }
+  }
+
+  private static String normalizeName(String value) {
+    return value == null ? "" : value.toLowerCase(Locale.ROOT);
+  }
+
+  private static String normalizeType(String value) {
+    if (value == null) {
+      return "";
+    }
+    return value.replaceAll("\\s+", "").toLowerCase(Locale.ROOT);
+  }
+
+  private static String toSparkSqlType(InternalField field) {
+    switch (field.getSchema().getDataType()) {
+      case ENUM:
+      case STRING:
+        return "string";
+      case INT:
+        return "int";
+      case LONG:
+        return "bigint";
+      case BYTES:
+      case FIXED:
+      case UUID:
+        return "binary";
+      case BOOLEAN:
+        return "boolean";
+      case FLOAT:
+        return "float";
+      case DATE:
+        return "date";
+      case TIMESTAMP:
+        return "timestamp";
+      case TIMESTAMP_NTZ:
+        return "timestamp_ntz";
+      case DOUBLE:
+        return "double";
+      case DECIMAL:
+        int precision =
+            (int) field.getSchema().getMetadata().get(InternalSchema.MetadataKey.DECIMAL_PRECISION);
+        int scale =
+            (int) field.getSchema().getMetadata().get(InternalSchema.MetadataKey.DECIMAL_SCALE);
+        return String.format("decimal(%d,%d)", precision, scale);
+      case RECORD:
+        return toStructType(field.getSchema());
+      case MAP:
+        InternalField key =
+            field.getSchema().getFields().stream()
+                .filter(
+                    mapField ->
+                        InternalField.Constants.MAP_KEY_FIELD_NAME.equals(mapField.getName()))
+                .findFirst()
+                .orElseThrow(() -> new CatalogSyncException("Invalid map schema"));
+        InternalField value =
+            field.getSchema().getFields().stream()
+                .filter(
+                    mapField ->
+                        InternalField.Constants.MAP_VALUE_FIELD_NAME.equals(mapField.getName()))
+                .findFirst()
+                .orElseThrow(() -> new CatalogSyncException("Invalid map schema"));
+        return String.format("map<%s,%s>", toSparkSqlType(key), toSparkSqlType(value));
+      case LIST:
+        InternalField element =
+            field.getSchema().getFields().stream()
+                .filter(
+                    arrayField ->
+                        InternalField.Constants.ARRAY_ELEMENT_FIELD_NAME.equals(
+                            arrayField.getName()))
+                .findFirst()
+                .orElseThrow(() -> new CatalogSyncException("Invalid array schema"));
+        return String.format("array<%s>", toSparkSqlType(element));
+      default:
+        throw new CatalogSyncException("Unsupported type: " + field.getSchema().getDataType());
+    }
+  }
+
+  private static String toStructType(InternalSchema schema) {

Review Comment:
   Nit: `toStructType` uses a manual `StringBuilder` + `first` flag. Prefer:
   ```java
   return schema.getFields().stream()
       .map(f -> f.getName() + ":" + toSparkSqlType(f))
       .collect(Collectors.joining(",", "struct<", ">"));
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org