You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "parisni (via GitHub)" <gi...@apache.org> on 2026/02/11 13:44:30 UTC
[PR] feat: databricks unity catalog sync [incubator-xtable]
parisni opened a new pull request, #802:
URL: https://github.com/apache/incubator-xtable/pull/802
## Important Read
- Please ensure the GitHub issue is mentioned at the beginning of the PR
## What is the purpose of the pull request
This pull request adds Databricks Unity Catalog (DATABRICKS_UC) catalog sync support to XTable, with a new xtable-databricks module, UC config wiring, UC sync client
implementation, and documentation updates.
## Brief change log
- Added new xtable-databricks module + ServiceLoader wiring.
- Added DatabricksUnityCatalogConfig and DatabricksUnityCatalogSyncClient (DDL-based UC registration).
- Implemented UC sync methods: hasDatabase, createDatabase, getTable, createTable, createOrReplaceTable, dropTable, refreshTable.
- Added UC-specific unit tests for all catalog sync methods.
- Documented Databricks UC sync flow, auth (OAuth M2M), limitations (Delta-only, unique location), and schema evolution strategy (drop/recreate).
## Verify this pull request
This change added tests and can be verified as follows:
- Added TestDatabricksUnityCatalogSyncClient in xtable-databricks.
- Run:
- mvn -Drat.skip=true -Dspotless.skip=true -Dtest=TestDatabricksUnityCatalogSyncClient -pl xtable-databricks test
--
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
Re: [PR] feat: built-in databricks unity catalog sync [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #802:
URL: https://github.com/apache/incubator-xtable/pull/802#discussion_r2841723788
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
+ * Execution API. Actual DDL execution and schema diffing should be implemented in a follow-up
+ * change.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+
+ private ExternalCatalogConfig catalogConfig;
+ private DatabricksUnityCatalogConfig databricksConfig;
+ private Configuration hadoopConf;
+ private String tableFormat;
+ private WorkspaceClient workspaceClient;
+ private StatementExecutionAPI statementExecution;
+ private TablesAPI tablesApi;
+ private SchemasAPI schemasApi;
+
+ // For loading the instance using ServiceLoader
+ 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() {
+ 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) {
+ 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)");
Review Comment:
`createOrReplaceTable` drops then recreates the table. This is not atomic — if `createTable` fails after `dropTable` succeeds, the table is gone. Consider using `CREATE OR REPLACE TABLE` DDL instead (if supported for external tables), or at minimum catch the create failure and log a clear error that the table was dropped but not recreated.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
+ * Execution API. Actual DDL execution and schema diffing should be implemented in a follow-up
+ * change.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+
+ private ExternalCatalogConfig catalogConfig;
+ private DatabricksUnityCatalogConfig databricksConfig;
+ private Configuration hadoopConf;
+ private String tableFormat;
+ private WorkspaceClient workspaceClient;
+ private StatementExecutionAPI statementExecution;
+ private TablesAPI tablesApi;
+ private SchemasAPI schemasApi;
+
+ // For loading the instance using ServiceLoader
+ 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() {
+ 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();
Review Comment:
`MSCK REPAIR TABLE ... SYNC METADATA` is a Hive-specific command. Does Databricks UC's SQL Warehouse support this syntax? The standard Databricks approach for refreshing external table metadata is `ALTER TABLE <table> SET TBLPROPERTIES` or reading from the Delta log directly.
If this doesn't actually work on Databricks, schema drift would silently be ignored. Has this been tested against a real SQL Warehouse?
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
+ * Execution API. Actual DDL execution and schema diffing should be implemented in a follow-up
+ * change.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+
+ private ExternalCatalogConfig catalogConfig;
+ private DatabricksUnityCatalogConfig databricksConfig;
+ private Configuration hadoopConf;
Review Comment:
All fields (`catalogConfig`, `databricksConfig`, `hadoopConf`, `tableFormat`, `workspaceClient`, etc.) are mutable and set via `init()`. This pattern is fragile — calling any method before `init()` would NPE.
Consider making these `final` and removing the no-arg constructor, or at least adding a null-check guard in methods that use these fields. The ServiceLoader no-arg constructor + `init()` pattern should be documented with a warning.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
+ * Execution API. Actual DDL execution and schema diffing should be implemented in a follow-up
+ * change.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+
+ private ExternalCatalogConfig catalogConfig;
+ private DatabricksUnityCatalogConfig databricksConfig;
+ private Configuration hadoopConf;
+ private String tableFormat;
+ private WorkspaceClient workspaceClient;
+ private StatementExecutionAPI statementExecution;
+ private TablesAPI tablesApi;
+ private SchemasAPI schemasApi;
+
+ // For loading the instance using ServiceLoader
+ public DatabricksUnityCatalogSyncClient() {}
+
+ public DatabricksUnityCatalogSyncClient(
+ ExternalCatalogConfig catalogConfig, String tableFormat, Configuration configuration) {
+ init(catalogConfig, tableFormat, configuration);
+ }
+
+ DatabricksUnityCatalogSyncClient(
+ ExternalCatalogConfig catalogConfig,
+ String tableFormat,
Review Comment:
SQL injection risk: `fullName` is derived from user-provided `tableIdentifier` and interpolated directly into the SQL string. While `escapeSqlString` is used for `location`, the table name itself is not sanitized.
If `hierarchicalId` contains `; DROP TABLE ...`, this would be injected. Consider validating that catalog/schema/table names match `[a-zA-Z0-9_]+` before interpolating them into DDL statements.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
Review Comment:
Class-level Javadoc says "This is a placeholder" and "Actual DDL execution and schema diffing should be implemented in a follow-up change" — but this PR implements both. The Javadoc is stale.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogConfig.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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";
+ public static final String TOKEN = "externalCatalog.uc.token";
+
+ String host;
+ String warehouseId;
+ String authType;
+ String clientId;
+ String clientSecret;
+ String token;
+
+ public static DatabricksUnityCatalogConfig from(ExternalCatalogConfig catalogConfig) {
+ Map<String, String> props = catalogConfig.getCatalogProperties();
+ return new DatabricksUnityCatalogConfig(
+ props.get(HOST),
+ props.get(WAREHOUSE_ID),
+ props.get(AUTH_TYPE),
+ props.get(CLIENT_ID),
+ props.get(CLIENT_SECRET),
+ props.get(TOKEN));
+ }
+}
Review Comment:
The `token` field is defined and parsed but the docs say PAT auth is "intentionally not wired" and "not supported yet". If it's not supported, don't add the field — it creates confusion. Remove `TOKEN` constant and `token` field until PAT auth is actually implemented.
##########
xtable-databricks/src/test/java/org/apache/xtable/databricks/TestDatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,570 @@
+/*
+ * 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
Review Comment:
Tests have massive boilerplate — every test method repeats the same 15-line client construction block. Extract a `createClient(String tableFormat)` helper method to reduce duplication and improve readability.
##########
xtable-databricks/pom.xml:
##########
@@ -0,0 +1,80 @@
+<?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>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
Review Comment:
Pin the Databricks SDK version to a property in the parent `pom.xml` instead of hardcoding `0.85.0` here. Follow the pattern used for other dependency versions in the project.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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 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 skeleton for CatalogSyncClient.
+ *
+ * <p>This is a placeholder to wire Databricks UC as a catalog target via the SQL Statement
+ * Execution API. Actual DDL execution and schema diffing should be implemented in a follow-up
+ * change.
+ */
+@Log4j2
+public class DatabricksUnityCatalogSyncClient implements CatalogSyncClient<TableInfo> {
+
+ private ExternalCatalogConfig catalogConfig;
+ private DatabricksUnityCatalogConfig databricksConfig;
+ private Configuration hadoopConf;
+ private String tableFormat;
+ private WorkspaceClient workspaceClient;
+ private StatementExecutionAPI statementExecution;
+ private TablesAPI tablesApi;
+ private SchemasAPI schemasApi;
+
+ // For loading the instance using ServiceLoader
+ 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() {
+ 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) {
+ 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 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);
+ }
+ }
Review Comment:
The `init` method can create up to 3 separate `WorkspaceClient` instances (lines 153, 158, 164) if `statementExecution`, `tablesApi`, and `schemasApi` are all null. Create the `WorkspaceClient` once at the top:
```java
if (this.workspaceClient == null) {
this.workspaceClient = new WorkspaceClient(buildConfig(databricksConfig));
}
if (this.statementExecution == null) {
this.statementExecution = workspaceClient.statementExecution();
}
...
```
##########
xtable-databricks/src/test/java/org/apache/xtable/databricks/TestDatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,570 @@
+/*
+ * 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.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+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 testCreateTableDelta_NoColumns() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DatabricksUnityCatalogConfig.HOST, "https://example.cloud.databricks.com");
+ props.put(DatabricksUnityCatalogConfig.WAREHOUSE_ID, "wh-1");
+ props.put(DatabricksUnityCatalogConfig.AUTH_TYPE, "oauth-m2m");
+ ExternalCatalogConfig config =
+ ExternalCatalogConfig.builder()
+ .catalogId("uc")
+ .catalogType(CatalogType.DATABRICKS_UC)
+ .catalogProperties(props)
+ .build();
+
+ DatabricksUnityCatalogSyncClient client =
+ new DatabricksUnityCatalogSyncClient(
+ config,
+ TableFormat.DELTA,
+ new Configuration(),
+ mockStatementExecution,
+ mockTablesApi,
+ mockSchemasApi);
+
+ when(mockStatementExecution.executeStatement(any(ExecuteStatementRequest.class)))
+ .thenReturn(
+ new StatementResponse()
+ .setStatus(new StatementStatus().setState(StatementState.SUCCEEDED)));
+
Review Comment:
Missing test: `refreshTable` with matching schema (no-op path). You test the schema-mismatch path but not the case where schemas match and no MSCK REPAIR is issued. Also missing: `refreshTable` with null `catalogTable` and `refreshTable` with null/empty schema.
##########
xtable-databricks/src/main/java/org/apache/xtable/databricks/DatabricksUnityCatalogSyncClient.java:
##########
@@ -0,0 +1,506 @@
+/*
+ * 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;
Review Comment:
Repeated validation logic for catalog/schema names in `hasDatabase`, `createDatabase`, and `getFullName`. Extract a shared method like `validateAndGetHierarchicalId(CatalogTableIdentifier)` that throws `CatalogSyncException` if catalog/schema are blank, and returns the `HierarchicalTableIdentifier`.
##########
website/sidebars.js:
##########
@@ -38,7 +38,7 @@ module.exports = {
items: [
'hms',
'glue-catalog',
- 'unity-catalog',
+ 'databricks-unity-catalog',
Review Comment:
The sidebar reference was changed from `unity-catalog` to `databricks-unity-catalog`, but the docs file is still named `unity-catalog.md` (not renamed). This will break the sidebar link. Either rename the md file or keep the sidebar reference as `unity-catalog`.
--
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
Re: [PR] feat: built-in databricks unity catalog sync [incubator-xtable]
Posted by "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