targetFormats =
+ Arrays.stream(cmd.getOptionValue(TARGETS).split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ String namespace = cmd.getOptionValue(NAMESPACE);
+
+ TableSyncSpec spec =
+ TableSyncSpec.builder()
+ .key(tableName)
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(cmd.getOptionValue(SOURCE_FORMAT).toUpperCase(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 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/XTableSyncService.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
new file mode 100644
index 000000000..6f9b62e19
--- /dev/null
+++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -0,0 +1,122 @@
+/*
+ * 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.hudi.HudiSourceConfig;
+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) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // The data files may live at a different path than the source table root (e.g. Iceberg keeps
+ // them under /data). Targets write their metadata alongside the data files, so the
+ // target base path is the data path (required by Hudi), defaulting to the source base path.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ List targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat ->
+ TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
+ .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..70d5c0fb5
--- /dev/null
+++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
@@ -0,0 +1,303 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.hudi.client.HoodieReadClient;
+
+import org.apache.xtable.GenericTable;
+import org.apache.xtable.hudi.HudiTestUtil;
+
+/**
+ * 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} for one case per direction, and
+ * asserting the target is data-equivalent to the source. Engines (and the Avro/Parquet versions
+ * they require) are supplied to the submit on a flat classpath via {@code
+ * spark.driver.extraClassPath} from {@code target/engine-classpath.txt}, never bundled; a flat
+ * classpath keeps a single Avro on one loader, matching a real cluster. Skipped when {@code
+ * SPARK_HOME} is not set.
+ *
+ * Directions are limited to Hudi and Iceberg: the root pom pins Delta 2.4.0 (Spark 3.4), which
+ * does not run on a Spark 3.5 {@code SPARK_HOME}. Delta conversion is covered in-process by {@code
+ * ITConversionController} (engine) and {@code ITXTableSyncListener} (listener).
+ */
+class ITXTableSparkRuntimeBundle {
+
+ @TempDir static java.nio.file.Path tempDir;
+
+ private static SparkSession sparkSession;
+ private static JavaSparkContext jsc;
+
+ @BeforeAll
+ static void setupSpark() {
+ sparkSession =
+ SparkSession.builder()
+ .config(HoodieReadClient.addHoodieSupport(HudiTestUtil.getSparkConf(tempDir)))
+ .getOrCreate();
+ jsc = JavaSparkContext.fromSparkContext(sparkSession.sparkContext());
+ }
+
+ @AfterAll
+ static void stopSpark() {
+ if (jsc != null) {
+ jsc.close();
+ }
+ if (sparkSession != null) {
+ sparkSession.stop();
+ }
+ }
+
+ private static Stream directions() {
+ // sourceFormat, targetFormat, hudiPartitionSpec (only used for a partitioned Hudi source)
+ return Stream.of(
+ Arguments.of("HUDI", "ICEBERG", "level:VALUE"), Arguments.of("ICEBERG", "HUDI", null));
+ }
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource("directions")
+ void bundleRunsSyncViaSparkSubmit(String sourceFormat, String targetFormat, String partitionSpec)
+ throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME is not set");
+
+ String tableName = sourceFormat.toLowerCase() + "_to_" + targetFormat.toLowerCase();
+ String basePath;
+ String dataPath;
+ List columns;
+ String orderByColumn;
+ try (GenericTable table =
+ GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, sourceFormat, true)) {
+ table.insertRows(100);
+ basePath = table.getBasePath();
+ // Targets write metadata alongside the data files; for Iceberg this is /data.
+ dataPath = table.getDataPath();
+ columns = table.getColumnsToSelect();
+ orderByColumn = table.getOrderByColumn();
+ }
+
+ File bundleJar = findBundleJar();
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " + sparkSubmit);
+
+ // Engines are provided on a FLAT classpath (single avro on one loader), NOT via --packages,
+ // which would layer them in a child classloader and break cross-boundary avro casts.
+ String engineClasspath = readEngineClasspath();
+
+ List command = new ArrayList<>();
+ command.add(sparkSubmit.getAbsolutePath());
+ command.add("--master");
+ command.add("local[2]");
+ command.add("--conf");
+ command.add("spark.driver.extraClassPath=" + engineClasspath);
+ command.add("--conf");
+ command.add("spark.executor.extraClassPath=" + engineClasspath);
+ command.add("--class");
+ command.add("org.apache.xtable.spark.XTableSparkSync");
+ command.add(bundleJar.getAbsolutePath());
+ command.add("--basePath");
+ command.add(basePath);
+ command.add("--dataPath");
+ command.add(dataPath);
+ command.add("--sourceFormat");
+ command.add(sourceFormat);
+ command.add("--targets");
+ command.add(targetFormat);
+ command.add("--tableName");
+ command.add(tableName);
+ if (partitionSpec != null) {
+ command.add("--partitionSpec");
+ command.add(partitionSpec);
+ }
+
+ ProcessBuilder pb = new ProcessBuilder(command);
+ 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);
+ }
+ assertEquals(0, process.exitValue(), "spark-submit failed. Output:\n" + output);
+
+ assertDatasetEquivalence(
+ sourceFormat, targetFormat, basePath, dataPath, columns, orderByColumn, output);
+ }
+
+ /**
+ * Reads the source (at its base path) and the target (at the data path, where the target metadata
+ * was written) back through Spark and asserts they are row-for-row equivalent, mirroring {@code
+ * ITConversionController.checkDatasetEquivalence}: a Hudi target is read with its metadata table
+ * enabled, and per-format column expressions normalize representation differences.
+ */
+ private void assertDatasetEquivalence(
+ String sourceFormat,
+ String targetFormat,
+ String basePath,
+ String dataPath,
+ List columns,
+ String orderByColumn,
+ String submitOutput) {
+ Dataset source =
+ sparkSession
+ .read()
+ .options(readOptions(sourceFormat))
+ .format(sourceFormat.toLowerCase())
+ .load(basePath)
+ .orderBy(orderByColumn);
+ // Compare scalar columns only. Nested/array/map read parity across writer and reader engines is
+ // covered in-process by ITConversionController; here it would exercise a Hudi-Parquet list
+ // encoding vs Iceberg reader quirk unrelated to whether the bundle jar runs the sync.
+ List scalarColumns = atomicColumns(source, columns);
+ List sourceRows =
+ source.selectExpr(selectColumns(scalarColumns, sourceFormat)).toJSON().collectAsList();
+ List targetRows =
+ sparkSession
+ .read()
+ .options(readOptions(targetFormat))
+ .format(targetFormat.toLowerCase())
+ .load(dataPath)
+ .orderBy(orderByColumn)
+ .selectExpr(selectColumns(scalarColumns, targetFormat))
+ .toJSON()
+ .collectAsList();
+ assertEquals(
+ 100,
+ targetRows.size(),
+ "Unexpected target row count. spark-submit output:\n" + submitOutput);
+ assertEquals(sourceRows, targetRows, "Target is not data-equivalent to the source");
+ }
+
+ // Keeps only the requested columns whose source type is atomic (not struct/array/map).
+ private static List atomicColumns(Dataset source, List columns) {
+ java.util.Set nonAtomic = new java.util.HashSet<>();
+ for (org.apache.spark.sql.types.StructField field : source.schema().fields()) {
+ org.apache.spark.sql.types.DataType type = field.dataType();
+ if (type instanceof org.apache.spark.sql.types.StructType
+ || type instanceof org.apache.spark.sql.types.ArrayType
+ || type instanceof org.apache.spark.sql.types.MapType) {
+ nonAtomic.add(field.name());
+ }
+ }
+ return columns.stream()
+ .filter(c -> !nonAtomic.contains(c))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ private static java.util.Map readOptions(String format) {
+ java.util.Map options = new java.util.HashMap<>();
+ if ("HUDI".equalsIgnoreCase(format)) {
+ options.put("hoodie.metadata.enable", "true");
+ options.put("hoodie.datasource.read.extract.partition.values.from.path", "true");
+ } else if ("ICEBERG".equalsIgnoreCase(format)) {
+ // Use the row-based reader: Iceberg's vectorized Arrow reader mis-casts timestamp columns
+ // written by Hudi (TimeStampMicroTZVector vs BigIntVector) for this cross-engine table.
+ options.put("vectorization-enabled", "false");
+ }
+ return options;
+ }
+
+ // Normalizes local-timestamp columns whose Hudi/Iceberg representations differ, matching
+ // ITConversionController.getSelectColumnsArr; other columns pass through unchanged.
+ private static String[] selectColumns(List columns, String format) {
+ boolean isHudi = "HUDI".equalsIgnoreCase(format);
+ boolean isIceberg = "ICEBERG".equalsIgnoreCase(format);
+ return columns.stream()
+ .map(
+ colName -> {
+ if (colName.startsWith("timestamp_local_millis")) {
+ if (isHudi) {
+ return String.format(
+ "unix_millis(CAST(%s AS TIMESTAMP)) AS %s", colName, colName);
+ } else if (isIceberg) {
+ return String.format("%s div 1000 AS %s", colName, colName);
+ }
+ return colName;
+ } else if (isHudi && colName.startsWith("timestamp_local_micros")) {
+ return String.format("unix_micros(CAST(%s AS TIMESTAMP)) AS %s", colName, colName);
+ }
+ return colName;
+ })
+ .toArray(String[]::new);
+ }
+
+ private String readEngineClasspath() throws Exception {
+ File cpFile = new File("target/engine-classpath.txt");
+ assertTrue(
+ cpFile.isFile(),
+ "engine classpath not built at "
+ + cpFile.getAbsolutePath()
+ + " (dependency:build-classpath)");
+ String cp =
+ new String(
+ java.nio.file.Files.readAllBytes(cpFile.toPath()),
+ java.nio.charset.StandardCharsets.UTF_8)
+ .trim();
+ assertTrue(!cp.isEmpty(), "engine classpath file is empty");
+ return cp;
+ }
+
+ 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();
+ }
+}