surrealdb-core 3.2.0

use std::sync::Arc;

use futures::StreamExt;
use tracing::instrument;

use super::pipeline::{
	build_field_state, determine_scan_direction, eval_limit_expr, kv_scan_stream,
};
use super::{FullTextScan, IndexScan, KnnScan};
use crate::catalog::{DatabaseId, NamespaceId, Permission};
use crate::err::Error;
use crate::exec::index::access_path::{AccessPath, select_access_path};
use crate::exec::index::analysis::IndexAnalyzer;
use crate::exec::operators::scan::pipeline::ScanPipeline;
use crate::exec::permission::{
	PhysicalPermission, convert_permission_to_physical_runtime, should_check_perms,
	validate_record_user_access,
};
use crate::exec::planner::util::{
	SELECT_ITERATION_PARAMS, fold_condition_expressions, index_covers_ordering,
	resolve_condition_params, resolve_projection_field_idioms, strip_knn_from_condition,
};
use crate::exec::pre_decode_filter::{PreDecodeFilterStatus, pre_decode_filter_for_execute};
use crate::exec::{
	AccessMode, ContextLevel, EvalContext, ExecOperator, ExecutionContext, FlowResult,
	OperatorMetrics, PhysicalExpr, ValueBatch, ValueBatchStream, monitor_stream,
};
use crate::expr::order::Ordering;
use crate::expr::with::With;
use crate::expr::{Cond, ControlFlow, ControlFlowExt};
use crate::iam::Action;
use crate::idx::planner::ScanDirection;
use crate::key::record;
use crate::val::{TableName, Value};

/// Full table scan - iterates over all records in a table.
///
/// Requires database-level context since it reads from a specific table
/// in the selected namespace and database.
///
/// Permission checking is performed at execution time by resolving the table
/// definition from the current transaction's schema view and filtering records
/// based on the SELECT permission.
///
/// When scanning a table, this operator can perform index selection based on
/// the provided WHERE condition, ORDER BY clause, and WITH hints.
///
/// The optional `predicate`, `limit`, and `start` fields allow the planner to
/// push the Filter, Limit, and Start operators down into the scan, reducing
/// pipeline overhead and enabling early termination for `WHERE ... LIMIT`
/// queries.
#[derive(Debug, Clone)]
pub struct DynamicScan {
	pub(crate) source: Arc<dyn PhysicalExpr>,
	/// Optional version timestamp for time-travel queries (VERSION clause)
	pub(crate) version: Option<Arc<dyn PhysicalExpr>>,
	/// Optional WHERE condition for index selection (AST form)
	pub(crate) cond: Option<Cond>,
	/// Optional ORDER BY for index selection and scan direction
	pub(crate) order: Option<Ordering>,
	/// Optional WITH INDEX/NOINDEX hints
	pub(crate) with: Option<With>,
	/// Fields needed by the query (projection + WHERE + ORDER + GROUP).
	/// `None` means all fields are needed (SELECT *).
	pub(crate) needed_fields: Option<std::collections::HashSet<String>>,
	/// Compiled WHERE predicate pushed down from the Filter operator.
	/// Applied after computed fields, before field-level permissions.
	pub(crate) predicate: Option<Arc<dyn PhysicalExpr>>,
	/// LIMIT expression pushed down from the Limit operator.
	/// Maximum number of rows to return after filtering.
	pub(crate) limit: Option<Arc<dyn PhysicalExpr>>,
	/// START offset expression pushed down from the Limit operator.
	/// Number of rows to skip (after filtering) before emitting.
	pub(crate) start: Option<Arc<dyn PhysicalExpr>>,
	/// Per-operator runtime metrics for EXPLAIN ANALYZE.
	pub(crate) metrics: Arc<OperatorMetrics>,
	/// KNN distance context, shared with IndexFunctionExec for vector::distance::knn().
	/// Populated by KnnScan during execution.
	pub(crate) knn_context: Option<Arc<crate::exec::function::KnnContext>>,
	/// Predicate pre-decode filter status (plan-time); see [`PreDecodeFilterStatus`].
	pub(crate) pre_decode_filter_status: PreDecodeFilterStatus,
}

