From fd3faeb7ed9653e1dae77308eef5d0ccdc0f0c4b Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:32:32 -0700 Subject: [PATCH] fix: backport scan statistics and partitioning fixes to release/5e02efa9 (#9783) * fix: preserve count correctness after repartition (#9154) * fix: preserve count correctness after repartition Signed-off-by: WenyXu * fix: restore full predicate guard for count statistics Signed-off-by: WenyXu * test: update count plans for full predicate guard Signed-off-by: WenyXu * fix: preserve count statistics for safe partition scans Signed-off-by: WenyXu * test: retain repartition home guard and cover staging flush Signed-off-by: WenyXu --------- Signed-off-by: WenyXu (cherry picked from commit d9b97796a6c3fa884c5ffab37d6d192b2b5c1e99) Signed-off-by: Lei, HUANG * fix: only push down aggregates grouped by the partition columns themselves (#9337) * fix: only push down aggregates grouped by the partition columns themselves Signed-off-by: Dennis Zhuang * fix: keep grouping sets on the frontend for partitioned tables Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang (cherry picked from commit def0c2ec5ca0992083d955f6bcbc5ffaf914fe90) Signed-off-by: Lei, HUANG * fix(table): require complete primary keys for hash partitioning (#9507) Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> (cherry picked from commit 28ec3ff5aad8d274d83ae42f96dc54f838fb6b20) Signed-off-by: Lei, HUANG * fix(table): stop claiming exact null and distinct counts for append scans (#9593) RegionScanExec reported null_count = 0 and distinct_count = total_rows for every column of an append-only table. DataFusion answers count(col) and count(DISTINCT col) from these statistics, so both returned the row count. Keep the exact row count and an exact zero null count for non-nullable columns, so count(*) is still answered from statistics. Signed-off-by: Dennis Zhuang (cherry picked from commit 5cfdc30a696b864fff75a078dcdecfb9ea74d212) Signed-off-by: Lei, HUANG * fix(table): stop claiming tag order for logical metric scans (#9522) * test: cover metric logical scan tag ordering Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(table): only claim projected primary key prefix ordering Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(table): keep physical key ordering for metric projections Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(table): compare unknown partition counts explicitly Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(table): model scoped internal TSID scan properties Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> (cherry picked from commit 75bc4ce46d0e7aedca53dc2b0375ca0658456ff6) Signed-off-by: Lei, HUANG * test: regenerate append_count_nulls result for release plan rendering Signed-off-by: Lei, HUANG --------- Signed-off-by: WenyXu Signed-off-by: Lei, HUANG Signed-off-by: Dennis Zhuang Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Co-authored-by: Weny Xu Co-authored-by: dennis zhuang Co-authored-by: discord9 --- src/mito2/src/read/scan_region.rs | 58 +++ src/mito2/src/read/seq_scan.rs | 3 +- src/mito2/src/read/series_scan.rs | 3 +- src/mito2/src/read/unordered_scan.rs | 3 +- src/query/src/dist_plan/analyzer/test.rs | 35 ++ src/query/src/dist_plan/commutativity.rs | 37 +- src/store-api/src/region_engine.rs | 15 + src/table/src/table/scan.rs | 461 ++++++++++++++++-- tests-integration/src/tests/instance_test.rs | 289 +++++++++++ tests-integration/tests/repartition.rs | 185 ++++++- .../aggregate/append_count_nulls.result | 165 +++++++ .../common/aggregate/append_count_nulls.sql | 67 +++ .../common/aggregate/multi_regions.result | 48 ++ .../common/aggregate/multi_regions.sql | 16 + .../aggregate/projected_pk_group_by.result | 69 +++ .../aggregate/projected_pk_group_by.sql | 33 ++ 16 files changed, 1442 insertions(+), 45 deletions(-) create mode 100644 tests/cases/standalone/common/aggregate/append_count_nulls.result create mode 100644 tests/cases/standalone/common/aggregate/append_count_nulls.sql create mode 100644 tests/cases/standalone/common/aggregate/projected_pk_group_by.result create mode 100644 tests/cases/standalone/common/aggregate/projected_pk_group_by.sql diff --git a/src/mito2/src/read/scan_region.rs b/src/mito2/src/read/scan_region.rs index c38e5c9c596..aea8aa19347 100644 --- a/src/mito2/src/read/scan_region.rs +++ b/src/mito2/src/read/scan_region.rs @@ -1688,6 +1688,27 @@ impl ScanInput { ); } + /// Whether physical source row counts equal the rows visible to this region, + /// before applying query predicates. + pub(crate) fn total_rows_is_exact(&self) -> bool { + if self.region_partition_expr.is_none() { + return true; + } + + // Extension ranges do not expose their partition expressions. + #[cfg(feature = "enterprise")] + if !self.extension_ranges.is_empty() { + return false; + } + + // Repartition flushes existing memtables before entering staging. New + // writes are routed under the target partition rules, whereas historical + // SSTs can be shared by regions and still need partition filtering. + self.files + .iter() + .all(|file| self.should_skip_region_partition(file)) + } + pub(crate) fn total_rows(&self) -> usize { let rows_in_files: usize = self.files.iter().map(|f| f.num_rows()).sum(); let rows_in_memtables: usize = self.memtables.iter().map(|m| m.stats().num_rows()).sum(); @@ -2325,6 +2346,43 @@ mod tests { .with_files(vec![file]) } + #[tokio::test] + async fn test_total_rows_is_exact_after_partition_filter() { + let expr = partition_col("k0").gt_eq(Value::String("foo".into())); + let other = partition_col("k0").gt_eq(Value::String("bar".into())); + for (region_expr, file_exprs, exact) in [ + (None, vec![None], true), + (None, vec![Some(expr.clone())], true), + (Some(expr.clone()), vec![], true), + (Some(expr.clone()), vec![Some(expr.clone())], true), + (Some(expr.clone()), vec![None], false), + (Some(expr.clone()), vec![Some(other)], false), + (Some(expr.clone()), vec![Some(expr), None], false), + ] { + let mut builder = + RegionMetadataBuilder::from_existing(metadata_with_primary_key(vec![0, 1], false)); + builder.partition_expr_json(region_expr.map(|expr| expr.as_json_str().unwrap())); + let metadata = Arc::new(builder.build_without_validation().unwrap()); + let files = file_exprs + .into_iter() + .map(|partition_expr| { + FileHandle::new( + FileMeta { + partition_expr, + ..Default::default() + }, + Arc::new(crate::sst::file_purger::NoopFilePurger), + ) + }) + .collect(); + let input = new_scan_input(metadata, vec![]) + .await + .with_files(files) + .build(); + assert_eq!(input.total_rows_is_exact(), exact); + } + } + /// Helper to create a timestamp millisecond literal. fn ts_lit(val: i64) -> datafusion_expr::Expr { lit(ScalarValue::TimestampMillisecond(Some(val), None)) diff --git a/src/mito2/src/read/seq_scan.rs b/src/mito2/src/read/seq_scan.rs index 29ea46f0f12..323704c31e2 100644 --- a/src/mito2/src/read/seq_scan.rs +++ b/src/mito2/src/read/seq_scan.rs @@ -76,7 +76,8 @@ impl SeqScan { pub(crate) fn new(input: ScanInput) -> Self { let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) - .with_total_rows(input.total_rows()); + .with_total_rows(input.total_rows()) + .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact()); if let Some(counters) = input.query_stat_counters.clone() { properties.set_query_stat_counters(counters); } diff --git a/src/mito2/src/read/series_scan.rs b/src/mito2/src/read/series_scan.rs index ca1d31b8804..bec48a14d8d 100644 --- a/src/mito2/src/read/series_scan.rs +++ b/src/mito2/src/read/series_scan.rs @@ -147,7 +147,8 @@ impl SeriesScan { }; let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) - .with_total_rows(input.total_rows()); + .with_total_rows(input.total_rows()) + .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact()); if let Some(counters) = input.query_stat_counters.clone() { properties.set_query_stat_counters(counters); } diff --git a/src/mito2/src/read/unordered_scan.rs b/src/mito2/src/read/unordered_scan.rs index 71e6b0a8acc..8bc4f2e0038 100644 --- a/src/mito2/src/read/unordered_scan.rs +++ b/src/mito2/src/read/unordered_scan.rs @@ -61,7 +61,8 @@ impl UnorderedScan { pub(crate) fn new(input: ScanInput) -> Self { let mut properties = ScannerProperties::default() .with_append_mode(input.append_mode) - .with_total_rows(input.total_rows()); + .with_total_rows(input.total_rows()) + .with_total_rows_is_exact(input.append_mode && input.total_rows_is_exact()); if let Some(counters) = input.query_stat_counters.clone() { properties.set_query_stat_counters(counters); } diff --git a/src/query/src/dist_plan/analyzer/test.rs b/src/query/src/dist_plan/analyzer/test.rs index c42eb0e285a..58e76deccfb 100644 --- a/src/query/src/dist_plan/analyzer/test.rs +++ b/src/query/src/dist_plan/analyzer/test.rs @@ -38,6 +38,7 @@ use datafusion_expr::{ }; use datafusion_functions::datetime::date_bin; use datafusion_functions::datetime::expr_fn::now; +use datafusion_functions::unicode::expr_fn::substring; use datafusion_sql::TableReference; use datatypes::data_type::ConcreteDataType; use datatypes::schema::{ColumnSchema, SchemaBuilder, SchemaRef}; @@ -1094,6 +1095,40 @@ fn expand_proj_alias_aliased_part_col_aggr() { assert_eq!(expected, result.to_string()); } +/// `substr(pk1, 1, 1)` maps rows from different partitions to the same group, so the +/// aggregate has to be merged on the frontend even though it references `pk1`. +#[test] +fn expand_expr_over_part_col_aggr() { + init_default_ut_logging(); + let test_table = TestTable::table_with_name(0, "t".to_string()); + let table_source = Arc::new(DefaultTableSource::new(Arc::new( + DfTableProviderAdapter::new(test_table), + ))); + let plan = LogicalPlanBuilder::scan_with_filters("t", table_source, None, vec![]) + .unwrap() + .aggregate( + vec![substring(col("pk1"), lit(1i64), lit(1i64)), col("pk2")], + vec![min(col("number"))], + ) + .unwrap() + .build() + .unwrap(); + + let config = ConfigOptions::default(); + let result = DistPlannerAnalyzer {}.analyze(plan, &config).unwrap(); + + let expected = [ + "Projection: substr(t.pk1,Int64(1),Int64(1)), t.pk2, min(t.number)", + " Aggregate: groupBy=[[substr(t.pk1,Int64(1),Int64(1)), t.pk2]], aggr=[[__min_merge(__min_state(t.number)) AS min(t.number)]]", + " MergeScan [is_placeholder=false, remote_input=[", + "Aggregate: groupBy=[[substr(t.pk1, Int64(1), Int64(1)), t.pk2]], aggr=[[__min_state(t.number)]]", + " TableScan: t", + "]]", + ] + .join("\n"); + assert_eq!(expected, result.to_string()); +} + /// notice that step aggr then part col aggr seems impossible as the partition columns for part col aggr /// can't pass through the step aggr without making step aggr also a part col aggr /// so here only test part col aggr -> step aggr case diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index 1a70fd8e1d9..1276debcb25 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -156,7 +156,14 @@ impl Categorizer { LogicalPlan::Filter(filter) => Self::check_expr(&filter.predicate), LogicalPlan::Window(_) => Commutativity::Unimplemented, LogicalPlan::Aggregate(aggr) => { - let is_all_steppable = is_all_aggr_exprs_steppable(&aggr.aggr_expr); + // The state/merge split maps each group expression to one output column, + // which doesn't hold for grouping sets. + let has_grouping_set = aggr + .group_expr + .iter() + .any(|expr| matches!(expr, Expr::GroupingSet(_))); + let is_all_steppable = + !has_grouping_set && is_all_aggr_exprs_steppable(&aggr.aggr_expr); let matches_partition = Self::check_partition(&aggr.group_expr, &partition_cols); if !matches_partition && is_all_steppable { debug!("Plan is steppable: {plan}"); @@ -345,26 +352,34 @@ impl Categorizer { /// Return true if the given expr and partition cols satisfied the rule. /// In this case the plan can be treated as fully commutative. /// - /// So only if all partition columns show up in `exprs`, return true. + /// So only if every partition column is itself one of `exprs`, return true. /// Otherwise return false. /// + /// An expression that only references a partition column, like `substr(host, 3, 1)` + /// or `k % 2`, doesn't count: it can put rows from different partitions into the same + /// group. fn check_partition(exprs: &[Expr], partition_cols: &AliasMapping) -> bool { - let mut ref_cols = HashSet::new(); - for expr in exprs { - expr.add_column_refs(&mut ref_cols); - } - let ref_cols = ref_cols - .into_iter() - .map(|c| c.name.clone()) + let group_cols = exprs + .iter() + .filter_map(|expr| { + let mut expr = expr; + while let Expr::Alias(alias) = expr { + expr = &alias.expr; + } + match expr { + Expr::Column(column) => Some(column.name.clone()), + _ => None, + } + }) .collect::>(); for all_alias in partition_cols.values() { let all_alias = all_alias .iter() .map(|c| c.name.clone()) .collect::>(); - // check if ref columns intersect with all alias of partition columns + // check if group columns intersect with all alias of partition columns // is empty, if it's empty, not all partition columns show up in `exprs` - if ref_cols.intersection(&all_alias).count() == 0 { + if group_cols.intersection(&all_alias).count() == 0 { return false; } } diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index 1a4351f279b..6a45d3db7d7 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -306,6 +306,9 @@ pub struct ScannerProperties { /// [ScannerProperties::append_mode] is true. total_rows: usize, + /// Whether total_rows is exact before applying query predicates. + total_rows_is_exact: bool, + /// Whether to yield an empty batch to distinguish partition ranges. pub distinguish_partition_range: bool, @@ -334,12 +337,19 @@ impl ScannerProperties { self } + /// Sets whether the source row count is exact before query predicates. + pub fn with_total_rows_is_exact(mut self, exact: bool) -> Self { + self.total_rows_is_exact = exact; + self + } + /// Creates a new [`ScannerProperties`] with the given partitioning. pub fn new(partitions: Vec>, append_mode: bool, total_rows: usize) -> Self { Self { partitions, append_mode, total_rows, + total_rows_is_exact: false, distinguish_partition_range: false, target_partitions: 0, logical_region: false, @@ -374,6 +384,11 @@ impl ScannerProperties { self.total_rows } + /// Returns whether the source row count is exact before query predicates. + pub fn total_rows_is_exact(&self) -> bool { + self.total_rows_is_exact + } + /// Returns whether the scanner is scanning a logical region. pub fn is_logical_region(&self) -> bool { self.logical_region diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index d1417bf7a95..4c04952c7ef 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -118,21 +118,34 @@ impl RegionScanExec { } let metadata = scanner.metadata(); - let mut pk_names = metadata + // Keep the primary key columns in the order reported by the metadata (for a metric + // region that is the physical row key order). Reordering them would let a projected + // logical tag look like a prefix of the row key, which yields ordering and + // partitioning claims the scanner cannot honour. + let pk_names = metadata .primary_key_columns() .map(|col| col.column_schema.name.clone()) .collect::>(); - // workaround for logical table - if scanner.properties().is_logical_region() { - pk_names.sort_unstable(); - } + // A region emits rows ordered by its primary key columns and then by the time + // index column. A primary key column that is not projected is absent from the + // arrow schema, and it also breaks the ordering of every primary key column that + // follows it (e.g. the physical `__table_id`/`__tsid` prefix of a metric region + // when reading a logical table). Thus only the longest fully projected prefix of + // the primary key can describe the output ordering, and that prefix can be empty. + let projected_pk_prefix_len = pk_names + .iter() + .position(|name| Column::new_with_schema(name, &arrow_schema).is_err()) + .unwrap_or(pk_names.len()); + // Whether the projected prefix covers the whole row key. + let all_pk_projected = projected_pk_prefix_len == pk_names.len(); + let projected_pk_names = &pk_names[..projected_pk_prefix_len]; let pk_columns = pk_names .iter() .filter_map( |col| Some(Arc::new(Column::new_with_schema(col, &arrow_schema).ok()?) as _), ) .collect::>(); - let pk_sort_columns: Vec = pk_names + let pk_sort_columns: Vec = projected_pk_names .iter() .filter_map(|col| { Some(PhysicalSortExpr::new( @@ -165,7 +178,9 @@ impl RegionScanExec { let mut orderings = Vec::with_capacity(2); let mut pk_time_ordering = pk_sort_columns.clone(); - if let Some(ts) = ts_col.clone() { + // The time index column extends the ordering only when the whole row key + // precedes it; otherwise rows of different series can interleave. + if all_pk_projected && let Some(ts) = ts_col.clone() { pk_time_ordering.push(ts); } orderings.push(pk_time_ordering); @@ -213,7 +228,10 @@ impl RegionScanExec { Ok(tsid_col) => { Partitioning::Hash(vec![Arc::new(tsid_col) as _], num_output_partition) } - Err(_) => Partitioning::Hash(pk_columns.clone(), num_output_partition), + Err(_) if pk_columns.len() == pk_names.len() => { + Partitioning::Hash(pk_columns.clone(), num_output_partition) + } + Err(_) => Partitioning::UnknownPartitioning(num_output_partition), } } Some(TimeSeriesDistribution::TimeWindowed) | None => { @@ -482,30 +500,40 @@ impl ExecutionPlan for RegionScanExec { } fn partition_statistics(&self, partition: Option) -> DfResult { - if partition.is_some() { + if partition.is_some() || !self.append_mode { return Ok(Statistics::new_unknown(self.schema().as_ref())); } - let statistics = - if self.append_mode && !self.scanner.lock().unwrap().has_predicate_without_region() { - 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 - ..Default::default() - }) - .collect(); - Statistics { - num_rows: Precision::Exact(self.total_rows), - total_byte_size: Default::default(), - column_statistics, - } - } else { - Statistics::new_unknown(&self.arrow_schema) - }; + let scanner = self.scanner.lock().unwrap(); + 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(|field| ColumnStatistics { + null_count: if field.is_nullable() { + Precision::Absent + } else { + Precision::Exact(0) + }, + ..Default::default() + }) + .collect(); + Statistics { + num_rows: Precision::Exact(self.total_rows), + total_byte_size: Default::default(), + column_statistics, + } + } else { + Statistics::new_unknown(&self.arrow_schema) + }; Ok(statistics) } @@ -679,10 +707,12 @@ mod test { use datatypes::vectors::{Int32Vector, TimestampMillisecondVector, VectorRef}; use futures::TryStreamExt; use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder}; + use store_api::metric_engine_consts::DATA_SCHEMA_TABLE_ID_COLUMN_NAME; use store_api::region_engine::{ PrepareRequest, QueryScanContext, RegionScanner, ScannerProperties, SinglePartitionScanner, }; - use store_api::storage::RegionId; + use store_api::storage::consts::ReservedColumnId; + use store_api::storage::{RegionId, TimeSeriesDistribution}; use super::*; @@ -845,6 +875,377 @@ mod test { ) } + #[test] + fn test_per_series_partitioning_requires_all_primary_keys() { + let full_schema = Arc::new(Schema::new(vec![ + ColumnSchema::new("marker", ConcreteDataType::int32_datatype(), false), + ColumnSchema::new("host", ConcreteDataType::string_datatype(), false), + ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ])); + let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5683)); + builder + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "marker", + ConcreteDataType::int32_datatype(), + false, + ), + semantic_type: SemanticType::Tag, + column_id: 1, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "host", + ConcreteDataType::string_datatype(), + false, + ), + semantic_type: SemanticType::Tag, + column_id: 2, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 3, + }) + .primary_key(vec![1, 2]); + let metadata = Arc::new(builder.build().unwrap()); + + for (schema, expected) in [ + ( + full_schema.clone(), + Partitioning::Hash( + vec![ + Arc::new( + Column::new_with_schema("marker", full_schema.arrow_schema()).unwrap(), + ) as _, + Arc::new( + Column::new_with_schema("host", full_schema.arrow_schema()).unwrap(), + ) as _, + ], + 2, + ), + ), + ( + Arc::new(Schema::new(vec![ + ColumnSchema::new("marker", ConcreteDataType::int32_datatype(), false), + ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ])), + Partitioning::UnknownPartitioning(2), + ), + ] { + let batches = RecordBatches::try_new(schema, vec![]).unwrap(); + let scanner = Box::new(RepeatableScanner { + batches, + properties: ScannerProperties::new(vec![vec![], vec![]], false, 0), + metadata: metadata.clone(), + dynamic_filters: Arc::new(Mutex::new(Vec::new())), + }); + let request = ScanRequest { + distribution: Some(TimeSeriesDistribution::PerSeries), + ..Default::default() + }; + let plan = RegionScanExec::new(scanner, request, None).unwrap(); + let actual = plan.properties().output_partitioning(); + match (&actual, &expected) { + ( + Partitioning::UnknownPartitioning(actual_count), + Partitioning::UnknownPartitioning(expected_count), + ) => assert_eq!(actual_count, expected_count), + _ => assert_eq!(actual, &expected), + } + } + } + + /// Returns whether the scan claims the given output ordering. + fn claims_ordering(plan: &RegionScanExec, names: &[&str]) -> bool { + plan.properties() + .equivalence_properties() + .ordering_satisfy(names.iter().map(|name| { + PhysicalSortExpr::new( + Arc::new(Column::new_with_schema(name, &plan.arrow_schema).unwrap()), + SortOptions { + descending: false, + nulls_first: true, + }, + ) + })) + .unwrap() + } + + /// Column schema used by the ordering fixtures below. + fn column_schema_of(name: &str) -> ColumnSchema { + let data_type = match name { + "ts" => ConcreteDataType::timestamp_millisecond_datatype(), + DATA_SCHEMA_TABLE_ID_COLUMN_NAME => ConcreteDataType::uint32_datatype(), + DATA_SCHEMA_TSID_COLUMN_NAME => ConcreteDataType::uint64_datatype(), + _ => ConcreteDataType::string_datatype(), + }; + ColumnSchema::new(name, data_type, false) + } + + /// The scan must describe its output ordering with the primary key order reported by + /// the metadata. Sorting the names (the removed logical-table workaround) would let a + /// projected column look like a row key prefix and claim an ordering the scanner does + /// not produce; the time index only extends a fully projected row key. + #[test] + fn test_per_series_ordering_follows_metadata_pk_order() { + // Deliberately not alphabetical: sorting these names would give `[a, b, z]`. + let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5690)); + for (index, name) in ["z", "a", "b", "ts"].into_iter().enumerate() { + builder.push_column_metadata(ColumnMetadata { + column_schema: column_schema_of(name), + semantic_type: if name == "ts" { + SemanticType::Timestamp + } else { + SemanticType::Tag + }, + column_id: index as u32 + 1, + }); + } + builder.primary_key(vec![1, 2, 3]); + let metadata = Arc::new(builder.build().unwrap()); + + // The claims must not depend on the scanner reporting a logical region. + for logical_region in [false, true] { + for (projection, claimed, unclaimed) in [ + // Missing the first key: neither the remaining keys nor the time index are + // ordered. + ( + vec!["a", "b", "ts"], + vec![], + vec![vec!["a"], vec!["b"], vec!["ts"], vec!["a", "b"]], + ), + // Missing a key in the middle: only the prefix before it is ordered. + ( + vec!["z", "b", "ts"], + vec![vec!["z"]], + vec![vec!["b"], vec!["z", "b"], vec!["z", "ts"]], + ), + // A valid but incomplete prefix: the time index is not ordered after it. + ( + vec!["z", "a", "ts"], + vec![vec!["z"], vec!["z", "a"]], + vec![vec!["a", "z"], vec!["z", "a", "ts"]], + ), + // The whole row key followed by the time index. + ( + vec!["z", "a", "b", "ts"], + vec![vec!["z", "a", "b"], vec!["z", "a", "b", "ts"]], + vec![vec!["a"], vec!["a", "b", "z"]], + ), + ] { + let schema = Arc::new(Schema::new( + projection + .iter() + .map(|name| column_schema_of(name)) + .collect(), + )); + let mut properties = ScannerProperties::new(vec![vec![], vec![]], false, 0); + properties.set_logical_region(logical_region); + let scanner = Box::new(RepeatableScanner { + batches: RecordBatches::try_new(schema, vec![]).unwrap(), + properties, + metadata: metadata.clone(), + dynamic_filters: Arc::new(Mutex::new(Vec::new())), + }); + let plan = RegionScanExec::new( + scanner, + ScanRequest { + distribution: Some(TimeSeriesDistribution::PerSeries), + ..Default::default() + }, + None, + ) + .unwrap(); + + for ordering in claimed { + assert!( + claims_ordering(&plan, &ordering), + "logical_region={logical_region} projection {projection:?} must claim \ + ordering {ordering:?}" + ); + } + for ordering in unclaimed { + assert!( + !claims_ordering(&plan, &ordering), + "logical_region={logical_region} projection {projection:?} must not \ + claim ordering {ordering:?}" + ); + } + } + } + } + + /// Metric TSID properties require projecting `__tsid` and restricting the scan to one + /// `__table_id`; hiding the physical row-key prefix must not claim a visible tag or time as + /// order or hash partition columns. + #[test] + fn test_metric_projection_claims_only_projected_row_key_prefix() { + let mut builder = RegionMetadataBuilder::new(RegionId::new(1234, 5691)); + for (name, column_id) in [ + ( + DATA_SCHEMA_TABLE_ID_COLUMN_NAME, + ReservedColumnId::table_id(), + ), + (DATA_SCHEMA_TSID_COLUMN_NAME, ReservedColumnId::tsid()), + ("A", 1), + ("marker", 2), + ("ts", 3), + ] { + builder.push_column_metadata(ColumnMetadata { + column_schema: column_schema_of(name), + semantic_type: if name == "ts" { + SemanticType::Timestamp + } else { + SemanticType::Tag + }, + column_id, + }); + } + // The physical row key order: `__table_id`, `__tsid`, then the tag columns. + builder.primary_key(vec![ + ReservedColumnId::table_id(), + ReservedColumnId::tsid(), + 1, + 2, + ]); + let metadata = Arc::new(builder.build().unwrap()); + + for (projection, partitioned_by_tsid, claimed, unclaimed) in [ + // The logical projection drops the row key prefix, so `A` is not a prefix of the + // row key and neither `A` nor `ts` is ordered. + ( + vec!["A", "marker", "ts"], + false, + vec![], + vec![vec!["A"], vec!["marker"], vec!["ts"], vec!["A", "ts"]], + ), + // Projecting internal `__tsid` plus the single-table filter preserves metric TSID properties. + ( + vec![DATA_SCHEMA_TSID_COLUMN_NAME, "A", "marker", "ts"], + true, + vec![ + vec![DATA_SCHEMA_TSID_COLUMN_NAME], + vec![DATA_SCHEMA_TSID_COLUMN_NAME, "ts"], + ], + vec![vec!["A"], vec!["ts"], vec!["A", "ts"]], + ), + ] { + let schema = Arc::new(Schema::new( + projection + .iter() + .map(|name| column_schema_of(name)) + .collect(), + )); + let mut properties = ScannerProperties::new(vec![vec![], vec![]], false, 0); + properties.set_logical_region(true); + let scanner = Box::new(RepeatableScanner { + batches: RecordBatches::try_new(schema, vec![]).unwrap(), + properties, + metadata: metadata.clone(), + dynamic_filters: Arc::new(Mutex::new(Vec::new())), + }); + let plan = RegionScanExec::new( + scanner, + ScanRequest { + distribution: Some(TimeSeriesDistribution::PerSeries), + filters: if partitioned_by_tsid { + vec![ + datafusion_expr::col(DATA_SCHEMA_TABLE_ID_COLUMN_NAME) + .eq(datafusion_expr::lit(42u32)), + ] + } else { + vec![] + }, + ..Default::default() + }, + None, + ) + .unwrap(); + + let expected_partitioning = if partitioned_by_tsid { + Partitioning::Hash( + vec![Arc::new( + Column::new_with_schema(DATA_SCHEMA_TSID_COLUMN_NAME, &plan.arrow_schema) + .unwrap(), + ) as _], + 2, + ) + } else { + Partitioning::UnknownPartitioning(2) + }; + match (&plan.properties.partitioning, &expected_partitioning) { + ( + Partitioning::UnknownPartitioning(actual_count), + Partitioning::UnknownPartitioning(expected_count), + ) => assert_eq!(actual_count, expected_count, "projection {projection:?}"), + (actual, expected) => assert_eq!(actual, expected, "projection {projection:?}"), + } + + for ordering in claimed { + assert!( + claims_ordering(&plan, &ordering), + "projection {projection:?} must claim ordering {ordering:?}" + ); + } + for ordering in unclaimed { + assert!( + !claims_ordering(&plan, &ordering), + "projection {projection:?} must not claim ordering {ordering:?}" + ); + } + } + } + + #[test] + fn test_count_statistics_require_exact_source_rows() { + for (append_mode, exact, expected) in [ + (true, false, Precision::Absent), + (true, true, Precision::Exact(3)), + (false, true, Precision::Absent), + ] { + // 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() + .with_append_mode(append_mode) + .with_total_rows(3) + .with_total_rows_is_exact(exact), + metadata, + dynamic_filters: Arc::new(Mutex::new(Vec::new())), + }); + let plan = RegionScanExec::new(scanner, ScanRequest::default(), None).unwrap(); + let stats = plan.partition_statistics(None).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!( + plan.partition_statistics(Some(0)).unwrap().num_rows, + Precision::Absent, + ); + } + } + #[tokio::test] async fn test_reset_state_clears_dynamic_filters_for_repeatable_scanner() { let ctx = SessionContext::new(); diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 9c0e22b2a51..db1df3b339e 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -258,6 +258,295 @@ WHERE greptime_timestamp = ( ); } +#[apply(both_instances_cases)] +async fn test_metric_logical_scan_tag_order(instance: Arc) { + let frontend = instance.frontend(); + + execute_sql( + &frontend, + r#"CREATE TABLE f04_scan_phy ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX + ) ENGINE = metric WITH (physical_metric_table = 'true')"#, + ) + .await; + execute_sql( + &frontend, + r#"CREATE TABLE f04_scan_logical ( + host STRING PRIMARY KEY, + val DOUBLE, + ts TIMESTAMP TIME INDEX + ) ENGINE = metric WITH (on_physical_table = 'f04_scan_phy')"#, + ) + .await; + execute_sql( + &frontend, + "INSERT INTO f04_scan_logical (host, val, ts) VALUES \ + ('a', 1, 0), ('b', 2, 0), ('c', 3, 0), ('d', 4, 0), \ + ('e', 5, 0), ('f', 6, 0), ('g', 7, 0), ('h', 8, 0)", + ) + .await; + + // The ordered rows are compared exactly as printed. Sorting them before + // comparing (or comparing them unordered) would accept a scan that returns + // the physical table's tag order instead of the requested `host` order. + async fn ordered_rows(instance: &Arc, sql: &str, parallelism: usize) -> String { + execute_sql_with_query_parallelism(instance, sql, parallelism) + .await + .data + .pretty_print() + .await + } + + let all_hosts = "\ ++------+---------------------+-----+ +| host | ts | val | ++------+---------------------+-----+ +| a | 1970-01-01T00:00:00 | 1.0 | +| b | 1970-01-01T00:00:00 | 2.0 | +| c | 1970-01-01T00:00:00 | 3.0 | +| d | 1970-01-01T00:00:00 | 4.0 | +| e | 1970-01-01T00:00:00 | 5.0 | +| f | 1970-01-01T00:00:00 | 6.0 | +| g | 1970-01-01T00:00:00 | 7.0 | +| h | 1970-01-01T00:00:00 | 8.0 | ++------+---------------------+-----+"; + let first_three_hosts = "\ ++------+---------------------+-----+ +| host | ts | val | ++------+---------------------+-----+ +| a | 1970-01-01T00:00:00 | 1.0 | +| b | 1970-01-01T00:00:00 | 2.0 | +| c | 1970-01-01T00:00:00 | 3.0 | ++------+---------------------+-----+"; + + // `host` is NOT NULL, so `NULLS FIRST` and the SQL default `NULLS LAST` return the same + // rows. The `NULLS FIRST` queries spell out the ordering the scan used to claim, so the + // optimizer may drop the sort and expose a wrong tag order; the `NULLS LAST` queries + // keep the sort. Both are replayed in memory and after flushing the physical table. + let mut mismatches = Vec::new(); + for flushed in [false, true] { + if flushed { + execute_sql(&frontend, "ADMIN FLUSH_TABLE('f04_scan_phy')").await; + } + for parallelism in [1, 4] { + for (query, expected) in [ + ( + "SELECT host, ts, val FROM f04_scan_logical ORDER BY host NULLS FIRST", + all_hosts, + ), + ( + "SELECT host, ts, val FROM f04_scan_logical ORDER BY host NULLS FIRST LIMIT 3", + first_three_hosts, + ), + ( + "SELECT host, ts, val FROM f04_scan_logical ORDER BY host NULLS LAST", + all_hosts, + ), + ( + "SELECT host, ts, val FROM f04_scan_logical ORDER BY host NULLS LAST LIMIT 3", + first_three_hosts, + ), + ] { + let actual = ordered_rows(&frontend, query, parallelism).await; + if actual.trim() != expected.trim() { + mismatches.push(format!( + "query `{query}` with query_parallelism={parallelism} flushed={flushed} \ + returned:\n{actual}" + )); + } + } + } + } + assert!( + mismatches.is_empty(), + "logical metric table rows are not ordered by host, expected `a`..`h`:\n{}", + mismatches.join("\n---\n") + ); +} + +/// The uppercase tag `"A"` sorts before the physical `__table_id`/`__tsid` row key prefix +/// by name, but it is not a prefix of the row key. A scan that claims it as ordering or +/// partitioning lets the optimizer skip the sort or the repartition the query needs, so the +/// per-host count comes back split across scan partitions in physical tsid order. +#[apply(both_instances_cases)] +async fn test_metric_logical_scan_uppercase_tag_order(instance: Arc) { + let frontend = instance.frontend(); + + execute_sql( + &frontend, + r#"CREATE TABLE f04_upper_phy ( + "A" STRING, + marker STRING, + val DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY ("A", marker) + ) ENGINE = metric WITH (physical_metric_table = 'true')"#, + ) + .await; + execute_sql( + &frontend, + r#"CREATE TABLE f04_upper_logical ( + "A" STRING, + marker STRING, + val DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY ("A", marker) + ) ENGINE = metric WITH (on_physical_table = 'f04_upper_phy')"#, + ) + .await; + + // 8 hosts x 4 markers x 2 timestamps = 64 rows. Every host spans four series, so a count + // computed per scan partition would return several rows per host with partial counts. + let mut rows = Vec::with_capacity(64); + for host in 0..8 { + for marker in ["a", "b", "c", "d"] { + for ts in [1000, 2000] { + rows.push(format!("('h{host}', '{marker}', {host}, {ts})")); + } + } + } + execute_sql( + &frontend, + &format!( + "INSERT INTO f04_upper_logical (\"A\", marker, val, ts) VALUES {}", + rows.join(", ") + ), + ) + .await; + + let expected = "\ ++----+---+ +| A | n | ++----+---+ +| h0 | 8 | +| h1 | 8 | +| h2 | 8 | +| h3 | 8 | +| h4 | 8 | +| h5 | 8 | +| h6 | 8 | +| h7 | 8 | ++----+---+"; + + // The rows are compared exactly as printed: sorting them before comparing would accept + // a wrong count or a wrong row order. `NULLS FIRST` spells out the ordering the scan + // used to claim, so the optimizer may drop the sort; the default `NULLS LAST` keeps it + // and a failure there can only come from a wrong partitioning claim. Both orders are + // replayed in memory and after flushing the physical table. + let queries = [ + "SELECT \"A\", count(*) AS n FROM f04_upper_logical GROUP BY \"A\" ORDER BY \"A\"", + "SELECT \"A\", count(*) AS n FROM f04_upper_logical GROUP BY \"A\" ORDER BY \"A\" NULLS FIRST", + ]; + + let mut mismatches = Vec::new(); + for flushed in [false, true] { + if flushed { + execute_sql(&frontend, "ADMIN FLUSH_TABLE('f04_upper_phy')").await; + } + for parallelism in [1, 4] { + for query in queries { + let actual = execute_sql_with_query_parallelism(&frontend, query, parallelism) + .await + .data + .pretty_print() + .await; + if actual.trim() != expected.trim() { + mismatches.push(format!( + "query `{query}` with query_parallelism={parallelism} flushed={flushed} \ + returned:\n{actual}" + )); + } + } + } + } + assert!( + mismatches.is_empty(), + "logical metric table must return one row of eight per host in `A` order:\n{}", + mismatches.join("\n---\n") + ); +} + +/// The small shape of the same problem: four rows and two groups. Each host has two series, +/// so the count must not be computed per scan partition, and `ORDER BY "A"` pins the exact +/// row order at query_parallelism 4 as well as at 1. +#[apply(both_instances_cases)] +async fn test_metric_logical_scan_small_group_count(instance: Arc) { + let frontend = instance.frontend(); + + execute_sql( + &frontend, + r#"CREATE TABLE f04_upper_min_phy ( + "A" STRING, + marker STRING, + val DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY ("A", marker) + ) ENGINE = metric WITH (physical_metric_table = 'true')"#, + ) + .await; + execute_sql( + &frontend, + r#"CREATE TABLE f04_upper_min_logical ( + "A" STRING, + marker STRING, + val DOUBLE, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY ("A", marker) + ) ENGINE = metric WITH (on_physical_table = 'f04_upper_min_phy')"#, + ) + .await; + execute_sql( + &frontend, + "INSERT INTO f04_upper_min_logical (\"A\", marker, val, ts) VALUES \ + ('h0', 'a', 1, 1000), ('h0', 'b', 2, 1000), \ + ('h1', 'a', 3, 1000), ('h1', 'b', 4, 1000)", + ) + .await; + + let expected = "\ ++----+---+ +| A | n | ++----+---+ +| h0 | 2 | +| h1 | 2 | ++----+---+"; + // `NULLS FIRST` spells out the ordering the scan used to claim and is replayed next to + // the default `NULLS LAST` order, in memory and after flushing the physical table. + let queries = [ + "SELECT \"A\", count(*) AS n FROM f04_upper_min_logical GROUP BY \"A\" ORDER BY \"A\"", + "SELECT \"A\", count(*) AS n FROM f04_upper_min_logical GROUP BY \"A\" ORDER BY \"A\" NULLS FIRST", + ]; + + let mut mismatches = Vec::new(); + for flushed in [false, true] { + if flushed { + execute_sql(&frontend, "ADMIN FLUSH_TABLE('f04_upper_min_phy')").await; + } + for parallelism in [1, 4] { + for query in queries { + let actual = execute_sql_with_query_parallelism(&frontend, query, parallelism) + .await + .data + .pretty_print() + .await; + if actual.trim() != expected.trim() { + mismatches.push(format!( + "query `{query}` with query_parallelism={parallelism} flushed={flushed} \ + returned:\n{actual}" + )); + } + } + } + } + assert!( + mismatches.is_empty(), + "small logical metric grouping must return [h0, 2] and [h1, 2]:\n{}", + mismatches.join("\n---\n") + ); +} + #[apply(both_instances_cases)] async fn test_partitioned_merge_scan_left_join_repartitions(instance: Arc) { let frontend = instance.frontend(); diff --git a/tests-integration/tests/repartition.rs b/tests-integration/tests/repartition.rs index cf64f40aa00..fae78b94f85 100644 --- a/tests-integration/tests/repartition.rs +++ b/tests-integration/tests/repartition.rs @@ -24,7 +24,7 @@ use common_procedure::{ProcedureContext, ProcedureWithId, watcher}; use common_query::Output; use common_telemetry::info; use common_test_util::recordbatch::check_output_stream; -use common_test_util::temp_dir::create_temp_dir; +use common_test_util::temp_dir::{TempDir, create_temp_dir}; use common_wal::config::DatanodeWalConfig; use frontend::instance::Instance; use meta_srv::gc::{self, BatchGcProcedure, GcSchedulerOptions, GcTickerRef}; @@ -117,6 +117,189 @@ macro_rules! repartition_tests { }; } +/// COUNT must use visible rows rather than the full row counts of shared SSTs. +#[tokio::test(flavor = "multi_thread")] +async fn test_repartition_append_count_file() { + let (cluster, _home_guard) = append_count_cluster("repartition_append_count").await; + let instance = cluster.fe_instance(); + let table = "count_repartition"; + prepare_append_count_table(instance, table, "").await; + assert_append_count(instance, table, 100, 1).await; + flush_append_count_table(instance, table).await; + assert_append_count(instance, table, 100, 1).await; + + let sql = "ALTER TABLE count_repartition PARTITION ON COLUMNS (device_id) \ + (device_id < 50, device_id >= 50)"; + run_sql(instance, sql, QueryContext::arc()).await.unwrap(); + wait_for_append_count_regions(instance, table, 2).await; + assert_append_count(instance, table, 100, 0).await; + check_append_count_after_write(instance, table, 0).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_split_append_count_file() { + let (cluster, _home_guard) = append_count_cluster("split_append_count").await; + let instance = cluster.fe_instance(); + let table = "count_split"; + prepare_append_count_table( + instance, + table, + "PARTITION ON COLUMNS (device_id) (device_id < 50, device_id >= 50)", + ) + .await; + assert_append_count(instance, table, 100, 2).await; + flush_append_count_table(instance, table).await; + assert_append_count(instance, table, 100, 2).await; + + let sql = "ALTER TABLE count_split SPLIT PARTITION (device_id < 50) INTO \ + (device_id < 25, device_id >= 25 AND device_id < 50)"; + run_sql(instance, sql, QueryContext::arc()).await.unwrap(); + wait_for_append_count_regions(instance, table, 3).await; + assert_append_count(instance, table, 100, 1).await; + check_append_count_after_write(instance, table, 1).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_repartition_append_count_memtable_file() { + let (cluster, home_guard) = append_count_cluster("repartition_append_count_memtable").await; + assert!(home_guard.path().is_dir()); + let instance = cluster.fe_instance(); + let table = "count_repartition_memtable"; + prepare_append_count_table(instance, table, "").await; + assert_append_count(instance, table, 100, 1).await; + + // Entering staging must flush the populated memtable before changing partitions. + let sql = "ALTER TABLE count_repartition_memtable PARTITION ON COLUMNS (device_id) \ + (device_id < 50, device_id >= 50)"; + run_sql(instance, sql, QueryContext::arc()).await.unwrap(); + wait_for_append_count_regions(instance, table, 2).await; + assert_append_count(instance, table, 100, 0).await; + check_append_count_after_write(instance, table, 0).await; + assert!(home_guard.path().is_dir()); +} + +async fn append_count_cluster(name: &str) -> (GreptimeDbCluster, Arc) { + common_telemetry::init_default_ut_logging(); + let (store_config, _guard) = get_test_store_config(&StorageType::File); + let home_dir = Arc::new(create_temp_dir(name)); + let cluster = GreptimeDbClusterBuilder::new(name) + .await + .with_shared_home_dir(Arc::clone(&home_dir)) + .with_datanodes(3) + .with_store_config(store_config) + .with_datanode_wal_config(DatanodeWalConfig::Noop) + .with_metasrv_gc_config(GcSchedulerOptions { + enable: true, + ..Default::default() + }) + .with_datanode_gc_config(GcConfig { + enable: true, + ..Default::default() + }) + .build(true) + .await; + (cluster, home_dir) +} + +async fn prepare_append_count_table(instance: &Arc, table: &str, partitions: &str) { + let sql = format!( + "CREATE TABLE {table} (ts TIMESTAMP TIME INDEX, device_id INT) \ + {partitions} ENGINE=mito WITH(append_mode='true')" + ); + run_sql(instance, &sql, QueryContext::arc()).await.unwrap(); + let sql = format!( + "INSERT INTO {table} SELECT to_timestamp_millis(value), CAST(value AS INT) \ + FROM generate_series(0, 99)" + ); + run_sql(instance, &sql, QueryContext::arc()).await.unwrap(); +} + +async fn flush_append_count_table(instance: &Arc, table: &str) { + run_sql( + instance, + &format!("ADMIN flush_table('{table}')"), + QueryContext::arc(), + ) + .await + .unwrap(); +} + +async fn check_append_count_after_write( + instance: &Arc, + table: &str, + statistics_regions: usize, +) { + run_sql( + instance, + &format!("INSERT INTO {table} VALUES (to_timestamp_millis(100), 100)"), + QueryContext::arc(), + ) + .await + .unwrap(); + assert_append_count(instance, table, 101, statistics_regions).await; + flush_append_count_table(instance, table).await; + assert_append_count(instance, table, 101, statistics_regions).await; +} + +async fn wait_for_append_count_regions(instance: &Arc, table: &str, expected: usize) { + // Wait for frontend routing, not for COUNT to become correct. + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let plan = + append_count_query(instance, &format!("EXPLAIN ANALYZE SELECT * FROM {table}")) + .await; + if plan.matches("UnorderedScan: region=").count() == expected { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("new partition routes must become visible"); +} + +async fn append_count_query(instance: &Arc, sql: &str) -> String { + let output = run_sql(instance, sql, QueryContext::arc()).await.unwrap(); + let batches = match output.data { + common_query::OutputData::Stream(stream) => { + common_recordbatch::RecordBatches::try_collect(stream) + .await + .unwrap() + } + common_query::OutputData::RecordBatches(batches) => batches, + _ => panic!("expected query output: {sql}"), + }; + batches.pretty_print().unwrap() +} + +async fn assert_append_count( + instance: &Arc, + table: &str, + rows: usize, + statistics_regions: usize, +) { + let expected = format!("+-----+\n| n |\n+-----+\n| {rows:<3} |\n+-----+"); + let count = append_count_query(instance, &format!("SELECT count(*) AS n FROM {table}")).await; + assert_eq!(count, expected, "unbounded COUNT for {table}"); + let scanned_count = append_count_query( + instance, + &format!("SELECT count(*) AS n FROM (SELECT * FROM {table} LIMIT 1000)"), + ) + .await; + assert_eq!(scanned_count, expected, "scanned COUNT for {table}"); + // Only regions whose source row counts are exact may use statistics. + let plan = append_count_query( + instance, + &format!("EXPLAIN ANALYZE SELECT count(*) FROM {table}"), + ) + .await; + assert_eq!( + plan.matches("PlaceholderRowExec").count(), + statistics_regions, + "{plan}" + ); +} + #[tokio::test(flavor = "multi_thread")] async fn test_repartition_multi_hop_fulltext_index_gc_file() { if StorageType::File.test_on() { 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..801e218d0be --- /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(Int64(1))] REDACTED +|_|_|_CoalescePartitionsExec REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[], aggr=[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; diff --git a/tests/cases/standalone/common/aggregate/multi_regions.result b/tests/cases/standalone/common/aggregate/multi_regions.result index 8627b83c153..538252d28da 100644 --- a/tests/cases/standalone/common/aggregate/multi_regions.result +++ b/tests/cases/standalone/common/aggregate/multi_regions.result @@ -113,6 +113,54 @@ select sum(val) from t group by idc; |_|_| Total rows: 0_| +-+-+-+ +insert into t values + (1000, 1, '1000-x', 'a'), + (1000, 2, '1000-y', 'a'), + (1000, 3, '2000-x', 'a'), + (1000, 4, '2000-y', 'a'); + +Affected Rows: 4 + +-- group keys derived from the partition column span regions +select substr(host, 6, 1) as g, count(*), sum(val) from t group by g order by g; + ++---+----------+------------+ +| g | count(*) | sum(t.val) | ++---+----------+------------+ +| x | 2 | 4.0 | +| y | 2 | 6.0 | ++---+----------+------------+ + +-- grouping sets are computed on the frontend, only the PostgreSQL dialect parses them +-- SQLNESS PROTOCOL POSTGRES +select host, idc, sum(val) from t group by grouping sets ((host, idc), (host)) order by host, idc; + ++--------+-----+------------+ +| host | idc | sum(t.val) | ++--------+-----+------------+ +| 1000-x | a | 1.0 | +| 1000-x | | 1.0 | +| 1000-y | a | 2.0 | +| 1000-y | | 2.0 | +| 2000-x | a | 3.0 | +| 2000-x | | 3.0 | +| 2000-y | a | 4.0 | +| 2000-y | | 4.0 | ++--------+-----+------------+ + +-- SQLNESS PROTOCOL POSTGRES +select host, sum(val) from t group by rollup(host) order by host; + ++--------+------------+ +| host | sum(t.val) | ++--------+------------+ +| 1000-x | 1.0 | +| 1000-y | 2.0 | +| 2000-x | 3.0 | +| 2000-y | 4.0 | +| | 10.0 | ++--------+------------+ + drop table t; Affected Rows: 0 diff --git a/tests/cases/standalone/common/aggregate/multi_regions.sql b/tests/cases/standalone/common/aggregate/multi_regions.sql index 3396bdddf19..4f99e1ed465 100644 --- a/tests/cases/standalone/common/aggregate/multi_regions.sql +++ b/tests/cases/standalone/common/aggregate/multi_regions.sql @@ -45,4 +45,20 @@ select sum(val) from t; explain analyze select sum(val) from t group by idc; +insert into t values + (1000, 1, '1000-x', 'a'), + (1000, 2, '1000-y', 'a'), + (1000, 3, '2000-x', 'a'), + (1000, 4, '2000-y', 'a'); + +-- group keys derived from the partition column span regions +select substr(host, 6, 1) as g, count(*), sum(val) from t group by g order by g; + +-- grouping sets are computed on the frontend, only the PostgreSQL dialect parses them +-- SQLNESS PROTOCOL POSTGRES +select host, idc, sum(val) from t group by grouping sets ((host, idc), (host)) order by host, idc; + +-- SQLNESS PROTOCOL POSTGRES +select host, sum(val) from t group by rollup(host) order by host; + drop table t; diff --git a/tests/cases/standalone/common/aggregate/projected_pk_group_by.result b/tests/cases/standalone/common/aggregate/projected_pk_group_by.result new file mode 100644 index 00000000000..7ca3f1e5826 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/projected_pk_group_by.result @@ -0,0 +1,69 @@ +CREATE TABLE projected_pk_group_by ( + marker TINYINT UNSIGNED, + host STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (marker, host) +); + +Affected Rows: 0 + +INSERT INTO projected_pk_group_by VALUES + (0, 'a', 1000), + (0, 'a', 2000), + (0, 'b', 1000), + (0, 'b', 2000), + (0, 'c', 1000), + (1, NULL, 1000); + +Affected Rows: 6 + +SELECT marker, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker +ORDER BY marker; + ++--------+-----------+ +| marker | row_count | ++--------+-----------+ +| 0 | 5 | +| 1 | 1 | ++--------+-----------+ + +SELECT marker, host, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker, host +ORDER BY marker, host; + ++--------+------+-----------+ +| marker | host | row_count | ++--------+------+-----------+ +| 0 | a | 2 | +| 0 | b | 2 | +| 0 | c | 1 | +| 1 | | 1 | ++--------+------+-----------+ + +ADMIN FLUSH_TABLE('projected_pk_group_by'); + ++--------------------------------------------+ +| ADMIN FLUSH_TABLE('projected_pk_group_by') | ++--------------------------------------------+ +| 0 | ++--------------------------------------------+ + +SELECT marker, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker +ORDER BY marker; + ++--------+-----------+ +| marker | row_count | ++--------+-----------+ +| 0 | 5 | +| 1 | 1 | ++--------+-----------+ + +DROP TABLE projected_pk_group_by; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/common/aggregate/projected_pk_group_by.sql b/tests/cases/standalone/common/aggregate/projected_pk_group_by.sql new file mode 100644 index 00000000000..d525e16bc55 --- /dev/null +++ b/tests/cases/standalone/common/aggregate/projected_pk_group_by.sql @@ -0,0 +1,33 @@ +CREATE TABLE projected_pk_group_by ( + marker TINYINT UNSIGNED, + host STRING, + ts TIMESTAMP TIME INDEX, + PRIMARY KEY (marker, host) +); + +INSERT INTO projected_pk_group_by VALUES + (0, 'a', 1000), + (0, 'a', 2000), + (0, 'b', 1000), + (0, 'b', 2000), + (0, 'c', 1000), + (1, NULL, 1000); + +SELECT marker, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker +ORDER BY marker; + +SELECT marker, host, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker, host +ORDER BY marker, host; + +ADMIN FLUSH_TABLE('projected_pk_group_by'); + +SELECT marker, count(*) AS row_count +FROM projected_pk_group_by +GROUP BY marker +ORDER BY marker; + +DROP TABLE projected_pk_group_by;