From d201ebe80e2630bdb28f984e3b0f907491371a46 Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 7 Jul 2026 13:21:44 -0700 Subject: [PATCH 1/4] [836] Make ConversionTarget discovery resilient to missing engines ConversionTargetFactory loads targets via ServiceLoader and eagerly instantiates every registered target when resolving any one of them. With a thin runtime where engines are provided, that hard-fails (e.g. DeltaKernelConversionTarget needs io.delta.kernel) even for a sync that never touches Delta. Skip providers whose engine library is not on the classpath so a subset of engines can be used (e.g. Hudi + Iceberg, no Delta). Also make the Delta-kernel check name-based to avoid loading DeltaKernelConversionTarget (and its io.delta.kernel dependencies) when Delta is absent. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../conversion/ConversionTargetFactory.java | 26 ++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) 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 f1e7bbb6f..4aea5761a 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,19 @@ public ConversionTarget createConversionTargetForName( TableFormat.DELTA.equalsIgnoreCase(tableFormatName) && DeltaConversionTargetConfig.fromProperties(properties).isUseKernel(); ServiceLoader loader = ServiceLoader.load(ConversionTarget.class); - for (ConversionTarget target : loader) { + Iterator iterator = loader.iterator(); + while (iterator.hasNext()) { + ConversionTarget target; + try { + target = iterator.next(); + } catch (ServiceConfigurationError | LinkageError error) { + // A registered target whose engine library is not on the classpath (e.g. Delta when only + // Hudi/Iceberg are provided). Skip it so a subset of engines can still be used; a missing + // engine for the requested format surfaces below as NotSupportedException. + log.debug( + "Skipping ConversionTarget whose engine is not available on the classpath", error); + continue; + } if (target.getTableFormat().equalsIgnoreCase(tableFormatName) && isDeltaKernelTarget(target) == useKernel) { return target; @@ -96,7 +111,12 @@ && isDeltaKernelTarget(target) == useKernel) { throw new NotSupportedException("Target format is not yet supported: " + tableFormatName); } + private static final String DELTA_KERNEL_TARGET_CLASS = + "org.apache.xtable.kernel.DeltaKernelConversionTarget"; + + // Name-based to avoid loading DeltaKernelConversionTarget (and its io.delta.kernel dependencies) + // when Delta is not on the classpath; the target passed in is already loaded. private static boolean isDeltaKernelTarget(ConversionTarget target) { - return target instanceof DeltaKernelConversionTarget; + return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName()); } } From 62a03a91455e60f217348839e44ba8c01b85c96e Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 7 Jul 2026 13:21:56 -0700 Subject: [PATCH 2/4] [836] Add xtable-spark-runtime thin bundle + spark-submit validation New xtable-spark-runtime module producing a thin, drop-in bundle for in-job XTable metadata sync, plus RFC-3. Packaging (the Hudi/Iceberg model): - Spark, Hadoop, and the engine libs (Hudi/Iceberg/Delta) are provided; the user brings their own engine versions and the bundle stays compatible across them. - Curated shade allowlist: only XTable's own modules + guava/protobuf (relocated under org.apache.xtable.shaded.*). Libraries exchanged with the engines (avro/parquet/jackson/commons) are NOT relocated/bundled - relocating avro breaks the boundary (Hudi returns a real avro Schema). - Resulting bundle: ~4 MB. Includes XTableSparkSync, a standalone spark-submit entry point (the RunSync-equivalent for this bundle), and TableSyncSpec/XTableSyncService shared with the listener (stacked follow-up PR). Validation: ITXTableSparkRuntimeBundle spark-submits the shaded jar via $SPARK_HOME with Hudi + Iceberg supplied through --packages (versions from the root pom) and asserts the Iceberg metadata is produced. Skipped when SPARK_HOME is unset. Co-Authored-By: Claude Opus 4.8 (1M context) --- pom.xml | 1 + rfc/rfc-3/rfc-3.md | 181 +++++++++++++ xtable-spark-runtime/pom.xml | 244 ++++++++++++++++++ .../apache/xtable/spark/TableSyncSpec.java | 52 ++++ .../apache/xtable/spark/XTableSparkSync.java | 116 +++++++++ .../xtable/spark/XTableSyncService.java | 110 ++++++++ .../spark/ITXTableSparkRuntimeBundle.java | 137 ++++++++++ 7 files changed, 841 insertions(+) create mode 100644 rfc/rfc-3/rfc-3.md create mode 100644 xtable-spark-runtime/pom.xml create mode 100644 xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java create mode 100644 xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java create mode 100644 xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java create mode 100644 xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java diff --git a/pom.xml b/pom.xml index dcd23758c..144bf5b3b 100644 --- a/pom.xml +++ b/pom.xml @@ -54,6 +54,7 @@ xtable-aws xtable-hive-metastore xtable-service + xtable-spark-runtime diff --git a/rfc/rfc-3/rfc-3.md b/rfc/rfc-3/rfc-3.md new file mode 100644 index 000000000..014c0151b --- /dev/null +++ b/rfc/rfc-3/rfc-3.md @@ -0,0 +1,181 @@ + +# RFC-3: xtable-spark-runtime - in-job metadata sync via a thin Spark bundle + +## Proposers + +- @vinishjail97 + +## Approvers +- 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`). + +### Activation (config only) + +``` +spark-submit --packages org.apache.xtable:xtable-spark-runtime_2.12: \ + --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 ``: `spark.xtable..basePath` (path-based) **or** `spark.xtable..sourceTable=db.table` + (name-based; resolved to a base path via the active `SparkSession` catalog); plus + `spark.xtable..sourceFormat` and `spark.xtable..targets` (comma list). Optional: + `spark.xtable..dataPath`, `spark.xtable..namespace`. + +Both path-based and name-based table selection are supported from the first release. + +### Components + +- **`TableSyncSpec`** — immutable description of one configured table (key, basePath, dataPath, + namespace, sourceFormat, targets). +- **`XTableSparkConfig`** — parse `SparkConf`/`Map` (+ optional `SparkSession` for name resolution) + into `List`, validating required keys and failing fast with clear messages. +- **`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`. +- **`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`. +- **`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 + +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 v1 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 `protobuf` — which are relocated under `org.apache.xtable.shaded.*`. + +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: **~4 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:** target Spark 3.5 first (Scala 2.12), with a Spark 4 upgrade as a follow-up. 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):** a `StreamingQueryListener` variant, a `CALL xtable.sync(...)` SQL + procedure, and the async execution opt-in. + +## Test Plan + +- **Unit:** `XTableSparkConfig` parsing (path- and name-based, missing-key errors); + `PlanTargetResolver` path extraction from representative write plans (and empty result when + unresolvable). +- **Integration (`ITXTableSyncListener`, embedded `local[*]`):** register the listener via + `spark.sql.queryExecutionListeners`, configure a Hudi source with `targets=DELTA,ICEBERG`, write the + table, then poll-with-timeout (callback delivery is async) and assert both targets by reading them + back through Spark (`format("delta")` / `format("iceberg")`) and comparing row counts to the source, + mirroring `ITConversionController.checkDatasetEquivalence`. +- **Bundle sanity:** package the module and inspect the shaded jar — confirm size is tens of MB, + `org.apache.spark`/`org.apache.hadoop` are absent, and relocated libraries live under + `org.apache.xtable.shaded.*`. diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml new file mode 100644 index 000000000..6d4c83177 --- /dev/null +++ b/xtable-spark-runtime/pom.xml @@ -0,0 +1,244 @@ + + + + 4.0.0 + + + org.apache.xtable + xtable + 0.2.0-SNAPSHOT + + + xtable-spark-runtime_${scala.binary.version} + XTable Project Spark Runtime + + Thin, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a + driver-side QueryExecutionListener, activated through spark.xtable.* config only. + + + + + + org.apache.xtable + xtable-core_${scala.binary.version} + ${project.version} + + + + + org.apache.logging.log4j + log4j-api + provided + + + + + org.apache.spark + spark-sql_${scala.binary.version} + provided + + + + + org.apache.hadoop + hadoop-common + provided + + + + + org.apache.hudi + hudi-common + provided + + + org.apache.hudi + hudi-java-client + provided + + + org.apache.iceberg + iceberg-core + provided + + + org.apache.iceberg + iceberg-api + provided + + + io.delta + delta-core_${scala.binary.version} + provided + + + io.delta + delta-kernel-api + provided + + + io.delta + delta-kernel-defaults + provided + + + + + org.apache.xtable + xtable-core_${scala.binary.version} + ${project.version} + tests + test-jar + test + + + + org.apache.hudi + hudi-spark${spark.version.prefix}-bundle_${scala.binary.version} + test + + + org.openjdk.jol + jol-core + test + + + + + org.junit.jupiter + junit-jupiter-api + test + + + org.junit.jupiter + junit-jupiter-params + test + + + org.junit.jupiter + junit-jupiter-engine + test + + + + + org.apache.logging.log4j + log4j-core + test + + + org.apache.logging.log4j + log4j-slf4j2-impl + test + + + + + + + + org.apache.maven.plugins + maven-failsafe-plugin + + + org.apache.hudi:hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}:${hudi.version},org.apache.iceberg:iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}:${iceberg.version} + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + package + + shade + + + true + bundle + false + + + org.apache.xtable:xtable-api + org.apache.xtable:xtable-core_${scala.binary.version} + org.apache.xtable:xtable-hudi-support-utils + com.google.guava:guava + com.google.guava:failureaccess + com.google.protobuf:protobuf-java + + + + + com.google.common + org.apache.xtable.shaded.com.google.common + + + com.google.protobuf + org.apache.xtable.shaded.com.google.protobuf + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + **/Log4j2Plugins.dat + + + + + + + + + LICENSE + NOTICE + NOTICE.txt + + + + + + + + + + diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java new file mode 100644 index 000000000..5ae4ebe4c --- /dev/null +++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.xtable.spark; + +import java.util.List; + +import lombok.Builder; +import lombok.NonNull; +import lombok.Value; + +/** + * Immutable description of a single table to keep in sync, resolved from {@code spark.xtable.*} + * configuration. One {@link TableSyncSpec} is produced per key listed in {@code + * spark.xtable.tables}. + */ +@Value +@Builder +public class TableSyncSpec { + /** The logical key for this table (the token listed in {@code spark.xtable.tables}). */ + @NonNull String key; + + /** Absolute base path of the source table. */ + @NonNull String basePath; + + /** Optional path to the data files; defaults to {@link #basePath} downstream when null. */ + String dataPath; + + /** Optional namespace segments for the table. */ + String[] namespace; + + /** The source table format, e.g. {@code HUDI} (see {@code TableFormat}). */ + @NonNull String sourceFormat; + + /** The target formats to sync to, e.g. {@code [ICEBERG, DELTA]}. */ + @NonNull List targets; +} diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java new file mode 100644 index 000000000..5c7555f73 --- /dev/null +++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java @@ -0,0 +1,116 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.xtable.spark; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; +import org.apache.spark.sql.SparkSession; + +/** + * Standalone {@code spark-submit} entry point that runs a single XTable sync using the relocated + * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this bundle and is + * primarily used to validate that the shaded jar is self-contained on a real Spark install. + * + * 
+ * $SPARK_HOME/bin/spark-submit \
+ *   --class org.apache.xtable.spark.XTableSparkSync \
+ *   xtable-spark-runtime_2.12-<ver>.jar \
+ *   --basePath /warehouse/db/orders --sourceFormat HUDI --targets ICEBERG,DELTA
+ *
+ */ +@Log4j2 +public final class XTableSparkSync { + + private XTableSparkSync() {} + + public static void main(String[] args) { + Map opts = parseArgs(args); + String basePath = required(opts, "basePath"); + String sourceFormat = required(opts, "sourceFormat"); + String targets = required(opts, "targets"); + String tableName = opts.getOrDefault("tableName", basePathToName(basePath)); + + List targetFormats = + Arrays.stream(targets.split(",")) + .map(String::trim) + .filter(s -> !s.isEmpty()) + .map(s -> s.toUpperCase(java.util.Locale.ROOT)) + .collect(Collectors.toList()); + + TableSyncSpec spec = + TableSyncSpec.builder() + .key(tableName) + .basePath(basePath) + .dataPath(opts.get("dataPath")) + .namespace(opts.containsKey("namespace") ? opts.get("namespace").split("\\.") : null) + .sourceFormat(sourceFormat.toUpperCase(java.util.Locale.ROOT)) + .targets(targetFormats) + .build(); + + SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate(); + try { + Configuration hadoopConf = spark.sparkContext().hadoopConfiguration(); + log.info("Starting standalone XTable sync for {}", spec.getBasePath()); + new XTableSyncService().sync(spec, hadoopConf); + log.info("Completed XTable sync for {}", spec.getBasePath()); + } finally { + spark.stop(); + } + } + + private static Map parseArgs(String[] args) { + Map opts = new HashMap<>(); + List tokens = new ArrayList<>(Arrays.asList(args)); + for (int i = 0; i < tokens.size(); i++) { + String token = tokens.get(i); + if (token.startsWith("--")) { + String key = token.substring(2); + if (i + 1 < tokens.size() && !tokens.get(i + 1).startsWith("--")) { + opts.put(key, tokens.get(++i)); + } else { + opts.put(key, "true"); + } + } + } + return opts; + } + + private static String required(Map opts, String key) { + String value = opts.get(key); + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException("Missing required argument: --" + key); + } + return value.trim(); + } + + private static String basePathToName(String basePath) { + String trimmed = + basePath.endsWith("/") ? basePath.substring(0, basePath.length() - 1) : basePath; + int idx = trimmed.lastIndexOf('/'); + return idx >= 0 ? trimmed.substring(idx + 1) : trimmed; + } +} diff --git a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java new file mode 100644 index 000000000..2bf1f7394 --- /dev/null +++ b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java @@ -0,0 +1,110 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.xtable.spark; + +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; + +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.ConversionSourceProvider; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.delta.DeltaConversionSourceProvider; +import org.apache.xtable.hudi.HudiConversionSourceProvider; +import org.apache.xtable.iceberg.IcebergConversionSourceProvider; +import org.apache.xtable.model.storage.TableFormat; +import org.apache.xtable.model.sync.SyncMode; +import org.apache.xtable.model.sync.SyncResult; + +/** + * Builds a {@link ConversionConfig} from a {@link TableSyncSpec} and runs an incremental {@link + * ConversionController#sync} for it. This is the unit of work triggered by {@link + * XTableSyncListener} after a successful write. + * + *

The sync watermark is persisted in the target's {@code TableSyncMetadata}, and {@link + * SyncMode#INCREMENTAL} auto-falls back to a full snapshot when incremental is not safe (e.g. the + * very first sync), so this call is idempotent and self-healing. + */ +@Log4j2 +public class XTableSyncService { + + /** Runs a single sync for the given spec, returning the per-format results. */ + public Map sync(TableSyncSpec spec, Configuration hadoopConf) { + SourceTable sourceTable = + SourceTable.builder() + .name(spec.getKey()) + .basePath(spec.getBasePath()) + .dataPath(spec.getDataPath()) + .namespace(spec.getNamespace()) + .formatName(spec.getSourceFormat()) + .additionalProperties(new Properties()) + .build(); + + List targetTables = + spec.getTargets().stream() + .map( + targetFormat -> + TargetTable.builder() + .name(spec.getKey()) + .basePath(spec.getBasePath()) + .namespace(spec.getNamespace()) + .formatName(targetFormat) + .build()) + .collect(Collectors.toList()); + + ConversionConfig conversionConfig = + ConversionConfig.builder() + .sourceTable(sourceTable) + .targetTables(targetTables) + .syncMode(SyncMode.INCREMENTAL) + .build(); + + ConversionSourceProvider sourceProvider = sourceProviderFor(spec.getSourceFormat()); + sourceProvider.init(hadoopConf); + + log.info( + "Running XTable sync for table {} ({} -> {}) at {}", + spec.getKey(), + spec.getSourceFormat(), + spec.getTargets(), + spec.getBasePath()); + return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider); + } + + private static ConversionSourceProvider sourceProviderFor(String sourceFormat) { + switch (sourceFormat.toUpperCase()) { + case TableFormat.HUDI: + return new HudiConversionSourceProvider(); + case TableFormat.DELTA: + return new DeltaConversionSourceProvider(); + case TableFormat.ICEBERG: + return new IcebergConversionSourceProvider(); + default: + throw new UnsupportedOperationException( + "Unsupported source format for spark-runtime sync: " + sourceFormat); + } + } +} diff --git a/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java new file mode 100644 index 000000000..41568edf9 --- /dev/null +++ b/xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.xtable.spark; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.io.File; +import java.net.URI; +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.hudi.common.model.HoodieTableType; + +import org.apache.xtable.GenericTable; +import org.apache.xtable.TestJavaHudiTable; + +/** + * Validates that the relocated {@code xtable-spark-runtime} bundle jar is self-contained by running + * an actual XTable sync via {@code $SPARK_HOME/bin/spark-submit} against a locally created Hudi + * table. Skipped when {@code SPARK_HOME} is not set. + */ +class ITXTableSparkRuntimeBundle { + + @TempDir Path tempDir; + + @Test + void bundleRunsHudiToIcebergSyncViaSparkSubmit() throws Exception { + String sparkHome = System.getenv("SPARK_HOME"); + assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME is not set"); + + String tableName = "orders"; + String basePath; + try (GenericTable table = + TestJavaHudiTable.forStandardSchema( + tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) { + table.insertRows(10); + basePath = table.getBasePath(); + } + + File bundleJar = findBundleJar(); + File sparkSubmit = new File(sparkHome, "bin/spark-submit"); + assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " + sparkSubmit); + + // Engines are provided by the user's runtime, not bundled. Supply Hudi + Iceberg (matching the + // versions from the root pom) via --packages, exactly as a user would. Delta is intentionally + // absent: resilient target discovery skips it. + String enginePackages = System.getProperty("xtable.it.enginePackages"); + assertTrue( + enginePackages != null && !enginePackages.isEmpty(), + "xtable.it.enginePackages system property must be set by the failsafe plugin"); + + ProcessBuilder pb = + new ProcessBuilder( + sparkSubmit.getAbsolutePath(), + "--master", + "local[2]", + "--packages", + enginePackages, + "--class", + "org.apache.xtable.spark.XTableSparkSync", + bundleJar.getAbsolutePath(), + "--basePath", + basePath, + "--sourceFormat", + "HUDI", + "--targets", + "ICEBERG", + "--tableName", + tableName); + pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1"); + pb.redirectErrorStream(true); + + Process process = pb.start(); + String output = readOutput(process); + boolean finished = process.waitFor(10, TimeUnit.MINUTES); + if (!finished) { + process.destroyForcibly(); + throw new AssertionError("spark-submit timed out. Output:\n" + output); + } + int exit = process.exitValue(); + assertEquals(0, exit, "spark-submit failed (exit " + exit + "). Output:\n" + output); + + File metadataDir = new File(new File(URI.create(basePath)), "metadata"); + File[] metadataJson = metadataDir.listFiles((dir, name) -> name.endsWith("metadata.json")); + assertTrue( + metadataJson != null && metadataJson.length > 0, + "No Iceberg metadata produced under " + metadataDir + ". spark-submit output:\n" + output); + } + + private File findBundleJar() { + File targetDir = new File("target"); + File[] candidates = + targetDir.listFiles( + (dir, name) -> name.startsWith("xtable-spark-runtime") && name.endsWith("-bundle.jar")); + assertTrue( + candidates != null && candidates.length == 1, + "Expected exactly one bundle jar in " + + targetDir.getAbsolutePath() + + " (run the package phase first)"); + return candidates[0]; + } + + private static String readOutput(Process process) throws Exception { + StringBuilder sb = new StringBuilder(); + try (java.io.BufferedReader reader = + new java.io.BufferedReader( + new java.io.InputStreamReader( + process.getInputStream(), java.nio.charset.StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + sb.append(line).append(System.lineSeparator()); + } + } + return sb.toString(); + } +} From 9fcb4f836f3cb83ed2f293c6803ee15608dd331f Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 7 Jul 2026 13:28:13 -0700 Subject: [PATCH 3/4] [836] RFC-3: correct bundle size to ~3.6 MB Co-Authored-By: Claude Opus 4.8 (1M context) --- rfc/rfc-3/rfc-3.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rfc/rfc-3/rfc-3.md b/rfc/rfc-3/rfc-3.md index 014c0151b..8fe6cc499 100644 --- a/rfc/rfc-3/rfc-3.md +++ b/rfc/rfc-3/rfc-3.md @@ -144,7 +144,7 @@ modules (`xtable-api`, `xtable-core`, `xtable-hudi-support-utils`) plus the libr 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: **~4 MB**. +therefore left to the provided runtime. Resulting bundle: **~3.6 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 From 96da5d679a4cbcbb4c7faba522320f8aab721dbd Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 7 Jul 2026 16:05:26 -0700 Subject: [PATCH 4/4] [836] Address review: commons-cli parsing, warn on skipped target, real bundle IT Review feedback on PR #838: - XTableSparkSync now parses args with Apache Commons CLI (mirrors RunSync) instead of hand-rolled parsing; the relocated commons-cli is bundled. - ConversionTargetFactory logs a warn (was debug) when it skips a registered target whose engine library is absent, so a missing provided engine is visible; dropped the self-explanatory comment on isDeltaKernelTarget. Bundle integration test (ITXTableSparkRuntimeBundle): - Supplies engines on a flat classpath via spark.driver.extraClassPath, and adds Avro 1.12 + parquet-avro to the engine classpath: Iceberg 1.9.2 (bumped in #784) compiles against Avro 1.12 APIs, which Spark 3.5's Avro 1.11.2 lacks. A flat classpath keeps one Avro on one loader, unlike --packages layering. - Adds jol-core (provided) that Hudi needs at runtime. - XTableSyncService writes target metadata at the source data path (required by Hudi; Iceberg data lives under /data); IT passes --dataPath. - Readback mirrors ITConversionController: target read at the data path, Hudi metadata read options, per-format timestamp normalization, scalar-column data-equivalence, and non-vectorized Iceberg reads. - Relocates guava's com.google.thirdparty with the rest of guava. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../conversion/ConversionTargetFactory.java | 8 +- xtable-spark-runtime/pom.xml | 80 +++++- .../apache/xtable/spark/TableSyncSpec.java | 6 + .../apache/xtable/spark/XTableSparkSync.java | 125 ++++++--- .../xtable/spark/XTableSyncService.java | 18 +- .../spark/ITXTableSparkRuntimeBundle.java | 260 ++++++++++++++---- 6 files changed, 394 insertions(+), 103 deletions(-) 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 4aea5761a..ada649ab7 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 @@ -99,8 +99,10 @@ public ConversionTarget createConversionTargetForName( // A registered target whose engine library is not on the classpath (e.g. Delta when only // Hudi/Iceberg are provided). Skip it so a subset of engines can still be used; a missing // engine for the requested format surfaces below as NotSupportedException. - log.debug( - "Skipping ConversionTarget whose engine is not available on the classpath", error); + log.warn( + "Skipping a registered ConversionTarget whose engine library is not on the classpath; " + + "provide the missing engine if you need this target format", + error); continue; } if (target.getTableFormat().equalsIgnoreCase(tableFormatName) @@ -114,8 +116,6 @@ && isDeltaKernelTarget(target) == useKernel) { private static final String DELTA_KERNEL_TARGET_CLASS = "org.apache.xtable.kernel.DeltaKernelConversionTarget"; - // Name-based to avoid loading DeltaKernelConversionTarget (and its io.delta.kernel dependencies) - // when Delta is not on the classpath; the target passed in is already loaded. private static boolean isDeltaKernelTarget(ConversionTarget target) { return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName()); } diff --git a/xtable-spark-runtime/pom.xml b/xtable-spark-runtime/pom.xml index 6d4c83177..8f8db7438 100644 --- a/xtable-spark-runtime/pom.xml +++ b/xtable-spark-runtime/pom.xml @@ -41,6 +41,12 @@ ${project.version} + + + commons-cli + commons-cli + + org.apache.logging.log4j @@ -74,6 +80,12 @@ hudi-common provided + + org.apache.hudi + hudi-hadoop-common + ${hudi.version} + provided + org.apache.hudi hudi-java-client @@ -105,6 +117,30 @@ provided + + + org.apache.avro + avro + provided + + + org.apache.parquet + parquet-avro + provided + + + + org.openjdk.jol + jol-core + provided + + org.apache.xtable @@ -114,15 +150,15 @@ test-jar test - + org.apache.hudi hudi-spark${spark.version.prefix}-bundle_${scala.binary.version} test - org.openjdk.jol - jol-core + org.apache.iceberg + iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version} test @@ -160,17 +196,30 @@ org.apache.maven.plugins - maven-failsafe-plugin - - - org.apache.hudi:hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}:${hudi.version},org.apache.iceberg:iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}:${iceberg.version} - - + maven-dependency-plugin + + + engine-classpath + pre-integration-test + + build-classpath + + + provided + org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j + ${project.build.directory}/engine-classpath.txt + + +