This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cad041961c [core] add a `deleted_record_count` field in
PartitionsTable (#8526)
cad041961c is described below
commit cad041961c3e577d706d6f8341ef7dd82ee66fa9
Author: Faiz <[email protected]>
AuthorDate: Mon Jul 13 23:25:58 2026 +0800
[core] add a `deleted_record_count` field in PartitionsTable (#8526)
---
docs/docs/concepts/system-tables.mdx | 11 ++-
docs/docs/pypaimon/system-tables.md | 1 +
.../paimon/table/system/PartitionsTable.java | 74 ++++++++++++++-
.../paimon/table/system/PartitionsTableTest.java | 105 +++++++++++++++++++++
.../pypaimon/table/system/partitions_table.py | 42 +++++++++
.../pypaimon/tests/system/partitions_table_test.py | 67 ++++++++++++-
6 files changed, 290 insertions(+), 10 deletions(-)
diff --git a/docs/docs/concepts/system-tables.mdx
b/docs/docs/concepts/system-tables.mdx
index 29806d7435..d70bc6c39b 100644
--- a/docs/docs/concepts/system-tables.mdx
+++ b/docs/docs/concepts/system-tables.mdx
@@ -439,16 +439,17 @@ You can query the partition files of the table.
SELECT * FROM my_table$partitions;
/*
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
-| partition | record_count | file_size_in_bytes| file_count | last_update_time
| created_at | created_by | updated_by | options | total_buckets |
done |
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
-| {1} | 1 | 645 | 1 | 2024-06-24
10:25:57 | 2024-06-24 10:20:00 | admin | test_user | {} |
1 | false |
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
+| partition | record_count | file_size_in_bytes| file_count | last_update_time
| created_at | created_by | updated_by | options | total_buckets |
done | deleted_record_count |
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
+| {1} | 1 | 645 | 1 | 2024-06-24
10:25:57 | 2024-06-24 10:20:00 | admin | test_user | {} |
1 | false | 0 |
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
*/
```
**Note**:
- The `created_by`, `created_at`, `updated_by`, and `options` fields are
populated from REST catalog audit information. For non-REST catalogs, these
fields will be `NULL`.
+- The `deleted_record_count` field is the number of records marked deleted by
deletion vectors in the partition. It is `0` when the partition has no deletion
vectors and `NULL` when legacy deletion vector metadata does not contain
cardinality.
### Buckets Table
diff --git a/docs/docs/pypaimon/system-tables.md
b/docs/docs/pypaimon/system-tables.md
index 5f82f6c351..fdbf287745 100644
--- a/docs/docs/pypaimon/system-tables.md
+++ b/docs/docs/pypaimon/system-tables.md
@@ -178,6 +178,7 @@ Aggregated partition statistics for the latest snapshot.
| `options` | STRING | Filesystem path returns
`NULL` |
| `total_buckets` | INT NOT NULL |
|
| `done` | BOOLEAN NOT NULL | Filesystem path returns
`False`|
+| `deleted_record_count` | BIGINT | Records marked deleted by
deletion vectors; `0` when none exist and `NULL` when legacy metadata lacks
cardinality |
### `$tags`
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
index 1291308e5a..14b02c86e1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/PartitionsTable.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.system;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.casting.CastExecutor;
import org.apache.paimon.casting.CastExecutors;
import org.apache.paimon.catalog.Catalog;
@@ -32,10 +33,15 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.index.DeletionVectorMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateVisitor;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.ReadonlyTable;
@@ -70,7 +76,9 @@ import java.time.ZoneId;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -78,6 +86,7 @@ import java.util.OptionalLong;
import java.util.stream.Collectors;
import static org.apache.paimon.catalog.Identifier.SYSTEM_TABLE_SPLITTER;
+import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
/** A {@link Table} for showing partitions info. */
public class PartitionsTable implements ReadonlyTable {
@@ -99,7 +108,8 @@ public class PartitionsTable implements ReadonlyTable {
new DataField(7, "updated_by", DataTypes.STRING()),
new DataField(8, "options", DataTypes.STRING()),
new DataField(9, "total_buckets",
DataTypes.INT().notNull()),
- new DataField(10, "done",
DataTypes.BOOLEAN().notNull())));
+ new DataField(10, "done",
DataTypes.BOOLEAN().notNull()),
+ new DataField(11, "deleted_record_count", new
BigIntType(true))));
private final FileStoreTable storeTable;
@@ -224,6 +234,8 @@ public class PartitionsTable implements ReadonlyTable {
.map(CastExecutors::resolveToString)
.collect(Collectors.toList());
+ Map<BinaryRow, Long> deletionNums = deletionNums();
+
// sorted by partition
Iterator<InternalRow> iterator =
partitions.stream()
@@ -236,7 +248,8 @@ public class PartitionsTable implements ReadonlyTable {
fieldGetters,
fileStoreTable
.coreOptions()
-
.partitionDefaultName()))
+
.partitionDefaultName(),
+ deletionNums))
.sorted(Comparator.comparing(row ->
row.getString(0)))
.iterator();
@@ -260,7 +273,8 @@ public class PartitionsTable implements ReadonlyTable {
List<String> partitionKeys,
List<CastExecutor> castExecutors,
InternalRow.FieldGetter[] fieldGetters,
- String defaultPartitionName) {
+ String defaultPartitionName,
+ @Nullable Map<BinaryRow, Long> deletionNums) {
PartitionEntry entry = toPartitionEntry(partition);
StringBuilder partitionStringBuilder = new StringBuilder();
@@ -303,7 +317,59 @@ public class PartitionsTable implements ReadonlyTable {
updatedByString,
optionsString,
partition.totalBuckets(),
- partition.done());
+ partition.done(),
+ deletionNums == null ? null :
deletionNums.getOrDefault(entry.partition(), 0L));
+ }
+
+ @Nullable
+ private Map<BinaryRow, Long> deletionNums() {
+ if (!fileStoreTable.coreOptions().deletionVectorsEnabled()) {
+ return Collections.emptyMap();
+ }
+
+ // do not scan deletion indices if query doesn't need them.
+ if (readType != null
+ && !readType.containsField("deleted_record_count")
+ && !PredicateVisitor.collectFieldNames(predicate)
+ .contains("deleted_record_count")) {
+ return null;
+ }
+
+ Snapshot snapshot =
TimeTravelUtil.tryTravelOrLatest(fileStoreTable);
+ if (snapshot == null || snapshot.indexManifest() == null) {
+ return Collections.emptyMap();
+ }
+
+ IndexFileHandler indexFileHandler =
fileStoreTable.store().newIndexFileHandler();
+ Map<BinaryRow, Long> result = new HashMap<>();
+ for (IndexManifestEntry entry :
+ indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX)) {
+ accumulateDeletionNum(result, entry.partition(),
entry.indexFile());
+ }
+ return result;
+ }
+
+ private void accumulateDeletionNum(
+ Map<BinaryRow, Long> result, BinaryRow partition,
IndexFileMeta indexFile) {
+ if (result.containsKey(partition) && result.get(partition) ==
null) {
+ return;
+ }
+
+ LinkedHashMap<String, DeletionVectorMeta> dvRanges =
indexFile.dvRanges();
+ if (dvRanges == null || dvRanges.isEmpty()) {
+ return;
+ }
+
+ long count = result.getOrDefault(partition, 0L);
+ for (DeletionVectorMeta dvMeta : dvRanges.values()) {
+ Long cardinality = dvMeta.cardinality();
+ if (cardinality == null) {
+ result.put(partition.copy(), null);
+ return;
+ }
+ count += cardinality;
+ }
+ result.put(partition.copy(), count);
}
private PartitionEntry toPartitionEntry(Partition partition) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
index f79c7811b5..ef18580dc6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/PartitionsTableTest.java
@@ -38,6 +38,7 @@ import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -153,6 +154,46 @@ public class PartitionsTableTest extends TableTestBase {
assertThat(result).containsExactlyInAnyOrderElementsOf(expectedRow);
}
+ @Test
+ public void testPartitionDeletionNumWithoutDeletionVectors() throws
Exception {
+ PredicateBuilder builder = new
PredicateBuilder(PartitionsTable.TABLE_TYPE);
+
+ List<InternalRow> expectedRow = new ArrayList<>();
+ expectedRow.add(GenericRow.of(BinaryString.fromString("pt=1"), 0L));
+ expectedRow.add(GenericRow.of(BinaryString.fromString("pt=2"), 0L));
+ expectedRow.add(GenericRow.of(BinaryString.fromString("pt=3"), 0L));
+
+ assertThat(read(partitionsTable, new int[] {0, 11}))
+ .containsExactlyInAnyOrderElementsOf(expectedRow);
+ assertThat(
+ readProjectedPartitions(
+ partitionsTable, builder.greaterThan(11, 0L),
new int[] {0}))
+ .isEmpty();
+ assertThat(readProjectedPartitions(partitionsTable, builder.equal(11,
0L), new int[] {0}))
+ .containsExactlyInAnyOrder("pt=1", "pt=2", "pt=3");
+ }
+
+ @Test
+ public void testPartitionDeletionNumWithDeletionVectors() throws Exception
{
+ PartitionsTable deletionVectorPartitionsTable =
createDeletionVectorPartitionsTable();
+ PredicateBuilder builder = new
PredicateBuilder(PartitionsTable.TABLE_TYPE);
+
+ assertThat(readPartitionAndDeletionNum(deletionVectorPartitionsTable,
null))
+ .containsExactlyInAnyOrder("pt=1-2", "pt=2-1", "pt=3-0");
+ assertThat(
+ readProjectedPartitions(
+ deletionVectorPartitionsTable,
+ builder.greaterThan(11, 0L),
+ new int[] {0}))
+ .containsExactlyInAnyOrder("pt=1", "pt=2");
+ assertThat(
+ readProjectedPartitions(
+ deletionVectorPartitionsTable,
+ builder.equal(11, 0L),
+ new int[] {0}))
+ .containsExactlyInAnyOrder("pt=3");
+ }
+
@Test
void testPartitionWithLegacyPartitionName() throws Exception {
String testTableName = "TestLegacyTable";
@@ -246,4 +287,68 @@ public class PartitionsTableTest extends TableTestBase {
}
return rows;
}
+
+ private PartitionsTable createDeletionVectorPartitionsTable() throws
Exception {
+ String testTableName = "DeletionVectorTable";
+ Schema testSchema =
+ Schema.newBuilder()
+ .column("pk", DataTypes.INT())
+ .column("pt", DataTypes.INT())
+ .column("col1", DataTypes.INT())
+ .partitionKeys("pt")
+ .primaryKey("pk", "pt")
+ .option(CoreOptions.CHANGELOG_PRODUCER.key(), "input")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .option("bucket", "1")
+ .build();
+
+ Identifier testTableId = identifier(testTableName);
+ catalog.createTable(testTableId, testSchema, true);
+ FileStoreTable deletionVectorTable = (FileStoreTable)
catalog.getTable(testTableId);
+
+ write(
+ deletionVectorTable,
+ ioManager,
+ GenericRow.of(1, 1, 1),
+ GenericRow.of(2, 1, 2),
+ GenericRow.of(3, 2, 3),
+ GenericRow.of(5, 1, 5),
+ GenericRow.of(6, 2, 6),
+ GenericRow.of(4, 3, 4));
+ write(
+ deletionVectorTable,
+ ioManager,
+ GenericRow.ofKind(RowKind.DELETE, 1, 1, 1),
+ GenericRow.ofKind(RowKind.DELETE, 2, 1, 2),
+ GenericRow.ofKind(RowKind.DELETE, 3, 2, 3));
+
+ Identifier partitionsTableId =
+ identifier(testTableName + SYSTEM_TABLE_SPLITTER +
PartitionsTable.PARTITIONS);
+ return (PartitionsTable) catalog.getTable(partitionsTableId);
+ }
+
+ private List<String> readProjectedPartitions(
+ PartitionsTable table, Predicate predicate, int[] projection)
throws IOException {
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ readBuilder.withProjection(projection);
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(row ->
rows.add(row.getString(0).toString()));
+ }
+ return rows;
+ }
+
+ private List<String> readPartitionAndDeletionNum(PartitionsTable table,
Predicate predicate)
+ throws IOException {
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ readBuilder.withProjection(new int[] {0, 11});
+ List<String> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan())) {
+ reader.forEachRemaining(
+ row -> rows.add(row.getString(0).toString() + "-" +
row.getLong(1)));
+ }
+ return rows;
+ }
}
diff --git a/paimon-python/pypaimon/table/system/partitions_table.py
b/paimon-python/pypaimon/table/system/partitions_table.py
index 583e4fd94f..70685fee0b 100644
--- a/paimon-python/pypaimon/table/system/partitions_table.py
+++ b/paimon-python/pypaimon/table/system/partitions_table.py
@@ -21,6 +21,8 @@ from typing import List, Optional
import pyarrow
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.manifest.index_manifest_file import IndexManifestFile
from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.schema.data_types import AtomicType, DataField, RowType
@@ -39,6 +41,7 @@ TABLE_TYPE = RowType(False, [
DataField(8, "options", AtomicType("STRING", nullable=True)),
DataField(9, "total_buckets", AtomicType("INT", nullable=False)),
DataField(10, "done", AtomicType("BOOLEAN", nullable=False)),
+ DataField(11, "deleted_record_count", AtomicType("BIGINT", nullable=True)),
])
@@ -73,9 +76,11 @@ class PartitionsTable(SystemTable):
manifest_file_manager = ManifestFileManager(self.base_table)
entries = manifest_file_manager.read_entries_parallel(
manifest_files, drop_stats=True)
+ deleted_record_counts = self._deleted_record_counts(snapshot)
partition_map: dict = {}
for entry in entries:
+ partition_key = tuple(entry.partition.values)
spec_items = tuple(
(field.name, str(value))
for field, value in zip(
@@ -86,6 +91,7 @@ class PartitionsTable(SystemTable):
if stats is None:
stats = {
"spec_items": spec_items,
+ "partition_key": partition_key,
"record_count": 0,
"file_size_in_bytes": 0,
"file_count": 0,
@@ -110,6 +116,7 @@ class PartitionsTable(SystemTable):
file_counts: List[int] = []
last_update_times: List[Optional[int]] = []
total_buckets: List[int] = []
+ deleted_counts: List[Optional[int]] = []
for stats in partition_map.values():
partition_strings.append(_render_partition(stats["spec_items"]))
@@ -118,6 +125,8 @@ class PartitionsTable(SystemTable):
file_counts.append(stats["file_count"])
last_update_times.append(stats["last_update_time"])
total_buckets.append(len(stats["buckets"]))
+ deleted_counts.append(
+ deleted_record_counts.get(stats["partition_key"], 0))
n = len(partition_map)
return pyarrow.table({
@@ -136,8 +145,40 @@ class PartitionsTable(SystemTable):
"options": pyarrow.array([None] * n, type=pyarrow.string()),
"total_buckets": pyarrow.array(total_buckets,
type=pyarrow.int32()),
"done": pyarrow.array([False] * n, type=pyarrow.bool_()),
+ "deleted_record_count": pyarrow.array(
+ deleted_counts, type=pyarrow.int64()),
})
+ def _deleted_record_counts(self, snapshot) -> dict:
+ if not self.base_table.options.deletion_vectors_enabled(False):
+ return {}
+
+ entries = IndexFileHandler(self.base_table).scan(
+ snapshot,
+ lambda entry: entry.index_file.index_type
+ == IndexManifestFile.DELETION_VECTORS_INDEX,
+ )
+ result = {}
+ for entry in entries:
+ partition_key = tuple(entry.partition.values)
+ if partition_key in result and result[partition_key] is None:
+ continue
+
+ dv_ranges = entry.index_file.dv_ranges
+ if not dv_ranges:
+ continue
+
+ count = result.get(partition_key, 0)
+ for dv_meta in dv_ranges.values():
+ if dv_meta.cardinality is None:
+ result[partition_key] = None
+ break
+ count += int(dv_meta.cardinality)
+ else:
+ result[partition_key] = count
+
+ return result
+
@staticmethod
def _empty_table() -> pyarrow.Table:
return pyarrow.table({
@@ -152,6 +193,7 @@ class PartitionsTable(SystemTable):
"options": pyarrow.array([], type=pyarrow.string()),
"total_buckets": pyarrow.array([], type=pyarrow.int32()),
"done": pyarrow.array([], type=pyarrow.bool_()),
+ "deleted_record_count": pyarrow.array([], type=pyarrow.int64()),
})
diff --git a/paimon-python/pypaimon/tests/system/partitions_table_test.py
b/paimon-python/pypaimon/tests/system/partitions_table_test.py
index 8f07fd5f2a..108cc6d358 100644
--- a/paimon-python/pypaimon/tests/system/partitions_table_test.py
+++ b/paimon-python/pypaimon/tests/system/partitions_table_test.py
@@ -21,11 +21,17 @@ import os
import shutil
import tempfile
import unittest
+from unittest.mock import patch
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
+from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.manifest.index_manifest_file import IndexManifestFile
from pypaimon.schema.data_types import DataField
+from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.system.partitions_table import PartitionsTable
@@ -86,7 +92,7 @@ class PartitionsTableTest(unittest.TestCase):
("last_update_time", True), ("created_at", True),
("created_by", True), ("updated_by", True),
("options", True), ("total_buckets", False),
- ("done", False),
+ ("done", False), ("deleted_record_count", True),
]
self.assertEqual([n for n, _ in expected],
[f.name for f in row_type.fields])
@@ -122,11 +128,70 @@ class PartitionsTableTest(unittest.TestCase):
# FileSystem catalog does not maintain a "done" flag.
for done in arrow_table.column("done").to_pylist():
self.assertFalse(done)
+ for count in arrow_table.column("deleted_record_count").to_pylist():
+ self.assertEqual(0, count)
# Catalog-managed fields are not surfaced by the filesystem path.
for field in ("created_by", "updated_by", "options", "created_at"):
for value in arrow_table.column(field).to_pylist():
self.assertIsNone(value)
+ def test_aggregates_deleted_record_count(self):
+ self._create_partitioned_table()
+ self._write_two_partitions()
+
+ table = self.catalog.get_table("db.t$partitions")
+ partition_fields = table.base_table.partition_keys_fields
+ entries = [
+ IndexManifestEntry(
+ kind=0,
+ partition=GenericRow(["2024-01-01"], partition_fields),
+ bucket=0,
+ index_file=IndexFileMeta(
+ IndexManifestFile.DELETION_VECTORS_INDEX,
+ "dv-1",
+ 1,
+ 2,
+ dv_ranges={
+ "file-1": DeletionVectorMeta("file-1", 0, 1, 1),
+ "file-2": DeletionVectorMeta("file-2", 1, 1, 2),
+ },
+ ),
+ ),
+ IndexManifestEntry(
+ kind=0,
+ partition=GenericRow(["2024-01-02"], partition_fields),
+ bucket=0,
+ index_file=IndexFileMeta(
+ IndexManifestFile.DELETION_VECTORS_INDEX,
+ "dv-2",
+ 1,
+ 1,
+ dv_ranges={
+ "legacy-file": DeletionVectorMeta(
+ "legacy-file", 0, 1, None),
+ },
+ ),
+ ),
+ ]
+
+ with patch.object(
+ table.base_table.options,
+ "deletion_vectors_enabled",
+ return_value=True,
+ ), patch(
+ "pypaimon.table.system.partitions_table.IndexFileHandler.scan",
+ return_value=entries,
+ ):
+ arrow_table = _read(table)
+
+ partitions = arrow_table.column("partition").to_pylist()
+ deleted_counts = dict(zip(
+ partitions,
+ arrow_table.column("deleted_record_count").to_pylist(),
+ ))
+ self.assertEqual(3, deleted_counts["dt=2024-01-01"])
+ self.assertIsNone(deleted_counts["dt=2024-01-02"])
+
if __name__ == "__main__":
unittest.main()