impl DynamicScan {
	/// Create a new Scan operator with fresh metrics.
	#[allow(clippy::too_many_arguments)]
	pub(crate) fn new(
		source: Arc<dyn PhysicalExpr>,
		version: Option<Arc<dyn PhysicalExpr>>,
		cond: Option<Cond>,
		order: Option<Ordering>,
		with: Option<With>,
		needed_fields: Option<std::collections::HashSet<String>>,
		predicate: Option<Arc<dyn PhysicalExpr>>,
		limit: Option<Arc<dyn PhysicalExpr>>,
		start: Option<Arc<dyn PhysicalExpr>>,
	) -> Self {
		Self {
			source,
			version,
			cond,
			order,
			with,
			needed_fields,
			predicate,
			limit,
			start,
			metrics: Arc::new(OperatorMetrics::new()),
			knn_context: None,
			pre_decode_filter_status: PreDecodeFilterStatus::NotApplicable,
		}
	}

	/// Set plan-time pre-decode filter status for EXPLAIN and execution.
	pub(crate) fn with_pre_decode_filter(mut self, status: PreDecodeFilterStatus) -> Self {
		self.pre_decode_filter_status = status;
		self
	}

	/// Set the KNN context for distance propagation.
	pub(crate) fn with_knn_context(
		mut self,
		knn_context: Option<Arc<crate::exec::function::KnnContext>>,
	) -> Self {
		self.knn_context = knn_context;
		self
	}
}
impl ExecOperator for DynamicScan {
	fn name(&self) -> &'static str {
		"DynamicScan"
	}

	fn attrs(&self) -> Vec<(String, String)> {
		let mut attrs = vec![("source".to_string(), self.source.to_sql())];
		if let Some(ref pred) = self.predicate {
			attrs.push(("predicate".to_string(), pred.to_sql()));
		}
		if let Some(ref limit) = self.limit {
			attrs.push(("limit".to_string(), limit.to_sql()));
		}
		if let Some(ref start) = self.start {
			attrs.push(("offset".to_string(), start.to_sql()));
		}
		if let Some(s) = self.pre_decode_filter_status.explain_text() {
			attrs.push(("pre_decode_filter".to_string(), s.to_string()));
		}
		attrs
	}

	fn required_context(&self) -> ContextLevel {
		// Scan needs database context for table access, combined with expression contexts
		let exprs_ctx = [
			Some(&self.source),
			self.version.as_ref(),
			self.predicate.as_ref(),
			self.limit.as_ref(),
			self.start.as_ref(),
		]
		.into_iter()
		.flatten()
		.map(|e| e.required_context())
		.max()
		.unwrap_or(ContextLevel::Root);
		exprs_ctx.max(ContextLevel::Database)
	}

	fn metrics(&self) -> Option<&OperatorMetrics> {
		Some(&self.metrics)
	}

	fn expressions(&self) -> Vec<(&str, &Arc<dyn PhysicalExpr>)> {
		let mut exprs = vec![("source", &self.source)];
		if let Some(ref version) = self.version {
			exprs.push(("version", version));
		}
		if let Some(ref pred) = self.predicate {
			exprs.push(("predicate", pred));
		}
		if let Some(ref limit) = self.limit {
			exprs.push(("limit", limit));
		}
		if let Some(ref start) = self.start {
			exprs.push(("start", start));
		}
		exprs
	}

	fn access_mode(&self) -> AccessMode {
		// Scan is read-only, but expressions could contain subqueries with mutations.
		let mut mode = self.source.access_mode();
		if let Some(ref version) = self.version {
			mode = mode.combine(version.access_mode());
		}
		if let Some(ref pred) = self.predicate {
			mode = mode.combine(pred.access_mode());
		}
		if let Some(ref limit) = self.limit {
			mode = mode.combine(limit.access_mode());
		}
		if let Some(ref start) = self.start {
			mode = mode.combine(start.access_mode());
		}
		mode
	}

	#[instrument(name = "Scan::execute", level = "trace", skip_all)]
	fn execute(&self, ctx: &ExecutionContext) -> FlowResult<ValueBatchStream> {
		// Get database context - we declared Database level, so this should succeed
		let db_ctx = ctx.database()?.clone();

		// Validate record user has access to this namespace/database
		validate_record_user_access(&db_ctx)?;

		// Check if we need to enforce permissions
		let check_perms = should_check_perms(&db_ctx, Action::View)?;

		// Clone for the async block
		let source_expr = Arc::clone(&self.source);
		let version = self.version.clone();
		let cond = self.cond.clone();
		let order = self.order.clone();
		let with = self.with.clone();
		let needed_fields = self.needed_fields.clone();
		let predicate = self.predicate.clone();
		let limit_expr = self.limit.clone();
		let start_expr = self.start.clone();
		let knn_context = self.knn_context.clone();
		let pre_decode_filter_status = self.pre_decode_filter_status.clone();
		let ctx = ctx.clone();

		let stream = async_stream::try_stream! {
			let db_ctx = ctx.database().context("Scan requires database context")?;
			let ns = Arc::clone(&db_ctx.ns_ctx.ns);
			let db = Arc::clone(&db_ctx.db);

			// Evaluate table expression
			let eval_ctx = EvalContext::from_exec_ctx(&ctx);
			let table_value = source_expr.evaluate(eval_ctx).await?;

			// Determine scan target: either a table name or a record ID
			let table_name = match table_value {
				Value::Table(t) => t,
			Value::RecordId(rid) => {
				// === RECORD LOOKUP (point or range) ===
				// Delegate to the shared execute_record_lookup helper which
				// handles both point lookups and range scans. For plan-time-
				// known RecordIds the planner emits RecordLookup directly;
				// this path handles runtime-discovered RecordIds (e.g. from
				// `type::thing(...)` or other dynamic expressions).
				//
				// The planner marks predicate/limit/start as consumed for
				// FunctionCall/Postfix sources, so the outer Filter/Limit
				// operators are removed. We must forward the pushdowns here
				// to ensure WHERE/LIMIT/START are still applied.

				// Evaluate VERSION expression
				let version: Option<u64> = match &version {
					Some(expr) => {
						let eval_ctx = EvalContext::from_exec_ctx(&ctx);
						let v = expr.evaluate(eval_ctx).await?;
						Some(
							v.cast_to::<crate::val::Datetime>()
								.map_err(|e| anyhow::anyhow!("{e}"))?
								.to_version_stamp(ctx.txn().timestamp_impl().as_ref())?,
						)
					}
					None => ctx.version_stamp(),
				};

				// Evaluate pushed-down LIMIT and START expressions
				let limit_val: Option<usize> = match &limit_expr {
					Some(expr) => Some(eval_limit_expr(&**expr, &ctx).await?),
					None => None,
				};
				let start_val: usize = match &start_expr {
					Some(expr) => eval_limit_expr(&**expr, &ctx).await?,
					None => 0,
				};

				// Early exit if limit is 0
				if limit_val == Some(0) {
					return;
				}

				let results = super::record_id::execute_record_lookup(
					&rid, version, check_perms, needed_fields.as_ref(), &ctx,
					predicate.as_ref(), limit_val, start_val, None,
					&pre_decode_filter_status,
				).await?;

				if !results.is_empty() {
					yield ValueBatch { values: results };
				}
				return;
			}
			Value::Array(arr) => {
				// === ARRAY SOURCE ===
				// The planner marks predicate/limit/start as consumed for
				// FunctionCall/Postfix sources, so the outer Filter/Limit
				// operators are removed. We must apply them here.

				// Evaluate pushed-down LIMIT and START expressions
				let limit_val: Option<usize> = match &limit_expr {
					Some(expr) => Some(eval_limit_expr(&**expr, &ctx).await?),
					None => None,
				};
				let start_val: usize = match &start_expr {
					Some(expr) => eval_limit_expr(&**expr, &ctx).await?,
					None => 0,
				};

				// Early exit if limit is 0
				if limit_val == Some(0) {
					return;
				}

				// Apply pushed-down predicate
				let mut values = arr.0;
				if let Some(ref pred) = predicate {
					let mut write_idx = 0;
					for read_idx in 0..values.len() {
						let eval_ctx = EvalContext::from_exec_ctx(&ctx).with_value(&values[read_idx]);
						if pred.evaluate(eval_ctx).await?.is_truthy() {
							if write_idx != read_idx {
								values.swap(write_idx, read_idx);
							}
							write_idx += 1;
						}
					}
					values.truncate(write_idx);
				}

				// Apply start offset
				if start_val > 0 {
					if start_val >= values.len() {
						return;
					}
					values.drain(..start_val);
				}

				// Apply limit
				if let Some(limit) = limit_val {
					values.truncate(limit);
				}

				if !values.is_empty() {
					yield ValueBatch { values };
				}
				return;
			}
			// For any other value type, yield as a single row.
			// This matches legacy FROM behavior for non-table values.
			other => {
				// === SCALAR SOURCE ===
				// Same pushdown logic as the array branch above.

				// Evaluate pushed-down LIMIT and START expressions
				let limit_val: Option<usize> = match &limit_expr {
					Some(expr) => Some(eval_limit_expr(&**expr, &ctx).await?),
					None => None,
				};
				let start_val: usize = match &start_expr {
					Some(expr) => eval_limit_expr(&**expr, &ctx).await?,
					None => 0,
				};

				// Early exit if limit is 0 or start skips past the single value
				if limit_val == Some(0) || start_val > 0 {
					return;
				}

				// Apply pushed-down predicate
				if let Some(ref pred) = predicate {
					let eval_ctx = EvalContext::from_exec_ctx(&ctx).with_value(&other);
					if !pred.evaluate(eval_ctx).await?.is_truthy() {
						return;
					}
				}

				yield ValueBatch { values: vec![other] };
				return;
			}
			};

			// === TABLE SCAN PATH ===
			// Everything below is for table-based scans only.

			// Evaluate pushed-down LIMIT and START expressions
			let limit_val: Option<usize> = match &limit_expr {
				Some(expr) => Some(eval_limit_expr(&**expr, &ctx).await?),
				None => None,
			};
			let start_val: usize = match &start_expr {
				Some(expr) => eval_limit_expr(&**expr, &ctx).await?,
				None => 0,
			};

			// VERSION stamp for metadata lookups (same evaluation as `resolve_table_scan_stream`).
			let version_stamp: Option<u64> = match &version {
				Some(expr) => {
					let eval_ctx = EvalContext::from_exec_ctx(&ctx);
					let v = expr.evaluate(eval_ctx).await?;
					Some(
						v.cast_to::<crate::val::Datetime>()
							.map_err(|e| anyhow::anyhow!("{e}"))?
							.to_version_stamp(ctx.txn().timestamp_impl().as_ref())?,
					)
				}
				None => ctx.version_stamp(),
			};

			// Check table existence and resolve SELECT permission
			let table_def = db_ctx
				.get_table_def(&table_name, version_stamp)
				.await
				.context("Failed to get table")?;

			if table_def.is_none() {
				Err(ControlFlow::Err(anyhow::Error::new(Error::TbNotFound {
					name: table_name.clone(),
				})))?;
			}

			let select_permission = if check_perms {
				let catalog_perm = match &table_def {
					Some(def) => def.permissions.select.clone(),
					None => Permission::None,
				};
				convert_permission_to_physical_runtime(&catalog_perm, ctx.ctx())
					.await
					.context("Failed to convert permission")?
			} else {
				PhysicalPermission::Allow
			};

			// Early exit if denied
			if matches!(select_permission, PhysicalPermission::Deny) {
				return;
			}

			if limit_val == Some(0) {
				return;
			}

			// Eagerly initialize field state (computed fields + field permissions)
			let field_state = build_field_state(&ctx, &table_name, check_perms, needed_fields.as_ref()).await?;

			// SECURITY (value-ordering oracle): runtime counterpart to the
			// plan-time guard in `planner/select`. The plan-time guard only
			// runs when the access path is resolved at plan time; this
			// `DynamicScan` is also reached when planning was txn-less or when
			// plan-time access-path resolution returned `None`/`Err`, and it
			// resolves the access path below using `order`. If `order` targets
			// a field the current actor cannot fully read, an index that walks
			// it in true value order would emit rows in that order; field
			// reduction then nulls the value, but the row order would still
			// encode the hidden values' relative ordering. Withholding `order`
			// here keeps the scan in record-id order and forces the outer Sort
			// to run over the reduced (NULL) keys. Unlike the plan-time path,
			// this uses the actor's *resolved* field permissions
			// (`field_state.field_permissions`, populated only when permissions
			// are enforced), so it never over-applies to privileged users.
			let order = if order_touches_restricted_select_field(
				order.as_ref(),
				&field_state.field_permissions,
			) {
				None
			} else {
				order
			};

			// Row-filtering (permissions, WHERE) prevents positional pushdown;
			// row-modifying ops (computed fields, field perms) do not.
			let needs_row_filtering = ScanPipeline::compute_needs_row_filtering(
				&select_permission, predicate.as_ref(),
			);

			let pre_skip = if !needs_row_filtering { start_val } else { 0 };
			let effective_storage_limit = if !needs_row_filtering { limit_val } else { None };

			let direction = determine_scan_direction(order.as_ref());

			// Create the source stream based on scan type.
			// `applied_pre_skip` tracks how many rows the source will skip
			// before decoding, so the pipeline can adjust its start accordingly.
			let pre_decode_filter = pre_decode_filter_for_execute(
				&pre_decode_filter_status,
				&field_state,
				check_perms,
				ctx.ctx().config.idiom_recursion_limit,
			);

			let (mut source, applied_pre_skip) = {
				// Table scan (with runtime index selection)
				resolve_table_scan_stream(
					&ctx, TableScanConfig {
						ns_id: ns.namespace_id,
						db_id: db.database_id,
						table_name,
						cond,
						order: order.clone(),
						with,
						direction,
						version,
						storage_limit: effective_storage_limit,
						pre_skip,
						has_pushed_limit: effective_storage_limit.is_some(),
						limit_hint: limit_val.map(|l| (l + start_val).min(u32::MAX as usize) as u32),
						knn_context: knn_context.clone(),
						pre_decode_filter,
					},
				).await?
			};

			// Build the pipeline with start adjusted for any pre-skipping.
			let mut pipeline = ScanPipeline::new(
				select_permission, predicate, field_state,
				check_perms, limit_val, start_val.saturating_sub(applied_pre_skip),
			);

			// Unified consumption loop for all stream-based sources.
			while let Some(batch_result) = source.next().await {
				// Check for cancellation between batches
				if ctx.cancellation().is_cancelled() {
					Err(ControlFlow::Err(
						anyhow::anyhow!(crate::err::Error::QueryCancelled),
					))?;
				}
				let mut batch = batch_result?;
				let cont = pipeline.process_batch(&mut batch.values, &ctx).await?;
				if !batch.values.is_empty() {
					yield ValueBatch { values: batch.values };
				}
				if !cont {
					break;
				}
			}
		};

		Ok(monitor_stream(Box::pin(stream), "Scan", &self.metrics))
	}
}

