You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by "vinishjail97 (via GitHub)" <gi...@apache.org> on 2026/10/08 21:09:45 UTC
[PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
vinishjail97 opened a new pull request, #964:
URL: https://github.com/apache/incubator-xtable/pull/964
## What is the purpose of the pull request
Adds the `xtable-spark-runtime` module to `main`: a thin, drop-in Spark bundle that runs XTable metadata sync, validated end to end with `spark-submit`.
The same module was reviewed and merged on `branch-0.4` in #843. This PR brings that version to `main` on Hudi 1.2.1 and replaces #838, which had drifted too far from `main`. The RFC lives in its own PR, #960.
## Brief change log
- New module `xtable-spark-runtime_${scala.binary.version}`, a thin shaded bundle (~4.5 MB).
- `XTableSparkSync`, a `spark-submit` entry point, the equivalent of `RunSync` for this bundle:
- single-table mode (`--basepath`, `--sourceformat`, `--targets`, ...) or `--datasetconfig <yaml>` for several tables, on a local or cloud path;
- `--usedeltakernel` for the Spark-free Delta Kernel, enabled automatically on Spark 3.5+;
- Hudi, Iceberg, Delta, Paimon and Parquet sources; Hudi, Iceberg and Delta targets;
- formats are validated before Spark starts, and one failing table does not stop the rest.
- `TableSyncSpec` / `XTableSyncService`, the sync building blocks for use inside a Spark job.
- `ConversionTargetFactory` skips registered targets whose engine is not on the classpath, so a runtime with only some engines still works. `hasNext()` is inside the guard too, because it loads providers lazily.
- `LICENSE-bundled` / `NOTICE-bundled` for the bundled guava, commons-cli and snakeyaml.
- `.github/workflows/spark-runtime-validation.yml` runs the bundle IT on Spark 3.4.3 and 3.5.9.
The Java sources are the same as on `branch-0.4`. The differences from #843 are:
- The Hudi 1.2.1 version and the `HoodieCleanStat` builder already landed on `main` with #835.
- The module's shade `<filters>` use `combine.self="override"`. Without it, Maven merges the parent's `software.amazon.awssdk:bundle` filter into the snakeyaml filter by position and drops every snakeyaml class.
- The shade plugin writes `dependency-reduced-pom.xml` next to the module pom, so it is now in `.gitignore`, and its spotless exclude is in the POM format block.
- Shorter comments.
### Packaging model
Engines are not bundled. Spark, Hadoop and Hudi/Iceberg/Delta are `provided`, so users bring their own engine versions.
- The shaded jar is the main artifact with a dependency-reduced POM, the same shape as `iceberg-spark-runtime` and `hudi-spark-bundle`.
- The shade allowlist is XTable plus guava, commons-cli and snakeyaml, relocated under `org.apache.xtable.shaded.*`.
- Libraries XTable exchanges with the engines (`avro`, `parquet`, `jackson`) are not bundled or relocated, because a relocated Avro breaks the boundary with Hudi.
## Verify this pull request
- `XTableSparkSyncTest`: 45 unit tests for option parsing, format validation and the dataset YAML.
- `ITXTableSparkRuntimeBundle` (skipped unless `SPARK_HOME` is set) `spark-submit`s a real sync for one case per direction across Hudi, Iceberg and Delta, and checks the target data against the source. It passes 4/4 locally on Spark 3.4.2 and 3.5.2:
```
SPARK_LOCAL_IP=127.0.0.1 SPARK_HOME=<spark> ./mvnw -pl xtable-spark-runtime clean verify
```
- `install -DskipITs` for `xtable-core` and this module, `spotless:check` and `validate_shaded_license_coverage.sh` for this module pass.
- Also ran the bundle as Spark jobs on Kubernetes against S3, on Spark 3.4.2 and on Spark 3.5.2 with the Hudi 3.5 bundle and `delta-spark` 3.2.1. Hudi to Iceberg/Delta covered three TPC-DS-style tables up to 6.87M rows in 2,180 partitioned files, and Iceberg to Hudi/Delta covered one table. For every table, the target metadata lists the same files, sizes and record counts as the source, and reading each format back gives the same row count and content hash.
Part of #836.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4251848295
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
Review Comment:
Updated in f7fb5d70: Iceberg and Delta targets now go to the base path. Hudi stays on the data path, since it derives partitions relative to its base path and an Iceberg `data/` directory would otherwise show up in the partition values (`data/level=ERROR`); the bundle IT now also covers Iceberg to Delta.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4238792218
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .additionalProperties(targetProperties)
+ .build();
+ })
+ .collect(Collectors.toList());
+
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+
+ ConversionSourceProvider<?> sourceProvider =
+ sourceProviderFor(spec.getSourceFormat(), spec.isUseDeltaKernel());
+ sourceProvider.init(hadoopConf);
+
+ log.info(
+ "Running XTable sync for table {} ({} -> {}) at {}",
+ spec.getKey(),
+ spec.getSourceFormat(),
+ spec.getTargets(),
+ spec.getBasePath());
+ return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider);
+ }
+
+ private static ConversionSourceProvider<?> sourceProviderFor(
Review Comment:
I'm guessing something similar exists in our RunSync code, is there any easy way to share this setup code?
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,25 @@ public ConversionTarget createConversionTargetForName(
TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
&& DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
- for (ConversionTarget target : loader) {
+ Iterator<ConversionTarget> iterator = loader.iterator();
+ while (true) {
+ ConversionTarget target;
+ try {
+ // hasNext() loads provider classes lazily, so it can throw too.
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
+ // The target's engine is not on the classpath. Skip it; if it is the requested format,
+ // the NotSupportedException below reports it.
+ log.warn(
+ "Skipping a ConversionTarget whose engine is not on the classpath ({}: {})",
+ error.getClass().getName(),
+ error.getMessage(),
+ error);
Review Comment:
I think we can remove the `error.getMessage`, that will be duplicated in the `error` right?
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,86 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+name: Spark Runtime Bundle Validation
+
+# Runs ITXTableSparkRuntimeBundle, which spark-submits the shaded bundle. The IT skips itself
+# unless SPARK_HOME is set, so it only runs here.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# No path filter, so the check always reports and can be a required status check.
+on:
+ push:
+ branches:
+ - "main"
+
+ pull_request:
+ branches:
+ - "main"
+
+ workflow_dispatch:
+
+env:
+ SPARK_HADOOP_PROFILE: "hadoop3"
+
+jobs:
+ validate-spark-runtime-bundle:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ # 3.4: Delta via delta-core. 3.5: Delta auto-switches to the Spark-free Delta Kernel.
+ spark_version: ["3.4.3", "3.5.9"]
+ env:
+ SPARK_VERSION: ${{ matrix.spark_version }}
+ steps:
+ - uses: actions/checkout@v6
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v5
+ with:
+ java-version: '11'
+ distribution: 'temurin'
+ cache: maven
+
+ - name: Cache Spark distribution
Review Comment:
Should we also pull the Hudi 3.X bundle based on the spark_version provided? If so, same question applies to other formats.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
Review Comment:
Same here, is an immutable set more fitting?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
Review Comment:
Don't we want to target the base path here?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
Review Comment:
Should we make this an immutable set?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .desc("The base path of the source table (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .desc(
+ "Comma-separated target formats to sync to, e.g. ICEBERG,DELTA "
+ + "(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATASET_CONFIG)
+ .hasArg()
+ .argName("path")
+ .desc(
+ "Path to a YAML dataset config (sourceFormat, targetFormats, datasets[]) to "
+ + "sync multiple tables in one run; may be a local or cloud (s3/gcs/abfs) "
+ + "path. Mutually exclusive with --basepath/--sourceformat/--targets.")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATA_PATH)
+ .hasArg()
+ .desc("The path of the data files if different from the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TABLE_NAME)
+ .hasArg()
+ .desc("The table name; defaults to the last segment of the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(NAMESPACE)
+ .hasArg()
+ .desc("The dot-separated table namespace")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(PARTITION_SPEC)
+ .hasArg()
+ .desc("The Hudi source partition field spec, e.g. level:VALUE")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(USE_DELTA_KERNEL)
+ .desc(
+ "Use the Spark-free Delta Kernel implementation for the Delta source and/or "
+ + "target instead of the default delta-core path. When omitted this is "
+ + "auto-enabled on Spark 3.5+ (where the bundled delta-core does not "
+ + "run); pass it explicitly to force Kernel on any Spark version.")
+ .build())
+ .addOption(Option.builder().longOpt(HELP).desc("Displays help information").build());
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ throw new IllegalArgumentException("Failed to parse arguments", e);
+ }
+ if (cmd.hasOption(HELP)) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ return;
+ }
+
+ boolean datasetMode = cmd.hasOption(DATASET_CONFIG);
+ if (datasetMode
+ && (cmd.hasOption(BASE_PATH) || cmd.hasOption(SOURCE_FORMAT) || cmd.hasOption(TARGETS))) {
+ throw new IllegalArgumentException(
+ "--datasetconfig is mutually exclusive with --basepath/--sourceformat/--targets");
+ }
+ // Validate single-table flags before starting Spark. A --datasetconfig is validated after,
+ // because it is read through the Spark Hadoop config.
+ if (!datasetMode) {
+ requireOption(cmd, BASE_PATH);
+ requireOption(cmd, SOURCE_FORMAT);
+ requireOption(cmd, TARGETS);
+ validateFormat(
+ SOURCE_FORMAT,
+ cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT),
+ SUPPORTED_SOURCE_FORMATS);
+ parseTargets(cmd.getOptionValue(TARGETS)); // validates formats and non-emptiness
+ }
+
+ boolean useDeltaKernelFlag = cmd.hasOption(USE_DELTA_KERNEL);
+
+ SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+
+ String sourceFormat;
+ List<String> targetFormats;
+ List<TableSyncSpec.TableSyncSpecBuilder> builders = new ArrayList<>();
+ if (datasetMode) {
+ DatasetConfig config = readDatasetConfig(cmd.getOptionValue(DATASET_CONFIG), hadoopConf);
+ sourceFormat = config.sourceFormat;
+ targetFormats = config.targetFormats;
+ for (DatasetConfig.Table table : config.datasets) {
+ builders.add(
+ TableSyncSpec.builder()
+ .key(
+ table.tableName != null
+ ? table.tableName
+ : basePathToTableName(table.tableBasePath))
+ .basePath(table.tableBasePath)
+ .dataPath(table.tableDataPath)
+ .namespace(table.namespace == null ? null : table.namespace.split("\\."))
+ .partitionSpec(table.partitionSpec)
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+ } else {
+ sourceFormat = cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT);
+ targetFormats = parseTargets(cmd.getOptionValue(TARGETS));
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String namespace = cmd.getOptionValue(NAMESPACE);
+ builders.add(
+ TableSyncSpec.builder()
+ .key(cmd.getOptionValue(TABLE_NAME, basePathToTableName(basePath)))
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+
+ // delta-core does not run on Spark 3.5+, so use Delta Kernel there.
+ boolean deltaInvolved =
+ TableFormat.DELTA.equals(sourceFormat) || targetFormats.contains(TableFormat.DELTA);
+ boolean useDeltaKernel = useDeltaKernelFlag;
+ if (!useDeltaKernel && deltaInvolved && isSparkAtLeast35(spark.version())) {
+ useDeltaKernel = true;
+ log.info(
+ "Spark {} detected (>= 3.5); auto-enabling the Delta Kernel implementation for Delta "
+ + "source/target. Pass --usedeltakernel to force it on any Spark version.",
+ spark.version());
+ }
+
+ XTableSyncService service = new XTableSyncService();
+ List<TableSyncSpec> specs = new ArrayList<>();
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ specs.add(builder.useDeltaKernel(useDeltaKernel).build());
+ }
+ int failures = syncAll(specs, spec -> service.sync(spec, hadoopConf));
+ if (failures > 0) {
+ throw new IllegalStateException(
+ failures + " of " + specs.size() + " table sync(s) failed; see logs above");
+ }
+ } finally {
+ spark.stop();
+ }
+ }
+
+ /**
+ * Syncs every spec and returns how many failed. A table fails when its sync throws or when any
+ * target reports a non-SUCCESS status; one failure does not abort the rest of the batch.
+ */
+ // package-private for unit testing
+ static int syncAll(
+ List<TableSyncSpec> specs, Function<TableSyncSpec, Map<String, SyncResult>> syncer) {
+ int failures = 0;
+ for (TableSyncSpec spec : specs) {
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(), spec.getBasePath());
+ // TableFormatSync reports a target failure as an ERROR status rather than throwing.
+ Map<String, String> failedTargets =
+ syncer.apply(spec).entrySet().stream()
+ .filter(
+ entry ->
+ entry.getValue().getTableFormatSyncStatus().getStatusCode()
+ != SyncStatusCode.SUCCESS)
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry -> describe(entry.getValue().getTableFormatSyncStatus()),
+ (a, b) -> a,
+ TreeMap::new));
+ if (failedTargets.isEmpty()) {
+ log.info("Completed XTable sync for {}", spec.getKey());
+ } else {
+ failures++;
+ log.error(
+ "XTable sync failed for {} ({}) on target(s) {}",
+ spec.getKey(),
+ spec.getBasePath(),
+ failedTargets);
+ }
+ } catch (RuntimeException e) {
+ failures++;
+ log.error("XTable sync failed for {} ({})", spec.getKey(), spec.getBasePath(), e);
+ }
+ }
+ return failures;
+ }
+
+ private static String describe(SyncResult.SyncStatus status) {
+ ErrorDetails details = status.getErrorDetails();
+ return details == null || details.getErrorMessage() == null
+ ? String.valueOf(status.getStatusCode())
+ : status.getStatusCode() + ": " + details.getErrorMessage();
+ }
+
+ private static void requireOption(CommandLine cmd, String option) {
+ if (!cmd.hasOption(option)) {
+ throw new IllegalArgumentException(
+ "--" + option + " is required unless --" + DATASET_CONFIG + " is provided");
+ }
+ }
+
+ // package-private for unit testing
+ static List<String> parseTargets(String targetsValue) {
+ List<String> targetFormats =
+ Arrays.stream(targetsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (targetFormats.isEmpty()) {
+ throw new IllegalArgumentException(
+ "--targets resolved to an empty list; provide at least one target format, e.g. ICEBERG,DELTA");
+ }
+ targetFormats.forEach(target -> validateFormat(TARGETS, target, SUPPORTED_TARGET_FORMATS));
+ return targetFormats;
+ }
+
+ static DatasetConfig readDatasetConfig(String path, Configuration hadoopConf) {
+ Path configPath = new Path(path);
+ try (InputStream in = configPath.getFileSystem(hadoopConf).open(configPath)) {
+ return parseDatasetConfig(in);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to read --datasetconfig at " + path, e);
+ }
+ }
+
+ // SafeConstructor builds only maps, lists and scalars. package-private for unit testing.
+ @SuppressWarnings("unchecked")
+ static DatasetConfig parseDatasetConfig(InputStream in) {
Review Comment:
Why do we need the custom logic in this instead of trying to read the yaml into the DatasetConfig directly with jackson object mapper?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4224719319
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,425 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .desc("The base path of the source table (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .desc(
+ "Comma-separated target formats to sync to, e.g. ICEBERG,DELTA "
+ + "(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATASET_CONFIG)
+ .hasArg()
+ .argName("path")
+ .desc(
+ "Path to a YAML dataset config (sourceFormat, targetFormats, datasets[]) to "
+ + "sync multiple tables in one run; may be a local or cloud (s3/gcs/abfs) "
+ + "path. Mutually exclusive with --basepath/--sourceformat/--targets.")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATA_PATH)
+ .hasArg()
+ .desc("The path of the data files if different from the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TABLE_NAME)
+ .hasArg()
+ .desc("The table name; defaults to the last segment of the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(NAMESPACE)
+ .hasArg()
+ .desc("The dot-separated table namespace")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(PARTITION_SPEC)
+ .hasArg()
+ .desc("The Hudi source partition field spec, e.g. level:VALUE")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(USE_DELTA_KERNEL)
+ .desc(
+ "Use the Spark-free Delta Kernel implementation for the Delta source and/or "
+ + "target instead of the default delta-core path. When omitted this is "
+ + "auto-enabled on Spark 3.5+ (where the bundled delta-core does not "
+ + "run); pass it explicitly to force Kernel on any Spark version.")
+ .build())
+ .addOption(Option.builder().longOpt(HELP).desc("Displays help information").build());
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ throw new IllegalArgumentException("Failed to parse arguments", e);
+ }
+ if (cmd.hasOption(HELP)) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ return;
+ }
+
+ boolean datasetMode = cmd.hasOption(DATASET_CONFIG);
+ if (datasetMode
+ && (cmd.hasOption(BASE_PATH) || cmd.hasOption(SOURCE_FORMAT) || cmd.hasOption(TARGETS))) {
+ throw new IllegalArgumentException(
+ "--datasetconfig is mutually exclusive with --basepath/--sourceformat/--targets");
+ }
+ // Validate single-table flags before starting Spark. A --datasetconfig is validated after,
+ // because it is read through the Spark Hadoop config.
+ if (!datasetMode) {
+ requireOption(cmd, BASE_PATH);
+ requireOption(cmd, SOURCE_FORMAT);
+ requireOption(cmd, TARGETS);
+ validateFormat(
+ SOURCE_FORMAT,
+ cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT),
+ SUPPORTED_SOURCE_FORMATS);
+ parseTargets(cmd.getOptionValue(TARGETS)); // validates formats and non-emptiness
+ }
+
+ boolean useDeltaKernelFlag = cmd.hasOption(USE_DELTA_KERNEL);
+
+ SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+
+ String sourceFormat;
+ List<String> targetFormats;
+ List<TableSyncSpec.TableSyncSpecBuilder> builders = new ArrayList<>();
+ if (datasetMode) {
+ DatasetConfig config = readDatasetConfig(cmd.getOptionValue(DATASET_CONFIG), hadoopConf);
+ sourceFormat = config.sourceFormat;
+ targetFormats = config.targetFormats;
+ for (DatasetConfig.Table table : config.datasets) {
+ builders.add(
+ TableSyncSpec.builder()
+ .key(
+ table.tableName != null
+ ? table.tableName
+ : basePathToTableName(table.tableBasePath))
+ .basePath(table.tableBasePath)
+ .dataPath(table.tableDataPath)
+ .namespace(table.namespace == null ? null : table.namespace.split("\\."))
+ .partitionSpec(table.partitionSpec)
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+ } else {
+ sourceFormat = cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT);
+ targetFormats = parseTargets(cmd.getOptionValue(TARGETS));
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String namespace = cmd.getOptionValue(NAMESPACE);
+ builders.add(
+ TableSyncSpec.builder()
+ .key(cmd.getOptionValue(TABLE_NAME, basePathToTableName(basePath)))
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+
+ // delta-core does not run on Spark 3.5+, so use Delta Kernel there.
+ boolean deltaInvolved =
+ TableFormat.DELTA.equals(sourceFormat) || targetFormats.contains(TableFormat.DELTA);
+ boolean useDeltaKernel = useDeltaKernelFlag;
+ if (!useDeltaKernel && deltaInvolved && isSparkAtLeast35(spark.version())) {
+ useDeltaKernel = true;
+ log.info(
+ "Spark {} detected (>= 3.5); auto-enabling the Delta Kernel implementation for Delta "
+ + "source/target. Pass --usedeltakernel to force it on any Spark version.",
+ spark.version());
+ }
+
+ XTableSyncService service = new XTableSyncService();
+ int failures = 0;
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ TableSyncSpec spec = builder.useDeltaKernel(useDeltaKernel).build();
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(), spec.getBasePath());
+ service.sync(spec, hadoopConf);
Review Comment:
Fixed in 013dd2ab. The batch loop is now `syncAll`, which counts a table as failed when the sync throws or any target returns a non-SUCCESS status, keeps going, and fails at the end; `syncAllCountsReturnedErrorsAndThrownFailuresAndContinuesBatch` covers the mixed case.
##########
xtable-spark-runtime/pom.xml:
##########
@@ -0,0 +1,325 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable</artifactId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>Thin, relocated Spark bundle that runs XTable metadata sync.</description>
+
+ <dependencies>
+ <!-- Bundled -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>2.2</version>
+ </dependency>
+
+
+ <!-- Provided by the Spark runtime -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+
+ <!-- Engines (Hudi, Iceberg, Delta) are provided by the user's runtime, not bundled -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <!-- The Hudi target uses the Java client; hudi-common comes from the hudi-spark bundle -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-java-client</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-core_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-defaults</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Iceberg 1.9.2 needs Avro 1.12, newer than Spark 3.4's -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.openjdk.jol</groupId>
+ <artifactId>jol-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Transitive via xtable-core but shipped by Spark; provided keeps them out of the bundle -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test Dependencies -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Engine classpath for ITXTableSparkRuntimeBundle, passed as extraClassPath -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>engine-classpath</id>
+ <phase>pre-integration-test</phase>
+ <goals>
+ <goal>build-classpath</goal>
+ </goals>
+ <configuration>
+ <includeScope>provided</includeScope>
+ <excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j,org.scala-lang,com.fasterxml.jackson.core,com.fasterxml.jackson.datatype,com.fasterxml.jackson.module</excludeGroupIds>
+ <outputFile>${project.build.directory}/engine-classpath.txt</outputFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--
+ The shaded jar is the main artifact, like hudi-spark-bundle and iceberg-spark-runtime.
+ It bundles only XTable and the libraries XTable uses internally, relocated. Libraries
+ XTable exchanges with the engines (avro, parquet, jackson) stay provided.
+ -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadedArtifactAttached>false</shadedArtifactAttached>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <!-- Must match the runtime dependency tree (validate_shaded_license_coverage.sh) -->
+ <artifactSet>
+ <includes>
+ <include>org.apache.xtable:xtable-api</include>
+ <include>org.apache.xtable:xtable-core_${scala.binary.version}</include>
+ <include>org.apache.xtable:xtable-hudi-support-utils</include>
+ <include>com.google.guava:guava</include>
+ <include>com.google.guava:failureaccess</include>
+ <include>com.google.guava:listenablefuture</include>
+ <include>com.google.errorprone:error_prone_annotations</include>
+ <include>com.google.j2objc:j2objc-annotations</include>
+ <include>org.checkerframework:checker-qual</include>
+ <include>commons-cli:commons-cli</include>
+ <include>org.yaml:snakeyaml</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>com.google.common</pattern>
+ <shadedPattern>org.apache.xtable.shaded.com.google.common</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google.thirdparty</pattern>
+ <shadedPattern>org.apache.xtable.shaded.com.google.thirdparty</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.commons.cli</pattern>
+ <shadedPattern>org.apache.xtable.shaded.org.apache.commons.cli</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.yaml.snakeyaml</pattern>
+ <shadedPattern>org.apache.xtable.shaded.org.yaml.snakeyaml</shadedPattern>
+ </relocation>
+ </relocations>
+ <!-- Override: a positional merge with the parent's awssdk filter drops snakeyaml -->
+ <filters combine.self="override">
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>**/Log4j2Plugins.dat</exclude>
+ </excludes>
+ </filter>
+ <!-- Shade does not relocate multi-release classes -->
+ <filter>
+ <artifact>org.yaml:snakeyaml</artifact>
+ <excludes>
+ <exclude>META-INF/versions/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheLicenseResourceTransformer"/>
Review Comment:
Fixed in 013dd2ab. Added `META-INF/licenses/LICENSE-checker-qual`, and License Check now builds `xtable-spark-runtime` and `validate_bundled_license_texts.py` picks up its main jar.
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,25 @@ public ConversionTarget createConversionTargetForName(
TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
&& DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
- for (ConversionTarget target : loader) {
+ Iterator<ConversionTarget> iterator = loader.iterator();
+ while (true) {
+ ConversionTarget target;
+ try {
+ // hasNext() loads provider classes lazily, so it can throw too.
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
Review Comment:
Added in 013dd2ab. Two tests run the factory under a classloader with a missing provider class, a Delta target with a hidden superclass (fails in `hasNext()`), and one with a hidden constructor dependency (fails in `next()`): Iceberg is still found, and DELTA throws `NotSupportedException` within a timeout.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4251454142
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
Review Comment:
This matches `ITConversionController` and `ConversionService`, which also target the source data path. For an Iceberg source with a `data/` directory, a table-root target would put `data/` into every Hudi partition path, so `RunSync` looks like the outlier here.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4224233540
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,25 @@ public ConversionTarget createConversionTargetForName(
TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
&& DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
- for (ConversionTarget target : loader) {
+ Iterator<ConversionTarget> iterator = loader.iterator();
+ while (true) {
+ ConversionTarget target;
+ try {
+ // hasNext() loads provider classes lazily, so it can throw too.
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
Review Comment:
[P2] Test discovery with missing engines
Current tests supply all engines, so they miss this recovery path. An isolated classloader probe passed. Add repository tests for available-target discovery and bounded failure when the requested engine is absent.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,425 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .desc("The base path of the source table (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .desc(
+ "Comma-separated target formats to sync to, e.g. ICEBERG,DELTA "
+ + "(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATASET_CONFIG)
+ .hasArg()
+ .argName("path")
+ .desc(
+ "Path to a YAML dataset config (sourceFormat, targetFormats, datasets[]) to "
+ + "sync multiple tables in one run; may be a local or cloud (s3/gcs/abfs) "
+ + "path. Mutually exclusive with --basepath/--sourceformat/--targets.")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATA_PATH)
+ .hasArg()
+ .desc("The path of the data files if different from the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TABLE_NAME)
+ .hasArg()
+ .desc("The table name; defaults to the last segment of the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(NAMESPACE)
+ .hasArg()
+ .desc("The dot-separated table namespace")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(PARTITION_SPEC)
+ .hasArg()
+ .desc("The Hudi source partition field spec, e.g. level:VALUE")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(USE_DELTA_KERNEL)
+ .desc(
+ "Use the Spark-free Delta Kernel implementation for the Delta source and/or "
+ + "target instead of the default delta-core path. When omitted this is "
+ + "auto-enabled on Spark 3.5+ (where the bundled delta-core does not "
+ + "run); pass it explicitly to force Kernel on any Spark version.")
+ .build())
+ .addOption(Option.builder().longOpt(HELP).desc("Displays help information").build());
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ throw new IllegalArgumentException("Failed to parse arguments", e);
+ }
+ if (cmd.hasOption(HELP)) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ return;
+ }
+
+ boolean datasetMode = cmd.hasOption(DATASET_CONFIG);
+ if (datasetMode
+ && (cmd.hasOption(BASE_PATH) || cmd.hasOption(SOURCE_FORMAT) || cmd.hasOption(TARGETS))) {
+ throw new IllegalArgumentException(
+ "--datasetconfig is mutually exclusive with --basepath/--sourceformat/--targets");
+ }
+ // Validate single-table flags before starting Spark. A --datasetconfig is validated after,
+ // because it is read through the Spark Hadoop config.
+ if (!datasetMode) {
+ requireOption(cmd, BASE_PATH);
+ requireOption(cmd, SOURCE_FORMAT);
+ requireOption(cmd, TARGETS);
+ validateFormat(
+ SOURCE_FORMAT,
+ cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT),
+ SUPPORTED_SOURCE_FORMATS);
+ parseTargets(cmd.getOptionValue(TARGETS)); // validates formats and non-emptiness
+ }
+
+ boolean useDeltaKernelFlag = cmd.hasOption(USE_DELTA_KERNEL);
+
+ SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+
+ String sourceFormat;
+ List<String> targetFormats;
+ List<TableSyncSpec.TableSyncSpecBuilder> builders = new ArrayList<>();
+ if (datasetMode) {
+ DatasetConfig config = readDatasetConfig(cmd.getOptionValue(DATASET_CONFIG), hadoopConf);
+ sourceFormat = config.sourceFormat;
+ targetFormats = config.targetFormats;
+ for (DatasetConfig.Table table : config.datasets) {
+ builders.add(
+ TableSyncSpec.builder()
+ .key(
+ table.tableName != null
+ ? table.tableName
+ : basePathToTableName(table.tableBasePath))
+ .basePath(table.tableBasePath)
+ .dataPath(table.tableDataPath)
+ .namespace(table.namespace == null ? null : table.namespace.split("\\."))
+ .partitionSpec(table.partitionSpec)
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+ } else {
+ sourceFormat = cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT);
+ targetFormats = parseTargets(cmd.getOptionValue(TARGETS));
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String namespace = cmd.getOptionValue(NAMESPACE);
+ builders.add(
+ TableSyncSpec.builder()
+ .key(cmd.getOptionValue(TABLE_NAME, basePathToTableName(basePath)))
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+
+ // delta-core does not run on Spark 3.5+, so use Delta Kernel there.
+ boolean deltaInvolved =
+ TableFormat.DELTA.equals(sourceFormat) || targetFormats.contains(TableFormat.DELTA);
+ boolean useDeltaKernel = useDeltaKernelFlag;
+ if (!useDeltaKernel && deltaInvolved && isSparkAtLeast35(spark.version())) {
+ useDeltaKernel = true;
+ log.info(
+ "Spark {} detected (>= 3.5); auto-enabling the Delta Kernel implementation for Delta "
+ + "source/target. Pass --usedeltakernel to force it on any Spark version.",
+ spark.version());
+ }
+
+ XTableSyncService service = new XTableSyncService();
+ int failures = 0;
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ TableSyncSpec spec = builder.useDeltaKernel(useDeltaKernel).build();
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(), spec.getBasePath());
+ service.sync(spec, hadoopConf);
Review Comment:
[P1] Handle returned sync errors
This call discards the SyncResult map. TableFormatSync returns ERROR for target failures without throwing, so the CLI can report success. Count returned errors, continue the batch, and fail at the end. Add a mixed-result regression test.
##########
xtable-spark-runtime/pom.xml:
##########
@@ -0,0 +1,325 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable</artifactId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>Thin, relocated Spark bundle that runs XTable metadata sync.</description>
+
+ <dependencies>
+ <!-- Bundled -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.yaml</groupId>
+ <artifactId>snakeyaml</artifactId>
+ <version>2.2</version>
+ </dependency>
+
+
+ <!-- Provided by the Spark runtime -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+
+ <!-- Engines (Hudi, Iceberg, Delta) are provided by the user's runtime, not bundled -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <!-- The Hudi target uses the Java client; hudi-common comes from the hudi-spark bundle -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-java-client</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-common</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-core_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-kernel-defaults</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Iceberg 1.9.2 needs Avro 1.12, newer than Spark 3.4's -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.openjdk.jol</groupId>
+ <artifactId>jol-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Transitive via xtable-core but shipped by Spark; provided keeps them out of the bundle -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.module</groupId>
+ <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Test Dependencies -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-spark-runtime-${spark.version.prefix}_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-engine</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Engine classpath for ITXTableSparkRuntimeBundle, passed as extraClassPath -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>engine-classpath</id>
+ <phase>pre-integration-test</phase>
+ <goals>
+ <goal>build-classpath</goal>
+ </goals>
+ <configuration>
+ <includeScope>provided</includeScope>
+ <excludeGroupIds>org.apache.spark,org.apache.hadoop,org.apache.xtable,org.projectlombok,org.slf4j,org.apache.logging.log4j,org.scala-lang,com.fasterxml.jackson.core,com.fasterxml.jackson.datatype,com.fasterxml.jackson.module</excludeGroupIds>
+ <outputFile>${project.build.directory}/engine-classpath.txt</outputFile>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <!--
+ The shaded jar is the main artifact, like hudi-spark-bundle and iceberg-spark-runtime.
+ It bundles only XTable and the libraries XTable uses internally, relocated. Libraries
+ XTable exchanges with the engines (avro, parquet, jackson) stay provided.
+ -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <shadedArtifactAttached>false</shadedArtifactAttached>
+ <createDependencyReducedPom>true</createDependencyReducedPom>
+ <!-- Must match the runtime dependency tree (validate_shaded_license_coverage.sh) -->
+ <artifactSet>
+ <includes>
+ <include>org.apache.xtable:xtable-api</include>
+ <include>org.apache.xtable:xtable-core_${scala.binary.version}</include>
+ <include>org.apache.xtable:xtable-hudi-support-utils</include>
+ <include>com.google.guava:guava</include>
+ <include>com.google.guava:failureaccess</include>
+ <include>com.google.guava:listenablefuture</include>
+ <include>com.google.errorprone:error_prone_annotations</include>
+ <include>com.google.j2objc:j2objc-annotations</include>
+ <include>org.checkerframework:checker-qual</include>
+ <include>commons-cli:commons-cli</include>
+ <include>org.yaml:snakeyaml</include>
+ </includes>
+ </artifactSet>
+ <relocations>
+ <relocation>
+ <pattern>com.google.common</pattern>
+ <shadedPattern>org.apache.xtable.shaded.com.google.common</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google.thirdparty</pattern>
+ <shadedPattern>org.apache.xtable.shaded.com.google.thirdparty</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.apache.commons.cli</pattern>
+ <shadedPattern>org.apache.xtable.shaded.org.apache.commons.cli</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>org.yaml.snakeyaml</pattern>
+ <shadedPattern>org.apache.xtable.shaded.org.yaml.snakeyaml</shadedPattern>
+ </relocation>
+ </relocations>
+ <!-- Override: a positional merge with the parent's awssdk filter drops snakeyaml -->
+ <filters combine.self="override">
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ <exclude>**/Log4j2Plugins.dat</exclude>
+ </excludes>
+ </filter>
+ <!-- Shade does not relocate multi-release classes -->
+ <filter>
+ <artifact>org.yaml:snakeyaml</artifact>
+ <excludes>
+ <exclude>META-INF/versions/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheLicenseResourceTransformer"/>
Review Comment:
[P1] Include checker-qual license text
The built JAR contains checker-qual classes but omits its MIT copyright and permission text. validate_bundled_license_texts.py confirms this. Include META-INF/licenses/LICENSE-checker-qual and validate the main artifact in CI.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 merged PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4251453626
##########
xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.hudi.client.HoodieReadClient;
+
+import org.apache.xtable.GenericTable;
+import org.apache.xtable.hudi.HudiTestUtil;
+
+/**
+ * Runs a sync with the shaded bundle through {@code $SPARK_HOME/bin/spark-submit}, one case per
+ * direction, and checks the target data equals the source. The engines come from {@code
+ * target/engine-classpath.txt}. Skipped when {@code SPARK_HOME} is not set.
+ */
+class ITXTableSparkRuntimeBundle {
+
+ @TempDir static java.nio.file.Path tempDir;
+
+ private static SparkSession sparkSession;
+ private static JavaSparkContext jsc;
+
+ @BeforeAll
+ static void setupSpark() {
+ sparkSession =
+ SparkSession.builder()
+ .config(HoodieReadClient.addHoodieSupport(HudiTestUtil.getSparkConf(tempDir)))
+ .getOrCreate();
+ jsc = JavaSparkContext.fromSparkContext(sparkSession.sparkContext());
+ }
+
+ @AfterAll
+ static void stopSpark() {
+ if (jsc != null) {
+ jsc.close();
+ }
+ if (sparkSession != null) {
+ sparkSession.stop();
+ }
+ }
+
+ private static Stream<Arguments> directions() {
+ // sourceFormat, targetFormat, hudiPartitionSpec
+ return Stream.of(
+ Arguments.of("HUDI", "ICEBERG", "level:VALUE"),
+ Arguments.of("ICEBERG", "HUDI", null),
+ Arguments.of("HUDI", "DELTA", "level:VALUE"),
+ Arguments.of("DELTA", "HUDI", null));
+ }
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource("directions")
+ void bundleRunsSyncViaSparkSubmit(String sourceFormat, String targetFormat, String partitionSpec)
+ throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME is not set");
+
+ String tableName = sourceFormat.toLowerCase() + "_to_" + targetFormat.toLowerCase();
+ String basePath;
+ String dataPath;
+ List<String> columns;
+ String orderByColumn;
+ try (GenericTable<?, ?> table =
+ GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, sourceFormat, true)) {
+ table.insertRows(100);
+ basePath = table.getBasePath();
+ dataPath = table.getDataPath();
+ columns = table.getColumnsToSelect();
+ orderByColumn = table.getOrderByColumn();
+ }
+
+ File bundleJar = findBundleJar();
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " + sparkSubmit);
+
+ // A flat classpath, not --packages: a child classloader breaks Avro casts across the boundary.
+ String engineClasspath = readEngineClasspath();
+
+ List<String> command = new ArrayList<>();
+ command.add(sparkSubmit.getAbsolutePath());
+ command.add("--master");
+ command.add("local[2]");
+ command.add("--conf");
+ command.add("spark.driver.extraClassPath=" + engineClasspath);
+ command.add("--conf");
+ command.add("spark.executor.extraClassPath=" + engineClasspath);
+ command.add("--class");
+ command.add("org.apache.xtable.spark.XTableSparkSync");
+ command.add(bundleJar.getAbsolutePath());
+ command.add("--basepath");
+ command.add(basePath);
+ command.add("--datapath");
+ command.add(dataPath);
+ command.add("--sourceformat");
+ command.add(sourceFormat);
+ command.add("--targets");
+ command.add(targetFormat);
+ command.add("--tablename");
+ command.add(tableName);
+ if (partitionSpec != null) {
+ command.add("--partitionspec");
+ command.add(partitionSpec);
+ }
+
+ ProcessBuilder pb = new ProcessBuilder(command);
+ pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1");
+ pb.redirectErrorStream(true);
+ Process process = pb.start();
+ // Drain on a thread so a hung spark-submit cannot block the timeout below.
+ StringBuilder outputBuffer = new StringBuilder();
+ Thread drainer = new Thread(() -> readOutputInto(process, outputBuffer), "spark-submit-stdout");
+ drainer.setDaemon(true);
+ drainer.start();
+ boolean finished = process.waitFor(10, TimeUnit.MINUTES);
+ if (!finished) {
+ process.destroyForcibly();
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ synchronized (outputBuffer) {
+ throw new AssertionError("spark-submit timed out. Output:\n" + outputBuffer);
+ }
+ }
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ String output;
+ synchronized (outputBuffer) {
+ output = outputBuffer.toString();
+ }
+ assertEquals(0, process.exitValue(), "spark-submit failed. Output:\n" + output);
+
+ assertDatasetEquivalence(
+ sourceFormat, targetFormat, basePath, dataPath, columns, orderByColumn, output);
+ }
+
+ /** Reads the source and the target back through Spark and asserts the scalar columns match. */
+ private void assertDatasetEquivalence(
+ String sourceFormat,
+ String targetFormat,
+ String basePath,
+ String dataPath,
+ List<String> columns,
+ String orderByColumn,
+ String submitOutput) {
+ Dataset<Row> source =
+ sparkSession
+ .read()
+ .options(readOptions(sourceFormat))
+ .format(sourceFormat.toLowerCase())
+ .load(basePath)
+ .orderBy(orderByColumn);
+ String[] projection = comparableColumns(source, columns).toArray(new String[0]);
+ List<String> sourceRows = source.selectExpr(projection).toJSON().collectAsList();
+ List<String> targetRows =
+ sparkSession
+ .read()
+ .options(readOptions(targetFormat))
+ .format(targetFormat.toLowerCase())
+ .load(dataPath)
+ .orderBy(orderByColumn)
+ .selectExpr(projection)
+ .toJSON()
+ .collectAsList();
+ assertEquals(
+ 100,
+ targetRows.size(),
+ "Unexpected target row count. spark-submit output:\n" + submitOutput);
+ assertEquals(sourceRows, targetRows, "Target is not data-equivalent to the source");
+ }
+
+ // Drops nested columns (covered by ITConversionController) and local timestamps (HUDI-7088).
+ private static List<String> comparableColumns(Dataset<Row> source, List<String> columns) {
+ java.util.Set<String> excluded = new java.util.HashSet<>();
+ for (org.apache.spark.sql.types.StructField field : source.schema().fields()) {
+ org.apache.spark.sql.types.DataType type = field.dataType();
+ if (type instanceof org.apache.spark.sql.types.StructType
+ || type instanceof org.apache.spark.sql.types.ArrayType
+ || type instanceof org.apache.spark.sql.types.MapType) {
+ excluded.add(field.name());
+ }
+ }
+ return columns.stream()
+ .filter(c -> !excluded.contains(c))
+ .filter(c -> !c.startsWith("timestamp_local_"))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ private static java.util.Map<String, String> readOptions(String format) {
+ java.util.Map<String, String> options = new java.util.HashMap<>();
+ if ("HUDI".equalsIgnoreCase(format)) {
+ options.put("hoodie.metadata.enable", "true");
+ options.put("hoodie.datasource.read.extract.partition.values.from.path", "true");
+ } else if ("ICEBERG".equalsIgnoreCase(format)) {
+ // The vectorized reader mis-casts timestamps written by Hudi.
+ options.put("vectorization-enabled", "false");
+ }
+ return options;
+ }
+
+ private String readEngineClasspath() throws Exception {
+ File cpFile = new File("target/engine-classpath.txt");
+ assertTrue(
+ cpFile.isFile(),
+ "engine classpath not built at "
+ + cpFile.getAbsolutePath()
+ + " (dependency:build-classpath)");
+ String cp =
+ new String(
+ java.nio.file.Files.readAllBytes(cpFile.toPath()),
+ java.nio.charset.StandardCharsets.UTF_8)
+ .trim();
+ assertTrue(!cp.isEmpty(), "engine classpath file is empty");
+ return cp;
+ }
+
+ // The shaded jar is the main artifact; skip original-* and the sources, javadoc and tests jars.
+ private File findBundleJar() {
+ File targetDir = new File("target");
+ File[] candidates =
+ targetDir.listFiles(
+ (dir, name) ->
+ name.startsWith("xtable-spark-runtime")
+ && name.endsWith(".jar")
+ && !name.startsWith("original-")
+ && !name.contains("-sources")
+ && !name.contains("-javadoc")
+ && !name.contains("-tests"));
+ assertTrue(
+ candidates != null && candidates.length == 1,
+ "Expected exactly one bundle jar in "
+ + targetDir.getAbsolutePath()
+ + " (run the package phase first)");
+ return candidates[0];
+ }
+
+ private static void readOutputInto(Process process, StringBuilder sb) {
+ try (java.io.BufferedReader reader =
+ new java.io.BufferedReader(
+ new java.io.InputStreamReader(
Review Comment:
Fixed in f99cf05a, the IT now imports these.
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,25 @@ public ConversionTarget createConversionTargetForName(
TableFormat.DELTA.equalsIgnoreCase(tableFormatName)
&& DeltaConversionTargetConfig.fromProperties(properties).isUseKernel();
ServiceLoader<ConversionTarget> loader = ServiceLoader.load(ConversionTarget.class);
- for (ConversionTarget target : loader) {
+ Iterator<ConversionTarget> iterator = loader.iterator();
+ while (true) {
+ ConversionTarget target;
+ try {
+ // hasNext() loads provider classes lazily, so it can throw too.
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
+ // The target's engine is not on the classpath. Skip it; if it is the requested format,
+ // the NotSupportedException below reports it.
+ log.warn(
+ "Skipping a ConversionTarget whose engine is not on the classpath ({}: {})",
+ error.getClass().getName(),
+ error.getMessage(),
+ error);
Review Comment:
Fixed in f99cf05a, the warning now logs only the error.
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,86 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+name: Spark Runtime Bundle Validation
+
+# Runs ITXTableSparkRuntimeBundle, which spark-submits the shaded bundle. The IT skips itself
+# unless SPARK_HOME is set, so it only runs here.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# No path filter, so the check always reports and can be a required status check.
+on:
+ push:
+ branches:
+ - "main"
+
+ pull_request:
+ branches:
+ - "main"
+
+ workflow_dispatch:
+
+env:
+ SPARK_HADOOP_PROFILE: "hadoop3"
+
+jobs:
+ validate-spark-runtime-bundle:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ # 3.4: Delta via delta-core. 3.5: Delta auto-switches to the Spark-free Delta Kernel.
+ spark_version: ["3.4.3", "3.5.9"]
+ env:
+ SPARK_VERSION: ${{ matrix.spark_version }}
+ steps:
+ - uses: actions/checkout@v6
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v5
+ with:
+ java-version: '11'
+ distribution: 'temurin'
+ cache: maven
+
+ - name: Cache Spark distribution
Review Comment:
The sync only uses `hudi-common` and `hudi-java-client` from the bundle, and Iceberg (`iceberg-core`) and Delta Kernel are Spark-free, so the 3.4 bundle covers the same code. Switching it per Spark version needs the test JVM to stay on 3.4 for writing the Delta source tables, so it would be good to take that as a follow-up.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
Review Comment:
This matches `ITConversionController` and `ConversionService`, which also target the source data path. For an Iceberg source with a `data/` directory, a table-root target would put `data/` into every Hudi partition path, so `RunSync` looks like the outlier here.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .additionalProperties(targetProperties)
+ .build();
+ })
+ .collect(Collectors.toList());
+
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+
+ ConversionSourceProvider<?> sourceProvider =
+ sourceProviderFor(spec.getSourceFormat(), spec.isUseDeltaKernel());
+ sourceProvider.init(hadoopConf);
+
+ log.info(
+ "Running XTable sync for table {} ({} -> {}) at {}",
+ spec.getKey(),
+ spec.getSourceFormat(),
+ spec.getTargets(),
+ spec.getBasePath());
+ return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider);
+ }
+
+ private static ConversionSourceProvider<?> sourceProviderFor(
Review Comment:
Sharing this first needs `RunSync` to agree on the target path, a source provider lookup with Kernel and Parquet, and the `SyncResult` map returned. It would be good to handle that as a follow-up that moves a small builder into `xtable-core`.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
Review Comment:
Fixed in f99cf05a, both sets are now `ImmutableSet`.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
Review Comment:
Fixed in f99cf05a, same change as above.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,471 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.yaml.snakeyaml.LoaderOptions;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.ErrorDetails;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+
+/**
+ * {@code spark-submit} entry point for the bundle, the equivalent of {@code RunSync}.
+ *
+ * <pre>
+ * $SPARK_HOME/bin/spark-submit \
+ * --class org.apache.xtable.spark.XTableSparkSync \
+ * xtable-spark-runtime_2.12-<ver>.jar \
+ * --basepath /warehouse/db/orders --sourceformat HUDI --targets ICEBERG,DELTA
+ * </pre>
+ */
+@Log4j2
+public final class XTableSparkSync {
+
+ private static final String BASE_PATH = "basepath";
+ private static final String DATA_PATH = "datapath";
+ private static final String SOURCE_FORMAT = "sourceformat";
+ private static final String TARGETS = "targets";
+ private static final String TABLE_NAME = "tablename";
+ private static final String NAMESPACE = "namespace";
+ private static final String PARTITION_SPEC = "partitionspec";
+ private static final String DATASET_CONFIG = "datasetconfig";
+ private static final String USE_DELTA_KERNEL = "usedeltakernel";
+ private static final String HELP = "help";
+
+ /**
+ * Formats usable as --sourceformat. Keep in sync with {@link
+ * XTableSyncService#sourceProviderFor}.
+ */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_SOURCE_FORMATS =
+ new LinkedHashSet<>(
+ Arrays.asList(
+ TableFormat.HUDI,
+ TableFormat.ICEBERG,
+ TableFormat.DELTA,
+ TableFormat.PAIMON,
+ TableFormat.PARQUET));
+
+ /** Formats usable as --targets. Paimon and Parquet are source-only. */
+ // package-private for unit testing
+ static final Set<String> SUPPORTED_TARGET_FORMATS =
+ new LinkedHashSet<>(Arrays.asList(TableFormat.HUDI, TableFormat.ICEBERG, TableFormat.DELTA));
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .desc("The base path of the source table (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG (single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .desc(
+ "Comma-separated target formats to sync to, e.g. ICEBERG,DELTA "
+ + "(single-table mode)")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATASET_CONFIG)
+ .hasArg()
+ .argName("path")
+ .desc(
+ "Path to a YAML dataset config (sourceFormat, targetFormats, datasets[]) to "
+ + "sync multiple tables in one run; may be a local or cloud (s3/gcs/abfs) "
+ + "path. Mutually exclusive with --basepath/--sourceformat/--targets.")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(DATA_PATH)
+ .hasArg()
+ .desc("The path of the data files if different from the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TABLE_NAME)
+ .hasArg()
+ .desc("The table name; defaults to the last segment of the base path")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(NAMESPACE)
+ .hasArg()
+ .desc("The dot-separated table namespace")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(PARTITION_SPEC)
+ .hasArg()
+ .desc("The Hudi source partition field spec, e.g. level:VALUE")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(USE_DELTA_KERNEL)
+ .desc(
+ "Use the Spark-free Delta Kernel implementation for the Delta source and/or "
+ + "target instead of the default delta-core path. When omitted this is "
+ + "auto-enabled on Spark 3.5+ (where the bundled delta-core does not "
+ + "run); pass it explicitly to force Kernel on any Spark version.")
+ .build())
+ .addOption(Option.builder().longOpt(HELP).desc("Displays help information").build());
+
+ private XTableSparkSync() {}
+
+ public static void main(String[] args) {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine cmd;
+ try {
+ cmd = parser.parse(OPTIONS, args);
+ } catch (ParseException e) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ throw new IllegalArgumentException("Failed to parse arguments", e);
+ }
+ if (cmd.hasOption(HELP)) {
+ new HelpFormatter().printHelp("xtable-spark-sync", OPTIONS, true);
+ return;
+ }
+
+ boolean datasetMode = cmd.hasOption(DATASET_CONFIG);
+ if (datasetMode
+ && (cmd.hasOption(BASE_PATH) || cmd.hasOption(SOURCE_FORMAT) || cmd.hasOption(TARGETS))) {
+ throw new IllegalArgumentException(
+ "--datasetconfig is mutually exclusive with --basepath/--sourceformat/--targets");
+ }
+ // Validate single-table flags before starting Spark. A --datasetconfig is validated after,
+ // because it is read through the Spark Hadoop config.
+ if (!datasetMode) {
+ requireOption(cmd, BASE_PATH);
+ requireOption(cmd, SOURCE_FORMAT);
+ requireOption(cmd, TARGETS);
+ validateFormat(
+ SOURCE_FORMAT,
+ cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT),
+ SUPPORTED_SOURCE_FORMATS);
+ parseTargets(cmd.getOptionValue(TARGETS)); // validates formats and non-emptiness
+ }
+
+ boolean useDeltaKernelFlag = cmd.hasOption(USE_DELTA_KERNEL);
+
+ SparkSession spark = SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+
+ String sourceFormat;
+ List<String> targetFormats;
+ List<TableSyncSpec.TableSyncSpecBuilder> builders = new ArrayList<>();
+ if (datasetMode) {
+ DatasetConfig config = readDatasetConfig(cmd.getOptionValue(DATASET_CONFIG), hadoopConf);
+ sourceFormat = config.sourceFormat;
+ targetFormats = config.targetFormats;
+ for (DatasetConfig.Table table : config.datasets) {
+ builders.add(
+ TableSyncSpec.builder()
+ .key(
+ table.tableName != null
+ ? table.tableName
+ : basePathToTableName(table.tableBasePath))
+ .basePath(table.tableBasePath)
+ .dataPath(table.tableDataPath)
+ .namespace(table.namespace == null ? null : table.namespace.split("\\."))
+ .partitionSpec(table.partitionSpec)
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+ } else {
+ sourceFormat = cmd.getOptionValue(SOURCE_FORMAT).trim().toUpperCase(Locale.ROOT);
+ targetFormats = parseTargets(cmd.getOptionValue(TARGETS));
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String namespace = cmd.getOptionValue(NAMESPACE);
+ builders.add(
+ TableSyncSpec.builder()
+ .key(cmd.getOptionValue(TABLE_NAME, basePathToTableName(basePath)))
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+ .sourceFormat(sourceFormat)
+ .targets(targetFormats));
+ }
+
+ // delta-core does not run on Spark 3.5+, so use Delta Kernel there.
+ boolean deltaInvolved =
+ TableFormat.DELTA.equals(sourceFormat) || targetFormats.contains(TableFormat.DELTA);
+ boolean useDeltaKernel = useDeltaKernelFlag;
+ if (!useDeltaKernel && deltaInvolved && isSparkAtLeast35(spark.version())) {
+ useDeltaKernel = true;
+ log.info(
+ "Spark {} detected (>= 3.5); auto-enabling the Delta Kernel implementation for Delta "
+ + "source/target. Pass --usedeltakernel to force it on any Spark version.",
+ spark.version());
+ }
+
+ XTableSyncService service = new XTableSyncService();
+ List<TableSyncSpec> specs = new ArrayList<>();
+ for (TableSyncSpec.TableSyncSpecBuilder builder : builders) {
+ specs.add(builder.useDeltaKernel(useDeltaKernel).build());
+ }
+ int failures = syncAll(specs, spec -> service.sync(spec, hadoopConf));
+ if (failures > 0) {
+ throw new IllegalStateException(
+ failures + " of " + specs.size() + " table sync(s) failed; see logs above");
+ }
+ } finally {
+ spark.stop();
+ }
+ }
+
+ /**
+ * Syncs every spec and returns how many failed. A table fails when its sync throws or when any
+ * target reports a non-SUCCESS status; one failure does not abort the rest of the batch.
+ */
+ // package-private for unit testing
+ static int syncAll(
+ List<TableSyncSpec> specs, Function<TableSyncSpec, Map<String, SyncResult>> syncer) {
+ int failures = 0;
+ for (TableSyncSpec spec : specs) {
+ try {
+ log.info("Starting XTable sync for {} ({})", spec.getKey(), spec.getBasePath());
+ // TableFormatSync reports a target failure as an ERROR status rather than throwing.
+ Map<String, String> failedTargets =
+ syncer.apply(spec).entrySet().stream()
+ .filter(
+ entry ->
+ entry.getValue().getTableFormatSyncStatus().getStatusCode()
+ != SyncStatusCode.SUCCESS)
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ entry -> describe(entry.getValue().getTableFormatSyncStatus()),
+ (a, b) -> a,
+ TreeMap::new));
+ if (failedTargets.isEmpty()) {
+ log.info("Completed XTable sync for {}", spec.getKey());
+ } else {
+ failures++;
+ log.error(
+ "XTable sync failed for {} ({}) on target(s) {}",
+ spec.getKey(),
+ spec.getBasePath(),
+ failedTargets);
+ }
+ } catch (RuntimeException e) {
+ failures++;
+ log.error("XTable sync failed for {} ({})", spec.getKey(), spec.getBasePath(), e);
+ }
+ }
+ return failures;
+ }
+
+ private static String describe(SyncResult.SyncStatus status) {
+ ErrorDetails details = status.getErrorDetails();
+ return details == null || details.getErrorMessage() == null
+ ? String.valueOf(status.getStatusCode())
+ : status.getStatusCode() + ": " + details.getErrorMessage();
+ }
+
+ private static void requireOption(CommandLine cmd, String option) {
+ if (!cmd.hasOption(option)) {
+ throw new IllegalArgumentException(
+ "--" + option + " is required unless --" + DATASET_CONFIG + " is provided");
+ }
+ }
+
+ // package-private for unit testing
+ static List<String> parseTargets(String targetsValue) {
+ List<String> targetFormats =
+ Arrays.stream(targetsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ if (targetFormats.isEmpty()) {
+ throw new IllegalArgumentException(
+ "--targets resolved to an empty list; provide at least one target format, e.g. ICEBERG,DELTA");
+ }
+ targetFormats.forEach(target -> validateFormat(TARGETS, target, SUPPORTED_TARGET_FORMATS));
+ return targetFormats;
+ }
+
+ static DatasetConfig readDatasetConfig(String path, Configuration hadoopConf) {
+ Path configPath = new Path(path);
+ try (InputStream in = configPath.getFileSystem(hadoopConf).open(configPath)) {
+ return parseDatasetConfig(in);
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Failed to read --datasetconfig at " + path, e);
+ }
+ }
+
+ // SafeConstructor builds only maps, lists and scalars. package-private for unit testing.
+ @SuppressWarnings("unchecked")
+ static DatasetConfig parseDatasetConfig(InputStream in) {
Review Comment:
Fixed in f99cf05a, Jackson now binds the parsed YAML to `DatasetConfig` and only the validation stays. It still parses with snakeyaml, since Spark does not always ship `jackson-dataformat-yaml`.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "the-other-tim-brown (via GitHub)" <gi...@apache.org>.
the-other-tim-brown commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4238741522
##########
xtable-spark-runtime/src/test/java/org/apache/xtable/spark/ITXTableSparkRuntimeBundle.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.hudi.client.HoodieReadClient;
+
+import org.apache.xtable.GenericTable;
+import org.apache.xtable.hudi.HudiTestUtil;
+
+/**
+ * Runs a sync with the shaded bundle through {@code $SPARK_HOME/bin/spark-submit}, one case per
+ * direction, and checks the target data equals the source. The engines come from {@code
+ * target/engine-classpath.txt}. Skipped when {@code SPARK_HOME} is not set.
+ */
+class ITXTableSparkRuntimeBundle {
+
+ @TempDir static java.nio.file.Path tempDir;
+
+ private static SparkSession sparkSession;
+ private static JavaSparkContext jsc;
+
+ @BeforeAll
+ static void setupSpark() {
+ sparkSession =
+ SparkSession.builder()
+ .config(HoodieReadClient.addHoodieSupport(HudiTestUtil.getSparkConf(tempDir)))
+ .getOrCreate();
+ jsc = JavaSparkContext.fromSparkContext(sparkSession.sparkContext());
+ }
+
+ @AfterAll
+ static void stopSpark() {
+ if (jsc != null) {
+ jsc.close();
+ }
+ if (sparkSession != null) {
+ sparkSession.stop();
+ }
+ }
+
+ private static Stream<Arguments> directions() {
+ // sourceFormat, targetFormat, hudiPartitionSpec
+ return Stream.of(
+ Arguments.of("HUDI", "ICEBERG", "level:VALUE"),
+ Arguments.of("ICEBERG", "HUDI", null),
+ Arguments.of("HUDI", "DELTA", "level:VALUE"),
+ Arguments.of("DELTA", "HUDI", null));
+ }
+
+ @ParameterizedTest(name = "{0} -> {1}")
+ @MethodSource("directions")
+ void bundleRunsSyncViaSparkSubmit(String sourceFormat, String targetFormat, String partitionSpec)
+ throws Exception {
+ String sparkHome = System.getenv("SPARK_HOME");
+ assumeTrue(sparkHome != null && !sparkHome.trim().isEmpty(), "SPARK_HOME is not set");
+
+ String tableName = sourceFormat.toLowerCase() + "_to_" + targetFormat.toLowerCase();
+ String basePath;
+ String dataPath;
+ List<String> columns;
+ String orderByColumn;
+ try (GenericTable<?, ?> table =
+ GenericTable.getInstance(tableName, tempDir, sparkSession, jsc, sourceFormat, true)) {
+ table.insertRows(100);
+ basePath = table.getBasePath();
+ dataPath = table.getDataPath();
+ columns = table.getColumnsToSelect();
+ orderByColumn = table.getOrderByColumn();
+ }
+
+ File bundleJar = findBundleJar();
+ File sparkSubmit = new File(sparkHome, "bin/spark-submit");
+ assertTrue(sparkSubmit.canExecute(), "spark-submit not executable at " + sparkSubmit);
+
+ // A flat classpath, not --packages: a child classloader breaks Avro casts across the boundary.
+ String engineClasspath = readEngineClasspath();
+
+ List<String> command = new ArrayList<>();
+ command.add(sparkSubmit.getAbsolutePath());
+ command.add("--master");
+ command.add("local[2]");
+ command.add("--conf");
+ command.add("spark.driver.extraClassPath=" + engineClasspath);
+ command.add("--conf");
+ command.add("spark.executor.extraClassPath=" + engineClasspath);
+ command.add("--class");
+ command.add("org.apache.xtable.spark.XTableSparkSync");
+ command.add(bundleJar.getAbsolutePath());
+ command.add("--basepath");
+ command.add(basePath);
+ command.add("--datapath");
+ command.add(dataPath);
+ command.add("--sourceformat");
+ command.add(sourceFormat);
+ command.add("--targets");
+ command.add(targetFormat);
+ command.add("--tablename");
+ command.add(tableName);
+ if (partitionSpec != null) {
+ command.add("--partitionspec");
+ command.add(partitionSpec);
+ }
+
+ ProcessBuilder pb = new ProcessBuilder(command);
+ pb.environment().put("SPARK_LOCAL_IP", "127.0.0.1");
+ pb.redirectErrorStream(true);
+ Process process = pb.start();
+ // Drain on a thread so a hung spark-submit cannot block the timeout below.
+ StringBuilder outputBuffer = new StringBuilder();
+ Thread drainer = new Thread(() -> readOutputInto(process, outputBuffer), "spark-submit-stdout");
+ drainer.setDaemon(true);
+ drainer.start();
+ boolean finished = process.waitFor(10, TimeUnit.MINUTES);
+ if (!finished) {
+ process.destroyForcibly();
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ synchronized (outputBuffer) {
+ throw new AssertionError("spark-submit timed out. Output:\n" + outputBuffer);
+ }
+ }
+ drainer.join(TimeUnit.SECONDS.toMillis(10));
+ String output;
+ synchronized (outputBuffer) {
+ output = outputBuffer.toString();
+ }
+ assertEquals(0, process.exitValue(), "spark-submit failed. Output:\n" + output);
+
+ assertDatasetEquivalence(
+ sourceFormat, targetFormat, basePath, dataPath, columns, orderByColumn, output);
+ }
+
+ /** Reads the source and the target back through Spark and asserts the scalar columns match. */
+ private void assertDatasetEquivalence(
+ String sourceFormat,
+ String targetFormat,
+ String basePath,
+ String dataPath,
+ List<String> columns,
+ String orderByColumn,
+ String submitOutput) {
+ Dataset<Row> source =
+ sparkSession
+ .read()
+ .options(readOptions(sourceFormat))
+ .format(sourceFormat.toLowerCase())
+ .load(basePath)
+ .orderBy(orderByColumn);
+ String[] projection = comparableColumns(source, columns).toArray(new String[0]);
+ List<String> sourceRows = source.selectExpr(projection).toJSON().collectAsList();
+ List<String> targetRows =
+ sparkSession
+ .read()
+ .options(readOptions(targetFormat))
+ .format(targetFormat.toLowerCase())
+ .load(dataPath)
+ .orderBy(orderByColumn)
+ .selectExpr(projection)
+ .toJSON()
+ .collectAsList();
+ assertEquals(
+ 100,
+ targetRows.size(),
+ "Unexpected target row count. spark-submit output:\n" + submitOutput);
+ assertEquals(sourceRows, targetRows, "Target is not data-equivalent to the source");
+ }
+
+ // Drops nested columns (covered by ITConversionController) and local timestamps (HUDI-7088).
+ private static List<String> comparableColumns(Dataset<Row> source, List<String> columns) {
+ java.util.Set<String> excluded = new java.util.HashSet<>();
+ for (org.apache.spark.sql.types.StructField field : source.schema().fields()) {
+ org.apache.spark.sql.types.DataType type = field.dataType();
+ if (type instanceof org.apache.spark.sql.types.StructType
+ || type instanceof org.apache.spark.sql.types.ArrayType
+ || type instanceof org.apache.spark.sql.types.MapType) {
+ excluded.add(field.name());
+ }
+ }
+ return columns.stream()
+ .filter(c -> !excluded.contains(c))
+ .filter(c -> !c.startsWith("timestamp_local_"))
+ .collect(java.util.stream.Collectors.toList());
+ }
+
+ private static java.util.Map<String, String> readOptions(String format) {
+ java.util.Map<String, String> options = new java.util.HashMap<>();
+ if ("HUDI".equalsIgnoreCase(format)) {
+ options.put("hoodie.metadata.enable", "true");
+ options.put("hoodie.datasource.read.extract.partition.values.from.path", "true");
+ } else if ("ICEBERG".equalsIgnoreCase(format)) {
+ // The vectorized reader mis-casts timestamps written by Hudi.
+ options.put("vectorization-enabled", "false");
+ }
+ return options;
+ }
+
+ private String readEngineClasspath() throws Exception {
+ File cpFile = new File("target/engine-classpath.txt");
+ assertTrue(
+ cpFile.isFile(),
+ "engine classpath not built at "
+ + cpFile.getAbsolutePath()
+ + " (dependency:build-classpath)");
+ String cp =
+ new String(
+ java.nio.file.Files.readAllBytes(cpFile.toPath()),
+ java.nio.charset.StandardCharsets.UTF_8)
+ .trim();
+ assertTrue(!cp.isEmpty(), "engine classpath file is empty");
+ return cp;
+ }
+
+ // The shaded jar is the main artifact; skip original-* and the sources, javadoc and tests jars.
+ private File findBundleJar() {
+ File targetDir = new File("target");
+ File[] candidates =
+ targetDir.listFiles(
+ (dir, name) ->
+ name.startsWith("xtable-spark-runtime")
+ && name.endsWith(".jar")
+ && !name.startsWith("original-")
+ && !name.contains("-sources")
+ && !name.contains("-javadoc")
+ && !name.contains("-tests"));
+ assertTrue(
+ candidates != null && candidates.length == 1,
+ "Expected exactly one bundle jar in "
+ + targetDir.getAbsolutePath()
+ + " (run the package phase first)");
+ return candidates[0];
+ }
+
+ private static void readOutputInto(Process process, StringBuilder sb) {
+ try (java.io.BufferedReader reader =
+ new java.io.BufferedReader(
+ new java.io.InputStreamReader(
Review Comment:
nitpick, can we import instead of using fully qualified path?
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [836] Add xtable-spark-runtime, a thin drop-in Spark bundle [incubator-xtable]
Posted by "vinishjail97 (via GitHub)" <gi...@apache.org>.
vinishjail97 commented on code in PR #964:
URL: https://github.com/apache/incubator-xtable/pull/964#discussion_r4251863236
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.spark;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.xtable.conversion.ConversionConfig;
+import org.apache.xtable.conversion.ConversionController;
+import org.apache.xtable.conversion.ConversionSourceProvider;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.delta.DeltaConversionSourceProvider;
+import org.apache.xtable.delta.DeltaConversionTargetConfig;
+import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
+import org.apache.xtable.kernel.DeltaKernelConversionSourceProvider;
+import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.sync.SyncMode;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.paimon.PaimonConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+
+/**
+ * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
+ * back to a full snapshot when an incremental sync is not safe, so it is safe to call after every
+ * write.
+ */
+@Log4j2
+public class XTableSyncService {
+
+ /** Runs a single sync for the given spec, returning the per-format results. */
+ public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
+ Properties sourceProperties = new Properties();
+ if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
+ sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+ }
+ // Targets write their metadata next to the data files, e.g. <basePath>/data for Iceberg.
+ String dataPath =
+ spec.getDataPath() != null && !spec.getDataPath().isEmpty()
+ ? spec.getDataPath()
+ : spec.getBasePath();
+ SourceTable sourceTable =
+ SourceTable.builder()
+ .name(spec.getKey())
+ .basePath(spec.getBasePath())
+ .dataPath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(spec.getSourceFormat())
+ .additionalProperties(sourceProperties)
+ .build();
+
+ // Read by ConversionTargetFactory to pick the Delta Kernel target.
+ List<TargetTable> targetTables =
+ spec.getTargets().stream()
+ .map(
+ targetFormat -> {
+ Properties targetProperties = new Properties();
+ if (spec.isUseDeltaKernel()) {
+ targetProperties.setProperty(
+ DeltaConversionTargetConfig.USE_KERNEL, Boolean.TRUE.toString());
+ }
+ return TargetTable.builder()
+ .name(spec.getKey())
+ .basePath(dataPath)
+ .namespace(spec.getNamespace())
+ .formatName(targetFormat)
+ .additionalProperties(targetProperties)
+ .build();
+ })
+ .collect(Collectors.toList());
+
+ ConversionConfig conversionConfig =
+ ConversionConfig.builder()
+ .sourceTable(sourceTable)
+ .targetTables(targetTables)
+ .syncMode(SyncMode.INCREMENTAL)
+ .build();
+
+ ConversionSourceProvider<?> sourceProvider =
+ sourceProviderFor(spec.getSourceFormat(), spec.isUseDeltaKernel());
+ sourceProvider.init(hadoopConf);
+
+ log.info(
+ "Running XTable sync for table {} ({} -> {}) at {}",
+ spec.getKey(),
+ spec.getSourceFormat(),
+ spec.getTargets(),
+ spec.getBasePath());
+ return new ConversionController(hadoopConf).sync(conversionConfig, sourceProvider);
+ }
+
+ private static ConversionSourceProvider<?> sourceProviderFor(
Review Comment:
Filed #968 to track this as a follow-up.
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,86 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+name: Spark Runtime Bundle Validation
+
+# Runs ITXTableSparkRuntimeBundle, which spark-submits the shaded bundle. The IT skips itself
+# unless SPARK_HOME is set, so it only runs here.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# No path filter, so the check always reports and can be a required status check.
+on:
+ push:
+ branches:
+ - "main"
+
+ pull_request:
+ branches:
+ - "main"
+
+ workflow_dispatch:
+
+env:
+ SPARK_HADOOP_PROFILE: "hadoop3"
+
+jobs:
+ validate-spark-runtime-bundle:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ # 3.4: Delta via delta-core. 3.5: Delta auto-switches to the Spark-free Delta Kernel.
+ spark_version: ["3.4.3", "3.5.9"]
+ env:
+ SPARK_VERSION: ${{ matrix.spark_version }}
+ steps:
+ - uses: actions/checkout@v6
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v5
+ with:
+ java-version: '11'
+ distribution: 'temurin'
+ cache: maven
+
+ - name: Cache Spark distribution
Review Comment:
Filed #969 to track this as a follow-up.
--
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: commits-unsubscribe@xtable.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org