vinishjail97 commented on code in PR #830:
URL: https://github.com/apache/incubator-xtable/pull/830#discussion_r3531968364
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTable.java:
##########
@@ -18,19 +18,25 @@
package org.apache.xtable.conversion;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.Properties;
import lombok.Builder;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
+import org.apache.hadoop.conf.Configuration;
+
@EqualsAndHashCode(callSuper = true)
@Getter
public class SourceTable extends ExternalTable {
/** The path to the data files, defaults to the basePath */
@NonNull private final String dataPath;
+ private final transient Configuration hadoopConf;
Review Comment:
`hadoopConf` is redeclared here while `ExternalTable` already has `protected
final Configuration hadoopConf`, and both get assigned in the ctor. Two
problems:
- With `@EqualsAndHashCode(callSuper = true)`, `hadoopConf` is now compared
in both parent and child, and `Configuration` doesn't override `equals()` — so
equality degrades to identity and two otherwise-identical `SourceTable`s built
with different `Configuration` instances become unequal (affects
`convertToSourceTable`, dedup, and test asserts).
- `transient` here but not on the parent field is inconsistent, and moot
since nothing implements `Serializable`.
Drop the child field and use the inherited one.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTableFormatDetector.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.conversion;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+public class SourceTableFormatDetector {
+ // private constructor to prevent instantiation
+ private SourceTableFormatDetector() {
+ throw new UnsupportedOperationException("This class cannot be
instantiated");
+ }
+ // helper method to detect input format
+ public static String detectFormat(String pathStr, Configuration conf) throws
IOException {
+ String sanitizeBasePath = ExternalTable.sanitizeBasePath(pathStr);
+ Path basePath = new Path(sanitizeBasePath);
+ FileSystem fs = basePath.getFileSystem(conf);
+
+ List<String> matches = new ArrayList<>();
+
+ if (fs.exists(new Path(basePath, "_delta_log"))) {
+ matches.add(TableFormat.DELTA);
+ }
+
+ if (fs.exists(new Path(basePath, ".hoodie"))) {
+ matches.add(TableFormat.HUDI);
+ }
+
+ try {
+ HadoopTables tables = new HadoopTables(conf);
+ Table table = tables.load(pathStr);
+ if (table != null) {
+ matches.add(TableFormat.ICEBERG);
+ }
+ } catch (Exception e) {
Review Comment:
`catch (Exception e)` swallows everything with no logging, and leaves a
commented-out `throw` (dead code — please remove). A real IO/permission/network
error while probing Iceberg is now indistinguishable from "not an Iceberg
table", so a misconfiguration is silently dropped and detection falls through
to "Unable to detect". At minimum log at DEBUG with the path + exception, and
ideally distinguish "table not found" from hard failures.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/TargetTable.java:
##########
@@ -39,8 +41,9 @@ public TargetTable(
String[] namespace,
CatalogConfig catalogConfig,
Duration metadataRetention,
- Properties additionalProperties) {
- super(name, formatName, basePath, namespace, catalogConfig,
additionalProperties);
+ Properties additionalProperties,
+ Configuration hadoopConf) {
Review Comment:
`hadoopConf` is now threaded through `TargetTable`, but only source
detection uses it — targets never auto-detect. This adds a required constructor
param with no consumer. Consider not propagating `hadoopConf` to `TargetTable`
(or clearly making it optional) to avoid unnecessary API churn.
##########
xtable-api/pom.xml:
##########
@@ -29,6 +29,10 @@
<name>XTable Project API</name>
<dependencies>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
Review Comment:
`commons-lang3` added here but I don't see a usage in the diff — is it
actually needed, or leftover?
##########
xtable-api/pom.xml:
##########
@@ -88,5 +92,13 @@
<groupId>org.mockito</groupId>
<artifactId>mockito-junit-jupiter</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-core</artifactId>
Review Comment:
This pulls `iceberg-core`/`iceberg-api` into `xtable-api`, the thin API
module everything depends on transitively. Format detection (and its Iceberg
dependency) fits better in `xtable-core` — putting Iceberg on the API surface
bloats every downstream consumer, which cuts against the thin-runtime direction
in #836. Can the detector live in xtable-core instead?
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -214,10 +232,28 @@ static CatalogConfig getIcebergCatalogConfig(String
icebergCatalogConfigPath) th
static ConversionSourceProvider<?> getConversionSourceProvider(
String conversionProviderConfigpath, DatasetConfig datasetConfig,
Configuration hadoopConf)
throws IOException {
- // Process source format
String sourceFormat = datasetConfig.sourceFormat;
byte[] customConfig =
getCustomConfigurations(conversionProviderConfigpath);
TableFormatConverters tableFormatConverters =
loadTableFormatConversionConfigs(customConfig);
+ if (sourceFormat == null
+ && datasetConfig.getDatasets() != null
+ && !datasetConfig.getDatasets().isEmpty()) {
+ DatasetConfig.Table firstTable = datasetConfig.getDatasets().get(0);
+ if (firstTable.getTableBasePath() != null) {
+ try {
+ String tablePath = firstTable.getTableBasePath();
+ sourceFormat = SourceTableFormatDetector.detectFormat(tablePath,
hadoopConf);
Review Comment:
Format is detected twice on the no-format path: once here to pick the
provider, and again inside the `SourceTable` no-format constructor (via
`sourceTableBuilder`). That's two FS scans + two Iceberg loads per table.
Detect once and thread the resolved format through to `sourceTableBuilder`.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTableFormatDetector.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.conversion;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+public class SourceTableFormatDetector {
+ // private constructor to prevent instantiation
+ private SourceTableFormatDetector() {
+ throw new UnsupportedOperationException("This class cannot be
instantiated");
+ }
+ // helper method to detect input format
+ public static String detectFormat(String pathStr, Configuration conf) throws
IOException {
+ String sanitizeBasePath = ExternalTable.sanitizeBasePath(pathStr);
+ Path basePath = new Path(sanitizeBasePath);
+ FileSystem fs = basePath.getFileSystem(conf);
+
+ List<String> matches = new ArrayList<>();
+
+ if (fs.exists(new Path(basePath, "_delta_log"))) {
+ matches.add(TableFormat.DELTA);
+ }
+
+ if (fs.exists(new Path(basePath, ".hoodie"))) {
+ matches.add(TableFormat.HUDI);
+ }
+
+ try {
+ HadoopTables tables = new HadoopTables(conf);
+ Table table = tables.load(pathStr);
+ if (table != null) {
+ matches.add(TableFormat.ICEBERG);
+ }
+ } catch (Exception e) {
+ // throw new IllegalArgumentException("Failed to inspect Iceberg table
at " + pathStr, e);
+ }
+
+ if (matches.size() == 1) {
+ return matches.get(0);
+ }
+
+ if (matches.size() > 1) {
Review Comment:
Ambiguity concern for XTable's own use case: a sync writes target metadata
into the *same* base path (e.g. a Hudi source synced to Iceberg ends up with
both `.hoodie` and `metadata/` under one path). On a re-run — or the in-job
listener in #836 — this method would then match multiple formats and throw
"Multiple table formats detected", breaking the idempotent re-sync path. How
should detection tell the *source* format apart from previously-written
*target* metadata? (e.g. only auto-detect before the first sync, or read
xtable's sync metadata to identify the source.)
##########
xtable-utilities/src/main/java/org/apache/xtable/utilities/RunSync.java:
##########
@@ -214,10 +232,28 @@ static CatalogConfig getIcebergCatalogConfig(String
icebergCatalogConfigPath) th
static ConversionSourceProvider<?> getConversionSourceProvider(
String conversionProviderConfigpath, DatasetConfig datasetConfig,
Configuration hadoopConf)
throws IOException {
- // Process source format
String sourceFormat = datasetConfig.sourceFormat;
byte[] customConfig =
getCustomConfigurations(conversionProviderConfigpath);
TableFormatConverters tableFormatConverters =
loadTableFormatConversionConfigs(customConfig);
+ if (sourceFormat == null
+ && datasetConfig.getDatasets() != null
+ && !datasetConfig.getDatasets().isEmpty()) {
+ DatasetConfig.Table firstTable = datasetConfig.getDatasets().get(0);
+ if (firstTable.getTableBasePath() != null) {
+ try {
+ String tablePath = firstTable.getTableBasePath();
+ sourceFormat = SourceTableFormatDetector.detectFormat(tablePath,
hadoopConf);
+ log.info(
+ "Source format was omitted in config. Auto-detected table
format: {}", sourceFormat);
+ } catch (Exception e) {
Review Comment:
Two issues in this catch:
1. We only reach this block when `datasetConfig.sourceFormat == null`, so
`sourceFormat` is still `null` — the message renders "Source format null is not
supported", which misdescribes the failure (it's *detection* that failed, not
an unsupported format).
2. The actual cause `e` is dropped, losing the stack trace of why detection
failed.
Report a detection-failure message and chain the cause, e.g. `throw new
IllegalArgumentException("Failed to auto-detect source format for path " +
tablePath, e);`.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTable.java:
##########
@@ -39,8 +45,38 @@ public SourceTable(
String dataPath,
String[] namespace,
CatalogConfig catalogConfig,
- Properties additionalProperties) {
- super(name, formatName, basePath, namespace, catalogConfig,
additionalProperties);
+ Properties additionalProperties,
+ Configuration hadoopConf) {
+ super(name, formatName, basePath, namespace, catalogConfig,
additionalProperties, hadoopConf);
this.dataPath = dataPath == null ? this.getBasePath() :
sanitizeBasePath(dataPath);
+ this.hadoopConf = hadoopConf;
+ }
+
+ public SourceTable(
+ @NonNull String name,
+ @NonNull String basePath,
+ String dataPath,
+ String[] namespace,
+ CatalogConfig catalogConfig,
+ Properties additionalProperties,
+ Configuration hadoopConf) {
+ super(
+ name,
+ resolveFormatOrThrow(basePath, hadoopConf),
+ basePath,
+ namespace,
+ catalogConfig,
+ additionalProperties,
+ hadoopConf);
+ this.dataPath = dataPath == null ? this.getBasePath() :
sanitizeBasePath(dataPath);
+ this.hadoopConf = hadoopConf;
+ }
+
+ private static String resolveFormatOrThrow(String basePath, Configuration
hadoopConf) {
Review Comment:
This makes `new SourceTable(...)` perform remote I/O at construction
(`fs.exists` + Iceberg `load` against S3/HDFS) and throw `UncheckedIOException`
from a constructor. Construction triggering network calls is surprising and
hard to test/mock. Prefer an explicit factory (e.g.
`SourceTable.withDetectedFormat(...)`) or keep detection in the RunSync/builder
layer so plain construction stays side-effect-free.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTableFormatDetector.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.conversion;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+public class SourceTableFormatDetector {
+ // private constructor to prevent instantiation
+ private SourceTableFormatDetector() {
+ throw new UnsupportedOperationException("This class cannot be
instantiated");
+ }
+ // helper method to detect input format
+ public static String detectFormat(String pathStr, Configuration conf) throws
IOException {
+ String sanitizeBasePath = ExternalTable.sanitizeBasePath(pathStr);
+ Path basePath = new Path(sanitizeBasePath);
+ FileSystem fs = basePath.getFileSystem(conf);
+
+ List<String> matches = new ArrayList<>();
+
+ if (fs.exists(new Path(basePath, "_delta_log"))) {
+ matches.add(TableFormat.DELTA);
+ }
+
+ if (fs.exists(new Path(basePath, ".hoodie"))) {
+ matches.add(TableFormat.HUDI);
+ }
+
+ try {
+ HadoopTables tables = new HadoopTables(conf);
+ Table table = tables.load(pathStr);
Review Comment:
Iceberg detection relies solely on `HadoopTables.load(pathStr)`, which only
resolves the *Hadoop-tables* layout (`<basePath>/metadata/version-hint.text`).
Two cases it can't cover:
1. **Catalog-managed Iceberg** (Hive/Glue/REST/Nessie). XTable's own
`IcebergTableManager.getTable` resolves these via `catalog.loadTable(...)` and
only falls back to `HadoopTables` when *no* catalog is configured. Here we
never consult `catalogConfig`, so a catalog table XTable can actually sync
won't be detected.
2. **Relocated metadata** — Iceberg metadata can live outside the base path
(`write.metadata.path`, or a catalog-tracked metadata pointer). A
`load(basePath)` / prefix check can't find it.
Since `detectFormat(pathStr, conf)` doesn't take `catalogConfig`,
catalog-based detection isn't even possible today. Can we thread
`catalogConfig` through and reuse the same resolution order as
`IcebergTableManager` (catalog first, HadoopTables fallback)? If the intended
scope really is path-based HadoopTables only, that limitation should be
documented explicitly.
##########
xtable-api/src/main/java/org/apache/xtable/conversion/SourceTableFormatDetector.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.conversion;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import org.apache.xtable.model.storage.TableFormat;
+
+public class SourceTableFormatDetector {
+ // private constructor to prevent instantiation
+ private SourceTableFormatDetector() {
+ throw new UnsupportedOperationException("This class cannot be
instantiated");
+ }
+ // helper method to detect input format
+ public static String detectFormat(String pathStr, Configuration conf) throws
IOException {
Review Comment:
Public API method with no Javadoc (just an inline `// helper` comment) —
please document the return value, the thrown exceptions, and the detection
scope. Minor: the delta/hudi checks use the sanitized `basePath`, but
`HadoopTables.load(pathStr)` uses the raw arg — use the sanitized path
consistently.
##########
xtable-utilities/src/test/java/org/apache/xtable/utilities/ITRunSync.java:
##########
@@ -93,6 +94,32 @@ void testContinuousSyncMode(@TempDir Path tempDir) throws
IOException {
}
}
+ @Test
+ void testSingleSyncModeWithoutInputTableFormat(@TempDir Path tempDir) throws
IOException {
+ String tableName = "test-table";
+ try (GenericTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+ table.insertRows(10);
+ File configFile = writeConfigFileWithoutSourceTableFormat(tempDir,
table, tableName);
+ String[] args = new String[] {"--datasetConfig", configFile.getPath()};
+ RunSync.main(args);
+ Path icebergMetadataPath = Paths.get(URI.create(table.getBasePath() +
"/metadata"));
+ waitForNumIcebergCommits(icebergMetadataPath, 3);
Review Comment:
Coverage gaps: only Hudi-source auto-detection is exercised. Please add
detection tests for Delta and Iceberg sources, and for the "multiple formats
detected" ambiguity branch (the case most likely to bite re-syncs). Also this
asserts on commit *count* (`waitForNumIcebergCommits(..., 3)`) rather than data
— a parameterized per-format test asserting the resolved format would be
stronger.
--
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]