// ---------------------------------------------------------------------------
// Source helpers
// ---------------------------------------------------------------------------

/// SECURITY (value-ordering oracle): returns `true` when any top-level
/// `ORDER BY` idiom references a field carrying a non-`Full` SELECT permission
/// for the current actor.
///
/// `field_permissions` are the per-field SELECT permissions resolved into
/// [`crate::exec::operators::scan::pipeline::FieldState`]; `Permission::Full`
/// fields are intentionally absent and the list is empty when permission
/// enforcement is skipped (owner/root), so a privileged actor never matches
/// here. The match is shallow — each `Order.value` is a plain top-level idiom
/// and an index-ordering leak requires the index to cover the field directly,
/// so a restricted field referenced only inside an idiom filter (e.g.
/// `ORDER BY foo[WHERE code = …]`) is not an indexable ordering and cannot leak
/// through this vector. Mirrors the plan-time `RestrictedPrefixes::order_touches`
/// guard in `planner/select`, including its out-of-scope parent/child
/// nested-field gap (ordering by a *parent* of a restricted child is not
/// caught — see that guard's docs).
fn order_touches_restricted_select_field(
	order: Option<&Ordering>,
	field_permissions: &[(crate::expr::Idiom, PhysicalPermission)],
) -> bool {
	// `ORDER BY RAND()` references no field and cannot leak ordering.
	let Some(Ordering::Order(order_list)) = order else {
		return false;
	};
	if field_permissions.is_empty() {
		return false;
	}
	order_list
		.iter()
		.any(|o| field_permissions.iter().any(|(field, _)| o.value.starts_with(field.0.as_slice())))
}

