diff --git a/core/src/main/java/org/apache/iceberg/PartitionsTable.java b/core/src/main/java/org/apache/iceberg/PartitionsTable.java index 09c6e7893b7e..1b7ea1ef9076 100644 --- a/core/src/main/java/org/apache/iceberg/PartitionsTable.java +++ b/core/src/main/java/org/apache/iceberg/PartitionsTable.java @@ -71,6 +71,11 @@ public class PartitionsTable extends BaseMetadataTable { "position_delete_file_count", Types.IntegerType.get(), "Count of position delete files"), + Types.NestedField.required( + 12, + "total_position_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of position delete files"), Types.NestedField.required( 7, "equality_delete_record_count", @@ -81,6 +86,11 @@ public class PartitionsTable extends BaseMetadataTable { "equality_delete_file_count", Types.IntegerType.get(), "Count of equality delete files"), + Types.NestedField.required( + 13, + "total_equality_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of equality delete files"), Types.NestedField.optional( 9, "last_updated_at", @@ -108,8 +118,10 @@ public Schema schema() { "total_data_file_size_in_bytes", "position_delete_record_count", "position_delete_file_count", + "total_position_delete_file_size_in_bytes", "equality_delete_record_count", "equality_delete_file_count", + "total_equality_delete_file_size_in_bytes", "last_updated_at", "last_updated_snapshot_id"); } @@ -137,8 +149,10 @@ private DataTask task(StaticTableScan scan) { root.dataFileSizeInBytes, root.posDeleteRecordCount, root.posDeleteFileCount, + root.posDeleteFileSizeInBytes, root.eqDeleteRecordCount, root.eqDeleteFileCount, + root.eqDeleteFileSizeInBytes, root.lastUpdatedAt, root.lastUpdatedSnapshotId)); } else { @@ -160,8 +174,10 @@ private static StaticDataTask.Row convertPartition(Partition partition) { partition.dataFileSizeInBytes, partition.posDeleteRecordCount, partition.posDeleteFileCount, + partition.posDeleteFileSizeInBytes, partition.eqDeleteRecordCount, partition.eqDeleteFileCount, + partition.eqDeleteFileSizeInBytes, partition.lastUpdatedAt, partition.lastUpdatedSnapshotId); } @@ -277,8 +293,10 @@ static class Partition { private long dataFileSizeInBytes; private long posDeleteRecordCount; private int posDeleteFileCount; + private long posDeleteFileSizeInBytes; private long eqDeleteRecordCount; private int eqDeleteFileCount; + private long eqDeleteFileSizeInBytes; private Long lastUpdatedAt; private Long lastUpdatedSnapshotId; @@ -290,8 +308,10 @@ static class Partition { this.dataFileSizeInBytes = 0L; this.posDeleteRecordCount = 0L; this.posDeleteFileCount = 0; + this.posDeleteFileSizeInBytes = 0L; this.eqDeleteRecordCount = 0L; this.eqDeleteFileCount = 0; + this.eqDeleteFileSizeInBytes = 0L; } void update(ContentFile file, Snapshot snapshot) { @@ -314,10 +334,12 @@ void update(ContentFile file, Snapshot snapshot) { case POSITION_DELETES: this.posDeleteRecordCount += file.recordCount(); this.posDeleteFileCount += 1; + this.posDeleteFileSizeInBytes += file.fileSizeInBytes(); break; case EQUALITY_DELETES: this.eqDeleteRecordCount += file.recordCount(); this.eqDeleteFileCount += 1; + this.eqDeleteFileSizeInBytes += file.fileSizeInBytes(); break; default: throw new UnsupportedOperationException( diff --git a/docs/docs/flink-queries.md b/docs/docs/flink-queries.md index 8518f2f40339..5681719e1a87 100644 --- a/docs/docs/flink-queries.md +++ b/docs/docs/flink-queries.md @@ -492,12 +492,12 @@ To show a table's current partitions: SELECT * FROM prod.db.table$partitions; ``` -| partition | spec_id | record_count | file_count | total_data_file_size_in_bytes | position_delete_record_count | position_delete_file_count | equality_delete_record_count | equality_delete_file_count | last_updated_at(μs) | last_updated_snapshot_id | -| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------|----------------------------|---------------------|--------------------------| -| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 0 | 0 | 1633086034192000 | 9205185327307503337 | -| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 0 | 0 | 1633172537358000 | 867027598972211003 | -| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | -| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 1 | 1 | 1633169159489000 | 6941468797545315876 | +| partition | spec_id | record_count | file_count | total_data_file_size_in_bytes | position_delete_record_count | position_delete_file_count | total_position_delete_file_size_in_bytes | equality_delete_record_count | equality_delete_file_count | total_equality_delete_file_size_in_bytes | last_updated_at(μs) | last_updated_snapshot_id | +| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------------------|------------------------------|----------------------------|------------------------------------------|---------------------|--------------------------| +| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 50 | 0 | 0 | 0 | 1633086034192000 | 9205185327307503337 | +| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 100 | 0 | 0 | 0 | 1633172537358000 | 867027598972211003 | +| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | +| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 0 | 1 | 1 | 400 | 1633169159489000 | 6941468797545315876 | Note: For unpartitioned tables, the partitions table will not contain the partition and spec_id fields. diff --git a/docs/docs/spark-queries.md b/docs/docs/spark-queries.md index ce626f42b14b..12d94705a12d 100644 --- a/docs/docs/spark-queries.md +++ b/docs/docs/spark-queries.md @@ -427,12 +427,12 @@ To show a table's current partitions: SELECT * FROM prod.db.table.partitions; ``` -| partition | spec_id | record_count | file_count | total_data_file_size_in_bytes | position_delete_record_count | position_delete_file_count | equality_delete_record_count | equality_delete_file_count | last_updated_at(μs) | last_updated_snapshot_id | -| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------|----------------------------|---------------------|--------------------------| -| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 0 | 0 | 1633086034192000 | 9205185327307503337 | -| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 0 | 0 | 1633172537358000 | 867027598972211003 | -| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | -| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 1 | 1 | 1633169159489000 | 6941468797545315876 | +| partition | spec_id | record_count | file_count | total_data_file_size_in_bytes | position_delete_record_count | position_delete_file_count | total_position_delete_file_size_in_bytes | equality_delete_record_count | equality_delete_file_count | total_equality_delete_file_size_in_bytes | last_updated_at(μs) | last_updated_snapshot_id | +| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------------------|------------------------------|----------------------------|------------------------------------------|---------------------|--------------------------| +| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 50 | 0 | 0 | 0 | 1633086034192000 | 9205185327307503337 | +| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 100 | 0 | 0 | 0 | 1633172537358000 | 867027598972211003 | +| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | +| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 0 | 1 | 1 | 400 | 1633169159489000 | 6941468797545315876 | Note: diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java index 6ae7bfb53a2e..334b83d76128 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java @@ -37,6 +37,7 @@ import org.apache.flink.configuration.CoreOptions; import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.types.Row; +import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; @@ -51,6 +52,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.avro.Avro; @@ -59,11 +61,15 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.flink.CatalogTestBase; import org.apache.iceberg.flink.TestHelpers; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Types; @@ -419,6 +425,43 @@ public void testPartitionedTable() throws Exception { assertThat(actualPartitionsWithProjection.get(i).getField(0)).isEqualTo(1); } + // check partitions table schema + List actualPartitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(actualPartitions).hasSize(2); + + Row partitionA = actualPartitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + Row partitionB = actualPartitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + // Check files table List expectedFiles = Stream.concat(expectedDataFiles.stream(), expectedDeleteFiles.stream()) @@ -592,6 +635,149 @@ public void testAllFilesPartitioned() throws Exception { TestHelpers.assertEquals(filesTableSchema, expectedFiles, actualFiles); } + @TestTemplate + public void testPartitionsTable() { + assumeThat(isPartition).isTrue(); + + // 2 inserts for partition 'a' and 'b' in @BeforeEach, 2 records each + // Check partitions table + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(partitions).hasSize(2); + + // Verify partition 'a' + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + // Verify partition 'b' + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + + long partitionASnapshotId = (Long) partitionA.getField("last_updated_snapshot_id"); + long partitionBSnapshotId = (Long) partitionB.getField("last_updated_snapshot_id"); + + // partitions 'a' and 'b' were inserted in separate commits + assertThat(partitionASnapshotId).isNotEqualTo(partitionBSnapshotId); + + // Test filtering by partition + List filteredPartitions = + sql("SELECT * FROM %s$partitions WHERE `partition`.`data` = 'a'", TABLE_NAME); + assertThat(filteredPartitions).hasSize(1); + assertThat(filteredPartitions.get(0).getField("partition")).isEqualTo(Row.of("a")); + + // Test projection + List projectedPartitions = + sql( + "SELECT record_count, file_count FROM %s$partitions ORDER BY `partition`.`data`", + TABLE_NAME); + assertThat(projectedPartitions).hasSize(2); + assertThat(projectedPartitions.get(0).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(0).getField("file_count")).isEqualTo(1); + assertThat(projectedPartitions.get(1).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(1).getField("file_count")).isEqualTo(1); + } + + @TestTemplate + public void testPartitionsTableWithDVs() throws IOException { + assumeThat(isPartition).isTrue(); + + String dvTableName = "test_dv_table"; + sql( + "CREATE TABLE %s (id INT, data VARCHAR,d DOUBLE) PARTITIONED BY (data) WITH ('format-version'='3', 'write.format.default'='%s')", + dvTableName, format.name()); + try { + sql("INSERT INTO %s VALUES (1,'a',10),(2,'a',20)", dvTableName); + sql("INSERT INTO %s VALUES (1,'b',10),(2,'b',20)", dvTableName); + + Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName)); + + // Write DV against the data file in partition 'b' + DataFile dataFile = + Iterables.getFirst(table.currentSnapshot().addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 1); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", dvTableName); + assertThat(partitions).hasSize(2); + + // partition 'a' should have no deletes + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + + // partition 'b' should have DV delete stats + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_position_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + } finally { + sql("DROP TABLE IF EXISTS %s", dvTableName); + } + } + + @TestTemplate + public void testPartitionsTableForUnpartitionedTable() { + assumeThat(isPartition).isFalse(); + + // For unpartitioned tables, there should be one row representing the whole table + List partitions = sql("SELECT * FROM %s$partitions", TABLE_NAME); + assertThat(partitions).hasSize(1); + + Row partition = partitions.get(0); + // Unpartitioned table should have record count = 4 (3 from first insert + 1 from second insert) + assertThat(partition.getField("record_count")).isEqualTo(4L); + assertThat(partition.getField("file_count")).isEqualTo(2); + assertThat((Long) partition.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partition.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("last_updated_at")).isNotNull(); + assertThat(partition.getField("last_updated_snapshot_id")).isNotNull(); + } + @TestTemplate public void testMetadataLogEntries() { Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME)); @@ -813,4 +999,18 @@ private List allDataManifests(Table table) { private List deleteManifests(Table table) { return table.currentSnapshot().deleteManifests(table.io()); } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java index 6ae7bfb53a2e..334b83d76128 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java @@ -37,6 +37,7 @@ import org.apache.flink.configuration.CoreOptions; import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.types.Row; +import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; @@ -51,6 +52,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.avro.Avro; @@ -59,11 +61,15 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.flink.CatalogTestBase; import org.apache.iceberg.flink.TestHelpers; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Types; @@ -419,6 +425,43 @@ public void testPartitionedTable() throws Exception { assertThat(actualPartitionsWithProjection.get(i).getField(0)).isEqualTo(1); } + // check partitions table schema + List actualPartitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(actualPartitions).hasSize(2); + + Row partitionA = actualPartitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + Row partitionB = actualPartitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + // Check files table List expectedFiles = Stream.concat(expectedDataFiles.stream(), expectedDeleteFiles.stream()) @@ -592,6 +635,149 @@ public void testAllFilesPartitioned() throws Exception { TestHelpers.assertEquals(filesTableSchema, expectedFiles, actualFiles); } + @TestTemplate + public void testPartitionsTable() { + assumeThat(isPartition).isTrue(); + + // 2 inserts for partition 'a' and 'b' in @BeforeEach, 2 records each + // Check partitions table + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(partitions).hasSize(2); + + // Verify partition 'a' + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + // Verify partition 'b' + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + + long partitionASnapshotId = (Long) partitionA.getField("last_updated_snapshot_id"); + long partitionBSnapshotId = (Long) partitionB.getField("last_updated_snapshot_id"); + + // partitions 'a' and 'b' were inserted in separate commits + assertThat(partitionASnapshotId).isNotEqualTo(partitionBSnapshotId); + + // Test filtering by partition + List filteredPartitions = + sql("SELECT * FROM %s$partitions WHERE `partition`.`data` = 'a'", TABLE_NAME); + assertThat(filteredPartitions).hasSize(1); + assertThat(filteredPartitions.get(0).getField("partition")).isEqualTo(Row.of("a")); + + // Test projection + List projectedPartitions = + sql( + "SELECT record_count, file_count FROM %s$partitions ORDER BY `partition`.`data`", + TABLE_NAME); + assertThat(projectedPartitions).hasSize(2); + assertThat(projectedPartitions.get(0).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(0).getField("file_count")).isEqualTo(1); + assertThat(projectedPartitions.get(1).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(1).getField("file_count")).isEqualTo(1); + } + + @TestTemplate + public void testPartitionsTableWithDVs() throws IOException { + assumeThat(isPartition).isTrue(); + + String dvTableName = "test_dv_table"; + sql( + "CREATE TABLE %s (id INT, data VARCHAR,d DOUBLE) PARTITIONED BY (data) WITH ('format-version'='3', 'write.format.default'='%s')", + dvTableName, format.name()); + try { + sql("INSERT INTO %s VALUES (1,'a',10),(2,'a',20)", dvTableName); + sql("INSERT INTO %s VALUES (1,'b',10),(2,'b',20)", dvTableName); + + Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName)); + + // Write DV against the data file in partition 'b' + DataFile dataFile = + Iterables.getFirst(table.currentSnapshot().addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 1); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", dvTableName); + assertThat(partitions).hasSize(2); + + // partition 'a' should have no deletes + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + + // partition 'b' should have DV delete stats + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_position_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + } finally { + sql("DROP TABLE IF EXISTS %s", dvTableName); + } + } + + @TestTemplate + public void testPartitionsTableForUnpartitionedTable() { + assumeThat(isPartition).isFalse(); + + // For unpartitioned tables, there should be one row representing the whole table + List partitions = sql("SELECT * FROM %s$partitions", TABLE_NAME); + assertThat(partitions).hasSize(1); + + Row partition = partitions.get(0); + // Unpartitioned table should have record count = 4 (3 from first insert + 1 from second insert) + assertThat(partition.getField("record_count")).isEqualTo(4L); + assertThat(partition.getField("file_count")).isEqualTo(2); + assertThat((Long) partition.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partition.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("last_updated_at")).isNotNull(); + assertThat(partition.getField("last_updated_snapshot_id")).isNotNull(); + } + @TestTemplate public void testMetadataLogEntries() { Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME)); @@ -813,4 +999,18 @@ private List allDataManifests(Table table) { private List deleteManifests(Table table) { return table.currentSnapshot().deleteManifests(table.io()); } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java index 6ae7bfb53a2e..334b83d76128 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java @@ -37,6 +37,7 @@ import org.apache.flink.configuration.CoreOptions; import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.types.Row; +import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; @@ -51,6 +52,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.StructLike; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.avro.Avro; @@ -59,11 +61,15 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.flink.CatalogTestBase; import org.apache.iceberg.flink.TestHelpers; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Types; @@ -419,6 +425,43 @@ public void testPartitionedTable() throws Exception { assertThat(actualPartitionsWithProjection.get(i).getField(0)).isEqualTo(1); } + // check partitions table schema + List actualPartitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(actualPartitions).hasSize(2); + + Row partitionA = actualPartitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + Row partitionB = actualPartitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_equality_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + // Check files table List expectedFiles = Stream.concat(expectedDataFiles.stream(), expectedDeleteFiles.stream()) @@ -592,6 +635,149 @@ public void testAllFilesPartitioned() throws Exception { TestHelpers.assertEquals(filesTableSchema, expectedFiles, actualFiles); } + @TestTemplate + public void testPartitionsTable() { + assumeThat(isPartition).isTrue(); + + // 2 inserts for partition 'a' and 'b' in @BeforeEach, 2 records each + // Check partitions table + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", TABLE_NAME); + assertThat(partitions).hasSize(2); + + // Verify partition 'a' + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("spec_id")).isEqualTo(0); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionA.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("last_updated_at")).isNotNull(); + assertThat(partitionA.getField("last_updated_snapshot_id")).isNotNull(); + + // Verify partition 'b' + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("spec_id")).isEqualTo(0); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionB.getField("last_updated_at")).isNotNull(); + assertThat(partitionB.getField("last_updated_snapshot_id")).isNotNull(); + + long partitionASnapshotId = (Long) partitionA.getField("last_updated_snapshot_id"); + long partitionBSnapshotId = (Long) partitionB.getField("last_updated_snapshot_id"); + + // partitions 'a' and 'b' were inserted in separate commits + assertThat(partitionASnapshotId).isNotEqualTo(partitionBSnapshotId); + + // Test filtering by partition + List filteredPartitions = + sql("SELECT * FROM %s$partitions WHERE `partition`.`data` = 'a'", TABLE_NAME); + assertThat(filteredPartitions).hasSize(1); + assertThat(filteredPartitions.get(0).getField("partition")).isEqualTo(Row.of("a")); + + // Test projection + List projectedPartitions = + sql( + "SELECT record_count, file_count FROM %s$partitions ORDER BY `partition`.`data`", + TABLE_NAME); + assertThat(projectedPartitions).hasSize(2); + assertThat(projectedPartitions.get(0).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(0).getField("file_count")).isEqualTo(1); + assertThat(projectedPartitions.get(1).getField("record_count")).isEqualTo(2L); + assertThat(projectedPartitions.get(1).getField("file_count")).isEqualTo(1); + } + + @TestTemplate + public void testPartitionsTableWithDVs() throws IOException { + assumeThat(isPartition).isTrue(); + + String dvTableName = "test_dv_table"; + sql( + "CREATE TABLE %s (id INT, data VARCHAR,d DOUBLE) PARTITIONED BY (data) WITH ('format-version'='3', 'write.format.default'='%s')", + dvTableName, format.name()); + try { + sql("INSERT INTO %s VALUES (1,'a',10),(2,'a',20)", dvTableName); + sql("INSERT INTO %s VALUES (1,'b',10),(2,'b',20)", dvTableName); + + Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName)); + + // Write DV against the data file in partition 'b' + DataFile dataFile = + Iterables.getFirst(table.currentSnapshot().addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 1); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + + List partitions = + sql("SELECT * FROM %s$partitions ORDER BY `partition`.`data`", dvTableName); + assertThat(partitions).hasSize(2); + + // partition 'a' should have no deletes + Row partitionA = partitions.get(0); + assertThat(partitionA.getField("partition")).isEqualTo(Row.of("a")); + assertThat(partitionA.getField("record_count")).isEqualTo(2L); + assertThat(partitionA.getField("file_count")).isEqualTo(1); + assertThat(partitionA.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionA.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionA.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + + // partition 'b' should have DV delete stats + Row partitionB = partitions.get(1); + assertThat(partitionB.getField("partition")).isEqualTo(Row.of("b")); + assertThat(partitionB.getField("record_count")).isEqualTo(2L); + assertThat(partitionB.getField("file_count")).isEqualTo(1); + assertThat(partitionB.getField("position_delete_record_count")).isEqualTo(1L); + assertThat(partitionB.getField("position_delete_file_count")).isEqualTo(1); + assertThat((Long) partitionB.getField("total_position_delete_file_size_in_bytes")) + .isGreaterThan(0L); + assertThat(partitionB.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partitionB.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partitionB.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + } finally { + sql("DROP TABLE IF EXISTS %s", dvTableName); + } + } + + @TestTemplate + public void testPartitionsTableForUnpartitionedTable() { + assumeThat(isPartition).isFalse(); + + // For unpartitioned tables, there should be one row representing the whole table + List partitions = sql("SELECT * FROM %s$partitions", TABLE_NAME); + assertThat(partitions).hasSize(1); + + Row partition = partitions.get(0); + // Unpartitioned table should have record count = 4 (3 from first insert + 1 from second insert) + assertThat(partition.getField("record_count")).isEqualTo(4L); + assertThat(partition.getField("file_count")).isEqualTo(2); + assertThat((Long) partition.getField("total_data_file_size_in_bytes")).isGreaterThan(0L); + assertThat(partition.getField("position_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("position_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_position_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_record_count")).isEqualTo(0L); + assertThat(partition.getField("equality_delete_file_count")).isEqualTo(0); + assertThat(partition.getField("total_equality_delete_file_size_in_bytes")).isEqualTo(0L); + assertThat(partition.getField("last_updated_at")).isNotNull(); + assertThat(partition.getField("last_updated_snapshot_id")).isNotNull(); + } + @TestTemplate public void testMetadataLogEntries() { Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME)); @@ -813,4 +999,18 @@ private List allDataManifests(Table table) { private List deleteManifests(Table table) { return table.currentSnapshot().deleteManifests(table.io()); } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 53c9ac6b2257..1a17401cc4ba 100644 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -37,6 +37,7 @@ import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.StringJoiner; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -45,6 +46,8 @@ import org.apache.avro.generic.GenericRecordBuilder; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.ManifestFile; @@ -62,6 +65,8 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.deletes.PositionDeleteWriter; import org.apache.iceberg.encryption.EncryptedOutputFile; @@ -1285,6 +1290,11 @@ public void testUnpartitionedPartitionsTable() { "position_delete_file_count", Types.IntegerType.get(), "Count of position delete files"), + required( + 12, + "total_position_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of position delete files"), required( 7, "equality_delete_record_count", @@ -1295,6 +1305,11 @@ public void testUnpartitionedPartitionsTable() { "equality_delete_file_count", Types.IntegerType.get(), "Count of equality delete files"), + required( + 13, + "total_equality_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of equality delete files"), optional( 9, "last_updated_at", @@ -1325,8 +1340,10 @@ public void testUnpartitionedPartitionsTable() { totalSizeInBytes(table.currentSnapshot().addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .build(); List actual = @@ -1394,8 +1411,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1410,8 +1429,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1530,8 +1551,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1546,8 +1569,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { dataFiles.get(1).fileSizeInBytes() + dataFiles.get(2).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1584,8 +1609,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", null) .set("last_updated_snapshot_id", null) @@ -1674,8 +1701,10 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1690,8 +1719,14 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 2L) // should be incremented now .set("position_delete_file_count", 2) // should be incremented now + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(posDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", posDeleteCommitId) @@ -1725,8 +1760,14 @@ public void testPartitionsTableDeleteStats() { .set("file_count", 1) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 2L) // should be incremented now .set("equality_delete_file_count", 2) // should be incremented now + .set( + "total_equality_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(eqDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.EQUALITY_DELETES)) .set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", eqDeleteCommitId) .build()); @@ -1736,6 +1777,115 @@ public void testPartitionsTableDeleteStats() { } } + @Test + public void testPartitionsTableDeleteStatsWithDVs() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test"); + Table table = + createTable( + tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3")); + Table partitionsTable = loadTable(tableIdentifier, "partitions"); + + Dataset df1 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(1, "a"), new SimpleRecord(1, "b"), new SimpleRecord(1, "c")), + SimpleRecord.class); + Dataset df2 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(2, "d"), new SimpleRecord(2, "e"), new SimpleRecord(2, "f")), + SimpleRecord.class); + + df1.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long firstCommitId = table.currentSnapshot().snapshotId(); + + // add a second file + df2.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long secondCommitId = table.currentSnapshot().snapshotId(); + + // write DVs against the data file in partition id=2 + DataFile dataFile = + Iterables.getFirst(table.snapshot(secondCommitId).addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 2); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + long dvCommitId = table.currentSnapshot().snapshotId(); + + List actual = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .orderBy("partition.id") + .collectAsList(); + assertThat(actual).as("Actual results should have two rows").hasSize(2); + + GenericRecordBuilder builder = + new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions")); + GenericRecordBuilder partitionBuilder = + new GenericRecordBuilder( + AvroSchemaUtil.convert( + partitionsTable.schema().findType("partition").asStructType(), "partition")); + List expected = Lists.newArrayList(); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 1).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 0L) + .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", firstCommitId) + .build()); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 2).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 2L) + .set("position_delete_file_count", 1) + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(dvCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(dvCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", dvCommitId) + .build()); + + for (int i = 0; i < 2; i += 1) { + TestHelpers.assertEqualsSafe( + partitionsTable.schema().asStruct(), expected.get(i), actual.get(i)); + } + } + @Test public synchronized void testSnapshotReadAfterAddColumn() { TableIdentifier tableIdentifier = TableIdentifier.of("db", "table"); @@ -2446,6 +2596,15 @@ private long totalSizeInBytes(Iterable dataFiles) { return Lists.newArrayList(dataFiles).stream().mapToLong(DataFile::fileSizeInBytes).sum(); } + private static long totalDeleteFileSizeInBytes( + Iterable deleteFiles, FileContent content) { + return Lists.newArrayList(deleteFiles).stream() + .filter(Objects::nonNull) + .filter(f -> f.content() == content) + .mapToLong(DeleteFile::fileSizeInBytes) + .sum(); + } + private void assertDataFilePartitions( List dataFiles, List expectedPartitionIds) { assertThat(dataFiles) @@ -2458,4 +2617,18 @@ private void assertDataFilePartitions( .isEqualTo(expectedPartitionIds.get(i).intValue()); } } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 67647925e59d..0958da203e48 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -37,6 +37,7 @@ import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.StringJoiner; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -45,6 +46,8 @@ import org.apache.avro.generic.GenericRecordBuilder; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.ManifestFile; @@ -62,6 +65,8 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.deletes.PositionDeleteWriter; import org.apache.iceberg.encryption.EncryptedOutputFile; @@ -1286,6 +1291,11 @@ public void testUnpartitionedPartitionsTable() { "position_delete_file_count", Types.IntegerType.get(), "Count of position delete files"), + required( + 12, + "total_position_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of position delete files"), required( 7, "equality_delete_record_count", @@ -1296,6 +1306,11 @@ public void testUnpartitionedPartitionsTable() { "equality_delete_file_count", Types.IntegerType.get(), "Count of equality delete files"), + required( + 13, + "total_equality_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of equality delete files"), optional( 9, "last_updated_at", @@ -1326,8 +1341,10 @@ public void testUnpartitionedPartitionsTable() { totalSizeInBytes(table.currentSnapshot().addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .build(); List actual = @@ -1395,8 +1412,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1411,8 +1430,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1533,8 +1554,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1549,8 +1572,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { dataFiles.get(1).fileSizeInBytes() + dataFiles.get(2).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1587,8 +1612,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", null) .set("last_updated_snapshot_id", null) @@ -1677,8 +1704,10 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1693,8 +1722,14 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 2L) // should be incremented now .set("position_delete_file_count", 2) // should be incremented now + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(posDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", posDeleteCommitId) @@ -1728,8 +1763,14 @@ public void testPartitionsTableDeleteStats() { .set("file_count", 1) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 2L) // should be incremented now .set("equality_delete_file_count", 2) // should be incremented now + .set( + "total_equality_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(eqDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.EQUALITY_DELETES)) .set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", eqDeleteCommitId) .build()); @@ -1739,6 +1780,115 @@ public void testPartitionsTableDeleteStats() { } } + @Test + public void testPartitionsTableDeleteStatsWithDVs() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test"); + Table table = + createTable( + tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3")); + Table partitionsTable = loadTable(tableIdentifier, "partitions"); + + Dataset df1 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(1, "a"), new SimpleRecord(1, "b"), new SimpleRecord(1, "c")), + SimpleRecord.class); + Dataset df2 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(2, "d"), new SimpleRecord(2, "e"), new SimpleRecord(2, "f")), + SimpleRecord.class); + + df1.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long firstCommitId = table.currentSnapshot().snapshotId(); + + // add a second file + df2.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long secondCommitId = table.currentSnapshot().snapshotId(); + + // write DVs against the data file in partition id=2 + DataFile dataFile = + Iterables.getFirst(table.snapshot(secondCommitId).addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 2); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + long dvCommitId = table.currentSnapshot().snapshotId(); + + List actual = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .orderBy("partition.id") + .collectAsList(); + assertThat(actual).as("Actual results should have two rows").hasSize(2); + + GenericRecordBuilder builder = + new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions")); + GenericRecordBuilder partitionBuilder = + new GenericRecordBuilder( + AvroSchemaUtil.convert( + partitionsTable.schema().findType("partition").asStructType(), "partition")); + List expected = Lists.newArrayList(); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 1).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 0L) + .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", firstCommitId) + .build()); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 2).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 2L) + .set("position_delete_file_count", 1) + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(dvCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(dvCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", dvCommitId) + .build()); + + for (int i = 0; i < 2; i += 1) { + TestHelpers.assertEqualsSafe( + partitionsTable.schema().asStruct(), expected.get(i), actual.get(i)); + } + } + @Test public synchronized void testSnapshotReadAfterAddColumn() { TableIdentifier tableIdentifier = TableIdentifier.of("db", "table"); @@ -2449,6 +2599,15 @@ private long totalSizeInBytes(Iterable dataFiles) { return Lists.newArrayList(dataFiles).stream().mapToLong(DataFile::fileSizeInBytes).sum(); } + private static long totalDeleteFileSizeInBytes( + Iterable deleteFiles, FileContent content) { + return Lists.newArrayList(deleteFiles).stream() + .filter(Objects::nonNull) + .filter(f -> f.content() == content) + .mapToLong(DeleteFile::fileSizeInBytes) + .sum(); + } + private void assertDataFilePartitions( List dataFiles, List expectedPartitionIds) { assertThat(dataFiles) @@ -2461,4 +2620,18 @@ private void assertDataFilePartitions( .isEqualTo(expectedPartitionIds.get(i).intValue()); } } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 67647925e59d..0958da203e48 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -37,6 +37,7 @@ import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.StringJoiner; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -45,6 +46,8 @@ import org.apache.avro.generic.GenericRecordBuilder; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.ManifestFile; @@ -62,6 +65,8 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.deletes.PositionDeleteWriter; import org.apache.iceberg.encryption.EncryptedOutputFile; @@ -1286,6 +1291,11 @@ public void testUnpartitionedPartitionsTable() { "position_delete_file_count", Types.IntegerType.get(), "Count of position delete files"), + required( + 12, + "total_position_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of position delete files"), required( 7, "equality_delete_record_count", @@ -1296,6 +1306,11 @@ public void testUnpartitionedPartitionsTable() { "equality_delete_file_count", Types.IntegerType.get(), "Count of equality delete files"), + required( + 13, + "total_equality_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of equality delete files"), optional( 9, "last_updated_at", @@ -1326,8 +1341,10 @@ public void testUnpartitionedPartitionsTable() { totalSizeInBytes(table.currentSnapshot().addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .build(); List actual = @@ -1395,8 +1412,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1411,8 +1430,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1533,8 +1554,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1549,8 +1572,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { dataFiles.get(1).fileSizeInBytes() + dataFiles.get(2).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1587,8 +1612,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", null) .set("last_updated_snapshot_id", null) @@ -1677,8 +1704,10 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1693,8 +1722,14 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 2L) // should be incremented now .set("position_delete_file_count", 2) // should be incremented now + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(posDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", posDeleteCommitId) @@ -1728,8 +1763,14 @@ public void testPartitionsTableDeleteStats() { .set("file_count", 1) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 2L) // should be incremented now .set("equality_delete_file_count", 2) // should be incremented now + .set( + "total_equality_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(eqDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.EQUALITY_DELETES)) .set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", eqDeleteCommitId) .build()); @@ -1739,6 +1780,115 @@ public void testPartitionsTableDeleteStats() { } } + @Test + public void testPartitionsTableDeleteStatsWithDVs() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test"); + Table table = + createTable( + tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3")); + Table partitionsTable = loadTable(tableIdentifier, "partitions"); + + Dataset df1 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(1, "a"), new SimpleRecord(1, "b"), new SimpleRecord(1, "c")), + SimpleRecord.class); + Dataset df2 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(2, "d"), new SimpleRecord(2, "e"), new SimpleRecord(2, "f")), + SimpleRecord.class); + + df1.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long firstCommitId = table.currentSnapshot().snapshotId(); + + // add a second file + df2.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long secondCommitId = table.currentSnapshot().snapshotId(); + + // write DVs against the data file in partition id=2 + DataFile dataFile = + Iterables.getFirst(table.snapshot(secondCommitId).addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 2); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + long dvCommitId = table.currentSnapshot().snapshotId(); + + List actual = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .orderBy("partition.id") + .collectAsList(); + assertThat(actual).as("Actual results should have two rows").hasSize(2); + + GenericRecordBuilder builder = + new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions")); + GenericRecordBuilder partitionBuilder = + new GenericRecordBuilder( + AvroSchemaUtil.convert( + partitionsTable.schema().findType("partition").asStructType(), "partition")); + List expected = Lists.newArrayList(); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 1).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 0L) + .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", firstCommitId) + .build()); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 2).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 2L) + .set("position_delete_file_count", 1) + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(dvCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(dvCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", dvCommitId) + .build()); + + for (int i = 0; i < 2; i += 1) { + TestHelpers.assertEqualsSafe( + partitionsTable.schema().asStruct(), expected.get(i), actual.get(i)); + } + } + @Test public synchronized void testSnapshotReadAfterAddColumn() { TableIdentifier tableIdentifier = TableIdentifier.of("db", "table"); @@ -2449,6 +2599,15 @@ private long totalSizeInBytes(Iterable dataFiles) { return Lists.newArrayList(dataFiles).stream().mapToLong(DataFile::fileSizeInBytes).sum(); } + private static long totalDeleteFileSizeInBytes( + Iterable deleteFiles, FileContent content) { + return Lists.newArrayList(deleteFiles).stream() + .filter(Objects::nonNull) + .filter(f -> f.content() == content) + .mapToLong(DeleteFile::fileSizeInBytes) + .sum(); + } + private void assertDataFilePartitions( List dataFiles, List expectedPartitionIds) { assertThat(dataFiles) @@ -2461,4 +2620,18 @@ private void assertDataFilePartitions( .isEqualTo(expectedPartitionIds.get(i).intValue()); } } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 67647925e59d..0958da203e48 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -37,6 +37,7 @@ import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.StringJoiner; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -45,6 +46,8 @@ import org.apache.avro.generic.GenericRecordBuilder; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.ManifestFile; @@ -62,6 +65,8 @@ import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.BaseDVFileWriter; +import org.apache.iceberg.deletes.DVFileWriter; import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.deletes.PositionDeleteWriter; import org.apache.iceberg.encryption.EncryptedOutputFile; @@ -1286,6 +1291,11 @@ public void testUnpartitionedPartitionsTable() { "position_delete_file_count", Types.IntegerType.get(), "Count of position delete files"), + required( + 12, + "total_position_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of position delete files"), required( 7, "equality_delete_record_count", @@ -1296,6 +1306,11 @@ public void testUnpartitionedPartitionsTable() { "equality_delete_file_count", Types.IntegerType.get(), "Count of equality delete files"), + required( + 13, + "total_equality_delete_file_size_in_bytes", + Types.LongType.get(), + "Total size in bytes of equality delete files"), optional( 9, "last_updated_at", @@ -1326,8 +1341,10 @@ public void testUnpartitionedPartitionsTable() { totalSizeInBytes(table.currentSnapshot().addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .build(); List actual = @@ -1395,8 +1412,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1411,8 +1430,10 @@ public void testPartitionsTable() { totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1533,8 +1554,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1549,8 +1572,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { dataFiles.get(1).fileSizeInBytes() + dataFiles.get(2).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", secondCommitId) @@ -1587,8 +1612,10 @@ public void testPartitionsTableLastUpdatedSnapshot() { .set("total_data_file_size_in_bytes", dataFiles.get(0).fileSizeInBytes()) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", null) .set("last_updated_snapshot_id", null) @@ -1677,8 +1704,10 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", firstCommitId) @@ -1693,8 +1722,14 @@ public void testPartitionsTableDeleteStats() { totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) .set("position_delete_record_count", 2L) // should be incremented now .set("position_delete_file_count", 2) // should be incremented now + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(posDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) .set("equality_delete_record_count", 0L) .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) .set("spec_id", 0) .set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", posDeleteCommitId) @@ -1728,8 +1763,14 @@ public void testPartitionsTableDeleteStats() { .set("file_count", 1) .set("position_delete_record_count", 0L) .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) .set("equality_delete_record_count", 2L) // should be incremented now .set("equality_delete_file_count", 2) // should be incremented now + .set( + "total_equality_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(eqDeleteCommitId).addedDeleteFiles(table.io()), + FileContent.EQUALITY_DELETES)) .set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000) .set("last_updated_snapshot_id", eqDeleteCommitId) .build()); @@ -1739,6 +1780,115 @@ public void testPartitionsTableDeleteStats() { } } + @Test + public void testPartitionsTableDeleteStatsWithDVs() throws IOException { + TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test"); + Table table = + createTable( + tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3")); + Table partitionsTable = loadTable(tableIdentifier, "partitions"); + + Dataset df1 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(1, "a"), new SimpleRecord(1, "b"), new SimpleRecord(1, "c")), + SimpleRecord.class); + Dataset df2 = + spark.createDataFrame( + Lists.newArrayList( + new SimpleRecord(2, "d"), new SimpleRecord(2, "e"), new SimpleRecord(2, "f")), + SimpleRecord.class); + + df1.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long firstCommitId = table.currentSnapshot().snapshotId(); + + // add a second file + df2.select("id", "data") + .write() + .format("iceberg") + .mode("append") + .save(loadLocation(tableIdentifier)); + + table.refresh(); + long secondCommitId = table.currentSnapshot().snapshotId(); + + // write DVs against the data file in partition id=2 + DataFile dataFile = + Iterables.getFirst(table.snapshot(secondCommitId).addedDataFiles(table.io()), null); + List dvs = writeDV(table, dataFile.partition(), dataFile.location(), 2); + table.newRowDelta().addDeletes(dvs.get(0)).commit(); + table.refresh(); + long dvCommitId = table.currentSnapshot().snapshotId(); + + List actual = + spark + .read() + .format("iceberg") + .load(loadLocation(tableIdentifier, "partitions")) + .orderBy("partition.id") + .collectAsList(); + assertThat(actual).as("Actual results should have two rows").hasSize(2); + + GenericRecordBuilder builder = + new GenericRecordBuilder(AvroSchemaUtil.convert(partitionsTable.schema(), "partitions")); + GenericRecordBuilder partitionBuilder = + new GenericRecordBuilder( + AvroSchemaUtil.convert( + partitionsTable.schema().findType("partition").asStructType(), "partition")); + List expected = Lists.newArrayList(); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 1).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(firstCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 0L) + .set("position_delete_file_count", 0) + .set("total_position_delete_file_size_in_bytes", 0L) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", firstCommitId) + .build()); + expected.add( + builder + .set("partition", partitionBuilder.set("id", 2).build()) + .set("record_count", 3L) + .set("file_count", 1) + .set( + "total_data_file_size_in_bytes", + totalSizeInBytes(table.snapshot(secondCommitId).addedDataFiles(table.io()))) + .set("position_delete_record_count", 2L) + .set("position_delete_file_count", 1) + .set( + "total_position_delete_file_size_in_bytes", + totalDeleteFileSizeInBytes( + table.snapshot(dvCommitId).addedDeleteFiles(table.io()), + FileContent.POSITION_DELETES)) + .set("equality_delete_record_count", 0L) + .set("equality_delete_file_count", 0) + .set("total_equality_delete_file_size_in_bytes", 0L) + .set("spec_id", 0) + .set("last_updated_at", table.snapshot(dvCommitId).timestampMillis() * 1000) + .set("last_updated_snapshot_id", dvCommitId) + .build()); + + for (int i = 0; i < 2; i += 1) { + TestHelpers.assertEqualsSafe( + partitionsTable.schema().asStruct(), expected.get(i), actual.get(i)); + } + } + @Test public synchronized void testSnapshotReadAfterAddColumn() { TableIdentifier tableIdentifier = TableIdentifier.of("db", "table"); @@ -2449,6 +2599,15 @@ private long totalSizeInBytes(Iterable dataFiles) { return Lists.newArrayList(dataFiles).stream().mapToLong(DataFile::fileSizeInBytes).sum(); } + private static long totalDeleteFileSizeInBytes( + Iterable deleteFiles, FileContent content) { + return Lists.newArrayList(deleteFiles).stream() + .filter(Objects::nonNull) + .filter(f -> f.content() == content) + .mapToLong(DeleteFile::fileSizeInBytes) + .sum(); + } + private void assertDataFilePartitions( List dataFiles, List expectedPartitionIds) { assertThat(dataFiles) @@ -2461,4 +2620,18 @@ private void assertDataFilePartitions( .isEqualTo(expectedPartitionIds.get(i).intValue()); } } + + private List writeDV( + Table table, StructLike partition, String path, int numPositionsToDelete) throws IOException { + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + DVFileWriter writer = new BaseDVFileWriter(fileFactory, p -> null); + try (DVFileWriter closeableWriter = writer) { + for (int row = 0; row < numPositionsToDelete; row++) { + closeableWriter.delete(path, row, table.spec(), partition); + } + } + + return writer.result().deleteFiles(); + } }