This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/branch-0.4 by this push:
new f2d98733 [836] xtable-spark-runtime: thin drop-in Spark bundle for
0.4.x (Spark 3.4 + 3.5) (#843)
f2d98733 is described below
commit f2d987330a84d8c559310ac1c8f31cfb32f6e025
Author: Vinish Reddy <[email protected]>
AuthorDate: Wed Jul 22 00:20:38 2026 +0530
[836] xtable-spark-runtime: thin drop-in Spark bundle for 0.4.x (Spark 3.4
+ 3.5) (#843)
* [836] xtable-spark-runtime: thin drop-in Spark bundle (Hudi 0.x)
Ports the xtable-spark-runtime packaging slice onto the Hudi 0.x line
(main-hudi-0x, Hudi 0.14 / Spark 3.4). New thin, relocated bundle that
runs an incremental XTable sync inside a Spark job; engines are provided
by the cluster, never bundled.
Module:
- xtable-spark-runtime_${scala.binary.version}: xtable-core compile;
Spark/Hadoop and the engines (Hudi/Iceberg/Delta) provided. Curated
shade allowlist (xtable modules + guava/protobuf/commons-cli relocated);
avro/parquet/jackson NOT relocated (exchanged with the engines). Thin
~3.7 MB bundle.
- XTableSparkSync: standalone spark-submit entry point (Apache Commons CLI).
- XTableSyncService / TableSyncSpec: build an INCREMENTAL ConversionConfig
and run ConversionController.sync; target metadata path = source data
path (required by Hudi; Iceberg data lives under <basePath>/data).
Hudi 0.14 specifics (vs the Hudi 1.x variant on main):
- No hudi-hadoop-common (that split is 1.x); hudi-common provides FSUtils.
- Engine classpath uses the hudi-spark bundle (its regenerated Avro model
classes link on Avro 1.12, which Iceberg 1.9.2 requires) plus
hudi-java-client for the Hudi target's Java write client, with the raw
hudi-common excluded so the bundle's clean DecimalWrapper wins.
ConversionTargetFactory: make ServiceLoader discovery resilient so a
subset of engines works when others are absent (warn + skip on
LinkageError/ServiceConfigurationError; name-based Delta-Kernel check).
ITXTableSparkRuntimeBundle: spark-submits the shaded jar for one case per
direction across Hudi, Iceberg and Delta (source and target), engines on
a flat classpath, asserting data-equivalence over comparable scalar
columns. Requires a Spark 3.4 SPARK_HOME; skipped otherwise.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* [836] xtable-spark-runtime: address review feedback (packaging,
licensing, RFC)
Publish the shaded jar as the MAIN artifact with a dependency-reduced POM
(shadedArtifactAttached=false, createDependencyReducedPom=true), matching
iceberg-spark-runtime / hudi-spark-bundle. Resolving the coordinate via a
Maven dependency or --packages now yields the relocated bundle and pulls no
un-relocated transitive deps; --jars is equivalent. The thin main jar +
separate -bundle classifier (which re-introduced the cluster guava clash)
is gone. IT findBundleJar() now picks the shaded main jar.
Pass release/scripts/validate_shaded_license_coverage.sh (the existing
allowlist gate): the shade <includes> must equal the runtime dependency
tree, so jackson / scala-library / log4j-1.2-api - which every Spark
runtime supplies - are declared provided (dropped from the tree, kept off
the shaded jar) and excluded from the IT's flat engine classpath so Spark's
own copies win. avro/parquet stay on the flat classpath (the engine's newer
avro must win over Spark 3.4's).
Bundled-dependency licensing: add META-INF/LICENSE-bundled and
NOTICE-bundled (wired via IncludeResourceTransformer) attributing the only
bundled third-party - guava's closure (Apache-2.0) and commons-cli
(Apache-2.0), plus checker-qual (MIT). Remove the dead protobuf-java
allowlist entry and relocation: protobuf resolves as provided and was never
bundled.
RFC-3: scope v1 to the CLI (XTableSparkSync); mark the config-only
XTableSyncListener on-ramp (and XTableSparkConfig / PlanTargetResolver) as a
deferred follow-up. Fix the activation example to the shipped CLI and
document the required engine Avro version + flat-classpath placement.
XTableSyncService: normalize sourceFormat with Locale.ROOT.
Root pom: exclude the shade-generated dependency-reduced-pom.xml from
spotless (no license header; CI runs clean install).
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* [836] xtable-spark-runtime: trim duplicated pom comments
Consolidate repeated rationale in the bundle pom (no functional change):
- Tell the hudi-common DecimalWrapper / Avro-1.8-1.9 story once (on the
hudi-spark bundle dep); the hudi-java-client exclusion just points to it.
- State "engine Avro must win on a flat classpath" once (engine-classpath
plugin); the avro dep and excludeGroupIds comments reference it.
- Explain jackson/scala-library are Spark-supplied once.
- Fix a stale "Spark 3.5" reference to Spark 3.4 (this is the Hudi 0.x
line).
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* [836] ci: trigger Maven CI + License Check on main-hudi-0x
Add main-hudi-0x to the push and pull_request branch filters of both
workflows (same change as #841) so CI runs on PRs targeting the Hudi 0.x
release line, including this one. A pull_request workflow's branch filter
takes effect from the PR's own branch, so this makes CI fire on #843.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* Address review feedback: guard ServiceLoader.hasNext() and drain
spark-submit stdout off-thread
- ConversionTargetFactory: hasNext() resolves provider classes lazily and
can
throw ServiceConfigurationError too, so move it inside the existing
ServiceConfigurationError|LinkageError guard alongside next().
- ITXTableSparkRuntimeBundle: drain spark-submit stdout on a background
thread
so a hung process is caught by waitFor(10min) instead of blocking forever
on
readOutput() reading until EOF.
* [836] ci: point Maven CI + License Check triggers at branch-0.4
main-hudi-0x was deleted and replaced by branch-0.4 (the 0.4.0 release
line),
so update the push/pull_request branch filters accordingly.
* [836] ci: validate xtable-spark-runtime bundle against a real Spark distro
ITXTableSparkRuntimeBundle spark-submits the shaded bundle jar and
self-skips unless SPARK_HOME is set, so the main CI never runs it. Add a
path-filtered workflow that installs a matching Spark 3.4 distribution
from the Apache archive, sets SPARK_HOME, and runs the failsafe IT.
* [836] ci: run spark-runtime validation on every branch PR (drop path
filter)
The workflow is intended to be a required status check on branch-0.4. A
required check whose workflow is skipped by a path filter never reports,
leaving PRs blocked on a pending check, so run it unconditionally on the
covered branches. The Spark distro is cached, so the added cost is the
reactor build plus the ~20s IT.
* [836] xtable-spark-runtime: address review feedback (CLI validation, unix
flags, docs)
- XTableSparkSync: validate --sourceformat/--targets up front and fail fast
(before SparkSession creation) on empty or unsupported values. Split the
allowed sets: sources are Hudi/Iceberg/Delta/Paimon/Parquet, targets are
Hudi/Iceberg/Delta (Paimon/Parquet are read-only, no ConversionTarget).
- XTableSyncService.sourceProviderFor: wire Paimon and Parquet source
providers
(previously threw UnsupportedOperationException); engines remain
cluster-provided.
- Rename CLI long-opts to unix-style lowercase (--basepath, --sourceformat,
--tablename, --datapath, --partitionspec); update javadoc, IT and RFC
example.
- basePathToName -> basePathToTableName: handle "/", trailing slashes and
null
by throwing with a "pass --tablename" hint instead of an empty table name.
- Add XTableSparkSyncTest covering table-name derivation and source/target
format validation.
- ConversionTargetFactory: log the skipped provider's error class/message so
operators can distinguish an intentionally-absent engine from a linkage
error.
- spark-runtime-validation.yml: add a workflow_dispatch spark_version input
and
document why the Spark 3.4 line is pinned (Delta 2.4.0 is Spark-3.4-only).
- RFC-3: add a supported-formats/engine-versions section, "from application
code" (Scala/Java + PySpark) activation examples, and list @vinothchandar
as
an approver.
* [836] xtable-spark-runtime: support Spark 3.5 via Delta Kernel
Delta source/target now auto-switch from delta-core to the Spark-free
Delta Kernel implementation on Spark 3.5+ (where the bundled delta-core
does not run), controlled by a single --usedeltakernel toggle that is
auto-enabled by Spark version. Hudi/Iceberg sync was already Spark-free.
Also runs the spark-runtime bundle IT on both Spark 3.4.3 and 3.5.9 via
a CI matrix.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* [836] Give Maven CI Build and License Check unique job names
Both workflows used a job named "build", so both reported the same
status-check context and could not be required distinctly. Set unique
job names so branch protection (see apache/incubator-xtable#848) can
require each on branch-0.4.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
* [836] xtable-spark-runtime: add --datasetconfig for multi-table sync
XTableSparkSync accepts a --datasetconfig YAML (sourceFormat, targetFormats,
datasets[]) to sync multiple tables in one run, mutually exclusive with the
single-table --basepath/--sourceformat/--targets flags. The config is read
through the Spark Hadoop config, so it may live on a local or cloud
(s3/gcs/abfs) path, and reuses the same schema as the RunSync utility.
Parsed with SnakeYAML's SafeConstructor (plain maps/lists only, no arbitrary
type instantiation). SnakeYAML is bundled and relocated to
org.apache.xtable.shaded (like guava/commons-cli) because Spark ships its
own
version (1.33 on 3.4, 2.0 on 3.5) that would otherwise clash; its
multi-release
classes are filtered out so no un-relocated org.yaml classes remain.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.github/workflows/mvn-ci-build.yml | 1 +
.github/workflows/mvn-license-check.yml | 1 +
.github/workflows/spark-runtime-validation.yml | 93 +++++
pom.xml | 3 +
rfc/rfc-3/rfc-3.md | 297 ++++++++++++++
.../xtable/conversion/ConversionTargetFactory.java | 36 +-
xtable-spark-runtime/pom.xml | 407 +++++++++++++++++++
.../org/apache/xtable/spark/TableSyncSpec.java | 61 +++
.../org/apache/xtable/spark/XTableSparkSync.java | 440 +++++++++++++++++++++
.../org/apache/xtable/spark/XTableSyncService.java | 145 +++++++
.../src/main/resources/META-INF/LICENSE-bundled | 222 +++++++++++
.../src/main/resources/META-INF/NOTICE-bundled | 22 ++
.../xtable/spark/ITXTableSparkRuntimeBundle.java | 318 +++++++++++++++
.../apache/xtable/spark/XTableSparkSyncTest.java | 197 +++++++++
14 files changed, 2240 insertions(+), 3 deletions(-)
diff --git a/.github/workflows/mvn-ci-build.yml
b/.github/workflows/mvn-ci-build.yml
index 5e3e1b76..3a7050de 100644
--- a/.github/workflows/mvn-ci-build.yml
+++ b/.github/workflows/mvn-ci-build.yml
@@ -33,6 +33,7 @@ on:
jobs:
build:
+ name: Maven CI Build
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
diff --git a/.github/workflows/mvn-license-check.yml
b/.github/workflows/mvn-license-check.yml
index 970e982d..193bc286 100644
--- a/.github/workflows/mvn-license-check.yml
+++ b/.github/workflows/mvn-license-check.yml
@@ -33,6 +33,7 @@ on:
jobs:
build:
+ name: License Check
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
diff --git a/.github/workflows/spark-runtime-validation.yml
b/.github/workflows/spark-runtime-validation.yml
new file mode 100644
index 00000000..a817d0a5
--- /dev/null
+++ b/.github/workflows/spark-runtime-validation.yml
@@ -0,0 +1,93 @@
+#
+# 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
+
+# Validates the shaded xtable-spark-runtime bundle end-to-end:
ITXTableSparkRuntimeBundle
+# spark-submits the bundled jar against a real Spark distribution for one case
per direction
+# across Hudi, Iceberg and Delta. The IT self-skips unless SPARK_HOME is set,
so the main CI
+# never exercises it; this workflow installs each Spark distro in the matrix
and runs it.
+# Runs on both Spark 3.4 and 3.5: Hudi/Iceberg sync is Spark-free, and a Delta
target/source
+# auto-switches from delta-core to the Spark-free Delta Kernel on Spark 3.5+.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# Runs on every push/PR to the covered branches (no path filter) so the check
always
+# reports — required status checks that never trigger leave a PR blocked on a
pending check.
+on:
+ push:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ pull_request:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ 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/pom.xml b/pom.xml
index 434bb21e..3478a7aa 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>
@@ -937,6 +938,8 @@
<excludes>
<exclude>.idea/**</exclude>
<exclude>${target.dir.pattern}</exclude>
+ <!-- Generated by maven-shade-plugin (no license
header); not source. -->
+ <exclude>**/dependency-reduced-pom.xml</exclude>
<exclude>website/node_modules/**</exclude>
<exclude>website/build/**</exclude>
<exclude>website/.docusaurus/**</exclude>
diff --git a/rfc/rfc-3/rfc-3.md b/rfc/rfc-3/rfc-3.md
new file mode 100644
index 00000000..bb41415d
--- /dev/null
+++ b/rfc/rfc-3/rfc-3.md
@@ -0,0 +1,297 @@
+<!--
+ 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.
+-->
+# RFC-3: xtable-spark-runtime - in-job metadata sync via a thin Spark bundle
+
+## Proposers
+
+- @vinishjail97
+
+## Approvers
+- @vinothchandar
+- Anyone from the XTable community can approve/add feedback.
+
+## Status
+
+GH Feature Request: https://github.com/apache/incubator-xtable/issues/836
+
+> Please keep the status updated in `rfc/README.md`.
+
+## Abstract
+
+XTable's modules are published to Maven, so a user can depend on `xtable-core`
and assemble their own
+runtime today. But there is no maintained, thin, drop-in artifact — everyone
re-solves the same
+shading/classpath problem, or uses `xtable-utilities`, an unshaded ~1 GB fat
jar that is not practical
+to add to a Spark job and requires running a separate `RunSync` process with
YAML config files.
+
+XTable conversion is metadata-only and lightweight, so the common case — "I
already write this table
+with Spark, keep it in sync in other formats" — should be a one-dependency,
config-only addition to an
+existing pipeline. This RFC proposes `xtable-spark-runtime`: a thin, relocated
Spark bundle that
+registers a driver-side listener and, after each successful write to a source
table, runs an
+incremental `ConversionController.sync(...)` for the configured targets. It
complements — does not
+replace — the standalone CLI (`RunSync`).
+
+## Background
+
+Everything needed to run a sync already exists in the engine; the gap is
purely packaging and a trigger:
+
+- The entire sync entry surface is Hadoop-`Configuration`-only: `new
ConversionController(conf).sync(config, sourceProvider)`.
+ The caller supplies only the **source** `ConversionSourceProvider`; target
providers are created
+ internally by `ConversionTargetFactory` (`ConversionController.java`).
Source providers instantiate
+ directly (`new HudiConversionSourceProvider()` /
`IcebergConversionSourceProvider` /
+ `DeltaConversionSourceProvider`) and take `.init(conf)`.
+- `SyncMode.INCREMENTAL` is self-healing: `ConversionController` auto-falls
back to a full snapshot
+ per-target when there is no prior sync metadata or incremental is not safe
(`isIncrementalSyncSufficient`).
+ First sync is effectively full; subsequent syncs are incremental.
+- The sync watermark (last-synced instant, pending commits) is persisted in
the **target's**
+ `TableSyncMetadata`. There is no additional state to keep on the client side.
+- `xtable-hudi-support-extensions` is a pure-Java precedent for a
Spark-adjacent module whose
+ `XTableSyncTool` already builds `SourceTable`/`TargetTable` and calls
`ConversionController.sync(...)`;
+ it declares Spark deps as `provided` and uses a Scala-suffixed artifactId.
+- Delta conversion currently runs Delta-on-Spark in-process (`delta-core`);
`DeltaConversionUtils.buildSparkSession`
+ uses `SparkSession.builder()...getOrCreate()` with no `master`, so it reuses
the host job's active
+ session rather than starting a second `SparkContext`. (A Spark-free Delta
**Kernel** path exists and,
+ once it becomes the default, removes the per-Spark-version coupling —
tracked separately.)
+
+## Implementation
+
+A new **pure-Java** module `xtable-spark-runtime_${scala.binary.version}`
(package `org.apache.xtable.spark`).
+
+### What ships in v1
+
+This PR delivers the **thin bundle plus the standalone `spark-submit` entry
point
+`XTableSparkSync`** — enough to run a relocated, in-cluster sync from one
command. The config-only,
+driver-side auto-sync (`XTableSyncListener` registered via
`spark.sql.queryExecutionListeners`, with
+`XTableSparkConfig` and `PlanTargetResolver`) is the headline on-ramp this RFC
designs toward, but it
+is **deferred to a follow-up PR** — those classes are not in v1. The listener
design is retained
+below as the target so the entry point and config schema stay aligned.
+
+### Supported formats and engine versions (v1)
+
+The bundle can read from Hudi, Iceberg, Delta, Paimon and Parquet, and write
to Hudi, Iceberg and
+Delta. Paimon and Parquet are **source-only** (there is no Paimon/Parquet
`ConversionTarget`), so they
+are valid for `--sourceformat` but not `--targets`. Syncing a format to itself
is a no-op and not a
+supported combination.
+
+Source \ Target:
+
+- Hudi source: to Iceberg yes, to Delta yes, to Hudi no (self)
+- Iceberg source: to Hudi yes, to Delta yes, to Iceberg no (self)
+- Delta source: to Hudi yes, to Iceberg yes, to Delta no (self)
+- Paimon source (read-only): to Hudi yes, to Iceberg yes, to Delta yes
+- Parquet source (read-only): to Hudi yes, to Iceberg yes, to Delta yes
+
+Engines are **provided by the cluster**, not bundled — the user supplies the
engine jars on the
+classpath (see Packaging / classpath notes below). This bundle targets the
Spark 3.4 line, pinned to a
+mutually-compatible engine set: Spark 3.4.x, Scala 2.12, Hudi 0.14.0, Iceberg
1.9.2, Delta 2.4.0.
+Delta 2.4.0 is Spark-3.4-only, which is why the line is pinned to Spark 3.4
rather than 3.5. The
+spark-submit smoke test (`ITXTableSparkRuntimeBundle`) currently exercises one
case per direction for
+Hudi/Iceberg/Delta; Paimon and Parquet sources are wired but not yet covered
by the smoke test.
+
+### Activation — v1 (CLI entry point)
+
+```
+spark-submit \
+ --jars /path/to/xtable-spark-runtime_2.12-<ver>.jar \
+ --class org.apache.xtable.spark.XTableSparkSync \
+ /path/to/xtable-spark-runtime_2.12-<ver>.jar \
+ --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
--tablename orders
+```
+
+Because the shaded jar is the **main artifact with a dependency-reduced POM**
(see Packaging),
+`--packages org.apache.xtable:xtable-spark-runtime_2.12:<ver>` resolves the
same relocated bundle and
+pulls no un-relocated transitive deps; `--jars` with the built jar is
equivalent.
+
+**Engine Avro / classpath placement (required).** The bundle does not carry
avro/parquet — it uses
+the engine's. Iceberg 1.9.2 compiles against Avro >= 1.12, newer than the Avro
Spark 3.4 ships (1.11),
+so the engine's Avro must win on the classpath. Supply the engine jars on a
**flat** classpath
+(`spark.driver.extraClassPath`/`spark.executor.extraClassPath`, or `--jars`),
not layered under a
+child classloader: `--packages` puts a second Avro in a child loader, which
breaks cross-boundary Avro
+casts (`NoSuchMethodError`/`ClassCastException`). This is why the bundle smoke
test builds a single
+flat `spark.driver.extraClassPath`.
+
+### Activation — v1 (from application code)
+
+The same sync can be triggered from an existing Spark job (Scala/Java or
PySpark) instead of a
+separate `spark-submit`, reusing the active `SparkSession`'s Hadoop
configuration. This is the
+programmatic equivalent of the CLI above and runs after your own write
completes.
+
+Scala / Java:
+
+```scala
+import org.apache.xtable.spark.{TableSyncSpec, XTableSyncService}
+
+// ... after writing the source table with `spark` ...
+val spec = TableSyncSpec.builder()
+ .key("orders")
+ .basePath("/warehouse/db/orders")
+ .sourceFormat("HUDI")
+ .targets(java.util.Arrays.asList("ICEBERG", "DELTA"))
+ .build()
+
+new XTableSyncService().sync(spec, spark.sparkContext.hadoopConfiguration)
+```
+
+PySpark (via the JVM gateway, with the bundle on the driver classpath):
+
+```python
+# after writing the source table with `spark` ...
+jvm = spark._jvm
+spec = (jvm.org.apache.xtable.spark.TableSyncSpec.builder()
+ .key("orders")
+ .basePath("/warehouse/db/orders")
+ .sourceFormat("HUDI")
+ .targets(jvm.java.util.Arrays.asList("ICEBERG", "DELTA"))
+ .build())
+hadoop_conf = spark._jsc.hadoopConfiguration()
+jvm.org.apache.xtable.spark.XTableSyncService().sync(spec, hadoop_conf)
+```
+
+These examples, plus a `--packages`/`--jars` quickstart, should be folded into
the docs-site
+quickstart in a docs follow-up.
+
+### Planned (follow-up): config-only listener activation
+
+```
+spark-submit --packages org.apache.xtable:xtable-spark-runtime_2.12:<ver> \
+ --conf
spark.sql.queryExecutionListeners=org.apache.xtable.spark.XTableSyncListener \
+ --conf spark.xtable.tables=orders \
+ --conf spark.xtable.orders.basePath=/warehouse/db/orders \
+ --conf spark.xtable.orders.sourceFormat=HUDI \
+ --conf spark.xtable.orders.targets=ICEBERG,DELTA
+```
+
+Config schema (`spark.xtable.*`):
+- `spark.xtable.tables` — comma-separated per-table keys (logical names).
+- Per key `<k>`: `spark.xtable.<k>.basePath` (path-based) **or**
`spark.xtable.<k>.sourceTable=db.table`
+ (name-based; resolved to a base path via the active `SparkSession` catalog);
plus
+ `spark.xtable.<k>.sourceFormat` and `spark.xtable.<k>.targets` (comma list).
Optional:
+ `spark.xtable.<k>.dataPath`, `spark.xtable.<k>.namespace`.
+
+Both path-based and name-based table selection are planned for the listener
on-ramp.
+
+### Components
+
+Shipped in v1:
+- **`XTableSyncService`** — for one `TableSyncSpec`: build `SourceTable` +
`TargetTable`s +
+ `ConversionConfig(syncMode=INCREMENTAL)`, pick the source provider via a
small
+ `sourceProviderFor(format)` factory, and run
`ConversionController.sync(...)`. Mirrors `XTableSyncTool`.
+- **`TableSyncSpec`** — immutable description of one table to sync (key,
basePath, dataPath,
+ namespace, sourceFormat, targets).
+- **`XTableSparkSync`** — the standalone `spark-submit` entry point. Parses
CLI args (commons-cli)
+ into a `TableSyncSpec` and drives `XTableSyncService`.
+
+Planned (follow-up, **not in v1**):
+- **`XTableSparkConfig`** — parse `SparkConf`/`Map` (+ optional `SparkSession`
for name resolution)
+ into `List<TableSyncSpec>`, validating required keys and failing fast with
clear messages.
+- **`PlanTargetResolver`** — best-effort extraction of the written output path
from `qe.analyzed()`
+ (`InsertIntoHadoopFsRelationCommand`, `SaveIntoDataSourceCommand`, and the
DataSource-V2
+ `AppendData`/`OverwriteByExpression` nodes used by Iceberg). Isolated and
unit-testable; returns
+ `Optional<String>`.
+- **`XTableSyncListener implements QueryExecutionListener`** — **stateless**
(only immutable parsed
+ config). On `onSuccess`, resolve the written path (best-effort) and, if it
matches a configured
+ table (basePath equals or is a prefix of the written path), call
`XTableSyncService.sync(...)`
+ **inline (synchronous)**. Reads and writes to unconfigured tables are
ignored — the resolver only
+ returns a path for recognized write commands, so a read never triggers a
sync. Per-table failures
+ are caught (as `Throwable`) so one table can't stop the others or
destabilize Spark's listener bus.
+ `onFailure` logs only.
+
+### Execution model: synchronous and stateless (planned listener)
+
+The sync watermark already lives in the target's `TableSyncMetadata`, and sync
is incremental +
+idempotent, so a client-side dirty/pending/single-flight structure would only
duplicate authoritative
+state and could not be more correct. Therefore the listener keeps no execution
state.
+
+Spark delivers `QueryExecutionListener` callbacks asynchronously via the
driver's `LiveListenerBus`
+(off `SparkListenerSQLExecutionEnd`), not on the thread that ran `df.write` —
so a listener cannot
+block the write from returning, and the bus processes events on a single
dispatch thread. Running the
+sync inline on the callback therefore (a) needs no executor or locking, (b)
gets single-flight per
+table for free (the bus is single-threaded), and (c) completes before JVM exit
for a normally
+terminating job, because `SparkContext.stop()` drains the bus. A hard `kill
-9` may skip a queued
+callback; the next commit's sync self-heals it.
+
+The cost of inline execution is that it occupies the bus dispatch thread for
the sync duration —
+acceptable for fast metadata-only syncs, which is the common case.
+
+**Async / "when to trigger" is intentionally left to the user.** Whether to
offload the sync to a
+background thread, and even whether/when to trigger at all (only after the
final write, batching
+several writes, latency tolerance), depends on the job's DAG, which only the
user knows. Async is an
+opt-in follow-up (`spark.xtable.sync.async`), not the default.
+
+### Packaging (thin bundle)
+
+Scala-suffixed artifactId. **Spark, Hadoop, and the engine libraries (Hudi /
Iceberg / Delta) are all
+`provided`** — the user brings their own engine versions via the cluster or
the submit
+packages/jars flags, and the thin XTable bundle stays compatible across engine
versions (Hudi 1.1 /
+1.2, etc.). The `maven-shade-plugin` uses a **curated `artifactSet`
allowlist**: only XTable's own
+modules (`xtable-api`, `xtable-core`, `xtable-hudi-support-utils`) plus the
libraries XTable uses
+*purely internally* — `guava` (and its `failureaccess`) and `commons-cli` for
the CLI — which are
+relocated under `org.apache.xtable.shaded.*`. The allowlist is include-only,
so its implicit
+"provided by the cluster" assumptions (e.g. `jackson-datatype-jsr310` via
`xtable-api`,
+`log4j-1.2-api` via `xtable-core`) are documented in the pom; a
`dependency:analyze`/enforcer gate to
+keep it from silently under-bundling is a tracked follow-up.
+
+The shaded jar is published as the **main artifact with a dependency-reduced
POM**
+(`shadedArtifactAttached=false`, `createDependencyReducedPom=true`) — the same
shape as
+`iceberg-spark-runtime` and `hudi-spark-bundle`, so resolving the coordinate
(Maven dep or
+`--packages`) yields the relocated bundle and no un-relocated transitive deps.
Because the jar
+relocates and ships `guava`/`failureaccess`/`commons-cli` (all Apache-2.0),
its `LICENSE`/`NOTICE`
+carry those attributions (`LICENSE-bundled`/`NOTICE-bundled` via
`IncludeResourceTransformer`).
+`protobuf` is **not** bundled — it resolves as `provided` (supplied by the
engines), so it needs no
+relocation or notice.
+
+Critically, libraries XTable **exchanges across the engine API boundary must
not be relocated or
+bundled** — e.g. Hudi returns a real `org.apache.avro.Schema`, so a
relocated/bundled `avro` in the
+XTable jar produces a `NoSuchMethodError` at runtime. `avro` / `parquet` /
`jackson` / `commons` are
+therefore left to the provided runtime. Resulting bundle: **~3.8 MB**.
+
+Because `ConversionTargetFactory` discovers targets via `ServiceLoader` and
would otherwise fail if
+any registered engine (e.g. Delta) is absent, target discovery is made
**resilient**: providers whose
+engine library is not on the classpath are skipped, so a user can run with
just the engines they use
+(e.g. Hudi + Iceberg, no Delta). This is a small `xtable-core` change and is
validated by the bundle
+smoke test.
+
+A standalone `spark-submit` entry point (`XTableSparkSync`, the
RunSync-equivalent for this bundle)
+lets the shaded jar run a sync directly and is what the jar-validation test
drives.
+
+## Rollout/Adoption Plan
+
+- **No breaking changes.** This is a new, additive module; existing
`RunSync`/`xtable-utilities` and
+ `xtable-core` behavior is unchanged.
+- **No impact on existing users** unless they opt in by adding the jar and
setting `spark.xtable.*`.
+- **Spark support:** this bundle targets the Spark line of its engine build
(Spark 3.4, Scala 2.12),
+ with newer Spark lines as follow-ups. The per-Spark-version coupling
+ comes from the Delta-on-Spark path; making Delta Kernel the default (tracked
separately) would let a
+ single bundle serve multiple Spark lines.
+- **Deferred (follow-up RFCs/PRs):** the config-only `XTableSyncListener`
on-ramp, a
+ `StreamingQueryListener` variant, a `CALL xtable.sync(...)` SQL procedure,
and the async execution
+ opt-in.
+
+## Test Plan
+
+- **Bundle smoke test (`ITXTableSparkRuntimeBundle`), v1:** `spark-submit` the
shaded main jar via
+ `XTableSparkSync` for one case per direction (Hudi↔Iceberg, Hudi↔Delta),
supplying the engines on a
+ **flat** `spark.driver.extraClassPath` (single Avro on one loader), and
assert the target is
+ data-equivalent to the source by reading both back through Spark (Hudi with
metadata enabled;
+ Iceberg non-vectorized to avoid a cross-engine timestamp Arrow-cast).
`SPARK_HOME`-gated. This
+ validates that the relocated jar is self-contained and runs a real sync.
+- **Planned (with the listener follow-up):** unit tests for
`XTableSparkConfig` parsing (path- and
+ name-based, missing-key errors) and `PlanTargetResolver` path extraction;
and an
+ `ITXTableSyncListener` embedded `local[*]` test that registers the listener
via
+ `spark.sql.queryExecutionListeners` and poll-asserts both targets appear
after a write.
+- **Bundle sanity:** package the module and inspect the shaded main jar —
confirm it is a few MB
+ (~3.8 MB), `org.apache.spark`/`org.apache.hadoop`/`org.apache.avro` are
absent, and the relocated
+ libraries live under `org.apache.xtable.shaded.*`.
diff --git
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
index f1e7bbb6..0d924d95 100644
---
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
+++
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java
@@ -18,20 +18,23 @@
package org.apache.xtable.conversion;
+import java.util.Iterator;
import java.util.Properties;
+import java.util.ServiceConfigurationError;
import java.util.ServiceLoader;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
+import lombok.extern.log4j.Log4j2;
import org.apache.hadoop.conf.Configuration;
import org.apache.xtable.delta.DeltaConversionTargetConfig;
import org.apache.xtable.exception.NotSupportedException;
-import org.apache.xtable.kernel.DeltaKernelConversionTarget;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.ConversionTarget;
+@Log4j2
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class ConversionTargetFactory {
private static final ConversionTargetFactory INSTANCE = new
ConversionTargetFactory();
@@ -87,7 +90,31 @@ public class ConversionTargetFactory {
TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
&&
DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
ServiceLoader<ConversionTarget> loader =
ServiceLoader.load(ConversionTarget.class);
- for (ConversionTarget target : loader) {
+ Iterator<ConversionTarget> iterator = loader.iterator();
+ while (true) {
+ ConversionTarget target;
+ try {
+ // hasNext() also resolves provider classes lazily, so it can throw
+ // ServiceConfigurationError too - it must be inside the guard
alongside next().
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
+ // A registered target whose engine library is not on the classpath
(e.g. Delta when only
+ // Hudi/Iceberg are provided). Skip it so a subset of engines can
still be used; a missing
+ // engine for the requested format surfaces below as
NotSupportedException. The offending
+ // provider is consumed before the error is thrown, so the next
hasNext() advances past it.
+ log.warn(
+ "Skipping a registered ConversionTarget whose engine library is
not on the classpath "
+ + "({}: {}); provide the missing engine if you need this
target format. This is "
+ + "expected when an engine is intentionally absent, but
indicates a linkage problem "
+ + "if the engine is present.",
+ error.getClass().getName(),
+ error.getMessage(),
+ error);
+ continue;
+ }
if (target.getTableFormat().equalsIgnoreCase(tableFormatName)
&& isDeltaKernelTarget(target) == useKernel) {
return target;
@@ -96,7 +123,10 @@ public class ConversionTargetFactory {
throw new NotSupportedException("Target format is not yet supported: " +
tableFormatName);
}
+ private static final String DELTA_KERNEL_TARGET_CLASS =
+ "org.apache.xtable.kernel.DeltaKernelConversionTarget";
+
private static boolean isDeltaKernelTarget(ConversionTarget target) {
- return target instanceof DeltaKernelConversionTarget;
+ return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName());
}
}
diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml
new file mode 100644
index 00000000..e1c07f02
--- /dev/null
+++ b/xtable-spark-runtime/pom.xml
@@ -0,0 +1,407 @@
+<?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.2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>
+ Thin, relocated, drop-in Spark bundle that runs incremental XTable
metadata sync. v1 ships
+ the standalone spark-submit entry point (XTableSparkSync); the
config-only, driver-side
+ QueryExecutionListener on-ramp is a planned follow-up (see rfc/rfc-3).
+ </description>
+
+ <dependencies>
+ <!-- XTable engine (bundled) -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Command-line parsing for the standalone XTableSparkSync entry
point (bundled). -->
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <!-- Parses the optional dataset-config YAML. Bundled (relocated
below) because it is not
+ reliably on a Spark runtime classpath; Apache-2.0 and has no
transitive dependencies. -->
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>2.2</version>
+ </dependency>
+
+ <!-- Logging API (lombok @Log4j2); provided by the cluster runtime -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Spark: provided by the cluster. Brings QueryExecutionListener,
QueryExecution,
+ Catalyst LogicalPlan, and SparkSession. -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Hadoop: provided by the cluster. -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!--
+ Engine libraries (Hudi / Iceberg / Delta) are provided, not bundled:
the user brings their
+ own engine version and the thin bundle stays compatible across
versions (see the shade
+ plugin below). Hudi is supplied as the hudi-spark bundle, whose
regenerated DecimalWrapper
+ links cleanly on Avro >=1.10; raw hudi-common's
Avro-1.8/1.9-generated model classes
+ reference org.apache.avro.data.TimeConversions.TimeConversion,
removed in Avro 1.10 and thus
+ absent from the Avro 1.12 that Iceberg 1.9.2 requires.
+ -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+
<artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <!--
+ XTable's Hudi TARGET writes via the Java client
(HoodieJavaEngineContext), which the Spark
+ bundle does not ship. Exclude its transitive hudi-common so the
bundle's clean DecimalWrapper
+ is the only one on the classpath (see above).
+ -->
+ <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>
+
+ <!--
+ avro/parquet-avro are provided by the engine runtime (Iceberg 1.9.2
needs Avro >=1.12, newer
+ than Spark 3.4's). Kept provided so the smoke test puts them on the
flat engine classpath;
+ why the engine's Avro must win is explained on the engine-classpath
plugin below.
+ -->
+ <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>
+ <!-- Hudi uses jol for in-memory sizing; the user's Hudi runtime
supplies it. Provided so it
+ is on the flat engine classpath for the smoke test. -->
+ <dependency>
+ <groupId>org.openjdk.jol</groupId>
+ <artifactId>jol-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!--
+ Jackson, scala-library and the log4j 1.2 bridge arrive transitively
through xtable-core but
+ are supplied by every Spark runtime (Spark itself depends on them).
Declaring them provided
+ here (nearest-wins over the transitive compile scope) keeps them out
of the shaded jar and
+ out of this module's runtime dependency tree, so the shade allowlist
stays limited to what
+ XTable actually bundles - which is what
release/scripts/validate_shaded_license_coverage.sh
+ enforces (shade <includes> must equal the runtime tree, and every
bundled dep must be listed
+ in META-INF/LICENSE-bundled). guava (used internally, relocated) is
intentionally left to
+ bundle. Versions come from the parent's dependencyManagement.
+ -->
+ <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>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <!-- Iceberg Spark runtime to create the Iceberg source table in-JVM
and read targets back.
+ Hudi's in-JVM table creation uses the provided hudi-spark bundle
above. -->
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+
<artifactId>iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Junit -->
+ <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>
+
+ <!-- Test logging -->
+ <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>
+ <!--
+ The bundle smoke test (ITXTableSparkRuntimeBundle) spark-submits
the shaded jar with
+ the engine libraries on a FLAT classpath
(spark.driver.extraClassPath), not via the
+ submit packages/jars flags. A layered packages/jars classloader
puts the engines' avro
+ in a child loader separate from Spark's own avro, which breaks
cross-boundary casts
+ (e.g. iceberg PartitionData to avro IndexedRecord). Resolve the
engine (provided-scope)
+ jars to a flat classpath here, excluding Spark/Hadoop (the
cluster provides them) and
+ xtable (bundled), so a single avro is on one loader, matching a
real cluster.
+ -->
+ <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>
+ <!--
+ Exclude Spark/Hadoop, XTable's own modules,
logging, and the libraries
+ Spark itself supplies (jackson, scala-library)
so cluster copies win;
+ avro/parquet stay so the engine's newer avro
wins (see above).
+ -->
+
<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>
+ <!--
+ Thin, curated bundle. Spark/Hadoop and the engine libraries
(Hudi/Iceberg/Delta) are
+ provided by the user's runtime, and so are the libraries XTable
exchanges with them
+ (avro, parquet, jackson, commons) - relocating those would break
the API boundary
+ (e.g. Hudi returning a real org.apache.avro.Schema to relocated
XTable code). The
+ bundle therefore contains only XTable's own modules plus the
libraries it uses purely
+ internally (guava), which are relocated to avoid clashes with
the cluster.
+
+ The shaded jar is the MAIN artifact
(shadedArtifactAttached=false) with a
+ dependency-reduced POM, matching how iceberg-spark-runtime and
hudi-spark-bundle
+ publish: resolving the coordinate (a plain Maven dependency or
the submit packages flag)
+ yields the relocated bundle and pulls no un-relocated transitive
deps. Consuming a thin,
+ un-shaded main jar plus a separate classifier would re-introduce
the cluster guava clash
+ that relocation exists to prevent.
+ -->
+ <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>
+ <!--
+ Include-only allowlist. Everything NOT listed
here is assumed to be
+ provided by the user's Spark/engine runtime and
is deliberately left out
+ of the jar. Known implicit "provided by cluster"
assumptions carried by
+ the bundled xtable modules:
jackson-datatype-jsr310 (via xtable-api) and
+ log4j-1.2-api (via xtable-core). If a future
compile dependency of
+ xtable-api/xtable-core is NOT cluster-provided,
it must be added here or
+ it will silently drop out and surface only as a
runtime
+ ClassNotFoundException. protobuf-java is
intentionally absent: it
+ resolves as provided (via the engines), so it is
never bundled and needs
+ no relocation. A hard verification gate
(dependency:analyze / enforcer)
+ over this allowlist is a tracked follow-up.
+ -->
+ <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>
+ <filters>
+ <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>
+ <!--
+ Drop SnakeYAML's multi-release (Java 9+)
copies: shade does not
+ relocate classes under META-INF/versions, so
keeping them would
+ leave un-relocated org.yaml.snakeyaml
classes in the bundle. The
+ base (relocated) classes work on all
supported Java versions.
+ -->
+ <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>
+ <!--
+ Bundled-dependency attribution: the shaded
jar relocates and ships
+ guava, failureaccess and commons-cli (all
Apache-2.0), so its
+ LICENSE/NOTICE must carry their attributions
per ASF policy.
+ -->
+ <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..37f7f56c
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+
+import lombok.Builder;
+import lombok.NonNull;
+import lombok.Value;
+
+/**
+ * Immutable description of a single table to keep in sync, resolved from
{@code spark.xtable.*}
+ * configuration. One {@link TableSyncSpec} is produced per key listed in
{@code
+ * spark.xtable.tables}.
+ */
+@Value
+@Builder
+public class TableSyncSpec {
+ /** The logical key for this table (the token listed in {@code
spark.xtable.tables}). */
+ @NonNull String key;
+
+ /** Absolute base path of the source table. */
+ @NonNull String basePath;
+
+ /** Optional path to the data files; defaults to {@link #basePath}
downstream when null. */
+ String dataPath;
+
+ /** Optional namespace segments for the table. */
+ String[] namespace;
+
+ /**
+ * Optional Hudi source partition spec (e.g. {@code level:VALUE}); only
applies to a partitioned
+ * Hudi source. Maps to {@code
xtable.hudi.source.partition_field_spec_config}.
+ */
+ 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..7360144e
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
@@ -0,0 +1,440 @@
+/*
+ * 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.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+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 org.apache.xtable.model.storage.TableFormat;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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. Must stay in sync with the source
providers wired in {@link
+ * XTableSyncService#sourceProviderFor}. Paimon and Parquet are read-only
source formats (no
+ * corresponding ConversionTarget exists).
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /**
+ * Formats usable as --targets. Must stay in sync with the {@code
ConversionTarget}s registered on
+ * the classpath (Hudi/Iceberg/Delta); Paimon and Parquet have no target
implementation. Used with
+ * {@link #SUPPORTED_SOURCE_FORMATS} to fail fast on invalid formats before
a SparkSession starts.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(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");
+ }
+ // Single-table flags are validated up front so an invalid value fails
fast, before a
+ // SparkSession is created. A --datasetconfig is on a local or cloud path,
so it is read through
+ // the Spark Hadoop config below and validated there.
+ 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));
+ }
+
+ // Use the Spark-free Delta Kernel when --usedeltakernel is passed, or
auto-enable it when
+ // Delta is involved and Spark is 3.5+ (where the bundled delta-core
does not run). One
+ // decision for the run, applied to every table.
+ 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();
+ int failures = 0;
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ TableSyncSpec spec = builder.useDeltaKernel(useDeltaKernel).build();
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(),
spec.getBasePath());
+ service.sync(spec, hadoopConf);
+ log.info("Completed XTable sync for {}", spec.getKey());
+ } catch (RuntimeException e) {
+ // One table failing should not abort the rest of the batch; report
at the end.
+ failures++;
+ log.error("XTable sync failed for {} ({})", spec.getKey(),
spec.getBasePath(), e);
+ }
+ }
+ if (failures > 0) {
+ throw new IllegalStateException(
+ failures + " of " + builders.size() + " table sync(s) failed; see
logs above");
+ }
+ } finally {
+ spark.stop();
+ }
+ }
+
+ private static void requireOption(CommandLine cmd, String option) {
+ if (!cmd.hasOption(option)) {
+ throw new IllegalArgumentException(
+ "--" + option + " is required unless --" + DATASET_CONFIG + " is
provided");
+ }
+ }
+
+ // Splits the comma-separated --targets value, upper-cases and validates
each, and requires at
+ // least one. 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;
+ }
+
+ // Reads the --datasetconfig YAML (local or cloud path) through the Spark
Hadoop config.
+ 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);
+ }
+ }
+
+ // Parses the dataset YAML with SnakeYAML's SafeConstructor (plain
maps/lists/scalars only, no
+ // arbitrary-type instantiation) and validates required fields.
package-private for unit testing.
+ @SuppressWarnings("unchecked")
+ 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");
+ }
+ Map<String, Object> root = (Map<String, Object>) loaded;
+
+ DatasetConfig config = new DatasetConfig();
+ config.sourceFormat = requireString(root,
"sourceFormat").trim().toUpperCase(Locale.ROOT);
+ validateFormat(SOURCE_FORMAT, config.sourceFormat,
SUPPORTED_SOURCE_FORMATS);
+
+ Object targetsRaw = root.get("targetFormats");
+ if (!(targetsRaw instanceof List) || ((List<?>) targetsRaw).isEmpty()) {
+ throw new IllegalArgumentException(
+ "--datasetconfig 'targetFormats' must be a non-empty list");
+ }
+ config.targetFormats =
+ ((List<Object>) targetsRaw)
+ .stream()
+ .map(String::valueOf)
+ .map(s -> s.trim().toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ config.targetFormats.forEach(t -> validateFormat(TARGETS, t,
SUPPORTED_TARGET_FORMATS));
+
+ Object datasetsRaw = root.get("datasets");
+ if (!(datasetsRaw instanceof List) || ((List<?>) datasetsRaw).isEmpty()) {
+ throw new IllegalArgumentException("--datasetconfig 'datasets' must be a
non-empty list");
+ }
+ config.datasets = new ArrayList<>();
+ for (Object entry : (List<Object>) datasetsRaw) {
+ if (!(entry instanceof Map)) {
+ throw new IllegalArgumentException("--datasetconfig 'datasets' entries
must be mappings");
+ }
+ Map<String, Object> raw = (Map<String, Object>) entry;
+ DatasetConfig.Table table = new DatasetConfig.Table();
+ table.tableBasePath = requireString(raw, "tableBasePath");
+ table.tableDataPath = asString(raw.get("tableDataPath"));
+ table.tableName = asString(raw.get("tableName"));
+ table.partitionSpec = asString(raw.get("partitionSpec"));
+ table.namespace = asString(raw.get("namespace"));
+ config.datasets.add(table);
+ }
+ return config;
+ }
+
+ private static String requireString(Map<String, Object> map, String key) {
+ String value = asString(map.get(key));
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException("--datasetconfig is missing required
field '" + key + "'");
+ }
+ return value;
+ }
+
+ private static String asString(Object value) {
+ return value == null ? null : String.valueOf(value);
+ }
+
+ /** 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);
+ }
+ }
+
+ /**
+ * Derives a table name from the last path segment of {@code basePath}, e.g.
{@code
+ * /warehouse/db/orders -> orders}. Trailing slashes are ignored. Throws
when no usable segment
+ * can be derived (e.g. {@code "/"} or an empty/blank path) so the caller is
told to pass {@code
+ * --tablename} explicitly instead of silently getting an empty table name.
+ */
+ // 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..e8650708
--- /dev/null
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -0,0 +1,145 @@
+/*
+ * 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;
+
+/**
+ * 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.
+ *
+ * <p>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<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());
+ }
+ // The data files may live at a different path than the source table root
(e.g. Iceberg keeps
+ // them under <basePath>/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();
+
+ // In Kernel mode, set the property ConversionTargetFactory reads to pick
the Delta Kernel
+ // writer (ignored for non-Delta targets).
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL,
Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
+ .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..cfe0372c
--- /dev/null
+++ b/xtable-spark-runtime/src/main/resources/META-INF/LICENSE-bundled
@@ -0,0 +1,222 @@
+ 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.
+
+
+Apache License 2.0
+------------------
+com.google.guava:guava:32.1.3-jre
+com.google.guava:failureaccess:1.0.1
+com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava
+com.google.errorprone:error_prone_annotations:2.21.1
+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/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..ce41f5d9
--- /dev/null
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java
@@ -0,0 +1,318 @@
+/*
+ * 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.
+ *
+ * <p>Covers Hudi, Iceberg and Delta as both source and target (one case per
direction). Hudi is
+ * supplied via the {@code hudi-spark} bundle (its regenerated Avro model
classes link on Avro 1.12,
+ * unlike raw hudi-common), so a {@code SPARK_HOME} matching the engine build
(Spark 3.4) is
+ * required. Local-timestamp columns are excluded from the equivalence check
due to known
+ * Hudi/Iceberg representation differences (HUDI-7088).
+ */
+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<Arguments> directions() {
+ // sourceFormat, targetFormat, hudiPartitionSpec (only used for a
partitioned Hudi source).
+ // Delta uses delta-core on Spark 3.4 and auto-switches to the Spark-free
Delta Kernel on Spark
+ // 3.5+ (see XTableSparkSync), so every direction runs on either Spark
line.
+ return Stream.of(
+ Arguments.of("HUDI", "ICEBERG", "level:VALUE"),
+ Arguments.of("ICEBERG", "HUDI", 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();
+ // Targets write metadata alongside the data files; for Iceberg this is
<basePath>/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<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.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);
+ }
+ // No kernel flag passed on purpose: XTableSparkSync auto-selects Delta
Kernel on Spark 3.5+.
+
+ ProcessBuilder pb = new ProcessBuilder(command);
+ pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1");
+ pb.redirectErrorStream(true);
+ Process process = pb.start();
+ // Drain stdout on a separate thread: reading until EOF only returns when
the process exits, so
+ // reading inline before waitFor() would block forever on a hung
spark-submit and never consult
+ // the timeout below. The background drain also keeps the pipe buffer from
filling and stalling
+ // the child.
+ 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);
+
+ 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 over the comparable
+ * scalar columns. A Hudi target is read with its metadata table enabled; an
Iceberg target is
+ * read non-vectorized to avoid a timestamp Arrow-vector cast on
cross-engine tables.
+ */
+ 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);
+ // Compare well-behaved scalar columns. Nested/array/map read parity
across writer and reader
+ // engines is covered in-process by ITConversionController;
local-timestamp columns are excluded
+ // as they have known Hudi/Iceberg representation differences (HUDI-7088).
Neither is what this
+ // smoke test validates - that the bundle jar runs the sync and produces
equivalent data.
+ String[] projection = comparableColumns(source, columns).toArray(new
String[0]);
+ List<String> sourceRows =
source.selectExpr(projection).toJSON().collectAsList();
+ List<String> targetRows =
+ sparkSession
+ .read()
+ .options(readOptions(targetFormat))
+ .format(targetFormat.toLowerCase())
+ .load(dataPath)
+ .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");
+ }
+
+ // Keeps the requested columns that compare cleanly across engines: drops
struct/array/map (reader
+ // parity covered by ITConversionController) and local-timestamp columns
(HUDI-7088 representation
+ // differences between Hudi and Iceberg).
+ private static List<String> comparableColumns(Dataset<Row> source,
List<String> columns) {
+ java.util.Set<String> excluded = 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) {
+ excluded.add(field.name());
+ }
+ }
+ return columns.stream()
+ .filter(c -> !excluded.contains(c))
+ .filter(c -> !c.startsWith("timestamp_local_"))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ private static java.util.Map<String, String> readOptions(String format) {
+ java.util.Map<String, String> 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;
+ }
+
+ 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;
+ }
+
+ // The shaded jar is the main artifact (shadedArtifactAttached=false), so it
carries no
+ // classifier.
+ // Skip the pre-shade jar the shade plugin renames to original-*, and the
sources/javadoc/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 (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) {
+ synchronized (sb) {
+ sb.append(line).append(System.lineSeparator());
+ }
+ }
+ } catch (java.io.IOException e) {
+ // Stream closed early (e.g. destroyForcibly on timeout) - stop
draining; whatever was
+ // captured so far is still surfaced in the failure message.
+ }
+ }
+}
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..9f41c922
--- /dev/null
+++
b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/XTableSparkSyncTest.java
@@ -0,0 +1,197 @@
+/*
+ * 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 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;
+
+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));
+ }
+
+ 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")));
+ }
+}
