Introduce xtable-spark-runtime: thin, relocated Spark bundle for in-job metadata sync

[vinishjail97]

Motivation

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 ~1GB fat
jar that isn't practical to add to a Spark job and requires running a
separate sync process with multiple 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.

Proposal

A new module xtable-spark-runtime producing a thin, relocated bundle that:

• is added via --jars / --packages,
• is activated through config only (no user code change),
• runs incremental ConversionController.sync(...) on the driver after each
  successful write to a source table.

Usage

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=/warehouse/db/orders \
  --conf spark.xtable.orders.sourceFormat=HUDI \
  --conf spark.xtable.orders.targets=ICEBERG,DELTA

Packaging (Hudi/Iceberg model)

• Spark / Hadoop → provided (on the cluster, not bundled).
• Curated dependency allowlist (no bundling of the full transitive closure).
• Relocate guava / jackson / protobuf / avro / commons under
  org.apache.xtable.shaded.* to avoid classpath clashes with the cluster.
• Target size: tens of MB.

Design

• XTableSyncListener implements QueryExecutionListener (batch) and a
  StreamingQueryListener variant.
• On write-success: resolve target table(s) from spark.xtable.*, submit an
  incremental sync to a debounced, single-flight-per-table driver executor.
• Sync is idempotent + incremental, so a missed trigger self-heals on the
  next commit.

Scope / non-goals

• No data rewriting — metadata translation only.
• Does not replace the standalone CLI (RunSync); complements it.

Open questions

• Table selection: config-declared (proposed) vs. plan-inspection auto-detect.
• Batch jobs: async (needs shutdown-hook drain) vs. blocking.
• Also ship a CALL xtable.sync(...) SQL procedure in the same jar?

Tasks

• Scaffold xtable-spark-runtime module (pom: provided engines, allowlist, relocations).
• Implement XTableSyncListener + spark.xtable.* config parsing.
• Debounced driver executor + shutdown drain.
• ITXTableSyncListener (embedded local[*], Hudi→Delta/Iceberg).
• spark-shell smoke test + README section.
• (optional) CALL xtable.sync(...) procedure.

Related

• DISCUSS thread: https://lists.apache.org/thread/w9lql69ovy0rd8mkljtly9f7z7kyb751
• Bundle/licensing history: dev@ thread "[DISCUSS] Excluding xtable-utilities from the release process", Review LICENSE files for xtable based on feedback provided in general@incubator.apache.org #536.

[vaibhavk1992]

Hi @vinishjail97
Thanks for initiating this discussion.

I have made a flowchart please confirm if my understanding is correct. I’ve also listed a few questions below.

---

  ┌─────────────────────────────────────────────────────┐                                                                  
  │                  Spark Driver JVM                   │                                                                  
  │                                                     │                                                                  
  │   Your job code                                     │                                                                  
  │       df.write.format("hudi").save("/orders")       │                                                                  
  │              │                                      │                                                                  
  │              ▼ write completes                      │                                                                  
  │   Spark fires onSuccess()                           │                                                                  
  │              │                                      │                                                                  
  │              ▼  ← still inside same JVM             │                                                                  
  │   XTableSyncListener                                │                                                                  
  │       ConversionController.sync()                   │                                                                  
  │              │                                      │                                                                  
  │              ▼                                      │                                                                  
  │   Iceberg/Delta metadata written                    │                                                                  
  │                                                     │                                                                  
  └─────────────────────────────────────────────────────┘ 

• Also which Spark versions will be supported — will there be separate artifacts for Spark 3.3, 3.4, 3.5 like Hudi does
  (spark3.3-bundle, spark3.4-bundle)?
• If CALL xtable.sync(table, targets) is shipped in the same jar, what is the intended use case — ad-hoc backfill,
  testing, or something else?
• Would the procedure support full sync (not just incremental) for cases where you need to rebuild from scratch?
• How does this handle catalog-based tables (Unity Catalog, Hive Metastore) where the table is referenced by name rather
  than path?

On the single-flight / concurrency model

• If a sync is already running for table X and another write completes for table X, is the second sync queued or dropped?
  If dropped, how is it guaranteed to run later?
• What is the thread model for the driver executor — is it a single dedicated thread per table or a shared pool?
• For batch jobs, what is the shutdown guarantee — if the job finishes while a sync is still running, does Spark wait for
  it or kill it?

[vinishjail97]

Thanks @vaibhavk1992 — flowchart's right on the shape. One refinement: the QueryExecutionListener callback doesn't run on the thread that executed df.write — Spark delivers it asynchronously via the driver's LiveListenerBus. The listener resolves the configured table and runs ConversionController.sync() inline on that callback, same driver JVM (see the concurrency note below for the exact model).

Spark versions: Right now the version matters mainly via the delta-core path (Delta-on-Spark, in-process). But we already have a full Delta Kernel implementation (org.apache.xtable.kernel.*, Spark-free) — @vaibhavk1992, since you authored the Kernel path: what are the remaining blockers to making Delta Kernel the default for Delta conversion? If we can close those, this runtime stops depending on a specific Spark line for Delta, and a single xtable-spark-runtime_2.12 bundle could serve both 3.5 and 4 (the listener APIs are stable across versions) — which would make the whole "per-Spark-version artifact" question mostly moot. Until Kernel is the default, we'd target Spark 3.5 first with Spark 4 as a follow-up.

CALL xtable.sync(...): ad-hoc/imperative path — onboarding an existing table, re-sync after a schema change, notebook/spark-sql, CI. Complements the listener; optional/stretch.

Full sync: yes — engine already has SyncMode.FULL vs INCREMENTAL (and INCREMENTAL auto-falls back to full when unsafe). "Rebuild from scratch" is just syncMode=FULL.

Catalog tables (UC/HMS by name): we'll support both from the start, path-based and name-based. The engine already handles name-based target catalogs (syncTableAcrossCatalogs, CatalogSync, Glue/HMS modules). UC (resolution + permissions) carries a bit more nuance but is in scope.

Concurrency / when to sync (revised):

• v1 is deliberately synchronous: on write-success the listener runs ConversionController.sync(...) inline on the callback. No executor, no queue, no dirty/single-flight state — the sync watermark already lives in the target's TableSyncMetadata, so re-tracking it in the listener would only duplicate it and couldn't be more correct. Single-flight falls out for free: Spark delivers QueryExecutionListener callbacks on the single-threaded LiveListenerBus, so two syncs for the same table can't overlap.
• What "synchronous" means precisely: Spark delivers the callback asynchronously off SparkListenerSQLExecutionEnd, so df.write returns before the sync runs — a listener can't block the write itself. Running inline means the sync occupies the bus thread until it completes; callbacks still queued at job end are drained by SparkContext.stop(). Because sync is incremental + idempotent, a genuinely missed trigger (e.g. kill -9) self-heals on the next commit.
• Async / "when to trigger" is left to the user, on purpose. 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. So async is an opt-in knob (follow-up), not a default; the default stays safe and deterministic within the job lifecycle. The user can write a custom listener/trigger depending on their DAG logic.

[vinishjail97]

Opened a draft PR with RFC-3 + a first working implementation: #837 (stateless, synchronous QueryExecutionListener that runs incremental ConversionController.sync(...) in-job, activated via spark.xtable.*; unit tests + an embedded local[*] Hudi→Delta/Iceberg IT all green).

Keeping it in draft until I've tested the bundled jar in a real Spark + K8s job and run the license checks. Will mark it ready for review after that.