cqlite_core/query/select_executor/execute.rs
1//! Async pipeline entry points for the SELECT executor (issue #1174).
2//!
3//! This submodule continues the [`SelectExecutor`](super::SelectExecutor) `impl`
4//! with the large `async` pipeline methods that drive query execution:
5//! - [`SelectExecutor::execute`] — the materializing plan runner,
6//! - [`SelectExecutor::execute_sstable_scan`] — the SSTable-scan step.
7//!
8//! (The streaming producer, `execute_streaming_background`, moved to the sibling
9//! `streaming` submodule in issue #1578's file-size split.)
10//!
11//! These were relocated verbatim from `mod.rs` (epic #1116 file-size split); the
12//! per-step helpers they call (`execute_filter`, `execute_sort`, etc.) and the
13//! `ExecutionContext` bookkeeping struct remain in `mod.rs`. As a child module,
14//! this file can reach `mod.rs`'s private items directly — the logic, ordering,
15//! and error handling are unchanged.
16
17use super::schemaless_point::classify_schemaless_point_lookup;
18use super::{
19 apply_forcing, build_row_from_scan_cached, classify_partition_lookup,
20 collect_capped_materialized, column_info_from_type_str, full_forbids_schemaless_seek,
21 honest_targeted_path, parse_table_id, point_forbids_fallback, point_requires_engaged,
22 project_expr_reshapes_row, scan_pushdown_cap, select_has_writetime_ttl,
23 sort_metadata_rows_by_token, sort_rows_by_token, validate_token_predicates, ForcedPlan,
24 PartitionLookupOutcome, SSTablePredicate,
25};
26use super::{
27 AccessPath, ColumnInfo, ExecutionContext, ExecutionStep, FallbackReason, OptimizedQueryPlan,
28 ProjectionFlags, QueryResult, QueryRow, Result, SelectExecutor, TableId, TableSchema,
29};
30use crate::query::result_budget::enforce_materialized_rows;
31use std::sync::Arc;
32
33impl SelectExecutor {
34 /// Execute an optimized query plan.
35 ///
36 /// Instrumented as `query.select.plan` (issue #1035): this span covers the
37 /// modern SELECT pipeline — SSTable scan, filtering, projection, aggregation,
38 /// and WRITETIME/TTL metadata extraction — and is the parent under which the
39 /// read-path spans (issue #1034) nest. On completion it emits
40 /// [`catalog::QUERY_ROWS_SCANNED`] (rows the scan step examined) dimensioned by
41 /// the honest access path, so the rows-scanned vs rows-returned gap is
42 /// observable. The bounded access-path attribute is recorded on the span; the
43 /// query text and key values never are.
44 #[tracing::instrument(
45 name = "query.select.plan",
46 skip_all,
47 fields(
48 cqlite.query.access_path = tracing::field::Empty,
49 cqlite.query.rows_scanned = tracing::field::Empty,
50 cqlite.query.rows = tracing::field::Empty,
51 )
52 )]
53 pub async fn execute(&self, plan: OptimizedQueryPlan) -> Result<QueryResult> {
54 // Issue #960: clear the global access-path probe so a stale value from a
55 // previous query cannot satisfy a test assertion against this one.
56 crate::query::access_path::reset();
57
58 let table_id = if let Some(ref from_clause) = plan.statement.from_clause {
59 self.extract_table_id(from_clause)?
60 } else {
61 // For queries without FROM clause (like SELECT 1), use a dummy table ID
62 TableId::new("_dummy_")
63 };
64
65 // Issue #1587 (E5): resolve the table's schema ONCE per query into a shared
66 // `Arc<TableSchema>`. Column-metadata building, the SSTable scan, and the
67 // SELECT-* metadata fallback all borrow this same schema (ref-count bump),
68 // instead of each independently re-locking the registry and deep-cloning a
69 // fresh `TableSchema` (2–4 deep clones per query before this).
70 let query_schema: Option<Arc<TableSchema>> = if plan.statement.from_clause.is_some() {
71 self.resolve_table_schema(&table_id).await
72 } else {
73 None
74 };
75
76 // Issue #692: detect whether any WRITETIME/TTL select items are present
77 // during planning and set the opt-in flag so the reader threads per-cell
78 // metadata. This is the "planning" half of the executor wiring; the
79 // "evaluation" half lives in `evaluate_select_expression`.
80 let projection_flags = ProjectionFlags {
81 include_cell_metadata: select_has_writetime_ttl(&plan.statement),
82 };
83 tracing::debug!(
84 "Query plan: include_cell_metadata={}",
85 projection_flags.include_cell_metadata
86 );
87
88 let mut context = ExecutionContext {
89 table_id,
90 columns: self.get_result_columns(&plan.statement, query_schema.as_deref())?,
91 rows_processed: 0,
92 scan_rows: 0,
93 projection_flags,
94 access_path: None,
95 reverse_served: false,
96 };
97
98 // Handle queries without FROM clause (like SELECT 1)
99 if plan.statement.from_clause.is_none() {
100 let mut result = self.execute_constant_query(&plan.statement, &context)?;
101 // Issue #1582 (roborev): apply the statement's LIMIT/OFFSET to the
102 // constant rows BEFORE the byte + row-count budget check, so the budget
103 // is enforced on the rows ACTUALLY returned (post LIMIT/OFFSET) —
104 // consistent with the table-backed path below. In particular `LIMIT 0`
105 // must return empty, never `ResultTooLarge`; an over-budget constant
106 // SELECT with NO limit still trips the guard on its final rows.
107 let offset = plan.statement.offset.unwrap_or(0) as usize;
108 let limit = plan
109 .statement
110 .limit
111 .as_ref()
112 .map(|l| l.count as usize)
113 .unwrap_or(usize::MAX);
114 result.rows = result.rows.into_iter().skip(offset).take(limit).collect();
115 // Keep the row-count metadata consistent with the returned rows.
116 let returned = result.rows.len() as u64;
117 result.rows_affected = returned;
118 result.metadata.total_rows = Some(returned);
119 enforce_materialized_rows(&result.rows, self.max_result_bytes, self.max_result_rows)?;
120 return Ok(result);
121 }
122
123 // Execute the plan step by step
124 let mut intermediate_results = Vec::new();
125
126 // If no execution steps are provided, add a default table scan
127 let execution_steps = if plan.execution_steps.is_empty() {
128 vec![ExecutionStep::SSTableScan {
129 table: context.table_id.clone(),
130 predicates: vec![],
131 projection: context.columns.iter().map(|c| c.name.clone()).collect(),
132 }]
133 } else {
134 plan.execution_steps.clone()
135 };
136
137 // Issue #1577 (D1): decide ONCE — from the plan alone — whether the
138 // SSTable scan may stop early at `LIMIT + OFFSET` accepted rows. `None`
139 // (any Sort/Aggregate/PerPartitionLimit/residual-Filter, DISTINCT, or no
140 // LIMIT) leaves the scan unbounded, exactly as before. Computed here so it
141 // is available inside the step loop below (unused, harmlessly, when the
142 // #1578 global-aggregate fast path takes the scan's place).
143 let scan_cap = scan_pushdown_cap(&execution_steps, &plan.statement.select_clause);
144
145 // Issue #1578 (D2): fold a GROUP-BY-free aggregate over a full table scan
146 // into an O(1) accumulator instead of buffering the whole table here.
147 // Returns None for any plan shape it does not model (GROUP BY, targeted
148 // lookup, WRITETIME/TTL, or a Sort/Project/Limit step) → the buffered step
149 // loop below runs unchanged.
150 if let Some(rows) = self
151 .try_execute_global_aggregate(&execution_steps, query_schema.as_deref(), &mut context)
152 .await?
153 {
154 intermediate_results = rows;
155 } else {
156 for step in &execution_steps {
157 match step {
158 ExecutionStep::SSTableScan {
159 table,
160 predicates,
161 projection,
162 ..
163 } => {
164 let rows = self
165 .execute_sstable_scan(
166 table,
167 predicates,
168 projection,
169 plan.statement.order_by.as_ref(),
170 query_schema.as_deref(),
171 scan_cap,
172 &mut context,
173 )
174 .await?;
175 intermediate_results = rows;
176 }
177 ExecutionStep::Filter { expression, .. } => {
178 intermediate_results =
179 self.execute_filter(intermediate_results, expression, &mut context)?;
180 }
181 ExecutionStep::Sort { order_by, .. } => {
182 // Issue #1184: when the BIG reverse partition iterator already
183 // produced the rows in descending clustering order, skip the
184 // in-memory sort entirely (it remains the fallback otherwise).
185 if !context.reverse_served {
186 intermediate_results =
187 self.execute_sort(intermediate_results, order_by, &mut context)?;
188 } else {
189 // Issue #1307 (hardening): skipping this Sort is sound ONLY
190 // because `reverse_served` is set exclusively by
191 // `targeted_partition_rows`, which serves the reverse
192 // promoted-index iterator precisely when `statement.order_by`
193 // requests the reverse of the stored clustering order — and
194 // the planner emits EXACTLY ONE `Sort` step, cloned from that
195 // same `statement.order_by` (see `select_optimizer.rs`). So
196 // the reverse scan's ordering matches this step's `order_by`,
197 // and skipping it drops a redundant sort rather than a
198 // different ordering. What would break the invariant: a
199 // multi-table / join plan (or any plan) that reused the flag
200 // across a Sort NOT derived from the reverse-served scan's
201 // `statement.order_by` — such a plan must clear
202 // `reverse_served` before this step. The debug_assert pins
203 // the property (the skipped Sort's key equals the statement's
204 // order_by); it is debug-only and never alters release
205 // behavior.
206 debug_assert!(
207 plan.statement.order_by.as_ref() == Some(order_by),
208 "reverse_served Sort-skip invariant violated: the skipped \
209 Sort's order_by must be the statement's order_by that drove \
210 the reverse-served scan (single-table plan); a plan whose \
211 Sort is not the reverse scan's matching Sort must clear \
212 reverse_served first",
213 );
214 }
215 }
216 ExecutionStep::Aggregate { plan: agg_plan, .. } => {
217 intermediate_results = self.execute_aggregation(
218 intermediate_results,
219 agg_plan,
220 query_schema.as_deref(),
221 &mut context,
222 )?;
223 }
224 ExecutionStep::PerPartitionLimit { count } => {
225 intermediate_results =
226 Self::execute_per_partition_limit(intermediate_results, *count);
227 }
228 ExecutionStep::Limit { count, offset } => {
229 intermediate_results = self.execute_limit(
230 intermediate_results,
231 *count,
232 *offset,
233 &mut context,
234 )?;
235 }
236 ExecutionStep::Project { columns } => {
237 // Issue #1952 (round-6 fix): branch on whether the projection
238 // RESHAPES the row. A plain-column Project only trims the
239 // #1952-widened helper columns — route it through the
240 // key-preserving `trim_projection` so the row keeps its real
241 // RowKey / metadata and a sparse row's absent selected cell is
242 // omitted rather than erroring. Only a reshaping / computed
243 // projection (alias, arithmetic, aggregate, function,
244 // writetime/ttl, collection-access) goes through
245 // `execute_projection`, whose empty-RowKey + name-derivation is
246 // correct for a computed row that has no natural stored key.
247 intermediate_results = if columns.iter().any(project_expr_reshapes_row) {
248 self.execute_projection(intermediate_results, columns, &mut context)?
249 } else {
250 self.trim_projection(intermediate_results, columns)
251 };
252 }
253 }
254 }
255 }
256
257 // Issue #1582 (D6, narrow subset): enforce the byte-bounded result budget
258 // (primary) + the row-count safety valve (secondary) with a SINGLE robust
259 // check on the FINAL materialized result — AFTER every execution step
260 // (Limit/Offset/Filter/Sort/Aggregate/Project) has produced the rows that
261 // will actually be returned. Because this sees ONLY the returned rows
262 // (post LIMIT/OFFSET), a `LIMIT 10` query never trips and OFFSET-skipped
263 // rows are never charged. This replaces the earlier during-collection
264 // machinery (LIMIT/OFFSET pushdown, per-row early-stop, storage-layer row limit),
265 // which kept generating correctness edge cases; a single final-result
266 // check has no such edges. Reuses the shared `estimate_value_size`
267 // estimator (via `enforce_materialized_rows`).
268 //
269 // SCOPE (owner-accepted boundaries, tracked on #1582 — NOT bugs to fix in
270 // this narrow subset):
271 // * Does NOT bound PEAK scan memory: `storage.scan` still materializes
272 // each reader's matching rows before this point (deferred to #1897).
273 // * Does NOT cover the LEGACY point-lookup `QueryExecutor` path
274 // (`WHERE id = ?` short lookups route there, not through this modern
275 // executor) — deferred to the D6 redesign.
276 // Issue #1578 (D2): demote the row-count valve to a genuine safety valve.
277 // A query with an EXPLICIT `LIMIT` already bounds its own result, so it is
278 // exempt from the crude row-count ceiling (the user accepted the count);
279 // the byte budget still guards memory. Without an explicit LIMIT the valve
280 // remains a real net against unbounded materialization.
281 let effective_max_rows = if plan.statement.limit.is_some() {
282 usize::MAX
283 } else {
284 self.max_result_rows
285 };
286 enforce_materialized_rows(
287 &intermediate_results,
288 self.max_result_bytes,
289 effective_max_rows,
290 )?;
291
292 let total_rows = intermediate_results.len() as u64;
293
294 // CRITICAL FIX (Issue #129/#140): Populate metadata.columns for SELECT *
295 // When SELECT * is used and no schema was found, context.columns is empty.
296 // Fall back to inferring column names from the first row's HashMap keys.
297 // IMPORTANT: Must be sorted alphabetically for deterministic JSON output (Issue #129)!
298 let mut columns = context.columns;
299 if columns.is_empty() && !intermediate_results.is_empty() {
300 // Issue #1587 (E5): reuse the schema resolved once at the top of the
301 // query rather than re-locking the registry + deep-cloning here.
302 let schema_opt = query_schema.as_deref();
303
304 let first_row = &intermediate_results[0];
305 let mut col_names: Vec<_> = first_row.values.keys().collect();
306 col_names.sort(); // Sort alphabetically for deterministic ordering (Issue #129)
307
308 let table_name_for_meta = schema_opt.map(|s| format!("{}.{}", s.keyspace, s.table));
309
310 for (idx, col_name) in col_names.iter().enumerate() {
311 let col_name: &str = col_name;
312 // Look up CQL type from schema; derive flat DataType from it (Issue #674).
313 let col_info = match schema_opt
314 .and_then(|schema| schema.columns.iter().find(|c| c.name.as_str() == col_name))
315 {
316 Some(schema_col) => column_info_from_type_str(
317 col_name.to_string(),
318 &schema_col.data_type,
319 idx,
320 table_name_for_meta.clone(),
321 ),
322 None => ColumnInfo {
323 name: col_name.to_string(),
324 data_type: crate::types::DataType::Text,
325 nullable: true,
326 position: idx,
327 table_name: table_name_for_meta.clone(),
328 cql_type: None,
329 },
330 };
331 columns.push(col_info);
332 }
333 }
334
335 // Observability (issue #1035): the `query.select.plan` span declared
336 // `access_path`/`rows_scanned`/`rows` but never recorded them, and
337 // `QUERY_ROWS_SCANNED` was never emitted. Do both here, sourced from the
338 // honest per-query signal (`context.access_path`, set by the SSTable-scan
339 // step) and the rows the scan examined (`context.rows_processed`). Bounded
340 // attributes only — never the query text or key values.
341 //
342 // Issue #2162: this stays a SINGLE-SHOT emission deliberately. The Flight
343 // `do_get` merge/scan path gained incremental `QUERY_ROWS_SCANNED` deltas
344 // (`cqlite-flight/src/scan_progress.rs`), but that path never runs through
345 // `SelectExecutor` — Flight drives `KWayMerger` directly. Making THIS site
346 // incremental too would additionally have to interact with
347 // `limit_pushdown`'s `scan_rows` rebaseline/decode-stop accounting
348 // (`limit_pushdown/mod.rs`), which re-runs a capped scan and resets
349 // `context.scan_rows` — a materially riskier change with no scenario in
350 // #2162's spec requiring it.
351 {
352 use crate::observability::{self as obs, catalog, AttrValue};
353
354 let access_path_label: &'static str = context
355 .access_path
356 .as_ref()
357 .map(|p| p.label())
358 .unwrap_or("unknown");
359
360 obs::add_counter(
361 catalog::QUERY_ROWS_SCANNED,
362 context.scan_rows,
363 &[(
364 catalog::attr::ACCESS_PATH,
365 AttrValue::StaticStr(access_path_label),
366 )],
367 );
368
369 let span = tracing::Span::current();
370 span.record(catalog::attr::ACCESS_PATH, access_path_label);
371 span.record("cqlite.query.rows_scanned", context.scan_rows);
372 span.record("cqlite.query.rows", total_rows);
373 }
374
375 // Issue #1035: carry a bounded plan family on the result so the engine's
376 // single observability chokepoint reports a real plan type for SELECTs
377 // (the modern executor previously always returned `plan_info: None`,
378 // forcing plan_type to "unknown").
379 let plan_info = Self::select_plan_info(&plan, context.access_path.as_ref());
380
381 Ok(QueryResult {
382 rows: intermediate_results,
383 rows_affected: total_rows, // Use actual number of rows returned
384 execution_time_ms: 0, // Will be set by the engine
385 metadata: crate::query::result::QueryMetadata {
386 columns,
387 total_rows: Some(total_rows),
388 plan_info: Some(plan_info),
389 performance: Default::default(),
390 warnings: vec![],
391 // Issue #960: surface the access path the SSTable-scan step chose
392 // on the result from PER-QUERY state (not the global probe), so a
393 // concurrent SELECT cannot overwrite it between record() and here.
394 access_path: context.access_path.clone(),
395 },
396 })
397 }
398
399 /// Execute SSTable scan with predicate pushdown.
400 ///
401 /// Per-row work (build row, decode partition key, evaluate predicates) is
402 /// handled by the free helpers `build_row_from_scan` and
403 /// `evaluate_predicates`, which are shared with the streaming background
404 /// task to keep the two execution paths in lockstep.
405 ///
406 /// Issue #1582 (D6, narrow subset): the single byte/row budget check is
407 /// applied ONCE by [`SelectExecutor::execute`] on the FINAL result, after the
408 /// whole step pipeline (post LIMIT/OFFSET), never mid-collection here.
409 ///
410 /// Issue #1577 (D1): when the caller determines the plan is LIMIT-pushdown
411 /// safe it passes `scan_cap = Some(limit + offset)` — the number of ACCEPTED
412 /// (post-marker, post-predicate) rows the downstream `Limit` needs. This scan
413 /// then stops after that many accepted rows: the full-scan fallback stops
414 /// DECODING early (via [`capped_fallback_scan`](Self::capped_fallback_scan)'s
415 /// bounded stream), and the partition-targeted paths stop the per-row build
416 /// loop. `scan_cap` is `None` (unbounded, as before) whenever a
417 /// Sort/Aggregate/PerPartitionLimit/residual-Filter step or DISTINCT follows,
418 /// or there is no `LIMIT` — so a `WHERE non_pk = ? LIMIT N` still counts
419 /// ACCEPTED rows, never raw rows, and can never silently drop matches.
420 #[cfg_attr(feature = "tombstones", allow(unused_variables))]
421 pub(super) async fn execute_sstable_scan(
422 &self,
423 table: &TableId,
424 predicates: &[SSTablePredicate],
425 projection: &[String],
426 order_by: Option<&crate::query::select_ast::OrderByClause>,
427 // Issue #1587 (E5): schema resolved ONCE per query by the caller and
428 // shared by reference — no per-scan registry lock + deep clone.
429 schema_opt: Option<&TableSchema>,
430 // Issue #1577 (D1): `Some(limit + offset)` when LIMIT pushdown is safe.
431 scan_cap: Option<usize>,
432 context: &mut ExecutionContext,
433 ) -> Result<Vec<QueryRow>> {
434 // FINDING 2 (Issue #955 follow-up): a `token(...)` predicate is evaluated
435 // by hashing the row's raw partition key, so its argument columns MUST be
436 // the full partition key in declared order or the result is silently
437 // wrong. Reject (Cassandra-style) before scanning/evaluating.
438 validate_token_predicates(predicates, schema_opt)?;
439
440 // Issue #1918: resolve the read-path forcing mode ONCE for this scan step
441 // (config over env over auto). An invalid env value fails the query loudly
442 // here rather than silently running under auto.
443 let mode = self.resolved_read_path_mode()?;
444
445 // Data-safety (issue #1694): log the SHAPE of the scan — predicate count
446 // and the constrained column names — never the predicate literals/values.
447 tracing::debug!(
448 "Executing SSTableScan: table=\"{}\", predicates={} on [{}], include_cell_metadata={}",
449 table,
450 predicates.len(),
451 predicates
452 .iter()
453 .map(|p| p.column.as_str())
454 .collect::<Vec<_>>()
455 .join(", "),
456 context.projection_flags.include_cell_metadata,
457 );
458
459 let (keyspace, table_name) = parse_table_id(table);
460
461 match schema_opt {
462 Some(schema) => tracing::debug!(
463 "Found schema for {}.{} with {} columns",
464 schema.keyspace,
465 schema.table,
466 schema.columns.len()
467 ),
468 None => tracing::debug!(
469 "No schema found for {}.{}, proceeding without schema-aware parsing",
470 keyspace.as_deref().unwrap_or("unknown"),
471 table_name
472 ),
473 }
474
475 // Issue #1750: a SCHEMA-LESS `WHERE pk = <literal>` point read can be served
476 // by a key-byte-targeted seek ONLY when the equality column is a
477 // metadata-CONFIRMED sole partition key (single-component pk, no clustering
478 // keys, column absent from the authoritative non-key column names). Resolve
479 // that authoritative shape from the Statistics.db SerializationHeader (the
480 // metadata a schema-less reader HAS) and classify BY ELIMINATION — never a
481 // pk-name/text guess (#28). `None` (metadata unavailable, or the column is a
482 // non-pk column) keeps the honest full-scan path, which correctly matches
483 // regular-column equalities. Resolved only when there is no CQL schema and no
484 // WRITETIME/TTL metadata projection (the two branches below own those).
485 // Issue #1918: `point`/`auto` keep the schema-less sole-pk targeted seek
486 // (#1750). Under forced `full` the query cannot take the general full-scan
487 // path for THIS shape: with no schema the per-row predicate backstop in
488 // `collect_capped_materialized` cannot reconstruct the pk column to match
489 // the literal, so a full scan would silently return 0 rows instead of the
490 // row `auto` returns — violating the spec's "identical to `auto`" SHALL.
491 // Fail closed with a clear error (mirror of `point_forbids_fallback`: full
492 // forbidding a shape only the specialized non-full seek can serve).
493 let schemaless_seek: Option<super::schemaless_point::SchemalessPointSeek> =
494 if schema_opt.is_none() && !context.projection_flags.include_cell_metadata {
495 let shape = self.storage.partition_key_shape(table).await;
496 let seek = classify_schemaless_point_lookup(predicates, shape.as_ref());
497 full_forbids_schemaless_seek(mode, seek.is_some())?;
498 // Under `auto`/`point` keep the seek; under `full` `seek` is `None`
499 // here only when the query did not qualify (the qualifying case
500 // errored above), so it correctly falls through to the full scan.
501 if mode == crate::config::ReadPathMode::Full {
502 None
503 } else {
504 seek
505 }
506 } else {
507 None
508 };
509
510 // Issue #693: When WRITETIME(col) or TTL(col) is in the SELECT, use the
511 // metadata-carrying scan so per-cell timestamps reach the QueryRow.
512 let results = if context.projection_flags.include_cell_metadata {
513 // Issue #962: route a fully-constrained `WHERE pk = ?` WRITETIME/TTL
514 // projection through a partition-targeted metadata lookup that prunes
515 // SSTables (bloom/BTI) before decoding, instead of full-scanning every
516 // SSTable for the table. Reuses the SAME `classify_partition_lookup`
517 // decision the non-metadata path uses (the shared resolved
518 // partition-lookup representation). The per-row predicate evaluation
519 // below is unchanged, so the pk equality itself is still applied as a
520 // correctness backstop and any bloom/BTI over-inclusion is filtered out.
521 // Issue #1918: the single forcing gate wraps the classifier outcome.
522 let outcome = classify_partition_lookup(predicates, schema_opt);
523 let scan_results = match apply_forcing(outcome, mode)? {
524 // Forced `full`: run the same full metadata scan the organic
525 // fallback uses, recorded with the distinct forced reason.
526 ForcedPlan::ForceFullScan => {
527 let path = AccessPath::FallbackFullScan {
528 reason: FallbackReason::ForcedFullScan,
529 };
530 context.access_path = Some(path.clone());
531 crate::query::access_path::record(path);
532 self.storage
533 .scan_with_cell_metadata(table, None, None, None, schema_opt)
534 .await?
535 }
536 ForcedPlan::Proceed(PartitionLookupOutcome::Targeted(pk_bytes)) => {
537 tracing::debug!(
538 "SSTableScan(metadata): partition-key point lookup (key len={}) for \"{}\"",
539 pk_bytes.len(),
540 table
541 );
542 // Epic #951 (honest paths): the `tombstones` build's metadata
543 // lookup is a full metadata scan + retain with NO prune,
544 // reported via `engaged == false`; claim
545 // `MetadataPartitionLookup` only when it really pruned, else
546 // report the honest `TombstonesBuildNoPrune` fallback (the
547 // rows are byte-identical either way).
548 let (rows, engaged) = self
549 .storage
550 .scan_partition_with_cell_metadata(table, &pk_bytes, schema_opt)
551 .await?;
552 // Issue #1918: under `point` a post-call no-prune (`engaged ==
553 // false`, tombstones build) is a non-targeted execution → fail
554 // closed rather than silently full-scanning.
555 point_requires_engaged(mode, engaged, FallbackReason::TombstonesBuildNoPrune)?;
556 let path = honest_targeted_path(AccessPath::MetadataPartitionLookup, engaged);
557 context.access_path = Some(path.clone());
558 crate::query::access_path::record(path);
559 rows
560 }
561 // Issue #1916: `WHERE pk IN (...)` on the metadata path is the union
562 // of N independent partition-targeted metadata lookups, each of which
563 // prunes SSTables (bloom/BTI) before decoding — mirroring the plain
564 // `MultiTargeted` fan-out (see the non-metadata arm below). #962
565 // shipped the single-key metadata fast path and left this fan-out as a
566 // documented follow-up; this arm is that follow-up. Each per-key call
567 // goes through the SAME single/multi-generation reconciliation the
568 // single-key `Targeted` arm uses (#1741), so the merged WRITETIME/TTL
569 // values are byte-identical to the old full-scan result.
570 //
571 // Epic #951 (honest paths): on the `tombstones` build each lookup
572 // full-scans + retains with NO prune (`engaged == false`); report
573 // `MultiPartitionLookup` only when the lookups actually pruned, else
574 // the honest `TombstonesBuildNoPrune` fallback. Rows are unchanged.
575 ForcedPlan::Proceed(PartitionLookupOutcome::MultiTargeted(pk_keys)) => {
576 tracing::debug!(
577 "SSTableScan(metadata): multi-partition lookup ({} keys) for \"{}\"",
578 pk_keys.len(),
579 table
580 );
581 let mut combined = Vec::new();
582 let mut all_engaged = true;
583 for pk_bytes in &pk_keys {
584 let (rows, engaged) = self
585 .storage
586 .scan_partition_with_cell_metadata(table, pk_bytes, schema_opt)
587 .await?;
588 all_engaged &= engaged;
589 combined.extend(rows);
590 }
591 // Issue #1918: `point` fails closed when the fan-out did not prune
592 // (tombstones build) rather than silently full-scanning.
593 point_requires_engaged(
594 mode,
595 all_engaged,
596 FallbackReason::TombstonesBuildNoPrune,
597 )?;
598 let path = honest_targeted_path(AccessPath::MultiPartitionLookup, all_engaged);
599 context.access_path = Some(path.clone());
600 crate::query::access_path::record(path);
601 // Order the union with the IDENTICAL token-then-raw-bytes rule the
602 // plain `MultiTargeted` arm uses (`sort_rows_by_token`), so the
603 // metadata IN result equals a full scan filtered to these keys.
604 sort_metadata_rows_by_token(&mut combined);
605 combined
606 }
607 // A genuinely unclassifiable metadata projection (no usable
608 // restriction) still full-scans; report that honestly
609 // (MetadataScanPath) rather than faking a targeted path.
610 // Issue #1918: under `point` a classification `Fallback` already
611 // errored up front in `apply_forcing`, so this arm is reached only
612 // under `auto`/`full`; `point_forbids_fallback` keeps the fail-closed
613 // contract explicit here.
614 ForcedPlan::Proceed(PartitionLookupOutcome::Fallback(_)) => {
615 point_forbids_fallback(mode, FallbackReason::MetadataScanPath)?;
616 let metadata_path = AccessPath::FallbackFullScan {
617 reason: FallbackReason::MetadataScanPath,
618 };
619 context.access_path = Some(metadata_path.clone());
620 crate::query::access_path::record(metadata_path);
621 self.storage
622 .scan_with_cell_metadata(table, None, None, None, schema_opt)
623 .await?
624 }
625 };
626
627 tracing::debug!("Scan (with metadata) returned {} rows", scan_results.len());
628
629 // Issue #1577 (D1 + roborev metric-accounting fix): this metadata scan
630 // is ALREADY fully materialized — the storage layer decoded every row
631 // before returning it. `collect_capped_materialized` charges
632 // `context.scan_rows` (→ `QUERY_ROWS_SCANNED`) with the TRUE decoded
633 // count up front, so the metric reflects real scan work, while the
634 // LIMIT/OFFSET `scan_cap` only bounds the per-row build/predicate work.
635 // Counting ACCEPTED rows toward the cap never drops a matching row.
636 // Issue #1817: hoist the partition-key decode across the rows of a
637 // partition (the scan yields a partition's rows consecutively).
638 let mut pk_cache = super::PartitionKeyCache::default();
639 collect_capped_materialized(
640 scan_results,
641 scan_cap,
642 predicates,
643 context,
644 |(key, value, cell_meta)| {
645 let mut row = build_row_from_scan_cached(
646 key,
647 value,
648 projection,
649 schema_opt,
650 &mut pk_cache,
651 )?;
652 // Attach per-cell metadata so evaluate_writetime_ttl can read it.
653 if !cell_meta.is_empty() {
654 row.set_cell_metadata(cell_meta);
655 }
656 Some(row)
657 },
658 )?
659 } else if let Some(seek) = schemaless_seek {
660 // Issue #1750 (regression fix, re-scoped): a SCHEMA-LESS `WHERE pk =
661 // <literal>` point read whose equality column is a metadata-CONFIRMED
662 // sole partition key. With no schema the full-scan path CANNOT
663 // reconstruct the partition-key column, so the shared per-row
664 // `evaluate_predicates` backstop below would reject EVERY row on the pk
665 // equality and return 0 rows — the regression from rerouting this read
666 // off the legacy `QueryExecutor` (which looked up by key bytes, never
667 // re-evaluating the predicate). Serve it here by the SAME
668 // key-byte-targeted seek the schema-aware Targeted arm uses
669 // (`scan_partition`), which is self-verifying: it returns ONLY rows whose
670 // raw partition key equals `seek.bytes`, so a wrong/absent key yields
671 // nothing (never another partition's rows). The key is encoded through
672 // the TYPED single-component codec for the pk's AUTHORITATIVE type (from
673 // the Statistics.db SerializationHeader `keyType`), so a parsed integer
674 // literal (`Value::BigInt`) builds the width Cassandra wrote — an `int`
675 // pk's 4-byte key, not an 8-byte one (roborev 3784 FINDING 2). The
676 // decision is made from AUTHORITATIVE metadata by ELIMINATION — never a
677 // pk-name/substring/token-count text guess (#28). A `WHERE <regular_col>
678 // = <literal>` does NOT reach here (the classifier returns `None` because
679 // the column is one of the authoritative non-key names), so it keeps the
680 // honest full-scan path and correctly matches the regular-column cell.
681 //
682 // Post-seek guard (roborev 3784 FINDING 1): the classifier admits the
683 // predicate column by ELIMINATION, which also admits a
684 // nonexistent/misspelled column whose literal happens to encode to a real
685 // partition's key. `finalize_schemaless_seek_row` reconstructs the sole
686 // pk column under its authoritative type/name and RE-EVALUATES the
687 // predicate, so such a column yields `Unknown` and rejects the row
688 // (correct 0 rows) — the pk-name predicate still matches. The schema-less
689 // `SELECT *` column metadata still comes from the first row in `execute`.
690 let (rows, engaged) = self
691 .storage
692 .scan_partition(table, &seek.bytes, None)
693 .await?;
694 // Issue #1918: `point` fails closed on a post-call no-prune. (`full`
695 // never reaches here — the seek is skipped above under forced `full`.)
696 point_requires_engaged(mode, engaged, FallbackReason::TombstonesBuildNoPrune)?;
697 let path = honest_targeted_path(AccessPath::PartitionLookup, engaged);
698 context.access_path = Some(path.clone());
699 crate::query::access_path::record(path);
700 context.scan_rows += rows.len() as u64;
701 context.rows_processed += rows.len() as u64;
702 let mut out = Vec::with_capacity(rows.len());
703 for (key, value) in rows {
704 if let Some(row) = super::schemaless_point::finalize_schemaless_seek_row(
705 key, value, projection, predicates, &seek,
706 ) {
707 out.push(row);
708 }
709 }
710 out
711 } else {
712 // Issue #949: a fully-constrained `WHERE pk = ?` is served by a
713 // partition-targeted lookup that prunes SSTables via bloom/BTI and only
714 // parses the candidates, instead of scanning every SSTable for the
715 // table. Falls back to a full scan when the partition key isn't fully
716 // pinned or can't be encoded. The per-row predicate evaluation below is
717 // unchanged, so clustering predicates and the pk equality itself are
718 // still applied (and any over-inclusion is filtered out).
719 // Issue #1918: the single forcing gate wraps the classifier outcome.
720 let outcome = classify_partition_lookup(predicates, schema_opt);
721 let scan_results = match apply_forcing(outcome, mode)? {
722 // Forced `full`: run the SAME full-scan + reconciliation code the
723 // organic fallback uses (so rows/order match `auto`), recorded with
724 // the distinct `ForcedFullScan` reason so it is never mistaken for
725 // an organic fallback.
726 ForcedPlan::ForceFullScan => {
727 let path = AccessPath::FallbackFullScan {
728 reason: FallbackReason::ForcedFullScan,
729 };
730 context.access_path = Some(path.clone());
731 crate::query::access_path::record(path);
732 if let Some(cap) = scan_cap {
733 return self
734 .capped_fallback_scan(
735 table, predicates, projection, schema_opt, cap, context,
736 )
737 .await;
738 }
739 self.storage
740 .scan(table, None, None, None, schema_opt)
741 .await?
742 }
743 ForcedPlan::Proceed(PartitionLookupOutcome::Targeted(pk_bytes)) => {
744 tracing::debug!(
745 "SSTableScan: partition-key point lookup (key len={}) for \"{}\"",
746 pk_bytes.len(),
747 table
748 );
749 // Issue #954: when a single-column clustering restriction is
750 // present, push it down to a within-partition seek so a wide
751 // partition's slice decodes O(matched rows + index), not the
752 // whole partition. The seek reports whether the clustering
753 // narrowing actually engaged; the per-row backstop below applies
754 // the exact bound so output is byte-identical either way.
755 //
756 // Issue #960: report the HONEST access path — `ClusteringSlice`
757 // only when the seek engaged, else `PartitionLookup`. The
758 // clustering seek exists only on the default build; the
759 // `tombstones` build uses the plain partition lookup.
760 // Issue #954/#960/#1184: forward clustering-slice seek OR (for
761 // `ORDER BY <ck>` reverse-of-stored) the BIG reverse iterator,
762 // with the honest access path recorded inside the helper.
763 #[cfg(not(feature = "tombstones"))]
764 {
765 self.targeted_partition_rows(
766 table, &pk_bytes, predicates, order_by, schema_opt, context,
767 )
768 .await?
769 }
770 #[cfg(feature = "tombstones")]
771 {
772 // Epic #951 (honest paths): the `tombstones` build's
773 // `scan_partition` is a full scan + retain with NO prune,
774 // reported via `engaged == false`. Report the honest
775 // fallback rather than a fake `PartitionLookup`; the rows
776 // are byte-identical to the pruned build.
777 let (rows, engaged) = self
778 .storage
779 .scan_partition(table, &pk_bytes, schema_opt)
780 .await?;
781 // Issue #1918: `point` fails closed on the tombstones-build
782 // no-prune rather than silently full-scanning.
783 point_requires_engaged(
784 mode,
785 engaged,
786 FallbackReason::TombstonesBuildNoPrune,
787 )?;
788 let path = honest_targeted_path(AccessPath::PartitionLookup, engaged);
789 context.access_path = Some(path.clone());
790 crate::query::access_path::record(path);
791 rows
792 }
793 }
794 ForcedPlan::Proceed(PartitionLookupOutcome::MultiTargeted(pk_keys)) => {
795 tracing::debug!(
796 "SSTableScan: multi-partition lookup ({} keys) for \"{}\"",
797 pk_keys.len(),
798 table
799 );
800 // Issue #955/#960: `WHERE pk IN (...)` over the complete key
801 // is the union of N independent partition-targeted lookups,
802 // each of which prunes SSTables. Epic #951 (honest paths): on
803 // the `tombstones` build each lookup full-scans + retains with
804 // NO prune (`engaged == false`); report `MultiPartitionLookup`
805 // only when the lookups actually pruned, else the honest
806 // `TombstonesBuildNoPrune` fallback. Rows are unchanged.
807 let mut combined = Vec::new();
808 let mut all_engaged = true;
809 for pk_bytes in &pk_keys {
810 let (rows, engaged) = self
811 .storage
812 .scan_partition(table, pk_bytes, schema_opt)
813 .await?;
814 all_engaged &= engaged;
815 combined.extend(rows);
816 }
817 // Issue #1918: `point` fails closed when the fan-out did not
818 // prune (tombstones build) rather than silently full-scanning.
819 point_requires_engaged(
820 mode,
821 all_engaged,
822 FallbackReason::TombstonesBuildNoPrune,
823 )?;
824 let path = honest_targeted_path(AccessPath::MultiPartitionLookup, all_engaged);
825 context.access_path = Some(path.clone());
826 crate::query::access_path::record(path);
827 // Order the union to equal a full scan filtered to these keys:
828 // partitions are stored token-ordered, so sort the combined
829 // rows by (partition token, raw key bytes). A *stable* sort
830 // keeps each partition's clustering order (rows for one key
831 // arrive contiguously from one `scan_partition`) intact.
832 sort_rows_by_token(&mut combined);
833 combined
834 }
835 ForcedPlan::Proceed(PartitionLookupOutcome::Fallback(reason)) => {
836 // Issue #960: report the honest reason a full scan was chosen.
837 // (Under `point` a `Fallback` already errored in `apply_forcing`,
838 // so this arm is reached only under `auto`.)
839 context.access_path = Some(AccessPath::FallbackFullScan { reason });
840 crate::query::access_path::record(AccessPath::FallbackFullScan { reason });
841 // Issue #1577 (D1): when the plan is LIMIT-pushdown safe, stop
842 // DECODING the table once `cap` rows are accepted. This streams
843 // (in lockstep with `scan`) and drops the stream at the cap,
844 // returning the final rows directly — the shared per-row loop
845 // below is bypassed for this branch.
846 if let Some(cap) = scan_cap {
847 return self
848 .capped_fallback_scan(
849 table, predicates, projection, schema_opt, cap, context,
850 )
851 .await;
852 }
853 // Issue #1582 (D6, narrow subset): unbounded full scan; the sole
854 // budget check is applied once on the FINAL result in `execute`.
855 self.storage
856 .scan(table, None, None, None, schema_opt)
857 .await?
858 }
859 };
860
861 tracing::debug!("Scan returned {} rows", scan_results.len());
862
863 // Issue #1577 (D1 + roborev metric-accounting fix): these results are
864 // ALREADY materialized — the partition-targeted paths decoded their
865 // (partition-bounded) rows, and the TRULY decode-bounded full-scan
866 // fallback already returned early above via `capped_fallback_scan`.
867 // `collect_capped_materialized` charges `context.scan_rows` with the
868 // full decoded count up front (so `QUERY_ROWS_SCANNED` reflects the
869 // real scan), while the `scan_cap` only bounds per-row build work.
870 // Counting ACCEPTED rows toward the cap (build_row_from_scan returns
871 // None for tombstoned/null rows, Issue #191) never drops a match.
872 // Issue #1817: hoist the partition-key decode across a partition's rows.
873 let mut pk_cache = super::PartitionKeyCache::default();
874 collect_capped_materialized(
875 scan_results,
876 scan_cap,
877 predicates,
878 context,
879 |(key, value)| {
880 build_row_from_scan_cached(key, value, projection, schema_opt, &mut pk_cache)
881 },
882 )?
883 };
884
885 Ok(results)
886 }
887}