From 75bc4ce46d0e7aedca53dc2b0375ca0658456ff6 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 14 Oct 2026 03:35:58 +0000 Subject: [PATCH] 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> --- src/table/src/table/scan.rs | 272 ++++++++++++++++- tests-integration/src/tests/instance_test.rs | 289 +++++++++++++++++++ 2 files changed, 554 insertions(+), 7 deletions(-) diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index 3e0eeab84a9..1bfaea0d1df 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); @@ -694,9 +709,11 @@ 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::consts::ReservedColumnId; use store_api::storage::{RegionId, TimeSeriesDistribution}; use super::*; @@ -953,6 +970,247 @@ mod test { } } + /// 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 [ diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 1a9cc1790f7..22a0b993d17 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();