feat: lower SQL through unified operators and scalars

[zzylol]

Problem: no frontend produces the #511 OperatorNode / ScalarExpr IR

#511 (decoupling_op_and_expr.md, §2.1 Operator nodes) says:

Names are resolved to ColumnId before constructing these nodes. Parsing and unresolved ColumnRef handling remain frontend concerns.

§3.2 (SQL semantic mapping) then lists how each SQL construct must map onto those nodes, for example "FROM t, WHERE x > 1, SELECT x * 2 → Scan, Filter, Project; scalar Compare, Arithmetic", "VALUES (1), (2); SELECT 1 → Values; Project over one empty row", "ORDER BY, LIMIT, offset-only queries → Sort, Limit … n = None means no fetch limit", and "Uncorrelated scalar subquery, EXISTS, IN / NOT IN (SELECT ...) → explicit scalar plan-reading variants". #509 §0 (Language-specific frontends) adds that the frontend preserves source-language behavior and rejects "a construct that cannot be represented faithfully".

Before this PR, #537 has the resolved IR (OperatorNode, NonASAPOp, ScalarExpr) in asap_types::ir, but no crate outside crates/types builds it. The SQL frontend still emits the legacy mixed graph: UnresolvedQueryExpr (QueryExpr<ColumnRef>), resolved by asap_types::pre_asap::resolve_root into QueryExpr<ColumnId>, where a scalar is just another QueryExpr node:

SELECT service FROM metrics WHERE service IN (SELECT service FROM hosts)

legacy frontend (before)                         this PR, unresolved tree
Project                                          Project
└── Join(Semi, pred = Compare(service,           └── Filter(pred = InSubquery {
    │      Rc<QueryExpr> = __asap_in_key))               expr: Column(service),
    ├── Scan metrics                                     subquery: Rc<UnresolvedOp>,  ← own scope
    └── Project(__asap_in_key) ← synthetic name          negated: false })
        └── Scan hosts                               └── Scan metrics

Concrete gaps in the legacy SQL path that §3.2 calls out:

Query	Legacy frontend	§3.2 requires
... WHERE service NOT IN (SELECT service FROM hosts)	rejected: "NOT IN (subquery): its NULL semantics are not an anti-join"	InSubquery { negated: true } with SQL NOT IN NULL rules
SELECT (SELECT max(latency) FROM metrics) FROM hosts	rejected: "subquery-valued expression in predicate"	ScalarSubquery(Rc<OperatorNode>)
SELECT 1, VALUES (1,'a'),(2,'b')	no EmptyRelation / Values arm	Project over Values with one empty row; Values
... OFFSET 5 with no LIMIT	n: usize::MAX	Limit { n: None }
WHERE ts < NOW()	no TypeCoercion pass, so the implicit conversion is not recorded	explicit Cast(CurrentTimestamp → Timestamp)

What this PR covers: the SQL part of #511 §2.1/§2.2/§3.2, plus the shared name resolver that §2.1 leaves to the frontend. It adds a common frontend crate with a name-based tree and a resolver, and a complete second SQL lowering under asap_frontend_sql::unified that emits it.

