surrealdb-core 3.3.1

use std::borrow::Cow;
use std::sync::Arc;

use common::future::stream::{self, Yielder};
use futures::StreamExt;
use tracing::instrument;

use super::pipeline::{
	build_field_state, determine_scan_direction, eval_limit_expr, kv_scan_stream,
	table_read_restricted_fields,
};
use super::{FullTextScan, IndexScan, KnnScan};
use crate::catalog::providers::TableProvider;
use crate::catalog::{DatabaseId, Error, NamespaceId, table_select_permission};
use crate::err::EngineError;
use crate::exec::index::access_path::{
	AccessPath, ElementColumns, may_have_array_columns, select_access_path,
	with_array_columns_as_elements, without_btree_indexes,
};
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_and_matches_from_condition, 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::key::schema::RecordPrefix;
use crate::kvs::Direction;
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,
	/// Whether the scan is a locked read (`SELECT ... FOR UPDATE`). The
	/// source must then resolve to a specific record id at execution time;
	/// every other value shape is rejected.
	pub(crate) for_update: bool,
}

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,
			for_update: false,
		}
	}

	/// 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 whether the scan performs a locked read (`FOR UPDATE`).
	pub(crate) fn with_for_update(mut self, for_update: bool) -> Self {
		self.for_update = for_update;
		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()));
		}
		if self.for_update {
			attrs.push(("for_update".to_string(), "true".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, and a locked read registers conflict-detection state on
		// the storage engine, which requires a writeable transaction.
		let mut mode = if self.for_update {
			AccessMode::ReadWrite
		} else {
			AccessMode::ReadOnly
		};
		mode = mode.combine(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 for_update = self.for_update;
		let ctx = ctx.clone();

		let stream = stream::try_async_stream(async move |mut yielder: Yielder<_>| {
			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?;

			// FOR UPDATE targets must resolve to specific record ids so the
			// locked read is never silently skipped for other shapes.
			if for_update && !matches!(table_value, Value::RecordId(_)) {
				return Err(ControlFlow::Err(anyhow::Error::new(Error::Query {
					message: crate::dbs::FOR_UPDATE_TARGETS_ERROR.to_string(),
				})));
			}

			// 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. A FOR UPDATE source must still
					// reach `execute_record_lookup` so the locked read
					// registers the key for conflict detection: the lookup
					// performs the read first and discards the row for a zero
					// limit afterwards, keeping this plan in agreement with
					// the literal RecordIdScan plan for the same query.
					if limit_val == Some(0) && !for_update {
						return Ok(());
					}

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

					if !results.is_empty() {
						yielder.emit(ValueBatch::new(results)).await;
					}
					return Ok(());
				}
				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 Ok(());
					}

					// 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 Ok(());
						}
						values.drain(..start_val);
					}

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

					if !values.is_empty() {
						yielder.emit(ValueBatch::new(values)).await;
					}
					return Ok(());
				}
				// 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 Ok(());
					}

					// 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 Ok(());
						}
					}

					yielder.emit(ValueBatch::new(vec![other])).await;
					return Ok(());
				}
			};

			// === 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 = table_select_permission(table_def.as_deref());
				convert_permission_to_physical_runtime(catalog_perm, &ctx)
					.await
					.context("Failed to convert permission")?
			} else {
				PhysicalPermission::Allow
			};

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

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

			// 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.exec.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,
						field_permissions: Arc::clone(&field_state.field_permissions),
					},
				)
				.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!(EngineError::QueryCancelled)))?;
				}
				let mut batch = batch_result?;
				let cont = pipeline.process_batch(batch.values_mut(), &ctx).await?;
				if !batch.is_empty() {
					yielder.emit(batch).await;
				}
				if !cont {
					break;
				}
			}
			Ok(())
		});

		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())))
}

