vinishjail97 commented on code in PR #894:
URL: https://github.com/apache/incubator-xtable/pull/894#discussion_r3825121923
##########
xtable-api/src/main/java/org/apache/xtable/model/InternalTable.java:
##########
@@ -52,4 +52,6 @@ public class InternalTable {
Instant latestCommitTime;
// Path to latest metadata
String latestMetadataPath;
+ // latest operation on the table.
+ String latestTableOperationId;
Review Comment:
If I traced this right, the value written here is
`MAPPER.writeValueAsString(InstantDTO.fromInstant(instant))` from
`HudiTableExtractor#generateTableOperationId`, and three classes in the new
module read it back with `MAPPER.readValue(..., InstantDTO.class)`.
That would mean a model in `xtable-api` carrying a Hudi-serialised type as
an opaque `String`. I do not think it breaks anything today, and I understand
the appeal of not inventing a new abstraction for one caller. My only worry is
the next reader: the name does not say what the payload is, and nothing warns
that Hudi could change the `InstantDTO` shape.
Would you mind expanding `// latest operation on the table.` into a short
Javadoc, saying who writes it, who reads it, and that the contents are
format-specific and other targets should treat it as opaque? That would be
enough for me.
##########
xtable-api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java:
##########
@@ -56,6 +56,7 @@ public class TableSyncMetadata {
int version;
String sourceTableFormat;
String sourceIdentifier;
+ String latestTableOperationId;
Review Comment:
I may be misreading the compatibility story here, so please correct me.
`CURRENT_VERSION` stays at `0`, and `MAPPER` in this class does not set
`FAIL_ON_UNKNOWN_PROPERTIES` to false. If I follow that correctly, a 0.4.0
reader that opens a table a newer writer has synced would find
`latestTableOperationId` in the payload, fail on the unknown property, and
surface it as `ParseException` from `fromJson`. Since this blob lives in the
Iceberg snapshot summary and in the Delta table properties, that would affect
targets beyond Iceberg.
Would it be reasonable to disable `FAIL_ON_UNKNOWN_PROPERTIES` on this
mapper, and add a round-trip test for a version-0 payload both with and without
the new field? I am flagging it mainly because it is the one change in the PR
that a user could not work around by reverting the jar, so it seemed worth
being sure about.
##########
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:
A question rather than a finding, since I am not certain of the Iceberg
semantics here.
My understanding is that `Table#snapshots()` returns every snapshot retained
in table metadata rather than the ancestry of `currentSnapshot()`. If that is
right, then after `IcebergRollbackExecutor` calls
`rollbackTo(parentSnapshotId)` the rolled-back snapshot would remain in this
iteration until it is expired, and its instant would keep being reported as
completed by this timeline.
Can we add a test that rolls back and then asserts the timeline is expected?
I looked through `ITIcebergVariousActions` and could not find one.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -145,6 +164,110 @@ public InternalFilesDiff getDiffForCommit(
return
InternalFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
}
+ public InternalFilesDiff getDiffForCommit(
Review Comment:
Readability suggestion only. This overloads `getDiffForCommit` with a
different parameter set and also a different order: the existing one on line
147 is `(HoodieInstant, InternalTable, HoodieInstant, HoodieTimeline)`, this
one `(InternalTable, HoodieCommitMetadata, HoodieInstant)`. I misread a call
site on my first pass because `InternalTable` moves from second to first
position.
Since the semantics differ too, this one deriving the diff from commit
metadata rather than by diffing timeline state, would a distinct name such as
`getDiffFromCommitMetadata` be clearer? Same thought for
`getDiffForReplaceCommit`.
Both new methods are public and undocumented; a Javadoc noting the required
constructor and that log files are skipped would help a future reader.
##########
xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/timeline/IcebergTimelineArchiver.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import lombok.SneakyThrows;
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.dto.InstantDTO;
+
+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.IcebergConversionTarget;
+import org.apache.xtable.iceberg.IcebergTableManager;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
+
+@Log4j2
+public class IcebergTimelineArchiver {
+ 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 IcebergTimelineArchiver(HoodieTableMetaClient metaClient,
IcebergConversionTarget target) {
+ this.metaClient = metaClient;
+ this.target = target;
+ this.tableManager =
+ IcebergTableManager.of((Configuration)
metaClient.getStorageConf().unwrap());
+ }
+
+ @SneakyThrows
+ public void archiveInstants(InternalTable internalTable, List<HoodieInstant>
archivedInstants) {
+ TableIdentifier tableIdentifier =
+ TableIdentifier.of(metaClient.getTableConfig().getTableName());
+ if (tableManager.tableExists(null, tableIdentifier,
metaClient.getBasePath().toString())) {
+ Table table =
+ tableManager.getTable(null, tableIdentifier,
metaClient.getBasePath().toString());
+ Set<String> savepointedInstantTimes =
+ metaClient
+ .getActiveTimeline()
+ .getSavePointTimeline()
+ .filterCompletedInstants()
+ .getInstantsAsStream()
+ .map(HoodieInstant::requestedTime)
+ .collect(Collectors.toSet());
+ List<Long> expireSnapshots = new ArrayList<>();
+ for (Snapshot snapshot : table.snapshots()) {
+ TableSyncMetadata syncMetadata =
+
TableSyncMetadata.fromJson(snapshot.summary().get(TableSyncMetadata.XTABLE_METADATA))
+ .get();
+ HoodieInstant hoodieInstant =
+ InstantDTO.toInstant(
+ MAPPER.readValue(syncMetadata.getLatestTableOperationId(),
InstantDTO.class),
+ metaClient.getInstantGenerator());
+ // A savepoint changes no data, so no snapshot carries the savepoint
action itself. The
+ // savepointed commit's snapshot and everything newer has to survive
for a restore to it to
+ // remain possible.
+ if (savepointedInstantTimes.contains(hoodieInstant.requestedTime())) {
+ log.warn(
+ "Not expiring the snapshot for {} or any newer snapshot because
it is savepointed",
+ hoodieInstant);
+ break;
Review Comment:
I would like to check an assumption behind this `break`.
It looks like it relies on `table.snapshots()` being ordered oldest-first.
That holds for a simple append history, since it is the order in
`TableMetadata.snapshots`, but as far as I can tell Iceberg does not document
it as a guarantee. If the order ever differed, an early `break` would keep
snapshots that could have expired, which is harmless, but a late one would
expire a snapshot protecting a savepoint, which would cost a restore.
Given the second outcome, would sorting explicitly on
`Snapshot#sequenceNumber` (or `timestampMillis`) before the loop be worth the
small cost? A test that savepoints a middle commit and asserts the exact
expired set would also make the intent clear.
Separately, and only a preference: on line 96 a savepoint stopping expiry
looks like expected steady-state behaviour rather than something an operator
must act on, so `log.info` might fit better than `log.warn`.
##########
xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java:
##########
@@ -347,6 +348,26 @@ public Optional<String> getTargetCommitIdentifier(String
sourceIdentifier) {
return Optional.empty();
}
+ public void expireSnapshotIds(List<Long> snapshotIds) {
Review Comment:
Three small things on the two new public methods, all easy to disagree with.
Both are public on a class other modules now depend on, and neither has a
Javadoc. A line stating that `beginSync` must have run first would help, since
both dereference `transaction` without a null check and an NPE there would not
point at the real mistake.
The teardown (`commitTransaction`, then nulling `transaction`,
`internalTableState` and `tableSyncMetadata`) now appears three times in this
class counting `completeSync`. A private `resetTransactionState()` might be
worth it.
And `expireSnapshotIds(Collections.emptyList())` still calls `commit()` and
`commitTransaction()`, which I think writes a metadata version that changes
nothing. `IcebergTimelineArchiver` calls this on every archive round, so an
early return on an empty list may be worthwhile.
##########
xtable-api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java:
##########
@@ -178,6 +179,11 @@ private SyncResult getSyncResult(
fileSyncMethod.sync(conversionTarget);
conversionTarget.completeSync();
+ log.info(
Review Comment:
I don't think log is required..we already have success and failure logs in
`TableFormatSync` and we can track duration there itself?
##########
xtable-core/src/main/java/org/apache/xtable/conversion/ConversionTargetFactory.java:
##########
@@ -87,7 +90,31 @@ 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
Review Comment:
Can we prune these comments? The code is pretty obvious here and can be
shortened to 1-2 lines.
##########
xtable-api/src/main/java/org/apache/xtable/model/InternalTable.java:
##########
@@ -52,4 +52,6 @@ public class InternalTable {
Instant latestCommitTime;
// Path to latest metadata
String latestMetadataPath;
+ // latest operation on the table.
+ String latestTableOperationId;
Review Comment:
@yihua Can we enhance the java doc for this field?
##########
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:
@yihua Let's document and add validation on which hudi partition types are
supported? Do we have test coverage for different partition types supported?
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiIncrementalTableChangeExtractor.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.hudi;
+
+import java.util.Collections;
+import java.util.Iterator;
+
+import lombok.AllArgsConstructor;
+import lombok.Value;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+
+import org.apache.xtable.model.IncrementalTableChanges;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.TableChange;
+import org.apache.xtable.model.storage.InternalFilesDiff;
+
+/**
+ * Computes {@link org.apache.xtable.model.IncrementalTableChanges} between
current state of the
+ * table and new completed instant added to the timeline.
+ */
+@Value
+@AllArgsConstructor
+public class HudiIncrementalTableChangeExtractor {
+ HoodieTableMetaClient metaClient;
+ HudiTableExtractor tableExtractor;
+ HudiDataFileExtractor dataFileExtractor;
+
+ public IncrementalTableChanges extractTableChanges(
+ HoodieCommitMetadata commitMetadata, HoodieInstant completedInstant) {
+ InternalTable internalTable =
+ tableExtractor.table(metaClient, commitMetadata, completedInstant);
+ InternalFilesDiff dataFilesDiff;
+ if (commitMetadata instanceof HoodieReplaceCommitMetadata) {
+ dataFilesDiff =
+ dataFileExtractor.getDiffForReplaceCommit(
+ internalTable, (HoodieReplaceCommitMetadata) commitMetadata,
completedInstant);
+ } else {
+ dataFilesDiff =
+ dataFileExtractor.getDiffForCommit(internalTable, commitMetadata,
completedInstant);
+ }
+
+ Iterator<TableChange> tableChangeIterator =
+ Collections.singleton(
+ TableChange.builder()
+ .tableAsOfChange(internalTable)
+ .filesDiff(dataFilesDiff)
+ .sourceIdentifier(completedInstant.getCompletionTime())
+ .build())
+ .iterator();
+ return IncrementalTableChanges.builder()
+ .tableChanges(tableChangeIterator)
+ .pendingCommits(Collections.emptyList())
+ .build();
+ }
+
+ public IncrementalTableChanges extractTableChanges(HoodieInstant
completedInstant) {
+ InternalTable internalTable = tableExtractor.table(metaClient,
completedInstant);
+ Iterator<TableChange> tableChangeIterator =
+ Collections.singleton(
+ TableChange.builder()
+ .tableAsOfChange(internalTable)
+ .filesDiff(
+ InternalFilesDiff.from(Collections.emptyList(),
Collections.emptyList()))
Review Comment:
@yihua For pluggable table format..can we add a test assertion that hudi
clean action should maps to iceberg's expireSnapshot and files have been
deleted both in hudi and iceberg metadata?
https://github.com/apache/incubator-xtable/blob/main/xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java#L292
##########
xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/timeline/IcebergRollbackExecutor.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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.getLatestTableOperationId(),
InstantDTO.class),
+ metaClient.getInstantGenerator());
+ if (latestHoodieInstantInIceberg.equals(instantToRollback)) {
+ // The instant to rollback is the one the current snapshot records, so
un-publish it by
+ // making its parent current again.
+ // NOTE: This is equivalent to hudi restore and should be performed by
killing all active
+ // writers.
+ Long parentSnapshotId = table.currentSnapshot().parentId();
+ if (parentSnapshotId == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot roll back instant '%s' because the snapshot
recording it is the first "
+ + "snapshot of the table and has no parent to fall back
to.",
+ instantToRollback));
+ }
+ target.beginSync(internalTable);
+ target.rollbackToSnapshotId(parentSnapshotId);
+ } else if (InstantComparison.compareTimestamps(
+ latestHoodieInstantInIceberg.getCompletionTime(),
+ InstantComparison.LESSER_THAN,
+ instantToRollback.getCompletionTime())) {
+ // In this case, instantToRollback was not committed in iceberg, so we
can will be ignoring
+ // it.
+ log.info(
+ "Ignoring rollback to instant {}' because it is not committed in
Iceberg. Latest committed instant in Iceberg {}'",
Review Comment:
Tiny nit, no action needed if you would rather not touch it: there are two
stray apostrophes in this message, `instant {}'` and `Iceberg {}'`. The comment
a few lines up also reads "we can will be ignoring it".
##########
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:
I wanted to check whether this difference is deliberate. `latestCommitTime`
is set from `completedInstant.getCompletionTime()` here, and from
`commit.requestedTime()` in the existing overload on line 97. Those are
different clocks, and the field name does not distinguish them.
My concern is downstream: `TableFormatSync` writes this into
`TableSyncMetadata.lastInstantSynced`, which the incremental-sync decision
reads, so the two paths would report times for the same instant that do not
line up.
If the difference is intentional, would a comment here and a Javadoc on
`InternalTable.latestCommitTime` saying which clock it holds be enough? If not,
it may be worth making them agree.
##########
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(
+
HudiInstantUtils.parseFromInstantTime(completedInstant.getCompletionTime()))
+ .latestTableOperationId(generateTableOperationId(completedInstant))
+ .build();
+ }
+
+ private InternalSchema getCanonicalSchema(HoodieCommitMetadata
commitMetadata) {
+ HoodieSchema writerSchema =
+ HoodieSchema.parse(commitMetadata.getExtraMetadata().get(SCHEMA_KEY));
Review Comment:
`getExtraMetadata().get(SCHEMA_KEY)` returns null when a commit carries no
writer schema, and `HoodieSchema.parse(null)` would then fail with an NPE
naming neither the table nor the instant.
The thing that drew my eye is the asymmetry: the sibling
`getCanonicalSchema(metaClient, commit)` just below wraps its failure in
`SchemaExtractorException` with the table name, and this overload has no
wrapper at all, so the same class of failure would be reported quite
differently depending on the path.
Would a null check with a message naming the table and instant, plus
wrapping in `SchemaExtractorException` to match, be reasonable?
Two smaller notes, take or leave:
- the `false` in `HoodieSchemaUtils.addMetadataFields(writerSchema, false)`
on line 134 was opaque to me; a named local would make the call site
self-explaining.
- the two `getCanonicalSchema` overloads resolve the schema from genuinely
different sources, so distinct names might save the next reader a lookup.
##########
xtable-hudi-support/xtable-iceberg-pluggable-tf/src/main/java/org/apache/xtable/metadata/IcebergBackedTableMetadata.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.metadata;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.metadata.FileSystemBackedTableMetadata;
+import org.apache.hudi.storage.HoodieStorage;
+
+public class IcebergBackedTableMetadata extends FileSystemBackedTableMetadata {
Review Comment:
As far as I can tell this subclass adds no behaviour: it forwards the
constructor and overrides nothing, so `IcebergMetadataFactory` could return
`FileSystemBackedTableMetadata` directly today.
I assume the point is to reserve the type for the manifest-backed
implementation the description mentions, which seems sensible. Would you mind
adding a class Javadoc saying that, so a future reader does not remove it as
dead weight? Noting that it lists the filesystem deliberately for now, that
this is why the metadata table has to stay off, and that Iceberg manifests
already carry the per-column bounds it should eventually read, would cover it.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiIncrementalTableChangeExtractor.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.hudi;
+
+import java.util.Collections;
+import java.util.Iterator;
+
+import lombok.AllArgsConstructor;
+import lombok.Value;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+
+import org.apache.xtable.model.IncrementalTableChanges;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.TableChange;
+import org.apache.xtable.model.storage.InternalFilesDiff;
+
+/**
+ * Computes {@link org.apache.xtable.model.IncrementalTableChanges} between
current state of the
+ * table and new completed instant added to the timeline.
+ */
+@Value
+@AllArgsConstructor
+public class HudiIncrementalTableChangeExtractor {
+ HoodieTableMetaClient metaClient;
+ HudiTableExtractor tableExtractor;
+ HudiDataFileExtractor dataFileExtractor;
+
+ public IncrementalTableChanges extractTableChanges(
+ HoodieCommitMetadata commitMetadata, HoodieInstant completedInstant) {
+ InternalTable internalTable =
+ tableExtractor.table(metaClient, commitMetadata, completedInstant);
+ InternalFilesDiff dataFilesDiff;
+ if (commitMetadata instanceof HoodieReplaceCommitMetadata) {
+ dataFilesDiff =
+ dataFileExtractor.getDiffForReplaceCommit(
+ internalTable, (HoodieReplaceCommitMetadata) commitMetadata,
completedInstant);
+ } else {
+ dataFilesDiff =
+ dataFileExtractor.getDiffForCommit(internalTable, commitMetadata,
completedInstant);
+ }
+
+ Iterator<TableChange> tableChangeIterator =
+ Collections.singleton(
+ TableChange.builder()
+ .tableAsOfChange(internalTable)
+ .filesDiff(dataFilesDiff)
+ .sourceIdentifier(completedInstant.getCompletionTime())
+ .build())
+ .iterator();
+ return IncrementalTableChanges.builder()
+ .tableChanges(tableChangeIterator)
+ .pendingCommits(Collections.emptyList())
+ .build();
+ }
+
+ public IncrementalTableChanges extractTableChanges(HoodieInstant
completedInstant) {
+ InternalTable internalTable = tableExtractor.table(metaClient,
completedInstant);
+ Iterator<TableChange> tableChangeIterator =
+ Collections.singleton(
+ TableChange.builder()
+ .tableAsOfChange(internalTable)
+ .filesDiff(
+ InternalFilesDiff.from(Collections.emptyList(),
Collections.emptyList()))
Review Comment:
This is the comment I am least sure about, so I would value your read on it.
The clean path emits an empty `InternalFilesDiff`, so a Hudi CLEAN appears
to produce an Iceberg snapshot that advances the recorded instant while
removing nothing. If that is right, the files the clean deleted would stay in
the Iceberg metadata as live data files pointing at paths that no longer exist,
and an Iceberg reader would hit them on the next scan.
It looks like the same class of issue as #824, which is still open, except
applying to every clean rather than an incremental-sync edge case.
`HoodieCleanMetadata` does carry the deleted paths, and this overload does not
take that parameter.
Could the clean path emit those as `filesRemoved`? If it is better as a
follow-up, would you mind adding it to the limitations list in the description,
since a reader seeing missing files feels different in kind from a missing
feature.
Unrelated nit: `@Value` already generates an all-args constructor, so
`@AllArgsConstructor` on line 42 may be redundant.
##########
xtable-api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java:
##########
@@ -56,6 +56,7 @@ public class TableSyncMetadata {
int version;
String sourceTableFormat;
String sourceIdentifier;
+ String latestTableOperationId;
Review Comment:
@yihua Can we rename this to `latestTableOperationIdentifier`? It's
following the same pattern as `sourceIdentifier`.
##########
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:
Question about partition types rather than a problem I can demonstrate.
`String.format("%s:VALUE", p)` maps every Hudi partition field to the `VALUE`
type.
For a table partitioned on a date-formatted string column, does that lose
the date interpretation `PathBasedPartitionSpecExtractor` would otherwise
apply, and produce a string-typed Iceberg partition field? If non-`VALUE` types
are out of scope for now, would it be worth a line in the limitations list and
a `// TODO` here, so the code does not read as covering them?
Minor: a named constant for `"VALUE"` might read better than the inline
format string, and I think the extractor already has one.
##########
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:
@yihua I think should we make a new class `HudiTableExtractorV9` for
pluggable table format? The latestCommitTime used will be different for v6 and
v9..the original PR probably just added it here for overloading and avoiding
new classes.
##########
xtable-hudi-support/xtable-iceberg-pluggable-tf/pom.xml:
##########
@@ -0,0 +1,205 @@
+<?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-hudi-support</artifactId>
+ <version>0.5.0-SNAPSHOT</version>
+ </parent>
+
+
<artifactId>xtable-iceberg-pluggable-tf_${scala.binary.version}</artifactId>
+ <name>XTable Project Iceberg Pluggable Table Format</name>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.xtable</groupId>
+ <artifactId>xtable-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Logging API -->
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-client-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-sync-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Avro -->
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- Jackson -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
Review Comment:
Two dependencies the new code compiles against do not seem to be declared in
this module, so I think both resolve transitively through `xtable-core`:
- `jackson-datatype-jsr310`, since `JavaTimeModule` is imported in four of
the new classes, while `jackson-core` and `jackson-databind` are declared right
here.
- `guava`, since `IcebergTableFormat` imports
`com.google.common.collect.ImmutableMap`.
Nothing breaks today, so this is about not depending on `xtable-core`'s
scopes staying as they are. Would declaring jsr310 next to the other two be
reasonable? For guava, `Collections.singletonMap(target, tableSyncMetadata)` at
that one call site would avoid the dependency altogether, though I appreciate
that is a matter of taste.
A separate question while I am in this file: with no shade or assembly
plugin here, and `iceberg-core` at compile scope while the rest is `provided`,
how do you picture a user putting this on a Hudi writer's classpath? I ask
mainly because it will come up in the release discussion. If assembling it
themselves is the plan for now, a note in the description would help; if a
bundle is intended to follow, naming it as a follow-up would be enough.
--
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]
