vinothchandar commented on code in PR #843:
URL: https://github.com/apache/incubator-xtable/pull/843#discussion_r3607590820
##########
.github/workflows/mvn-ci-build.yml:
##########
@@ -24,10 +24,12 @@ on:
push:
branches:
- "main"
+ - "branch-0.4"
Review Comment:
I was a little confused at first to see the version number. But it seems you
already cut the release branch
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,85 @@
+#
+# 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
+
+# Validates the shaded xtable-spark-runtime bundle end-to-end:
ITXTableSparkRuntimeBundle
+# spark-submits the bundled jar against a real Spark distribution for one case
per direction
+# across Hudi, Iceberg and Delta. The IT self-skips unless SPARK_HOME is set,
so the main CI
+# never exercises it; this workflow installs a matching Spark 3.4 distro and
runs it.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# Runs on every push/PR to the covered branches (no path filter) so the check
always
+# reports — required status checks that never trigger leave a PR blocked on a
pending check.
+on:
+ push:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ pull_request:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ workflow_dispatch:
+
+env:
+ SPARK_VERSION: "3.4.3"
Review Comment:
This cannot be 3.5 or something.?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Locale;
+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.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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 HELP = "help";
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .required()
+ .desc("The base path of the source table")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .required()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .required()
+ .desc("Comma-separated target formats to sync to, e.g.
ICEBERG,DELTA")
+ .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(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;
+ }
+
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String tableName = cmd.getOptionValue(TABLE_NAME,
basePathToName(basePath));
+ List<String> targetFormats =
+ Arrays.stream(cmd.getOptionValue(TARGETS).split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
Review Comment:
The resolved target list is never validated to be non-empty. An input like
`--targets ","` (or all-blank entries) passes through trim/filter and yields an
empty `targetFormats`, so the sync silently produces no target tables and does
nothing. Consider failing fast with a clear message when the resulting list is
empty.
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Locale;
+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.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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 HELP = "help";
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .required()
+ .desc("The base path of the source table")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .required()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .required()
+ .desc("Comma-separated target formats to sync to, e.g.
ICEBERG,DELTA")
+ .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(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;
+ }
+
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String tableName = cmd.getOptionValue(TABLE_NAME,
basePathToName(basePath));
+ List<String> targetFormats =
+ Arrays.stream(cmd.getOptionValue(TARGETS).split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ String namespace = cmd.getOptionValue(NAMESPACE);
+
+ TableSyncSpec spec =
+ TableSyncSpec.builder()
+ .key(tableName)
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+
.sourceFormat(cmd.getOptionValue(SOURCE_FORMAT).toUpperCase(Locale.ROOT))
Review Comment:
An unsupported `--sourceFormat` isn't rejected here; it surfaces only later
in `XTableSyncService.sourceProviderFor(...)` as an
`UnsupportedOperationException`, after the SparkSession has already been
created via `getOrCreate()`. Consider validating the source format (and
required args) up front in the CLI so an invalid value fails fast without
spinning up Spark.
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,27 @@ 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() also resolves provider classes lazily, so it can throw
+ // ServiceConfigurationError too - it must be inside the guard
alongside next().
+ if (!iterator.hasNext()) {
+ break;
+ }
+ target = iterator.next();
+ } catch (ServiceConfigurationError | LinkageError error) {
+ // A registered target whose engine library is not on the classpath
(e.g. Delta when only
+ // Hudi/Iceberg are provided). Skip it so a subset of engines can
still be used; a missing
+ // engine for the requested format surfaces below as
NotSupportedException. The offending
+ // provider is consumed before the error is thrown, so the next
hasNext() advances past it.
+ log.warn(
Review Comment:
The `warn` log is good; consider also logging the caught error's
class/message (or the provider name) so operators can distinguish "engine
intentionally not on classpath" from "engine present but failing to link.?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Locale;
+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.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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 HELP = "help";
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .required()
+ .desc("The base path of the source table")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
+ .hasArg()
+ .required()
+ .desc("The source table format, e.g. HUDI, DELTA or ICEBERG")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(TARGETS)
+ .hasArg()
+ .required()
+ .desc("Comma-separated target formats to sync to, e.g.
ICEBERG,DELTA")
+ .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(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;
+ }
+
+ String basePath = cmd.getOptionValue(BASE_PATH);
+ String tableName = cmd.getOptionValue(TABLE_NAME,
basePathToName(basePath));
+ List<String> targetFormats =
+ Arrays.stream(cmd.getOptionValue(TARGETS).split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .map(s -> s.toUpperCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ String namespace = cmd.getOptionValue(NAMESPACE);
+
+ TableSyncSpec spec =
+ TableSyncSpec.builder()
+ .key(tableName)
+ .basePath(basePath)
+ .dataPath(cmd.getOptionValue(DATA_PATH))
+ .namespace(namespace == null ? null : namespace.split("\\."))
+ .partitionSpec(cmd.getOptionValue(PARTITION_SPEC))
+
.sourceFormat(cmd.getOptionValue(SOURCE_FORMAT).toUpperCase(Locale.ROOT))
+ .targets(targetFormats)
+ .build();
+
+ SparkSession spark =
SparkSession.builder().appName("xtable-spark-sync").getOrCreate();
+ try {
+ Configuration hadoopConf = spark.sparkContext().hadoopConfiguration();
+ log.info("Starting standalone XTable sync for {}", spec.getBasePath());
+ new XTableSyncService().sync(spec, hadoopConf);
+ log.info("Completed XTable sync for {}", spec.getBasePath());
+ } finally {
+ spark.stop();
+ }
+ }
+
+ private static String basePathToName(String basePath) {
Review Comment:
throw errors or handle edge cases like these?: a `basePath` of `"/"` (or one
that is only a trailing slash) yields an empty table name, and a path with no
`/` returns the whole string. add UTs
rename: basePathToTableName? (is that what you intend?)
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -96,7 +119,10 @@ && isDeltaKernelTarget(target) == useKernel) {
throw new NotSupportedException("Target format is not yet supported: " +
tableFormatName);
}
+ private static final String DELTA_KERNEL_TARGET_CLASS =
+ "org.apache.xtable.kernel.DeltaKernelConversionTarget";
+
private static boolean isDeltaKernelTarget(ConversionTarget target) {
- return target instanceof DeltaKernelConversionTarget;
+ return DELTA_KERNEL_TARGET_CLASS.equals(target.getClass().getName());
Review Comment:
It won't fail at compile time if the class is renamed/moved. I assume tests
would fail? to catch this during dev?
##########
xtable-spark-runtime/pom.xml:
##########
@@ -0,0 +1,382 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable</artifactId>
+ <version>0.2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>
+ Thin, relocated, drop-in Spark bundle that runs incremental XTable
metadata sync. v1 ships
+ the standalone spark-submit entry point (XTableSparkSync); the
config-only, driver-side
+ QueryExecutionListener on-ramp is a planned follow-up (see rfc/rfc-3).
Review Comment:
So right now, the sync needs to be manually invoked?
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Locale;
+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.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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 HELP = "help";
+
+ private static final Options OPTIONS =
+ new Options()
+ .addOption(
+ Option.builder()
+ .longOpt(BASE_PATH)
+ .hasArg()
+ .required()
+ .desc("The base path of the source table")
+ .build())
+ .addOption(
+ Option.builder()
+ .longOpt(SOURCE_FORMAT)
Review Comment:
do we have a standard enum, where this can mapped from a format string back
the internal int/short? (why long)
##########
xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * 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.Arrays;
+import java.util.List;
+import java.util.Locale;
+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.spark.sql.SparkSession;
+
+/**
+ * Standalone {@code spark-submit} entry point that runs a single XTable sync
using the relocated
+ * {@code xtable-spark-runtime} bundle. It is the RunSync-equivalent for this
bundle and is
+ * primarily used to validate that the shaded jar is self-contained on a real
Spark install.
+ *
+ * <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";
Review Comment:
if these are cli options - can we drop the camel case to be more unix like?
i.e --basepath vs --basePath?
##########
rfc/rfc-3/rfc-3.md:
##########
@@ -0,0 +1,234 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+# RFC-3: xtable-spark-runtime - in-job metadata sync via a thin Spark bundle
+
+## Proposers
+
+- @vinishjail97
+
+## Approvers
+- Anyone from the XTable community can approve/add feedback.
Review Comment:
You can pencil me in
##########
rfc/rfc-3/rfc-3.md:
##########
@@ -0,0 +1,234 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+# RFC-3: xtable-spark-runtime - in-job metadata sync via a thin Spark bundle
+
+## Proposers
+
+- @vinishjail97
+
+## Approvers
+- Anyone from the XTable community can approve/add feedback.
+
+## Status
Review Comment:
the rfc looks good, given the implementation is in the same PR as well, you
can look at my comments on those
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,85 @@
+#
+# 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
+
+# Validates the shaded xtable-spark-runtime bundle end-to-end:
ITXTableSparkRuntimeBundle
+# spark-submits the bundled jar against a real Spark distribution for one case
per direction
+# across Hudi, Iceberg and Delta. The IT self-skips unless SPARK_HOME is set,
so the main CI
+# never exercises it; this workflow installs a matching Spark 3.4 distro and
runs it.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# Runs on every push/PR to the covered branches (no path filter) so the check
always
+# reports — required status checks that never trigger leave a PR blocked on a
pending check.
+on:
+ push:
+ branches:
+ - "main"
Review Comment:
The Spark version is hardcoded to `3.4.3`. Since this module targets the
Hudi 0.x / Spark 3.4 line, that's reasonable, but consider parameterizing it
(matrix or workflow input) or at least adding a comment explaining why 3.4.x is
pinned, so the validation stays easy to bump. (A maintainer also raised this
above.)
##########
.github/workflows/spark-runtime-validation.yml:
##########
@@ -0,0 +1,85 @@
+#
+# 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
+
+# Validates the shaded xtable-spark-runtime bundle end-to-end:
ITXTableSparkRuntimeBundle
+# spark-submits the bundled jar against a real Spark distribution for one case
per direction
+# across Hudi, Iceberg and Delta. The IT self-skips unless SPARK_HOME is set,
so the main CI
+# never exercises it; this workflow installs a matching Spark 3.4 distro and
runs it.
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: ${{ !contains(github.ref, 'main') }}
+
+# Runs on every push/PR to the covered branches (no path filter) so the check
always
+# reports — required status checks that never trigger leave a PR blocked on a
pending check.
+on:
+ push:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ pull_request:
+ branches:
+ - "main"
+ - "branch-0.4"
+
+ workflow_dispatch:
+
+env:
+ SPARK_VERSION: "3.4.3"
+ SPARK_HADOOP_PROFILE: "hadoop3"
+
+jobs:
+ validate-spark-runtime-bundle:
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ 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
+ uses: actions/cache@v4
+ with:
+ path: ~/spark
+ key: spark-${{ env.SPARK_VERSION }}-bin-${{ env.SPARK_HADOOP_PROFILE
}}
+
+ - name: Install Spark
+ run: |
+ set -euo pipefail
+ SPARK_DIR="spark-${SPARK_VERSION}-bin-${SPARK_HADOOP_PROFILE}"
Review Comment:
curious, any way to specify any caching so that we don't install spark each
run?
Not blocking - sth to think about
##########
xtable-spark-runtime/src/main/resources/META-INF/NOTICE-bundled:
##########
@@ -0,0 +1,22 @@
+Apache XTable (incubating)
+Copyright 2024-2026 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (https://www.apache.org/).
+
+--------------------------------------------------------------------------------
+
+This binary artifact bundles the following projects with NOTICE:
+
+--------------------------------------------------------------------------------
+
+Group: commons-cli Name: commons-cli Version: 1.5.0
Review Comment:
are we shading this fully. In my adventures, commons version mismatches left
and right with hadoop/spark classes/installs all the time
##########
xtable-spark-runtime/pom.xml:
##########
@@ -0,0 +1,382 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable</artifactId>
+ <version>0.2.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>xtable-spark-runtime_${scala.binary.version}</artifactId>
+ <name>XTable Project Spark Runtime</name>
+ <description>
+ Thin, relocated, drop-in Spark bundle that runs incremental XTable
metadata sync. v1 ships
+ the standalone spark-submit entry point (XTableSparkSync); the
config-only, driver-side
+ QueryExecutionListener on-ramp is a planned follow-up (see rfc/rfc-3).
+ </description>
+
+ <dependencies>
+ <!-- XTable engine (bundled) -->
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Command-line parsing for the standalone XTableSparkSync entry
point (bundled). -->
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ </dependency>
+
+ <!-- Logging API (lombok @Log4j2); provided by the cluster runtime -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Spark: provided by the cluster. Brings QueryExecutionListener,
QueryExecution,
+ Catalyst LogicalPlan, and SparkSession. -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-sql_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Hadoop: provided by the cluster. -->
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!--
+ Engine libraries (Hudi / Iceberg / Delta) are provided, not bundled:
the user brings their
+ own engine version and the thin bundle stays compatible across
versions (see the shade
+ plugin below). Hudi is supplied as the hudi-spark bundle, whose
regenerated DecimalWrapper
+ links cleanly on Avro >=1.10; raw hudi-common's
Avro-1.8/1.9-generated model classes
+ reference org.apache.avro.data.TimeConversions.TimeConversion,
removed in Avro 1.10 and thus
+ absent from the Avro 1.12 that Iceberg 1.9.2 requires.
+ -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+
<artifactId>hudi-spark${spark.version.prefix}-bundle_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <!--
+ XTable's Hudi TARGET writes via the Java client
(HoodieJavaEngineContext), which the Spark
Review Comment:
+1 . we should keep it this way, and publish the support matrix of what
format versions works for source -> target? (could we add docs for that?)
##########
rfc/rfc-3/rfc-3.md:
##########
@@ -0,0 +1,234 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+# RFC-3: xtable-spark-runtime - in-job metadata sync via a thin Spark bundle
+
+## Proposers
+
+- @vinishjail97
+
+## Approvers
+- Anyone from the XTable community can approve/add feedback.
+
+## Status
+
+GH Feature Request: https://github.com/apache/incubator-xtable/issues/836
+
+> Please keep the status updated in `rfc/README.md`.
+
+## Abstract
+
+XTable's modules are published to Maven, so a user can depend on `xtable-core`
and assemble their own
+runtime today. But there is no maintained, thin, drop-in artifact — everyone
re-solves the same
+shading/classpath problem, or uses `xtable-utilities`, an unshaded ~1 GB fat
jar that is not practical
+to add to a Spark job and requires running a separate `RunSync` process with
YAML config files.
+
+XTable conversion is metadata-only and lightweight, so the common case — "I
already write this table
+with Spark, keep it in sync in other formats" — should be a one-dependency,
config-only addition to an
+existing pipeline. This RFC proposes `xtable-spark-runtime`: a thin, relocated
Spark bundle that
+registers a driver-side listener and, after each successful write to a source
table, runs an
+incremental `ConversionController.sync(...)` for the configured targets. It
complements — does not
+replace — the standalone CLI (`RunSync`).
+
+## Background
+
+Everything needed to run a sync already exists in the engine; the gap is
purely packaging and a trigger:
+
+- The entire sync entry surface is Hadoop-`Configuration`-only: `new
ConversionController(conf).sync(config, sourceProvider)`.
+ The caller supplies only the **source** `ConversionSourceProvider`; target
providers are created
+ internally by `ConversionTargetFactory` (`ConversionController.java`).
Source providers instantiate
+ directly (`new HudiConversionSourceProvider()` /
`IcebergConversionSourceProvider` /
+ `DeltaConversionSourceProvider`) and take `.init(conf)`.
+- `SyncMode.INCREMENTAL` is self-healing: `ConversionController` auto-falls
back to a full snapshot
+ per-target when there is no prior sync metadata or incremental is not safe
(`isIncrementalSyncSufficient`).
+ First sync is effectively full; subsequent syncs are incremental.
+- The sync watermark (last-synced instant, pending commits) is persisted in
the **target's**
+ `TableSyncMetadata`. There is no additional state to keep on the client side.
+- `xtable-hudi-support-extensions` is a pure-Java precedent for a
Spark-adjacent module whose
+ `XTableSyncTool` already builds `SourceTable`/`TargetTable` and calls
`ConversionController.sync(...)`;
+ it declares Spark deps as `provided` and uses a Scala-suffixed artifactId.
+- Delta conversion currently runs Delta-on-Spark in-process (`delta-core`);
`DeltaConversionUtils.buildSparkSession`
+ uses `SparkSession.builder()...getOrCreate()` with no `master`, so it reuses
the host job's active
+ session rather than starting a second `SparkContext`. (A Spark-free Delta
**Kernel** path exists and,
+ once it becomes the default, removes the per-Spark-version coupling —
tracked separately.)
+
+## Implementation
+
+A new **pure-Java** module `xtable-spark-runtime_${scala.binary.version}`
(package `org.apache.xtable.spark`).
+
+### What ships in v1
+
+This PR delivers the **thin bundle plus the standalone `spark-submit` entry
point
+`XTableSparkSync`** — enough to run a relocated, in-cluster sync from one
command. The config-only,
+driver-side auto-sync (`XTableSyncListener` registered via
`spark.sql.queryExecutionListeners`, with
+`XTableSparkConfig` and `PlanTargetResolver`) is the headline on-ramp this RFC
designs toward, but it
+is **deferred to a follow-up PR** — those classes are not in v1. The listener
design is retained
+below as the target so the entry point and config schema stay aligned.
+
+### Activation — v1 (CLI entry point)
Review Comment:
Can we also document how to call this from code i.e from within pySpark or
spark code? These could go into the quickstart as well. Can we revamp that
--
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]