/// SECURITY: whether any column of an index is governed by a non-`Full` SELECT
/// permission for the current actor.
///
/// An index scan answers its predicate from index entries and drops the covered
/// condition, so the document-level reduction never runs on the data the
/// decision was made from — index membership then reports something about a
/// value the actor may not read. Withholding the index leaves the predicate to
/// be answered from the reduced document.
///
/// A column is governed at, beneath or above a restricted field: an ancestor's
/// reduction reaches the column, and a column over an ancestor stores the
/// restricted descendant's values as entries (an index on `acl` holds the
/// elements a restricted `acl.*` hides).
///
/// `field_permissions` is empty when permission enforcement is skipped
/// (owner/root), so a privileged actor never matches here. Mirrors the
/// plan-time `RestrictedPrefixes::index_touches` guard in `planner/select`.
fn index_touches_restricted_select_field(
	cols: &[crate::expr::Idiom],
	field_permissions: &[(crate::expr::Idiom, PhysicalPermission)],
) -> bool {
	if field_permissions.is_empty() {
		return false;
	}
	cols.iter().any(|col| {
		field_permissions
			.iter()
			.any(|(field, _)| crate::exec::planner::util::paths_overlap(&col.0, &field.0))
	})
}

/// 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: Direction,
	/// 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>>,
	/// The actor's resolved per-field SELECT permissions, shared with the
	/// scan's [`FieldState`] rather than copied. Empty when enforcement is
	/// skipped (owner/root). Used to withhold indexes whose columns the actor
	/// may not read from access-path selection.
	///
	/// [`FieldState`]: crate::exec::operators::scan::pipeline::FieldState
	field_permissions: Arc<Vec<(crate::expr::Idiom, PhysicalPermission)>>,
}

