You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/08/26 13:09:22 UTC
(iceberg) branch revert-16125-dv_count_partition created (now 8eaf4a505b)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a change to branch revert-16125-dv_count_partition
in repository https://gitbox.apache.org/repos/asf/iceberg.git
at 8eaf4a505b Revert "Core: Add dv_count column to PartitionsTable metadata table (#16125)"
This branch includes the following new commits:
new 8eaf4a505b Revert "Core: Add dv_count column to PartitionsTable metadata table (#16125)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg) 01/01: Revert "Core: Add dv_count column to PartitionsTable metadata table (#16125)"
Posted by pv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch revert-16125-dv_count_partition
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 8eaf4a505b73289024964d5c8cdc05fc283f7308
Author: pvary <pe...@gmail.com>
AuthorDate: Wed Aug 26 15:09:16 2026 +0200
Revert "Core: Add dv_count column to PartitionsTable metadata table (#16125)"
This reverts commit 68a7f2767b8da025a2663e67fc5321febf3c8dd1.
---
.../java/org/apache/iceberg/PartitionsTable.java | 33 +++-----------
.../src/test/java/org/apache/iceberg/TestBase.java | 11 -----
.../org/apache/iceberg/TestMetadataTableScans.java | 25 -----------
.../flink/source/TestFlinkMetaDataTable.java | 29 ------------
.../flink/source/TestFlinkMetaDataTable.java | 29 ------------
.../flink/source/TestFlinkMetaDataTable.java | 29 ------------
.../spark/source/TestIcebergSourceTablesBase.java | 51 +---------------------
.../spark/source/TestIcebergSourceTablesBase.java | 51 +---------------------
.../spark/source/TestIcebergSourceTablesBase.java | 51 +---------------------
9 files changed, 8 insertions(+), 301 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/PartitionsTable.java b/core/src/main/java/org/apache/iceberg/PartitionsTable.java
index 81f4ebfdb6..b5dd5c284c 100644
--- a/core/src/main/java/org/apache/iceberg/PartitionsTable.java
+++ b/core/src/main/java/org/apache/iceberg/PartitionsTable.java
@@ -90,9 +90,6 @@ public class PartitionsTable extends BaseMetadataTable {
"last_updated_snapshot_id",
Types.LongType.get(),
"Id of snapshot that last updated this partition");
- private static final Types.NestedField DV_COUNT =
- Types.NestedField.required(
- 12, "dv_count", Types.IntegerType.get(), "Count of deletion vectors");
private final Schema schema;
@@ -118,8 +115,7 @@ public class PartitionsTable extends BaseMetadataTable {
EQUALITY_DELETE_RECORD_COUNT,
EQUALITY_DELETE_FILE_COUNT,
LAST_UPDATED_AT,
- LAST_UPDATED_SNAPSHOT_ID,
- DV_COUNT);
+ LAST_UPDATED_SNAPSHOT_ID);
this.unpartitionedTable = Partitioning.partitionType(table).fields().isEmpty();
}
@@ -142,8 +138,7 @@ public class PartitionsTable extends BaseMetadataTable {
EQUALITY_DELETE_RECORD_COUNT.fieldId(),
EQUALITY_DELETE_FILE_COUNT.fieldId(),
LAST_UPDATED_AT.fieldId(),
- LAST_UPDATED_SNAPSHOT_ID.fieldId(),
- DV_COUNT.fieldId()));
+ LAST_UPDATED_SNAPSHOT_ID.fieldId()));
}
return schema;
}
@@ -172,8 +167,7 @@ public class PartitionsTable extends BaseMetadataTable {
root.eqDeleteRecordCount,
root.eqDeleteFileCount,
root.lastUpdatedAt,
- root.lastUpdatedSnapshotId,
- root.dvCount));
+ root.lastUpdatedSnapshotId));
} else {
return StaticDataTask.of(
io().newInputFile(table().operations().current().metadataFileLocation()),
@@ -196,8 +190,7 @@ public class PartitionsTable extends BaseMetadataTable {
partition.eqDeleteRecordCount,
partition.eqDeleteFileCount,
partition.lastUpdatedAt,
- partition.lastUpdatedSnapshotId,
- partition.dvCount);
+ partition.lastUpdatedSnapshotId);
}
@VisibleForTesting
@@ -317,7 +310,6 @@ public class PartitionsTable extends BaseMetadataTable {
private int eqDeleteFileCount;
private Long lastUpdatedAt;
private Long lastUpdatedSnapshotId;
- private int dvCount;
Partition(StructLike key, PartitionData partitionDataTemplate) {
this.partitionData = toPartitionData(key, partitionDataTemplate);
@@ -329,7 +321,6 @@ public class PartitionsTable extends BaseMetadataTable {
this.posDeleteFileCount = 0;
this.eqDeleteRecordCount = 0L;
this.eqDeleteFileCount = 0;
- this.dvCount = 0;
}
@VisibleForTesting
@@ -337,16 +328,6 @@ public class PartitionsTable extends BaseMetadataTable {
return partitionData;
}
- @VisibleForTesting
- int dvCount() {
- return dvCount;
- }
-
- @VisibleForTesting
- int posDeleteFileCount() {
- return posDeleteFileCount;
- }
-
void update(ContentFile<?> file, Snapshot snapshot) {
if (snapshot != null) {
long snapshotCommitTime = snapshot.timestampMillis() * 1000;
@@ -366,11 +347,7 @@ public class PartitionsTable extends BaseMetadataTable {
break;
case POSITION_DELETES:
this.posDeleteRecordCount += file.recordCount();
- if (file.format() == FileFormat.PUFFIN) {
- this.dvCount += 1;
- } else {
- this.posDeleteFileCount += 1;
- }
+ this.posDeleteFileCount += 1;
break;
case EQUALITY_DELETES:
this.eqDeleteRecordCount += file.recordCount();
diff --git a/core/src/test/java/org/apache/iceberg/TestBase.java b/core/src/test/java/org/apache/iceberg/TestBase.java
index 9ced8eaf91..6865d1fb1b 100644
--- a/core/src/test/java/org/apache/iceberg/TestBase.java
+++ b/core/src/test/java/org/apache/iceberg/TestBase.java
@@ -114,17 +114,6 @@ public class TestBase {
.withContentOffset(4)
.withContentSizeInBytes(6)
.build();
- static final DeleteFile FILE_A2_DV =
- FileMetadata.deleteFileBuilder(SPEC)
- .ofPositionDeletes()
- .withPath("/path/to/data-a2-deletes.puffin")
- .withFileSizeInBytes(10)
- .withPartitionPath("data_bucket=0")
- .withRecordCount(5)
- .withReferencedDataFile(FILE_A2.location())
- .withContentOffset(4)
- .withContentSizeInBytes(6)
- .build();
// Equality delete files.
static final DeleteFile FILE_A2_DELETES =
FileMetadata.deleteFileBuilder(SPEC)
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
index 319c241ed6..ded6dd643d 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
@@ -703,31 +703,6 @@ public class TestMetadataTableScans extends MetadataTableScanTestBase {
assertThat(partitionSchemas).allSatisfy(schema -> assertThat(schema).isSameAs(expectedSchema));
}
- @TestTemplate
- public void testPartitionsTableDvCount() {
- assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
-
- // FILE_A and FILE_A2 are both in data_bucket=0; FILE_B is in data_bucket=1
- table.newFastAppend().appendFile(FILE_A).appendFile(FILE_A2).appendFile(FILE_B).commit();
- // Add two DVs to data_bucket=0 (recordCount=1 and recordCount=5) and one DV to data_bucket=1
- // to confirm dv_count tracks file count, not record count
- table.newRowDelta().addDeletes(FILE_A_DV).addDeletes(FILE_A2_DV).addDeletes(FILE_B_DV).commit();
-
- Table partitionsTable = new PartitionsTable(table);
- StaticTableScan scan = (StaticTableScan) partitionsTable.newScan();
- List<PartitionsTable.Partition> partitions =
- Lists.newArrayList(PartitionsTable.partitions(table, scan));
- partitions.sort(Comparator.comparing(p -> p.partitionData().get(0, Integer.class)));
-
- assertThat(partitions).hasSize(2);
- // data_bucket=0 has two DVs (with different record counts) → dv_count=2, not the record sum
- assertThat(partitions.get(0).dvCount()).isEqualTo(2);
- assertThat(partitions.get(0).posDeleteFileCount()).isEqualTo(0);
- // data_bucket=1 has one DV → dv_count=1
- assertThat(partitions.get(1).dvCount()).isEqualTo(1);
- assertThat(partitions.get(1).posDeleteFileCount()).isEqualTo(0);
- }
-
@TestTemplate
public void partitionsTableScanNoStats() {
table.newFastAppend().appendFile(FILE_WITH_STATS).commit();
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 ae726e16b8..6ae7bfb53a 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,12 +37,9 @@ import org.apache.flink.configuration.Configuration;
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;
-import org.apache.iceberg.FileGenerationUtil;
-import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -670,32 +667,6 @@ public class TestFlinkMetaDataTable extends CatalogTestBase {
}
}
- @TestTemplate
- public void testPartitionsTableDvCount() {
- String dvTableName = "dv_test_table";
- sql(
- "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) WITH ('format-version'='3')",
- dvTableName);
- sql("INSERT INTO %s VALUES (1,'a')", dvTableName);
- sql("INSERT INTO %s VALUES (1,'b')", dvTableName);
-
- Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName));
- List<DataFile> dataFiles =
- Lists.newArrayList(
- CloseableIterable.transform(table.newScan().planFiles(), FileScanTask::file));
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
-
- List<Row> actual = sql("SELECT dv_count FROM %s$partitions", dvTableName);
-
- assertThat(actual).hasSize(2);
- assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
- assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
- }
-
@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
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 ae726e16b8..6ae7bfb53a 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,12 +37,9 @@ import org.apache.flink.configuration.Configuration;
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;
-import org.apache.iceberg.FileGenerationUtil;
-import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -670,32 +667,6 @@ public class TestFlinkMetaDataTable extends CatalogTestBase {
}
}
- @TestTemplate
- public void testPartitionsTableDvCount() {
- String dvTableName = "dv_test_table";
- sql(
- "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) WITH ('format-version'='3')",
- dvTableName);
- sql("INSERT INTO %s VALUES (1,'a')", dvTableName);
- sql("INSERT INTO %s VALUES (1,'b')", dvTableName);
-
- Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName));
- List<DataFile> dataFiles =
- Lists.newArrayList(
- CloseableIterable.transform(table.newScan().planFiles(), FileScanTask::file));
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
-
- List<Row> actual = sql("SELECT dv_count FROM %s$partitions", dvTableName);
-
- assertThat(actual).hasSize(2);
- assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
- assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
- }
-
@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
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 ae726e16b8..6ae7bfb53a 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,12 +37,9 @@ import org.apache.flink.configuration.Configuration;
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;
-import org.apache.iceberg.FileGenerationUtil;
-import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -670,32 +667,6 @@ public class TestFlinkMetaDataTable extends CatalogTestBase {
}
}
- @TestTemplate
- public void testPartitionsTableDvCount() {
- String dvTableName = "dv_test_table";
- sql(
- "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) WITH ('format-version'='3')",
- dvTableName);
- sql("INSERT INTO %s VALUES (1,'a')", dvTableName);
- sql("INSERT INTO %s VALUES (1,'b')", dvTableName);
-
- Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, dvTableName));
- List<DataFile> dataFiles =
- Lists.newArrayList(
- CloseableIterable.transform(table.newScan().planFiles(), FileScanTask::file));
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
-
- List<Row> actual = sql("SELECT dv_count FROM %s$partitions", dvTableName);
-
- assertThat(actual).hasSize(2);
- assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
- assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
-
- sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
- }
-
@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
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 436dca3c8e..88113be2dd 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
@@ -45,7 +45,6 @@ import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileGenerationUtil;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -1304,8 +1303,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
10,
"last_updated_snapshot_id",
Types.LongType.get(),
- "Id of snapshot that last updated this partition"),
- required(12, "dv_count", Types.IntegerType.get(), "Count of deletion vectors"));
+ "Id of snapshot that last updated this partition"));
Table partitionsTable = loadTable(tableIdentifier, "partitions");
@@ -1328,7 +1326,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.build();
List<Row> actual =
@@ -1402,7 +1399,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1423,7 +1419,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1546,7 +1541,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1563,7 +1557,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1602,7 +1595,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", null)
.set("last_updated_snapshot_id", null)
@@ -1697,7 +1689,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1718,7 +1709,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 2) // should be incremented now
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", posDeleteCommitId)
@@ -1754,7 +1744,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 2L) // should be incremented now
.set("equality_delete_file_count", 2) // should be incremented now
- .set("dv_count", 0)
.set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", eqDeleteCommitId)
.build());
@@ -1764,44 +1753,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
}
}
- @Test
- public void testPartitionsTableDvCount() {
- TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test");
- Table table =
- createTable(
- tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "a")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "b")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- table.refresh();
- List<DataFile> dataFiles = TestHelpers.dataFiles(table);
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
- table.refresh();
-
- List<Row> actual =
- spark
- .read()
- .format("iceberg")
- .load(loadLocation(tableIdentifier, "partitions"))
- .collectAsList();
-
- assertThat(actual).hasSize(1);
- assertThat((int) actual.get(0).getAs("dv_count")).isEqualTo(2);
- }
-
@Test
public synchronized void testSnapshotReadAfterAddColumn() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
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 436dca3c8e..88113be2dd 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
@@ -45,7 +45,6 @@ import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileGenerationUtil;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -1304,8 +1303,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
10,
"last_updated_snapshot_id",
Types.LongType.get(),
- "Id of snapshot that last updated this partition"),
- required(12, "dv_count", Types.IntegerType.get(), "Count of deletion vectors"));
+ "Id of snapshot that last updated this partition"));
Table partitionsTable = loadTable(tableIdentifier, "partitions");
@@ -1328,7 +1326,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.build();
List<Row> actual =
@@ -1402,7 +1399,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1423,7 +1419,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1546,7 +1541,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1563,7 +1557,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1602,7 +1595,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", null)
.set("last_updated_snapshot_id", null)
@@ -1697,7 +1689,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1718,7 +1709,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 2) // should be incremented now
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", posDeleteCommitId)
@@ -1754,7 +1744,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 2L) // should be incremented now
.set("equality_delete_file_count", 2) // should be incremented now
- .set("dv_count", 0)
.set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", eqDeleteCommitId)
.build());
@@ -1764,44 +1753,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
}
}
- @Test
- public void testPartitionsTableDvCount() {
- TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test");
- Table table =
- createTable(
- tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "a")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "b")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- table.refresh();
- List<DataFile> dataFiles = TestHelpers.dataFiles(table);
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
- table.refresh();
-
- List<Row> actual =
- spark
- .read()
- .format("iceberg")
- .load(loadLocation(tableIdentifier, "partitions"))
- .collectAsList();
-
- assertThat(actual).hasSize(1);
- assertThat((int) actual.get(0).getAs("dv_count")).isEqualTo(2);
- }
-
@Test
public synchronized void testSnapshotReadAfterAddColumn() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");
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 cecfd1a82b..c322730440 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
@@ -47,7 +47,6 @@ import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileGenerationUtil;
import org.apache.iceberg.Files;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.ManifestFile;
@@ -1309,8 +1308,7 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
10,
"last_updated_snapshot_id",
Types.LongType.get(),
- "Id of snapshot that last updated this partition"),
- required(12, "dv_count", Types.IntegerType.get(), "Count of deletion vectors"));
+ "Id of snapshot that last updated this partition"));
Table partitionsTable = loadTable(tableIdentifier, "partitions");
@@ -1333,7 +1331,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.build();
List<Row> actual =
@@ -1407,7 +1404,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1428,7 +1424,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1551,7 +1546,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1568,7 +1562,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(secondCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", secondCommitId)
@@ -1607,7 +1600,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", null)
.set("last_updated_snapshot_id", null)
@@ -1702,7 +1694,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(firstCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", firstCommitId)
@@ -1723,7 +1714,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 2) // should be incremented now
.set("equality_delete_record_count", 0L)
.set("equality_delete_file_count", 0)
- .set("dv_count", 0)
.set("spec_id", 0)
.set("last_updated_at", table.snapshot(posDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", posDeleteCommitId)
@@ -1759,7 +1749,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
.set("position_delete_file_count", 0)
.set("equality_delete_record_count", 2L) // should be incremented now
.set("equality_delete_file_count", 2) // should be incremented now
- .set("dv_count", 0)
.set("last_updated_at", table.snapshot(eqDeleteCommitId).timestampMillis() * 1000)
.set("last_updated_snapshot_id", eqDeleteCommitId)
.build());
@@ -1769,44 +1758,6 @@ public abstract class TestIcebergSourceTablesBase extends TestBase {
}
}
- @Test
- public void testPartitionsTableDvCount() {
- TableIdentifier tableIdentifier = TableIdentifier.of("db", "partitions_dv_test");
- Table table =
- createTable(
- tableIdentifier, SCHEMA, SPEC, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "a")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- spark
- .createDataFrame(Lists.newArrayList(new SimpleRecord(1, "b")), SimpleRecord.class)
- .select("id", "data")
- .write()
- .format("iceberg")
- .mode("append")
- .save(loadLocation(tableIdentifier));
- table.refresh();
- List<DataFile> dataFiles = TestHelpers.dataFiles(table);
- DeleteFile dv1 = FileGenerationUtil.generateDV(table, dataFiles.get(0));
- DeleteFile dv2 = FileGenerationUtil.generateDV(table, dataFiles.get(1));
- table.newRowDelta().addDeletes(dv1).addDeletes(dv2).commit();
- table.refresh();
-
- List<Row> actual =
- spark
- .read()
- .format("iceberg")
- .load(loadLocation(tableIdentifier, "partitions"))
- .collectAsList();
-
- assertThat(actual).hasSize(1);
- assertThat((int) actual.get(0).getAs("dv_count")).isEqualTo(2);
- }
-
@Test
public synchronized void testSnapshotReadAfterAddColumn() {
TableIdentifier tableIdentifier = TableIdentifier.of("db", "table");