iterator = loader.iterator();
+ while (iterator.hasNext()) {
+ ConversionTarget target;
+ try {
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
+ // A registered target whose engine library is not on the classpath (e.g. Delta when only
+ // Hudi/Iceberg are provided). Skip it so a subset of engines can still be used; a missing
+ // engine for the requested format surfaces below as NotSupportedException.
+ log.debug(
+ "Skipping ConversionTarget whose engine is not available on the classpath", error);
+ continue;
+ }
if (target.getTableFormat().equalsIgnoreCase(tableFormatName)
&& isDeltaKernelTarget(target) == useKernel) {
return target;
@@ -96,7 +111,12 @@ && isDeltaKernelTarget(target) == useKernel) {
throw new NotSupportedException("Target format is not yet supported: " + tableFormatName);
}
+ private static final String DELTA_KERNEL_TARGET_CLASS =
+ "org.apache.xtable.kernel.DeltaKernelConversionTarget";
+
+ // Name-based to avoid loading DeltaKernelConversionTarget (and its io.delta.kernel dependencies)
+ // when Delta is not on the classpath; the target passed in is already loaded.
private static boolean isDeltaKernelTarget(ConversionTarget target) {
- return target instanceof DeltaKernelConversionTarget;
+ return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName());
}
}
diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml
new file mode 100644
index 000000000..8b7f1c8c1
--- /dev/null
+++ b/xtable-spark-runtime/pom.xml
@@ -0,0 +1,275 @@
+
+
+
+ 4.0.0
+
+
+ org.apache.xtable
+ xtable
+ 0.2.0-SNAPSHOT
+
+
+ xtable-spark-runtime_${scala.binary.version}
+ XTable Project Spark Runtime
+
+ Thin, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a
+ driver-side QueryExecutionListener, activated through spark.xtable.* config only.
+
+
+
+
+
+ org.apache.xtable
+ xtable-core_${scala.binary.version}
+ ${project.version}
+
+
+
+
+ org.apache.logging.log4j
+ log4j-api
+ provided
+
+
+
+
+ org.apache.spark
+ spark-sql_${scala.binary.version}
+ provided
+
+
+
+
+ org.apache.hadoop
+ hadoop-common
+ provided
+
+
+
+
+ org.apache.hudi
+ hudi-common
+ provided
+
+
+ org.apache.hudi
+ hudi-java-client
+ provided
+
+
+ org.apache.iceberg
+ iceberg-core
+ provided
+
+
+ org.apache.iceberg
+ iceberg-api
+ provided
+
+
+ io.delta
+ delta-core_${scala.binary.version}
+ provided
+
+
+ io.delta
+ delta-kernel-api
+ provided
+
+
+ io.delta
+ delta-kernel-defaults
+ provided
+
+
+
+
+ org.apache.xtable
+ xtable-core_${scala.binary.version}
+ ${project.version}
+ tests
+ test-jar
+ test
+
+
+
+ org.apache.hudi
+ hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}
+ test
+
+
+ org.openjdk.jol
+ jol-core
+ test
+
+
+
+ org.apache.iceberg
+ iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}
+ test
+
+
+ io.delta
+ delta-core_${scala.binary.version}
+ test
+
+
+ io.delta
+ delta-kernel-api
+ test
+
+
+ io.delta
+ delta-kernel-defaults
+ test
+
+
+ org.mockito
+ mockito-core
+ test
+
+
+ org.mockito
+ mockito-junit-jupiter
+ test
+
+
+
+
+ org.junit.jupiter
+ junit-jupiter-api
+ test
+
+
+ org.junit.jupiter
+ junit-jupiter-params
+ test
+
+
+ org.junit.jupiter
+ junit-jupiter-engine
+ test
+
+
+
+
+ org.apache.logging.log4j
+ log4j-core
+ test
+
+
+ org.apache.logging.log4j
+ log4j-slf4j2-impl
+ test
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+
+
+ org.apache.hudi:hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}:${hudi.version},org.apache.iceberg:iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}:${iceberg.version}
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-shade-plugin
+
+
+ package
+
+ shade
+
+
+ true
+ bundle
+ false
+
+
+ org.apache.xtable:xtable-api
+ org.apache.xtable:xtable-core_${scala.binary.version}
+ org.apache.xtable:xtable-hudi-support-utils
+ com.google.guava:guava
+ com.google.guava:failureaccess
+ com.google.protobuf:protobuf-java
+
+
+
+
+ com.google.common
+ org.apache.xtable.shaded.com.google.common
+
+
+ com.google.protobuf
+ org.apache.xtable.shaded.com.google.protobuf
+
+
+
+
+ *:*
+
+ META-INF/*.SF
+ META-INF/*.DSA
+ META-INF/*.RSA
+ **/Log4j2Plugins.dat
+
+
+
+
+
+
+
+
+ LICENSE
+ NOTICE
+ NOTICE.txt
+
+
+
+
+
+
+
+
+
+
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/PlanTargetResolver.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/PlanTargetResolver.java
new file mode 100644
index 000000000..09236b93d
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/PlanTargetResolver.java
@@ -0,0 +1,72 @@
+/*
+ * 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.spark;
+
+import java.net.URI;
+import java.util.Optional;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.spark.sql.catalyst.catalog.CatalogTable;
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.execution.QueryExecution;
+import org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand;
+import org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand;
+import org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand;
+
+/**
+ * Best-effort extraction of the base path written by a query, from its analyzed {@link
+ * LogicalPlan}.
+ *
+ * Returns the written path only for recognized write commands; reads and unrecognized plans
+ * yield an empty result and the caller skips them (so a read never triggers a sync). The fragile
+ * dependence on Spark internal command types is deliberately isolated here.
+ */
+@Log4j2
+public final class PlanTargetResolver {
+
+ private PlanTargetResolver() {}
+
+ /** Returns the output base path of a write command at the root of the analyzed plan, if known. */
+ public static Optional resolveWrittenPath(QueryExecution qe) {
+ if (qe == null) {
+ return Optional.empty();
+ }
+ try {
+ LogicalPlan plan = qe.analyzed();
+ if (plan instanceof InsertIntoHadoopFsRelationCommand) {
+ return Optional.ofNullable(((InsertIntoHadoopFsRelationCommand) plan).outputPath())
+ .map(Object::toString);
+ }
+ if (plan instanceof SaveIntoDataSourceCommand) {
+ scala.Option path = ((SaveIntoDataSourceCommand) plan).options().get("path");
+ return path.isDefined() ? Optional.of(path.get()) : Optional.empty();
+ }
+ if (plan instanceof CreateDataSourceTableAsSelectCommand) {
+ CatalogTable table = ((CreateDataSourceTableAsSelectCommand) plan).table();
+ scala.Option location = table.storage().locationUri();
+ return location.isDefined() ? Optional.of(location.get().toString()) : Optional.empty();
+ }
+ } catch (Throwable t) {
+ // Best-effort only: any failure to introspect the plan falls back to the all-tables path.
+ log.debug("Could not resolve written path from query plan; falling back to all tables", t);
+ }
+ return Optional.empty();
+ }
+}
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
new file mode 100644
index 000000000..5ae4ebe4c
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
@@ -0,0 +1,52 @@
+/*
+ * 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.spark;
+
+import java.util.List;
+
+import lombok.Builder;
+import lombok.NonNull;
+import lombok.Value;
+
+/**
+ * Immutable description of a single table to keep in sync, resolved from {@code spark.xtable.*}
+ * configuration. One {@link TableSyncSpec} is produced per key listed in {@code
+ * spark.xtable.tables}.
+ */
+@Value
+@Builder
+public class TableSyncSpec {
+ /** The logical key for this table (the token listed in {@code spark.xtable.tables}). */
+ @NonNull String key;
+
+ /** Absolute base path of the source table. */
+ @NonNull String basePath;
+
+ /** Optional path to the data files; defaults to {@link #basePath} downstream when null. */
+ String dataPath;
+
+ /** Optional namespace segments for the table. */
+ String[] namespace;
+
+ /** The source table format, e.g. {@code HUDI} (see {@code TableFormat}). */
+ @NonNull String sourceFormat;
+
+ /** The target formats to sync to, e.g. {@code [ICEBERG, DELTA]}. */
+ @NonNull List targets;
+}
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkConfig.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkConfig.java
new file mode 100644
index 000000000..89b7ddc8f
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkConfig.java
@@ -0,0 +1,120 @@
+/*
+ * 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.spark;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
+
+/**
+ * Parses {@code spark.xtable.*} configuration into a list of {@link TableSyncSpec}.
+ *
+ * Schema:
+ *
+ *
+ * - {@code spark.xtable.tables} - comma-separated per-table keys.
+ *
- {@code spark.xtable..basePath} - source base path (path-based selection), OR
+ *
- {@code spark.xtable..sourceTable} - {@code db.table} (name-based; resolved to a base
+ * path via the supplied {@code nameResolver}).
+ *
- {@code spark.xtable..sourceFormat} - e.g. {@code HUDI}.
+ *
- {@code spark.xtable..targets} - comma-separated target formats, e.g. {@code
+ * ICEBERG,DELTA}.
+ *
- {@code spark.xtable..dataPath} (optional), {@code spark.xtable..namespace}
+ * (optional, dot-separated).
+ *
+ */
+public final class XTableSparkConfig {
+ static final String PREFIX = "spark.xtable.";
+ static final String TABLES_KEY = PREFIX + "tables";
+
+ private XTableSparkConfig() {}
+
+ /**
+ * @param conf resolves a config key to its value, or {@code null} if unset (e.g. backed by
+ * Spark's {@code RuntimeConfig}).
+ * @param nameResolver resolves a {@code db.table} identifier to an absolute base path; only
+ * invoked for name-based table entries.
+ */
+ public static List parse(
+ UnaryOperator conf, UnaryOperator nameResolver) {
+ String tables = conf.apply(TABLES_KEY);
+ if (isBlank(tables)) {
+ return java.util.Collections.emptyList();
+ }
+ List specs = new ArrayList<>();
+ for (String rawKey : tables.split(",")) {
+ String key = rawKey.trim();
+ if (key.isEmpty()) {
+ continue;
+ }
+ specs.add(parseTable(key, conf, nameResolver));
+ }
+ return specs;
+ }
+
+ private static TableSyncSpec parseTable(
+ String key, UnaryOperator conf, UnaryOperator nameResolver) {
+ String p = PREFIX + key + ".";
+ String basePath = conf.apply(p + "basePath");
+ String sourceTableName = conf.apply(p + "sourceTable");
+ if (isBlank(basePath) && !isBlank(sourceTableName)) {
+ basePath = nameResolver.apply(sourceTableName.trim());
+ }
+ require(!isBlank(basePath), key, "requires '" + p + "basePath' or '" + p + "sourceTable'");
+
+ String sourceFormat = conf.apply(p + "sourceFormat");
+ require(!isBlank(sourceFormat), key, "requires '" + p + "sourceFormat'");
+
+ String targetsStr = conf.apply(p + "targets");
+ require(!isBlank(targetsStr), key, "requires '" + p + "targets'");
+ List targets =
+ Arrays.stream(targetsStr.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(java.util.Locale.ROOT))
+ .collect(Collectors.toList());
+ require(!targets.isEmpty(), key, "requires at least one target in '" + p + "targets'");
+
+ String dataPath = conf.apply(p + "dataPath");
+ String namespaceStr = conf.apply(p + "namespace");
+ String[] namespace = isBlank(namespaceStr) ? null : namespaceStr.trim().split("\\.");
+
+ return TableSyncSpec.builder()
+ .key(key)
+ .basePath(basePath.trim())
+ .dataPath(isBlank(dataPath) ? null : dataPath.trim())
+ .namespace(namespace)
+ .sourceFormat(sourceFormat.trim().toUpperCase(java.util.Locale.ROOT))
+ .targets(targets)
+ .build();
+ }
+
+ private static void require(boolean condition, String key, String message) {
+ if (!condition) {
+ throw new IllegalArgumentException(
+ "Invalid spark.xtable config for table '" + key + "': " + message);
+ }
+ }
+
+ private static boolean isBlank(String s) {
+ return s == null || s.trim().isEmpty();
+ }
+}
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
new file mode 100644
index 000000000..5c7555f73
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
@@ -0,0 +1,116 @@
+/*
+ * 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.spark;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real Spark install.
+ *
+ *
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basePath /warehouse/db/orders --sourceFormat HUDI --targets ICEBERG,DELTA
+ *
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ Map opts = parseArgs(args);
+ String basePath = required(opts, "basePath");
+ String sourceFormat = required(opts, "sourceFormat");
+ String targets = required(opts, "targets");
+ String tableName = opts.getOrDefault("tableName", basePathToName(basePath));
+
+ List targetFormats =
+ Arrays.stream(targets.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(java.util.Locale.ROOT))
+ .collect(Collectors.toList());
+
+ TableSyncSpec spec =
+ TableSyncSpec.builder()
+ .key(tableName)
+ .basePath(basePath)
+ .dataPath(opts.get("dataPath"))
+ .namespace(opts.containsKey("namespace") ? opts.get("namespace").split("\\.") : null)
+ .sourceFormat(sourceFormat.toUpperCase(java.util.Locale.ROOT))
+ .targets(targetFormats)
+ .build();
+
+ SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+ log.info("Starting standalone XTable sync for {}", spec.getBasePath());
+ new XTableSyncService().sync(spec, hadoopConf);
+ log.info("Completed XTable sync for {}", spec.getBasePath());
+ } finally {
+ spark.stop();
+ }
+ }
+
+ private static Map parseArgs(String[] args) {
+ Map opts = new HashMap<>();
+ List tokens = new ArrayList<>(Arrays.asList(args));
+ for (int i = 0; i < tokens.size(); i++) {
+ String token = tokens.get(i);
+ if (token.startsWith("--")) {
+ String key = token.substring(2);
+ if (i + 1 < tokens.size() && !tokens.get(i + 1).startsWith("--")) {
+ opts.put(key, tokens.get(++i));
+ } else {
+ opts.put(key, "true");
+ }
+ }
+ }
+ return opts;
+ }
+
+ private static String required(Map opts, String key) {
+ String value = opts.get(key);
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException("Missing required argument: --" + key);
+ }
+ return value.trim();
+ }
+
+ private static String basePathToName(String basePath) {
+ String trimmed =
+ basePath.endsWith("/") ? basePath.substring(0, basePath.length() - 1) : basePath;
+ int idx = trimmed.lastIndexOf('/');
+ return idx >= 0 ? trimmed.substring(idx + 1) : trimmed;
+ }
+}
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncListener.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncListener.java
new file mode 100644
index 000000000..d3e854583
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncListener.java
@@ -0,0 +1,177 @@
+/*
+ * 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.spark;
+
+import java.net.URI;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.catalyst.TableIdentifier;
+import org.apache.spark.sql.catalyst.catalog.CatalogTable;
+import org.apache.spark.sql.execution.QueryExecution;
+import org.apache.spark.sql.util.QueryExecutionListener;
+
+/**
+ * Driver-side {@link QueryExecutionListener} that runs an incremental XTable sync after a
+ * successful write to a configured source table. Register it via {@code
+ * spark.sql.queryExecutionListeners=org.apache.xtable.spark.XTableSyncListener} and declare the
+ * tables through {@code spark.xtable.*} (see {@link XTableSparkConfig}).
+ *
+ * The listener is stateless apart from the immutable parsed config: the sync watermark lives in
+ * the target's metadata, and the sync runs inline on the callback (see RFC-3 for the execution
+ * model). It triggers only on recognized write commands whose output path matches a configured
+ * table, so reads and writes to other tables are ignored.
+ */
+@Log4j2
+public class XTableSyncListener implements QueryExecutionListener {
+
+ private static final ThreadLocal SYNCING = ThreadLocal.withInitial(() -> Boolean.FALSE);
+
+ private final XTableSyncService syncService = new XTableSyncService();
+ private volatile List specs;
+
+ @Override
+ public void onSuccess(String funcName, QueryExecution qe, long durationNs) {
+ if (SYNCING.get()) {
+ // Ignore writes generated by our own sync running inline on this thread.
+ return;
+ }
+ Optional writtenPath = PlanTargetResolver.resolveWrittenPath(qe);
+ if (!writtenPath.isPresent()) {
+ return;
+ }
+ SparkSession session = activeSession();
+ if (session == null) {
+ return;
+ }
+ List matched =
+ specs(session).stream()
+ .filter(spec -> pathsMatch(writtenPath.get(), spec.getBasePath()))
+ .collect(Collectors.toList());
+ if (matched.isEmpty()) {
+ log.debug("Write to {} did not match any configured spark.xtable table", writtenPath.get());
+ return;
+ }
+ Configuration hadoopConf = session.sparkContext().hadoopConfiguration();
+ SYNCING.set(Boolean.TRUE);
+ try {
+ for (TableSyncSpec spec : matched) {
+ try {
+ syncService.sync(spec, hadoopConf);
+ } catch (Throwable t) {
+ // Isolate failures so one table does not stop the others and a listener never
+ // destabilizes Spark's listener bus; the next commit self-heals.
+ log.error("XTable sync failed for table {} at {}", spec.getKey(), spec.getBasePath(), t);
+ }
+ }
+ } finally {
+ SYNCING.remove();
+ }
+ }
+
+ @Override
+ public void onFailure(String funcName, QueryExecution qe, Exception exception) {
+ // No-op: only successful writes trigger a sync.
+ }
+
+ private List specs(SparkSession session) {
+ List local = specs;
+ if (local == null) {
+ synchronized (this) {
+ local = specs;
+ if (local == null) {
+ local = XTableSparkConfig.parse(confGetter(session), nameResolver(session));
+ specs = local;
+ log.info(
+ "XTable spark-runtime configured for {} table(s): {}",
+ local.size(),
+ local.stream().map(TableSyncSpec::getKey).collect(Collectors.toList()));
+ }
+ }
+ }
+ return local;
+ }
+
+ private static UnaryOperator confGetter(SparkSession session) {
+ return key -> {
+ try {
+ return session.conf().get(key, null);
+ } catch (Exception e) {
+ return null;
+ }
+ };
+ }
+
+ private static UnaryOperator nameResolver(SparkSession session) {
+ return name -> {
+ try {
+ TableIdentifier id = session.sessionState().sqlParser().parseTableIdentifier(name);
+ CatalogTable meta = session.sessionState().catalog().getTableMetadata(id);
+ scala.Option location = meta.storage().locationUri();
+ if (location.isDefined()) {
+ return location.get().toString();
+ }
+ throw new IllegalArgumentException("No location URI for table '" + name + "'");
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to resolve table '" + name + "' via catalog", e);
+ }
+ };
+ }
+
+ private static SparkSession activeSession() {
+ scala.Option active = SparkSession.getActiveSession();
+ if (active.isDefined()) {
+ return active.get();
+ }
+ scala.Option dflt = SparkSession.getDefaultSession();
+ return dflt.isDefined() ? dflt.get() : null;
+ }
+
+ /** Lenient path match ignoring scheme, trailing slashes, and partition sub-paths. */
+ static boolean pathsMatch(String written, String configured) {
+ URI a = new Path(written).toUri();
+ URI b = new Path(configured).toUri();
+ String pa = stripTrailingSlash(a.getPath());
+ String pb = stripTrailingSlash(b.getPath());
+ boolean pathRelated = pa.equals(pb) || pa.startsWith(pb + "/") || pb.startsWith(pa + "/");
+ if (!pathRelated) {
+ return false;
+ }
+ if (a.getAuthority() != null && b.getAuthority() != null) {
+ return a.getAuthority().equals(b.getAuthority());
+ }
+ return true;
+ }
+
+ private static String stripTrailingSlash(String p) {
+ if (p == null) {
+ return "";
+ }
+ return (p.length() > 1 && p.endsWith("/")) ? p.substring(0, p.length() - 1) : p;
+ }
+}
diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
new file mode 100644
index 000000000..2bf1f7394
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -0,0 +1,110 @@
+/*
+ * 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.spark;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+
+/**
+ * Builds a {@link ConversionConfig} from a {@link TableSyncSpec} and runs an incremental {@link
+ * ConversionController#sync} for it. This is the unit of work triggered by {@link
+ * XTableSyncListener} after a successful write.
+ *
+ * The sync watermark is persisted in the target's {@code TableSyncMetadata}, and {@link
+ * SyncMode#INCREMENTAL} auto-falls back to a full snapshot when incremental is not safe (e.g. the
+ * very first sync), so this call is idempotent and self-healing.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map sync(TableSyncSpec spec, Configuration hadoopConf) {
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(spec.getDataPath())
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(new Properties())
+ .build();
+
+ List targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat ->
+ TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .build())
+ .collect(Collectors.toList());
+
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+
+ ConversionSourceProvider sourceProvider = sourceProviderFor(spec.getSourceFormat());
+ sourceProvider.init(hadoopConf);
+
+ log.info(
+ "Running XTable sync for table {} ({} -> {}) at {}",
+ spec.getKey(),
+ spec.getSourceFormat(),
+ spec.getTargets(),
+ spec.getBasePath());
+ return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider);
+ }
+
+ private static ConversionSourceProvider sourceProviderFor(String sourceFormat) {
+ switch (sourceFormat.toUpperCase()) {
+ case TableFormat.HUDI:
+ return new HudiConversionSourceProvider();
+ case TableFormat.DELTA:
+ return new DeltaConversionSourceProvider();
+ case TableFormat.ICEBERG:
+ return new IcebergConversionSourceProvider();
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported source format for spark-runtime sync: " + sourceFormat);
+ }
+ }
+}
diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
new file mode 100644
index 000000000..41568edf9
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
@@ -0,0 +1,137 @@
+/*
+ * 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.spark;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.File;
+import java.net.URI;
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.hudi.common.model.HoodieTableType;
+
+import org.apache.xtable.GenericTable;
+import org.apache.xtable.TestJavaHudiTable;
+
+/**
+ * Validates that the relocated {@code xtable-spark-runtime} bundle jar is self-contained by running
+ * an actual XTable sync via {@code $SPARK_HOME/bin/spark-submit} against a locally created Hudi
+ * table. Skipped when {@code SPARK_HOME} is not set.
+ */
+class ITXTableSparkRuntimeBundle {
+
+ @TempDir Path tempDir;
+
+ @Test
+ void bundleRunsHudiToIcebergSyncViaSparkSubmit() throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME is not set");
+
+ String tableName = "orders";
+ String basePath;
+ try (GenericTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+ table.insertRows(10);
+ basePath = table.getBasePath();
+ }
+
+ File bundleJar = findBundleJar();
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " + sparkSubmit);
+
+ // Engines are provided by the user's runtime, not bundled. Supply Hudi + Iceberg (matching the
+ // versions from the root pom) via --packages, exactly as a user would. Delta is intentionally
+ // absent: resilient target discovery skips it.
+ String enginePackages = System.getProperty("xtable.it.enginePackages");
+ assertTrue(
+ enginePackages != null && !enginePackages.isEmpty(),
+ "xtable.it.enginePackages system property must be set by the failsafe plugin");
+
+ ProcessBuilder pb =
+ new ProcessBuilder(
+ sparkSubmit.getAbsolutePath(),
+ "--master",
+ "local[2]",
+ "--packages",
+ enginePackages,
+ "--class",
+ "org.apache.xtable.spark.XTableSparkSync",
+ bundleJar.getAbsolutePath(),
+ "--basePath",
+ basePath,
+ "--sourceFormat",
+ "HUDI",
+ "--targets",
+ "ICEBERG",
+ "--tableName",
+ tableName);
+ pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1");
+ pb.redirectErrorStream(true);
+
+ Process process = pb.start();
+ String output = readOutput(process);
+ boolean finished = process.waitFor(10, TimeUnit.MINUTES);
+ if (!finished) {
+ process.destroyForcibly();
+ throw new AssertionError("spark-submit timed out. Output:\n" + output);
+ }
+ int exit = process.exitValue();
+ assertEquals(0, exit, "spark-submit failed (exit " + exit + "). Output:\n" + output);
+
+ File metadataDir = new File(new File(URI.create(basePath)), "metadata");
+ File[] metadataJson = metadataDir.listFiles((dir, name) -> name.endsWith("metadata.json"));
+ assertTrue(
+ metadataJson != null && metadataJson.length > 0,
+ "No Iceberg metadata produced under " + metadataDir + ". spark-submit output:\n" + output);
+ }
+
+ private File findBundleJar() {
+ File targetDir = new File("target");
+ File[] candidates =
+ targetDir.listFiles(
+ (dir, name) -> name.startsWith("xtable-spark-runtime") && name.endsWith("-bundle.jar"));
+ assertTrue(
+ candidates != null && candidates.length == 1,
+ "Expected exactly one bundle jar in "
+ + targetDir.getAbsolutePath()
+ + " (run the package phase first)");
+ return candidates[0];
+ }
+
+ private static String readOutput(Process process) throws Exception {
+ StringBuilder sb = new StringBuilder();
+ try (java.io.BufferedReader reader =
+ new java.io.BufferedReader(
+ new java.io.InputStreamReader(
+ process.getInputStream(), java.nio.charset.StandardCharsets.UTF_8))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ sb.append(line).append(System.lineSeparator());
+ }
+ }
+ return sb.toString();
+ }
+}
diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSyncListener.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSyncListener.java
new file mode 100644
index 000000000..24adfc220
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSyncListener.java
@@ -0,0 +1,147 @@
+/*
+ * 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.spark;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.hudi.client.HoodieReadClient;
+
+import org.apache.xtable.hudi.HudiTestUtil;
+
+/**
+ * End-to-end test: a Hudi write through the Spark datasource triggers {@link XTableSyncListener},
+ * which synchronizes the table to Delta and Iceberg in the same driver JVM.
+ */
+class ITXTableSyncListener {
+
+ @TempDir static java.nio.file.Path tempDir;
+
+ private SparkSession spark;
+
+ @AfterEach
+ void stopSpark() {
+ if (spark != null) {
+ spark.stop();
+ spark = null;
+ }
+ SparkSession.clearActiveSession();
+ SparkSession.clearDefaultSession();
+ }
+
+ @Test
+ void hudiWriteSyncsToDeltaAndIceberg() throws Exception {
+ String basePath = tempDir.resolve("orders").toUri().toString();
+
+ SparkConf sparkConf =
+ HudiTestUtil.getSparkConf(tempDir)
+ .set("spark.sql.queryExecutionListeners", XTableSyncListener.class.getName())
+ .set("spark.xtable.tables", "orders")
+ .set("spark.xtable.orders.basePath", basePath)
+ .set("spark.xtable.orders.sourceFormat", "HUDI")
+ .set("spark.xtable.orders.targets", "DELTA,ICEBERG");
+ spark =
+ SparkSession.builder().config(HoodieReadClient.addHoodieSupport(sparkConf)).getOrCreate();
+
+ List rows =
+ Arrays.asList(
+ RowFactory.create("1", "alice", 1L),
+ RowFactory.create("2", "bob", 2L),
+ RowFactory.create("3", "carol", 3L),
+ RowFactory.create("4", "dave", 4L),
+ RowFactory.create("5", "erin", 5L));
+ StructType schema =
+ new StructType()
+ .add("id", DataTypes.StringType, false)
+ .add("name", DataTypes.StringType, true)
+ .add("ts", DataTypes.LongType, false);
+
+ // Write through the Hudi Spark datasource so the write flows through a QueryExecution and fires
+ // the listener (unlike the write-client-based test helpers).
+ spark
+ .createDataFrame(rows, schema)
+ .write()
+ .format("hudi")
+ .option("hoodie.table.name", "orders")
+ .option("hoodie.datasource.write.recordkey.field", "id")
+ .option("hoodie.datasource.write.precombine.field", "ts")
+ .option("hoodie.datasource.write.partitionpath.field", "")
+ .option(
+ "hoodie.datasource.write.keygenerator.class",
+ "org.apache.hudi.keygen.NonpartitionedKeyGenerator")
+ .option("hoodie.datasource.write.operation", "insert")
+ .mode(SaveMode.Append)
+ .save(basePath);
+
+ // Callback delivery is async even though the sync itself runs inline, so wait for the targets.
+ waitForTargets(basePath);
+
+ Set expectedIds = rows.stream().map(r -> r.getString(0)).collect(Collectors.toSet());
+ assertEquals(expectedIds, idsFrom(basePath, "delta"));
+ assertEquals(expectedIds, idsFrom(basePath, "iceberg"));
+ }
+
+ private Set idsFrom(String basePath, String format) {
+ Dataset df = spark.read().format(format).load(basePath);
+ return df.select("id").collectAsList().stream()
+ .map(r -> r.getString(0))
+ .collect(Collectors.toSet());
+ }
+
+ private static void waitForTargets(String basePath) throws InterruptedException {
+ File local = new File(java.net.URI.create(basePath));
+ File deltaLog = new File(local, "_delta_log");
+ File icebergMetadata = new File(local, "metadata");
+ long deadline = System.currentTimeMillis() + 120_000;
+ while (System.currentTimeMillis() < deadline) {
+ if (hasDeltaCommit(deltaLog) && hasIcebergMetadata(icebergMetadata)) {
+ return;
+ }
+ Thread.sleep(2_000);
+ }
+ throw new AssertionError("Timed out waiting for Delta/Iceberg metadata under " + basePath);
+ }
+
+ private static boolean hasDeltaCommit(File deltaLog) {
+ File[] files = deltaLog.listFiles((dir, name) -> name.endsWith(".json"));
+ return files != null && files.length > 0;
+ }
+
+ private static boolean hasIcebergMetadata(File metadataDir) {
+ File[] files = metadataDir.listFiles((dir, name) -> name.endsWith("metadata.json"));
+ return files != null && files.length > 0;
+ }
+}
diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestPlanTargetResolver.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestPlanTargetResolver.java
new file mode 100644
index 000000000..853bb7226
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestPlanTargetResolver.java
@@ -0,0 +1,44 @@
+/*
+ * 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.spark;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
+import org.apache.spark.sql.execution.QueryExecution;
+import org.junit.jupiter.api.Test;
+
+class TestPlanTargetResolver {
+
+ @Test
+ void nullQueryExecutionYieldsEmpty() {
+ assertFalse(PlanTargetResolver.resolveWrittenPath(null).isPresent());
+ }
+
+ @Test
+ void nonWritePlanYieldsEmpty() {
+ QueryExecution qe = mock(QueryExecution.class);
+ LogicalPlan plan = mock(LogicalPlan.class);
+ when(qe.analyzed()).thenReturn(plan);
+ // A read/aggregate plan is not one of the recognized write commands.
+ assertFalse(PlanTargetResolver.resolveWrittenPath(qe).isPresent());
+ }
+}
diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSparkConfig.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSparkConfig.java
new file mode 100644
index 000000000..52f68d78f
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSparkConfig.java
@@ -0,0 +1,158 @@
+/*
+ * 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.spark;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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 java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.UnaryOperator;
+
+import org.junit.jupiter.api.Test;
+
+class TestXTableSparkConfig {
+
+ private static final UnaryOperator FAIL_RESOLVER =
+ name -> {
+ throw new AssertionError("nameResolver should not be called: " + name);
+ };
+
+ private static UnaryOperator confOf(Map map) {
+ return map::get;
+ }
+
+ @Test
+ void parsesPathBasedTableWithMultipleTargets() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.basePath", "/warehouse/db/orders");
+ conf.put("spark.xtable.orders.sourceFormat", "hudi");
+ conf.put("spark.xtable.orders.targets", "iceberg, delta");
+
+ List specs = XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER);
+
+ assertEquals(1, specs.size());
+ TableSyncSpec spec = specs.get(0);
+ assertEquals("orders", spec.getKey());
+ assertEquals("/warehouse/db/orders", spec.getBasePath());
+ assertEquals("HUDI", spec.getSourceFormat());
+ assertEquals(Arrays.asList("ICEBERG", "DELTA"), spec.getTargets());
+ assertNull(spec.getDataPath());
+ assertNull(spec.getNamespace());
+ }
+
+ @Test
+ void parsesMultipleTablesWithOptionalFields() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders , customers");
+ conf.put("spark.xtable.orders.basePath", "/wh/orders");
+ conf.put("spark.xtable.orders.sourceFormat", "HUDI");
+ conf.put("spark.xtable.orders.targets", "ICEBERG");
+ conf.put("spark.xtable.orders.dataPath", "/wh/orders/data");
+ conf.put("spark.xtable.orders.namespace", "prod.sales");
+ conf.put("spark.xtable.customers.basePath", "/wh/customers");
+ conf.put("spark.xtable.customers.sourceFormat", "HUDI");
+ conf.put("spark.xtable.customers.targets", "DELTA");
+
+ List specs = XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER);
+
+ assertEquals(2, specs.size());
+ TableSyncSpec orders = specs.get(0);
+ assertEquals("/wh/orders/data", orders.getDataPath());
+ assertArrayEquals(new String[] {"prod", "sales"}, orders.getNamespace());
+ assertEquals("customers", specs.get(1).getKey());
+ }
+
+ @Test
+ void resolvesNameBasedTableViaResolver() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.sourceTable", "db.orders");
+ conf.put("spark.xtable.orders.sourceFormat", "HUDI");
+ conf.put("spark.xtable.orders.targets", "ICEBERG");
+
+ UnaryOperator resolver =
+ name -> "db.orders".equals(name) ? "s3://bucket/warehouse/orders" : null;
+
+ List specs = XTableSparkConfig.parse(confOf(conf), resolver);
+
+ assertEquals(1, specs.size());
+ assertEquals("s3://bucket/warehouse/orders", specs.get(0).getBasePath());
+ }
+
+ @Test
+ void basePathTakesPrecedenceOverNameSoResolverNotCalled() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.basePath", "/wh/orders");
+ conf.put("spark.xtable.orders.sourceTable", "db.orders");
+ conf.put("spark.xtable.orders.sourceFormat", "HUDI");
+ conf.put("spark.xtable.orders.targets", "ICEBERG");
+
+ List specs = XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER);
+ assertEquals("/wh/orders", specs.get(0).getBasePath());
+ }
+
+ @Test
+ void emptyOrMissingTablesYieldsEmptyList() {
+ assertTrue(XTableSparkConfig.parse(confOf(new HashMap<>()), FAIL_RESOLVER).isEmpty());
+ Map blank = new HashMap<>();
+ blank.put("spark.xtable.tables", " ");
+ assertTrue(XTableSparkConfig.parse(confOf(blank), FAIL_RESOLVER).isEmpty());
+ }
+
+ @Test
+ void missingBasePathAndSourceTableThrows() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.sourceFormat", "HUDI");
+ conf.put("spark.xtable.orders.targets", "ICEBERG");
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER));
+ assertTrue(e.getMessage().contains("basePath"));
+ }
+
+ @Test
+ void missingSourceFormatThrows() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.basePath", "/wh/orders");
+ conf.put("spark.xtable.orders.targets", "ICEBERG");
+ assertThrows(
+ IllegalArgumentException.class, () -> XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER));
+ }
+
+ @Test
+ void missingTargetsThrows() {
+ Map conf = new HashMap<>();
+ conf.put("spark.xtable.tables", "orders");
+ conf.put("spark.xtable.orders.basePath", "/wh/orders");
+ conf.put("spark.xtable.orders.sourceFormat", "HUDI");
+ assertThrows(
+ IllegalArgumentException.class, () -> XTableSparkConfig.parse(confOf(conf), FAIL_RESOLVER));
+ }
+}
diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSyncListener.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSyncListener.java
new file mode 100644
index 000000000..57a4b3bd7
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/TestXTableSyncListener.java
@@ -0,0 +1,62 @@
+/*
+ * 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.spark;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+
+class TestXTableSyncListener {
+
+ @Test
+ void matchesIdenticalAndTrailingSlash() {
+ assertTrue(XTableSyncListener.pathsMatch("file:///wh/orders", "file:///wh/orders"));
+ assertTrue(XTableSyncListener.pathsMatch("file:///wh/orders/", "file:///wh/orders"));
+ }
+
+ @Test
+ void matchesPartitionSubPathEitherDirection() {
+ assertTrue(XTableSyncListener.pathsMatch("file:///wh/orders/dt=2024", "file:///wh/orders"));
+ assertTrue(XTableSyncListener.pathsMatch("file:///wh/orders", "file:///wh/orders/dt=2024"));
+ }
+
+ @Test
+ void matchesAcrossSchemeWhenOneAuthorityMissing() {
+ assertTrue(XTableSyncListener.pathsMatch("file:/wh/orders", "/wh/orders"));
+ }
+
+ @Test
+ void doesNotMatchDifferentPaths() {
+ assertFalse(XTableSyncListener.pathsMatch("file:///wh/orders", "file:///wh/customers"));
+ // sibling prefix that is not a path boundary must not match
+ assertFalse(XTableSyncListener.pathsMatch("file:///wh/orders_v2", "file:///wh/orders"));
+ }
+
+ @Test
+ void doesNotMatchDifferentAuthorities() {
+ assertFalse(XTableSyncListener.pathsMatch("s3://bucketA/wh/orders", "s3://bucketB/wh/orders"));
+ }
+
+ @Test
+ void matchesSameAuthority() {
+ assertTrue(
+ XTableSyncListener.pathsMatch("s3://bucket/wh/orders/dt=1", "s3://bucket/wh/orders"));
+ }
+}