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