What it leaves out: PromQL lowering (§3.3, #540; this PR only adds the resolver arms PromQL will use), planner cutover (callers stay on the legacy API), and the §3.4 gaps (correlated IN / scalar subqueries, GROUPS frames, DISTINCT ON), which are rejected.

Proposed method

All of this is frontend (#509 stage 0). Nothing runs in Pass 1/Pass 2, physical planning, selection or runtime.

1. Name-based tree (asap-frontend-common::unresolved). UnresolvedOp / UnresolvedScalar mirror NonASAPOp / ScalarExpr variant for variant. There are three differences: every ColumnId is a name-based ColumnRef, Scan.schema is Option<Schema> (known for a catalog SQL table, None for a schemaless PromQL leaf), and children are Rc<UnresolvedOp> because no schema exists yet. Two extra variants (PromqlMap, PromqlScalarOp) exist only here; the resolver expands them into Project / Filter.

2. SQL → tree (frontend-sql::unified::sql). SqlLowerer::lower parses (DataFusion dialect with aggregate FILTER, or ClickHouse), plans with DataFusion, runs ApplyFunctionRewrites and then DataFusion's TypeCoercion analyzer, so implicit conversions become explicit Casts. It rejects temporal subtraction that would produce a Duration type. It then walks the unoptimized LogicalPlan:

   • TableScan → Scan { schema: Some(catalog schema) }; a WHERE directly over a bare scan folds into Scan.predicates (filter_or_fold).
   • EmptyRelation → Values with one empty row (or zero rows); Values → Values typed from DataFusion's schema.
   • Every Compare / Arithmetic / Negative carries ExprSemantics::Sql. BETWEEN becomes >= AND <= (NOT BETWEEN becomes < OR >). NOW() / CURRENT_TIMESTAMP become CurrentTimestamp.
   • count(x) over a nullable x keeps SQL NULL skipping as a per-measure filter x IS NOT NULL. ROLLUP / CUBE / grouping sets expand to one Aggregate per level under a Concat.
   • LIMIT with no literal fetch → n: None.
   • Subqueries: an uncorrelated ScalarSubquery / EXISTS / IN becomes the matching UnresolvedScalar variant, and its plan is lowered as a separate root. Each subquery conjunct of a WHERE becomes its own Filter above the ordinary predicate. Only a correlated EXISTS is still lowered to a semi/anti join in the frontend, because its correlation predicate needs both sides. Other correlated subqueries are rejected.

3. Resolve (asap-frontend-common::resolve). resolve_root walks the tree bottom-up. Each child becomes an OperatorNode first, and its derived schema is the scope the parent's ColumnRefs resolve against. Scope rules:

   • Aggregate.having resolves against the aggregate's output schema.
   • Join / SetOp sides and plan-reading scalars (ScalarSubquery, Exists, InSubquery, PromqlScalarFromVector) each resolve as a root in their own scope. A Join predicate sees left ++ right.
   • A BinaryOp side is its own root but inherits label names that the enclosing scope references (issue L3: sum by (__name__) fails column resolution (metric-name label not in usage-derived schema) #52).
   • A Scan with schema: None binds to the usage-derived schema from SchemaResolver: ts, value, plus one nullable Utf8 field per referenced name, closed: false.

   The result then goes through the existing canonicalize pass (which turns Filter(Exists | IN) conjuncts into semi/anti joins and recognizes count-ranked top-k) and validate_structure.

4. Entry points. unified::lower_sql / lower_sql_dialect / lower_sql_batch mirror the legacy root API but return Rc<OperatorNode>. The legacy asap_frontend_sql::{lower_sql, …} is untouched. No bridge operators and no conversion between the two graphs are added.

Key code interfaces

Name-based tree — crates/frontend-common/src/unresolved.rs:

pub struct UnresolvedPredicate(pub UnresolvedScalar);
pub struct UnresolvedProjectItem { pub alias: Option<String>, pub expr: UnresolvedScalar }
pub struct UnresolvedSortKey { pub expr: UnresolvedScalar, pub ascending: bool, pub nulls_first: bool }

pub enum UnresolvedScalar {
    Column(ColumnRef),
    // Literal, Negative, Compare, BoolAnd, BoolOr, Not, IsNull, IsNotNull, Cast, InList,
    // FunctionCall, Arithmetic, Case, CurrentTimestamp, EvalTimestamp:
    // same fields as ScalarExpr, with Box<UnresolvedScalar> children.
    PromqlScalarFromVector(Rc<UnresolvedOp>),
    ScalarSubquery(Rc<UnresolvedOp>),
    Exists { subquery: Rc<UnresolvedOp>, negated: bool },
    InSubquery { expr: Box<UnresolvedScalar>, subquery: Rc<UnresolvedOp>, negated: bool },
}

pub enum UnresolvedOp {
    Scan { source: Source, predicates: Vec<UnresolvedPredicate>, schema: Option<Schema> },
    Values { rows: Vec<Vec<UnresolvedScalar>>, schema: Schema },
    Limit { n: Option<usize>, offset: usize, partition_by: GroupKeys<ColumnRef>, child: Rc<UnresolvedOp> },
    // Filter, Project, Aggregate, Join, SetOp, Concat, Dedup, Sort, BinaryOp, SQLWindowFunc,
    // TimeRange, TimeShift, PromqlVectorFromScalar, PromqlRelabel, PromqlInfoEnrich,
    // PromqlSeriesSample, PromqlSubquery: same fields as NonASAPOp, ColumnRef for ColumnId.
    PromqlMap { child: Rc<UnresolvedOp>, sample: UnresolvedScalar, drop_metric_name: bool },
    PromqlScalarOp {
        child: Rc<UnresolvedOp>, scalar: UnresolvedScalar, op: BinaryOpKind,
        scalar_left: bool, return_bool: bool,
    },
}

impl UnresolvedScalar {
    pub fn children(&self) -> Vec<&UnresolvedScalar>;
    pub fn columns_referenced(&self) -> Vec<&ColumnRef>;
    pub fn operator_refs(&self) -> Vec<&Rc<UnresolvedOp>>;
}
impl UnresolvedOp {
    pub fn concat(children: Vec<UnresolvedOp>) -> Self;
    pub fn concat_with_discriminator(children: Vec<UnresolvedOp>, discriminator: ColumnRef, inner_key: Vec<ColumnRef>) -> Self;
    pub fn scalar_exprs(&self) -> Vec<&UnresolvedScalar>;
}

Resolver — crates/frontend-common/src/resolve.rs, schema_resolver.rs:

pub fn resolve_root(tree: &UnresolvedOp) -> Result<Rc<OperatorNode>, ResolveDAGError>;
pub fn resolve_expr(expr: &UnresolvedScalar, schema: &Schema) -> Result<ScalarExpr, ResolveDAGError>;
pub fn resolve_scalar_root(tree: &UnresolvedScalar) -> Result<ScalarExpr, ResolveDAGError>;

pub enum ResolveDAGError {
    Resolve(ResolveError),
    Schema(SchemaDerivationError),
}

pub trait SchemaCatalog { fn columns_for(&self, source: &str) -> Option<Vec<Field>>; }
pub struct UsageDerivedCatalog;                       // knows nothing
pub struct SchemaResolver<C: SchemaCatalog = UsageDerivedCatalog> { catalog: C }
impl SchemaResolver<UsageDerivedCatalog> { pub fn new() -> Self; }
impl<C: SchemaCatalog> SchemaResolver<C> {
    pub fn with_catalog(catalog: C) -> Self;
    pub fn resolve_schema(&self, tree: &UnresolvedOp) -> Schema;
    pub fn resolve_schema_with_inherited(&self, tree: &UnresolvedOp, inherited: &[String]) -> Schema;
}
pub fn collect_referenced_columns(tree: &UnresolvedOp) -> Vec<String>;

SQL frontend — crates/frontend-sql/src/unified/mod.rs, unified/sql/mod.rs, unified/sql/types.rs, unified/error.rs:

pub async fn lower_sql(query: &str, catalog: &SqlCatalog, accuracy: AccuracyTarget)
    -> Result<Rc<OperatorNode>, SqlError>;
pub async fn lower_sql_dialect(query: &str, catalog: &SqlCatalog, dialect: SqlDialect, accuracy: AccuracyTarget)
    -> Result<Rc<OperatorNode>, SqlError>;
pub async fn lower_sql_batch(workload: &QueryWorkload, catalog: &SqlCatalog)
    -> Vec<Result<Rc<OperatorNode>, SqlError>>;

pub struct SqlLowerer<'a> { catalog: &'a SqlCatalog, dialect: SqlDialect }
impl<'a> SqlLowerer<'a> {
    pub fn new(catalog: &'a SqlCatalog) -> Self;
    pub fn with_dialect(catalog: &'a SqlCatalog, dialect: SqlDialect) -> Self;
    pub async fn lower(&self, sql: &str, accuracy: &AccuracyTarget) -> Result<UnresolvedOp, SqlError>;
}

pub struct SqlCatalog { pub tables: HashMap<String, Schema> }
impl SqlCatalog {
    pub fn new() -> Self;
    pub fn with_table(self, name: impl Into<String>, schema: Schema) -> Self;
}

pub enum SqlError {
    DataFusion(DataFusionError), TableNotFound(String), UnsupportedAggregate(String),
    InvalidExpression(String), UnsupportedDialect(String), UnsupportedFeature(String),
    WrongLanguage(String), Convert(ResolveDAGError),
}

Series identity — crates/types/src/ir/schema_support.rs:

pub fn with_promql_series_identity(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, String>;

Usage (from the tests):

let root: Rc<OperatorNode> =
    asap_frontend_sql::unified::lower_sql("SELECT 1", &catalog, AccuracyTarget::Exact).await?;
// or, to inspect the tree before resolution:
let tree: UnresolvedOp = SqlLowerer::new(&catalog).lower(sql, &AccuracyTarget::Exact).await?;
let root = asap_frontend_common::resolve_root(&tree)?;

Other new items, not public API: collection_planning, dialect, clickhouse_ast (copied from the legacy SQL module; collection_planning is changed slightly), and the pre_asap::SchemaDerivationError re-export.

Fields

Tree wrappers

Item	Type	Meaning
UnresolvedPredicate.0	UnresolvedScalar	A Boolean condition. Becomes ir::Predicate.
UnresolvedProjectItem.alias	Option<String>	Output name. None keeps the name derived from the expression.
UnresolvedProjectItem.expr	UnresolvedScalar	The projected value.
UnresolvedSortKey.expr / ascending / nulls_first	scalar / bool / bool	Sort expression, direction and NULL placement, as in ir::SortKey.

UnresolvedScalar (the mirrored variants have the meaning of the same ScalarExpr variant)

Variant / field	Meaning
Column(ColumnRef)	Name-based reference: Named, Qualified { table, name }, SampleValue or Wildcard. SQL keeps DataFusion's relation qualifier, so a.k and b.k across a join resolve to the right side. Resolved to ScalarExpr::Column(ColumnId).
PromqlScalarFromVector(op)	PromQL scalar(v). op resolves as its own root. Not emitted by SQL.
ScalarSubquery(op)	Uncorrelated (SELECT …) in a value position. op resolves as its own root. The IR contract: zero rows gives typed NULL, more than one row is an error.
Exists { subquery, negated }	[NOT] EXISTS (subquery). negated = true for NOT EXISTS.
InSubquery { expr, subquery, negated }	expr [NOT] IN (subquery). expr resolves in the outer scope, subquery in its own scope. The SQL frontend requires the subquery to have exactly one column. negated = true is NOT IN, kept with SQL NULL rules rather than lowered to an anti-join.

UnresolvedOp (only the fields that differ from NonASAPOp, plus frontend-only variants)

Variant / field	Type	Meaning
Scan.source	Source	Table { table_ref } for SQL, TimeSeries { metric } for PromQL.
Scan.predicates	Vec<UnresolvedPredicate>	Row filters folded onto the leaf. They resolve against the scan schema.
Scan.schema	Option<Schema>	Some = catalog-backed SQL table (closed schema). None = bind to the SchemaResolver usage-derived schema.
Values.rows	Vec<Vec<UnresolvedScalar>>	One inner vector per row. Expressions have no input-column scope (resolved against an empty schema). vec![vec![]] is the one empty row of SELECT 1. vec![] is an empty relation.
Values.schema	Schema	Declared row type, from DataFusion's planned schema. Always closed: true, no time index or keys.
Limit.n	Option<usize>	Fetch count. None = offset-only, no limit (was usize::MAX in the legacy tree).
Limit.offset	usize	Rows skipped. 0 when there is no OFFSET.
Limit.partition_by	GroupKeys<ColumnRef>	Per-group limit keys. SQL always sets GroupKeys::none().
Limit.child	Rc<UnresolvedOp>	Input.
PromqlMap.child / sample / drop_metric_name	op / scalar / bool	Frontend-only: rewrite the sample column with sample and keep every other field. The resolver first adds the full series identity (if the child schema is open), then emits a Project. drop_metric_name removes __name__. Used by PromQL (#540).
PromqlScalarOp.child / scalar / op / scalar_left / return_bool	op / scalar / BinaryOpKind / bool / bool	Frontend-only vector–scalar operation. scalar resolves in an empty scope. scalar_left gives the operand order. Arithmetic becomes a Project. A comparison becomes a Filter, or a Project of Case(cmp, 1.0, 0.0) when return_bool is set or op is CompareBool. Set ops are rejected. Used by PromQL (#540).

Helper methods: children returns the direct scalar sub-expressions (not the operators they read). columns_referenced returns every ColumnRef in this expression, not inside the operators it reads. operator_refs returns the operators the expression reads, transitively. concat builds a Concat with no unique-key claim. concat_with_discriminator builds a Concat with a caller-proven (discriminator, inner_key) unique key; nothing verifies the claim. scalar_exprs returns every scalar the operator owns.

Resolver

Item	Meaning
resolve_root(tree)	Resolve the whole tree with no inherited names, then canonicalize and validate_structure. Returns the resolved root.
resolve_expr(expr, schema)	Resolve one scalar against schema. Operators it reads become their own roots and inherit schema's label names (all fields except ts and value).
resolve_scalar_root(tree)	Resolve a standalone scalar in an empty column scope and type-check it. A free column is an error.
ResolveDAGError::Resolve	A ColumnRef did not resolve against its scope (missing or ambiguous name).
ResolveDAGError::Schema	Schema derivation failed while building a node, including invalid scalar signatures and SQL result-type checks.
SchemaCatalog::columns_for(source)	Known fields for a metric/table name. None = unknown, so fall back to usage.
UsageDerivedCatalog	Default catalog that always returns None.
SchemaResolver.catalog (private)	The catalog consulted for the leftmost Scan source. Set by new() (usage-derived) or with_catalog(c).
resolve_schema(tree)	The binding schema for a schemaless leaf: catalog fields or (ts: Timestamp, value: Float64), plus one nullable Utf8 field per referenced name. time_index = position of ts, no unique keys, closed: false.
resolve_schema_with_inherited(tree, inherited)	The same schema, plus label names referenced only by an enclosing scope (job in sum by (job)(a or b), issue #52).
collect_referenced_columns(tree)	Sorted, deduplicated names referenced anywhere in the tree, including inside operators read from scalar positions.

SQL frontend

Item	Meaning
lower_sql(query, catalog, accuracy)	lower_sql_dialect with SqlDialect::DataFusionSQL.
lower_sql_dialect(…, dialect, …)	SqlLowerer::with_dialect(…).lower(…), then resolve_root. ClickhouseSQL parses with sqlparser's ClickHouse dialect.
lower_sql_batch(workload, catalog)	One Result per query_batch entry; errors do not stop the batch. Each entry uses its own requirements.accuracy.target(). Non-SQL language → WrongLanguage for every entry. ElasticSQL → UnsupportedDialect. Empty batch → vec![].
SqlLowerer.catalog (private)	Tables registered with DataFusion as empty MemTables; their schemas become Scan.schema.
SqlLowerer.dialect (private)	Parser dialect. new uses DataFusionSQL; with_dialect sets it.
SqlLowerer::lower(sql, accuracy)	Parse, plan, coerce, then build the UnresolvedOp tree. accuracy is attached to every approximate intent (Count, Quantile, Cardinality). Exactly one statement per query.
SqlCatalog.tables	Table name → Schema. with_table adds one entry.
SqlError::DataFusion	Parse/plan/analyzer failure, including the temporal-subtraction check.
SqlError::TableNotFound	Table missing from the catalog.
SqlError::UnsupportedAggregate	Aggregate function not mapped.
SqlError::InvalidExpression	Expression that cannot be lowered (for example, a multi-column IN (subquery)).
SqlError::UnsupportedDialect	ElasticSQL.
SqlError::UnsupportedFeature	Rejected construct: correlated IN / scalar subquery, DISTINCT ON, GROUPS frames, unsupported join types, more than one statement.
SqlError::WrongLanguage	Batch language is not SQL.
SqlError::Convert	resolve_root failed (ResolveDAGError).

with_promql_series_identity(root) rebuilds the DAG (memoized by node pointer) so that every open TimeSeries scan gets a non-null Utf8 series-identity field and becomes closed. Errors: a closed scan without it, a without ranking Sort, or an operator with no rule for it. guarantee and timing are copied to the rebuilt nodes. The resolver calls it for PromqlMap / PromqlScalarOp over an open child.

Examples

End to end: SELECT service FROM metrics WHERE service NOT IN (SELECT service FROM hosts) (test not_in_subquery_is_rejected_rather_than_mislowered_as_an_anti_join; the test name is from the legacy behavior, but it now asserts the query is accepted). Catalog: metrics(ts, service, latency, bytes), hosts(service, region).

1. SqlLowerer::lower          → Project[service]
                                └── Filter(InSubquery { expr: Column(Named "service"),
                                    │                   subquery: <Project[service] ← Scan hosts>,
                                    │                   negated: true })
                                    └── Scan metrics (schema: Some(..))
2. resolve_root               → service → Column(1) against metrics' schema;
                                subquery resolved as its own root (service → Column(0) of hosts)
3. canonicalize               → NOT IN is not lowered to an anti-join; the Filter stays
4. validate_structure         → ok

The positive IN form goes through the same steps, and canonicalize turns it into Join(Semi) whose predicate is Column(1) = Column(4). Column 4 is the subquery key, bound by position, not by the shared name service (in_subquery_lowers_to_a_semi_join).

Accepted and rejected cases (all in crates/frontend-sql/tests/unified_sql_lowering.rs):

SQL	Result	Test
SELECT * FROM metrics WHERE service = 'api'	Scan with 1 predicate, closed schema	select_star_with_where_folds_predicate_onto_scan
SELECT bytes * 8 FROM metrics WHERE latency > 1.5	Arithmetic / Compare with ExprSemantics::Sql	sql_comparisons_and_arithmetic_carry_sql_semantics
SELECT 1	Project[Literal(Int64(1))] over Values { rows: [[]] }	select_without_from_projects_over_one_empty_row
SELECT * FROM (VALUES (1,'a'),(2,'b')) AS v(n, s)	Values with 2 rows, Int64/Utf8; output n, s	values_lowers_to_one_row_per_values_row
SELECT * FROM metrics WHERE ts < NOW()	Compare(ts, Cast(CurrentTimestamp → Timestamp))	now_in_predicate_lowers_to_current_timestamp
SELECT (SELECT max(latency) FROM metrics) FROM hosts	Project[ScalarSubquery(..)] over Scan hosts	scalar_subquery_in_projection_lowers_to_a_scalar_subquery_item
… WHERE bytes > (SELECT AVG(bytes) FROM metrics)	Filter(Compare(.., ScalarSubquery)) over Scan	scalar_subquery_in_predicate_lowers_through_a_cross_join
… WHERE EXISTS (SELECT 1 FROM hosts)	Filter(Exists), then Join(Semi, true) after canonicalize	exists_and_in_subqueries_lower_to_scalar_filter_conjuncts, an_uncorrelated_exists_is_an_unconditional_semi_join
correlated NOT EXISTS (… h.service = m.service)	Join(Anti) built in the frontend	not_exists_lowers_to_an_anti_join
count(nullable_value)	Count with measure filter IsNotNull	count_null_semantics_become_a_measure_filter
SUM(bytes * 2), MIN(bytes * 2)	one shared derived column	a_shared_expression_is_materialized_once
IN (SELECT service, region FROM hosts)	rejected: "exactly one column"	a_multi_column_in_subquery_is_rejected
correlated IN (… WHERE h.region = m.service)	rejected: "correlated IN"	a_correlated_in_subquery_is_rejected
… OVER (… GROUPS BETWEEN 2 PRECEDING AND CURRENT ROW)	rejected: GROUPS	groups_frame_is_rejected
count(nullable_value) … GROUP BY ROLLUP(value)	UnsupportedFeature (no measure filter inside multi-level grouping)	count_null_semantics_become_a_measure_filter

Out of scope

• PromQL lowering (feat(promql): lower queries to unified operator and scalar IR #540). This PR adds only the shared resolver arms for it.
• Promoting unified to the root API (planner cutover) and deleting the legacy SQL path (cleanup PR).
• Correlated subqueries other than EXISTS, DISTINCT ON, GROUPS frames and the other docs: define unified operators and SQL/PromQL scalar boundaries #511 §3.4 gaps. These are rejected.
• Physical runtime, materialization and workload selection.

Stack and validation

Revised logical foundation 3/5 · Previous: #537 · Next: #540 · Tracker: #528. Part of #528, #511, #509.

Review order: #567 → #560 → #537 → #539 → #540 → #561. This restack follows #560 and phase-free #537. SQL imports follow #536's scalar_type_rules naming.

Validation: the common-resolver tests pass, and both the existing and the unified SQL lowering suites pass, including 109 unified SQL tests. Workspace compilation, formatting, and workspace/all-target/all-feature Clippy pass. The full first-five tip passes 1,961 tests/doctests and workspace Clippy.

🤖 Generated with Claude Code