diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index 1bfaea0d1df..fc7aaa24dc6 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -522,13 +522,21 @@ impl ExecutionPlan for RegionScanExec { let statistics = if scanner.properties().total_rows_is_exact() && !scanner.has_predicate_without_region() { + // Only the row count is known. DataFusion answers `count(col)` as + // `num_rows - null_count` and `count(DISTINCT col)` as `distinct_count` + // without scanning, so these must stay absent unless the schema + // guarantees them. Non-nullable columns (including the time index that + // `count(*)` is rewritten to) keep an exact zero null count. let column_statistics = self .arrow_schema .fields .iter() - .map(|_| ColumnStatistics { - distinct_count: Precision::Exact(self.total_rows), - null_count: Precision::Exact(0), // all null rows are counted for append-only table + .map(|field| ColumnStatistics { + null_count: if field.is_nullable() { + Precision::Absent + } else { + Precision::Exact(0) + }, ..Default::default() }) .collect(); @@ -1218,7 +1226,8 @@ mod test { (true, true, Precision::Exact(3)), (false, true, Precision::Absent), ] { - let (schema, batch, metadata) = dynamic_filter_test_data(5685, false); + // Column `a` is nullable and `ts` is not. + let (schema, batch, metadata) = dynamic_filter_test_data(5685, true); let scanner = Box::new(RepeatableScanner { batches: RecordBatches::try_new(schema, vec![batch]).unwrap(), properties: ScannerProperties::default() @@ -1229,13 +1238,19 @@ mod test { dynamic_filters: Arc::new(Mutex::new(Vec::new())), }); let plan = RegionScanExec::new(scanner, ScanRequest::default(), None).unwrap(); - assert_eq!( - StatisticsContext::new() - .compute(&plan, &StatisticsArgs::new()) - .unwrap() - .num_rows, - expected, - ); + let stats = StatisticsContext::new() + .compute(&plan, &StatisticsArgs::new()) + .unwrap(); + assert_eq!(stats.num_rows, expected); + if expected != Precision::Absent { + let [a, ts] = stats.column_statistics.as_slice() else { + panic!("unexpected column statistics: {stats:?}"); + }; + assert_eq!(a.null_count, Precision::Absent); + assert_eq!(ts.null_count, Precision::Exact(0)); + assert_eq!(a.distinct_count, Precision::Absent); + assert_eq!(ts.distinct_count, Precision::Absent); + } assert_eq!( StatisticsContext::new() .compute(&plan, &StatisticsArgs::new().with_partition(Some(0))) diff --git a/tests/cases/standalone/common/aggregate/append_count_nulls.result b/tests/cases/standalone/common/aggregate/append_count_nulls.result new file mode 100644 index 00000000000..acb4714224b --- /dev/null +++ b/tests/cases/standalone/common/aggregate/append_count_nulls.result @@ -0,0 +1,165 @@ +-- count(col) and count(DISTINCT col) on append-only tables must not be answered +-- from row-count statistics, while count(*) still is. +CREATE TABLE n1 (ts TIMESTAMP TIME INDEX, v BIGINT, s STRING) WITH (append_mode='true'); + +Affected Rows: 0 + +INSERT INTO n1 VALUES (1000, 1, 'a'), (2000, NULL, 'a'), (3000, NULL, NULL), (4000, 4, 'b'); + +Affected Rows: 4 + +SELECT count(*), count(v), count(s), count(DISTINCT s) FROM n1; + ++----------+-------------+-------------+----------------------+ +| count(*) | count(n1.v) | count(n1.s) | count(DISTINCT n1.s) | ++----------+-------------+-------------+----------------------+ +| 4 | 2 | 3 | 2 | ++----------+-------------+-------------+----------------------+ + +SELECT count(v) FROM n1; + ++-------------+ +| count(n1.v) | ++-------------+ +| 2 | ++-------------+ + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n1; + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_CooperativeExec REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_ProjectionExec: expr=[4 as count(*)] REDACTED +|_|_|_PlaceholderRowExec REDACTED +|_|_|_| +|_|_| Total rows: 1_| ++-+-+-+ + +ADMIN FLUSH_TABLE('n1'); + ++-------------------------+ +| ADMIN FLUSH_TABLE('n1') | ++-------------------------+ +| 0 | ++-------------------------+ + +SELECT count(*), count(v), count(s), count(DISTINCT s) FROM n1; + ++----------+-------------+-------------+----------------------+ +| count(*) | count(n1.v) | count(n1.s) | count(DISTINCT n1.s) | ++----------+-------------+-------------+----------------------+ +| 4 | 2 | 3 | 2 | ++----------+-------------+-------------+----------------------+ + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n1; + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_CooperativeExec REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_ProjectionExec: expr=[4 as count(*)] REDACTED +|_|_|_PlaceholderRowExec REDACTED +|_|_|_| +|_|_| Total rows: 1_| ++-+-+-+ + +DROP TABLE n1; + +Affected Rows: 0 + +CREATE TABLE n2 ( + host STRING, + ts TIMESTAMP TIME INDEX, + v BIGINT, + s STRING, + PRIMARY KEY (host) +) +PARTITION ON COLUMNS (host) ( + host < 'm', + host >= 'm' +) +WITH (append_mode='true'); + +Affected Rows: 0 + +INSERT INTO n2 VALUES + ('a', 1000, 1, 'a'), + ('b', 2000, NULL, 'a'), + ('x', 3000, NULL, NULL), + ('y', 4000, 4, 'b'); + +Affected Rows: 4 + +-- count is pushed down to each region as a state aggregate. +SELECT count(v), count(s) FROM n2; + ++-------------+-------------+ +| count(n2.v) | count(n2.s) | ++-------------+-------------+ +| 2 | 3 | ++-------------+-------------+ + +ADMIN FLUSH_TABLE('n2'); + ++-------------------------+ +| ADMIN FLUSH_TABLE('n2') | ++-------------------------+ +| 0 | ++-------------------------+ + +SELECT count(v), count(s) FROM n2; + ++-------------+-------------+ +| count(n2.v) | count(n2.s) | ++-------------+-------------+ +| 2 | 3 | ++-------------+-------------+ + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n2; + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_ProjectionExec: expr=[count(Int64(1))@0 as count(*)] REDACTED +|_|_|_AggregateExec: mode=Final, gby=[], aggr=[__count_merge(__count_state(n2.ts)) as count(Int64(1))] REDACTED +|_|_|_CoalescePartitionsExec REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__count_merge(__count_state(n2.ts)) as count(Int64(1))] REDACTED +|_|_|_RepartitionExec: partitioning=REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_ProjectionExec: expr=[{count[count]:2} as __count_state(n2.ts)] REDACTED +|_|_|_PlaceholderRowExec REDACTED +|_|_|_| +| 1_| 1_|_ProjectionExec: expr=[{count[count]:2} as __count_state(n2.ts)] REDACTED +|_|_|_PlaceholderRowExec REDACTED +|_|_|_| +|_|_| Total rows: 1_| ++-+-+-+ + +DROP TABLE n2; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/aggregate/append_count_nulls.sql b/tests/cases/standalone/common/aggregate/append_count_nulls.sql new file mode 100644 index 00000000000..8f2e1f5c2f7 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/append_count_nulls.sql @@ -0,0 +1,67 @@ +-- count(col) and count(DISTINCT col) on append-only tables must not be answered +-- from row-count statistics, while count(*) still is. +CREATE TABLE n1 (ts TIMESTAMP TIME INDEX, v BIGINT, s STRING) WITH (append_mode='true'); + +INSERT INTO n1 VALUES (1000, 1, 'a'), (2000, NULL, 'a'), (3000, NULL, NULL), (4000, 4, 'b'); + +SELECT count(*), count(v), count(s), count(DISTINCT s) FROM n1; + +SELECT count(v) FROM n1; + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n1; + +ADMIN FLUSH_TABLE('n1'); + +SELECT count(*), count(v), count(s), count(DISTINCT s) FROM n1; + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n1; + +DROP TABLE n1; + +CREATE TABLE n2 ( + host STRING, + ts TIMESTAMP TIME INDEX, + v BIGINT, + s STRING, + PRIMARY KEY (host) +) +PARTITION ON COLUMNS (host) ( + host < 'm', + host >= 'm' +) +WITH (append_mode='true'); + +INSERT INTO n2 VALUES + ('a', 1000, 1, 'a'), + ('b', 2000, NULL, 'a'), + ('x', 3000, NULL, NULL), + ('y', 4000, 4, 'b'); + +-- count is pushed down to each region as a state aggregate. +SELECT count(v), count(s) FROM n2; + +ADMIN FLUSH_TABLE('n2'); + +SELECT count(v), count(s) FROM n2; + +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +EXPLAIN ANALYZE SELECT count(*) FROM n2; + +DROP TABLE n2;