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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
