yihua commented on code in PR #894: URL: https://github.com/apache/incubator-xtable/pull/894#discussion_r4225249292
########## xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/timeline/IcebergRollbackExecutor.java: ########## @@ -0,0 +1,101 @@ +/* + * 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.timeline; + +import lombok.SneakyThrows; +import lombok.extern.log4j.Log4j2; + +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.InstantComparison; +import org.apache.hudi.common.table.timeline.dto.InstantDTO; + +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; + +import org.apache.xtable.iceberg.IcebergConversionTarget; +import org.apache.xtable.iceberg.IcebergTableManager; +import org.apache.xtable.model.InternalTable; +import org.apache.xtable.model.metadata.TableSyncMetadata; + +@Log4j2 +public class IcebergRollbackExecutor { + private static final ObjectMapper MAPPER = + new ObjectMapper() + .registerModule(new JavaTimeModule()) + .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false) + .setSerializationInclusion(JsonInclude.Include.NON_NULL); + + private final HoodieTableMetaClient metaClient; + private final IcebergConversionTarget target; + private final IcebergTableManager tableManager; + + public IcebergRollbackExecutor(HoodieTableMetaClient metaClient, IcebergConversionTarget target) { + this.metaClient = metaClient; + this.target = target; + this.tableManager = + IcebergTableManager.of( + (org.apache.hadoop.conf.Configuration) metaClient.getStorageConf().unwrap()); + } + + @SneakyThrows + public void rollbackSnapshot(InternalTable internalTable, HoodieInstant instantToRollback) { + TableIdentifier tableIdentifier = + TableIdentifier.of(metaClient.getTableConfig().getTableName()); + if (tableManager.tableExists(null, tableIdentifier, metaClient.getBasePath().toString())) { + Table table = + tableManager.getTable(null, tableIdentifier, metaClient.getBasePath().toString()); + TableSyncMetadata syncMetadata = + TableSyncMetadata.fromJson( + table.currentSnapshot().summary().get(TableSyncMetadata.XTABLE_METADATA)) + .get(); + HoodieInstant latestHoodieInstantInIceberg = + InstantDTO.toInstant( + MAPPER.readValue(syncMetadata.getLatestTableOperationIdentifier(), InstantDTO.class), + metaClient.getInstantGenerator()); + if (latestHoodieInstantInIceberg.equals(instantToRollback)) { + // The instant to rollback is committed in iceberg, so rollback to previous snapshot. + // NOTE: This is equivalent to hudi restore and should be performed by killing all active + // writers. + target.beginSync(internalTable); + target.rollbackToSnapshotId(table.currentSnapshot().snapshotId()); Review Comment: Fixed. Rolling back the instant the current snapshot records now calls `rollbackTo(parentId)` and fails loudly when there is no parent; `testsForRollbacks` asserts Iceberg plans exactly the base files Hudi kept and that each one exists on storage. ########## xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/timeline/IcebergActiveTimeline.java: ########## @@ -0,0 +1,166 @@ +/* + * 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.timeline; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import lombok.SneakyThrows; + +import org.apache.hadoop.conf.Configuration; + +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.dto.InstantDTO; +import org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2; +import org.apache.hudi.common.table.timeline.versioning.v2.InstantComparatorV2; + +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; + +import org.apache.xtable.iceberg.IcebergTableManager; +import org.apache.xtable.model.metadata.TableSyncMetadata; + +public class IcebergActiveTimeline extends ActiveTimelineV2 { + private static final ObjectMapper MAPPER = + new ObjectMapper() + .registerModule(new JavaTimeModule()) + .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false) + .setSerializationInclusion(JsonInclude.Include.NON_NULL); + + public IcebergActiveTimeline( + HoodieTableMetaClient metaClient, + Set<String> includedExtensions, + boolean applyLayoutFilters) { + this.setInstants(getInstantsFromFileSystem(metaClient, includedExtensions, applyLayoutFilters)); + this.metaClient = metaClient; + } + + public IcebergActiveTimeline(HoodieTableMetaClient metaClient) { + this(metaClient, Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE), true); + } + + public IcebergActiveTimeline(HoodieTableMetaClient metaClient, boolean applyLayoutFilters) { + this( + metaClient, + Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE), + applyLayoutFilters); + } + + public IcebergActiveTimeline() {} + + @Override + public HoodieActiveTimeline reload() { + return new IcebergActiveTimeline(metaClient); + } + + /** + * Whether an action adds or removes data files, and therefore has to be recorded in an Iceberg + * snapshot before it counts as completed. Savepoint and restore change no data, so an Iceberg + * snapshot never records them and their state is taken from the Hudi timeline as-is. + */ + static boolean changesDataFiles(String action) { + return !HoodieTimeline.SAVEPOINT_ACTION.equals(action) + && !HoodieTimeline.RESTORE_ACTION.equals(action); + } + + /** + * Requested time alone does not identify an instant: savepointing a commit produces a savepoint + * instant at that commit's own requested time, so the action has to be part of the key or the two + * collide and one is dropped from the reconstructed timeline. + */ + static String instantKey(HoodieInstant instant) { + return instant.requestedTime() + "." + instant.getAction(); + } + + @SneakyThrows + protected List<HoodieInstant> getInstantsFromFileSystem( + HoodieTableMetaClient metaClient, + Set<String> includedExtensions, + boolean applyLayoutFilters) { + List<HoodieInstant> instantsFromHoodieTimeline = + super.getInstantsFromFileSystem(metaClient, includedExtensions, applyLayoutFilters); + IcebergTableManager icebergTableManager = + IcebergTableManager.of((Configuration) metaClient.getStorageConf().unwrap()); + TableIdentifier tableIdentifier = + TableIdentifier.of(metaClient.getTableConfig().getTableName()); + if (!icebergTableManager.tableExists( + null, tableIdentifier, metaClient.getBasePath().toString())) { + return Collections.emptyList(); + } + Table icebergTable = + icebergTableManager.getTable(null, tableIdentifier, metaClient.getBasePath().toString()); + Map<String, HoodieInstant> instantsFromIceberg = new HashMap<>(); + for (Snapshot snapshot : icebergTable.snapshots()) { Review Comment: Addressed. `IcebergActiveTimeline` and `IcebergTimelineArchiver` now walk `SnapshotUtil.currentAncestors(table)` instead of every retained snapshot, the archiver expires the snapshots a rollback leaves behind, and `testsForRollbacks` asserts the reconstructed timeline after the rollback, after the next commit, and after archival. ########## xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/IcebergTableFormat.java: ########## @@ -0,0 +1,248 @@ +/* + * 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.time.Instant; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +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.common.HoodieTableFormat; +import org.apache.hudi.common.config.HoodieConfig; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.model.HoodieCommitMetadata; +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.metadata.TableMetadataFactory; + +import com.google.common.collect.ImmutableMap; + +import org.apache.xtable.conversion.ConversionTargetFactory; +import org.apache.xtable.conversion.TargetTable; +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.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.IcebergConversionTarget; +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.spi.sync.TableFormatSync; +import org.apache.xtable.timeline.IcebergRollbackExecutor; +import org.apache.xtable.timeline.IcebergTimelineArchiver; +import org.apache.xtable.timeline.IcebergTimelineFactory; + +public class IcebergTableFormat implements HoodieTableFormat { + private transient TableFormatSync tableFormatSync; + + public IcebergTableFormat() {} + + @Override + public void init(Properties properties) { + this.tableFormatSync = TableFormatSync.getInstance(); + } + + @Override + public String getName() { + return org.apache.xtable.model.storage.TableFormat.ICEBERG; + } + + @Override + public void commit( + HoodieCommitMetadata commitMetadata, + HoodieInstant completedInstant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) { + HudiIncrementalTableChangeExtractor hudiTableExtractor = + getHudiTableExtractor(metaClient, viewManager); + completeInstant( + metaClient, hudiTableExtractor.extractTableChanges(commitMetadata, completedInstant)); + } + + @Override + public void clean( + HoodieCleanMetadata cleanMetadata, + HoodieInstant completedInstant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) { + HudiIncrementalTableChangeExtractor hudiTableExtractor = + getHudiTableExtractor(metaClient, viewManager); + completeInstant(metaClient, hudiTableExtractor.extractTableChanges(completedInstant)); + } + + @Override + public void archive( + Supplier<List<HoodieInstant>> archivedInstants, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) { + HudiIncrementalTableChangeExtractor hudiTableExtractor = + getHudiTableExtractor(metaClient, viewManager); + InternalTable internalTable = + hudiTableExtractor + .getTableExtractor() + .table( + metaClient, + metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().get()); + archiveInstants(metaClient, internalTable, archivedInstants.get()); + } + + @Override + public void rollback( + HoodieInstant completedInstant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) { + HudiIncrementalTableChangeExtractor hudiTableExtractor = + getHudiTableExtractor(metaClient, viewManager); + InternalTable internalTable = + hudiTableExtractor + .getTableExtractor() + .table( + metaClient, + metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().get()); + IcebergRollbackExecutor rollbackExecutor = + new IcebergRollbackExecutor(metaClient, getIcebergConversionTarget(metaClient)); + rollbackExecutor.rollbackSnapshot(internalTable, completedInstant); + } + + @Override + public void completedRollback( + HoodieInstant rollbackInstant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) { + metaClient.reloadActiveTimeline(); + HudiIncrementalTableChangeExtractor hudiTableExtractor = + getHudiTableExtractor(metaClient, viewManager); + completeInstant(metaClient, hudiTableExtractor.extractTableChanges(rollbackInstant)); + } + + /** + * A savepoint changes no data files, so there is nothing for Iceberg to record. Materializing it + * as a snapshot would put a non-data snapshot at the tip whose completion time is later than the + * commits it protects, which breaks the rollback path. {@link + * org.apache.xtable.timeline.IcebergActiveTimeline} takes savepoint instants from the Hudi + * timeline directly instead. + */ + @Override + public void savepoint( + HoodieInstant instant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) {} + + /** + * The per-instant rollbacks a restore performs already moved the Iceberg table back, and the + * restore instant itself changes no data files. + */ + @Override + public void restore( + HoodieInstant restoreCompletedInstant, + HoodieEngineContext engineContext, + HoodieTableMetaClient metaClient, + FileSystemViewManager viewManager) {} + + @Override + public TimelineFactory getTimelineFactory() { + return new IcebergTimelineFactory(new HoodieConfig()); + } + + @Override + public TableMetadataFactory getMetadataFactory() { + return IcebergMetadataFactory.getInstance(); + } + + private void completeInstant(HoodieTableMetaClient metaClient, IncrementalTableChanges changes) { + IcebergConversionTarget target = getIcebergConversionTarget(metaClient); + TableSyncMetadata tableSyncMetadata = + target + .getTableMetadata() + .orElse(TableSyncMetadata.of(Instant.MIN, Collections.emptyList())); + try { + tableFormatSync.syncChanges(ImmutableMap.of(target, tableSyncMetadata), changes); + } catch (Exception e) { + throw new UpdateException("Failed to update iceberg metadata", e); + } + } + + private void archiveInstants( + HoodieTableMetaClient metaClient, + InternalTable internalTable, + List<HoodieInstant> archivedInstants) { + IcebergConversionTarget target = getIcebergConversionTarget(metaClient); + IcebergTimelineArchiver timelineArchiver = new IcebergTimelineArchiver(metaClient, target); + timelineArchiver.archiveInstants(internalTable, archivedInstants); + } + + private HudiIncrementalTableChangeExtractor getHudiTableExtractor( + HoodieTableMetaClient metaClient, FileSystemViewManager viewManager) { + String partitionSpec = + metaClient + .getTableConfig() + .getPartitionFields() + .map( + partitionPaths -> + Arrays.stream(partitionPaths) + .map(p -> String.format("%s:VALUE", p)) Review Comment: Done. `IcebergTableFormat.partitionFieldSpec` maps each field through the `PartitionTransformType.VALUE` constant and rejects a timestamp-based key generator, or a custom key generator with a non-simple field, naming the table and field; `TestIcebergTableFormatPartitioning` covers both rejected shapes and the partitioned ITs cover identity partitioning end to end. ########## xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableExtractor.java: ########## @@ -88,9 +95,62 @@ public InternalTable table(HoodieTableMetaClient metaClient, HoodieInstant commi .readSchema(canonicalSchema) .latestMetadataPath(metaClient.getMetaPath().toString()) .latestCommitTime(HudiInstantUtils.parseFromInstantTime(commit.requestedTime())) + .latestTableOperationId(generateTableOperationId(commit)) .build(); } + public InternalTable table( + HoodieTableMetaClient metaClient, + HoodieCommitMetadata commitMetadata, + HoodieInstant completedInstant) { + InternalSchema canonicalSchema = getCanonicalSchema(commitMetadata); + List<InternalPartitionField> partitionFields = partitionSpecExtractor.spec(canonicalSchema); + List<InternalField> recordKeyFields = getRecordKeyFields(metaClient, canonicalSchema); + if (!recordKeyFields.isEmpty()) { + canonicalSchema = canonicalSchema.toBuilder().recordKeyFields(recordKeyFields).build(); + } + DataLayoutStrategy dataLayoutStrategy = + partitionFields.size() > 0 + ? DataLayoutStrategy.DIR_HIERARCHY_PARTITION_VALUES + : DataLayoutStrategy.FLAT; + return InternalTable.builder() + .tableFormat(TableFormat.HUDI) + .basePath(metaClient.getBasePath().toString()) + .name(metaClient.getTableConfig().getTableName()) + .layoutStrategy(dataLayoutStrategy) + .partitioningFields(partitionFields) + .readSchema(canonicalSchema) + .latestMetadataPath(metaClient.getMetaPath().toString()) + .latestCommitTime( Review Comment: Addressed by the rebase. Main now has `HudiInstantUtils.getSyncInstant` from #835, which picks completion time on table version 8+ and requested time below, so both overloads call it and record the same time for the same instant, which makes a separate `HudiTableExtractorV9` unnecessary. ########## xtable-api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java: ########## @@ -46,6 +47,10 @@ public class TableSyncMetadata { new ObjectMapper() .registerModule(new JavaTimeModule()) .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false) + // A reader on an older version has to tolerate fields a newer writer added, since this + // blob is persisted in target-table metadata and is read back by whichever version + // happens to open the table next. + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) Review Comment: Done. Added `ignoresFieldsAddedByANewerWriter`, which parses a version-0 payload carrying an unknown field, plus a round-trip case with `latestTableOperationIdentifier` set. -- 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]
