You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "laskoviymishka (via GitHub)" <gi...@apache.org> on 2026/04/24 07:52:04 UTC
Re: [PR] Delta: Updating Delta to Iceberg conversion [iceberg]
laskoviymishka commented on code in PR #15407:
URL: https://github.com/apache/iceberg/pull/15407#discussion_r3136208821
##########
delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeKernelTableAction.java:
##########
@@ -0,0 +1,595 @@
+/*
+ * 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.iceberg.delta;
+
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import io.delta.kernel.internal.DeltaHistoryManager;
+import io.delta.kernel.internal.DeltaLogActionUtils;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.TableImpl;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterator;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class BaseSnapshotDeltaLakeKernelTableAction implements SnapshotDeltaLakeTable {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(BaseSnapshotDeltaLakeKernelTableAction.class);
+
+ private static final String SNAPSHOT_SOURCE_PROP = "snapshot_source";
+ private static final String DELTA_SOURCE_VALUE = "delta";
+ private static final String ORIGINAL_LOCATION_PROP = "original_location";
+ private static final String DELTA_VERSION_TAG_PREFIX = "delta-version-";
+ private static final String DELTA_TIMESTAMP_TAG_PREFIX = "delta-ts-";
+
+ private final ImmutableMap.Builder<String, String> icebergPropertiesBuilder =
+ ImmutableMap.builder();
+
+ private final String deltaTableLocation;
+ private Engine deltaEngine;
+ private TableImpl deltaTable;
+
+ private Catalog icebergCatalog;
+ private TableIdentifier newTableIdentifier;
+ private String newTableLocation;
+ private HadoopFileIO deltaLakeFileIO;
+ private DeletionVectorConverter deletionVectorConverter;
+ private OutputFileFactory icebergDVFileFactory;
+ private final Set<Long> deltaTimestampTags = Sets.newHashSet();
+
+ BaseSnapshotDeltaLakeKernelTableAction(String deltaTableLocation) {
+ this.deltaTableLocation = deltaTableLocation;
+ this.newTableLocation = deltaTableLocation;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperties(Map<String, String> properties) {
+ icebergPropertiesBuilder.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperty(String name, String value) {
+ icebergPropertiesBuilder.put(name, value);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableLocation(String location) {
+ newTableLocation = location;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable as(TableIdentifier identifier) {
+ newTableIdentifier = identifier;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable icebergCatalog(Catalog catalog) {
+ icebergCatalog = catalog;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable deltaLakeConfiguration(Configuration conf) {
+ deltaEngine = DefaultEngine.create(conf);
+ deltaLakeFileIO = new HadoopFileIO(conf);
+ deltaTable = (TableImpl) Table.forPath(deltaEngine, deltaTableLocation);
+ deletionVectorConverter = new DeletionVectorConverter(deltaEngine, deltaTableLocation);
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Preconditions.checkArgument(
+ icebergCatalog != null && newTableIdentifier != null,
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ Preconditions.checkArgument(
+ deltaTable != null && deltaLakeFileIO != null,
+ "Make sure to configure the action with a valid deltaLakeConfiguration");
+ assertDeltaColumnMappingDisabled(
+ "Conversion of Delta Lake tables with columnMapping feature is not supported.");
+
+ final long latestDeltaVersion = getLatestDeltaSnapshot().getVersion();
+ final long minimalAvailableDeltaVersion = getEarliestRecreatableDeltaLog();
+
+ SnapshotImpl initialDeltaSnapshot = getDeltaSnapshotAsOfVersion(minimalAvailableDeltaVersion);
+
+ LOG.info(
+ "Converting Delta Lake table at {} from version {} to version {} into Iceberg table {} ...",
+ deltaTableLocation,
+ minimalAvailableDeltaVersion,
+ latestDeltaVersion,
+ newTableIdentifier);
+
+ Schema icebergSchema = convertToIcebergSchema(initialDeltaSnapshot.getSchema());
+ PartitionSpec partitionSpec =
+ buildPartitionSpec(icebergSchema, initialDeltaSnapshot.getPartitionColumnNames());
+
+ Transaction transaction =
+ icebergCatalog.newCreateTableTransaction(
+ newTableIdentifier,
+ icebergSchema,
+ partitionSpec,
+ newTableLocation,
+ buildTablePropertiesWithDelta(initialDeltaSnapshot, deltaTableLocation));
+ setDefaultNamingMapping(transaction);
+ // Create an initial empty snapshot so currentSnapshot() is never null
+ transaction.newAppend().commit();
+
+ icebergDVFileFactory =
+ OutputFileFactory.builderFor(transaction.table(), 1, 1).format(FileFormat.PUFFIN).build();
+
+ Set<String> processedDataFiles = Sets.newHashSet();
+ try {
+ commitDeltaSnapshotToIcebergTransaction(
+ initialDeltaSnapshot, transaction, processedDataFiles);
+ convertEachDeltaVersion(
+ minimalAvailableDeltaVersion + 1, latestDeltaVersion, transaction, processedDataFiles);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ transaction.commitTransaction();
+
+ LOG.info(
+ "Successfully created Iceberg table {} from Delta Lake table at {}, processed data file count: {}",
+ newTableIdentifier,
+ deltaTableLocation,
+ processedDataFiles.size());
+ return processedDataFiles::size;
+ }
+
+ private void commitDeltaSnapshotToIcebergTransaction(
+ SnapshotImpl deltaSnapshot, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+ Scan scan = deltaSnapshot.getScanBuilder().build();
+ try (CloseableIterator<FilteredColumnarBatch> changes = scan.getScanFiles(deltaEngine)) {
+ while (changes.hasNext()) {
+ FilteredColumnarBatch columnarBatch = changes.next();
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch.getData(), transaction, processedDataFiles);
+ }
+ tagCurrentSnapshot(
+ deltaSnapshot.getVersion(), deltaSnapshot.getTimestamp(deltaEngine), transaction);
+ }
+ }
+
+ /**
+ * Current implementation uses the schema conversion mapping based on the logical names. Delta
+ * Lake supports three column mapping modes: none, name, id. So, the renames with columnMapping
+ * feature can lead to data corruption.
+ */
+ private void assertDeltaColumnMappingDisabled(String errorMessage) {
+ Map<String, String> configuration = getLatestDeltaSnapshot().getMetadata().getConfiguration();
+ String columnMappingMode = configuration.getOrDefault("delta.columnMapping.mode", "none");
+ if (!"none".equals(columnMappingMode)) {
+ throw new UnsupportedOperationException(errorMessage);
+ }
+ }
+
+ private static void setDefaultNamingMapping(Transaction transaction) {
+ transaction
+ .table()
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+ NameMappingParser.toJson(MappingUtil.create(transaction.table().schema())))
+ .commit();
+ }
+
+ private SnapshotImpl getDeltaSnapshotAsOfVersion(long deltaVersion) {
+ Snapshot snapshot =
+ Preconditions.checkNotNull(
+ deltaTable.getSnapshotAsOfVersion(deltaEngine, deltaVersion),
+ "Delta snapshot for version %s is unreachable.",
+ deltaVersion);
+ assertSnapshotImpl(snapshot);
+ return (SnapshotImpl) snapshot;
+ }
+
+ private void convertEachDeltaVersion(
+ long initialDeltaVersion,
+ long latestDeltaVersion,
+ Transaction transaction,
+ Set<String> processedDataFiles)
+ throws IOException {
+
+ for (long currDeltaVersion = initialDeltaVersion;
+ currDeltaVersion <= latestDeltaVersion;
+ currDeltaVersion++) {
+ try (CloseableIterator<ColumnarBatch> changes =
+ deltaTable.getChanges(
+ deltaEngine,
+ currDeltaVersion,
+ currDeltaVersion,
+ Set.of(DeltaLogActionUtils.DeltaAction.values()))) {
+ Long batchTimestamp = null;
+ while (changes.hasNext()) {
+ ColumnarBatch columnarBatch = changes.next();
+
+ Long commitTimestamp =
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch, transaction, processedDataFiles);
+ if (batchTimestamp == null) {
+ batchTimestamp = commitTimestamp;
+ }
+ }
+
+ tagCurrentSnapshot(currDeltaVersion, batchTimestamp, transaction);
+ }
+ }
+ }
+
+ /**
+ * Convert each delta log {@code ColumnarBatch} to Iceberg action and commit to the given {@code
+ * Transaction}. The complete <a
+ * href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Actions">spec</a> of delta
+ * actions. <br>
+ * Supported:
+ * <li>Add
+ *
+ * @return number of added data files
+ */
+ private Long commitDeltaColumnarBatchToIcebergTransaction(
+ ColumnarBatch columnarBatch, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+
+ Long originalCommitTimestamp = null;
+ List<DataFile> dataFilesToAdd = Lists.newArrayList();
+ List<DataFile> dataFilesToRemove = Lists.newArrayList();
+ List<DeleteFile> deleteFilesToAdd = Lists.newArrayList();
+
+ try (CloseableIterator<Row> rows = columnarBatch.getRows()) {
+ while (rows.hasNext()) {
+ Row row = rows.next();
+ if (DeltaLakeActionsTranslationUtil.isCommitInfo(row)) {
+ Row commitInfo = row.getStruct(row.getSchema().indexOf("commitInfo"));
+ originalCommitTimestamp = commitInfo.getLong(commitInfo.getSchema().indexOf("timestamp"));
+ } else if (DeltaLakeActionsTranslationUtil.isAdd(row)) {
+ AddFile addFile = DeltaLakeActionsTranslationUtil.toAdd(row);
+
+ DataFile dataFile = buildDataFileFromAddDeltaAction(addFile, transaction);
+ dataFilesToAdd.add(dataFile);
+
+ List<DeleteFile> deleteFiles =
+ convertDeltaDVsToIcebergDVs(transaction.table().spec(), addFile, dataFile);
+
+ deleteFilesToAdd.addAll(deleteFiles);
+ processedDataFiles.add(dataFile.location());
+ } else if (DeltaLakeActionsTranslationUtil.isRemove(row)) {
+ RemoveFile remove = DeltaLakeActionsTranslationUtil.toRemove(row);
+
+ DataFile dataFile = buildDataFileFromRemoveDeltaAction(remove, transaction);
+ dataFilesToRemove.add(dataFile);
+ processedDataFiles.add(dataFile.location());
+ }
+ }
+ }
+
+ // TODO support more actions
+ commitIcebergTransaction(transaction, dataFilesToAdd, dataFilesToRemove, deleteFilesToAdd);
+
+ return originalCommitTimestamp;
+ }
+
+ private List<DeleteFile> convertDeltaDVsToIcebergDVs(
+ PartitionSpec partitionSpec, AddFile addFile, DataFile dataFile) throws IOException {
+ if (addFile.getDeletionVector().isEmpty()) {
+ return List.of();
+ }
+
+ DVFileWriter dvWriter = new BaseDVFileWriter(icebergDVFileFactory, path -> null);
+ try (DVFileWriter closeableWriter = dvWriter) {
+ long[] positions =
+ deletionVectorConverter.readDeltaDVPositions(addFile.getDeletionVector().get());
+ for (long deletedRowIndex : positions) {
+ closeableWriter.delete(
+ dataFile.location(), deletedRowIndex, partitionSpec, dataFile.partition());
+ }
+ }
+
+ return dvWriter.result().deleteFiles();
+ }
+
+ /**
+ * CASES:
+ *
+ * <ol>
+ * <li>Append only
+ * <li>Delete only
+ * <li>Append and Delete => overwrite
+ * <li>RowDelta with deletes.
+ * <li>No Append, No Delete => No data changes, append tag or snapshot.
+ * </ol>
+ */
+ private static void commitIcebergTransaction(
+ Transaction transaction,
+ List<DataFile> dataFilesToAdd,
+ List<DataFile> dataFilesToRemove,
+ List<DeleteFile> deleteFilesToAdd) {
+ if (!deleteFilesToAdd.isEmpty()) {
+ // Row Delta
+ RowDelta rowDelta = transaction.newRowDelta();
+ // Avoid validation for multiple DVs added in transaction
+ // org/apache/iceberg/MergingSnapshotProducer.java:854
+ // since we do the conversion sequentially in a single Iceberg transaction
+ rowDelta.validateFromSnapshot(transaction.table().currentSnapshot().snapshotId());
Review Comment:
Setting the validate-from snapshot to currentSnapshot() disables RowDelta's conflict validation wholesale, which is what's masking the Add+Remove-same-path bug above in tests. Please remove this or scope it to the narrower concern (e.g. validateDataFilesExist on a specific set) and let the real validation run.
##########
delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeKernelTableAction.java:
##########
@@ -0,0 +1,595 @@
+/*
+ * 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.iceberg.delta;
+
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import io.delta.kernel.internal.DeltaHistoryManager;
+import io.delta.kernel.internal.DeltaLogActionUtils;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.TableImpl;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterator;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class BaseSnapshotDeltaLakeKernelTableAction implements SnapshotDeltaLakeTable {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(BaseSnapshotDeltaLakeKernelTableAction.class);
+
+ private static final String SNAPSHOT_SOURCE_PROP = "snapshot_source";
+ private static final String DELTA_SOURCE_VALUE = "delta";
+ private static final String ORIGINAL_LOCATION_PROP = "original_location";
+ private static final String DELTA_VERSION_TAG_PREFIX = "delta-version-";
+ private static final String DELTA_TIMESTAMP_TAG_PREFIX = "delta-ts-";
+
+ private final ImmutableMap.Builder<String, String> icebergPropertiesBuilder =
+ ImmutableMap.builder();
+
+ private final String deltaTableLocation;
+ private Engine deltaEngine;
+ private TableImpl deltaTable;
+
+ private Catalog icebergCatalog;
+ private TableIdentifier newTableIdentifier;
+ private String newTableLocation;
+ private HadoopFileIO deltaLakeFileIO;
+ private DeletionVectorConverter deletionVectorConverter;
+ private OutputFileFactory icebergDVFileFactory;
+ private final Set<Long> deltaTimestampTags = Sets.newHashSet();
+
+ BaseSnapshotDeltaLakeKernelTableAction(String deltaTableLocation) {
+ this.deltaTableLocation = deltaTableLocation;
+ this.newTableLocation = deltaTableLocation;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperties(Map<String, String> properties) {
+ icebergPropertiesBuilder.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperty(String name, String value) {
+ icebergPropertiesBuilder.put(name, value);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableLocation(String location) {
+ newTableLocation = location;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable as(TableIdentifier identifier) {
+ newTableIdentifier = identifier;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable icebergCatalog(Catalog catalog) {
+ icebergCatalog = catalog;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable deltaLakeConfiguration(Configuration conf) {
+ deltaEngine = DefaultEngine.create(conf);
+ deltaLakeFileIO = new HadoopFileIO(conf);
+ deltaTable = (TableImpl) Table.forPath(deltaEngine, deltaTableLocation);
+ deletionVectorConverter = new DeletionVectorConverter(deltaEngine, deltaTableLocation);
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Preconditions.checkArgument(
+ icebergCatalog != null && newTableIdentifier != null,
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ Preconditions.checkArgument(
+ deltaTable != null && deltaLakeFileIO != null,
+ "Make sure to configure the action with a valid deltaLakeConfiguration");
+ assertDeltaColumnMappingDisabled(
+ "Conversion of Delta Lake tables with columnMapping feature is not supported.");
+
+ final long latestDeltaVersion = getLatestDeltaSnapshot().getVersion();
+ final long minimalAvailableDeltaVersion = getEarliestRecreatableDeltaLog();
+
+ SnapshotImpl initialDeltaSnapshot = getDeltaSnapshotAsOfVersion(minimalAvailableDeltaVersion);
+
+ LOG.info(
+ "Converting Delta Lake table at {} from version {} to version {} into Iceberg table {} ...",
+ deltaTableLocation,
+ minimalAvailableDeltaVersion,
+ latestDeltaVersion,
+ newTableIdentifier);
+
+ Schema icebergSchema = convertToIcebergSchema(initialDeltaSnapshot.getSchema());
+ PartitionSpec partitionSpec =
+ buildPartitionSpec(icebergSchema, initialDeltaSnapshot.getPartitionColumnNames());
+
+ Transaction transaction =
+ icebergCatalog.newCreateTableTransaction(
+ newTableIdentifier,
+ icebergSchema,
+ partitionSpec,
+ newTableLocation,
+ buildTablePropertiesWithDelta(initialDeltaSnapshot, deltaTableLocation));
+ setDefaultNamingMapping(transaction);
+ // Create an initial empty snapshot so currentSnapshot() is never null
+ transaction.newAppend().commit();
+
+ icebergDVFileFactory =
+ OutputFileFactory.builderFor(transaction.table(), 1, 1).format(FileFormat.PUFFIN).build();
+
+ Set<String> processedDataFiles = Sets.newHashSet();
+ try {
+ commitDeltaSnapshotToIcebergTransaction(
+ initialDeltaSnapshot, transaction, processedDataFiles);
+ convertEachDeltaVersion(
+ minimalAvailableDeltaVersion + 1, latestDeltaVersion, transaction, processedDataFiles);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ transaction.commitTransaction();
+
+ LOG.info(
+ "Successfully created Iceberg table {} from Delta Lake table at {}, processed data file count: {}",
+ newTableIdentifier,
+ deltaTableLocation,
+ processedDataFiles.size());
+ return processedDataFiles::size;
+ }
+
+ private void commitDeltaSnapshotToIcebergTransaction(
+ SnapshotImpl deltaSnapshot, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+ Scan scan = deltaSnapshot.getScanBuilder().build();
+ try (CloseableIterator<FilteredColumnarBatch> changes = scan.getScanFiles(deltaEngine)) {
+ while (changes.hasNext()) {
+ FilteredColumnarBatch columnarBatch = changes.next();
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch.getData(), transaction, processedDataFiles);
+ }
+ tagCurrentSnapshot(
+ deltaSnapshot.getVersion(), deltaSnapshot.getTimestamp(deltaEngine), transaction);
+ }
+ }
+
+ /**
+ * Current implementation uses the schema conversion mapping based on the logical names. Delta
+ * Lake supports three column mapping modes: none, name, id. So, the renames with columnMapping
+ * feature can lead to data corruption.
+ */
+ private void assertDeltaColumnMappingDisabled(String errorMessage) {
+ Map<String, String> configuration = getLatestDeltaSnapshot().getMetadata().getConfiguration();
+ String columnMappingMode = configuration.getOrDefault("delta.columnMapping.mode", "none");
+ if (!"none".equals(columnMappingMode)) {
+ throw new UnsupportedOperationException(errorMessage);
+ }
+ }
+
+ private static void setDefaultNamingMapping(Transaction transaction) {
+ transaction
+ .table()
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+ NameMappingParser.toJson(MappingUtil.create(transaction.table().schema())))
+ .commit();
+ }
+
+ private SnapshotImpl getDeltaSnapshotAsOfVersion(long deltaVersion) {
+ Snapshot snapshot =
+ Preconditions.checkNotNull(
+ deltaTable.getSnapshotAsOfVersion(deltaEngine, deltaVersion),
+ "Delta snapshot for version %s is unreachable.",
+ deltaVersion);
+ assertSnapshotImpl(snapshot);
+ return (SnapshotImpl) snapshot;
+ }
+
+ private void convertEachDeltaVersion(
+ long initialDeltaVersion,
+ long latestDeltaVersion,
+ Transaction transaction,
+ Set<String> processedDataFiles)
+ throws IOException {
+
+ for (long currDeltaVersion = initialDeltaVersion;
+ currDeltaVersion <= latestDeltaVersion;
+ currDeltaVersion++) {
+ try (CloseableIterator<ColumnarBatch> changes =
+ deltaTable.getChanges(
+ deltaEngine,
+ currDeltaVersion,
+ currDeltaVersion,
+ Set.of(DeltaLogActionUtils.DeltaAction.values()))) {
+ Long batchTimestamp = null;
+ while (changes.hasNext()) {
+ ColumnarBatch columnarBatch = changes.next();
+
+ Long commitTimestamp =
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch, transaction, processedDataFiles);
+ if (batchTimestamp == null) {
+ batchTimestamp = commitTimestamp;
+ }
+ }
+
+ tagCurrentSnapshot(currDeltaVersion, batchTimestamp, transaction);
+ }
+ }
+ }
+
+ /**
+ * Convert each delta log {@code ColumnarBatch} to Iceberg action and commit to the given {@code
+ * Transaction}. The complete <a
+ * href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Actions">spec</a> of delta
+ * actions. <br>
+ * Supported:
+ * <li>Add
+ *
+ * @return number of added data files
+ */
+ private Long commitDeltaColumnarBatchToIcebergTransaction(
+ ColumnarBatch columnarBatch, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+
+ Long originalCommitTimestamp = null;
+ List<DataFile> dataFilesToAdd = Lists.newArrayList();
+ List<DataFile> dataFilesToRemove = Lists.newArrayList();
+ List<DeleteFile> deleteFilesToAdd = Lists.newArrayList();
+
+ try (CloseableIterator<Row> rows = columnarBatch.getRows()) {
+ while (rows.hasNext()) {
+ Row row = rows.next();
+ if (DeltaLakeActionsTranslationUtil.isCommitInfo(row)) {
+ Row commitInfo = row.getStruct(row.getSchema().indexOf("commitInfo"));
+ originalCommitTimestamp = commitInfo.getLong(commitInfo.getSchema().indexOf("timestamp"));
+ } else if (DeltaLakeActionsTranslationUtil.isAdd(row)) {
+ AddFile addFile = DeltaLakeActionsTranslationUtil.toAdd(row);
+
+ DataFile dataFile = buildDataFileFromAddDeltaAction(addFile, transaction);
+ dataFilesToAdd.add(dataFile);
+
+ List<DeleteFile> deleteFiles =
+ convertDeltaDVsToIcebergDVs(transaction.table().spec(), addFile, dataFile);
+
+ deleteFilesToAdd.addAll(deleteFiles);
+ processedDataFiles.add(dataFile.location());
+ } else if (DeltaLakeActionsTranslationUtil.isRemove(row)) {
+ RemoveFile remove = DeltaLakeActionsTranslationUtil.toRemove(row);
+
+ DataFile dataFile = buildDataFileFromRemoveDeltaAction(remove, transaction);
+ dataFilesToRemove.add(dataFile);
+ processedDataFiles.add(dataFile.location());
+ }
+ }
+ }
+
+ // TODO support more actions
+ commitIcebergTransaction(transaction, dataFilesToAdd, dataFilesToRemove, deleteFilesToAdd);
+
+ return originalCommitTimestamp;
+ }
+
+ private List<DeleteFile> convertDeltaDVsToIcebergDVs(
+ PartitionSpec partitionSpec, AddFile addFile, DataFile dataFile) throws IOException {
+ if (addFile.getDeletionVector().isEmpty()) {
+ return List.of();
+ }
+
+ DVFileWriter dvWriter = new BaseDVFileWriter(icebergDVFileFactory, path -> null);
+ try (DVFileWriter closeableWriter = dvWriter) {
+ long[] positions =
+ deletionVectorConverter.readDeltaDVPositions(addFile.getDeletionVector().get());
+ for (long deletedRowIndex : positions) {
+ closeableWriter.delete(
+ dataFile.location(), deletedRowIndex, partitionSpec, dataFile.partition());
+ }
+ }
+
+ return dvWriter.result().deleteFiles();
+ }
+
+ /**
+ * CASES:
+ *
+ * <ol>
+ * <li>Append only
+ * <li>Delete only
+ * <li>Append and Delete => overwrite
+ * <li>RowDelta with deletes.
+ * <li>No Append, No Delete => No data changes, append tag or snapshot.
+ * </ol>
+ */
+ private static void commitIcebergTransaction(
+ Transaction transaction,
+ List<DataFile> dataFilesToAdd,
+ List<DataFile> dataFilesToRemove,
+ List<DeleteFile> deleteFilesToAdd) {
+ if (!deleteFilesToAdd.isEmpty()) {
+ // Row Delta
+ RowDelta rowDelta = transaction.newRowDelta();
+ // Avoid validation for multiple DVs added in transaction
+ // org/apache/iceberg/MergingSnapshotProducer.java:854
+ // since we do the conversion sequentially in a single Iceberg transaction
+ rowDelta.validateFromSnapshot(transaction.table().currentSnapshot().snapshotId());
+
+ dataFilesToAdd.forEach(rowDelta::addRows);
+ dataFilesToRemove.forEach(rowDelta::removeRows);
+ deleteFilesToAdd.forEach(rowDelta::addDeletes);
+ rowDelta.commit();
Review Comment:
Per Delta PROTOCOL.md, a DV update emits Add(P, dv=new) + Remove(P, dv=old) for the same path in the same commit. Here that lands as rowDelta.addRows(P) + removeRows(P) + addDeletes(puffin_for_P). This either fails validation or produces a snapshot where the file is added and removed simultaneously and the new DV dangles.
Additionally, prior Puffin DVs written for earlier versions of P are never retired — there is no removeDeletes(...). Please detect same-path Add+Remove pairs and emit a DV-only update (addDeletes + removeDeletes) without addRows/removeRows. This is the central DELETE/MERGE/UPDATE path on DV-enabled Delta tables and is currently incorrect.
##########
delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeKernelTableAction.java:
##########
@@ -0,0 +1,595 @@
+/*
+ * 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.iceberg.delta;
+
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import io.delta.kernel.internal.DeltaHistoryManager;
+import io.delta.kernel.internal.DeltaLogActionUtils;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.TableImpl;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterator;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class BaseSnapshotDeltaLakeKernelTableAction implements SnapshotDeltaLakeTable {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(BaseSnapshotDeltaLakeKernelTableAction.class);
+
+ private static final String SNAPSHOT_SOURCE_PROP = "snapshot_source";
+ private static final String DELTA_SOURCE_VALUE = "delta";
+ private static final String ORIGINAL_LOCATION_PROP = "original_location";
+ private static final String DELTA_VERSION_TAG_PREFIX = "delta-version-";
+ private static final String DELTA_TIMESTAMP_TAG_PREFIX = "delta-ts-";
+
+ private final ImmutableMap.Builder<String, String> icebergPropertiesBuilder =
+ ImmutableMap.builder();
+
+ private final String deltaTableLocation;
+ private Engine deltaEngine;
+ private TableImpl deltaTable;
+
+ private Catalog icebergCatalog;
+ private TableIdentifier newTableIdentifier;
+ private String newTableLocation;
+ private HadoopFileIO deltaLakeFileIO;
+ private DeletionVectorConverter deletionVectorConverter;
+ private OutputFileFactory icebergDVFileFactory;
+ private final Set<Long> deltaTimestampTags = Sets.newHashSet();
+
+ BaseSnapshotDeltaLakeKernelTableAction(String deltaTableLocation) {
+ this.deltaTableLocation = deltaTableLocation;
+ this.newTableLocation = deltaTableLocation;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperties(Map<String, String> properties) {
+ icebergPropertiesBuilder.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperty(String name, String value) {
+ icebergPropertiesBuilder.put(name, value);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableLocation(String location) {
+ newTableLocation = location;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable as(TableIdentifier identifier) {
+ newTableIdentifier = identifier;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable icebergCatalog(Catalog catalog) {
+ icebergCatalog = catalog;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable deltaLakeConfiguration(Configuration conf) {
+ deltaEngine = DefaultEngine.create(conf);
+ deltaLakeFileIO = new HadoopFileIO(conf);
+ deltaTable = (TableImpl) Table.forPath(deltaEngine, deltaTableLocation);
+ deletionVectorConverter = new DeletionVectorConverter(deltaEngine, deltaTableLocation);
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Preconditions.checkArgument(
+ icebergCatalog != null && newTableIdentifier != null,
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ Preconditions.checkArgument(
+ deltaTable != null && deltaLakeFileIO != null,
+ "Make sure to configure the action with a valid deltaLakeConfiguration");
+ assertDeltaColumnMappingDisabled(
+ "Conversion of Delta Lake tables with columnMapping feature is not supported.");
+
+ final long latestDeltaVersion = getLatestDeltaSnapshot().getVersion();
+ final long minimalAvailableDeltaVersion = getEarliestRecreatableDeltaLog();
+
+ SnapshotImpl initialDeltaSnapshot = getDeltaSnapshotAsOfVersion(minimalAvailableDeltaVersion);
+
+ LOG.info(
+ "Converting Delta Lake table at {} from version {} to version {} into Iceberg table {} ...",
+ deltaTableLocation,
+ minimalAvailableDeltaVersion,
+ latestDeltaVersion,
+ newTableIdentifier);
+
+ Schema icebergSchema = convertToIcebergSchema(initialDeltaSnapshot.getSchema());
+ PartitionSpec partitionSpec =
+ buildPartitionSpec(icebergSchema, initialDeltaSnapshot.getPartitionColumnNames());
+
+ Transaction transaction =
+ icebergCatalog.newCreateTableTransaction(
+ newTableIdentifier,
+ icebergSchema,
+ partitionSpec,
+ newTableLocation,
+ buildTablePropertiesWithDelta(initialDeltaSnapshot, deltaTableLocation));
+ setDefaultNamingMapping(transaction);
+ // Create an initial empty snapshot so currentSnapshot() is never null
+ transaction.newAppend().commit();
+
+ icebergDVFileFactory =
+ OutputFileFactory.builderFor(transaction.table(), 1, 1).format(FileFormat.PUFFIN).build();
+
+ Set<String> processedDataFiles = Sets.newHashSet();
+ try {
+ commitDeltaSnapshotToIcebergTransaction(
+ initialDeltaSnapshot, transaction, processedDataFiles);
+ convertEachDeltaVersion(
+ minimalAvailableDeltaVersion + 1, latestDeltaVersion, transaction, processedDataFiles);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ transaction.commitTransaction();
+
+ LOG.info(
+ "Successfully created Iceberg table {} from Delta Lake table at {}, processed data file count: {}",
+ newTableIdentifier,
+ deltaTableLocation,
+ processedDataFiles.size());
+ return processedDataFiles::size;
+ }
+
+ private void commitDeltaSnapshotToIcebergTransaction(
+ SnapshotImpl deltaSnapshot, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+ Scan scan = deltaSnapshot.getScanBuilder().build();
+ try (CloseableIterator<FilteredColumnarBatch> changes = scan.getScanFiles(deltaEngine)) {
+ while (changes.hasNext()) {
+ FilteredColumnarBatch columnarBatch = changes.next();
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch.getData(), transaction, processedDataFiles);
+ }
+ tagCurrentSnapshot(
+ deltaSnapshot.getVersion(), deltaSnapshot.getTimestamp(deltaEngine), transaction);
+ }
+ }
+
+ /**
+ * Current implementation uses the schema conversion mapping based on the logical names. Delta
+ * Lake supports three column mapping modes: none, name, id. So, the renames with columnMapping
+ * feature can lead to data corruption.
+ */
+ private void assertDeltaColumnMappingDisabled(String errorMessage) {
+ Map<String, String> configuration = getLatestDeltaSnapshot().getMetadata().getConfiguration();
+ String columnMappingMode = configuration.getOrDefault("delta.columnMapping.mode", "none");
+ if (!"none".equals(columnMappingMode)) {
+ throw new UnsupportedOperationException(errorMessage);
+ }
+ }
+
+ private static void setDefaultNamingMapping(Transaction transaction) {
+ transaction
+ .table()
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+ NameMappingParser.toJson(MappingUtil.create(transaction.table().schema())))
+ .commit();
+ }
+
+ private SnapshotImpl getDeltaSnapshotAsOfVersion(long deltaVersion) {
+ Snapshot snapshot =
+ Preconditions.checkNotNull(
+ deltaTable.getSnapshotAsOfVersion(deltaEngine, deltaVersion),
+ "Delta snapshot for version %s is unreachable.",
+ deltaVersion);
+ assertSnapshotImpl(snapshot);
+ return (SnapshotImpl) snapshot;
+ }
+
+ private void convertEachDeltaVersion(
+ long initialDeltaVersion,
+ long latestDeltaVersion,
+ Transaction transaction,
+ Set<String> processedDataFiles)
+ throws IOException {
+
+ for (long currDeltaVersion = initialDeltaVersion;
+ currDeltaVersion <= latestDeltaVersion;
+ currDeltaVersion++) {
+ try (CloseableIterator<ColumnarBatch> changes =
+ deltaTable.getChanges(
+ deltaEngine,
+ currDeltaVersion,
+ currDeltaVersion,
+ Set.of(DeltaLogActionUtils.DeltaAction.values()))) {
+ Long batchTimestamp = null;
+ while (changes.hasNext()) {
+ ColumnarBatch columnarBatch = changes.next();
+
+ Long commitTimestamp =
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch, transaction, processedDataFiles);
+ if (batchTimestamp == null) {
+ batchTimestamp = commitTimestamp;
+ }
+ }
Review Comment:
Each ColumnarBatch triggers its own newAppend/newOverwrite/newRowDelta.commit().
A single Delta version can span multiple batches; an Add in batch 1 and its matching Remove in batch 2 produce two separate Iceberg snapshots with different sequence numbers (so DVs attach to the wrong add), and only the LAST snapshot gets the delta-version-N tag — time-travel by tag will see partial state. Please buffer all actions for a version across batches and emit one Iceberg commit per Delta version.
##########
delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeKernelTableAction.java:
##########
@@ -0,0 +1,595 @@
+/*
+ * 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.iceberg.delta;
+
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.data.ColumnarBatch;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import io.delta.kernel.internal.DeltaHistoryManager;
+import io.delta.kernel.internal.DeltaLogActionUtils;
+import io.delta.kernel.internal.SnapshotImpl;
+import io.delta.kernel.internal.TableImpl;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+import io.delta.kernel.internal.fs.Path;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterator;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nonnull;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.ManageSnapshots;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.deletes.BaseDVFileWriter;
+import org.apache.iceberg.deletes.DVFileWriter;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.parquet.ParquetUtil;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class BaseSnapshotDeltaLakeKernelTableAction implements SnapshotDeltaLakeTable {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(BaseSnapshotDeltaLakeKernelTableAction.class);
+
+ private static final String SNAPSHOT_SOURCE_PROP = "snapshot_source";
+ private static final String DELTA_SOURCE_VALUE = "delta";
+ private static final String ORIGINAL_LOCATION_PROP = "original_location";
+ private static final String DELTA_VERSION_TAG_PREFIX = "delta-version-";
+ private static final String DELTA_TIMESTAMP_TAG_PREFIX = "delta-ts-";
+
+ private final ImmutableMap.Builder<String, String> icebergPropertiesBuilder =
+ ImmutableMap.builder();
+
+ private final String deltaTableLocation;
+ private Engine deltaEngine;
+ private TableImpl deltaTable;
+
+ private Catalog icebergCatalog;
+ private TableIdentifier newTableIdentifier;
+ private String newTableLocation;
+ private HadoopFileIO deltaLakeFileIO;
+ private DeletionVectorConverter deletionVectorConverter;
+ private OutputFileFactory icebergDVFileFactory;
+ private final Set<Long> deltaTimestampTags = Sets.newHashSet();
+
+ BaseSnapshotDeltaLakeKernelTableAction(String deltaTableLocation) {
+ this.deltaTableLocation = deltaTableLocation;
+ this.newTableLocation = deltaTableLocation;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperties(Map<String, String> properties) {
+ icebergPropertiesBuilder.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableProperty(String name, String value) {
+ icebergPropertiesBuilder.put(name, value);
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable tableLocation(String location) {
+ newTableLocation = location;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable as(TableIdentifier identifier) {
+ newTableIdentifier = identifier;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable icebergCatalog(Catalog catalog) {
+ icebergCatalog = catalog;
+ return this;
+ }
+
+ @Override
+ public SnapshotDeltaLakeTable deltaLakeConfiguration(Configuration conf) {
+ deltaEngine = DefaultEngine.create(conf);
+ deltaLakeFileIO = new HadoopFileIO(conf);
+ deltaTable = (TableImpl) Table.forPath(deltaEngine, deltaTableLocation);
+ deletionVectorConverter = new DeletionVectorConverter(deltaEngine, deltaTableLocation);
+ return this;
+ }
+
+ @Override
+ public Result execute() {
+ Preconditions.checkArgument(
+ icebergCatalog != null && newTableIdentifier != null,
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ Preconditions.checkArgument(
+ deltaTable != null && deltaLakeFileIO != null,
+ "Make sure to configure the action with a valid deltaLakeConfiguration");
+ assertDeltaColumnMappingDisabled(
+ "Conversion of Delta Lake tables with columnMapping feature is not supported.");
+
+ final long latestDeltaVersion = getLatestDeltaSnapshot().getVersion();
+ final long minimalAvailableDeltaVersion = getEarliestRecreatableDeltaLog();
+
+ SnapshotImpl initialDeltaSnapshot = getDeltaSnapshotAsOfVersion(minimalAvailableDeltaVersion);
+
+ LOG.info(
+ "Converting Delta Lake table at {} from version {} to version {} into Iceberg table {} ...",
+ deltaTableLocation,
+ minimalAvailableDeltaVersion,
+ latestDeltaVersion,
+ newTableIdentifier);
+
+ Schema icebergSchema = convertToIcebergSchema(initialDeltaSnapshot.getSchema());
+ PartitionSpec partitionSpec =
+ buildPartitionSpec(icebergSchema, initialDeltaSnapshot.getPartitionColumnNames());
+
+ Transaction transaction =
+ icebergCatalog.newCreateTableTransaction(
+ newTableIdentifier,
+ icebergSchema,
+ partitionSpec,
+ newTableLocation,
+ buildTablePropertiesWithDelta(initialDeltaSnapshot, deltaTableLocation));
+ setDefaultNamingMapping(transaction);
+ // Create an initial empty snapshot so currentSnapshot() is never null
+ transaction.newAppend().commit();
+
+ icebergDVFileFactory =
+ OutputFileFactory.builderFor(transaction.table(), 1, 1).format(FileFormat.PUFFIN).build();
+
+ Set<String> processedDataFiles = Sets.newHashSet();
+ try {
+ commitDeltaSnapshotToIcebergTransaction(
+ initialDeltaSnapshot, transaction, processedDataFiles);
+ convertEachDeltaVersion(
+ minimalAvailableDeltaVersion + 1, latestDeltaVersion, transaction, processedDataFiles);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ transaction.commitTransaction();
+
+ LOG.info(
+ "Successfully created Iceberg table {} from Delta Lake table at {}, processed data file count: {}",
+ newTableIdentifier,
+ deltaTableLocation,
+ processedDataFiles.size());
+ return processedDataFiles::size;
+ }
+
+ private void commitDeltaSnapshotToIcebergTransaction(
+ SnapshotImpl deltaSnapshot, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+ Scan scan = deltaSnapshot.getScanBuilder().build();
+ try (CloseableIterator<FilteredColumnarBatch> changes = scan.getScanFiles(deltaEngine)) {
+ while (changes.hasNext()) {
+ FilteredColumnarBatch columnarBatch = changes.next();
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch.getData(), transaction, processedDataFiles);
+ }
+ tagCurrentSnapshot(
+ deltaSnapshot.getVersion(), deltaSnapshot.getTimestamp(deltaEngine), transaction);
+ }
+ }
+
+ /**
+ * Current implementation uses the schema conversion mapping based on the logical names. Delta
+ * Lake supports three column mapping modes: none, name, id. So, the renames with columnMapping
+ * feature can lead to data corruption.
+ */
+ private void assertDeltaColumnMappingDisabled(String errorMessage) {
+ Map<String, String> configuration = getLatestDeltaSnapshot().getMetadata().getConfiguration();
+ String columnMappingMode = configuration.getOrDefault("delta.columnMapping.mode", "none");
+ if (!"none".equals(columnMappingMode)) {
+ throw new UnsupportedOperationException(errorMessage);
+ }
+ }
+
+ private static void setDefaultNamingMapping(Transaction transaction) {
+ transaction
+ .table()
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
+ NameMappingParser.toJson(MappingUtil.create(transaction.table().schema())))
+ .commit();
+ }
+
+ private SnapshotImpl getDeltaSnapshotAsOfVersion(long deltaVersion) {
+ Snapshot snapshot =
+ Preconditions.checkNotNull(
+ deltaTable.getSnapshotAsOfVersion(deltaEngine, deltaVersion),
+ "Delta snapshot for version %s is unreachable.",
+ deltaVersion);
+ assertSnapshotImpl(snapshot);
+ return (SnapshotImpl) snapshot;
+ }
+
+ private void convertEachDeltaVersion(
+ long initialDeltaVersion,
+ long latestDeltaVersion,
+ Transaction transaction,
+ Set<String> processedDataFiles)
+ throws IOException {
+
+ for (long currDeltaVersion = initialDeltaVersion;
+ currDeltaVersion <= latestDeltaVersion;
+ currDeltaVersion++) {
+ try (CloseableIterator<ColumnarBatch> changes =
+ deltaTable.getChanges(
+ deltaEngine,
+ currDeltaVersion,
+ currDeltaVersion,
+ Set.of(DeltaLogActionUtils.DeltaAction.values()))) {
+ Long batchTimestamp = null;
+ while (changes.hasNext()) {
+ ColumnarBatch columnarBatch = changes.next();
+
+ Long commitTimestamp =
+ commitDeltaColumnarBatchToIcebergTransaction(
+ columnarBatch, transaction, processedDataFiles);
+ if (batchTimestamp == null) {
+ batchTimestamp = commitTimestamp;
+ }
+ }
+
+ tagCurrentSnapshot(currDeltaVersion, batchTimestamp, transaction);
+ }
+ }
+ }
+
+ /**
+ * Convert each delta log {@code ColumnarBatch} to Iceberg action and commit to the given {@code
+ * Transaction}. The complete <a
+ * href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Actions">spec</a> of delta
+ * actions. <br>
+ * Supported:
+ * <li>Add
+ *
+ * @return number of added data files
+ */
+ private Long commitDeltaColumnarBatchToIcebergTransaction(
+ ColumnarBatch columnarBatch, Transaction transaction, Set<String> processedDataFiles)
+ throws IOException {
+
+ Long originalCommitTimestamp = null;
+ List<DataFile> dataFilesToAdd = Lists.newArrayList();
+ List<DataFile> dataFilesToRemove = Lists.newArrayList();
+ List<DeleteFile> deleteFilesToAdd = Lists.newArrayList();
+
+ try (CloseableIterator<Row> rows = columnarBatch.getRows()) {
+ while (rows.hasNext()) {
+ Row row = rows.next();
+ if (DeltaLakeActionsTranslationUtil.isCommitInfo(row)) {
+ Row commitInfo = row.getStruct(row.getSchema().indexOf("commitInfo"));
+ originalCommitTimestamp = commitInfo.getLong(commitInfo.getSchema().indexOf("timestamp"));
+ } else if (DeltaLakeActionsTranslationUtil.isAdd(row)) {
+ AddFile addFile = DeltaLakeActionsTranslationUtil.toAdd(row);
+
+ DataFile dataFile = buildDataFileFromAddDeltaAction(addFile, transaction);
+ dataFilesToAdd.add(dataFile);
+
+ List<DeleteFile> deleteFiles =
+ convertDeltaDVsToIcebergDVs(transaction.table().spec(), addFile, dataFile);
+
+ deleteFilesToAdd.addAll(deleteFiles);
+ processedDataFiles.add(dataFile.location());
+ } else if (DeltaLakeActionsTranslationUtil.isRemove(row)) {
+ RemoveFile remove = DeltaLakeActionsTranslationUtil.toRemove(row);
+
+ DataFile dataFile = buildDataFileFromRemoveDeltaAction(remove, transaction);
+ dataFilesToRemove.add(dataFile);
+ processedDataFiles.add(dataFile.location());
+ }
+ }
+ }
Review Comment:
Only commitInfo, add, and remove are handled. metaData, protocol, cdc, txn are silently skipped even though DeltaLakeActionsTranslationUtil exposes isMetaData/isProtocol/isCdc/isTxn helpers (dead code today).
A mid-history metaData (ADD COLUMN, type widen) leaves Iceberg pinned to the earliest-recreatable schema while later AddFiles reference data written under a newer parquet schema → mis-mapped columns and wrong metrics at read time.
At minimum, fail fast when these appear at version > minimalAvailableDeltaVersion.
##########
delta-lake/src/test/java/org/apache/iceberg/delta/TestBaseSnapshotDeltaLakeKernelTableAction.java:
##########
@@ -0,0 +1,383 @@
+/*
+ * 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.iceberg.delta;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import io.delta.kernel.Operation;
+import io.delta.kernel.Table;
+import io.delta.kernel.Transaction;
+import io.delta.kernel.TransactionBuilder;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterable;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class TestBaseSnapshotDeltaLakeKernelTableAction {
+ private static final String DELTA_GOLDEN_TABLES_ROOT = "delta/golden/";
+
+ @TempDir private File sourceDeltaFolder;
+ @TempDir private File icebergWarehouseFolder;
+ private String sourceTableLocation;
+ private final Configuration testHadoopConf = new Configuration();
+ private String newTableLocation;
+ private Catalog testCatalog;
+
+ @BeforeEach
+ public void before() throws IOException {
+ sourceTableLocation = sourceDeltaFolder.toURI().toString();
+
+ Path icebergTablePath = icebergWarehouseFolder.toPath().resolve("iceberg_table");
+ newTableLocation = icebergTablePath.toString();
+ Files.createDirectories(icebergTablePath);
+
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(CatalogProperties.WAREHOUSE_LOCATION, icebergWarehouseFolder.toString());
+
+ HadoopCatalog icebergCatalog = new HadoopCatalog();
+ icebergCatalog.setConf(new Configuration());
+ icebergCatalog.initialize("hadoop_catalog", properties);
+ testCatalog = icebergCatalog;
+ }
+
+ @Test
+ public void testRequiredTableIdentifier() {
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .icebergCatalog(testCatalog)
+ .deltaLakeConfiguration(testHadoopConf)
+ .tableLocation(newTableLocation);
+ assertThatThrownBy(testAction::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ }
+
+ @Test
+ public void testRequiredIcebergCatalog() {
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(TableIdentifier.of("test", "test"))
+ .deltaLakeConfiguration(testHadoopConf)
+ .tableLocation(newTableLocation);
+ assertThatThrownBy(testAction::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Iceberg catalog and identifier cannot be null. Make sure to configure the action with a valid Iceberg catalog and identifier.");
+ }
+
+ @Test
+ public void testRequiredDeltaLakeConfiguration() {
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(TableIdentifier.of("test", "test"))
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+ assertThatThrownBy(testAction::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Make sure to configure the action with a valid deltaLakeConfiguration");
+ }
+
+ @Test
+ public void testDeltaTableColumnMappingFeatureDisabled() throws Exception {
+ Engine engine = DefaultEngine.create(testHadoopConf);
+ Transaction txn =
+ createEmptyDeltaTableTransaction(engine)
+ .withTableProperties(engine, Map.of("delta.columnMapping.mode", "name"))
+ .build(engine);
+ txn.commit(engine, CloseableIterable.emptyIterable());
+
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(TableIdentifier.of("iceberg_table"))
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+
+ // Act & check
+ assertThatThrownBy(testAction::execute)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Conversion of Delta Lake tables with columnMapping feature is not supported.");
+ }
+
+ @Test
+ public void testEmptyTableConversion() {
+ Engine engine = DefaultEngine.create(testHadoopConf);
+ createEmptyDeltaTableTransaction(engine)
+ .build(engine)
+ .commit(engine, CloseableIterable.emptyIterable());
+
+ TableIdentifier icebergTable = TableIdentifier.of("iceberg_table");
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(icebergTable)
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+
+ assertThat(testCatalog.tableExists(icebergTable)).isFalse();
+
+ testAction.execute();
+
+ assertThat(testCatalog.tableExists(icebergTable)).isTrue();
+
+ org.apache.iceberg.Table table = testCatalog.loadTable(icebergTable);
+ assertThat(((BaseTable) table).operations().current().formatVersion()).isEqualTo(3);
+ }
+
+ @Test
+ public void testTableCreated() throws Exception {
+ loadDeltaLakeGoldenTable("basic-decimal-table");
+
+ TableIdentifier icebergTable = TableIdentifier.of("iceberg_table");
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(icebergTable)
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+
+ assertThat(testCatalog.tableExists(icebergTable)).isFalse();
+
+ testAction.execute();
+
+ assertThat(testCatalog.tableExists(icebergTable)).isTrue();
+ }
+
+ @Test
+ public void testDefaultTableProperties() throws Exception {
+ loadDeltaLakeGoldenTable("basic-decimal-table");
+
+ TableIdentifier icebergTableIdentifier = TableIdentifier.of("iceberg_table");
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(icebergTableIdentifier)
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+
+ assertThat(testCatalog.tableExists(icebergTableIdentifier)).isFalse();
+
+ testAction.execute();
+
+ assertThat(testCatalog.tableExists(icebergTableIdentifier)).isTrue();
+
+ Map<String, String> properties = testCatalog.loadTable(icebergTableIdentifier).properties();
+ assertThat(properties.get("original_location")).isEqualTo(sourceTableLocation);
+ assertThat(properties.get("snapshot_source")).isEqualTo("delta");
+ assertThat(properties.containsKey(TableProperties.DEFAULT_NAME_MAPPING)).isTrue();
+ }
+
+ @Test
+ public void testCustomTableProperties() throws Exception {
+ loadDeltaLakeGoldenTable("basic-decimal-table");
+
+ TableIdentifier icebergTableIdentifier = TableIdentifier.of("iceberg_table");
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(icebergTableIdentifier)
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation)
+ .tableProperty("custom_prop_1", "custom val 1")
+ .tableProperty("custom_prop_2", "custom val 2")
+ .tableProperties(
+ ImmutableMap.of(
+ "custom_map_prop1", "val 1",
+ "custom_map_prop2", "val 2"));
+
+ assertThat(testCatalog.tableExists(icebergTableIdentifier)).isFalse();
+
+ testAction.execute();
+
+ assertThat(testCatalog.tableExists(icebergTableIdentifier)).isTrue();
+
+ Map<String, String> properties = testCatalog.loadTable(icebergTableIdentifier).properties();
+ assertThat(properties.get("custom_prop_1")).isEqualTo("custom val 1");
+ assertThat(properties.get("custom_prop_2")).isEqualTo("custom val 2");
+ assertThat(properties.get("custom_map_prop1")).isEqualTo("val 1");
+ assertThat(properties.get("custom_map_prop2")).isEqualTo("val 2");
+ }
+
+ @Test
+ void testDeltaTableNotExist() {
+ SnapshotDeltaLakeTable testAction =
+ new BaseSnapshotDeltaLakeKernelTableAction(sourceTableLocation)
+ .as(TableIdentifier.of("test", "test"))
+ .deltaLakeConfiguration(testHadoopConf)
+ .icebergCatalog(testCatalog)
+ .tableLocation(newTableLocation);
+
+ assertThatThrownBy(testAction::execute)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Delta Lake table does not exist at the given location: %s", sourceTableLocation);
+ }
+
+ @Test
+ public void testDeltaTableDVSupported() throws Exception {
Review Comment:
This only checks hasDeleteFiles == true and isDV(deleteFile). It never scans the converted Iceberg table to confirm row counts match the post-DV Delta row count, or that the specific rows Delta marked deleted are absent from Iceberg reads.
The dv-partitioned-with-checkpoint golden exercises DV updates and file removes (V6–V13) — please assert row-level parity (count + sampled content) so the bugs in comments 1–3 cannot pass silently.
Same concern for goldenDeltaTableConversion at line 322–335, which only asserts execute() doesn't throw.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org