mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-15 16:32:34 +00:00
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>
This commit is contained in:
discord9
+265
-7
@@ -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::<Vec<_>>();
|
||||
// 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::<Vec<_>>();
|
||||
let pk_sort_columns: Vec<PhysicalSortExpr> = pk_names
|
||||
let pk_sort_columns: Vec<PhysicalSortExpr> = 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 [
|
||||
|
||||
@@ -258,6 +258,295 @@ WHERE greptime_timestamp = (
|
||||
);
|
||||
}
|
||||
|
||||
#[apply(both_instances_cases)]
|
||||
async fn test_metric_logical_scan_tag_order(instance: Arc<dyn MockInstance>) {
|
||||
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<Instance>, 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<dyn MockInstance>) {
|
||||
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<dyn MockInstance>) {
|
||||
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<dyn MockInstance>) {
|
||||
let frontend = instance.frontend();
|
||||
|
||||
Reference in New Issue
Block a user