rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 06:46:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(query): preserve legacy semantics at native admission

Browse Source 
Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-10-24 18:58:51 +08:00
parent 990d932198
commit ebfbd2ae5d
11 changed files with 754 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/AGENTS.md
+18 -5
View File
@@ -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`.
src/query/src/dist_plan/commutativity.rs
+8 -14
View File
@@ -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()
src/query/src/planner.rs
+119 -2
View File
@@ -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<LogicalPlan> {
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::<DistPlannerOptions>()
.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::<bool>().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.
src/query/src/promql/planner.rs
-19
View File
@@ -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())
src/query/src/promql/planner/native.rs
+42 -13
View File
@@ -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;
src/query/src/promql/planner/test.rs
+192 -5
View File
@@ -972,6 +972,73 @@ async fn native_engine_provider(partition_key_indices: Vec<usize>) -> 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() {
src/sql/src/parsers/create_parser.rs
+1 -60
View File
@@ -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";
src/sql/src/statements/tql.rs
+1 -1
View File
@@ -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}")
tests-integration/src/tests/promql_test.rs
+373 -19
View File
@@ -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<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
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::<TimestampMillisecondArray>()
.unwrap();
let hosts = cast(
batch.column_by_name("host").unwrap().as_ref(),
&DataType::Utf8,
)
.unwrap();
let hosts = hosts.as_any().downcast_ref::<StringArray>().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::<StringArray>().unwrap();
let values = batch
.columns()
.iter()
.filter(|column| column.data_type() == &DataType::Float64)
.collect::<Vec<_>>();
assert_eq!(values.len(), 1, "expected one Float64 value column");
let values = values[0].as_any().downcast_ref::<Float64Array>().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::<Vec<_>>();
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<dyn datafusion::physical_plan::ExecutionPlan>) {
use datafusion::physical_plan::metrics::MetricValue;
@@ -1893,20 +1987,35 @@ fn assert_four_native_partitions(plan: &Arc<dyn datafusion::physical_plan::Execu
);
}
/// A standalone instance with the native route opted in, and an observer of the plans it runs.
async fn native_test_instance(
builder: GreptimeDbStandaloneBuilder,
) -> (
crate::standalone::GreptimeDbStandalone,
Arc<NativePlanObserver>,
) {
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<NativePlanObserver>,
) {
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"
);
}
}
tests/cases/standalone/common/view/tql_eval.result
-41
View File
@@ -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
tests/cases/standalone/common/view/tql_eval.sql
-18
View File
@@ -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;