Skip to content

Core: Add dv_count column to PartitionsTable #17851

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hemanthboyina wants to merge 2 commits into from
+376 −19
Closed

Core: Add dv_count column to PartitionsTable #17851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a69b214
rebase and tests modification
hemanthboyina Aug 31, 2026
fc0ec7d
fixed the doc changes
hemanthboyina Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions core/src/main/java/org/apache/iceberg/PartitionsTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,9 @@ 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;

Expand All @@ -115,7 +118,8 @@ public class PartitionsTable extends BaseMetadataTable {
EQUALITY_DELETE_RECORD_COUNT,
EQUALITY_DELETE_FILE_COUNT,
LAST_UPDATED_AT,
LAST_UPDATED_SNAPSHOT_ID);
LAST_UPDATED_SNAPSHOT_ID,
DV_COUNT);
this.unpartitionedTable = Partitioning.partitionType(table).fields().isEmpty();
}

Expand All @@ -138,7 +142,8 @@ public Schema schema() {
EQUALITY_DELETE_RECORD_COUNT.fieldId(),
EQUALITY_DELETE_FILE_COUNT.fieldId(),
LAST_UPDATED_AT.fieldId(),
LAST_UPDATED_SNAPSHOT_ID.fieldId()));
LAST_UPDATED_SNAPSHOT_ID.fieldId(),
DV_COUNT.fieldId()));
}
return schema;
}
Expand Down Expand Up @@ -167,7 +172,8 @@ private DataTask task(StaticTableScan scan) {
root.eqDeleteRecordCount,
root.eqDeleteFileCount,
root.lastUpdatedAt,
root.lastUpdatedSnapshotId));
root.lastUpdatedSnapshotId,
root.dvCount));
} else {
return StaticDataTask.of(
io().newInputFile(table().operations().current().metadataFileLocation()),
Expand All @@ -190,7 +196,8 @@ private static StaticDataTask.Row convertPartition(Partition partition) {
partition.eqDeleteRecordCount,
partition.eqDeleteFileCount,
partition.lastUpdatedAt,
partition.lastUpdatedSnapshotId);
partition.lastUpdatedSnapshotId,
partition.dvCount);
}