/// 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;
			let capabilities = ctx.root().ctx.get_capabilities();
			// This path scans a table, so every row reaching the condition is
			// a record. A field read off one is still only inert while
			// materialising that row evaluates nothing.
			// Mirror the scan's own enforcement decision: a session that skips
			// permission evaluation never runs those expressions.
			let check_perms = match ctx.database() {
				Ok(db_ctx) => should_check_perms(db_ctx, Action::View).unwrap_or(true),
				Err(_) => true,
			};
			let restricted_fields = table_read_restricted_fields(
				&txn,
				cfg.ns_id,
				cfg.db_id,
				&cfg.table_name,
				version_stamp,
				check_perms,
			)
			.await;
			fold_condition_expressions(
				&mut cond,
				ctx.function_registry(),
				&capabilities,
				restricted_fields.as_ref(),
			);
			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")?;
		// SECURITY: drop indexes whose columns the actor may not select, for
		// the reason given on `index_touches_restricted_select_field`. The
		// actor's real field permissions are known here, so this cannot
		// over-apply the way a conservative plan-time assumption would.
		let indexes: std::sync::Arc<[_]> = if indexes
			.iter()
			.any(|ix| index_touches_restricted_select_field(&ix.cols, &cfg.field_permissions))
		{
			indexes
				.iter()
				.filter(|ix| {
					!index_touches_restricted_select_field(&ix.cols, &cfg.field_permissions)
				})
				.cloned()
				.collect::<Vec<_>>()
				.into()
		} else {
			indexes
		};
		// A column that stores array elements is planned as the element
		// column it behaves as, as the static planner does. Without the field
		// list no column can be classified, so the b-tree indexes are
		// withheld and the scan answers.
		let columns = if may_have_array_columns(&indexes) {
			match txn.all_tb_fields(cfg.ns_id, cfg.db_id, &cfg.table_name, version_stamp).await {
				Ok(fields) => with_array_columns_as_elements(indexes, &fields),
				Err(e) => {
					tracing::warn!(
						table = %cfg.table_name,
						error = %e,
						"field list failed in DynamicScan; planning without b-tree indexes",
					);
					ElementColumns::unchanged(without_btree_indexes(indexes))
				}
			}
		} else {
			ElementColumns::unchanged(indexes)
		};

		let analyzer = columns.analyzer(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))
				// A pushed LIMIT counts rows, and a scan that fans out reads
				// several entries per row, so the pipeline would truncate on
				// entries. Leave that combination to the KV scan.
				&& !crate::exec::index::access_path::access_fans_out(&index_ref.cols, &access)) =>
		{
			// A key range leaving a fanning column open reaches a record
			// through several entries; the scan yields records, so the
			// repeats are collapsed before anything above counts them.
			let fans_out =
				crate::exec::index::access_path::access_fans_out(&index_ref.cols, &access);
			let operator: Arc<dyn ExecOperator> = Arc::new(IndexScan::new(
				index_ref,
				access,
				direction,
				cfg.table_name,
				None,
				None,
				cfg.version,
				None,
				None,
			));
			let stream = if fans_out {
				crate::exec::operators::DistinctRecords::new(operator).execute(ctx)?
			} else {
				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,
			// Runtime-resolved access paths never carry a prefilter — the
			// bitmap split is a plan-time-only analysis (#548).
			prefilter: _,
		}) => {
			// Strip KNN operators — and MATCHES conjuncts, which are not
			// evaluable inside the ANN search — from the resolved condition
			// to get the residual for HNSW pushdown.
			let residual_cond =
				resolved_cond.as_ref().and_then(strip_knn_and_matches_from_condition);
			// This path never builds an allow-list bitmap, so MATCHES
			// conjuncts must stay evaluable during the ANN traversal for
			// every index format — otherwise the top-k is chosen among all
			// records and matching neighbours outside it are unrecoverable.
			// Compile the KNN-stripped condition — MATCHES kept — into the
			// scan's matches condition, which supersedes the residual above.
			// The compile reproduces the select planner's MATCHES
			// registration scope (allowlist from both condition forms, this
			// table as the executor table); an unregistered MATCHES compiles
			// to an operator that answers `false` for every row.
			let matches_condition = match resolved_cond.as_ref().and_then(strip_knn_from_condition)
			{
				Some(c) => {
					let mut allowlist = std::collections::HashSet::new();
					if let Some(orig) = cfg.cond.as_ref() {
						crate::exec::physical_expr::collect_cond_matches(&orig.0, &mut allowlist);
					}
					crate::exec::physical_expr::collect_cond_matches(&c.0, &mut allowlist);
					let mut planner =
						crate::exec::planner::Planner::new(ctx.ctx(), ctx.function_registry());
					planner.set_matches_scope(Arc::new(crate::exec::physical_expr::MatchesScope {
						allowlist,
						executor_tables: Arc::from(vec![cfg.table_name.clone()]),
					}));
					Some(planner.physical_expr(c.0).await.map_err(|e| anyhow::anyhow!("{e}"))?)
				}
				None => None,
			};
			let knn_op = KnnScan::new(
				index_ref,
				vector,
				k,
				ef,
				cfg.table_name,
				cfg.version,
				cfg.knn_context.clone(),
				residual_cond,
				None,
				None,
			)
			.with_matches_condition(matches_condition);
			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.)
		_ => {
			// Fail closed: a record-range scan over a lightweight relation
			// would be wrong-empty, and a runtime-resolved source cannot fall
			// back to the legacy collectors.
			{
				if txn.get_tb(cfg.ns_id, cfg.db_id, &cfg.table_name, None).await?.is_some_and(
					|def| crate::kvs::lightweight::lightweight_relation(&def.table_type).is_some(),
				) {
					return Err(ControlFlow::Err(anyhow::anyhow!(
						"a LIGHTWEIGHT relation cannot be scanned by this execution path; \
						 reference the table statically so the record-less scan can serve it"
					)));
				}
			}
			let range = RecordPrefix {
				ns: cfg.ns_id,
				db: cfg.db_id,
				tb: Cow::Borrowed(&cfg.table_name),
			}
			.range()?;

			let stream = kv_scan_stream(
				txn,
				range,
				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,
			prefilter: _,
		} => {
			let residual_cond = resolved_cond.and_then(strip_knn_and_matches_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,
				None,
			))
		}
		// Provably empty: emit a single EmptyScan operator.
		AccessPath::EmptyScan => Arc::new(super::EmptyScan::new()),
		// TableScan, nested Union and BitmapFusion should not appear as
		// sub-paths (bitmap fusion is a plan-time-only path; the runtime
		// analyzer never emits it). Fall back to a table scan operator
		// which will produce all records (safe but sub-optimal).
		AccessPath::TableScan
		| AccessPath::Union {
			..
		}
		| AccessPath::BitmapFusion {
			..
		} => 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 registry = crate::exec::function::FunctionRegistry::with_builtins();
		let source = expr_to_physical_expr(
			crate::expr::Expr::Literal(crate::expr::literal::Literal::String(table_name.into())),
			&ctx,
			&registry,
		)
		.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, Direction::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, Direction::Forward));
	}
}