/// Configuration bundle for [`resolve_table_scan_stream`].
struct TableScanConfig {
	ns_id: NamespaceId,
	db_id: DatabaseId,
	table_name: TableName,
	cond: Option<Cond>,
	order: Option<Ordering>,
	with: Option<With>,
	direction: ScanDirection,
	/// VERSION expression for time-travel queries (evaluated inside
	/// [`resolve_table_scan_stream`] and passed to child operators).
	version: Option<Arc<dyn PhysicalExpr>>,
	storage_limit: Option<usize>,
	/// Number of KV pairs to skip before decoding (fast-path only).
	pre_skip: usize,
	/// Whether the LIMIT was actually pushed into the storage-level scan.
	/// When true the scan truncates rows, so we must verify that the
	/// runtime-selected BTree index covers the requested ordering.
	/// If it doesn't, we fall back to a KV scan for correctness.
	has_pushed_limit: bool,
	/// Hint for the scanner's initial batch size, typically `limit + start`.
	/// Caps the first fetch to avoid over-reading for small-limit queries.
	limit_hint: Option<u32>,
	/// KNN distance context for vector::distance::knn() support.
	knn_context: Option<Arc<crate::exec::function::KnnContext>>,
	/// Optional structural WHERE pre-decode filter for raw KV table scans.
	pre_decode_filter: Option<Arc<crate::exec::pre_decode_filter::PreDecodeFilter>>,
}

