yihua commented on code in PR #894:
URL: https://github.com/apache/incubator-xtable/pull/894#discussion_r4299919763
##########
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:
Fixed in 9d8a3ca, so no exception is needed. The replace commit's metadata
is already deleted when `completedRollback` runs, so the replaced file groups
now come from the data files the rolled-back commit's own Iceberg snapshot
removed, and each comes back with its latest base file. `ITIcebergRollbacks`
(clustering) and `ITIcebergVariousActions` (insert_overwrite, delete_partition)
fail without the fix the way you saw, and a local Spark run of all three
rollbacks, followed by new-key and update writes, keeps Hudi and Iceberg
identical.
##########
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:
Done in 9d8a3ca: the commit hook refuses a table whose `hoodie.properties`
records a built metadata table, naming `hoodie.metadata.enable`, and the next
writer with it disabled rolls the refused commit back. Only commits are
checked, since a clean or rollback has deleted its files by the time its hook
runs, and Hudi's Spark rollback procedures build a metadata table with their
default write config.
##########
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:
Done, moved in 431fcaa.
--
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]
