This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new 92003282 [836] Add xtable-spark-runtime, a thin drop-in Spark bundle
(#964)
92003282 is described below
commit 92003282425ede3e59193a8bf9b30987b18ff9fe
Author: Vinish Reddy <[email protected]>
AuthorDate: Mon Oct 19 06:28:54 2026 +0530
[836] Add xtable-spark-runtime, a thin drop-in Spark bundle (#964)
* [836] Add xtable-spark-runtime: thin drop-in Spark bundle on main (Hudi
1.2.1)
Ports the reviewed branch-0.4 module (#843) to main on Hudi 1.2.1:
- xtable-spark-runtime: a ~4 MB relocated bundle (xtable-api, xtable-core,
xtable-hudi-support-utils, guava, commons-cli, snakeyaml) with every
engine
dependency provided, a spark-submit entry point (XTableSparkSync) and an
in-job API (XTableSyncService / TableSyncSpec).
- ConversionTargetFactory: skip registered targets whose engine is absent
from the classpath (ServiceConfigurationError / LinkageError from both
hasNext() and next()) and resolve the Delta Kernel target by class name.
- spark-runtime-validation workflow: spark-submits the shaded bundle via
ITXTableSparkRuntimeBundle on Spark 3.4 and 3.5.
- Root pom: register the module and exclude the shade-generated
dependency-reduced-pom.xml from spotless; gitignore that file.
Differences from #843: hudi.version and HoodieCleanStat.builder() already
landed on main via #835; the module overrides the parent's shade filters
(combine.self="override") because Maven merges list entries by position and
the parent's awssdk filter otherwise drops every snakeyaml class.
* Address review feedback
- Fail the batch when a target returns a non-SUCCESS sync status, not only
on exceptions
- Ship the checker-qual license text and validate the spark-runtime main
jar in License Check
- Test target discovery and NotSupportedException when an engine is missing
- Drop the explicit tests classifier on the xtable-core test-jar, which the
build cache extension resolves as a .test-jar file
* Address review feedback
* Write non-Hudi targets at the table base path
* Test Iceberg with a separate data path syncing to Hudi
---
.github/workflows/mvn-license-check.yml | 4 +-
.github/workflows/spark-runtime-validation.yml | 86 ++++
.gitignore | 3 +
pom.xml | 3 +
release/scripts/validate_bundled_license_texts.py | 16 +-
.../conversion/TestConversionTargetFactory.java | 154 +++++++
xtable-spark-runtime/pom.xml | 324 ++++++++++++++
.../org/apache/xtable/spark/TableSyncSpec.java | 54 +++
.../org/apache/xtable/spark/XTableSparkSync.java | 472 +++++++++++++++++++++
.../org/apache/xtable/spark/XTableSyncService.java | 143 +++++++
.../src/main/resources/META-INF/LICENSE-bundled | 221 ++++++++++
.../src/main/resources/META-INF/NOTICE-bundled | 22 +
.../META-INF/licenses/LICENSE-checker-qual | 22 +
.../xtable/spark/ITXTableSparkRuntimeBundle.java | 387 +++++++++++++++++
.../apache/xtable/spark/XTableSparkSyncTest.java | 255 +++++++++++
15 files changed, 2162 insertions(+), 4 deletions(-)
diff --git a/.github/workflows/mvn-license-check.yml
b/.github/workflows/mvn-license-check.yml
index 4a29da3b..6a2e6843 100644
--- a/.github/workflows/mvn-license-check.yml
+++ b/.github/workflows/mvn-license-check.yml
@@ -100,7 +100,7 @@ jobs:
- name: Build Shaded Bundles
run: >
./mvnw -B -DskipTests -Dmaven.build.cache.enabled=false
- -pl
xtable-aws,xtable-hive-metastore,xtable-hudi-support/xtable-hudi-support-extensions
+ -pl
xtable-aws,xtable-hive-metastore,xtable-hudi-support/xtable-hudi-support-extensions,xtable-spark-runtime
-am package
dependency:tree -DoutputType=text
-DoutputFile=target/dependency-tree.txt
@@ -110,7 +110,7 @@ jobs:
# Nothing above this line resolves xtable-utilities or xtable-service:
# `apache-rat:check` requires no dependency resolution, `dependency:tree`
# collects poms rather than jars, and the shaded-bundle build reaches
only
- # the `-am` closure of the three modules it names, which excludes both
+ # the `-am` closure of the four modules it names, which excludes both
# leaves. Without this step the entry never gains the Quarkus stack or
# xtable-utilities' bundled dependencies, and Maven CI Build - which
# cannot write the cache - would re-download them from Central on every
diff --git a/.github/workflows/spark-runtime-validation.yml
b/.github/workflows/spark-runtime-validation.yml
new file mode 100644
index 00000000..2741ab5e
--- /dev/null
+++ b/.github/workflows/spark-runtime-validation.yml
@@ -0,0 +1,86 @@
+#
+# 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.
+#
+name: Spark Runtime Bundle Validation
+
+# Runs ITXTableSparkRuntimeBundle, which spark-submits the shaded bundle. The
IT skips itself
+# unless SPARK_HOME is set, so it only runs here.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# No path filter, so the check always reports and can be a required status
check.
+on:
+ push:
+ branches:
+ - "main"
+
+ pull_request:
+ branches:
+ - "main"
+
+ workflow_dispatch:
+
+env:
+ SPARK_HADOOP_PROFILE: "hadoop3"
+
+jobs:
+ validate-spark-runtime-bundle:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ # 3.4: Delta via delta-core. 3.5: Delta auto-switches to the
Spark-free Delta Kernel.
+ spark_version: ["3.4.3", "3.5.9"]
+ env:
+ SPARK_VERSION: ${{ matrix.spark_version }}
+ steps:
+ - uses: actions/checkout@v6
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v5
+ with:
+ java-version: '11'
+ distribution: 'temurin'
+ cache: maven
+
+ - name: Cache Spark distribution
+ uses: actions/cache@v4
+ with:
+ path: ~/spark
+ key: spark-${{ env.SPARK_VERSION }}-bin-${{ env.SPARK_HADOOP_PROFILE
}}
+
+ - name: Install Spark
+ run: |
+ set -euo pipefail
+ SPARK_DIR="spark-${SPARK_VERSION}-bin-${SPARK_HADOOP_PROFILE}"
+ if [ ! -d "$HOME/spark/$SPARK_DIR" ]; then
+ mkdir -p "$HOME/spark"
+ echo "Downloading $SPARK_DIR from the Apache archive"
+ curl -fsSL
"https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/${SPARK_DIR}.tgz";
\
+ | tar -xz -C "$HOME/spark"
+ fi
+ echo "SPARK_HOME=$HOME/spark/$SPARK_DIR" >> "$GITHUB_ENV"
+
+ - name: Build reactor (skip tests)
+ run: ./mvnw clean install -ntp -B -DskipTests
+
+ - name: Run xtable-spark-runtime bundle IT
+ env:
+ SPARK_LOCAL_IP: "127.0.0.1"
+ run: ./mvnw verify -ntp -B -pl xtable-spark-runtime
diff --git a/.gitignore b/.gitignore
index def30ad3..2b4cdd7c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -54,3 +54,6 @@ spec/generated
demo/notebook/spark-warehouse/
demo/notebook/metastore_db/
demo/notebook/derby.log
+
+# Written by maven-shade-plugin in xtable-spark-runtime
+dependency-reduced-pom.xml
diff --git a/pom.xml b/pom.xml
index b66c5887..6eaee41e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,6 +54,7 @@
<module>xtable-aws</module>
<module>xtable-hive-metastore</module>
<module>xtable-service</module>
+ <module>xtable-spark-runtime</module>
</modules>
<properties>
@@ -970,6 +971,8 @@
<excludes>
<exclude>.idea/**</exclude>
<exclude>${target.dir.pattern}</exclude>
+ <!-- Written by maven-shade-plugin in
xtable-spark-runtime -->
+ <exclude>**/dependency-reduced-pom.xml</exclude>
<exclude>website/node_modules/**</exclude>
<exclude>website/build/**</exclude>
<exclude>website/.docusaurus/**</exclude>
diff --git a/release/scripts/validate_bundled_license_texts.py
b/release/scripts/validate_bundled_license_texts.py
index a8148f20..905d7528 100644
--- a/release/scripts/validate_bundled_license_texts.py
+++ b/release/scripts/validate_bundled_license_texts.py
@@ -23,7 +23,7 @@ pom chain declares none. Anything not Apache-2.0 must have
META-INF/licenses/LICENSE-<artifactId>, and a text with no dependency behind it
is reported too.
-Flow, for each */target/*-bundled.jar:
+Flow, for each */target/*-bundled.jar, plus each module in
MAIN_ARTIFACT_BUNDLES:
main() find the jars, print each report, exit non-zero on any
failure
check() everything below, for one jar
@@ -56,6 +56,10 @@ M2 = pathlib.Path.home() / ".m2" / "repository"
ROOT = pathlib.Path(__file__).resolve().parents[2]
TREE = "target/dependency-tree.txt"
+# Modules whose shaded jar replaces the main artifact instead of a
*-bundled.jar.
+MAIN_ARTIFACT_BUNDLES =
("xtable-spark-runtime/target/xtable-spark-runtime_*.jar",)
+NOT_BUNDLES = ("-sources.jar", "-tests.jar", "-javadoc.jar")
+
# Shade cannot merge module descriptors, so it drops them from every
dependency.
DROPPED_BY_SHADE = {"module-info.class"}
@@ -295,9 +299,17 @@ REPORTS = [
]
+def default_jars() -> list[pathlib.Path]:
+ """Every */target/*-bundled.jar, plus the main jar of each
MAIN_ARTIFACT_BUNDLES module."""
+ jars = set(ROOT.glob("**/target/*-bundled.jar"))
+ for pattern in MAIN_ARTIFACT_BUNDLES:
+ jars.update(jar for jar in ROOT.glob(pattern) if not
jar.name.endswith(NOT_BUNDLES))
+ return sorted(jars)
+
+
def main() -> None:
"""Check every bundled jar, or only those named on the command line."""
- jars = [pathlib.Path(a) for a in sys.argv[1:]] or
sorted(ROOT.glob("**/target/*-bundled.jar"))
+ jars = [pathlib.Path(a) for a in sys.argv[1:]] or default_jars()
if not jars:
raise SystemExit(
"FAIL: no */target/*-bundled.jar found; nothing was validated.\n"
diff --git
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
index 86096255..ae679ed5 100644
---
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
+++
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionTargetFactory.java
@@ -23,17 +23,33 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
import java.util.Properties;
+import java.util.concurrent.Callable;
import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.apache.xtable.delta.DeltaConversionTarget;
import org.apache.xtable.delta.DeltaConversionTargetConfig;
import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.hudi.HudiConversionTarget;
+import org.apache.xtable.iceberg.IcebergConversionTarget;
import org.apache.xtable.kernel.DeltaKernelConversionTarget;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.ConversionTarget;
@@ -137,6 +153,144 @@ public class TestConversionTargetFactory {
.createConversionTargetForName(TableFormat.HUDI, properties));
}
+ @Test
+ public void testDiscoversAvailableTargetWhenOtherEnginesAreMissing(@TempDir
Path tempDir)
+ throws Exception {
+ ConversionTarget target =
+ withEngineMissingClassLoader(
+ tempDir,
+ () ->
+ ConversionTargetFactory.getInstance()
+ .createConversionTargetForName(TableFormat.ICEBERG));
+ assertEquals(IcebergConversionTarget.class.getName(),
target.getClass().getName());
+ }
+
+ @Test
+ public void
testMissingEngineForRequestedFormatFailsWithNotSupported(@TempDir Path tempDir)
{
+ NotSupportedException thrown =
+ assertThrows(
+ NotSupportedException.class,
+ () ->
+ withEngineMissingClassLoader(
+ tempDir,
+ () ->
+ ConversionTargetFactory.getInstance()
+
.createConversionTargetForName(TableFormat.DELTA)));
+ assertTrue(thrown.getMessage().contains(TableFormat.DELTA));
+ }
+
+ /**
+ * Runs the call with a context classloader that registers four providers:
one whose class does
+ * not exist, two Delta targets whose engine is absent (a missing superclass
fails in hasNext(), a
+ * missing constructor dependency fails in next()), and the Iceberg target.
The timeout guards
+ * against the provider iterator retrying a failing entry forever.
+ */
+ private static <T> T withEngineMissingClassLoader(Path tempDir, Callable<T>
call)
+ throws Exception {
+ Path services = tempDir.resolve(ConversionTarget.class.getName());
+ Files.write(
+ services,
+ Arrays.asList(
+ "org.apache.xtable.missing.NoSuchConversionTarget",
+ EngineSubclassDeltaTarget.class.getName(),
+ EngineFieldDeltaTarget.class.getName(),
+ IcebergConversionTarget.class.getName()),
+ StandardCharsets.UTF_8);
+ ClassLoader loader =
+ new EngineMissingClassLoader(
+ TestConversionTargetFactory.class.getClassLoader(),
services.toUri().toURL());
+ return assertTimeoutPreemptively(
+ Duration.ofSeconds(30),
+ () -> {
+ Thread thread = Thread.currentThread();
+ ClassLoader previous = thread.getContextClassLoader();
+ thread.setContextClassLoader(loader);
+ try {
+ return call.call();
+ } finally {
+ thread.setContextClassLoader(previous);
+ }
+ });
+ }
+
+ /** Stands in for an engine class; hidden by {@link
EngineMissingClassLoader}. */
+ public static class DeltaEngine {}
+
+ /** Stands in for an engine base class; hidden by {@link
EngineMissingClassLoader}. */
+ public static class DeltaEngineTarget extends DeltaConversionTarget {}
+
+ /** A Delta target whose superclass is in the absent engine. */
+ public static class EngineSubclassDeltaTarget extends DeltaEngineTarget {}
+
+ /** A Delta target whose constructor needs the absent engine. */
+ public static class EngineFieldDeltaTarget extends DeltaConversionTarget {
+ private final DeltaEngine engine = new DeltaEngine();
+ }
+
+ /**
+ * Serves a fixed ConversionTarget services file, hides the engine classes,
and defines the
+ * engine-backed targets itself so that their engine references resolve here.
+ */
+ private static class EngineMissingClassLoader extends ClassLoader {
+ private static final String SERVICES_RESOURCE =
+ "META-INF/services/" + ConversionTarget.class.getName();
+ private static final List<String> HIDDEN =
+ Arrays.asList(DeltaEngine.class.getName(),
DeltaEngineTarget.class.getName());
+ private static final List<String> CHILD_FIRST =
+ Arrays.asList(
+ EngineSubclassDeltaTarget.class.getName(),
EngineFieldDeltaTarget.class.getName());
+
+ private final URL services;
+
+ EngineMissingClassLoader(ClassLoader parent, URL services) {
+ super(parent);
+ this.services = services;
+ }
+
+ @Override
+ public Enumeration<URL> getResources(String name) throws IOException {
+ return SERVICES_RESOURCE.equals(name)
+ ? Collections.enumeration(Collections.singletonList(services))
+ : super.getResources(name);
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws
ClassNotFoundException {
+ synchronized (getClassLoadingLock(name)) {
+ if (HIDDEN.contains(name)) {
+ throw new ClassNotFoundException(name);
+ }
+ if (!CHILD_FIRST.contains(name)) {
+ return super.loadClass(name, resolve);
+ }
+ Class<?> loaded = findLoadedClass(name);
+ if (loaded == null) {
+ try (InputStream in =
+ getParent().getResourceAsStream(name.replace('.', '/') +
".class")) {
+ byte[] bytes = readAll(in);
+ loaded = defineClass(name, bytes, 0, bytes.length);
+ } catch (IOException e) {
+ throw new ClassNotFoundException(name, e);
+ }
+ }
+ if (resolve) {
+ resolveClass(loaded);
+ }
+ return loaded;
+ }
+ }
+
+ private static byte[] readAll(InputStream in) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] buffer = new byte[8192];
+ int read;
+ while ((read = in.read(buffer)) != -1) {
+ out.write(buffer, 0, read);
+ }
+ return out.toByteArray();
+ }
+ }
+
private TargetTable getPerTableConfig(String tableFormat) {
return TargetTable.builder()
.name(getTableName())
diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml
new file mode 100644
index 00000000..34a3de70
--- /dev/null
+++ b/xtable-spark-runtime/pom.xml
@@ -0,0 +1,324 @@
+<?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.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>Thin, relocated Spark bundle that runs XTable metadata
sync.</description>
+
+ <dependencies>
+ <!-- Bundled -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>2.2</version>
+ </dependency>
+
+
+ <!-- Provided by the Spark runtime -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+
+ <!-- Engines (Hudi, Iceberg, Delta) are provided by the user's
runtime, not bundled -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+
<artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <!-- The Hudi target uses the Java client; hudi-common comes from the
hudi-spark bundle -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-java-client</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-core_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-defaults</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Iceberg 1.9.2 needs Avro 1.12, newer than Spark 3.4's -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.openjdk.jol</groupId>
+ <artifactId>jol-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Transitive via xtable-core but shipped by Spark; provided keeps
them out of the bundle -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+
<artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test Dependencies -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+
<artifactId>iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Engine classpath for ITXTableSparkRuntimeBundle, passed as
extraClassPath -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>engine-classpath</id>
+ <phase>pre-integration-test</phase>
+ <goals>
+ <goal>build-classpath</goal>
+ </goals>
+ <configuration>
+ <includeScope>provided</includeScope>
+
<excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j,org.scala-lang,com.fasterxml.jackson.core,com.fasterxml.jackson.datatype,com.fasterxml.jackson.module</excludeGroupIds>
+
<outputFile>${project.build.directory}/engine-classpath.txt</outputFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--
+ The shaded jar is the main artifact, like hudi-spark-bundle and
iceberg-spark-runtime.
+ It bundles only XTable and the libraries XTable uses internally,
relocated. Libraries
+ XTable exchanges with the engines (avro, parquet, jackson) stay
provided.
+ -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+
<shadedArtifactAttached>false</shadedArtifactAttached>
+
<createDependencyReducedPom>true</createDependencyReducedPom>
+ <!-- Must match the runtime dependency tree
(validate_shaded_license_coverage.sh) -->
+ <artifactSet>
+ <includes>
+
<include>org.apache.xtable:xtable-api</include>
+
<include>org.apache.xtable:xtable-core_${scala.binary.version}</include>
+
<include>org.apache.xtable:xtable-hudi-support-utils</include>
+ <include>com.google.guava:guava</include>
+
<include>com.google.guava:failureaccess</include>
+
<include>com.google.guava:listenablefuture</include>
+
<include>com.google.errorprone:error_prone_annotations</include>
+
<include>com.google.j2objc:j2objc-annotations</include>
+
<include>org.checkerframework:checker-qual</include>
+ <include>commons-cli:commons-cli</include>
+ <include>org.yaml:snakeyaml</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>com.google.common</pattern>
+
<shadedPattern>org.apache.xtable.shaded.com.google.common</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google.thirdparty</pattern>
+
<shadedPattern>org.apache.xtable.shaded.com.google.thirdparty</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.commons.cli</pattern>
+
<shadedPattern>org.apache.xtable.shaded.org.apache.commons.cli</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.yaml.snakeyaml</pattern>
+
<shadedPattern>org.apache.xtable.shaded.org.yaml.snakeyaml</shadedPattern>
+ </relocation>
+ </relocations>
+ <!-- Override: a positional merge with the
parent's awssdk filter drops snakeyaml -->
+ <filters combine.self="override">
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>**/Log4j2Plugins.dat</exclude>
+ </excludes>
+ </filter>
+ <!-- Shade does not relocate multi-release
classes -->
+ <filter>
+ <artifact>org.yaml:snakeyaml</artifact>
+ <excludes>
+ <exclude>META-INF/versions/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ApacheLicenseResourceTransformer"/>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer">
+ <resources>
+ <resource>LICENSE</resource>
+ <resource>NOTICE</resource>
+ <resource>NOTICE.txt</resource>
+ </resources>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/LICENSE</resource>
+
<file>target/classes/META-INF/LICENSE-bundled</file>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/NOTICE</resource>
+
<file>target/classes/META-INF/NOTICE-bundled</file>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
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 00000000..75d35a32
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+/** One table to sync: its source, its target formats and where its data files
live. */
+@Value
+@Builder
+public class TableSyncSpec {
+ /** A name for the table, used in log messages. */
+ @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;
+
+ /** Optional partition spec for a partitioned Hudi source, e.g. {@code
level:VALUE}. */
+ String partitionSpec;
+
+ /** 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<String> targets;
+
+ /** Use the Spark-free Delta Kernel implementation for the Delta source
and/or target. */
+ boolean useDeltaKernel;
+}
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 00000000..84a8b863
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
@@ -0,0 +1,472 @@
+/*
+ * 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.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableSet;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code
RunSync}.
+ *
+ * <pre>
+ * $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
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ ImmutableSet.of(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET);
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ ImmutableSet.of(TableFormat.HUDI, TableFormat.ICEBERG,
TableFormat.DELTA);
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .desc("The base path of the source table (single-table
mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG
(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .desc(
+ "Comma-separated target formats to sync to, e.g.
ICEBERG,DELTA "
+ + "(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATASET_CONFIG)
+ .hasArg()
+ .argName("path")
+ .desc(
+ "Path to a YAML dataset config (sourceFormat,
targetFormats, datasets[]) to "
+ + "sync multiple tables in one run; may be a local
or cloud (s3/gcs/abfs) "
+ + "path. Mutually exclusive with
--basepath/--sourceformat/--targets.")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATA_PATH)
+ .hasArg()
+ .desc("The path of the data files if different from the base
path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TABLE_NAME)
+ .hasArg()
+ .desc("The table name; defaults to the last segment of the
base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(NAMESPACE)
+ .hasArg()
+ .desc("The dot-separated table namespace")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(PARTITION_SPEC)
+ .hasArg()
+ .desc("The Hudi source partition field spec, e.g.
level:VALUE")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(USE_DELTA_KERNEL)
+ .desc(
+ "Use the Spark-free Delta Kernel implementation for the
Delta source and/or "
+ + "target instead of the default delta-core path.
When omitted this is "
+ + "auto-enabled on Spark 3.5+ (where the bundled
delta-core does not "
+ + "run); pass it explicitly to force Kernel on any
Spark version.")
+ .build())
+ .addOption(Option.builder().longOpt(HELP).desc("Displays help
information").build());
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ throw new IllegalArgumentException("Failed to parse arguments", e);
+ }
+ if (cmd.hasOption(HELP)) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ return;
+ }
+
+ boolean datasetMode = cmd.hasOption(DATASET_CONFIG);
+ if (datasetMode
+ && (cmd.hasOption(BASE_PATH) || cmd.hasOption(SOURCE_FORMAT) ||
cmd.hasOption(TARGETS))) {
+ throw new IllegalArgumentException(
+ "--datasetconfig is mutually exclusive with
--basepath/--sourceformat/--targets");
+ }
+ // Validate single-table flags before starting Spark. A --datasetconfig is
validated after,
+ // because it is read through the Spark Hadoop config.
+ if (!datasetMode) {
+ requireOption(cmd, BASE_PATH);
+ requireOption(cmd, SOURCE_FORMAT);
+ requireOption(cmd, TARGETS);
+ validateFormat(
+ SOURCE_FORMAT,
+ cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT),
+ SUPPORTED_SOURCE_FORMATS);
+ parseTargets(cmd.getOptionValue(TARGETS)); // validates formats and
non-emptiness
+ }
+
+ boolean useDeltaKernelFlag = cmd.hasOption(USE_DELTA_KERNEL);
+
+ SparkSession spark =
SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+
+ String sourceFormat;
+ List<String> targetFormats;
+ List<TableSyncSpec.TableSyncSpecBuilder> builders = new ArrayList<>();
+ if (datasetMode) {
+ DatasetConfig config =
readDatasetConfig(cmd.getOptionValue(DATASET_CONFIG), hadoopConf);
+ sourceFormat = config.sourceFormat;
+ targetFormats = config.targetFormats;
+ for (DatasetConfig.Table table : config.datasets) {
+ builders.add(
+ TableSyncSpec.builder()
+ .key(
+ table.tableName != null
+ ? table.tableName
+ : basePathToTableName(table.tableBasePath))
+ .basePath(table.tableBasePath)
+ .dataPath(table.tableDataPath)
+ .namespace(table.namespace == null ? null :
table.namespace.split("\\."))
+ .partitionSpec(table.partitionSpec)
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+ } else {
+ sourceFormat =
cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT);
+ targetFormats = parseTargets(cmd.getOptionValue(TARGETS));
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String namespace = cmd.getOptionValue(NAMESPACE);
+ builders.add(
+ TableSyncSpec.builder()
+ .key(cmd.getOptionValue(TABLE_NAME,
basePathToTableName(basePath)))
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+
+ // delta-core does not run on Spark 3.5+, so use Delta Kernel there.
+ boolean deltaInvolved =
+ TableFormat.DELTA.equals(sourceFormat) ||
targetFormats.contains(TableFormat.DELTA);
+ boolean useDeltaKernel = useDeltaKernelFlag;
+ if (!useDeltaKernel && deltaInvolved &&
isSparkAtLeast35(spark.version())) {
+ useDeltaKernel = true;
+ log.info(
+ "Spark {} detected (>= 3.5); auto-enabling the Delta Kernel
implementation for Delta "
+ + "source/target. Pass --usedeltakernel to force it on any
Spark version.",
+ spark.version());
+ }
+
+ XTableSyncService service = new XTableSyncService();
+ List<TableSyncSpec> specs = new ArrayList<>();
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ specs.add(builder.useDeltaKernel(useDeltaKernel).build());
+ }
+ int failures = syncAll(specs, spec -> service.sync(spec, hadoopConf));
+ if (failures > 0) {
+ throw new IllegalStateException(
+ failures + " of " + specs.size() + " table sync(s) failed; see
logs above");
+ }
+ } finally {
+ spark.stop();
+ }
+ }
+
+ /**
+ * Syncs every spec and returns how many failed. A table fails when its sync
throws or when any
+ * target reports a non-SUCCESS status; one failure does not abort the rest
of the batch.
+ */
+ // package-private for unit testing
+ static int syncAll(
+ List<TableSyncSpec> specs, Function<TableSyncSpec, Map<String,
SyncResult>> syncer) {
+ int failures = 0;
+ for (TableSyncSpec spec : specs) {
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(),
spec.getBasePath());
+ // TableFormatSync reports a target failure as an ERROR status rather
than throwing.
+ Map<String, String> failedTargets =
+ syncer.apply(spec).entrySet().stream()
+ .filter(
+ entry ->
+
entry.getValue().getTableFormatSyncStatus().getStatusCode()
+ != SyncStatusCode.SUCCESS)
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry ->
describe(entry.getValue().getTableFormatSyncStatus()),
+ (a, b) -> a,
+ TreeMap::new));
+ if (failedTargets.isEmpty()) {
+ log.info("Completed XTable sync for {}", spec.getKey());
+ } else {
+ failures++;
+ log.error(
+ "XTable sync failed for {} ({}) on target(s) {}",
+ spec.getKey(),
+ spec.getBasePath(),
+ failedTargets);
+ }
+ } catch (RuntimeException e) {
+ failures++;
+ log.error("XTable sync failed for {} ({})", spec.getKey(),
spec.getBasePath(), e);
+ }
+ }
+ return failures;
+ }
+
+ private static String describe(SyncResult.SyncStatus status) {
+ ErrorDetails details = status.getErrorDetails();
+ return details == null || details.getErrorMessage() == null
+ ? String.valueOf(status.getStatusCode())
+ : status.getStatusCode() + ": " + details.getErrorMessage();
+ }
+
+ private static void requireOption(CommandLine cmd, String option) {
+ if (!cmd.hasOption(option)) {
+ throw new IllegalArgumentException(
+ "--" + option + " is required unless --" + DATASET_CONFIG + " is
provided");
+ }
+ }
+
+ // package-private for unit testing
+ static List<String> parseTargets(String targetsValue) {
+ List<String> targetFormats =
+ Arrays.stream(targetsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (targetFormats.isEmpty()) {
+ throw new IllegalArgumentException(
+ "--targets resolved to an empty list; provide at least one target
format, e.g. ICEBERG,DELTA");
+ }
+ targetFormats.forEach(target -> validateFormat(TARGETS, target,
SUPPORTED_TARGET_FORMATS));
+ return targetFormats;
+ }
+
+ static DatasetConfig readDatasetConfig(String path, Configuration
hadoopConf) {
+ Path configPath = new Path(path);
+ try (InputStream in =
configPath.getFileSystem(hadoopConf).open(configPath)) {
+ return parseDatasetConfig(in);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to read --datasetconfig at "
+ path, e);
+ }
+ }
+
+ /**
+ * SafeConstructor builds only maps, lists and scalars, and Jackson binds
them to {@link
+ * DatasetConfig}. Uses snakeyaml because Spark does not always ship
jackson-dataformat-yaml.
+ */
+ // package-private for unit testing
+ static DatasetConfig parseDatasetConfig(InputStream in) {
+ Object loaded = new Yaml(new SafeConstructor(new
LoaderOptions())).load(in);
+ if (!(loaded instanceof Map)) {
+ throw new IllegalArgumentException("--datasetconfig is empty or not a
YAML mapping");
+ }
+ DatasetConfig config;
+ try {
+ config = DATASET_CONFIG_MAPPER.convertValue(loaded, DatasetConfig.class);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("--datasetconfig is invalid: " +
e.getMessage(), e);
+ }
+
+ config.sourceFormat =
+ requireField(config.sourceFormat,
"sourceFormat").trim().toUpperCase(Locale.ROOT);
+ validateFormat(SOURCE_FORMAT, config.sourceFormat,
SUPPORTED_SOURCE_FORMATS);
+
+ if (config.targetFormats == null || config.targetFormats.isEmpty()) {
+ throw new IllegalArgumentException(
+ "--datasetconfig 'targetFormats' must be a non-empty list");
+ }
+ config.targetFormats =
+ config.targetFormats.stream()
+ .map(t -> String.valueOf(t).trim().toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ config.targetFormats.forEach(t -> validateFormat(TARGETS, t,
SUPPORTED_TARGET_FORMATS));
+
+ if (config.datasets == null || config.datasets.isEmpty()) {
+ throw new IllegalArgumentException("--datasetconfig 'datasets' must be a
non-empty list");
+ }
+ for (DatasetConfig.Table table : config.datasets) {
+ if (table == null) {
+ throw new IllegalArgumentException("--datasetconfig 'datasets' entries
must be mappings");
+ }
+ requireField(table.tableBasePath, "tableBasePath");
+ }
+ return config;
+ }
+
+ private static String requireField(String value, String key) {
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException("--datasetconfig is missing required
field '" + key + "'");
+ }
+ return value;
+ }
+
+ // Reads the package-private fields. Unknown keys are ignored, as before.
+ private static final ObjectMapper DATASET_CONFIG_MAPPER =
+ new ObjectMapper()
+ .setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY)
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+ /** Parsed {@code --datasetconfig}; mirrors the RunSync dataset YAML so
configs are shareable. */
+ static final class DatasetConfig {
+ String sourceFormat;
+ List<String> targetFormats;
+ List<Table> datasets;
+
+ static final class Table {
+ String tableBasePath;
+ String tableDataPath;
+ String tableName;
+ String partitionSpec;
+ String namespace;
+ }
+ }
+
+ /**
+ * True when the Spark version is 3.5 or newer; unparseable/null versions
are treated as older.
+ */
+ // package-private for unit testing
+ static boolean isSparkAtLeast35(String sparkVersion) {
+ if (sparkVersion == null || sparkVersion.trim().isEmpty()) {
+ return false;
+ }
+ String[] parts = sparkVersion.trim().split("\\.");
+ try {
+ int major = Integer.parseInt(parts[0]);
+ int minor = parts.length > 1 ?
Integer.parseInt(parts[1].replaceAll("[^0-9].*$", "")) : 0;
+ return major > 3 || (major == 3 && minor >= 5);
+ } catch (NumberFormatException e) {
+ return false;
+ }
+ }
+
+ // package-private for unit testing
+ static void validateFormat(String option, String format, Set<String>
supported) {
+ if (!supported.contains(format)) {
+ throw new IllegalArgumentException(
+ "Unsupported --" + option + " '" + format + "'; expected one of " +
supported);
+ }
+ }
+
+ /**
+ * Returns the last path segment, e.g. {@code /warehouse/db/orders ->
orders}. Throws when there
+ * is none, so the user passes {@code --tablename}.
+ */
+ // package-private for unit testing
+ static String basePathToTableName(String basePath) {
+ String trimmed = basePath == null ? "" : basePath;
+ while (trimmed.endsWith("/")) {
+ trimmed = trimmed.substring(0, trimmed.length() - 1);
+ }
+ int idx = trimmed.lastIndexOf('/');
+ String name = idx >= 0 ? trimmed.substring(idx + 1) : trimmed;
+ if (name.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Cannot derive a table name from --basepath '"
+ + basePath
+ + "'; pass --tablename explicitly");
+ }
+ return name;
+ }
+}
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 00000000..43eb720a
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -0,0 +1,143 @@
+/*
+ * 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.Locale;
+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.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link
TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe
to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results.
*/
+ public Map<String, SyncResult> 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());
+ }
+ 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();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL,
Boolean.TRUE.toString());
+ }
+ // Hudi resolves partitions relative to its base path, so a
Hudi target sits
+ // at the data path. Other targets sit at the table root.
+ String targetPath =
+ TableFormat.HUDI.equalsIgnoreCase(targetFormat)
+ ? dataPath
+ : spec.getBasePath();
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(targetPath)
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .additionalProperties(targetProperties)
+ .build();
+ })
+ .collect(Collectors.toList());
+
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+
+ ConversionSourceProvider<?> sourceProvider =
+ sourceProviderFor(spec.getSourceFormat(), spec.isUseDeltaKernel());
+ 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, boolean useDeltaKernel) {
+ switch (sourceFormat.toUpperCase(Locale.ROOT)) {
+ case TableFormat.HUDI:
+ return new HudiConversionSourceProvider();
+ case TableFormat.DELTA:
+ // The default delta-core reader is Spark-3.4-only; the Kernel reader
is Spark-free.
+ return useDeltaKernel
+ ? new DeltaKernelConversionSourceProvider()
+ : new DeltaConversionSourceProvider();
+ case TableFormat.ICEBERG:
+ return new IcebergConversionSourceProvider();
+ case TableFormat.PAIMON:
+ return new PaimonConversionSourceProvider();
+ case TableFormat.PARQUET:
+ return new ParquetConversionSourceProvider();
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported source format for spark-runtime sync: " +
sourceFormat);
+ }
+ }
+}
diff --git a/xtable-spark-runtime/src/main/resources/META-INF/LICENSE-bundled
b/xtable-spark-runtime/src/main/resources/META-INF/LICENSE-bundled
new file mode 100644
index 00000000..5574bfdc
--- /dev/null
+++ b/xtable-spark-runtime/src/main/resources/META-INF/LICENSE-bundled
@@ -0,0 +1,221 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
+---------------------------------------------------------
+
+This product bundles various third-party components under other open source
licenses.
+This section summarizes those components and their licenses. See licenses/ for
text of these licenses.
+
+
+Apache License 2.0
+------------------
+com.google.errorprone:error_prone_annotations:2.21.1
+com.google.guava:failureaccess:1.0.1
+com.google.guava:guava:32.1.3-jre
+com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava
+com.google.j2objc:j2objc-annotations:2.8
+commons-cli:commons-cli:1.5.0
+org.yaml:snakeyaml:2.2
+
+MIT License
+-----------
+org.checkerframework:checker-qual:3.37.0
diff --git a/xtable-spark-runtime/src/main/resources/META-INF/NOTICE-bundled
b/xtable-spark-runtime/src/main/resources/META-INF/NOTICE-bundled
new file mode 100644
index 00000000..04bfd54f
--- /dev/null
+++ b/xtable-spark-runtime/src/main/resources/META-INF/NOTICE-bundled
@@ -0,0 +1,22 @@
+Apache XTable (incubating)
+Copyright 2024-2026 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (https://www.apache.org/).
+
+--------------------------------------------------------------------------------
+
+This binary artifact bundles the following projects with NOTICE:
+
+--------------------------------------------------------------------------------
+
+Group: commons-cli Name: commons-cli Version: 1.5.0
+
+NOTICE:
+| Apache Commons CLI
+| Copyright 2002-2021 The Apache Software Foundation
+|
+| This product includes software developed at
+| The Apache Software Foundation (https://www.apache.org/).
+
+--------------------------------------------------------------------------------
diff --git
a/xtable-spark-runtime/src/main/resources/META-INF/licenses/LICENSE-checker-qual
b/xtable-spark-runtime/src/main/resources/META-INF/licenses/LICENSE-checker-qual
new file mode 100644
index 00000000..9837c6b6
--- /dev/null
+++
b/xtable-spark-runtime/src/main/resources/META-INF/licenses/LICENSE-checker-qual
@@ -0,0 +1,22 @@
+Checker Framework qualifiers
+Copyright 2004-present by the Checker Framework developers
+
+MIT License:
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
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 00000000..061d6e35
--- /dev/null
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
@@ -0,0 +1,387 @@
+/*
+ * 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.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+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.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.MapType;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+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.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+
+import org.apache.xtable.GenericTable;
+import org.apache.xtable.hudi.HudiTestUtil;
+
+/**
+ * Runs a sync with the shaded bundle through {@code
$SPARK_HOME/bin/spark-submit}, one case per
+ * direction, and checks the target data equals the source. The engines come
from {@code
+ * target/engine-classpath.txt}. Skipped when {@code SPARK_HOME} is not set.
+ */
+class ITXTableSparkRuntimeBundle {
+
+ @TempDir static 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<Arguments> directions() {
+ // sourceFormat, targetFormat, hudiPartitionSpec
+ return Stream.of(
+ Arguments.of("HUDI", "ICEBERG", "level:VALUE"),
+ Arguments.of("ICEBERG", "HUDI", null),
+ Arguments.of("ICEBERG", "DELTA", null),
+ Arguments.of("HUDI", "DELTA", "level:VALUE"),
+ Arguments.of("DELTA", "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<String> columns;
+ String orderByColumn;
+ try (GenericTable<?, ?> table =
+ GenericTable.getInstance(tableName, tempDir, sparkSession, jsc,
sourceFormat, true)) {
+ table.insertRows(100);
+ basePath = table.getBasePath();
+ dataPath = table.getDataPath();
+ columns = table.getColumnsToSelect();
+ orderByColumn = table.getOrderByColumn();
+ }
+
+ List<String> syncArgs = new ArrayList<>();
+ syncArgs.add("--basepath");
+ syncArgs.add(basePath);
+ syncArgs.add("--datapath");
+ syncArgs.add(dataPath);
+ syncArgs.add("--sourceformat");
+ syncArgs.add(sourceFormat);
+ syncArgs.add("--targets");
+ syncArgs.add(targetFormat);
+ syncArgs.add("--tablename");
+ syncArgs.add(tableName);
+ if (partitionSpec != null) {
+ syncArgs.add("--partitionspec");
+ syncArgs.add(partitionSpec);
+ }
+ String output = runBundleSync(sparkHome, syncArgs);
+
+ assertDatasetEquivalence(
+ sourceFormat, targetFormat, basePath, dataPath, columns,
orderByColumn, output);
+ }
+
+ /**
+ * Iceberg keeps its metadata at the base path but writes data files to a
separate {@code
+ * write.data.path}. The Hudi target goes to the data path, so {@code
.hoodie} lands next to the
+ * Iceberg data files and nothing is written under the Iceberg base path.
+ */
+ @Test
+ void icebergWithSeparateDataPathSyncsToHudi() throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME
is not set");
+
+ String tableName = "iceberg_split_paths_to_hudi";
+ Path metadataDir = tempDir.resolve(tableName + "_metadata");
+ Path dataDir = tempDir.resolve("separate_data").resolve(tableName);
+ String basePath = metadataDir.toUri().toString();
+ String dataPath = dataDir.toUri().toString();
+
+ Dataset<Row> rows =
+ sparkSession
+ .range(100)
+ .selectExpr(
+ "concat('id-', lpad(cast(id AS string), 3, '0')) AS id",
+ "element_at(array('INFO', 'WARN', 'ERROR'), cast(id % 3 + 1 AS
int)) AS level",
+ "cast(id AS double) * 1.5 AS amount",
+ "date_add(date '2026-01-01', cast(id AS int)) AS event_date");
+ Schema schema = SparkSchemaUtil.convert(rows.schema());
+ new HadoopTables(jsc.hadoopConfiguration())
+ .create(
+ schema,
+ PartitionSpec.builderFor(schema).identity("level").build(),
+ Collections.singletonMap(TableProperties.WRITE_DATA_LOCATION,
dataPath),
+ basePath);
+ rows.write().format("iceberg").mode("append").save(basePath);
+ assertTrue(countParquetFiles(dataDir) > 0, "Iceberg did not write data
files to " + dataDir);
+ assertEquals(0, countParquetFiles(metadataDir), "Iceberg wrote data files
under the base path");
+
+ List<String> syncArgs =
+ Arrays.asList(
+ "--basepath",
+ basePath,
+ "--datapath",
+ dataPath,
+ "--sourceformat",
+ "ICEBERG",
+ "--targets",
+ "HUDI",
+ "--tablename",
+ tableName);
+ String output = runBundleSync(sparkHome, syncArgs);
+
+ assertTrue(
+ Files.isDirectory(dataDir.resolve(".hoodie")),
+ "Hudi metadata is not at the data path. spark-submit output:\n" +
output);
+ assertTrue(
+ !Files.exists(metadataDir.resolve(".hoodie")),
+ "Hudi metadata was written to the base path");
+ assertDatasetEquivalence(
+ "ICEBERG",
+ "HUDI",
+ basePath,
+ dataPath,
+ Arrays.asList("id", "level", "amount", "event_date"),
+ "id",
+ output);
+ }
+
+ /** Runs {@code XTableSparkSync} from the bundle through spark-submit and
returns its output. */
+ private String runBundleSync(String sparkHome, List<String> syncArgs) throws
Exception {
+ File bundleJar = findBundleJar();
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " +
sparkSubmit);
+
+ // A flat classpath, not --packages: a child classloader breaks Avro casts
across the boundary.
+ String engineClasspath = readEngineClasspath();
+
+ List<String> 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.addAll(syncArgs);
+
+ ProcessBuilder pb = new ProcessBuilder(command);
+ pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1");
+ pb.redirectErrorStream(true);
+ Process process = pb.start();
+ // Drain on a thread so a hung spark-submit cannot block the timeout below.
+ StringBuilder outputBuffer = new StringBuilder();
+ Thread drainer = new Thread(() -> readOutputInto(process, outputBuffer),
"spark-submit-stdout");
+ drainer.setDaemon(true);
+ drainer.start();
+ boolean finished = process.waitFor(10, TimeUnit.MINUTES);
+ if (!finished) {
+ process.destroyForcibly();
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ synchronized (outputBuffer) {
+ throw new AssertionError("spark-submit timed out. Output:\n" +
outputBuffer);
+ }
+ }
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ String output;
+ synchronized (outputBuffer) {
+ output = outputBuffer.toString();
+ }
+ assertEquals(0, process.exitValue(), "spark-submit failed. Output:\n" +
output);
+ return output;
+ }
+
+ private static long countParquetFiles(Path dir) throws IOException {
+ if (!Files.isDirectory(dir)) {
+ return 0;
+ }
+ try (Stream<Path> files = Files.walk(dir)) {
+ return files.filter(f -> f.toString().endsWith(".parquet")).count();
+ }
+ }
+
+ /** Reads the source and the target back through Spark and asserts the
scalar columns match. */
+ private void assertDatasetEquivalence(
+ String sourceFormat,
+ String targetFormat,
+ String basePath,
+ String dataPath,
+ List<String> columns,
+ String orderByColumn,
+ String submitOutput) {
+ Dataset<Row> source =
+ sparkSession
+ .read()
+ .options(readOptions(sourceFormat))
+ .format(sourceFormat.toLowerCase())
+ .load(basePath)
+ .orderBy(orderByColumn);
+ String[] projection = comparableColumns(source, columns).toArray(new
String[0]);
+ List<String> sourceRows =
source.selectExpr(projection).toJSON().collectAsList();
+ // Same target location as XTableSyncService.
+ String targetPath = "HUDI".equalsIgnoreCase(targetFormat) ? dataPath :
basePath;
+ List<String> targetRows =
+ sparkSession
+ .read()
+ .options(readOptions(targetFormat))
+ .format(targetFormat.toLowerCase())
+ .load(targetPath)
+ .orderBy(orderByColumn)
+ .selectExpr(projection)
+ .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");
+ }
+
+ // Drops nested columns (covered by ITConversionController) and local
timestamps (HUDI-7088).
+ private static List<String> comparableColumns(Dataset<Row> source,
List<String> columns) {
+ Set<String> excluded = new HashSet<>();
+ for (StructField field : source.schema().fields()) {
+ DataType type = field.dataType();
+ if (type instanceof StructType || type instanceof ArrayType || type
instanceof MapType) {
+ excluded.add(field.name());
+ }
+ }
+ return columns.stream()
+ .filter(c -> !excluded.contains(c))
+ .filter(c -> !c.startsWith("timestamp_local_"))
+ .collect(Collectors.toList());
+ }
+
+ private static Map<String, String> readOptions(String format) {
+ Map<String, String> options = new 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)) {
+ // The vectorized reader mis-casts timestamps written by Hudi.
+ options.put("vectorization-enabled", "false");
+ }
+ return options;
+ }
+
+ 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(Files.readAllBytes(cpFile.toPath()),
StandardCharsets.UTF_8).trim();
+ assertTrue(!cp.isEmpty(), "engine classpath file is empty");
+ return cp;
+ }
+
+ // The shaded jar is the main artifact; skip original-* and the sources,
javadoc and tests jars.
+ private File findBundleJar() {
+ File targetDir = new File("target");
+ File[] candidates =
+ targetDir.listFiles(
+ (dir, name) ->
+ name.startsWith("xtable-spark-runtime")
+ && name.endsWith(".jar")
+ && !name.startsWith("original-")
+ && !name.contains("-sources")
+ && !name.contains("-javadoc")
+ && !name.contains("-tests"));
+ assertTrue(
+ candidates != null && candidates.length == 1,
+ "Expected exactly one bundle jar in "
+ + targetDir.getAbsolutePath()
+ + " (run the package phase first)");
+ return candidates[0];
+ }
+
+ private static void readOutputInto(Process process, StringBuilder sb) {
+ try (BufferedReader reader =
+ new BufferedReader(
+ new InputStreamReader(process.getInputStream(),
StandardCharsets.UTF_8))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ synchronized (sb) {
+ sb.append(line).append(System.lineSeparator());
+ }
+ }
+ } catch (IOException e) {
+ // Closed on timeout; the output so far is still reported.
+ }
+ }
+}
diff --git
a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
new file mode 100644
index 00000000..733c27e9
--- /dev/null
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.assertFalse;
+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.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+class XTableSparkSyncTest {
+
+ @ParameterizedTest
+ @CsvSource({
+ "/warehouse/db/orders, orders",
+ "/warehouse/db/orders/, orders",
+ "/warehouse/db/orders///, orders",
+ "orders, orders",
+ "hdfs://ns/warehouse/db/orders, orders",
+ })
+ void basePathToTableNameDerivesLastSegment(String basePath, String expected)
{
+ assertEquals(expected, XTableSparkSync.basePathToTableName(basePath));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"/", "//", ""})
+ void basePathToTableNameRejectsPathsWithNoUsableSegment(String basePath) {
+ assertThrows(
+ IllegalArgumentException.class, () ->
XTableSparkSync.basePathToTableName(basePath));
+ }
+
+ @Test
+ void basePathToTableNameRejectsNull() {
+ assertThrows(IllegalArgumentException.class, () ->
XTableSparkSync.basePathToTableName(null));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"HUDI", "ICEBERG", "DELTA", "PAIMON", "PARQUET"})
+ void validateFormatAcceptsSupportedSourceFormats(String format) {
+ XTableSparkSync.validateFormat(
+ "sourceformat", format, XTableSparkSync.SUPPORTED_SOURCE_FORMATS);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"HUDI", "ICEBERG", "DELTA"})
+ void validateFormatAcceptsSupportedTargetFormats(String format) {
+ XTableSparkSync.validateFormat("targets", format,
XTableSparkSync.SUPPORTED_TARGET_FORMATS);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"PAIMON", "PARQUET"})
+ void validateFormatRejectsSourceOnlyFormatsAsTargets(String format) {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.validateFormat(
+ "targets", format, XTableSparkSync.SUPPORTED_TARGET_FORMATS));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"HOODIE", "hudi", "", "ICEBERG,DELTA"})
+ void validateFormatRejectsUnsupportedFormats(String format) {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.validateFormat(
+ "sourceformat", format,
XTableSparkSync.SUPPORTED_SOURCE_FORMATS));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"3.5.0", "3.5.9", "3.6.0", "4.0.0", "4.0.0-preview",
"10.2.1"})
+ void isSparkAtLeast35TrueFor35AndNewer(String version) {
+ assertTrue(XTableSparkSync.isSparkAtLeast35(version));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"3.4.3", "3.4.0", "3.3.4", "2.4.8", "3.0.0"})
+ void isSparkAtLeast35FalseForOlder(String version) {
+ assertFalse(XTableSparkSync.isSparkAtLeast35(version));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", " ", "not-a-version", "x.y.z"})
+ void isSparkAtLeast35FalseForNullOrUnparseable(String version) {
+ assertFalse(XTableSparkSync.isSparkAtLeast35(version));
+ }
+
+ @Test
+ void isSparkAtLeast35FalseForNull() {
+ assertFalse(XTableSparkSync.isSparkAtLeast35(null));
+ }
+
+ @Test
+ void syncAllCountsReturnedErrorsAndThrownFailuresAndContinuesBatch() {
+ List<TableSyncSpec> specs =
+ Arrays.asList(spec("ok"), spec("target_error"), spec("throws"),
spec("ok_after"));
+ List<String> synced = new ArrayList<>();
+ int failures =
+ XTableSparkSync.syncAll(
+ specs,
+ spec -> {
+ synced.add(spec.getKey());
+ Map<String, SyncResult> results = new HashMap<>();
+ results.put("ICEBERG", syncResult(SyncStatusCode.SUCCESS));
+ switch (spec.getKey()) {
+ case "target_error":
+ results.put("DELTA", syncResult(SyncStatusCode.ERROR));
+ return results;
+ case "throws":
+ throw new IllegalStateException("source unreadable");
+ default:
+ results.put("DELTA", syncResult(SyncStatusCode.SUCCESS));
+ return results;
+ }
+ });
+ assertEquals(2, failures);
+ assertEquals(Arrays.asList("ok", "target_error", "throws", "ok_after"),
synced);
+ }
+
+ private static TableSyncSpec spec(String key) {
+ return TableSyncSpec.builder()
+ .key(key)
+ .basePath("/warehouse/" + key)
+ .sourceFormat("HUDI")
+ .targets(Arrays.asList("ICEBERG", "DELTA"))
+ .build();
+ }
+
+ private static SyncResult syncResult(SyncStatusCode statusCode) {
+ return SyncResult.builder()
+ .tableFormatSyncStatus(
+ SyncResult.SyncStatus.builder()
+ .statusCode(statusCode)
+ .errorDetails(
+ statusCode == SyncStatusCode.SUCCESS
+ ? null
+ : ErrorDetails.builder().errorMessage("commit
conflict").build())
+ .build())
+ .build();
+ }
+
+ private static InputStream yaml(String content) {
+ return new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void parseDatasetConfigReadsAndNormalizesFields() {
+ XTableSparkSync.DatasetConfig config =
+ XTableSparkSync.parseDatasetConfig(
+ yaml(
+ "sourceFormat: hudi\n"
+ + "targetFormats:\n"
+ + " - iceberg\n"
+ + " - DELTA\n"
+ + "datasets:\n"
+ + " - tableBasePath: /data/orders\n"
+ + " tableName: orders\n"
+ + " partitionSpec: level:VALUE\n"
+ + " - tableBasePath: /data/customers\n"
+ + " namespace: db.sales\n"));
+
+ assertEquals("HUDI", config.sourceFormat);
+ assertEquals(2, config.targetFormats.size());
+ assertEquals("ICEBERG", config.targetFormats.get(0));
+ assertEquals("DELTA", config.targetFormats.get(1));
+ assertEquals(2, config.datasets.size());
+ assertEquals("/data/orders", config.datasets.get(0).tableBasePath);
+ assertEquals("orders", config.datasets.get(0).tableName);
+ assertEquals("level:VALUE", config.datasets.get(0).partitionSpec);
+ assertEquals("/data/customers", config.datasets.get(1).tableBasePath);
+ assertEquals("db.sales", config.datasets.get(1).namespace);
+ assertNull(config.datasets.get(1).tableName);
+ }
+
+ @Test
+ void parseDatasetConfigRejectsMissingSourceFormat() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.parseDatasetConfig(
+ yaml("targetFormats: [ICEBERG]\ndatasets:\n - tableBasePath:
/data/orders\n")));
+ }
+
+ @Test
+ void parseDatasetConfigRejectsEmptyTargetFormats() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.parseDatasetConfig(
+ yaml("sourceFormat: HUDI\ntargetFormats: []\ndatasets:\n -
tableBasePath: /d\n")));
+ }
+
+ @Test
+ void parseDatasetConfigRejectsInvalidTargetFormat() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.parseDatasetConfig(
+ yaml(
+ "sourceFormat: HUDI\ntargetFormats: [PAIMON]\ndatasets:\n
- tableBasePath: /d\n")));
+ }
+
+ @Test
+ void parseDatasetConfigRejectsMissingDatasets() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.parseDatasetConfig(
+ yaml("sourceFormat: HUDI\ntargetFormats: [DELTA]\n")));
+ }
+
+ @Test
+ void parseDatasetConfigRejectsDatasetWithoutBasePath() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ XTableSparkSync.parseDatasetConfig(
+ yaml(
+ "sourceFormat: HUDI\ntargetFormats: [DELTA]\ndatasets:\n
- tableName: orders\n")));
+ }
+}
