HonahX commented on code in PR #15407: URL: https://github.com/apache/iceberg/pull/15407#discussion_r3139968476
########## 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 { Review Comment: In the old implementation, the [commitInitialDeltaSnapshotToIcebergTransaction](https://github.com/apache/iceberg/blob/57b7c2bd564e5dfa8ccb7cb9e6386e7e942bac0f/delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeTableAction.java#L241), will first to find the actual earliest reconstructable version that have all the data files available. That could help reduce the conversion failure due to VACUUM cleaned up files and breaks time travel ability. What's the design choice behind not using that in the new impl? ########## delta-lake/src/integration/java/org/apache/iceberg/delta/TestSnapshotDeltaLakeKernelTable.java: ########## @@ -0,0 +1,351 @@ +/* + * 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 io.delta.kernel.defaults.engine.DefaultEngine; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.sql.Timestamp; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkCatalog; +import org.apache.iceberg.util.LocationUtil; +import org.apache.spark.sql.DataFrameWriter; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.connector.catalog.CatalogPlugin; +import org.apache.spark.sql.delta.catalog.DeltaCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +public class TestSnapshotDeltaLakeKernelTable extends SparkDeltaLakeSnapshotTestBase { + private static final String NAMESPACE = "delta_conversion_test_ns"; + private static final String DEFAULT_SPARK_CATALOG = "spark_catalog"; + private static final String ICEBERG_CATALOG_NAME = "iceberg_hive"; + private static final Map<String, String> BASE_SPARK_CONFIG = + ImmutableMap.of( + "type", + "hive", + "default-namespace", + "default", + "parquet-enabled", + "true", + "cache-enabled", + "false" // Spark will delete tables using v1, leaving the cache out of sync + ); + + private static Dataset<Row> genericDataFrame; + + @TempDir private File sourceLocation; + @TempDir private File destinationLocation; + + public TestSnapshotDeltaLakeKernelTable() { + super(ICEBERG_CATALOG_NAME, SparkCatalog.class.getName(), BASE_SPARK_CONFIG); + spark.conf().set("spark.sql.catalog." + DEFAULT_SPARK_CATALOG, DeltaCatalog.class.getName()); + } + + @BeforeAll + public static void beforeClass() { + spark.sql(String.format("CREATE DATABASE IF NOT EXISTS %s", NAMESPACE)); + StructType schema = + new StructType( + new StructField[] { + new StructField("id", DataTypes.IntegerType, false, Metadata.empty()), + new StructField("created_at", DataTypes.TimestampType, true, Metadata.empty()), + new StructField("event_name", DataTypes.StringType, true, Metadata.empty()), + new StructField("is_active", DataTypes.BooleanType, true, Metadata.empty()), + new StructField("price", DataTypes.DoubleType, true, Metadata.empty()) + }); + List<Row> data = + Arrays.asList( + RowFactory.create(1, Timestamp.valueOf("2025-01-01 10:00:00"), "Signup", true, 0.00), + RowFactory.create( + 2, Timestamp.valueOf("2025-01-02 14:30:00"), "Purchase", true, 199.99), + RowFactory.create( + 3, Timestamp.valueOf("2025-01-03 09:15:00"), "Deactivation", false, 0.00), + RowFactory.create(4, Timestamp.valueOf("2025-01-03 09:16:00"), "Refund", false, 0.00), + RowFactory.create(5, Timestamp.valueOf("2025-01-03 09:17:00"), "Refund", false, 0.00)); + genericDataFrame = spark.createDataFrame(data, schema); + } + + @AfterAll + public static void afterClass() { + spark.sql(String.format("DROP DATABASE IF EXISTS %s CASCADE", NAMESPACE)); + } + + @Test + public void testBasicPartitionedInsertsOnly() { + String sourceTable = toFullTableName(DEFAULT_SPARK_CATALOG, "partitioned_table"); + String sourceTableLocation = sourceLocation.toURI().toString(); + + writeDeltaTable(genericDataFrame, sourceTable, sourceTableLocation, "id"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (10, current_date(), null, null, null);"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (11, current_date(), null, null, null);"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (12, current_date(), null, null, null);"); + + String newTableIdentifier = toFullTableName(ICEBERG_CATALOG_NAME, "iceberg_partitioned_table"); + + // Act + SnapshotDeltaLakeTable conversionAction = + DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeKernelTable( + spark, newTableIdentifier, sourceTableLocation); + SnapshotDeltaLakeTable.Result result = conversionAction.execute(); + + // Assert + checkLatestSnapshotIntegrity(sourceTable, newTableIdentifier); + checkTagContentAndOrder(sourceTable, sourceTableLocation, newTableIdentifier, 0); + checkIcebergTableLocation(newTableIdentifier, sourceTableLocation); + } + + @Test + public void testInsertUpdateDeleteSqls() { + String sourceTable = toFullTableName(DEFAULT_SPARK_CATALOG, "crud_table"); + String sourceTableLocation = sourceLocation.toURI().toString(); + + writeDeltaTable(genericDataFrame, sourceTable, sourceTableLocation, "id"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (10, current_date(), null, null, null);"); + spark.sql("DELETE FROM " + sourceTable + " WHERE id=3;"); + spark.sql("UPDATE " + sourceTable + " SET id=3 WHERE id=1;"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (11, current_date(), null, null, null);"); + + String newTableIdentifier = toFullTableName(ICEBERG_CATALOG_NAME, "iceberg_crud_table"); + + // Act + SnapshotDeltaLakeTable conversionAction = + DeltaLakeToIcebergMigrationSparkIntegration.snapshotDeltaLakeKernelTable( + spark, newTableIdentifier, sourceTableLocation); + SnapshotDeltaLakeTable.Result result = conversionAction.execute(); + + // Assert + checkLatestSnapshotIntegrity(sourceTable, newTableIdentifier); + checkTagContentAndOrder(sourceTable, sourceTableLocation, newTableIdentifier, 0); + checkIcebergTableLocation(newTableIdentifier, sourceTableLocation); + } + + @Test + public void testConversionAfterVacuum() throws IOException { + String sourceTable = toFullTableName(DEFAULT_SPARK_CATALOG, "vacuumed_table"); + String sourceTableLocation = sourceLocation.toURI().toString(); + + writeDeltaTable(genericDataFrame, sourceTable, sourceTableLocation, "id"); + for (int i = 0; i < 5; i++) { + spark.sql( + "UPDATE " + + sourceTable + + " SET price=" + + ThreadLocalRandom.current().nextDouble(1000) + + " where id=" + + i + + ";"); + } + spark.sql("UPDATE " + sourceTable + " SET created_at=current_date() ;"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (10, current_date(), null, null, null);"); + spark.sql("INSERT INTO " + sourceTable + " VALUES (11, current_date(), null, null, null);"); + spark.sql("DELETE FROM " + sourceTable + " WHERE id>=10;"); + spark.sql("VACUUM " + sourceTable + " RETAIN 0 HOURS"); + spark.sql( + "INSERT INTO " + sourceTable + " VALUES (12, current_date(), 'after_vacuum', null, null);"); + spark.sql("UPDATE " + sourceTable + " SET id=13 WHERE id=5;"); + + // Checkpoint generated. Simulate logs clean-up + assertThat(deleteDeltaLogFile("00000000000000000000.json")).isTrue(); + assertThat(deleteDeltaLogFile("00000000000000000001.json")).isTrue(); + assertThat(deleteDeltaLogFile("00000000000000000002.json")).isTrue(); Review Comment: Does the log clean-up always to happen with the VACUUM? Will there be case where the old logs are not cleaned up and our conversion fails because of the missing data file (deleted by VACUUM)? Seems this indicate some requirement for conversion to work when VACUUM is called on the source table, could you please elaborate on the scope of supporting vacuum in this PR? Thanks! ########## delta-lake/src/test/resources/delta/golden/README.md: ########## @@ -0,0 +1,69 @@ +<!-- + - 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. + --> + +# Delta Lake Golden Tables + +## Overview +This directory contains the **Delta Lake Golden Tables**, a set of reference data tables used to verify compliance with the Delta Lake protocol. + +These tables cover various features of the Delta Lake specification (e.g., different protocol versions, partition strategies, checkpoints, and data types). They act as a "compliance suite" to ensure that this project's reader/writer implementation can correctly handle data produced by the official Delta Lake reference implementation. + +## Origin & Attribution +These resources were sourced directly from the open-source Delta Lake project. + +* **Source Repository:** [delta-io/delta](https://github.com/delta-io/delta) +* **Module Path:** `connectors/golden-tables/src` +* **License:** Apache License 2.0 (See [LICENSE](https://github.com/delta-io/delta/blob/master/LICENSE.txt) in the project) + +## Purpose +The primary goal of these tables is **integration testing**. + +Each folder represents a specific "edge case" or feature configuration in Delta Lake. By running the conversion against +these tables, we verify that: +1. **Protocol Versioning:** We respect `minReaderVersion` and `minWriterVersion`. +2. **Feature Support:** We can read tables with `checkpoints`, `vacuumed` files, `deletion vectors`, or `column mapping`. +3. **Data Consistency:** We read the correct number of rows and column values as defined by the transaction log (`_delta_log`). + +## Table Inventory & Explanations + +The tables are typically named after the specific feature they test. Below are explanations of common patterns found in this suite: Review Comment: [nit] seems we only include basic-decimal-table and dv-partitioned-with-checkpoint. Could we clean up other not included table names here? ########## delta-lake/src/integration/java/org/apache/iceberg/delta/TestSnapshotDeltaLakeKernelTable.java: ########## @@ -0,0 +1,351 @@ +/* + * 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 io.delta.kernel.defaults.engine.DefaultEngine; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.sql.Timestamp; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.Spark3Util; +import org.apache.iceberg.spark.SparkCatalog; +import org.apache.iceberg.util.LocationUtil; +import org.apache.spark.sql.DataFrameWriter; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.connector.catalog.CatalogPlugin; +import org.apache.spark.sql.delta.catalog.DeltaCatalog; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.Metadata; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +public class TestSnapshotDeltaLakeKernelTable extends SparkDeltaLakeSnapshotTestBase { + private static final String NAMESPACE = "delta_conversion_test_ns"; + private static final String DEFAULT_SPARK_CATALOG = "spark_catalog"; + private static final String ICEBERG_CATALOG_NAME = "iceberg_hive"; + private static final Map<String, String> BASE_SPARK_CONFIG = + ImmutableMap.of( + "type", + "hive", + "default-namespace", + "default", + "parquet-enabled", + "true", + "cache-enabled", + "false" // Spark will delete tables using v1, leaving the cache out of sync + ); + + private static Dataset<Row> genericDataFrame; + + @TempDir private File sourceLocation; + @TempDir private File destinationLocation; + + public TestSnapshotDeltaLakeKernelTable() { Review Comment: Shall we add tests for unpartitioned table and a table with possible all data types? Also would be good if we have some check similar to `checkDataFilePathsIntegrity` to verify that data file path is absolute and matches those in delta table. ########## build.gradle: ########## @@ -638,6 +638,8 @@ project(':iceberg-delta-lake') { compileOnly libs.immutables.value compileOnly "io.delta:delta-standalone_${scalaVersion}:${libs.versions.delta.standalone.get()}" + compileOnly "io.delta:delta-kernel-api:4.0.0" Review Comment: [nit], Could we add the version number to https://github.com/apache/iceberg/blob/main/gradle/libs.versions.toml ? ########## delta-lake/src/main/java/org/apache/iceberg/delta/DeletionVectorConverter.java: ########## @@ -0,0 +1,42 @@ +/* + * 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.engine.Engine; +import io.delta.kernel.internal.actions.DeletionVectorDescriptor; +import io.delta.kernel.internal.deletionvectors.DeletionVectorUtils; +import io.delta.kernel.internal.deletionvectors.RoaringBitmapArray; +import io.delta.kernel.internal.util.Tuple2; + +public class DeletionVectorConverter { Review Comment: Could this also be package-private? ########## delta-lake/src/main/java/org/apache/iceberg/delta/BaseSnapshotDeltaLakeTableAction.java: ########## @@ -450,11 +448,12 @@ private void tagCurrentSnapshot(long deltaVersion, Transaction transaction) { */ private static String getFullFilePath(String path, String tableRoot) { URI dataFileUri = URI.create(path); - String decodedPath = URLDecoder.decode(path, StandardCharsets.UTF_8); if (dataFileUri.isAbsolute()) { - return decodedPath; + return dataFileUri.getScheme().equalsIgnoreCase("file") ? dataFileUri.getPath() : path; } else { - return tableRoot + File.separator + decodedPath; + String decodedPath = dataFileUri.getPath(); + String separator = tableRoot.contains(":/") ? "/" : File.separator; + return tableRoot + (tableRoot.endsWith(separator) ? "" : separator) + decodedPath; Review Comment: Is this change necessary? ########## 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(); Review Comment: `assertDeltacolumnMappingDisabled` also calls `getLatestDeltaSnapshot` once. Could we consolidate to only call it once every execution? ########## 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(); + } else if (!dataFilesToAdd.isEmpty() && dataFilesToRemove.isEmpty()) { + // Append only + AppendFiles appendFiles = transaction.newAppend(); + dataFilesToAdd.forEach(appendFiles::appendFile); + appendFiles.commit(); + } else if (dataFilesToAdd.isEmpty() && !dataFilesToRemove.isEmpty()) { + // Delete only + DeleteFiles deleteFiles = transaction.newDelete(); + dataFilesToRemove.forEach(deleteFiles::deleteFile); + deleteFiles.commit(); + } else if (!dataFilesToAdd.isEmpty() && !dataFilesToRemove.isEmpty()) { + // Overwrite + OverwriteFiles overwriteFiles = transaction.newOverwrite(); + dataFilesToAdd.forEach(overwriteFiles::addFile); + dataFilesToRemove.forEach(overwriteFiles::deleteFile); + overwriteFiles.commit(); + } else { + // Tag case + transaction.newAppend().commit(); + } + } + + private DataFile buildDataFileFromAddDeltaAction(AddFile addFile, Transaction transaction) { Review Comment: [nit] Most of the logic in this method is the same as `buildDataFileFromRemoveDeltaAction`, could we extract the common "buildDataFile" logic to reduce code duplication? -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