@VisibleForTesting
Expand Down Expand Up @@ -310,6 +317,7 @@ static class Partition {
private int eqDeleteFileCount;
private Long lastUpdatedAt;
private Long lastUpdatedSnapshotId;
private int dvCount;

Partition(StructLike key, PartitionData partitionDataTemplate) {
this.partitionData = toPartitionData(key, partitionDataTemplate);
Expand All @@ -321,13 +329,19 @@ static class Partition {
this.posDeleteFileCount = 0;
this.eqDeleteRecordCount = 0L;
this.eqDeleteFileCount = 0;
this.dvCount = 0;
}

@VisibleForTesting
PartitionData partitionData() {
return partitionData;
}

@VisibleForTesting
int dvCount() {
return dvCount;
}

void update(ContentFile<?> file, Snapshot snapshot) {
if (snapshot != null) {
long snapshotCommitTime = snapshot.timestampMillis() * 1000;
Expand All @@ -348,6 +362,9 @@ void update(ContentFile<?> file, Snapshot snapshot) {
case POSITION_DELETES:
this.posDeleteRecordCount += file.recordCount();
this.posDeleteFileCount += 1;
if (file.format() == FileFormat.PUFFIN) {
this.dvCount += 1;
}
Comment thread
hemanthboyina marked this conversation as resolved.
break;
case EQUALITY_DELETES:
this.eqDeleteRecordCount += file.recordCount();
Expand Down
36 changes: 36 additions & 0 deletions core/src/test/java/org/apache/iceberg/TestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,42 @@ 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();
// Two DVs packed into a single Puffin file (same path, different content offsets), each
// referencing a different data file in data_bucket=0. Used to verify that multiple DVs in one
// Puffin file are each counted once.
static final DeleteFile FILE_A_SHARED_DV =
FileMetadata.deleteFileBuilder(SPEC)
.ofPositionDeletes()
.withPath("/path/to/data-a-shared-deletes.puffin")
.withFileSizeInBytes(20)
.withPartitionPath("data_bucket=0")
.withRecordCount(1)
.withReferencedDataFile(FILE_A.location())
.withContentOffset(4)
.withContentSizeInBytes(6)
.build();
static final DeleteFile FILE_A2_SHARED_DV =
FileMetadata.deleteFileBuilder(SPEC)
.ofPositionDeletes()
.withPath("/path/to/data-a-shared-deletes.puffin")
.withFileSizeInBytes(20)
.withPartitionPath("data_bucket=0")
.withRecordCount(1)
.withReferencedDataFile(FILE_A2.location())
.withContentOffset(10)
.withContentSizeInBytes(6)
.build();
// Equality delete files.
static final DeleteFile FILE_A2_DELETES =
FileMetadata.deleteFileBuilder(SPEC)
Expand Down
45 changes: 45 additions & 0 deletions core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -720,6 +720,51 @@ public void partitionsTableScanNoStats() {
}
}

@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
// position_delete_file_count also includes DVs (non-breaking)
assertThat(partitions.get(0).dvCount()).isEqualTo(2);
// data_bucket=1 has one DV → dv_count=1
assertThat(partitions.get(1).dvCount()).isEqualTo(1);
}

@TestTemplate
public void testPartitionsTableDvCountSharedPuffinFile() {
assumeThat(formatVersion).isGreaterThanOrEqualTo(3);

// FILE_A and FILE_A2 are both in data_bucket=0
table.newFastAppend().appendFile(FILE_A).appendFile(FILE_A2).commit();
// FILE_A_SHARED_DV and FILE_A2_SHARED_DV are two DVs packed into a single Puffin file (same
// path, different content offsets), each referencing a different data file in data_bucket=0
table.newRowDelta().addDeletes(FILE_A_SHARED_DV).addDeletes(FILE_A2_SHARED_DV).commit();

Table partitionsTable = new PartitionsTable(table);
StaticTableScan scan = (StaticTableScan) partitionsTable.newScan();
List<PartitionsTable.Partition> partitions =
Lists.newArrayList(PartitionsTable.partitions(table, scan));

// Both DVs live in one Puffin file but are distinct manifest entries → each counted once, so
// dv_count=2 (not deduped to 1 by the shared Puffin file path)
assertThat(partitions).hasSize(1);
assertThat(partitions.get(0).dvCount()).isEqualTo(2);
}

@TestTemplate
public void partitionsTableScanAndFilter() {
preparePartitionedTable();
Expand Down
12 changes: 6 additions & 6 deletions docs/docs/flink-queries.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 | equality_delete_record_count | equality_delete_file_count | last_updated_at(μs) | last_updated_snapshot_id | dv_count |
| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------|----------------------------|---------------------|--------------------------|----------|
| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 0 | 0 | 1633086034192000 | 9205185327307503337 | 1 |
| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 0 | 0 | 1633172537358000 | 867027598972211003 | 1 |
| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | 0 |
| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 1 | 1 | 1633169159489000 | 6941468797545315876 | 0 |

Note:
For unpartitioned tables, the partitions table will not contain the partition and spec_id fields.
Expand Down
12 changes: 6 additions & 6 deletions docs/docs/spark-queries.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -486,12 +486,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 | equality_delete_record_count | equality_delete_file_count | last_updated_at(μs) | last_updated_snapshot_id | dv_count |
| -------------- |---------|---------------|------------|--------------------------|------------------------------|----------------------------|------------------------------|----------------------------|---------------------|--------------------------|----------|
| {20211001, 11} | 0 | 1 | 1 | 100 | 2 | 1 | 0 | 0 | 1633086034192000 | 9205185327307503337 | 1 |
| {20211002, 11} | 0 | 4 | 3 | 500 | 1 | 1 | 0 | 0 | 1633172537358000 | 867027598972211003 | 1 |
| {20211001, 10} | 0 | 7 | 4 | 700 | 0 | 0 | 0 | 0 | 1633082598716000 | 3280122546965981531 | 0 |
| {20211002, 10} | 0 | 3 | 2 | 400 | 0 | 0 | 1 | 1 | 1633169159489000 | 6941468797545315876 | 0 |

Note:

Expand Down
28 changes: 28 additions & 0 deletions flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,12 @@
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;
Expand Down Expand Up @@ -665,6 +668,31 @@ public void testMetadataLogEntries() {
}
}

@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, position_delete_file_count FROM %s$partitions", dvTableName);
assertThat(actual).hasSize(2);
assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
assertThat((int) actual.get(0).getField(1)).isEqualTo(1);
assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
assertThat((int) actual.get(1).getField(1)).isEqualTo(1);
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
}

@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
Expand Down
28 changes: 28 additions & 0 deletions flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,12 @@
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;
Expand Down Expand Up @@ -665,6 +668,31 @@ public void testMetadataLogEntries() {
}
}

@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, position_delete_file_count FROM %s$partitions", dvTableName);
assertThat(actual).hasSize(2);
assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
assertThat((int) actual.get(0).getField(1)).isEqualTo(1);
assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
assertThat((int) actual.get(1).getField(1)).isEqualTo(1);
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
}

@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
Expand Down
28 changes: 28 additions & 0 deletions flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkMetaDataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,12 @@
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;
Expand Down Expand Up @@ -665,6 +668,31 @@ public void testMetadataLogEntries() {
}
}

@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, position_delete_file_count FROM %s$partitions", dvTableName);
assertThat(actual).hasSize(2);
assertThat((int) actual.get(0).getField(0)).isEqualTo(1);
assertThat((int) actual.get(0).getField(1)).isEqualTo(1);
assertThat((int) actual.get(1).getField(0)).isEqualTo(1);
assertThat((int) actual.get(1).getField(1)).isEqualTo(1);
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, dvTableName);
}

@TestTemplate
public void testSnapshotReferencesMetatable() {
Table table = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
Expand Down
Loading
Loading