/// Resolve the optimal access path for a table scan and return the source
/// stream together with the number of rows that were pre-skipped at the
/// KV layer (zero for index / full-text sources).
async fn resolve_table_scan_stream(
	ctx: &ExecutionContext,
	cfg: TableScanConfig,
) -> Result<(ValueBatchStream, usize), ControlFlow> {
	let txn = ctx.txn();

	// Evaluate VERSION expression once for the fallback KV scan path.
	// Child operators receive the unevaluated expr and evaluate it themselves.
	let version_stamp: Option<u64> = match &cfg.version {
		Some(expr) => {
			let eval_ctx = EvalContext::from_exec_ctx(ctx);
			let v = expr.evaluate(eval_ctx).await?;
			Some(
				v.cast_to::<crate::val::Datetime>()
					.map_err(|e| anyhow::anyhow!("{e}"))?
					.to_version_stamp(txn.timestamp_impl().as_ref())?,
			)
		}
		None => ctx.version_stamp(),
	};

	// Resolve bind-parameter references so that index analysis sees
	// Expr::Literal instead of Expr::Param. Covers LET bindings, client
	// bind params, and DEFINE PARAM (via txn store).
	//
	// After param resolution, re-fold constant expressions (pure functions
	// whose arguments are now all literals) and rewrite type::field("name")
	// to Expr::Idiom so the index analyzer can match them.
	let resolved_cond = match cfg.cond.as_ref() {
		Some(c) => {
			let ns_db = Some((cfg.ns_id, cfg.db_id));
			let mut cond =
				resolve_condition_params(c, &ctx.root().ctx, ns_db, SELECT_ITERATION_PARAMS).await;
			fold_condition_expressions(&mut cond, ctx.function_registry());
			resolve_projection_field_idioms(&mut cond, ctx.function_registry());
			Some(cond)
		}
		None => None,
	};

	// If the WHERE folded to `false` (e.g. `field IN []` short-circuited by
	// `fold_condition_expressions`) the SELECT can produce no rows. Skip
	// index analysis and the KV scan entirely. Mirrors the static planner's
	// check at `planner/select/mod.rs` so dynamic FROM sources get the same
	// zero-I/O treatment as static `FROM table` sources.
	if let Some(c) = resolved_cond.as_ref()
		&& matches!(&c.0, crate::expr::Expr::Literal(crate::expr::literal::Literal::Bool(false)))
	{
		let op = super::EmptyScan::new();
		let stream = op.execute(ctx)?;
		return Ok((stream, 0));
	}

	let access_path = if matches!(&cfg.with, Some(With::NoIndex)) {
		None
	} else {
		let db_ctx =
			ctx.database().context("DynamicScan index analysis requires database context")?;
		let indexes = db_ctx
			.get_table_indexes(&cfg.table_name, version_stamp)
			.await
			.context("Failed to fetch indexes")?;

		let analyzer = IndexAnalyzer::new(indexes, cfg.with.as_ref());
		let candidates = analyzer.analyze(resolved_cond.as_ref(), cfg.order.as_ref());
		if candidates.is_empty() {
			// No single-index candidates -- try multi-index union for OR conditions
			analyzer
				.try_or_union(resolved_cond.as_ref(), cfg.direction)
				// Try expanding IN operators into union of equality lookups
				.or_else(|| analyzer.try_in_expansion(resolved_cond.as_ref(), cfg.direction))
				// Try expanding CONTAINSALL/CONTAINSANY into union of equality lookups
				.or_else(|| {
					analyzer.try_containment_expansion(resolved_cond.as_ref(), cfg.direction)
				})
		} else {
			let path = select_access_path(candidates, cfg.with.as_ref(), cfg.direction);
			// When the best single-index path is a full-range scan (ORDER BY
			// only), prefer a multi-index union for OR conditions if available.
			if path.is_full_range_scan() {
				analyzer.try_or_union(resolved_cond.as_ref(), cfg.direction).or(Some(path))
			} else {
				Some(path)
			}
		}
	};

	match access_path {
		// B-tree index scan (single-column and compound).
		// When the LIMIT was pushed into storage, the scan truncates rows
		// and assumes id-ordered output (via order_is_scan_compatible).
		// Reject the index if it doesn't cover the requested ordering so
		// we fall through to the KV scan fallback for correctness.
		Some(AccessPath::BTreeScan {
			index_ref,
			access,
			direction,
		}) if !cfg.has_pushed_limit
			|| cfg
				.order
				.as_ref()
				.is_none_or(|o| index_covers_ordering(&index_ref, &access, direction, o)) =>
		{
			let operator = IndexScan::new(
				index_ref,
				access,
				direction,
				cfg.table_name,
				None,
				None,
				cfg.version,
				None,
				None,
			);
			let stream = operator.execute(ctx)?;
			Ok((stream, 0))
		}

		// Full-text search
		Some(AccessPath::FullTextSearch {
			index_ref,
			query,
			operator,
		}) => {
			let ft_op =
				FullTextScan::new(index_ref, query, operator, cfg.table_name, cfg.version, None);
			let stream = ft_op.execute(ctx)?;
			Ok((stream, 0))
		}

		// KNN vector search via HNSW index
		Some(AccessPath::KnnSearch {
			index_ref,
			vector,
			k,
			ef,
		}) => {
			// Strip KNN operators from the resolved condition to get the
			// residual (non-KNN predicates) for HNSW pushdown.
			let residual_cond = resolved_cond.as_ref().and_then(strip_knn_from_condition);
			let knn_op = KnnScan::new(
				index_ref,
				vector,
				k,
				ef,
				cfg.table_name,
				cfg.version,
				cfg.knn_context.clone(),
				residual_cond,
				None,
			);
			let stream = knn_op.execute(ctx)?;
			Ok((stream, 0))
		}

		// Multi-index union for OR conditions — delegate to UnionIndexScan.
		// Permission handling is done by DynamicScan's ScanPipeline above.
		// The `dedupe` flag is informational here: this fallback path uses
		// the operator's default no-merge sequential mode, which already
		// dedupes by record id via a HashSet. The flag would matter if a
		// future `MergeMode::ByIndexKey{,Dedup}` were wired into this
		// fallback — see [`AccessPath::Union`].
		Some(AccessPath::Union {
			paths,
			dedupe: _,
		}) => {
			let mut sub_operators: Vec<Arc<dyn ExecOperator>> = Vec::with_capacity(paths.len());
			for path in paths {
				sub_operators.push(create_index_operator(&path, &cfg, resolved_cond.as_ref()));
			}
			let union_op = super::UnionIndexScan::new(cfg.table_name.clone(), sub_operators, None);
			let stream = union_op.execute(ctx)?;
			Ok((stream, 0))
		}

		// Provably empty result set — short-circuit with no storage I/O.
		Some(AccessPath::EmptyScan) => {
			let op = super::EmptyScan::new();
			let stream = op.execute(ctx)?;
			Ok((stream, 0))
		}

		// Fall back to table KV scan (NOINDEX, BTree rejected by ordering
		// check, etc.)
		_ => {
			let beg = record::prefix(cfg.ns_id, cfg.db_id, &cfg.table_name)?;
			let end = record::suffix(cfg.ns_id, cfg.db_id, &cfg.table_name)?;
			let stream = kv_scan_stream(
				txn,
				beg,
				end,
				version_stamp,
				cfg.storage_limit,
				cfg.direction,
				cfg.pre_skip,
				cfg.limit_hint,
				cfg.pre_decode_filter.clone(),
				// TopK threshold pushdown is plan-time-only (TableScan);
				// DynamicScan resolves its access path at runtime.
				None,
			);
			Ok((stream, cfg.pre_skip))
		}
	}
}

