From ebfbd2ae5da1a2e02ce27db98ede723e5a4308d1 Mon Sep 17 00:00:00 2001 From: discord9 Date: Sat, 24 Oct 2026 18:58:51 +0800 Subject: [PATCH] fix(query): preserve legacy semantics at native admission Signed-off-by: discord9 --- src/promql/AGENTS.md | 23 +- src/query/src/dist_plan/commutativity.rs | 22 +- src/query/src/planner.rs | 121 +++++- src/query/src/promql/planner.rs | 19 - src/query/src/promql/planner/native.rs | 55 ++- src/query/src/promql/planner/test.rs | 197 ++++++++- src/sql/src/parsers/create_parser.rs | 61 +-- src/sql/src/statements/tql.rs | 2 +- tests-integration/src/tests/promql_test.rs | 392 +++++++++++++++++- .../standalone/common/view/tql_eval.result | 41 -- .../cases/standalone/common/view/tql_eval.sql | 18 - 11 files changed, 754 insertions(+), 197 deletions(-) delete mode 100644 tests/cases/standalone/common/view/tql_eval.result delete mode 100644 tests/cases/standalone/common/view/tql_eval.sql diff --git a/src/promql/AGENTS.md b/src/promql/AGENTS.md index 3962add5e1b..2381461d3c0 100644 --- a/src/promql/AGENTS.md +++ b/src/promql/AGENTS.md @@ -34,8 +34,21 @@ operators. Statement-to-plan translation lives in `src/query/src/promql/`. requires one selector, a direct scan with a bound metric name, and no residual label matchers or `__field__` matcher that narrows the scan, plus an eligible raw-series table/field set and supported expression/grid. Metric scans retain - the static `__table_id` predicate. Hybrid CTEs and CREATE VIEW TQL creation - currently fall back to the existing planner. + the static `__table_id` predicate. Hybrid CTEs keep the existing plan. The + operator also disables native planning for a programmatic TQL view so its + persisted Substrait plan remains serializable; `CREATE VIEW ... AS TQL EVAL` + syntax is a separate change. +- **Admission and SQL output:** the guards live in + `src/query/src/promql/planner/native.rs`. Non-`Float64` fields, alternative + (`or`) matchers and `rate`/`increase` on temporality-tagged tables keep the + existing plan; the latter preserves legacy per-series delta summation. + Allowing query fallback through the engine option, session variable or + `query_fallback` hint also disables native admission. Native SQL/TQL output + has column order `time index, fields, tags`, raw field names with nullable + `Float64` values, and nullable `Utf8` tags (including integer table tags). + Unlike a legacy raw selector, it does not preserve tag types; legacy function + expressions can also use different field names and column order. The native + time index is a non-nullable millisecond timestamp without a time zone. ## Change coupling @@ -45,9 +58,9 @@ operators. Statement-to-plan translation lives in `src/query/src/promql/`. (`src/query/src/dist_plan/commutativity.rs`). Process-local `FusedSeries` is the exception: it is not serialized to a datanode. Its distributed placement is restricted to a direct table scan, a bound metric name and trees that read - one series at a time. The single-region placement warning in - `commutativity.rs` remains important: do not assume a series stays in one - region. + one series at a time, and it is only correct because native admission refuses a + table whose partition columns are not all tag columns: that is what keeps a + series inside one region. Keep that admission and this placement together. - A new logical node also needs its physical planner reachable from `DfQueryPlanner::create_physical_plan`, assembled in `src/query/src/query_engine/state.rs`. diff --git a/src/query/src/dist_plan/commutativity.rs b/src/query/src/dist_plan/commutativity.rs index c11e78cf819..4221b009390 100644 --- a/src/query/src/dist_plan/commutativity.rs +++ b/src/query/src/dist_plan/commutativity.rs @@ -263,20 +263,14 @@ impl Categorizer { return Commutativity::NonCommutative; } - // TEMPORARY, and this is the one thing here that is not proved: the placement is - // only correct while each series reaches exactly one region. A region evaluates the - // whole tree over the series its scan returns, so a series whose samples are split - // across regions is evaluated per fragment, and a range function over a fragment is - // not the range function over the series. Nothing checks that a series stays in one - // region: the storage partition key is whatever `partition_column_names()` names and - // may be a column that varies within one series, and this categorizer runs before - // region selection, so it has no selected-region vector to consult either. A single - // region deployment satisfies it; a partitioned table over several regions does not, - // and would return a wrong answer rather than an error. - // - // What would make this unconditional is a partition key the label set determines, - // checked against the table's physical partition expressions. Until then the - // placement is a proof of concept over one region. + // A region evaluates the whole tree over the series its scan returns, so this is + // only correct while each series reaches exactly one region: a series split + // across regions would be ranged per fragment, and a range over fragments is not + // the range over the series. Native admission + // (`promql::planner::native::try_plan_native_fused`) only builds the node when + // every partition column of the table is one of the binding's tag columns, so + // the partition key is a function of the label set and a series cannot straddle + // regions. Commutativity::Commutative } name if name == SeriesNormalize::name() diff --git a/src/query/src/planner.rs b/src/query/src/planner.rs index de45e55e8d8..a70e0108817 100644 --- a/src/query/src/planner.rs +++ b/src/query/src/planner.rs @@ -50,6 +50,8 @@ use sql::statements::statement::Statement; use sql::statements::tql::Tql; use sqlparser::ast::{AccessExpr, Value, visit_expressions_mut}; +use crate::datafusion::QUERY_FALLBACK_HINT; +use crate::dist_plan::DistPlannerOptions; use crate::error::{ CteColumnSchemaMismatchSnafu, PlanSqlSnafu, QueryPlanSnafu, Result, SqlSnafu, UnimplementedSnafu, @@ -323,6 +325,20 @@ impl DfLogicalPlanner { native_route_allowed: bool, ) -> Result { let mut scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?; + // Whether the distributed planner may fall back to the fallback plan rewriter. The + // execution engine resolves the same three sources the same way: the engine option, which + // it copies into the session state it plans with, and then, per query, the session + // variable or the `query_fallback` hint. Each source only ever turns the fallback on, so + // the value they end up with is the disjunction of them. + let allow_query_fallback = scheduled_state + .config_options() + .extensions + .get::() + .is_some_and(|options| options.allow_query_fallback) + || query_ctx.configuration_parameter().allow_query_fallback() + || query_ctx + .extension(QUERY_FALLBACK_HINT) + .is_some_and(|fallback| fallback.to_lowercase().parse::().unwrap_or(false)); let promql_annotations = query_ctx.remote_query_id().map(promql_annotation_collector); if let Some(collector) = &promql_annotations { scheduled_state @@ -350,7 +366,13 @@ impl DfLogicalPlanner { stmt, &self.engine_state, promql_annotations, + // A query allowed to fall back keeps the existing plan. The native node's physical + // planner needs the bare region scan as its input; the fallback rewriter instead wraps + // every table scan in `MergeScan`, so the node above it would have no physical plan and + // the query would fail after the fallback was already taken. Declining here keeps the + // whole query on the legacy path. native_route_allowed + && !allow_query_fallback && self.engine_state.experimental_native_promql() && self.engine_state.in_process_region_transport() && crate::promql::planner::native::native_promql_requested(&query_ctx), @@ -778,7 +800,9 @@ mod tests { use table::test_util::EmptyTable; use super::*; + use crate::datafusion::QUERY_FALLBACK_HINT; use crate::parser::{PromQuery, QueryLanguageParser}; + use crate::promql::planner::NATIVE_PROMQL_EXTENSION_KEY; use crate::{QueryEngineFactory, QueryEngineRef}; async fn create_test_engine() -> QueryEngineRef { @@ -837,6 +861,14 @@ mod tests { } fn create_promql_test_engine() -> QueryEngineRef { + create_promql_test_engine_with(crate::options::QueryOptions::default(), false) + } + + /// The PromQL fixture, additionally stating the engine options and how its regions are reached. + fn create_promql_test_engine_with( + query_options: crate::options::QueryOptions, + in_process_region_transport: bool, + ) -> QueryEngineRef { let catalog_manager = MemoryCatalogManager::with_default_setup(); let physical_table_name = "phy"; let physical_table_id = 999u32; @@ -927,15 +959,19 @@ mod tests { }) .unwrap(); - QueryEngineFactory::new( + QueryEngineFactory::try_new_with_plugins_and_region_transport( catalog_manager, None, None, None, None, + None, false, - crate::options::QueryOptions::default(), + in_process_region_transport, + Default::default(), + query_options, ) + .unwrap() .query_engine() } @@ -968,6 +1004,87 @@ mod tests { engine.planner().plan(&stmt, query_ctx).await.unwrap() } + /// Plans `rate(some_metric[5m])` on the fixture above with the fallback sources the caller + /// states — the engine option, the session variable and the `query_fallback` hint — while the + /// native route is enabled and requested, so only the fallback guard decides. + async fn plan_native_promql_with_fallback( + engine_fallback: bool, + session_fallback: bool, + hint: Option<&str>, + ) -> String { + let engine = create_promql_test_engine_with( + crate::options::QueryOptions { + allow_query_fallback: engine_fallback, + experimental_native_promql: true, + ..Default::default() + }, + // The native route plans a region input the region server in this process accepts. + true, + ); + // The native route request and the fallback hint both reach the planner as query context + // extensions, the way a client sends them. + let mut builder = QueryContextBuilder::default() + .set_extension(NATIVE_PROMQL_EXTENSION_KEY.to_string(), "true".to_string()); + if let Some(hint) = hint { + builder = builder.set_extension(QUERY_FALLBACK_HINT.to_string(), hint.to_string()); + } + let query_ctx = builder.build(); + query_ctx + .configuration_parameter() + .set_allow_query_fallback(session_fallback); + let query_ctx = Arc::new(query_ctx); + let stmt = QueryLanguageParser::parse_promql( + &PromQuery { + query: "rate(some_metric[5m])".to_string(), + start: "0".to_string(), + end: "10".to_string(), + step: "5s".to_string(), + lookback: "300s".to_string(), + alias: None, + }, + &query_ctx, + ) + .unwrap(); + + engine + .planner() + .plan(&stmt, query_ctx) + .await + .unwrap() + .display_indent() + .to_string() + } + + /// The native route is admitted only when the query cannot fall back, and every source the + /// execution engine resolves the fallback from decides that at plan time too. + #[tokio::test] + async fn test_plan_pql_declines_native_route_when_query_may_fall_back() { + // (engine option, session variable, hint, native route admitted) + for (engine, session, hint, native) in [ + // Nothing asks for the fallback, so the route is planned as before. + (false, false, None, true), + // A hint that does not parse as `true` leaves the fallback off, as it does at execution. + (false, false, Some("false"), true), + (false, false, Some("yes"), true), + // Either per-query source turns the fallback on, so the route is declined. + (false, true, None, false), + (false, false, Some("true"), false), + (false, false, Some("TRUE"), false), + // Each source only turns the fallback on: neither a `false` hint nor a `false` session + // variable clears the engine option, and a `false` hint does not clear the variable. + (true, false, None, false), + (true, false, Some("false"), false), + (true, true, Some("false"), false), + ] { + let plan = plan_native_promql_with_fallback(engine, session, hint).await; + assert_eq!( + plan.contains("FusedSeries"), + native, + "engine={engine}, session={session}, hint={hint:?}: {plan}" + ); + } + } + /// Plans `sql` and runs the DataFusion analyzer, which is where /// `InsertAssignmentRule` sits. Planning alone stops short of it, so these /// assertions would not see the assignment rewrite at all. diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 05acc1a9de5..18bb5f2c316 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -5517,25 +5517,6 @@ impl PromPlanner { .map(|idx| schema.field(idx).data_type()) } - /// Whether every field column carries a type the region's raw series preflight serves. - /// - /// The preflight decides on `ConcreteDataType::is_numeric`, which excludes the decimals, so - /// the Arrow type is converted rather than asked: Arrow's own `is_numeric` includes - /// `Decimal128`/`Decimal256` (and `Float16`, which has no column type here), and a decimal - /// field would then pass here and fail once the logical plan is fixed. Anything else - a - /// string, a boolean, a timestamp - has no raw framing either. - fn field_columns_are_raw_series_servable( - schema: &DFSchemaRef, - field_columns: &[String], - ) -> bool { - field_columns.iter().all(|field| { - Self::field_column_type(schema, field).is_some_and(|data_type| { - datatypes::prelude::ConcreteDataType::from_arrow_type(data_type).is_numeric() - || data_type == &Self::native_histogram_arrow_type() - }) - }) - } - fn field_column_is_native_histogram(schema: &DFSchemaRef, field_column: &str) -> bool { Self::field_column_type(schema, field_column) .is_some_and(|data_type| data_type == &Self::native_histogram_arrow_type()) diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index d2d23c8a123..9cd23a87cc3 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -31,6 +31,7 @@ use std::sync::Arc; use std::time::{Duration, UNIX_EPOCH}; use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE, MITO2_ENGINE}; +use common_query::prelude::OTLP_AGGREGATION_TEMPORALITY_LABEL; use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, TimeUnit as ArrowTimeUnit}; use datafusion::common::{DFSchema, DFSchemaRef}; use datafusion::error::DataFusionError; @@ -102,8 +103,9 @@ impl PromPlanner { /// caller plans it with the existing planner as if this method did not exist. The two ways that /// happens are a shape the compiler refuses (`DataFusionError::NotImplemented`, which names /// what it found) and a query this route cannot bind faithfully — more than one selector, a - /// label matcher the bare scan below the node would not apply, an `__field__` matcher that - /// narrows the field set, a missing table, a native histogram field, or an unusable grid. + /// label matcher the bare scan below the node would not apply, alternative (`or`) matchers, an + /// `__field__` matcher that narrows the field set, a missing table, a field that is not + /// `Float64`, a `rate`/`increase` over a delta-temporality table, or an unusable grid. pub(super) async fn try_plan_native_fused( &mut self, prom_expr: &PromExpr, @@ -125,6 +127,11 @@ impl PromPlanner { return Ok(None); }; + // These matchers are alternatives the one scan below the node cannot represent, so the + // query keeps the plan the existing path builds for it. + if !selector.matchers.or_matchers.is_empty() { + return Ok(None); + } // Resolve the selector the way the existing planner does, and take what is left as the // rows it selects. The scan below the node is the selector's own scan, unfiltered, so a // matcher that survives preprocessing selects fewer rows here than the scan reads. @@ -155,6 +162,18 @@ impl PromPlanner { if fields.is_empty() { return Ok(None); } + // On a delta-temporality table a series' samples are increments, which the existing + // planner sums over the window for `increase` and divides by the range for `rate` + // ([`PromPlanner::select_delta_range_math`]), per series because the tag can differ between + // them. A compiled tree computes both for a cumulative series, so any `rate`/`increase` + // over such a table keeps the existing plan; a bare selector still folds. + if tags + .iter() + .any(|tag| tag == OTLP_AGGREGATION_TEMPORALITY_LABEL) + && calls_rate_or_increase(prom_expr) + { + return Ok(None); + } // The selector's own table scan, built by the same method the existing path uses: the // physical planner turns it into the region scan the node reads, and building a scan here @@ -258,17 +277,13 @@ impl PromPlanner { }; scan.filters.push(filter); } - // A window over a native histogram yields a histogram, which the output schema below - // cannot declare; the existing path evaluates those with its own kernels. - if Self::field_columns_contain_native_histogram(input.schema(), &fields) { - return Ok(None); - } - // The region's raw series preflight serves a field only when it is numeric or a native - // histogram, and the physical layer turns that refusal into a planning error because the - // logical plan is already fixed by then. A table whose selected fields are, say, a string - // and a boolean therefore has to be declined here, while the existing plan is still the - // fallback - `stringsfield` over such a table answered normally before this route existed. - if !Self::field_columns_are_raw_series_servable(input.schema(), &fields) { + // The region's columnar series-index samples, one of the readers this scan may be served + // by, store `Float64` fields only, and the logical plan cannot tell which reader arrives. + // So every field has to be `Float64`: an integer, a `Float32`, a decimal, a string or a + // native histogram field keeps the existing plan, which the physical layer still serves. + if !fields.iter().all(|field| { + Self::field_column_type(input.schema(), field) == Some(&ArrowDataType::Float64) + }) { return Ok(None); } let Some(stmt) = self.native_statement(prom_expr) else { @@ -484,6 +499,20 @@ fn collect_selectors<'a>(expr: &'a PromExpr, out: &mut Vec<&'a VectorSelector>) } } +/// Whether `expr` calls `rate`/`increase`, the functions the existing planner rewrites per series +/// for a delta-temporality table. Only nested calls and parentheses can hold one in an expression +/// this route accepts; every other shape names no single selector or reads across series. +fn calls_rate_or_increase(expr: &PromExpr) -> bool { + match expr { + PromExpr::Call(call) => { + matches!(call.func.name, "rate" | "increase") + || call.args.args.iter().any(|arg| calls_rate_or_increase(arg)) + } + PromExpr::Paren(ParenExpr { expr }) => calls_rate_or_increase(expr), + _ => false, + } +} + #[cfg(test)] mod tests { use std::sync::Arc; diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 6a77ce48cc8..2e22ee59003 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -972,6 +972,73 @@ async fn native_engine_provider(partition_key_indices: Vec) -> DfTableSou .await } +/// A delta-temporality table on the engine the native route accepts. +/// +/// The OTLP receiver writes the aggregation temporality as a tag of every series, so a table with +/// that tag can hold delta and cumulative series side by side: the existing planner sums the +/// increments of a delta series instead of rating its counter. The value field is the `Float64` +/// this route folds. +async fn native_engine_temporality_marker_provider() -> DfTableSourceProvider { + let catalog_list = MemoryCatalogManager::with_default_setup(); + let schema = Arc::new(Schema::new(vec![ + ColumnSchema::new( + "tag_0".to_string(), + ConcreteDataType::string_datatype(), + false, + ), + ColumnSchema::new( + OTLP_AGGREGATION_TEMPORALITY_LABEL.to_string(), + ConcreteDataType::string_datatype(), + true, + ), + ColumnSchema::new( + greptime_timestamp().to_string(), + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ) + .with_time_index(true), + ColumnSchema::new( + greptime_value().to_string(), + ConcreteDataType::float64_datatype(), + true, + ), + ])); + let table_meta = TableMetaBuilder::empty() + .schema(schema) + .engine(NATIVE_ENGINE.to_string()) + .primary_key_indices(vec![0, 1]) + .value_indices(vec![3]) + .next_column_id(1024) + .build() + .unwrap(); + let table_info = TableInfoBuilder::default() + .name("some_metric") + .meta(table_meta) + .build() + .unwrap(); + let table = EmptyTable::from_table_info(&table_info); + + assert!( + catalog_list + .register_table_sync(RegisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: "some_metric".to_string(), + table_id: 1024, + table, + }) + .is_ok() + ); + + DfTableSourceProvider::new( + catalog_list, + false, + QueryContext::arc(), + DummyDecoder::arc(), + false, + ) +} + #[allow(clippy::too_many_arguments)] async fn build_test_table_provider_with_options( table_name_tuples: &[(String, String)], @@ -2132,19 +2199,23 @@ async fn native_fused_engine_without_raw_series_falls_back() { ); } -/// A value field the region's raw scan cannot serve keeps the existing plan. +/// A value field the raw-series readers cannot both serve keeps the existing plan. /// -/// The preflight serves numeric columns and native histograms only, and the physical layer turns -/// its refusal into a planning error once the logical plan is fixed. A bare selector over a string -/// field answered normally before this route existed, so it must keep doing that. +/// The region's preflight serves numeric columns and native histograms, while the columnar +/// series-index samples the scan may be read through store `Float64` fields only, and the logical +/// plan cannot tell which reader the region prepares. So only `Float64` is admitted; a bare +/// selector over any other field type answered normally before this route existed. #[tokio::test] -async fn native_fused_non_numeric_field_falls_back() { +async fn native_fused_unsupported_field_type_falls_back() { for field_type in [ ConcreteDataType::string_datatype(), ConcreteDataType::boolean_datatype(), // Arrow's `is_numeric` includes the decimals while the region's preflight does not, so // asking Arrow would let a decimal field through to a planning error. ConcreteDataType::decimal128_datatype(10, 2), + // Numeric, but not the `Float64` the columnar samples store. + ConcreteDataType::int64_datatype(), + ConcreteDataType::float32_datatype(), ] { let native = build_native_fused_plan_with_provider( "some_metric", @@ -2170,6 +2241,122 @@ async fn native_fused_non_numeric_field_falls_back() { } } +/// The `Float64` field every raw reader serves controls the field-type decline above. +#[tokio::test] +async fn native_fused_float64_field_is_accepted() { + let plan = build_native_fused_plan_with_provider( + "some_metric", + true, + build_test_table_provider_with_field_type(ConcreteDataType::float64_datatype()).await, + ) + .await; + + assert!( + find_fused_series(&plan).is_some(), + "a `Float64` field must still fold: {}", + plan.display_indent() + ); +} + +/// Alternative (`or`) matchers keep the existing plan. +/// +/// `a{x="1" or x="2"}` names alternative matchers, which the one scan below the node cannot +/// represent. These queries are only asserted to fall back: what the existing plan answers for +/// them is its own question, not one this route claims to have fixed. +#[tokio::test] +async fn native_fused_or_matchers_fall_back() { + for query in [ + r#"some_metric{tag_0="a" or tag_0="b"}"#, + r#"rate(some_metric{tag_0="a" or tag_0="b"}[5m])"#, + ] { + let native = build_native_fused_plan(query, true).await; + let without_flag = build_native_fused_plan(query, false).await; + assert!( + find_fused_series(&native).is_none(), + "`{query}` names alternative matchers: {}", + native.display_indent() + ); + assert_eq!( + native.display_indent().to_string(), + without_flag.display_indent().to_string(), + "`{query}` must keep the existing plan" + ); + } +} + +/// A `rate`/`increase` over a delta-temporality table keeps the existing plan, which sums each +/// series' increments. +/// +/// The samples of a series tagged `delta` are increments, which the existing planner sums over the +/// window for `increase` and divides by the range for `rate`, while a cumulative series is rated +/// from its counter; it picks that per series. A compiled tree rates every series the same way, so +/// these queries fall back. The same queries over a table without the marker still fold, which is +/// what makes the decline the marker's doing rather than the shape's. +#[tokio::test] +async fn native_fused_delta_temporality_rates_fall_back() { + for query in [ + "rate(some_metric[5m])", + "increase(some_metric[5m])", + // Below another call and inside parentheses, which are the same decline one level down. + "abs(rate(some_metric[5m]))", + "abs(increase(some_metric[5m]))", + "(rate(some_metric[5m]))", + ] { + let marked = build_native_fused_plan_with_provider( + query, + true, + native_engine_temporality_marker_provider().await, + ) + .await; + let marked_without_flag = build_native_fused_plan_with_provider( + query, + false, + native_engine_temporality_marker_provider().await, + ) + .await; + assert!( + find_fused_series(&marked).is_none(), + "`{query}` needs the per-series delta math: {}", + marked.display_indent() + ); + assert_eq!( + marked.display_indent().to_string(), + marked_without_flag.display_indent().to_string(), + "`{query}` must keep the existing plan, not the counter form for every series" + ); + + let unmarked = build_native_fused_plan_with_provider( + query, + true, + native_engine_provider(vec![]).await, + ) + .await; + assert!( + find_fused_series(&unmarked).is_some(), + "`{query}` over a table without the marker must still fold: {}", + unmarked.display_indent() + ); + } +} + +/// The marker alone does not decline a query: only `rate`/`increase` are refused. +#[tokio::test] +async fn native_fused_delta_temporality_non_counter_math_still_folds() { + for query in ["some_metric", "abs(some_metric)"] { + let plan = build_native_fused_plan_with_provider( + query, + true, + native_engine_temporality_marker_provider().await, + ) + .await; + assert!( + find_fused_series(&plan).is_some(), + "`{query}` needs no per-series delta math and must still fold: {}", + plan.display_indent() + ); + } +} + /// A whole-vector operation is not a per-series tree, so it stays on the existing plan. #[tokio::test] async fn native_fused_whole_vector_operation_falls_back() { diff --git a/src/sql/src/parsers/create_parser.rs b/src/sql/src/parsers/create_parser.rs index de9c5657358..20eba9f2087 100644 --- a/src/sql/src/parsers/create_parser.rs +++ b/src/sql/src/parsers/create_parser.rs @@ -55,7 +55,6 @@ use crate::statements::create::{ CreateTableLike, CreateView, Partitions, SqlOrTql, TableConstraint, VECTOR_OPT_DIM, }; use crate::statements::statement::Statement; -use crate::statements::tql::Tql; use crate::statements::transform::type_alias::get_data_type_by_alias_name; use crate::statements::{OptionMap, sql_data_type_to_concrete_data_type}; use crate::util::{OptionValue, location_to_index, parse_option_string}; @@ -149,18 +148,7 @@ impl<'a> ParserContext<'a> { .expect_keyword(Keyword::AS) .context(SyntaxSnafu)?; - let query = match self.parser.peek_token().token { - Token::Word(word) - if word.quote_style.is_none() - && word.value.eq_ignore_ascii_case(tql_parser::TQL) => - { - match self.parse_tql(false)? { - Statement::Tql(tql @ Tql::Eval(_)) => Statement::Tql(tql), - _ => return self.expected("TQL EVAL", self.parser.peek_token()), - } - } - _ => self.parse_query()?, - }; + let query = self.parse_query()?; Ok(Statement::CreateView(CreateView { name: view_name, @@ -3327,53 +3315,6 @@ CREATE TABLE spans ( } } - #[test] - fn test_parse_create_view_tql_eval() { - for (query, expected_lookback) in [ - ("TQL EVAL (0, 15, '5s') metric", None), - ("TQL EVAL (0, 15, '5s', '5m') metric", Some("5m")), - ("TQL EVAL (0, 15, '5s', '300') metric", Some("300")), - ] { - let sql = format!("CREATE VIEW test AS {query}"); - let result = ParserContext::create_with_dialect( - &sql, - &GreptimeDbDialect {}, - ParseOptions::default(), - ) - .unwrap(); - let Statement::CreateView(view) = &result[0] else { - panic!("expected CREATE VIEW, got {:?}", result[0]); - }; - let Statement::Tql(Tql::Eval(eval)) = view.query.as_ref() else { - panic!("expected TQL EVAL, got {}", view.query); - }; - assert_eq!(eval.lookback.as_deref(), expected_lookback); - - let formatted = result[0].to_string(); - let reparsed = ParserContext::create_with_dialect( - &formatted, - &GreptimeDbDialect {}, - ParseOptions::default(), - ) - .unwrap(); - assert_eq!(reparsed[0], result[0], "{formatted}"); - } - - for sql in [ - "CREATE VIEW test AS TQL EXPLAIN (0, 15, '5s') metric", - "CREATE VIEW test AS TQL ANALYZE (0, 15, '5s') metric", - ] { - assert!( - ParserContext::create_with_dialect( - sql, - &GreptimeDbDialect {}, - ParseOptions::default(), - ) - .is_err() - ); - } - } - #[test] fn test_parse_create_view_with_columns() { let sql = "CREATE VIEW test () AS SELECT * FROM NUMBERS"; diff --git a/src/sql/src/statements/tql.rs b/src/sql/src/statements/tql.rs index 948c31ddd8d..835a957eada 100644 --- a/src/sql/src/statements/tql.rs +++ b/src/sql/src/statements/tql.rs @@ -46,7 +46,7 @@ fn format_tql( ) -> std::fmt::Result { write!(f, "({start}, {end}, '{step}'")?; if let Some(lookback) = lookback { - write!(f, ", '{lookback}'")?; + write!(f, ", {lookback}")?; } if let Some(alias) = alias { write!(f, ") {query} AS {alias}") diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index dbce5829110..7f6e900bc1d 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -20,6 +20,7 @@ use api::v1::{ ColumnDataType, ColumnSchema, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType, }; use common_base::Plugins; +use common_query::prelude::{GREPTIME_TEMPORALITY_DELTA, OTLP_AGGREGATION_TEMPORALITY_LABEL}; use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS; use common_query::{Output, OutputData}; use common_recordbatch::util::collect_batches; @@ -30,7 +31,7 @@ use datatypes::arrow::array::{ use datatypes::arrow::compute::cast; use datatypes::arrow::datatypes::DataType; use frontend::instance::Instance; -use query::datafusion::QUERY_PARALLELISM_HINT; +use query::datafusion::{QUERY_FALLBACK_HINT, QUERY_PARALLELISM_HINT}; use query::options::QueryOptions; use query::parser::{PromQuery, QueryLanguageParser, QueryStatement}; use query::physical_wrapper::{PhysicalPlanWrapper, PhysicalPlanWrapperRef}; @@ -39,6 +40,10 @@ use rstest::rstest; use rstest_reuse::apply; use servers::query_handler::sql::SqlQueryHandler; use session::context::QueryContext; +use sql::ast::{Ident, ObjectName, ObjectNamePart}; +use sql::statements::create::CreateView; +use sql::statements::statement::Statement as SqlStatement; +use sql::statements::tql::{Tql, TqlEval}; use super::test_util::{ both_instances_cases, check_unordered_output_stream, distributed, standalone, @@ -1774,6 +1779,11 @@ impl NativePlanObserver { .take() .expect("native fused plan was captured") } + + /// The fused node the last wrapped plan contained, if any; `take` requires one instead. + fn fused(&self) -> Option> { + self.0.lock().unwrap().clone() + } } impl PhysicalPlanWrapper for NativePlanObserver { @@ -1867,6 +1877,90 @@ async fn native_result_rows(output: Output) -> Vec<(i64, String, f64)> { rows } +/// [`native_result_rows`] plus each row's OTLP temporality marker, sorted by label then time. +async fn native_marker_rows(output: Output) -> Vec<(i64, String, String, f64)> { + let batches = match output.data { + OutputData::Stream(stream) => collect_batches(stream).await.unwrap(), + OutputData::RecordBatches(batches) => batches, + _ => unreachable!("TQL EVAL returns record batches"), + }; + let mut rows = Vec::new(); + for batch in &batches { + let timestamps = batch + .column_by_name("ts") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let hosts = cast( + batch.column_by_name("host").unwrap().as_ref(), + &DataType::Utf8, + ) + .unwrap(); + let hosts = hosts.as_any().downcast_ref::().unwrap(); + let markers = cast( + batch + .column_by_name(OTLP_AGGREGATION_TEMPORALITY_LABEL) + .unwrap() + .as_ref(), + &DataType::Utf8, + ) + .unwrap(); + let markers = markers.as_any().downcast_ref::().unwrap(); + let values = batch + .columns() + .iter() + .filter(|column| column.data_type() == &DataType::Float64) + .collect::>(); + assert_eq!(values.len(), 1, "expected one Float64 value column"); + let values = values[0].as_any().downcast_ref::().unwrap(); + assert_eq!( + markers.null_count(), + 0, + "every row must keep its temporality label" + ); + assert_eq!(values.null_count(), 0); + for row in 0..batch.num_rows() { + rows.push(( + timestamps.value(row), + hosts.value(row).to_string(), + markers.value(row).to_string(), + values.value(row), + )); + } + } + rows.sort_by(|left, right| (left.1.as_str(), left.0).cmp(&(right.1.as_str(), right.0))); + rows +} + +/// Compares the whole answer against `expected`, in the same label order, key by key and value by +/// value; a duplicated or missing series therefore fails on the keys rather than on the count. +fn assert_delta_rows( + rows: &[(i64, String, String, f64)], + expected: &[(i64, &str, &str, f64)], + context: &str, +) { + let actual = rows + .iter() + .map(|(ts, host, marker, value)| (*ts, host.as_str(), marker.as_str(), *value)) + .collect::>(); + assert_eq!(actual.len(), expected.len(), "{context}: rows {actual:?}"); + for (actual, expected) in actual.iter().zip(expected) { + assert_eq!( + (actual.0, actual.1, actual.2), + (expected.0, expected.1, expected.2), + "{context}: row {actual:?}" + ); + assert!( + (actual.3 - expected.3).abs() <= 1e-9 * expected.3.abs().max(1.0), + "{context}: series {:?} must report {}, found {}", + actual.1, + expected.3, + actual.3 + ); + } +} + fn assert_four_native_partitions(plan: &Arc) { use datafusion::physical_plan::metrics::MetricValue; @@ -1893,20 +1987,35 @@ fn assert_four_native_partitions(plan: &Arc ( crate::standalone::GreptimeDbStandalone, Arc, +) { + native_test_instance_with_options( + builder, + QueryOptions { + experimental_native_promql: true, + ..Default::default() + }, + ) + .await +} + +/// [`native_test_instance`] with the engine-wide query options spelled out by the caller. +async fn native_test_instance_with_options( + builder: GreptimeDbStandaloneBuilder, + options: QueryOptions, +) -> ( + crate::standalone::GreptimeDbStandalone, + Arc, ) { let observer = Arc::new(NativePlanObserver::default()); let plugins = Plugins::new(); let wrapper: PhysicalPlanWrapperRef = observer.clone(); plugins.insert(wrapper); - let options = QueryOptions { - experimental_native_promql: true, - ..Default::default() - }; let standalone = builder .with_plugin(plugins) .with_query_options(options) @@ -2024,12 +2133,30 @@ async fn native_promql_view_fallback_preserves_native_hint() { execute_all(&instance, "CREATE TABLE native_view_source (ts TIMESTAMP TIME INDEX, host STRING, val DOUBLE, PRIMARY KEY(host)); INSERT INTO native_view_source VALUES (0, 'a', 2)", QueryContext::arc()).await; let ctx = native_query_context(true, 4); let expected_extensions = ctx.extensions(); - execute_all( - &instance, - "CREATE VIEW native_view AS TQL EVAL (0, 0, '1s', '5m') native_view_source", - ctx.clone(), - ) - .await; + // The AST the parser would build for `CREATE VIEW native_view AS TQL EVAL (0, 0, '1s', '5m') + // native_view_source`; the statement's spelling is covered by the separate view case. + let create_view = SqlStatement::CreateView(CreateView { + name: ObjectName(vec![ObjectNamePart::Identifier(Ident::new("native_view"))]), + columns: vec![], + query: Box::new(SqlStatement::Tql(Tql::Eval(TqlEval { + start: "0".to_string(), + end: "0".to_string(), + step: "1s".to_string(), + lookback: Some("5m".to_string()), + query: "native_view_source".to_string(), + alias: None, + }))), + or_replace: false, + if_not_exists: false, + }); + instance + .statement_executor() + .execute_stmt(QueryStatement::Sql(create_view), ctx.clone()) + .await + .unwrap(); + // The view's stored plan must not depend on the caller's native hint, so reading it runs the + // plan the view was created with rather than one that needs the hint to be readable. + observer.clear(); let view = instance .do_query("SELECT * FROM native_view", ctx.clone()) .await @@ -2038,14 +2165,11 @@ async fn native_promql_view_fallback_preserves_native_hint() { let view_rows = native_result_rows(view).await; assert_eq!(ctx.extensions(), expected_extensions); assert_eq!(view_rows, vec![(0, "a".to_string(), 2.0)]); - let create_view = instance - .do_query("SHOW CREATE VIEW native_view", ctx.clone()) - .await - .remove(0) - .unwrap(); - let create_view_text = native_output_text(create_view).await; - assert!(create_view_text.contains("TQL EVAL")); - assert!(create_view_text.contains("5m")); + assert!( + observer.fused().is_none(), + "a view plan must be readable without the native hint" + ); + // The hint is still the caller's: the same context keeps folding a direct query. observer.clear(); let direct = instance .do_query("TQL EVAL (0, 0, '1s') native_view_source", ctx) @@ -2102,3 +2226,233 @@ async fn native_promql_metric_logical_tables_are_isolated() { assert_eq!(native_rows, legacy_rows); } } + +/// OTLP delta temporality on the public TQL route: a `delta`-marked series answers the raw sample +/// sum (10 + 10 = 20) while a cumulative series with the same samples answers 0. The fused node +/// computes one cumulative window for every series, so it must decline a marker-bearing table +/// instead of reporting 0 for the delta series. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_delta_temporality_rate_and_increase_match_legacy() { + common_telemetry::init_default_ut_logging(); + const TABLE: &str = "native_delta_temporality"; + const CONTROL: &str = "native_delta_control"; + // Samples at 60s and 120s, evaluated at 120s over a 2m window: the `delta` series' samples are + // raw deltas, so its increase is their sum 10 + 10; the `cumulative` one did not rise, so 0. + const DELTA_INCREASE: f64 = 20.0; + const CUMULATIVE_INCREASE: f64 = 0.0; + const DELTA_RATE: f64 = 20.0 / 120.0; + const CUMULATIVE_RATE: f64 = 0.0; + + let (standalone, observer) = native_test_instance(GreptimeDbStandaloneBuilder::new( + "native_promql_delta_temporality", + )) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + &format!( + r#"CREATE TABLE {TABLE} ( + host STRING, + ts TIMESTAMP(3) TIME INDEX, + val DOUBLE, + otlp_aggregation_temporality STRING, + PRIMARY KEY (host, otlp_aggregation_temporality) + ) ENGINE=mito; + INSERT INTO {TABLE} (host, ts, val, otlp_aggregation_temporality) VALUES + ('delta', 60000, 10, '{GREPTIME_TEMPORALITY_DELTA}'), + ('delta', 120000, 10, '{GREPTIME_TEMPORALITY_DELTA}'), + ('cumulative', 60000, 10, 'cumulative'), + ('cumulative', 120000, 10, 'cumulative'); + CREATE TABLE {CONTROL} (host STRING, ts TIMESTAMP(3) TIME INDEX, val DOUBLE, PRIMARY KEY (host)) ENGINE=mito; + INSERT INTO {CONTROL} (host, ts, val) VALUES ('a', 60000, 10), ('a', 120000, 30);"# + ), + QueryContext::arc(), + ) + .await; + + for (query, delta, cumulative) in [ + ( + format!("increase({TABLE}[2m])"), + DELTA_INCREASE, + CUMULATIVE_INCREASE, + ), + (format!("rate({TABLE}[2m])"), DELTA_RATE, CUMULATIVE_RATE), + ( + format!("abs(increase({TABLE}[2m]))"), + DELTA_INCREASE, + CUMULATIVE_INCREASE, + ), + ( + format!("abs(rate({TABLE}[2m]))"), + DELTA_RATE, + CUMULATIVE_RATE, + ), + ] { + let sql = format!("TQL EVAL (120, 120, '1m') {query}"); + // Both series are in the window, in label order, with their own marker and value. + let expected = [ + (120_000, "cumulative", "cumulative", cumulative), + (120_000, "delta", GREPTIME_TEMPORALITY_DELTA, delta), + ]; + // Values before routes: a wrong number is the regression, a declined plan is not. + let legacy_rows = native_marker_rows(native_tql_output(&instance, &sql, false).await).await; + assert_delta_rows(&legacy_rows, &expected, &format!("native-off `{query}`")); + observer.clear(); + let native_rows = native_marker_rows(native_tql_output(&instance, &sql, true).await).await; + assert_delta_rows(&native_rows, &expected, &format!("native-on `{query}`")); + assert_eq!( + native_rows, legacy_rows, + "`{query}`: same rows on both routes" + ); + assert!( + observer.fused().is_none(), + "`{query}`: the native route must decline a table that carries the temporality marker" + ); + } + + // The control: a marker-free table (samples 10 and 30 here, unlike the table above) still fuses + // and agrees with the existing plan, so the decline is about the marker, not the table shape. + let control_sql = format!("TQL EVAL (120, 120, '1m') sum_over_time({CONTROL}[2m])"); + observer.clear(); + let control_native = + native_result_rows(native_tql_output(&instance, &control_sql, true).await).await; + assert_eq!( + control_native, + vec![(120_000, "a".to_string(), 40.0)], + "the fused control reports the sampled sum" + ); + assert_eq!(observer.take().name(), "FusedSeriesExec"); + let control_legacy = + native_result_rows(native_tql_output(&instance, &control_sql, false).await).await; + assert_eq!( + control_native, control_legacy, + "the control agrees with the existing plan" + ); +} + +/// Enabling query fallback — by the engine option or by the per-request `query_fallback` hint — must +/// keep an otherwise fusible query on the existing plan without changing its answer. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_fallback_options_keep_expected_answer() { + common_telemetry::init_default_ut_logging(); + const TABLE: &str = "native_fallback_control"; + // One eligible selector, so only the fallback settings decide the route. `sum_over_time` is + // exact over the two samples in the window: 10 + 30 for `a`, 10 + 70 for `b`. + let query = format!("sum_over_time({TABLE}[2m])"); + let expected = vec![ + (120_000, "a".to_string(), 40.0), + (120_000, "b".to_string(), 80.0), + ]; + for (tag, allow_query_fallback) in [("engine_off", false), ("engine_on", true)] { + let (standalone, observer) = native_test_instance_with_options( + GreptimeDbStandaloneBuilder::new(&format!("native_promql_fallback_{tag}")), + QueryOptions { + experimental_native_promql: true, + allow_query_fallback, + ..Default::default() + }, + ) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + &format!( + "CREATE TABLE {TABLE} (host STRING, ts TIMESTAMP(3) TIME INDEX, val DOUBLE, PRIMARY KEY (host)) ENGINE=mito; \ + INSERT INTO {TABLE} (host, ts, val) VALUES ('a', 60000, 10), ('a', 120000, 30), \ + ('b', 60000, 10), ('b', 120000, 70);" + ), + QueryContext::arc(), + ) + .await; + let mut hinted = native_query_context(true, 4).as_ref().clone(); + hinted.set_extension(QUERY_FALLBACK_HINT, "true"); + for (route, ctx) in [ + ("plain", native_query_context(true, 4)), + ("fallback-hint", Arc::new(hinted)), + ] { + let sql = format!("TQL EVAL (120, 120, '1m') {query}"); + observer.clear(); + let rows = + native_result_rows(instance.do_query(&sql, ctx).await.remove(0).unwrap()).await; + assert_eq!( + rows, expected, + "{tag} {route}: the answer must not depend on the fallback settings" + ); + if allow_query_fallback || route == "fallback-hint" { + // A fallback rewrites every table scan into `MergeScan`, which the fused node + // cannot read, so a query allowed to fall back keeps the existing plan. + assert!( + observer.fused().is_none(), + "{tag} {route}: enabling the fallback must keep the existing plan" + ); + } else { + assert_eq!( + observer.take().name(), + "FusedSeriesExec", + "{tag} {route}: with no fallback enabled this query must be fused" + ); + } + } + } +} + +/// A store configured for the columnar series format must not change the answer of a query over a +/// field that format cannot serve: the columnar samples carry `Float64` series only, so an integer +/// field has no raw frame to fold and the route has to decline it while the existing plan answers. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_columnar_config_falls_back_for_integer_field() { + common_telemetry::init_default_ut_logging(); + const TABLE: &str = "native_columnar_int"; + // Both functions are exact over the two samples inside the window: 10 + 30 and their mean. + const SUM: f64 = 40.0; + const AVG: f64 = 20.0; + let mito_config = mito2::config::MitoConfig { + experimental_enable_series_index: true, + experimental_series_index_format: mito2::config::SeriesIndexFormat::Columnar, + ..Default::default() + }; + let (standalone, observer) = native_test_instance( + GreptimeDbStandaloneBuilder::new("native_promql_columnar_fallback") + .with_mito_config(mito_config), + ) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + &format!( + "CREATE TABLE {TABLE} (host STRING, ts TIMESTAMP(3) TIME INDEX, val BIGINT, PRIMARY KEY (host)) ENGINE=mito; \ + INSERT INTO {TABLE} (host, ts, val) VALUES ('a', 60000, 10), ('a', 120000, 30);" + ), + QueryContext::arc(), + ) + .await; + for (query, expected) in [ + (format!("sum_over_time({TABLE}[2m])"), SUM), + (format!("avg_over_time({TABLE}[2m])"), AVG), + ] { + let sql = format!("TQL EVAL (120, 120, '1m') {query}"); + observer.clear(); + let native_rows = native_result_rows(native_tql_output(&instance, &sql, true).await).await; + assert_eq!( + native_rows.len(), + 1, + "native-on `{query}`: one row for the one series" + ); + assert_eq!(native_rows[0].0, 120_000, "native-on `{query}`: ts"); + assert_eq!(native_rows[0].1, "a", "native-on `{query}`: label"); + assert!( + (native_rows[0].2 - expected).abs() <= 1e-9 * expected.abs().max(1.0), + "native-on `{query}`: expected {expected}, found {}", + native_rows[0].2 + ); + let legacy_rows = native_result_rows(native_tql_output(&instance, &sql, false).await).await; + assert_eq!( + native_rows, legacy_rows, + "`{query}`: same rows on both routes" + ); + assert!( + observer.fused().is_none(), + "`{query}`: the native route must decline a field the columnar samples cannot serve" + ); + } +} diff --git a/tests/cases/standalone/common/view/tql_eval.result b/tests/cases/standalone/common/view/tql_eval.result deleted file mode 100644 index 704ba763969..00000000000 --- a/tests/cases/standalone/common/view/tql_eval.result +++ /dev/null @@ -1,41 +0,0 @@ -CREATE TABLE tql_view_source( - ts TIMESTAMP TIME INDEX, - host STRING, - val DOUBLE, - PRIMARY KEY (host) -); - -Affected Rows: 0 - -INSERT INTO tql_view_source VALUES (0, 'a', 2); - -Affected Rows: 1 - -CREATE VIEW tql_view AS TQL EVAL (0, 0, '1s', '5m') tql_view_source; - -Affected Rows: 0 - -SELECT * FROM tql_view; - -+---------------------+------+-----+ -| ts | host | val | -+---------------------+------+-----+ -| 1970-01-01T00:00:00 | a | 2.0 | -+---------------------+------+-----+ - -SHOW CREATE VIEW tql_view; - -+----------+---------------------------------------------------------------------+ -| View | Create View | -+----------+---------------------------------------------------------------------+ -| tql_view | CREATE VIEW tql_view AS TQL EVAL (0, 0, '1s', '5m') tql_view_source | -+----------+---------------------------------------------------------------------+ - -DROP VIEW tql_view; - -Affected Rows: 0 - -DROP TABLE tql_view_source; - -Affected Rows: 0 - diff --git a/tests/cases/standalone/common/view/tql_eval.sql b/tests/cases/standalone/common/view/tql_eval.sql deleted file mode 100644 index 367eecb6bde..00000000000 --- a/tests/cases/standalone/common/view/tql_eval.sql +++ /dev/null @@ -1,18 +0,0 @@ -CREATE TABLE tql_view_source( - ts TIMESTAMP TIME INDEX, - host STRING, - val DOUBLE, - PRIMARY KEY (host) -); - -INSERT INTO tql_view_source VALUES (0, 'a', 2); - -CREATE VIEW tql_view AS TQL EVAL (0, 0, '1s', '5m') tql_view_source; - -SELECT * FROM tql_view; - -SHOW CREATE VIEW tql_view; - -DROP VIEW tql_view; - -DROP TABLE tql_view_source;