vinishjail97 commented on code in PR #843: URL: https://github.com/apache/incubator-xtable/pull/843#discussion_r3555806247
########## xtable-spark-runtime/pom.xml: ########## @@ -0,0 +1,311 @@ +<?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, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a + driver-side QueryExecutionListener, activated through spark.xtable.* config only. + </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> + + <!-- 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 by the user's runtime, NOT bundled. + The user brings their own version via the cluster or the submit packages/jars flags, and + the thin XTable bundle stays compatible across engine versions. Declared provided so they + are on the compile/test classpath but excluded from the shaded jar. + --> + <!-- + Hudi is supplied by the cluster as the hudi-spark bundle (how Hudi ships for Spark). The + bundle carries a clean, regenerated DecimalWrapper; raw hudi-common's Avro-1.8/1.9-generated + model classes reference org.apache.avro.data.TimeConversions.TimeConversion, which Avro + removed in 1.10 and is therefore absent from the Avro 1.12 that Iceberg 1.9.2 requires. + Provided so it is on the compile/test/engine classpath but excluded from the shaded jar. + --> + <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 - raw hudi-common's DecimalWrapper is + Avro-1.8/1.9-generated and fails to link on the Avro >=1.10 required by Iceberg 1.9.2. + --> + <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 and parquet-avro pinned to the versions the engines require (Iceberg 1.9.2 compiles + against Avro 1.12 APIs), not the older Avro that Spark 3.5 ships. On a real Iceberg-1.9.2 + cluster the runtime must already supply these; declaring them provided puts them on the + flat engine classpath for the bundle smoke test so the engine's Avro wins over Spark's. + --> + <dependency> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.parquet</groupId> + <artifactId>parquet-avro</artifactId> + <scope>provided</scope> + </dependency> + <!-- 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> + + <!-- 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> + <excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j</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, protobuf), which are relocated to avoid clashes with the cluster. + --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedArtifactAttached>true</shadedArtifactAttached> Review Comment: Fixed in 4587fa77. The shaded jar is now the **main artifact** (`shadedArtifactAttached=false`) with `createDependencyReducedPom=true`, matching `iceberg-spark-runtime`/`hudi-spark-bundle`. Resolving the coordinate (Maven dep or `--packages`) now yields the relocated bundle with a dependency-reduced POM (only `provided`/`test` deps remain, none transitive), so no un-relocated guava/xtable-core is pulled onto the cluster classpath. The thin-jar + `-bundle` classifier is gone; the IT now picks up the shaded main jar. ########## rfc/rfc-3/rfc-3.md: ########## @@ -0,0 +1,181 @@ +<!-- + 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 +- 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:<ver> \ Review Comment: Fixed in 4587fa77. Now that the shaded jar is the main artifact with a dependency-reduced POM, `--packages org.apache.xtable:xtable-spark-runtime_2.12:<ver>` resolves the correct relocated bundle (verified: the deployed POM carries only `provided`/`test` deps, so nothing un-relocated is pulled). The RFC activation example was rewritten to the shipped CLI (`--jars ... --class XTableSparkSync`) and notes that `--packages` is equivalent. ########## rfc/rfc-3/rfc-3.md: ########## @@ -0,0 +1,181 @@ +<!-- + 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 +- 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:<ver> \ + --conf spark.sql.queryExecutionListeners=org.apache.xtable.spark.XTableSyncListener \ Review Comment: Fixed in 4587fa77. Scoped RFC-3 to what v1 actually ships: the standalone CLI `XTableSparkSync`. The config-only `XTableSyncListener` on-ramp (and `XTableSparkConfig`/`PlanTargetResolver`) is now clearly marked **deferred to a follow-up PR** in a "What ships in v1" section, the Components/Test-Plan sections, and the pom `<description>`. The `spark.sql.queryExecutionListeners=...XTableSyncListener` example moved under "Planned (follow-up)". ########## xtable-spark-runtime/pom.xml: ########## @@ -0,0 +1,311 @@ +<?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, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a + driver-side QueryExecutionListener, activated through spark.xtable.* config only. + </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> + + <!-- 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 by the user's runtime, NOT bundled. + The user brings their own version via the cluster or the submit packages/jars flags, and + the thin XTable bundle stays compatible across engine versions. Declared provided so they + are on the compile/test classpath but excluded from the shaded jar. + --> + <!-- + Hudi is supplied by the cluster as the hudi-spark bundle (how Hudi ships for Spark). The + bundle carries a clean, regenerated DecimalWrapper; raw hudi-common's Avro-1.8/1.9-generated + model classes reference org.apache.avro.data.TimeConversions.TimeConversion, which Avro + removed in 1.10 and is therefore absent from the Avro 1.12 that Iceberg 1.9.2 requires. + Provided so it is on the compile/test/engine classpath but excluded from the shaded jar. + --> + <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 - raw hudi-common's DecimalWrapper is + Avro-1.8/1.9-generated and fails to link on the Avro >=1.10 required by Iceberg 1.9.2. + --> + <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 and parquet-avro pinned to the versions the engines require (Iceberg 1.9.2 compiles + against Avro 1.12 APIs), not the older Avro that Spark 3.5 ships. On a real Iceberg-1.9.2 + cluster the runtime must already supply these; declaring them provided puts them on the + flat engine classpath for the bundle smoke test so the engine's Avro wins over Spark's. + --> + <dependency> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> Review Comment: Fixed in 4587fa77. Added a user-facing "Engine Avro / classpath placement (required)" section to RFC-3: the engine Avro (>=1.12 for Iceberg 1.9.2) must win over the Avro Spark 3.4 ships, so engines must go on a **flat** classpath (`spark.driver.extraClassPath`/`--jars`), not layered under `--packages` (a second Avro in a child loader breaks cross-boundary casts). This is exactly what the bundle smoke test does. ########## xtable-spark-runtime/pom.xml: ########## @@ -0,0 +1,311 @@ +<?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, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a + driver-side QueryExecutionListener, activated through spark.xtable.* config only. + </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> + + <!-- 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 by the user's runtime, NOT bundled. + The user brings their own version via the cluster or the submit packages/jars flags, and + the thin XTable bundle stays compatible across engine versions. Declared provided so they + are on the compile/test classpath but excluded from the shaded jar. + --> + <!-- + Hudi is supplied by the cluster as the hudi-spark bundle (how Hudi ships for Spark). The + bundle carries a clean, regenerated DecimalWrapper; raw hudi-common's Avro-1.8/1.9-generated + model classes reference org.apache.avro.data.TimeConversions.TimeConversion, which Avro + removed in 1.10 and is therefore absent from the Avro 1.12 that Iceberg 1.9.2 requires. + Provided so it is on the compile/test/engine classpath but excluded from the shaded jar. + --> + <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 - raw hudi-common's DecimalWrapper is + Avro-1.8/1.9-generated and fails to link on the Avro >=1.10 required by Iceberg 1.9.2. + --> + <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 and parquet-avro pinned to the versions the engines require (Iceberg 1.9.2 compiles + against Avro 1.12 APIs), not the older Avro that Spark 3.5 ships. On a real Iceberg-1.9.2 + cluster the runtime must already supply these; declaring them provided puts them on the + flat engine classpath for the bundle smoke test so the engine's Avro wins over Spark's. + --> + <dependency> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.parquet</groupId> + <artifactId>parquet-avro</artifactId> + <scope>provided</scope> + </dependency> + <!-- 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> + + <!-- 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> + <excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j</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, protobuf), which are relocated to avoid clashes with the cluster. + --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedArtifactAttached>true</shadedArtifactAttached> + <shadedClassifierName>bundle</shadedClassifierName> + <createDependencyReducedPom>false</createDependencyReducedPom> + <artifactSet> + <includes> + <include>org.apache.xtable:xtable-api</include> Review Comment: Addressed in 4587fa77. The gate already exists: `release/scripts/validate_shaded_license_coverage.sh` (run in the license-check workflow) enforces that the shade `<includes>` **exactly equal** the runtime dependency tree and that every bundled dep is listed in `META-INF/LICENSE-bundled`. I made the module pass it: the transitives you flagged as "assumed cluster-provided" (jackson, plus scala-library and log4j-1.2-api) are now declared `provided` (Spark supplies them), which drops them from the runtime tree so they can no longer silently slip in/out. The per-dep assumptions are documented in the pom, and the script reports `OK xtable-spark-runtime`. ########## xtable-spark-runtime/pom.xml: ########## @@ -0,0 +1,311 @@ +<?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, drop-in Spark bundle that runs incremental XTable metadata sync in-job via a + driver-side QueryExecutionListener, activated through spark.xtable.* config only. + </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> + + <!-- 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 by the user's runtime, NOT bundled. + The user brings their own version via the cluster or the submit packages/jars flags, and + the thin XTable bundle stays compatible across engine versions. Declared provided so they + are on the compile/test classpath but excluded from the shaded jar. + --> + <!-- + Hudi is supplied by the cluster as the hudi-spark bundle (how Hudi ships for Spark). The + bundle carries a clean, regenerated DecimalWrapper; raw hudi-common's Avro-1.8/1.9-generated + model classes reference org.apache.avro.data.TimeConversions.TimeConversion, which Avro + removed in 1.10 and is therefore absent from the Avro 1.12 that Iceberg 1.9.2 requires. + Provided so it is on the compile/test/engine classpath but excluded from the shaded jar. + --> + <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 - raw hudi-common's DecimalWrapper is + Avro-1.8/1.9-generated and fails to link on the Avro >=1.10 required by Iceberg 1.9.2. + --> + <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 and parquet-avro pinned to the versions the engines require (Iceberg 1.9.2 compiles + against Avro 1.12 APIs), not the older Avro that Spark 3.5 ships. On a real Iceberg-1.9.2 + cluster the runtime must already supply these; declaring them provided puts them on the + flat engine classpath for the bundle smoke test so the engine's Avro wins over Spark's. + --> + <dependency> + <groupId>org.apache.avro</groupId> + <artifactId>avro</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.parquet</groupId> + <artifactId>parquet-avro</artifactId> + <scope>provided</scope> + </dependency> + <!-- 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> + + <!-- 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> + <excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j</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, protobuf), which are relocated to avoid clashes with the cluster. + --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <shadedArtifactAttached>true</shadedArtifactAttached> + <shadedClassifierName>bundle</shadedClassifierName> + <createDependencyReducedPom>false</createDependencyReducedPom> + <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.protobuf:protobuf-java</include> + <include>commons-cli:commons-cli</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>com.google.protobuf</pattern> + <shadedPattern>org.apache.xtable.shaded.com.google.protobuf</shadedPattern> + </relocation> + <relocation> + <pattern>org.apache.commons.cli</pattern> + <shadedPattern>org.apache.xtable.shaded.org.apache.commons.cli</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> + </filters> + <transformers> + <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> + <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheLicenseResourceTransformer"/> Review Comment: Addressed in 4587fa77. Two parts: (1) **protobuf is not actually bundled** — it resolves as `provided` (via the engines), so it was never in the jar; I removed the dead `protobuf-java` allowlist entry and relocation that made it look shipped. (2) Added `META-INF/LICENSE-bundled` + `NOTICE-bundled` (via `IncludeResourceTransformer`, matching the other shaded modules) attributing the real bundled third-party: guava's closure + commons-cli (Apache-2.0) and checker-qual (MIT). `validate_shaded_license_coverage.sh` verifies coverage and license-family classification (all Category A). ########## xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java: ########## @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.xtable.spark; + +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + +import lombok.extern.log4j.Log4j2; + +import org.apache.hadoop.conf.Configuration; + +import org.apache.xtable.conversion.ConversionConfig; +import org.apache.xtable.conversion.ConversionController; +import org.apache.xtable.conversion.ConversionSourceProvider; +import org.apache.xtable.conversion.SourceTable; +import org.apache.xtable.conversion.TargetTable; +import org.apache.xtable.delta.DeltaConversionSourceProvider; +import org.apache.xtable.hudi.HudiConversionSourceProvider; +import org.apache.xtable.hudi.HudiSourceConfig; +import org.apache.xtable.iceberg.IcebergConversionSourceProvider; +import org.apache.xtable.model.storage.TableFormat; +import org.apache.xtable.model.sync.SyncMode; +import org.apache.xtable.model.sync.SyncResult; + +/** + * Builds a {@link ConversionConfig} from a {@link TableSyncSpec} and runs an incremental {@link + * ConversionController#sync} for it. This is the unit of work triggered by {@link + * XTableSyncListener} after a successful write. + * + * <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(); + + List<TargetTable> targetTables = + spec.getTargets().stream() + .map( + targetFormat -> + TargetTable.builder() + .name(spec.getKey()) + .basePath(dataPath) + .namespace(spec.getNamespace()) + .formatName(targetFormat) + .build()) + .collect(Collectors.toList()); + + ConversionConfig conversionConfig = + ConversionConfig.builder() + .sourceTable(sourceTable) + .targetTables(targetTables) + .syncMode(SyncMode.INCREMENTAL) + .build(); + + ConversionSourceProvider<?> sourceProvider = sourceProviderFor(spec.getSourceFormat()); + sourceProvider.init(hadoopConf); + + log.info( + "Running XTable sync for table {} ({} -> {}) at {}", + spec.getKey(), + spec.getSourceFormat(), + spec.getTargets(), + spec.getBasePath()); + return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider); + } + + private static ConversionSourceProvider<?> sourceProviderFor(String sourceFormat) { + switch (sourceFormat.toUpperCase()) { Review Comment: Fixed in 4587fa77. `sourceFormat` is now normalized with `Locale.ROOT`, consistent with `XTableSparkSync`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