/// Create an index scan operator for a single access path.
///
/// Used by the multi-index union handler to create individual operators
/// for each branch of an OR condition. The caller (typically
/// [`UnionIndexScan`](super::UnionIndexScan)) is responsible for
/// executing the operators and deduplicating results.
fn create_index_operator(
	path: &AccessPath,
	cfg: &TableScanConfig,
	resolved_cond: Option<&Cond>,
) -> Arc<dyn ExecOperator> {
	match path {
		AccessPath::BTreeScan {
			index_ref,
			access,
			direction,
		} => Arc::new(IndexScan::new(
			index_ref.clone(),
			access.clone(),
			*direction,
			cfg.table_name.clone(),
			None,
			None,
			cfg.version.clone(),
			None,
			None,
		)),
		AccessPath::FullTextSearch {
			index_ref,
			query,
			operator,
		} => Arc::new(FullTextScan::new(
			index_ref.clone(),
			query.clone(),
			operator.clone(),
			cfg.table_name.clone(),
			cfg.version.clone(),
			None,
		)),
		AccessPath::KnnSearch {
			index_ref,
			vector,
			k,
			ef,
		} => {
			let residual_cond = resolved_cond.and_then(strip_knn_from_condition);
			Arc::new(KnnScan::new(
				index_ref.clone(),
				vector.clone(),
				*k,
				*ef,
				cfg.table_name.clone(),
				cfg.version.clone(),
				cfg.knn_context.clone(),
				residual_cond,
				None,
			))
		}
		// Provably empty: emit a single EmptyScan operator.
		AccessPath::EmptyScan => Arc::new(super::EmptyScan::new()),
		// TableScan and nested Union should not appear as sub-paths.
		// Fall back to a table scan operator which will produce all
		// records (safe but sub-optimal).
		AccessPath::TableScan
		| AccessPath::Union {
			..
		} => Arc::new(super::TableScan::new(
			cfg.table_name.clone(),
			cfg.direction,
			None,
			None,
			None,
			None,
			None,
		)),
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use crate::ctx::Context;
	use crate::exec::planner::expr_to_physical_expr;

	/// Helper to create a Scan with all fields for testing
	async fn create_test_scan(table_name: &str, with_index_hints: bool) -> DynamicScan {
		let ctx = std::sync::Arc::new(Context::new_test());
		let source = expr_to_physical_expr(
			crate::expr::Expr::Literal(crate::expr::literal::Literal::String(table_name.into())),
			&ctx,
		)
		.await
		.expect("Failed to create physical expression");

		DynamicScan::new(
			source,
			None,
			None,
			None,
			if with_index_hints {
				Some(With::NoIndex)
			} else {
				None
			},
			None,
			None,
			None,
			None,
		)
	}

	#[tokio::test]
	async fn test_scan_struct_with_index_fields() {
		// Test that Scan can be created with all fields
		let scan = create_test_scan("test_table", false).await;
		assert!(scan.cond.is_none());
		assert!(scan.order.is_none());
		assert!(scan.with.is_none());
	}

	#[tokio::test]
	async fn test_scan_struct_with_noindex_hint() {
		// Test that Scan can be created with WITH NOINDEX
		let scan = create_test_scan("test_table", true).await;
		assert!(scan.with.is_some());
		assert!(matches!(scan.with, Some(With::NoIndex)));
	}

	#[tokio::test]
	async fn test_scan_operator_name() {
		let scan = create_test_scan("test_table", false).await;
		assert_eq!(scan.name(), "DynamicScan");
	}

	#[tokio::test]
	async fn test_scan_required_context() {
		let scan = create_test_scan("test_table", false).await;
		assert!(matches!(scan.required_context(), ContextLevel::Database));
	}

	#[test]
	fn test_determine_scan_direction_no_order() {
		// No order -> Forward
		let direction = determine_scan_direction(None);
		assert!(matches!(direction, ScanDirection::Forward));
	}

	#[test]
	fn test_determine_scan_direction_random_order() {
		use crate::expr::order::Ordering;

		// Random order -> Forward
		let order = Ordering::Random;
		let direction = determine_scan_direction(Some(&order));
		assert!(matches!(direction, ScanDirection::Forward));
	}
}