vinishjail97 commented on code in PR #894:
URL: https://github.com/apache/incubator-xtable/pull/894#discussion_r4262911941
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -145,6 +165,220 @@ public InternalFilesDiff getDiffForCommit(
return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
}
+ /**
+ * Derives the file diff from the metadata of the commit being written,
rather than by comparing
+ * two committed states as {@link #getDiffForCommit(HoodieInstant,
InternalTable, HoodieInstant,
+ * HoodieTimeline)} does. Requires the constructor taking a {@link
FileSystemViewManager}, since
+ * it reads the live file system view. Log files are skipped, so
merge-on-read updates are not
+ * represented.
+ */
+ public InternalFilesDiff getDiffFromCommitMetadata(
+ InternalTable table, HoodieCommitMetadata commitMetadata, HoodieInstant
commit) {
+ SyncableFileSystemView fsView =
fileSystemViewManager.getFileSystemView(metaClient);
+ List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+ List<InternalDataFile> filesToRemove = new ArrayList<>();
+ Map<String, StoragePathInfo> fullPathInfo =
+ commitMetadata.getFullPathToInfo(metaClient.getStorage(),
basePath.toString());
+ commitMetadata
+ .getPartitionToWriteStats()
+ .forEach(
+ (partitionPath, writeStats) -> {
+ List<PartitionValue> partitionValues =
+ partitionValuesExtractor.extractPartitionValues(
+ table.getPartitioningFields(), partitionPath);
+ Map<String, HoodieBaseFile> currentBaseFilesInPartition =
+ previousBaseFileVersions(fsView, partitionPath,
commit.requestedTime());
+ for (HoodieWriteStat writeStat : writeStats) {
+ if (FSUtils.isLogFile(new StoragePath(writeStat.getPath()))) {
+ continue;
+ }
+ StoragePath baseFileFullPath =
+ FSUtils.constructAbsolutePath(metaClient.getBasePath(),
writeStat.getPath());
+ if
(FSUtils.getCommitTimeWithFullPath(baseFileFullPath.toString())
+ .equals(commit.requestedTime())) {
+ // getFullPathToInfo keys the map by the absolute path, not
the file name
+ StoragePathInfo pathInfo =
fullPathInfo.get(baseFileFullPath.toString());
+ if (pathInfo == null) {
+ throw new ReadException(
+ "Commit metadata has no file info for base file " +
baseFileFullPath);
+ }
+ filesAddedWithoutStats.add(
+ buildFileWithoutStats(partitionValues, new
HoodieBaseFile(pathInfo)));
+ }
+ if
(currentBaseFilesInPartition.containsKey(writeStat.getFileId())) {
+ filesToRemove.add(
+ buildFileWithoutStats(
+ partitionValues,
currentBaseFilesInPartition.get(writeStat.getFileId())));
+ }
+ }
+ });
+ List<InternalDataFile> filesAdded =
+ fileStatsExtractor
+ .addStatsToFiles(tableMetadata, filesAddedWithoutStats.stream(),
table.getReadSchema())
+ .collect(Collectors.toList());
+ return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesToRemove).build();
+ }
+
+ /**
+ * Replace-commit counterpart of {@link #getDiffFromCommitMetadata}. Files
the replace commit
Review Comment:
Nit: this Javadoc belongs to `getDiffFromReplaceCommitMetadata` further
down, but it sits on top of the rollback method's Javadoc.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -145,6 +165,220 @@ public InternalFilesDiff getDiffForCommit(
return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
}
+ /**
+ * Derives the file diff from the metadata of the commit being written,
rather than by comparing
+ * two committed states as {@link #getDiffForCommit(HoodieInstant,
InternalTable, HoodieInstant,
+ * HoodieTimeline)} does. Requires the constructor taking a {@link
FileSystemViewManager}, since
+ * it reads the live file system view. Log files are skipped, so
merge-on-read updates are not
+ * represented.
+ */
+ public InternalFilesDiff getDiffFromCommitMetadata(
+ InternalTable table, HoodieCommitMetadata commitMetadata, HoodieInstant
commit) {
+ SyncableFileSystemView fsView =
fileSystemViewManager.getFileSystemView(metaClient);
+ List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+ List<InternalDataFile> filesToRemove = new ArrayList<>();
+ Map<String, StoragePathInfo> fullPathInfo =
+ commitMetadata.getFullPathToInfo(metaClient.getStorage(),
basePath.toString());
+ commitMetadata
+ .getPartitionToWriteStats()
+ .forEach(
+ (partitionPath, writeStats) -> {
+ List<PartitionValue> partitionValues =
+ partitionValuesExtractor.extractPartitionValues(
+ table.getPartitioningFields(), partitionPath);
+ Map<String, HoodieBaseFile> currentBaseFilesInPartition =
+ previousBaseFileVersions(fsView, partitionPath,
commit.requestedTime());
+ for (HoodieWriteStat writeStat : writeStats) {
+ if (FSUtils.isLogFile(new StoragePath(writeStat.getPath()))) {
+ continue;
+ }
+ StoragePath baseFileFullPath =
+ FSUtils.constructAbsolutePath(metaClient.getBasePath(),
writeStat.getPath());
+ if
(FSUtils.getCommitTimeWithFullPath(baseFileFullPath.toString())
+ .equals(commit.requestedTime())) {
+ // getFullPathToInfo keys the map by the absolute path, not
the file name
+ StoragePathInfo pathInfo =
fullPathInfo.get(baseFileFullPath.toString());
+ if (pathInfo == null) {
+ throw new ReadException(
+ "Commit metadata has no file info for base file " +
baseFileFullPath);
+ }
+ filesAddedWithoutStats.add(
+ buildFileWithoutStats(partitionValues, new
HoodieBaseFile(pathInfo)));
+ }
+ if
(currentBaseFilesInPartition.containsKey(writeStat.getFileId())) {
+ filesToRemove.add(
+ buildFileWithoutStats(
+ partitionValues,
currentBaseFilesInPartition.get(writeStat.getFileId())));
+ }
+ }
+ });
+ List<InternalDataFile> filesAdded =
+ fileStatsExtractor
+ .addStatsToFiles(tableMetadata, filesAddedWithoutStats.stream(),
table.getReadSchema())
+ .collect(Collectors.toList());
+ return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesToRemove).build();
+ }
+
+ /**
+ * Replace-commit counterpart of {@link #getDiffFromCommitMetadata}. Files
the replace commit
+ * supersedes are reported as removed, files it wrote as added.
+ */
+ /**
+ * Derives the file diff a completed rollback produces, from the rollback's
own metadata rather
+ * than by comparing timeline states: the base files the rollback deleted
leave, and for each file
+ * group they belonged to the previous base file, which the rolled-back
commit had superseded and
+ * which is the latest file slice again, comes back. Log files are skipped.
Only files of commits
+ * in {@code publishedCommitTimes} are removed, since a commit that never
reached the target has
+ * nothing there to remove.
+ *
+ * @param publishedCommitTimes requested times of the commits the target has
recorded
+ */
+ public InternalFilesDiff getDiffFromRollbackMetadata(
+ InternalTable table,
+ HoodieRollbackMetadata rollbackMetadata,
+ Set<String> publishedCommitTimes) {
+ SyncableFileSystemView fsView =
fileSystemViewManager.getFileSystemView(metaClient);
+ // The rollback has just deleted files and removed the instant, so refresh
the cached view.
+ fsView.sync();
+ List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+ List<InternalDataFile> filesToRemove = new ArrayList<>();
+ rollbackMetadata
+ .getPartitionMetadata()
+ .forEach(
+ (partitionPath, partitionMetadata) -> {
+ List<String> deletedBaseFiles =
+ partitionMetadata.getSuccessDeleteFiles().stream()
+ .filter(path -> FSUtils.isBaseFile(new
StoragePath(path)))
+ .filter(
+ path ->
+ publishedCommitTimes.contains(
+ FSUtils.getCommitTimeWithFullPath(path)))
+ .collect(Collectors.toList());
+ if (deletedBaseFiles.isEmpty()) {
+ return;
+ }
+ filesToRemove.addAll(
+ getRemovedFiles(partitionPath, deletedBaseFiles,
table.getPartitioningFields()));
+ Set<String> fileIdsRolledBack =
Review Comment:
I am not sure this covers the rollback of a completed replace commit. For
clustering, the deleted files belong to the new file groups, so the replaced
file groups that become live again in Hudi never match `fileIdsRolledBack`, and
Iceberg ends up with neither set. A `delete_partition` rollback deletes
nothing, so it returns early on line 258 with the same result.
I ran this on S3 with TPC-DS `customer` (12M rows) written through the Spark
datasource. After rolling back a completed clustering, Hudi reads 11,940,000
rows and Iceberg reads 0. After rolling back an `insert_overwrite` of one
partition, Iceberg is missing all 977,187 rows of that partition.
Would it work to read the rolled-back instant's `partitionToReplaceFileIds`
and add those groups' latest base files back? A cluster-then-rollback case in
`ITIcebergRollbacks` would cover it, and I would see this as blocking since
Iceberg readers lose the data.
##########
xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/IcebergTableFormat.java:
##########
@@ -0,0 +1,460 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.common.HoodieTableFormat;
+import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.TimelineFactory;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.keygen.constant.KeyGeneratorType;
+import org.apache.hudi.metadata.TableMetadataFactory;
+
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+import org.apache.xtable.conversion.ConversionTargetFactory;
+import org.apache.xtable.conversion.SourceTable;
+import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.exception.ReadException;
+import org.apache.xtable.exception.UpdateException;
+import org.apache.xtable.hudi.HudiDataFileExtractor;
+import org.apache.xtable.hudi.HudiFileStatsExtractor;
+import org.apache.xtable.hudi.HudiIncrementalTableChangeExtractor;
+import org.apache.xtable.hudi.HudiInstantUtils;
+import org.apache.xtable.hudi.HudiSchemaExtractor;
+import org.apache.xtable.hudi.HudiSourceConfig;
+import org.apache.xtable.hudi.HudiTableExtractor;
+import org.apache.xtable.hudi.PathBasedPartitionSpecExtractor;
+import org.apache.xtable.hudi.PathBasedPartitionValuesExtractor;
+import org.apache.xtable.iceberg.IcebergConversionSource;
+import org.apache.xtable.iceberg.IcebergConversionTarget;
+import org.apache.xtable.iceberg.IcebergTableManager;
+import org.apache.xtable.metadata.IcebergMetadataFactory;
+import org.apache.xtable.model.IncrementalTableChanges;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
+import org.apache.xtable.model.schema.PartitionTransformType;
+import org.apache.xtable.model.storage.DataLayoutStrategy;
+import org.apache.xtable.model.sync.SyncResult;
+import org.apache.xtable.model.sync.SyncStatusCode;
+import org.apache.xtable.spi.sync.TableFormatSync;
+import org.apache.xtable.timeline.IcebergActiveTimeline;
+import org.apache.xtable.timeline.IcebergSnapshotInstants;
+import org.apache.xtable.timeline.IcebergTimelineArchiver;
+import org.apache.xtable.timeline.IcebergTimelineFactory;
+
+public class IcebergTableFormat implements HoodieTableFormat {
+ /**
+ * The partition type a custom key generator records for a field whose value
is the partition path
+ * as is. See {@code CustomAvroKeyGenerator.PartitionKeyType} in Hudi.
+ */
+ private static final String CUSTOM_KEY_GENERATOR_SIMPLE_PARTITION_TYPE =
"SIMPLE";
+
+ private transient TableFormatSync tableFormatSync;
+
+ public IcebergTableFormat() {}
+
+ /**
+ * Hudi passes the table's {@code hoodie.properties} here on every meta
client load, which is the
+ * earliest point an unsupported table shape can be refused: before any
instant exists.
+ */
+ @Override
+ public void init(Properties properties) {
+ requireCopyOnWrite(new HoodieConfig(TypedProperties.copy(properties)));
Review Comment:
Hudi enables the metadata table by default, so a writer that only sets
`hoodie.table.format=ICEBERG` will hit the `Unsupported operation:
getColumnsStats` error the test harness works around. Would a fail-fast check
that names `hoodie.metadata.enable`, like `requireCopyOnWrite`, be reasonable
here or in the hooks?
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -145,6 +165,220 @@ public InternalFilesDiff getDiffForCommit(
return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
}
+ /**
+ * Derives the file diff from the metadata of the commit being written,
rather than by comparing
+ * two committed states as {@link #getDiffForCommit(HoodieInstant,
InternalTable, HoodieInstant,
+ * HoodieTimeline)} does. Requires the constructor taking a {@link
FileSystemViewManager}, since
+ * it reads the live file system view. Log files are skipped, so
merge-on-read updates are not
+ * represented.
+ */
+ public InternalFilesDiff getDiffFromCommitMetadata(
+ InternalTable table, HoodieCommitMetadata commitMetadata, HoodieInstant
commit) {
+ SyncableFileSystemView fsView =
fileSystemViewManager.getFileSystemView(metaClient);
+ List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+ List<InternalDataFile> filesToRemove = new ArrayList<>();
+ Map<String, StoragePathInfo> fullPathInfo =
+ commitMetadata.getFullPathToInfo(metaClient.getStorage(),
basePath.toString());
+ commitMetadata
+ .getPartitionToWriteStats()
+ .forEach(
+ (partitionPath, writeStats) -> {
+ List<PartitionValue> partitionValues =
+ partitionValuesExtractor.extractPartitionValues(
+ table.getPartitioningFields(), partitionPath);
+ Map<String, HoodieBaseFile> currentBaseFilesInPartition =
+ previousBaseFileVersions(fsView, partitionPath,
commit.requestedTime());
+ for (HoodieWriteStat writeStat : writeStats) {
+ if (FSUtils.isLogFile(new StoragePath(writeStat.getPath()))) {
+ continue;
+ }
+ StoragePath baseFileFullPath =
+ FSUtils.constructAbsolutePath(metaClient.getBasePath(),
writeStat.getPath());
+ if
(FSUtils.getCommitTimeWithFullPath(baseFileFullPath.toString())
+ .equals(commit.requestedTime())) {
+ // getFullPathToInfo keys the map by the absolute path, not
the file name
+ StoragePathInfo pathInfo =
fullPathInfo.get(baseFileFullPath.toString());
+ if (pathInfo == null) {
+ throw new ReadException(
+ "Commit metadata has no file info for base file " +
baseFileFullPath);
+ }
+ filesAddedWithoutStats.add(
+ buildFileWithoutStats(partitionValues, new
HoodieBaseFile(pathInfo)));
+ }
+ if
(currentBaseFilesInPartition.containsKey(writeStat.getFileId())) {
+ filesToRemove.add(
+ buildFileWithoutStats(
+ partitionValues,
currentBaseFilesInPartition.get(writeStat.getFileId())));
+ }
+ }
+ });
+ List<InternalDataFile> filesAdded =
+ fileStatsExtractor
+ .addStatsToFiles(tableMetadata, filesAddedWithoutStats.stream(),
table.getReadSchema())
+ .collect(Collectors.toList());
+ return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesToRemove).build();
+ }
+
+ /**
+ * Replace-commit counterpart of {@link #getDiffFromCommitMetadata}. Files
the replace commit
+ * supersedes are reported as removed, files it wrote as added.
+ */
+ /**
+ * Derives the file diff a completed rollback produces, from the rollback's
own metadata rather
+ * than by comparing timeline states: the base files the rollback deleted
leave, and for each file
+ * group they belonged to the previous base file, which the rolled-back
commit had superseded and
+ * which is the latest file slice again, comes back. Log files are skipped.
Only files of commits
+ * in {@code publishedCommitTimes} are removed, since a commit that never
reached the target has
+ * nothing there to remove.
+ *
+ * @param publishedCommitTimes requested times of the commits the target has
recorded
+ */
+ public InternalFilesDiff getDiffFromRollbackMetadata(
+ InternalTable table,
+ HoodieRollbackMetadata rollbackMetadata,
+ Set<String> publishedCommitTimes) {
+ SyncableFileSystemView fsView =
fileSystemViewManager.getFileSystemView(metaClient);
+ // The rollback has just deleted files and removed the instant, so refresh
the cached view.
+ fsView.sync();
+ List<InternalDataFile> filesAddedWithoutStats = new ArrayList<>();
+ List<InternalDataFile> filesToRemove = new ArrayList<>();
+ rollbackMetadata
+ .getPartitionMetadata()
+ .forEach(
+ (partitionPath, partitionMetadata) -> {
+ List<String> deletedBaseFiles =
+ partitionMetadata.getSuccessDeleteFiles().stream()
+ .filter(path -> FSUtils.isBaseFile(new
StoragePath(path)))
+ .filter(
+ path ->
+ publishedCommitTimes.contains(
+ FSUtils.getCommitTimeWithFullPath(path)))
+ .collect(Collectors.toList());
+ if (deletedBaseFiles.isEmpty()) {
+ return;
+ }
+ filesToRemove.addAll(
+ getRemovedFiles(partitionPath, deletedBaseFiles,
table.getPartitioningFields()));
+ Set<String> fileIdsRolledBack =
Review Comment:
@yihua If the rollback of a completed commit can't be supported let's throw
an exception instead of the data corruption?
--
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]
