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