This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 82cecf1c fix(table): stop answering COUNT(*) from a placeholder row
count (#624)
82cecf1c is described below
commit 82cecf1c617f7bbd394bfe163fb55acd76df320c
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Tue Jul 28 23:12:46 2026 +0800
fix(table): stop answering COUNT(*) from a placeholder row count (#624)
---
.../datafusion/tests/format_table_statistics.rs | 199 +++++++++++++++++++++
crates/paimon/src/spec/data_file.rs | 17 ++
crates/paimon/src/table/format_table_scan.rs | 2 +-
crates/paimon/src/table/source.rs | 78 +++++++-
4 files changed, 292 insertions(+), 4 deletions(-)
diff --git a/crates/integrations/datafusion/tests/format_table_statistics.rs
b/crates/integrations/datafusion/tests/format_table_statistics.rs
new file mode 100644
index 00000000..c5505fb1
--- /dev/null
+++ b/crates/integrations/datafusion/tests/format_table_statistics.rs
@@ -0,0 +1,199 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Statistics-driven aggregates over `type=format-table` tables.
+//!
+//! A format table has no manifest, so scan planning cannot fill in per-file
+//! row counts. Those counts must be reported as *unknown*; if the placeholder
+//! is reported as an exact statistic instead, DataFusion's
+//! `aggregate_statistics` rule answers `COUNT(*)` from it and never opens a
+//! single data file — a silent wrong answer.
+
+mod common;
+
+use std::path::Path;
+use std::sync::Arc;
+
+use arrow_array::{Int64Array, RecordBatch};
+use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+use paimon::catalog::{Catalog, Identifier};
+use paimon::spec::{BigIntType, DataType, Schema, SchemaBuilder, VarCharType};
+use paimon_datafusion::SQLContext;
+use parquet::arrow::ArrowWriter;
+use tempfile::TempDir;
+
+const DATABASE: &str = "test_db";
+const TABLE: &str = "events";
+
+/// Creates a `type=format-table` table in a filesystem catalog and returns the
+/// table directory, where callers drop raw data files.
+async fn setup_format_table() -> (TempDir, SQLContext, std::path::PathBuf) {
+ setup_table(Schema::builder().column("id",
DataType::BigInt(BigIntType::new()))).await
+}
+
+/// Same, partitioned by a single `dt` column (Hive layout: the partition value
+/// lives in the directory name, not in the data file).
+async fn setup_partitioned_format_table() -> (TempDir, SQLContext,
std::path::PathBuf) {
+ setup_table(
+ Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::new(32).unwrap()))
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .partition_keys(vec!["dt".to_string()]),
+ )
+ .await
+}
+
+async fn setup_table(builder: SchemaBuilder) -> (TempDir, SQLContext,
std::path::PathBuf) {
+ let (tmp, catalog) = common::create_test_env();
+ catalog
+ .create_database(DATABASE, false, Default::default())
+ .await
+ .expect("CREATE DATABASE failed");
+ let schema = builder
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap();
+ catalog
+ .create_table(&Identifier::new(DATABASE, TABLE), schema, false)
+ .await
+ .expect("CREATE TABLE failed");
+ let table_dir = tmp.path().join(format!("{DATABASE}.db")).join(TABLE);
+ let context = common::create_sql_context(catalog).await;
+ (tmp, context, table_dir)
+}
+
+fn write_parquet(path: &Path, values: &[i64]) {
+ let schema = Arc::new(ArrowSchema::new(vec![Field::new(
+ "id",
+ ArrowDataType::Int64,
+ true,
+ )]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int64Array::from(values.to_vec()))],
+ )
+ .unwrap();
+ let file = std::fs::File::create(path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, schema, None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+}
+
+async fn scalar_count(context: &SQLContext, sql: &str) -> i64 {
+ let batches = context.sql(sql).await.unwrap().collect().await.unwrap();
+ let batch = batches
+ .iter()
+ .find(|batch| batch.num_rows() > 0)
+ .expect("aggregate must return one row");
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("count column must be int64")
+ .value(0)
+}
+
+async fn scanned_rows(context: &SQLContext, sql: &str) -> i64 {
+ let batches = context.sql(sql).await.unwrap().collect().await.unwrap();
+ batches.iter().map(|batch| batch.num_rows() as i64).sum()
+}
+
+/// `COUNT(*)` on a format table must agree with what an actual scan returns.
+/// Regression for the silent `COUNT(*) = 0`: the placeholder
+/// `DataFileMeta::row_count` was reported as an exact statistic.
+#[tokio::test]
+async fn test_format_table_count_star_matches_scanned_rows() {
+ let (_tmp, context, table_dir) = setup_format_table().await;
+ write_parquet(&table_dir.join("part-0.parquet"), &[1, 2, 3]);
+ write_parquet(&table_dir.join("part-1.parquet"), &[4, 5]);
+
+ let scanned = scanned_rows(&context, "SELECT id FROM
paimon.test_db.events").await;
+ assert_eq!(scanned, 5, "the scan itself must see all rows");
+
+ let counted = scalar_count(&context, "SELECT COUNT(*) AS c FROM
paimon.test_db.events").await;
+ assert_eq!(
+ counted, scanned,
+ "COUNT(*) must not be answered from the placeholder row count"
+ );
+
+ let counted_col =
+ scalar_count(&context, "SELECT COUNT(id) AS c FROM
paimon.test_db.events").await;
+ assert_eq!(
+ counted_col, scanned,
+ "COUNT(col) must not be short-circuited"
+ );
+}
+
+/// A format table with no data files really has zero rows; reporting the row
+/// count as unknown must not turn that into a wrong answer either.
+#[tokio::test]
+async fn test_empty_format_table_counts_zero() {
+ let (_tmp, context, _table_dir) = setup_format_table().await;
+
+ assert_eq!(
+ scalar_count(&context, "SELECT COUNT(*) AS c FROM
paimon.test_db.events").await,
+ 0
+ );
+}
+
+/// The partition-pruned plan reaches `partition_statistics()` too, so a pruned
+/// `COUNT(*)` must also come from the data and not from the placeholder.
+///
+/// Only predicated queries are asserted here. An unfiltered scan over a
+/// partitioned format table on a `file://` warehouse currently finds no splits
+/// at all — `table_path` keeps the `file:///` form while the listed status
+/// paths come back as `file:/`, so the `strip_prefix` in
+/// `partition_row_from_path` (paimon/src/table/format_table_scan.rs:487-493)
+/// drops every file. That is a separate defect from the one under test.
+#[tokio::test]
+async fn test_partitioned_format_table_count_with_partition_predicate() {
+ let (_tmp, context, table_dir) = setup_partitioned_format_table().await;
+ for (dt, values) in [
+ ("2026-07-21", vec![1i64, 2, 3]),
+ ("2026-07-22", vec![4i64, 5]),
+ ] {
+ let partition_dir = table_dir.join(format!("dt={dt}"));
+ std::fs::create_dir_all(&partition_dir).unwrap();
+ write_parquet(&partition_dir.join("part-0.parquet"), &values);
+ }
+
+ for (dt, expected) in [("2026-07-21", 3), ("2026-07-22", 2),
("2026-07-23", 0)] {
+ let sql = format!("SELECT id FROM paimon.test_db.events WHERE dt =
'{dt}'");
+ let scanned = scanned_rows(&context, &sql).await;
+ assert_eq!(scanned, expected, "scan of dt={dt}");
+
+ let sql = format!("SELECT COUNT(*) AS c FROM paimon.test_db.events
WHERE dt = '{dt}'");
+ assert_eq!(
+ scalar_count(&context, &sql).await,
+ expected,
+ "count of dt={dt}"
+ );
+ }
+}
+
+/// A data file that genuinely holds zero rows must also count 0.
+#[tokio::test]
+async fn test_format_table_with_empty_file_counts_zero() {
+ let (_tmp, context, table_dir) = setup_format_table().await;
+ write_parquet(&table_dir.join("part-0.parquet"), &[]);
+
+ assert_eq!(
+ scalar_count(&context, "SELECT COUNT(*) AS c FROM
paimon.test_db.events").await,
+ 0
+ );
+}
diff --git a/crates/paimon/src/spec/data_file.rs
b/crates/paimon/src/spec/data_file.rs
index d2b901bd..314c6d15 100644
--- a/crates/paimon/src/spec/data_file.rs
+++ b/crates/paimon/src/spec/data_file.rs
@@ -47,6 +47,8 @@ pub struct DataFileMeta {
#[serde(rename = "_FILE_SIZE")]
pub file_size: i64,
// row_count tells the total number of rows (including add & delete) in
this file.
+ // A negative value means the producer does not know the row count; see
+ // [`DataFileMeta::ROW_COUNT_UNKNOWN`]. Never treat it as a real count.
#[serde(rename = "_ROW_COUNT")]
pub row_count: i64,
#[serde(rename = "_MIN_KEY", with = "serde_bytes")]
@@ -188,6 +190,21 @@ fn read_compact_millis_as_utc(
}
impl DataFileMeta {
+ /// Placeholder for `row_count` when the producer cannot know how many rows
+ /// a file holds.
+ ///
+ /// Metadata that comes from a manifest always carries a real count.
Sources
+ /// that plan over bare directory listings (for example
`type=format-table`,
+ /// which has no manifest) have nothing to fill in and must use this value:
+ /// a plain `0` is indistinguishable from a file that really is empty, and
+ /// every consumer that trusts it silently returns a wrong answer.
+ pub const ROW_COUNT_UNKNOWN: i64 = -1;
+
+ /// Whether `row_count` is a real count rather than
[`Self::ROW_COUNT_UNKNOWN`].
+ pub fn row_count_known(&self) -> bool {
+ self.row_count >= 0
+ }
+
/// Decode this file's manifest value statistics for a field in the
current schema.
///
/// Returns `None` when statistics are missing, malformed, or belong to a
different
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index 28907664..df2a48e0 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -619,7 +619,7 @@ fn data_file_meta(file_name: String, file_size: i64,
schema_id: i64) -> DataFile
DataFileMeta {
file_name,
file_size,
- row_count: 0,
+ row_count: DataFileMeta::ROW_COUNT_UNKNOWN,
min_key: Vec::new(),
max_key: Vec::new(),
key_stats: BinaryTableStats::empty(),
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 41836580..933cb441 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -561,9 +561,25 @@ impl DataSplit {
file.data_file_path(&self.bucket_path)
}
- /// Total row count of all data files in this split.
+ /// Sum of the physical row counts this split knows about.
+ ///
+ /// Files whose count is [`DataFileMeta::ROW_COUNT_UNKNOWN`] contribute
+ /// nothing, so the result is a lower bound, not a total. Ask
+ /// [`Self::row_counts_known`] before presenting it as one.
pub fn row_count(&self) -> i64 {
- self.data_files.iter().map(|f| f.row_count).sum()
+ self.data_files
+ .iter()
+ .filter(|f| f.row_count_known())
+ .map(|f| f.row_count)
+ .sum()
+ }
+
+ /// Whether every data file in this split carries a real row count.
+ ///
+ /// False for splits planned without a manifest (`type=format-table`),
where
+ /// the only honest answer about row counts is "unknown".
+ pub fn row_counts_known(&self) -> bool {
+ self.data_files.iter().all(DataFileMeta::row_count_known)
}
/// Returns the merged row count if it can be computed.
@@ -577,10 +593,15 @@ impl DataSplit {
/// 2. If all files have `first_row_id` (data evolution mode): merge
/// overlapping row ID ranges and take max row count per group.
///
- /// Returns `None` otherwise.
+ /// Returns `None` otherwise, and always `None` when any file's row count
is
+ /// [`DataFileMeta::ROW_COUNT_UNKNOWN`] — no arithmetic over a placeholder
+ /// produces a number a caller may trust.
///
/// Reference:
[DataSplit.mergedRowCount()](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/DataSplit.java#L133)
pub fn merged_row_count(&self) -> Option<i64> {
+ if !self.row_counts_known() {
+ return None;
+ }
if let Some(count) = self.raw_merged_row_count() {
return Some(count);
}
@@ -1367,6 +1388,57 @@ mod tests {
assert_eq!(s.merged_row_count(), Some(15));
}
+ /// A placeholder row count must surface as "unknown", not as a number.
+ /// Reporting `Some(0)` here lets an engine answer `COUNT(*)` with 0 for a
+ /// split that holds data.
+ #[test]
+ fn test_merged_row_count_unknown_when_a_file_has_no_row_count() {
+ let s = split(
+ vec![
+ file("a", DataFileMeta::ROW_COUNT_UNKNOWN, None),
+ file("b", DataFileMeta::ROW_COUNT_UNKNOWN, None),
+ ],
+ true,
+ );
+ assert!(!s.row_counts_known());
+ assert_eq!(s.merged_row_count(), None);
+ assert_eq!(s.row_count(), 0, "unknown files contribute nothing");
+ }
+
+ /// One unknown file poisons the whole split: the rest are still a lower
+ /// bound, never a total.
+ #[test]
+ fn test_merged_row_count_unknown_when_mixed_with_known_files() {
+ let s = split(
+ vec![
+ file("a", 10, None),
+ file("b", DataFileMeta::ROW_COUNT_UNKNOWN, None),
+ ],
+ true,
+ );
+ assert_eq!(s.merged_row_count(), None);
+ assert_eq!(s.row_count(), 10);
+ }
+
+ /// Unknown row counts must not be rescued by the data-evolution branch.
+ #[test]
+ fn test_merged_row_count_unknown_beats_data_evolution_branch() {
+ let s = split(
+ vec![file("a", DataFileMeta::ROW_COUNT_UNKNOWN, Some(0))],
+ false,
+ );
+ assert_eq!(s.merged_row_count(), None);
+ }
+
+ /// A file that really holds zero rows stays an exact 0 — "unknown" is a
+ /// distinct value, not a synonym for empty.
+ #[test]
+ fn test_merged_row_count_zero_rows_stays_exact() {
+ let s = split(vec![file("a", 0, None)], true);
+ assert!(s.row_counts_known());
+ assert_eq!(s.merged_row_count(), Some(0));
+ }
+
#[test]
fn test_fully_materialized_pk_dv_requires_compacted_files() {
let mut level_zero = file("a", 10, None);
