Skip to main content

radixdb_executor/expression/
vm.rs

1// Copyright 2026 RadixDB Contributors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15// Expression Virtual Machine
16//
17// The VM executes compiled Programs against row data.
18// Design goals:
19// - Zero allocation in hot path
20// - Linear instruction dispatch
21// - Reusable across rows (clear() between uses)
22// - No recursion
23
24use std::borrow::Cow;
25use std::sync::Arc;
26
27use smallvec::SmallVec;
28
29use super::execution_context::ExecuteContext;
30use super::ops::{CompiledPattern, Op};
31use super::program::Program;
32use radixdb_core::SmartString;
33use radixdb_core::{DataType, Error, Result, Value, NULL_VALUE};
34
35/// Stack value that can be borrowed (from row/constants) or owned (from operations)
36type StackValue<'a> = Cow<'a, Value>;
37
38/// Stack capacity for inline storage (avoids heap allocation for simple expressions)
39/// Most expressions need 4-8 stack slots, so 8 covers the common case.
40const STACK_INLINE_CAPACITY: usize = 16;
41
42/// Arithmetic operation type (used for safe wrapping operations)
43#[derive(Clone, Copy)]
44#[allow(dead_code)]
45enum ArithmeticOp {
46    Add,
47    Sub,
48    Mul,
49    Div,
50    Mod,
51}
52
53/// Expression Virtual Machine
54///
55/// Executes compiled Programs against row data.
56/// The VM is reusable - call execute() with different contexts.
57/// Capacity for reusable args buffer (most functions have <= 4 args)
58const ARGS_BUFFER_CAPACITY: usize = 8;
59
60/// Parsed interval value: either a fixed-length duration or a calendar-relative month count.
61/// Months/years require calendar-aware arithmetic (leap years, variable month lengths).
62enum IntervalValue {
63    Duration(chrono::Duration),
64    Months(i64),
65}
66
67pub struct ExprVM {
68    /// Evaluation stack (reused between executions)
69    /// Uses SmallVec to avoid heap allocation for simple expressions (stack depth <= 16)
70    stack: SmallVec<[Value; STACK_INLINE_CAPACITY]>,
71
72    /// Reusable buffer for function arguments (avoids allocation per call)
73    /// Uses SmallVec to avoid heap allocation for functions with <= 8 args
74    args_buffer: SmallVec<[Value; ARGS_BUFFER_CAPACITY]>,
75
76    /// Cache for dynamic LIKE patterns (avoids recompilation per row)
77    /// Stores (pattern_string, case_insensitive, escape_char, compiled_pattern)
78    cached_like: Option<(SmartString, bool, Option<char>, CompiledPattern)>,
79
80    /// Cache for dynamic GLOB patterns (separate from LIKE to avoid cross-contamination)
81    /// Stores (pattern_string, compiled_pattern)
82    cached_glob: Option<(SmartString, CompiledPattern)>,
83
84    /// Cache for dynamic REGEXP patterns (avoids recompilation per row)
85    /// Stores (pattern_string, compiled_regex)
86    cached_regexp: Option<(SmartString, regex::Regex)>,
87}
88
89impl ExprVM {
90    /// Create a new VM with default stack capacity
91    /// Uses inline storage for up to 16 stack values and 8 args (no heap allocation)
92    pub fn new() -> Self {
93        Self {
94            stack: SmallVec::new(),
95            args_buffer: SmallVec::new(),
96            cached_like: None,
97            cached_glob: None,
98            cached_regexp: None,
99        }
100    }
101
102    /// Create a VM with specific stack capacity
103    /// If capacity > 16, will spill to heap when needed
104    pub fn with_capacity(capacity: usize) -> Self {
105        Self {
106            stack: SmallVec::with_capacity(capacity),
107            args_buffer: SmallVec::new(),
108            cached_like: None,
109            cached_glob: None,
110            cached_regexp: None,
111        }
112    }
113
114    /// Execute a program and return the result
115    #[inline]
116    pub fn execute(&mut self, program: &Program, ctx: &ExecuteContext) -> Result<Value> {
117        // Ensure stack has enough capacity
118        if self.stack.capacity() < program.max_stack_depth() {
119            self.stack
120                .reserve(program.max_stack_depth() - self.stack.capacity());
121        }
122        self.stack.clear();
123
124        let ops = program.ops();
125        if ops.is_empty() {
126            return Ok(Value::null_unknown());
127        }
128
129        let mut pc: usize = 0;
130
131        // Main execution loop
132        loop {
133            if pc >= ops.len() {
134                break;
135            }
136
137            match &ops[pc] {
138                // =============================================================
139                // LOAD OPERATIONS
140                // =============================================================
141                Op::LoadColumn(idx) => {
142                    let idx = *idx as usize;
143                    let value = ctx
144                        .row
145                        .get(idx)
146                        .cloned()
147                        .unwrap_or_else(Value::null_unknown);
148                    self.stack.push(value);
149                    pc += 1;
150                }
151
152                Op::LoadColumn2(idx) => {
153                    let idx = *idx as usize;
154                    let value = ctx
155                        .row2
156                        .and_then(|r| r.get(idx).cloned())
157                        .unwrap_or_else(Value::null_unknown);
158                    self.stack.push(value);
159                    pc += 1;
160                }
161
162                Op::LoadOuterColumn(name) => {
163                    let value = ctx
164                        .outer_row
165                        .and_then(|r| r.get(name.as_ref()).cloned())
166                        .unwrap_or_else(Value::null_unknown);
167                    self.stack.push(value);
168                    pc += 1;
169                }
170
171                Op::LoadConst(value) => {
172                    self.stack.push(value.clone());
173                    pc += 1;
174                }
175
176                Op::LoadParam(idx) => {
177                    let idx = *idx as usize;
178                    let value = ctx
179                        .params
180                        .get(idx)
181                        .cloned()
182                        .unwrap_or_else(Value::null_unknown);
183                    self.stack.push(value);
184                    pc += 1;
185                }
186
187                Op::LoadNamedParam(name) => {
188                    let value = ctx
189                        .named_params
190                        .and_then(|p| p.get(name.as_ref()).cloned())
191                        .or_else(|| {
192                            // DEFAULT evaluation and catalog recovery compile
193                            // scalar expressions without a full statement
194                            // context. CURRENT_TIMESTAMP must remain a valid
195                            // system value there instead of degrading to NULL;
196                            // ordinary statement execution supplies one stable
197                            // value through named_params.
198                            (name.as_ref() == "CURRENT_STATEMENT_TIMESTAMP").then(|| {
199                                Value::timestamp(
200                                    radixdb_core::time_compat::system_time_now().into(),
201                                )
202                            })
203                        })
204                        .unwrap_or_else(Value::null_unknown);
205                    self.stack.push(value);
206                    pc += 1;
207                }
208
209                Op::LoadNull(dt) => {
210                    self.stack.push(Value::Null(*dt));
211                    pc += 1;
212                }
213
214                Op::LoadAggregateResult(idx) => {
215                    // Aggregate results are stored in the row at specific indices
216                    let idx = *idx as usize;
217                    let value = ctx
218                        .row
219                        .get(idx)
220                        .cloned()
221                        .unwrap_or_else(Value::null_unknown);
222                    self.stack.push(value);
223                    pc += 1;
224                }
225
226                Op::LoadTransactionId => {
227                    // Load current transaction ID, or NULL if not in a transaction
228                    let value = match ctx.transaction_id {
229                        Some(txn_id) => Value::Integer(i64::try_from(txn_id).map_err(|_| {
230                            radixdb_core::Error::invalid_argument(
231                                "transaction ID exceeds the SQL INTEGER domain",
232                            )
233                        })?),
234                        None => Value::null_unknown(),
235                    };
236                    self.stack.push(value);
237                    pc += 1;
238                }
239
240                // =============================================================
241                // COMPARISON OPERATIONS
242                // =============================================================
243                Op::Eq => {
244                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
245                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
246                    let result = Self::sql_equality_result(ctx, &a, &b, false)?;
247                    self.stack.push(result);
248                    pc += 1;
249                }
250
251                Op::Ne => {
252                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
253                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
254                    let result = Self::sql_equality_result(ctx, &a, &b, true)?;
255                    self.stack.push(result);
256                    pc += 1;
257                }
258
259                Op::Lt => {
260                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
261                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
262                    let result = Self::sql_order_result(ctx, &a, &b, |ordering| {
263                        ordering == std::cmp::Ordering::Less
264                    })?;
265                    self.stack.push(result);
266                    pc += 1;
267                }
268
269                Op::Le => {
270                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
271                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
272                    let result = Self::sql_order_result(ctx, &a, &b, |ordering| {
273                        ordering != std::cmp::Ordering::Greater
274                    })?;
275                    self.stack.push(result);
276                    pc += 1;
277                }
278
279                Op::Gt => {
280                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
281                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
282                    let result = Self::sql_order_result(ctx, &a, &b, |ordering| {
283                        ordering == std::cmp::Ordering::Greater
284                    })?;
285                    self.stack.push(result);
286                    pc += 1;
287                }
288
289                Op::Ge => {
290                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
291                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
292                    let result = Self::sql_order_result(ctx, &a, &b, |ordering| {
293                        ordering != std::cmp::Ordering::Less
294                    })?;
295                    self.stack.push(result);
296                    pc += 1;
297                }
298
299                Op::IsNull => {
300                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
301                    self.stack.push(Value::Boolean(v.is_null()));
302                    pc += 1;
303                }
304
305                Op::IsNotNull => {
306                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
307                    self.stack.push(Value::Boolean(!v.is_null()));
308                    pc += 1;
309                }
310
311                Op::IsDistinctFrom => {
312                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
313                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
314                    // NULL IS DISTINCT FROM NULL = FALSE
315                    // NULL IS DISTINCT FROM non-NULL = TRUE
316                    // non-NULL IS DISTINCT FROM NULL = TRUE
317                    // Otherwise use regular comparison
318                    let result = match (a.is_null(), b.is_null()) {
319                        (true, true) => false,
320                        (true, false) | (false, true) => true,
321                        (false, false) => !Self::sql_values_equal(ctx, &a, &b)?,
322                    };
323                    self.stack.push(Value::Boolean(result));
324                    pc += 1;
325                }
326
327                Op::IsNotDistinctFrom => {
328                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
329                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
330                    let result = match (a.is_null(), b.is_null()) {
331                        (true, true) => true,
332                        (true, false) | (false, true) => false,
333                        (false, false) => Self::sql_values_equal(ctx, &a, &b)?,
334                    };
335                    self.stack.push(Value::Boolean(result));
336                    pc += 1;
337                }
338
339                // =============================================================
340                // FUSED COMPARISON OPERATIONS
341                // Single instruction for column vs constant (avoids push/pop)
342                // =============================================================
343                Op::EqColumnConst(idx, val) => {
344                    let col_val = ctx
345                        .row
346                        .get(*idx as usize)
347                        .unwrap_or(&Value::Null(DataType::Null));
348                    let result = Self::sql_equality_result(ctx, col_val, val, false)?;
349                    self.stack.push(result);
350                    pc += 1;
351                }
352
353                Op::NeColumnConst(idx, val) => {
354                    let col_val = ctx
355                        .row
356                        .get(*idx as usize)
357                        .unwrap_or(&Value::Null(DataType::Null));
358                    let result = Self::sql_equality_result(ctx, col_val, val, true)?;
359                    self.stack.push(result);
360                    pc += 1;
361                }
362
363                Op::LtColumnConst(idx, val) => {
364                    let col_val = ctx
365                        .row
366                        .get(*idx as usize)
367                        .unwrap_or(&Value::Null(DataType::Null));
368                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
369                        ordering == std::cmp::Ordering::Less
370                    })?;
371                    self.stack.push(result);
372                    pc += 1;
373                }
374
375                Op::LeColumnConst(idx, val) => {
376                    let col_val = ctx
377                        .row
378                        .get(*idx as usize)
379                        .unwrap_or(&Value::Null(DataType::Null));
380                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
381                        ordering != std::cmp::Ordering::Greater
382                    })?;
383                    self.stack.push(result);
384                    pc += 1;
385                }
386
387                Op::GtColumnConst(idx, val) => {
388                    let col_val = ctx
389                        .row
390                        .get(*idx as usize)
391                        .unwrap_or(&Value::Null(DataType::Null));
392                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
393                        ordering == std::cmp::Ordering::Greater
394                    })?;
395                    self.stack.push(result);
396                    pc += 1;
397                }
398
399                Op::GeColumnConst(idx, val) => {
400                    let col_val = ctx
401                        .row
402                        .get(*idx as usize)
403                        .unwrap_or(&Value::Null(DataType::Null));
404                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
405                        ordering != std::cmp::Ordering::Less
406                    })?;
407                    self.stack.push(result);
408                    pc += 1;
409                }
410
411                Op::IsNullColumn(idx) => {
412                    let col_val = ctx
413                        .row
414                        .get(*idx as usize)
415                        .unwrap_or(&Value::Null(DataType::Null));
416                    self.stack.push(Value::Boolean(col_val.is_null()));
417                    pc += 1;
418                }
419
420                Op::IsNotNullColumn(idx) => {
421                    let col_val = ctx
422                        .row
423                        .get(*idx as usize)
424                        .unwrap_or(&Value::Null(DataType::Null));
425                    self.stack.push(Value::Boolean(!col_val.is_null()));
426                    pc += 1;
427                }
428
429                Op::LikeColumn(idx, pattern, case_insensitive) => {
430                    let col_val = ctx
431                        .row
432                        .get(*idx as usize)
433                        .unwrap_or(&Value::Null(DataType::Null));
434                    let result = match col_val {
435                        Value::Text(s) => Value::Boolean(pattern.matches(s, *case_insensitive)),
436                        Value::Null(_) => Value::Null(DataType::Boolean),
437                        _ => Value::Boolean(false),
438                    };
439                    self.stack.push(result);
440                    pc += 1;
441                }
442
443                Op::InSetColumn(idx, set, has_null) => {
444                    let col_val = ctx
445                        .row
446                        .get(*idx as usize)
447                        .unwrap_or(&Value::Null(DataType::Null));
448                    let result = if col_val.is_null() {
449                        Value::Null(DataType::Boolean)
450                    } else if Self::sql_set_contains(ctx, set, col_val)? {
451                        Value::Boolean(true)
452                    } else if *has_null {
453                        Value::Null(DataType::Boolean)
454                    } else {
455                        Value::Boolean(false)
456                    };
457                    self.stack.push(result);
458                    pc += 1;
459                }
460
461                Op::BetweenColumnConst(idx, low, high) => {
462                    let col_val = ctx
463                        .row
464                        .get(*idx as usize)
465                        .unwrap_or(&Value::Null(DataType::Null));
466                    let result = Self::sql_between_result(ctx, col_val, low, high, false)?;
467                    self.stack.push(result);
468                    pc += 1;
469                }
470
471                // =============================================================
472                // LOGICAL OPERATIONS
473                // =============================================================
474                Op::And(jump_target) => {
475                    // Short-circuit AND: if top is false, jump
476                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
477                    match top {
478                        Value::Boolean(false) => {
479                            // Result is false, jump to target
480                            pc = *jump_target as usize;
481                        }
482                        Value::Null(_) => {
483                            // Need to evaluate right side to check for false
484                            pc += 1;
485                        }
486                        _ => {
487                            // True or truthy, continue to evaluate right side
488                            pc += 1;
489                        }
490                    }
491                }
492
493                Op::Or(jump_target) => {
494                    // Short-circuit OR: if top is true, jump
495                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
496                    match top {
497                        Value::Boolean(true) => {
498                            // Result is true, jump to target
499                            pc = *jump_target as usize;
500                        }
501                        Value::Null(_) => {
502                            // Need to evaluate right side to check for true
503                            pc += 1;
504                        }
505                        _ => {
506                            // False or falsy, continue to evaluate right side
507                            pc += 1;
508                        }
509                    }
510                }
511
512                Op::AndFinalize => {
513                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
514                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
515                    let result = match (Self::to_tribool(&a), Self::to_tribool(&b)) {
516                        (Some(false), _) | (_, Some(false)) => Value::Boolean(false),
517                        (Some(true), Some(true)) => Value::Boolean(true),
518                        _ => Value::Null(DataType::Boolean),
519                    };
520                    self.stack.push(result);
521                    pc += 1;
522                }
523
524                Op::OrFinalize => {
525                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
526                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
527                    let result = match (Self::to_tribool(&a), Self::to_tribool(&b)) {
528                        (Some(true), _) | (_, Some(true)) => Value::Boolean(true),
529                        (Some(false), Some(false)) => Value::Boolean(false),
530                        _ => Value::Null(DataType::Boolean),
531                    };
532                    self.stack.push(result);
533                    pc += 1;
534                }
535
536                Op::Not => {
537                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
538                    let result = match Self::to_tribool(&v) {
539                        Some(b) => Value::Boolean(!b),
540                        None => Value::Null(DataType::Boolean),
541                    };
542                    self.stack.push(result);
543                    pc += 1;
544                }
545
546                Op::Xor => {
547                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
548                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
549                    let result = match (Self::to_tribool(&a), Self::to_tribool(&b)) {
550                        (Some(a), Some(b)) => Value::Boolean(a ^ b),
551                        _ => Value::Null(DataType::Boolean),
552                    };
553                    self.stack.push(result);
554                    pc += 1;
555                }
556
557                // =============================================================
558                // ARITHMETIC OPERATIONS
559                // =============================================================
560                Op::Add => {
561                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
562                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
563                    // Handle timestamp + interval or timestamp + integer (days)
564                    let result = match (&a, &b) {
565                        (Value::Timestamp(t), Value::Integer(days)) => {
566                            Self::timestamp_add_days(*t, *days)?
567                        }
568                        (Value::Integer(days), Value::Timestamp(t)) => {
569                            Self::timestamp_add_days(*t, *days)?
570                        }
571                        (Value::Extension(_), Value::Integer(days))
572                            if a.as_date_days().is_some() =>
573                        {
574                            Self::date_add_days(a.as_date_days().expect("date was checked"), *days)?
575                        }
576                        (Value::Integer(days), Value::Extension(_))
577                            if b.as_date_days().is_some() =>
578                        {
579                            Self::date_add_days(b.as_date_days().expect("date was checked"), *days)?
580                        }
581                        (Value::Timestamp(_), Value::Text(_)) => {
582                            // Parse interval string - pass references directly to avoid clone
583                            self.timestamp_add_interval(&a, &b, true)?
584                        }
585                        _ => Self::arithmetic_op(&a, &b, ArithmeticOp::Add, |x, y| x + y)?,
586                    };
587                    self.stack.push(result);
588                    pc += 1;
589                }
590
591                Op::Sub => {
592                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
593                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
594                    // Handle timestamp - interval, timestamp - integer (days), or timestamp - timestamp
595                    let result = match (&a, &b) {
596                        (Value::Timestamp(t1), Value::Timestamp(t2)) => {
597                            // Return interval text
598                            let duration = t1.signed_duration_since(*t2);
599                            Value::Text(SmartString::from_string(
600                                self.format_duration_as_interval(duration),
601                            ))
602                        }
603                        (Value::Timestamp(t), Value::Integer(days)) => Self::timestamp_add_days(
604                            *t,
605                            days.checked_neg().ok_or_else(|| {
606                                Error::Type("timestamp interval overflow".to_string())
607                            })?,
608                        )?,
609                        (Value::Extension(_), Value::Integer(days))
610                            if a.as_date_days().is_some() =>
611                        {
612                            Self::date_add_days(
613                                a.as_date_days().expect("date was checked"),
614                                days.checked_neg().ok_or_else(|| {
615                                    Error::Type("DATE arithmetic overflow".to_string())
616                                })?,
617                            )?
618                        }
619                        (Value::Extension(_), Value::Extension(_))
620                            if a.as_date_days().is_some() && b.as_date_days().is_some() =>
621                        {
622                            Value::Integer(
623                                i64::from(a.as_date_days().expect("date was checked"))
624                                    - i64::from(b.as_date_days().expect("date was checked")),
625                            )
626                        }
627                        (Value::Timestamp(_), Value::Text(_)) => {
628                            // Parse interval string - pass references directly to avoid clone
629                            self.timestamp_add_interval(&a, &b, false)?
630                        }
631                        _ => Self::arithmetic_op(&a, &b, ArithmeticOp::Sub, |x, y| x - y)?,
632                    };
633                    self.stack.push(result);
634                    pc += 1;
635                }
636
637                Op::Mul => {
638                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
639                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
640                    let result = Self::arithmetic_op(&a, &b, ArithmeticOp::Mul, |x, y| x * y)?;
641                    self.stack.push(result);
642                    pc += 1;
643                }
644
645                Op::Div => {
646                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
647                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
648                    let result = Self::div_op(&a, &b)?;
649                    self.stack.push(result);
650                    pc += 1;
651                }
652
653                Op::Mod => {
654                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
655                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
656                    let result = Self::mod_op(&a, &b)?;
657                    self.stack.push(result);
658                    pc += 1;
659                }
660
661                Op::Neg => {
662                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
663                    let result = match v {
664                        Value::Integer(i) => match i.checked_neg() {
665                            Some(neg) => Value::Integer(neg),
666                            None => Value::Null(DataType::Integer), // i64::MIN overflow
667                        },
668                        Value::Float(f) => Value::Float(-f),
669                        Value::Null(dt) => Value::Null(dt),
670                        _ => Value::Null(DataType::Null),
671                    };
672                    self.stack.push(result);
673                    pc += 1;
674                }
675
676                // =============================================================
677                // BITWISE OPERATIONS (inlined for performance)
678                // =============================================================
679                Op::BitAnd => {
680                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
681                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
682                    let result = match (&a, &b) {
683                        (Value::Integer(x), Value::Integer(y)) => Value::Integer(x & y),
684                        _ if a.is_null() || b.is_null() => Value::Null(DataType::Integer),
685                        _ => Value::Null(DataType::Null),
686                    };
687                    self.stack.push(result);
688                    pc += 1;
689                }
690
691                Op::BitOr => {
692                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
693                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
694                    let result = match (&a, &b) {
695                        (Value::Integer(x), Value::Integer(y)) => Value::Integer(x | y),
696                        _ if a.is_null() || b.is_null() => Value::Null(DataType::Integer),
697                        _ => Value::Null(DataType::Null),
698                    };
699                    self.stack.push(result);
700                    pc += 1;
701                }
702
703                Op::BitXor => {
704                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
705                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
706                    let result = match (&a, &b) {
707                        (Value::Integer(x), Value::Integer(y)) => Value::Integer(x ^ y),
708                        _ if a.is_null() || b.is_null() => Value::Null(DataType::Integer),
709                        _ => Value::Null(DataType::Null),
710                    };
711                    self.stack.push(result);
712                    pc += 1;
713                }
714
715                Op::BitNot => {
716                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
717                    let result = match v {
718                        Value::Integer(i) => Value::Integer(!i),
719                        Value::Null(dt) => Value::Null(dt),
720                        _ => Value::Null(DataType::Null),
721                    };
722                    self.stack.push(result);
723                    pc += 1;
724                }
725
726                Op::Shl => {
727                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
728                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
729                    let result = match (&a, &b) {
730                        (Value::Integer(x), Value::Integer(y)) => {
731                            Value::Integer(x.wrapping_shl(*y as u32))
732                        }
733                        _ if a.is_null() || b.is_null() => Value::Null(DataType::Integer),
734                        _ => Value::Null(DataType::Null),
735                    };
736                    self.stack.push(result);
737                    pc += 1;
738                }
739
740                Op::Shr => {
741                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
742                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
743                    let result = match (&a, &b) {
744                        (Value::Integer(x), Value::Integer(y)) => {
745                            Value::Integer(x.wrapping_shr(*y as u32))
746                        }
747                        _ if a.is_null() || b.is_null() => Value::Null(DataType::Integer),
748                        _ => Value::Null(DataType::Null),
749                    };
750                    self.stack.push(result);
751                    pc += 1;
752                }
753
754                // =============================================================
755                // STRING OPERATIONS
756                // =============================================================
757                Op::Concat => {
758                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
759                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
760                    let result = if a.is_null() || b.is_null() {
761                        Value::Null(DataType::Text)
762                    } else {
763                        // Fast path: both are Text
764                        match (&a, &b) {
765                            (Value::Text(a_str), Value::Text(b_str)) => {
766                                // Use optimized concat - handles inline and heap efficiently
767                                Value::Text(SmartString::concat(a_str, b_str))
768                            }
769                            (Value::Text(a_str), _) => {
770                                // a is Text, b needs conversion - use Arc to avoid shrink_to_fit
771                                use std::fmt::Write;
772                                let mut s = String::with_capacity(a_str.len() + 32);
773                                s.push_str(a_str);
774                                let _ = write!(s, "{}", b);
775                                Value::Text(SmartString::from_string_shared(s))
776                            }
777                            (_, Value::Text(b_str)) => {
778                                // a needs conversion, b is Text - use Arc to avoid shrink_to_fit
779                                use std::fmt::Write;
780                                let mut s = String::with_capacity(32 + b_str.len());
781                                let _ = write!(s, "{}", a);
782                                s.push_str(b_str);
783                                Value::Text(SmartString::from_string_shared(s))
784                            }
785                            _ => {
786                                // Both need conversion - use Arc to avoid shrink_to_fit
787                                use std::fmt::Write;
788                                let mut s = String::with_capacity(64);
789                                let _ = write!(s, "{}{}", a, b);
790                                Value::Text(SmartString::from_string_shared(s))
791                            }
792                        }
793                    };
794                    self.stack.push(result);
795                    pc += 1;
796                }
797
798                Op::ConcatN(n) => {
799                    let n = *n as usize;
800                    let start = self.stack.len().saturating_sub(n);
801
802                    // Single pass: check for NULL and calculate total length
803                    let mut total_len = 0usize;
804                    let mut has_null = false;
805                    let mut all_text = true;
806
807                    for v in &self.stack[start..] {
808                        match v {
809                            Value::Null(_) => {
810                                has_null = true;
811                                break;
812                            }
813                            Value::Text(s) => total_len += s.len(),
814                            _ => {
815                                all_text = false;
816                                total_len += 32;
817                            }
818                        }
819                    }
820
821                    if has_null {
822                        self.stack.truncate(start);
823                        self.stack.push(Value::Null(DataType::Text));
824                        pc += 1;
825                        continue;
826                    }
827
828                    // Build result - optimize for inline vs heap
829                    let result = if all_text && total_len <= 15 {
830                        // Fast path: build directly into inline SmartString (no heap allocation)
831                        let mut data = [0u8; 15];
832                        let mut pos = 0;
833                        for v in self.stack.drain(start..) {
834                            if let Value::Text(text) = v {
835                                let bytes = text.as_bytes();
836                                data[pos..pos + bytes.len()].copy_from_slice(bytes);
837                                pos += bytes.len();
838                            }
839                        }
840                        let text = std::str::from_utf8(&data[..total_len])
841                            .expect("concatenated Text values remain valid UTF-8");
842                        SmartString::new(text)
843                    } else if all_text {
844                        // Heap path: exact capacity, into_boxed_str is O(1) when len == capacity
845                        let mut s = String::with_capacity(total_len);
846                        for v in self.stack.drain(start..) {
847                            if let Value::Text(text) = v {
848                                s.push_str(&text);
849                            }
850                        }
851                        // len == capacity, so into_boxed_str is O(1)
852                        SmartString::from_string(s)
853                    } else {
854                        // Mixed types: capacity is estimate, use Arc to avoid shrink_to_fit
855                        let mut s = String::with_capacity(total_len);
856                        for v in self.stack.drain(start..) {
857                            match v {
858                                Value::Text(text) => s.push_str(&text),
859                                _ => {
860                                    use std::fmt::Write;
861                                    let _ = write!(s, "{}", v);
862                                }
863                            }
864                        }
865                        SmartString::from_string_shared(s)
866                    };
867                    self.stack.push(Value::Text(result));
868                    pc += 1;
869                }
870
871                Op::Like(pattern, case_insensitive) => {
872                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
873                    let result = match &v {
874                        Value::Text(s) => Value::Boolean(pattern.matches(s, *case_insensitive)),
875                        Value::Null(_) => Value::Null(DataType::Boolean),
876                        _ => Value::Boolean(false),
877                    };
878                    self.stack.push(result);
879                    pc += 1;
880                }
881
882                Op::Glob(pattern) => {
883                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
884                    let result = match &v {
885                        Value::Text(s) => Value::Boolean(pattern.matches(s, false)),
886                        Value::Null(_) => Value::Null(DataType::Boolean),
887                        _ => Value::Boolean(false),
888                    };
889                    self.stack.push(result);
890                    pc += 1;
891                }
892
893                Op::Regexp(regex) => {
894                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
895                    let result = match &v {
896                        Value::Text(s) => Value::Boolean(regex.is_match(s)),
897                        Value::Null(_) => Value::Null(DataType::Boolean),
898                        _ => Value::Boolean(false),
899                    };
900                    self.stack.push(result);
901                    pc += 1;
902                }
903
904                Op::LikeEscape(pattern, case_insensitive, _escape) => {
905                    // ESCAPE is already incorporated into the compiled pattern
906                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
907                    let result = match &v {
908                        Value::Text(s) => Value::Boolean(pattern.matches(s, *case_insensitive)),
909                        Value::Null(_) => Value::Null(DataType::Boolean),
910                        _ => Value::Boolean(false),
911                    };
912                    self.stack.push(result);
913                    pc += 1;
914                }
915
916                Op::LikeDynamic(case_insensitive) => {
917                    let pattern_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
918                    let text_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
919                    let ci = *case_insensitive;
920                    let result = match (&text_val, &pattern_val) {
921                        (Value::Text(text), Value::Text(pat)) => {
922                            // Cache compiled pattern: parameters are constant per query,
923                            // so recompiling every row is wasteful
924                            let need_compile = match &self.cached_like {
925                                Some((cached_pat, cached_ci, cached_esc, _)) => {
926                                    cached_pat.as_str() != pat.as_str()
927                                        || *cached_ci != ci
928                                        || cached_esc.is_some()
929                                }
930                                None => true,
931                            };
932                            if need_compile {
933                                let compiled = CompiledPattern::compile(pat, ci)?;
934                                self.cached_like = Some((pat.clone(), ci, None, compiled));
935                            }
936                            let (_, _, _, ref compiled) = self.cached_like.as_ref().unwrap();
937                            Value::Boolean(compiled.matches(text, ci))
938                        }
939                        (Value::Null(_), _) | (_, Value::Null(_)) => Value::Null(DataType::Boolean),
940                        _ => Value::Boolean(false),
941                    };
942                    self.stack.push(result);
943                    pc += 1;
944                }
945
946                Op::LikeDynamicEscape(case_insensitive, escape_char) => {
947                    let pattern_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
948                    let text_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
949                    let result = match (&text_val, &pattern_val) {
950                        (Value::Text(text), Value::Text(pat)) => {
951                            let ci = *case_insensitive;
952                            let esc = *escape_char;
953                            let need_compile = match &self.cached_like {
954                                Some((cached_pat, cached_ci, cached_esc, _)) => {
955                                    cached_pat.as_str() != pat.as_str()
956                                        || *cached_ci != ci
957                                        || *cached_esc != Some(esc)
958                                }
959                                None => true,
960                            };
961                            if need_compile {
962                                // Pre-process the escape character in the pattern at runtime,
963                                // converting e.g. !% -> \% so CompiledPattern treats it as literal
964                                let processed = process_like_escape_runtime(pat, esc);
965                                let compiled = CompiledPattern::compile(&processed, ci)?;
966                                self.cached_like = Some((pat.clone(), ci, Some(esc), compiled));
967                            }
968                            let (_, _, _, ref compiled) = self.cached_like.as_ref().unwrap();
969                            Value::Boolean(compiled.matches(text, ci))
970                        }
971                        (Value::Null(_), _) | (_, Value::Null(_)) => Value::Null(DataType::Boolean),
972                        _ => Value::Boolean(false),
973                    };
974                    self.stack.push(result);
975                    pc += 1;
976                }
977
978                Op::GlobDynamic => {
979                    let pattern_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
980                    let text_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
981                    let result = match (&text_val, &pattern_val) {
982                        (Value::Text(text), Value::Text(pat)) => {
983                            let need_compile = match &self.cached_glob {
984                                Some((cached_pat, _)) => cached_pat.as_str() != pat.as_str(),
985                                None => true,
986                            };
987                            if need_compile {
988                                let compiled = CompiledPattern::compile_glob(pat)?;
989                                self.cached_glob = Some((pat.clone(), compiled));
990                            }
991                            let (_, ref compiled) = self.cached_glob.as_ref().unwrap();
992                            Value::Boolean(compiled.matches(text, false))
993                        }
994                        (Value::Null(_), _) | (_, Value::Null(_)) => Value::Null(DataType::Boolean),
995                        _ => Value::Boolean(false),
996                    };
997                    self.stack.push(result);
998                    pc += 1;
999                }
1000
1001                Op::RegexpDynamic => {
1002                    let pattern_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1003                    let text_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1004                    let result = match (&text_val, &pattern_val) {
1005                        (Value::Text(text), Value::Text(pat)) => {
1006                            let need_compile = match &self.cached_regexp {
1007                                Some((cached_pat, _)) => cached_pat.as_str() != pat.as_str(),
1008                                None => true,
1009                            };
1010                            if need_compile {
1011                                match regex::Regex::new(pat) {
1012                                    Ok(re) => {
1013                                        self.cached_regexp = Some((pat.clone(), re));
1014                                    }
1015                                    Err(e) => {
1016                                        self.cached_regexp = None;
1017                                        // Return error for invalid regex, matching literal
1018                                        // REGEXP behavior where bad patterns fail at compile time
1019                                        return Err(radixdb_core::Error::invalid_argument(
1020                                            format!("Invalid regular expression '{}': {}", pat, e),
1021                                        ));
1022                                    }
1023                                }
1024                            }
1025                            let (_, ref re) = self.cached_regexp.as_ref().unwrap();
1026                            Value::Boolean(re.is_match(text))
1027                        }
1028                        (Value::Null(_), _) | (_, Value::Null(_)) => Value::Null(DataType::Boolean),
1029                        _ => Value::Boolean(false),
1030                    };
1031                    self.stack.push(result);
1032                    pc += 1;
1033                }
1034
1035                // =============================================================
1036                // JSON OPERATIONS
1037                // =============================================================
1038                Op::JsonAccess => {
1039                    let key = self.stack.pop().unwrap_or_else(Value::null_unknown);
1040                    let json_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1041                    let result = self.json_access(&json_val, &key, false);
1042                    self.stack.push(result);
1043                    pc += 1;
1044                }
1045
1046                Op::JsonAccessText => {
1047                    let key = self.stack.pop().unwrap_or_else(Value::null_unknown);
1048                    let json_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1049                    let result = self.json_access(&json_val, &key, true);
1050                    self.stack.push(result);
1051                    pc += 1;
1052                }
1053
1054                // =============================================================
1055                // TIMESTAMP OPERATIONS
1056                // =============================================================
1057                Op::TimestampAddInterval => {
1058                    let interval = self.stack.pop().unwrap_or_else(Value::null_unknown);
1059                    let ts = self.stack.pop().unwrap_or_else(Value::null_unknown);
1060                    let result = self.timestamp_add_interval(&ts, &interval, true)?;
1061                    self.stack.push(result);
1062                    pc += 1;
1063                }
1064
1065                Op::TimestampSubInterval => {
1066                    let interval = self.stack.pop().unwrap_or_else(Value::null_unknown);
1067                    let ts = self.stack.pop().unwrap_or_else(Value::null_unknown);
1068                    let result = self.timestamp_add_interval(&ts, &interval, false)?;
1069                    self.stack.push(result);
1070                    pc += 1;
1071                }
1072
1073                Op::TimestampDiff => {
1074                    let ts2 = self.stack.pop().unwrap_or_else(Value::null_unknown);
1075                    let ts1 = self.stack.pop().unwrap_or_else(Value::null_unknown);
1076                    let result = match (&ts1, &ts2) {
1077                        (Value::Timestamp(t1), Value::Timestamp(t2)) => {
1078                            let duration = t1.signed_duration_since(*t2);
1079                            Value::Text(SmartString::from_string(
1080                                self.format_duration_as_interval(duration),
1081                            ))
1082                        }
1083                        _ if ts1.is_null() || ts2.is_null() => Value::Null(DataType::Text),
1084                        _ => Value::Null(DataType::Text),
1085                    };
1086                    self.stack.push(result);
1087                    pc += 1;
1088                }
1089
1090                Op::TimestampAddDays => {
1091                    let days = self.stack.pop().unwrap_or_else(Value::null_unknown);
1092                    let ts = self.stack.pop().unwrap_or_else(Value::null_unknown);
1093                    let result = match (&ts, &days) {
1094                        (Value::Timestamp(t), Value::Integer(d)) => {
1095                            Value::Timestamp(*t + chrono::Duration::days(*d))
1096                        }
1097                        (Value::Extension(_), Value::Integer(d)) if ts.as_date_days().is_some() => {
1098                            Self::date_add_days(ts.as_date_days().expect("date was checked"), *d)?
1099                        }
1100                        _ if ts.is_null() || days.is_null() => Value::Null(DataType::Timestamp),
1101                        _ => Value::Null(DataType::Timestamp),
1102                    };
1103                    self.stack.push(result);
1104                    pc += 1;
1105                }
1106
1107                Op::TimestampSubDays => {
1108                    let days = self.stack.pop().unwrap_or_else(Value::null_unknown);
1109                    let ts = self.stack.pop().unwrap_or_else(Value::null_unknown);
1110                    let result = match (&ts, &days) {
1111                        (Value::Timestamp(t), Value::Integer(d)) => {
1112                            Value::Timestamp(*t - chrono::Duration::days(*d))
1113                        }
1114                        (Value::Extension(_), Value::Integer(d)) if ts.as_date_days().is_some() => {
1115                            Self::date_add_days(
1116                                ts.as_date_days().expect("date was checked"),
1117                                d.checked_neg().ok_or_else(|| {
1118                                    Error::Type("DATE arithmetic overflow".to_string())
1119                                })?,
1120                            )?
1121                        }
1122                        _ if ts.is_null() || days.is_null() => Value::Null(DataType::Timestamp),
1123                        _ => Value::Null(DataType::Timestamp),
1124                    };
1125                    self.stack.push(result);
1126                    pc += 1;
1127                }
1128
1129                // =============================================================
1130                // SET OPERATIONS
1131                // =============================================================
1132                Op::InSet(set, has_null) => {
1133                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1134                    let result = if v.is_null() {
1135                        Value::Null(DataType::Boolean)
1136                    } else if Self::sql_set_contains(ctx, set, &v)? {
1137                        Value::Boolean(true)
1138                    } else if *has_null {
1139                        Value::Null(DataType::Boolean)
1140                    } else {
1141                        Value::Boolean(false)
1142                    };
1143                    self.stack.push(result);
1144                    pc += 1;
1145                }
1146
1147                Op::NotInSet(set, has_null) => {
1148                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1149                    let result = if v.is_null() {
1150                        Value::Null(DataType::Boolean)
1151                    } else if Self::sql_set_contains(ctx, set, &v)? {
1152                        Value::Boolean(false)
1153                    } else if *has_null {
1154                        Value::Null(DataType::Boolean)
1155                    } else {
1156                        Value::Boolean(true)
1157                    };
1158                    self.stack.push(result);
1159                    pc += 1;
1160                }
1161
1162                Op::Between => {
1163                    let high = self.stack.pop().unwrap_or_else(Value::null_unknown);
1164                    let low = self.stack.pop().unwrap_or_else(Value::null_unknown);
1165                    let val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1166                    let result = Self::sql_between_result(ctx, &val, &low, &high, false)?;
1167                    self.stack.push(result);
1168                    pc += 1;
1169                }
1170
1171                Op::NotBetween => {
1172                    let high = self.stack.pop().unwrap_or_else(Value::null_unknown);
1173                    let low = self.stack.pop().unwrap_or_else(Value::null_unknown);
1174                    let val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1175                    let result = Self::sql_between_result(ctx, &val, &low, &high, true)?;
1176                    self.stack.push(result);
1177                    pc += 1;
1178                }
1179
1180                Op::InTupleSet {
1181                    tuple_size,
1182                    values,
1183                    negated,
1184                } => {
1185                    let tuple_size = *tuple_size as usize;
1186                    let start = self.stack.len().saturating_sub(tuple_size);
1187
1188                    // Reuse args_buffer to avoid allocation
1189                    self.args_buffer.clear();
1190                    self.args_buffer.extend(self.stack.drain(start..));
1191
1192                    // Check if any values are NULL
1193                    let has_null_in_tuple = self.args_buffer.iter().any(|v| v.is_null());
1194
1195                    if has_null_in_tuple {
1196                        // NULL in tuple -> result is NULL
1197                        self.stack.push(Value::Null(DataType::Boolean));
1198                    } else {
1199                        // Check membership
1200                        let mut found = false;
1201                        for tuple in values.iter() {
1202                            if tuple.len() != self.args_buffer.len() {
1203                                continue;
1204                            }
1205                            let mut equal = true;
1206                            for (left, right) in tuple.iter().zip(self.args_buffer.iter()) {
1207                                if !Self::sql_values_equal(ctx, left, right)? {
1208                                    equal = false;
1209                                    break;
1210                                }
1211                            }
1212                            if equal {
1213                                found = true;
1214                                break;
1215                            }
1216                        }
1217
1218                        let result = if *negated { !found } else { found };
1219                        self.stack.push(Value::Boolean(result));
1220                    }
1221                    pc += 1;
1222                }
1223
1224                // =============================================================
1225                // BOOLEAN CHECKS
1226                // =============================================================
1227                Op::IsTrue => {
1228                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1229                    let result = match v {
1230                        Value::Boolean(b) => Value::Boolean(b),
1231                        Value::Null(_) => Value::Boolean(false),
1232                        _ => Value::Boolean(false),
1233                    };
1234                    self.stack.push(result);
1235                    pc += 1;
1236                }
1237
1238                Op::IsNotTrue => {
1239                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1240                    let result = match v {
1241                        Value::Boolean(b) => Value::Boolean(!b),
1242                        Value::Null(_) => Value::Boolean(true),
1243                        _ => Value::Boolean(true),
1244                    };
1245                    self.stack.push(result);
1246                    pc += 1;
1247                }
1248
1249                Op::IsFalse => {
1250                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1251                    let result = match v {
1252                        Value::Boolean(b) => Value::Boolean(!b),
1253                        Value::Null(_) => Value::Boolean(false),
1254                        _ => Value::Boolean(false),
1255                    };
1256                    self.stack.push(result);
1257                    pc += 1;
1258                }
1259
1260                Op::IsNotFalse => {
1261                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1262                    let result = match v {
1263                        Value::Boolean(b) => Value::Boolean(b),
1264                        Value::Null(_) => Value::Boolean(true),
1265                        _ => Value::Boolean(true),
1266                    };
1267                    self.stack.push(result);
1268                    pc += 1;
1269                }
1270
1271                // =============================================================
1272                // FUNCTION CALLS
1273                // =============================================================
1274                Op::CallScalar { func, arg_count } => {
1275                    let arg_count = *arg_count as usize;
1276                    let start = self.stack.len().saturating_sub(arg_count);
1277
1278                    // Reuse args_buffer to avoid allocation
1279                    self.args_buffer.clear();
1280                    self.args_buffer.extend(self.stack.drain(start..));
1281
1282                    func.info().signature.validate_values(&self.args_buffer)?;
1283                    let result = crate::context::with_current_query_cancellation(|cancellation| {
1284                        func.evaluate_with_cancellation(&self.args_buffer, cancellation)
1285                    })?;
1286                    self.stack.push(result);
1287                    pc += 1;
1288                }
1289
1290                Op::CallStored { name, arg_count } => {
1291                    let arg_count = *arg_count as usize;
1292                    let start = self.stack.len().saturating_sub(arg_count);
1293                    self.args_buffer.clear();
1294                    self.args_buffer.extend(self.stack.drain(start..));
1295                    let invoker = ctx.stored_function_invoker.ok_or_else(|| {
1296                        Error::invalid_argument(format!(
1297                            "stored function {name} is unavailable in this execution context"
1298                        ))
1299                    })?;
1300                    let result = Arc::clone(invoker).invoke(name, &self.args_buffer)?;
1301                    self.stack.push(result);
1302                    pc += 1;
1303                }
1304
1305                Op::Coalesce(n) => {
1306                    let n = *n as usize;
1307                    let start = self.stack.len().saturating_sub(n);
1308
1309                    // Find first non-null using slice iteration (no intermediate buffer)
1310                    let result_idx = self.stack[start..]
1311                        .iter()
1312                        .position(|v| !v.is_null())
1313                        .map(|i| start + i);
1314
1315                    let result = if let Some(idx) = result_idx {
1316                        // Swap the result to end, pop it, then truncate the rest
1317                        let last = self.stack.len() - 1;
1318                        self.stack.swap(idx, last);
1319                        let result = self.stack.pop().unwrap_or_else(Value::null_unknown);
1320                        self.stack.truncate(start);
1321                        result
1322                    } else {
1323                        self.stack.truncate(start);
1324                        Value::null_unknown()
1325                    };
1326                    self.stack.push(result);
1327                    pc += 1;
1328                }
1329
1330                Op::NullIf => {
1331                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
1332                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
1333                    let result = if Self::sql_values_equal(ctx, &a, &b)? {
1334                        Value::null_unknown()
1335                    } else {
1336                        a
1337                    };
1338                    self.stack.push(result);
1339                    pc += 1;
1340                }
1341
1342                Op::Greatest(n) => {
1343                    let n = *n as usize;
1344                    let start = self.stack.len().saturating_sub(n);
1345
1346                    // Find max using slice iteration (no intermediate buffer)
1347                    let mut max_idx: Option<usize> = None;
1348                    for (i, v) in self.stack[start..].iter().enumerate() {
1349                        if !v.is_null() {
1350                            match max_idx {
1351                                None => max_idx = Some(start + i),
1352                                Some(mi) => {
1353                                    if matches!(
1354                                        Self::sql_ordering(ctx, v, &self.stack[mi])?,
1355                                        Some(std::cmp::Ordering::Greater)
1356                                    ) {
1357                                        max_idx = Some(start + i);
1358                                    }
1359                                }
1360                            }
1361                        }
1362                    }
1363
1364                    let result = if let Some(idx) = max_idx {
1365                        // Swap the result to end, pop it, then truncate the rest
1366                        let last = self.stack.len() - 1;
1367                        self.stack.swap(idx, last);
1368                        let result = self.stack.pop().unwrap_or_else(Value::null_unknown);
1369                        self.stack.truncate(start);
1370                        result
1371                    } else {
1372                        self.stack.truncate(start);
1373                        Value::null_unknown()
1374                    };
1375                    self.stack.push(result);
1376                    pc += 1;
1377                }
1378
1379                Op::Least(n) => {
1380                    let n = *n as usize;
1381                    let start = self.stack.len().saturating_sub(n);
1382
1383                    // Find min using slice iteration (no intermediate buffer)
1384                    let mut min_idx: Option<usize> = None;
1385                    for (i, v) in self.stack[start..].iter().enumerate() {
1386                        if !v.is_null() {
1387                            match min_idx {
1388                                None => min_idx = Some(start + i),
1389                                Some(mi) => {
1390                                    if matches!(
1391                                        Self::sql_ordering(ctx, v, &self.stack[mi])?,
1392                                        Some(std::cmp::Ordering::Less)
1393                                    ) {
1394                                        min_idx = Some(start + i);
1395                                    }
1396                                }
1397                            }
1398                        }
1399                    }
1400
1401                    let result = if let Some(idx) = min_idx {
1402                        // Swap the result to end, pop it, then truncate the rest
1403                        let last = self.stack.len() - 1;
1404                        self.stack.swap(idx, last);
1405                        let result = self.stack.pop().unwrap_or_else(Value::null_unknown);
1406                        self.stack.truncate(start);
1407                        result
1408                    } else {
1409                        self.stack.truncate(start);
1410                        Value::null_unknown()
1411                    };
1412                    self.stack.push(result);
1413                    pc += 1;
1414                }
1415
1416                // =============================================================
1417                // NATIVE SCALAR FUNCTIONS (direct function pointer call)
1418                // In-place mutation - no pop/push overhead
1419                // =============================================================
1420                Op::NativeFn1(func) => {
1421                    if let Some(v) = self.stack.last_mut() {
1422                        func(v);
1423                    }
1424                    pc += 1;
1425                }
1426
1427                // =============================================================
1428                // TYPE OPERATIONS
1429                // =============================================================
1430                Op::Cast(target_type) => {
1431                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1432                    let result = if v.as_external().is_some() {
1433                        let invoker = ctx.stored_function_invoker.ok_or_else(|| {
1434                            Error::invalid_argument(
1435                                "external type output is unavailable in this execution context",
1436                            )
1437                        })?;
1438                        invoker.external_output(&v, *target_type)?
1439                    } else {
1440                        v.try_coerce_to_type(*target_type)?
1441                    };
1442                    self.stack.push(result);
1443                    pc += 1;
1444                }
1445
1446                Op::CastExternal(type_name) => {
1447                    let value = self.stack.pop().unwrap_or_else(Value::null_unknown);
1448                    let invoker = ctx.stored_function_invoker.ok_or_else(|| {
1449                        Error::invalid_argument(
1450                            "external type input is unavailable in this execution context",
1451                        )
1452                    })?;
1453                    self.stack.push(invoker.external_input(type_name, &value)?);
1454                    pc += 1;
1455                }
1456
1457                Op::TruncateToDate => {
1458                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1459                    let result = match &v {
1460                        Value::Timestamp(t) => {
1461                            use chrono::{Datelike, TimeZone, Utc};
1462                            let truncated = Utc
1463                                .with_ymd_and_hms(t.year(), t.month(), t.day(), 0, 0, 0)
1464                                .single()
1465                                .unwrap_or(*t);
1466                            Value::Timestamp(truncated)
1467                        }
1468                        Value::Text(s) => match radixdb_core::parse_timestamp(s) {
1469                            Ok(t) => {
1470                                use chrono::{Datelike, TimeZone, Utc};
1471                                let truncated = Utc
1472                                    .with_ymd_and_hms(t.year(), t.month(), t.day(), 0, 0, 0)
1473                                    .single()
1474                                    .unwrap_or(t);
1475                                Value::Timestamp(truncated)
1476                            }
1477                            Err(_) => Value::Null(DataType::Timestamp),
1478                        },
1479                        Value::Integer(i) => {
1480                            use chrono::{Datelike, TimeZone, Utc};
1481                            match Utc.timestamp_opt(*i, 0) {
1482                                chrono::LocalResult::Single(t) => {
1483                                    let truncated = Utc
1484                                        .with_ymd_and_hms(t.year(), t.month(), t.day(), 0, 0, 0)
1485                                        .single()
1486                                        .unwrap_or(t);
1487                                    Value::Timestamp(truncated)
1488                                }
1489                                _ => Value::Null(DataType::Timestamp),
1490                            }
1491                        }
1492                        Value::Null(_) => Value::Null(DataType::Timestamp),
1493                        _ => Value::Null(DataType::Timestamp),
1494                    };
1495                    self.stack.push(result);
1496                    pc += 1;
1497                }
1498
1499                // =============================================================
1500                // CASE EXPRESSION
1501                // =============================================================
1502                Op::CaseStart => {
1503                    // Marker only, no operation
1504                    pc += 1;
1505                }
1506
1507                Op::CaseWhen(next_branch) => {
1508                    let cond = self.stack.pop().unwrap_or_else(Value::null_unknown);
1509                    if !Self::to_bool(&cond) {
1510                        pc = *next_branch as usize;
1511                    } else {
1512                        pc += 1;
1513                    }
1514                }
1515
1516                Op::CaseThen(end_pos) => {
1517                    // Result is on stack, jump to end
1518                    pc = *end_pos as usize;
1519                }
1520
1521                Op::CaseElse => {
1522                    // Marker only
1523                    pc += 1;
1524                }
1525
1526                Op::CaseEnd => {
1527                    // Marker only
1528                    pc += 1;
1529                }
1530
1531                Op::CaseCompare => {
1532                    let when_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1533                    let case_val = self.stack.pop().unwrap_or_else(Value::null_unknown);
1534                    let result = if !case_val.is_null() && !when_val.is_null() {
1535                        Value::Boolean(case_val == when_val)
1536                    } else {
1537                        Value::Boolean(false)
1538                    };
1539                    self.stack.push(result);
1540                    pc += 1;
1541                }
1542
1543                // =============================================================
1544                // CONTROL FLOW
1545                // =============================================================
1546                Op::Jump(target) => {
1547                    pc = *target as usize;
1548                }
1549
1550                Op::JumpIfTrue(target) => {
1551                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
1552                    if Self::to_bool(top) {
1553                        pc = *target as usize;
1554                    } else {
1555                        pc += 1;
1556                    }
1557                }
1558
1559                Op::JumpIfFalse(target) => {
1560                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
1561                    if !Self::to_bool(top) {
1562                        pc = *target as usize;
1563                    } else {
1564                        pc += 1;
1565                    }
1566                }
1567
1568                Op::JumpIfNull(target) => {
1569                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
1570                    if top.is_null() {
1571                        pc = *target as usize;
1572                    } else {
1573                        pc += 1;
1574                    }
1575                }
1576
1577                Op::JumpIfNotNull(target) => {
1578                    let top = self.stack.last().unwrap_or(&Value::Null(DataType::Boolean));
1579                    if !top.is_null() {
1580                        pc = *target as usize;
1581                    } else {
1582                        pc += 1;
1583                    }
1584                }
1585
1586                Op::PopJumpIfTrue(target) => {
1587                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1588                    if Self::to_bool(&v) {
1589                        pc = *target as usize;
1590                    } else {
1591                        pc += 1;
1592                    }
1593                }
1594
1595                Op::PopJumpIfFalse(target) => {
1596                    let v = self.stack.pop().unwrap_or_else(Value::null_unknown);
1597                    if !Self::to_bool(&v) {
1598                        pc = *target as usize;
1599                    } else {
1600                        pc += 1;
1601                    }
1602                }
1603
1604                Op::Dup => {
1605                    if let Some(v) = self.stack.last().cloned() {
1606                        self.stack.push(v);
1607                    }
1608                    pc += 1;
1609                }
1610
1611                Op::Pop => {
1612                    // Use truncate instead of pop to drop in-place without copying value out
1613                    let new_len = self.stack.len().saturating_sub(1);
1614                    self.stack.truncate(new_len);
1615                    pc += 1;
1616                }
1617
1618                Op::Swap => {
1619                    let len = self.stack.len();
1620                    if len >= 2 {
1621                        self.stack.swap(len - 1, len - 2);
1622                    }
1623                    pc += 1;
1624                }
1625
1626                // =============================================================
1627                // SPECIAL
1628                // =============================================================
1629                Op::Nop => {
1630                    pc += 1;
1631                }
1632
1633                Op::Return => {
1634                    break;
1635                }
1636
1637                Op::ReturnTrue => {
1638                    self.stack.clear();
1639                    self.stack.push(Value::Boolean(true));
1640                    break;
1641                }
1642
1643                Op::ReturnFalse => {
1644                    self.stack.clear();
1645                    self.stack.push(Value::Boolean(false));
1646                    break;
1647                }
1648
1649                Op::ReturnNull(dt) => {
1650                    self.stack.clear();
1651                    self.stack.push(Value::Null(*dt));
1652                    break;
1653                }
1654
1655                // =============================================================
1656                // VECTOR DISTANCE OPERATIONS
1657                // =============================================================
1658                Op::VectorDistanceL2 => {
1659                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
1660                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
1661                    let mut buf_a = Vec::new();
1662                    let mut buf_b = Vec::new();
1663                    if a.is_null() || b.is_null() {
1664                        self.stack.push(Value::null_unknown());
1665                    } else {
1666                        match (
1667                            extract_vector_bytes(&a, &mut buf_a),
1668                            extract_vector_bytes(&b, &mut buf_b),
1669                        ) {
1670                            (Some(ba), Some(bb)) => {
1671                                self.stack.push(Value::Float(
1672                                    radixdb_functions::scalar::vector::l2_distance_bytes(ba, bb)?,
1673                                ));
1674                            }
1675                            _ => {
1676                                return Err(Error::Type(
1677                                    "vector distance requires two valid VECTOR values".to_string(),
1678                                ));
1679                            }
1680                        }
1681                    }
1682                    pc += 1;
1683                }
1684
1685                Op::VectorDistanceCosine => {
1686                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
1687                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
1688                    let mut buf_a = Vec::new();
1689                    let mut buf_b = Vec::new();
1690                    if a.is_null() || b.is_null() {
1691                        self.stack.push(Value::null_unknown());
1692                    } else {
1693                        match (
1694                            extract_vector_bytes(&a, &mut buf_a),
1695                            extract_vector_bytes(&b, &mut buf_b),
1696                        ) {
1697                            (Some(ba), Some(bb)) => {
1698                                self.stack.push(Value::Float(
1699                                    radixdb_functions::scalar::vector::cosine_distance_bytes(
1700                                        ba, bb,
1701                                    )?,
1702                                ));
1703                            }
1704                            _ => {
1705                                return Err(Error::Type(
1706                                    "vector distance requires two valid VECTOR values".to_string(),
1707                                ));
1708                            }
1709                        }
1710                    }
1711                    pc += 1;
1712                }
1713
1714                Op::VectorDistanceIP => {
1715                    let b = self.stack.pop().unwrap_or_else(Value::null_unknown);
1716                    let a = self.stack.pop().unwrap_or_else(Value::null_unknown);
1717                    let mut buf_a = Vec::new();
1718                    let mut buf_b = Vec::new();
1719                    if a.is_null() || b.is_null() {
1720                        self.stack.push(Value::null_unknown());
1721                    } else {
1722                        match (
1723                            extract_vector_bytes(&a, &mut buf_a),
1724                            extract_vector_bytes(&b, &mut buf_b),
1725                        ) {
1726                            (Some(ba), Some(bb)) => {
1727                                self.stack.push(Value::Float(
1728                                    radixdb_functions::scalar::vector::ip_distance_bytes(ba, bb)?,
1729                                ));
1730                            }
1731                            _ => {
1732                                return Err(Error::Type(
1733                                    "vector distance requires two valid VECTOR values".to_string(),
1734                                ));
1735                            }
1736                        }
1737                    }
1738                    pc += 1;
1739                }
1740            }
1741        }
1742
1743        // Return top of stack or NULL
1744        Ok(self.stack.pop().unwrap_or_else(Value::null_unknown))
1745    }
1746
1747    /// Execute a program using borrowed values where possible (Cow-based stack)
1748    ///
1749    /// This version avoids cloning values from the row when possible.
1750    /// Values are only cloned when they need to be modified or passed to functions.
1751    #[inline]
1752    pub fn execute_cow<'a>(
1753        &mut self,
1754        program: &'a Program,
1755        ctx: &'a ExecuteContext<'a>,
1756    ) -> Result<Value> {
1757        // Choose the interpreter before executing any observable instruction.
1758        // Restarting after a partial Cow prefix would duplicate scalar calls.
1759        if !program.ops().iter().all(Self::cow_supports_op) {
1760            return self.execute(program, ctx);
1761        }
1762
1763        // Local stack with borrowed values - lifetime tied to this execution
1764        let mut stack: SmallVec<[StackValue<'a>; STACK_INLINE_CAPACITY]> = SmallVec::new();
1765
1766        let ops = program.ops();
1767        if ops.is_empty() {
1768            return Ok(NULL_VALUE.clone());
1769        }
1770
1771        let mut pc: usize = 0;
1772
1773        loop {
1774            if pc >= ops.len() {
1775                break;
1776            }
1777
1778            match &ops[pc] {
1779                // LOAD OPERATIONS - borrow instead of clone
1780                Op::LoadColumn(idx) => {
1781                    let idx = *idx as usize;
1782                    let value = ctx
1783                        .row
1784                        .get(idx)
1785                        .map(Cow::Borrowed)
1786                        .unwrap_or_else(|| Cow::Borrowed(&NULL_VALUE));
1787                    stack.push(value);
1788                    pc += 1;
1789                }
1790
1791                Op::LoadColumn2(idx) => {
1792                    let idx = *idx as usize;
1793                    let value = ctx
1794                        .row2
1795                        .and_then(|r| r.get(idx))
1796                        .map(Cow::Borrowed)
1797                        .unwrap_or_else(|| Cow::Borrowed(&NULL_VALUE));
1798                    stack.push(value);
1799                    pc += 1;
1800                }
1801
1802                Op::LoadConst(value) => {
1803                    stack.push(Cow::Borrowed(value));
1804                    pc += 1;
1805                }
1806
1807                Op::LoadParam(idx) => {
1808                    let idx = *idx as usize;
1809                    let value = ctx
1810                        .params
1811                        .get(idx)
1812                        .map(Cow::Borrowed)
1813                        .unwrap_or_else(|| Cow::Borrowed(&NULL_VALUE));
1814                    stack.push(value);
1815                    pc += 1;
1816                }
1817
1818                Op::LoadNull(dt) => {
1819                    stack.push(Cow::Owned(Value::Null(*dt)));
1820                    pc += 1;
1821                }
1822
1823                Op::LoadAggregateResult(idx) => {
1824                    let idx = *idx as usize;
1825                    let value = ctx
1826                        .row
1827                        .get(idx)
1828                        .map(Cow::Borrowed)
1829                        .unwrap_or_else(|| Cow::Borrowed(&NULL_VALUE));
1830                    stack.push(value);
1831                    pc += 1;
1832                }
1833
1834                // COMPARISON OPERATIONS - work with references
1835                Op::Eq => {
1836                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1837                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1838                    let result = Self::sql_equality_result(ctx, a.as_ref(), b.as_ref(), false)?;
1839                    stack.push(Cow::Owned(result));
1840                    pc += 1;
1841                }
1842
1843                Op::Ne => {
1844                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1845                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1846                    let result = Self::sql_equality_result(ctx, a.as_ref(), b.as_ref(), true)?;
1847                    stack.push(Cow::Owned(result));
1848                    pc += 1;
1849                }
1850
1851                Op::Lt => {
1852                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1853                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1854                    let result = Self::sql_order_result(ctx, a.as_ref(), b.as_ref(), |ordering| {
1855                        ordering == std::cmp::Ordering::Less
1856                    })?;
1857                    stack.push(Cow::Owned(result));
1858                    pc += 1;
1859                }
1860
1861                Op::Le => {
1862                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1863                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1864                    let result = Self::sql_order_result(ctx, a.as_ref(), b.as_ref(), |ordering| {
1865                        ordering != std::cmp::Ordering::Greater
1866                    })?;
1867                    stack.push(Cow::Owned(result));
1868                    pc += 1;
1869                }
1870
1871                Op::Gt => {
1872                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1873                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1874                    let result = Self::sql_order_result(ctx, a.as_ref(), b.as_ref(), |ordering| {
1875                        ordering == std::cmp::Ordering::Greater
1876                    })?;
1877                    stack.push(Cow::Owned(result));
1878                    pc += 1;
1879                }
1880
1881                Op::Ge => {
1882                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1883                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1884                    let result = Self::sql_order_result(ctx, a.as_ref(), b.as_ref(), |ordering| {
1885                        ordering != std::cmp::Ordering::Less
1886                    })?;
1887                    stack.push(Cow::Owned(result));
1888                    pc += 1;
1889                }
1890
1891                Op::IsNull => {
1892                    let v = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1893                    stack.push(Cow::Owned(Value::Boolean(v.is_null())));
1894                    pc += 1;
1895                }
1896
1897                Op::IsNotNull => {
1898                    let v = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1899                    stack.push(Cow::Owned(Value::Boolean(!v.is_null())));
1900                    pc += 1;
1901                }
1902
1903                // LOGICAL OPERATIONS
1904                Op::And(jump_target) => {
1905                    let top = stack.last().map(|v| &**v).unwrap_or(&NULL_VALUE);
1906                    match top {
1907                        Value::Boolean(false) => pc = *jump_target as usize,
1908                        Value::Null(_) => pc += 1,
1909                        _ => pc += 1,
1910                    }
1911                }
1912
1913                Op::Or(jump_target) => {
1914                    let top = stack.last().map(|v| &**v).unwrap_or(&NULL_VALUE);
1915                    match top {
1916                        Value::Boolean(true) => pc = *jump_target as usize,
1917                        Value::Null(_) => pc += 1,
1918                        _ => pc += 1,
1919                    }
1920                }
1921
1922                Op::AndFinalize => {
1923                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1924                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1925                    let result = match (Self::to_tribool(&a), Self::to_tribool(&b)) {
1926                        (Some(false), _) | (_, Some(false)) => Value::Boolean(false),
1927                        (Some(true), Some(true)) => Value::Boolean(true),
1928                        _ => Value::Null(DataType::Boolean),
1929                    };
1930                    stack.push(Cow::Owned(result));
1931                    pc += 1;
1932                }
1933
1934                Op::OrFinalize => {
1935                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1936                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1937                    let result = match (Self::to_tribool(&a), Self::to_tribool(&b)) {
1938                        (Some(true), _) | (_, Some(true)) => Value::Boolean(true),
1939                        (Some(false), Some(false)) => Value::Boolean(false),
1940                        _ => Value::Null(DataType::Boolean),
1941                    };
1942                    stack.push(Cow::Owned(result));
1943                    pc += 1;
1944                }
1945
1946                Op::Not => {
1947                    let v = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
1948                    let result = match Self::to_tribool(&v) {
1949                        Some(b) => Value::Boolean(!b),
1950                        None => Value::Null(DataType::Boolean),
1951                    };
1952                    stack.push(Cow::Owned(result));
1953                    pc += 1;
1954                }
1955
1956                // FUNCTION CALLS - need to convert to owned
1957                Op::CallScalar { func, arg_count } => {
1958                    let arg_count = *arg_count as usize;
1959                    let start = stack.len().saturating_sub(arg_count);
1960
1961                    // Convert Cow values to owned for function call
1962                    self.args_buffer.clear();
1963                    for cow_val in stack.drain(start..) {
1964                        self.args_buffer.push(cow_val.into_owned());
1965                    }
1966
1967                    func.info().signature.validate_values(&self.args_buffer)?;
1968                    let result = crate::context::with_current_query_cancellation(|cancellation| {
1969                        func.evaluate_with_cancellation(&self.args_buffer, cancellation)
1970                    })?;
1971                    stack.push(Cow::Owned(result));
1972                    pc += 1;
1973                }
1974
1975                Op::CallStored { name, arg_count } => {
1976                    let arg_count = *arg_count as usize;
1977                    let start = stack.len().saturating_sub(arg_count);
1978                    self.args_buffer.clear();
1979                    for value in stack.drain(start..) {
1980                        self.args_buffer.push(value.into_owned());
1981                    }
1982                    let invoker = ctx.stored_function_invoker.ok_or_else(|| {
1983                        Error::invalid_argument(format!(
1984                            "stored function {name} is unavailable in this execution context"
1985                        ))
1986                    })?;
1987                    let result = Arc::clone(invoker).invoke(name, &self.args_buffer)?;
1988                    stack.push(Cow::Owned(result));
1989                    pc += 1;
1990                }
1991
1992                // FUSED OPERATIONS (already optimized, no stack involvement)
1993                Op::GtColumnConst(idx, val) => {
1994                    let col_val = ctx.row.get(*idx as usize).unwrap_or(&NULL_VALUE);
1995                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
1996                        ordering == std::cmp::Ordering::Greater
1997                    })?;
1998                    stack.push(Cow::Owned(result));
1999                    pc += 1;
2000                }
2001
2002                Op::LtColumnConst(idx, val) => {
2003                    let col_val = ctx.row.get(*idx as usize).unwrap_or(&NULL_VALUE);
2004                    let result = Self::sql_order_result(ctx, col_val, val, |ordering| {
2005                        ordering == std::cmp::Ordering::Less
2006                    })?;
2007                    stack.push(Cow::Owned(result));
2008                    pc += 1;
2009                }
2010
2011                Op::EqColumnConst(idx, val) => {
2012                    let col_val = ctx.row.get(*idx as usize).unwrap_or(&NULL_VALUE);
2013                    let result = Self::sql_equality_result(ctx, col_val, val, false)?;
2014                    stack.push(Cow::Owned(result));
2015                    pc += 1;
2016                }
2017
2018                // COALESCE - return first non-null value
2019                Op::Coalesce(n) => {
2020                    let n = *n as usize;
2021                    let start = stack.len().saturating_sub(n);
2022
2023                    // Find first non-null value index
2024                    let result_idx = stack[start..]
2025                        .iter()
2026                        .position(|v| !v.is_null())
2027                        .map(|i| start + i);
2028
2029                    let result = if let Some(idx) = result_idx {
2030                        // Move the result out, swap to end, pop, then truncate
2031                        let last = stack.len() - 1;
2032                        stack.swap(idx, last);
2033                        // pop() should always succeed since we found an index
2034                        stack.pop().expect("stack underflow in COALESCE")
2035                    } else {
2036                        Cow::Borrowed(&NULL_VALUE)
2037                    };
2038                    stack.truncate(start);
2039                    stack.push(result);
2040                    pc += 1;
2041                }
2042
2043                // NULLIF - return NULL if both args are equal
2044                Op::NullIf => {
2045                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2046                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2047                    let result = if Self::sql_values_equal(ctx, a.as_ref(), b.as_ref())? {
2048                        Cow::Borrowed(&NULL_VALUE)
2049                    } else {
2050                        a
2051                    };
2052                    stack.push(result);
2053                    pc += 1;
2054                }
2055
2056                // GREATEST - return maximum non-null value
2057                Op::Greatest(n) => {
2058                    let n = *n as usize;
2059                    let start = stack.len().saturating_sub(n);
2060
2061                    // Find max value index
2062                    let mut max_idx: Option<usize> = None;
2063                    for (i, v) in stack[start..].iter().enumerate() {
2064                        if !v.is_null() {
2065                            match max_idx {
2066                                None => max_idx = Some(start + i),
2067                                Some(mi) => {
2068                                    if matches!(
2069                                        Self::sql_ordering(ctx, v.as_ref(), stack[mi].as_ref())?,
2070                                        Some(std::cmp::Ordering::Greater)
2071                                    ) {
2072                                        max_idx = Some(start + i);
2073                                    }
2074                                }
2075                            }
2076                        }
2077                    }
2078
2079                    let result = if let Some(idx) = max_idx {
2080                        let last = stack.len() - 1;
2081                        stack.swap(idx, last);
2082                        stack.pop().expect("stack underflow in GREATEST")
2083                    } else {
2084                        Cow::Borrowed(&NULL_VALUE)
2085                    };
2086                    stack.truncate(start);
2087                    stack.push(result);
2088                    pc += 1;
2089                }
2090
2091                // LEAST - return minimum non-null value
2092                Op::Least(n) => {
2093                    let n = *n as usize;
2094                    let start = stack.len().saturating_sub(n);
2095
2096                    // Find min value index
2097                    let mut min_idx: Option<usize> = None;
2098                    for (i, v) in stack[start..].iter().enumerate() {
2099                        if !v.is_null() {
2100                            match min_idx {
2101                                None => min_idx = Some(start + i),
2102                                Some(mi) => {
2103                                    if matches!(
2104                                        Self::sql_ordering(ctx, v.as_ref(), stack[mi].as_ref())?,
2105                                        Some(std::cmp::Ordering::Less)
2106                                    ) {
2107                                        min_idx = Some(start + i);
2108                                    }
2109                                }
2110                            }
2111                        }
2112                    }
2113
2114                    let result = if let Some(idx) = min_idx {
2115                        let last = stack.len() - 1;
2116                        stack.swap(idx, last);
2117                        stack.pop().expect("stack underflow in LEAST")
2118                    } else {
2119                        Cow::Borrowed(&NULL_VALUE)
2120                    };
2121                    stack.truncate(start);
2122                    stack.push(result);
2123                    pc += 1;
2124                }
2125
2126                // JUMP/CONTROL FLOW for short-circuit evaluation (used by COALESCE)
2127                Op::JumpIfNotNull(target) => {
2128                    if let Some(top) = stack.last() {
2129                        if !top.is_null() {
2130                            pc = *target as usize;
2131                            continue;
2132                        }
2133                    }
2134                    pc += 1;
2135                }
2136
2137                Op::Pop => {
2138                    // Use truncate instead of pop to drop in-place without copying value out
2139                    let new_len = stack.len().saturating_sub(1);
2140                    stack.truncate(new_len);
2141                    pc += 1;
2142                }
2143
2144                Op::Jump(target) => {
2145                    pc = *target as usize;
2146                }
2147
2148                Op::JumpIfTrue(target) => {
2149                    if let Some(top) = stack.last() {
2150                        if Self::to_bool(top) {
2151                            pc = *target as usize;
2152                            continue;
2153                        }
2154                    }
2155                    pc += 1;
2156                }
2157
2158                Op::JumpIfFalse(target) => {
2159                    if let Some(top) = stack.last() {
2160                        if !Self::to_bool(top) {
2161                            pc = *target as usize;
2162                            continue;
2163                        }
2164                    } else {
2165                        // Empty stack → treat as NULL → falsy → jump
2166                        pc = *target as usize;
2167                        continue;
2168                    }
2169                    pc += 1;
2170                }
2171
2172                Op::JumpIfNull(target) => {
2173                    if let Some(top) = stack.last() {
2174                        if top.is_null() {
2175                            pc = *target as usize;
2176                            continue;
2177                        }
2178                    }
2179                    pc += 1;
2180                }
2181
2182                Op::PopJumpIfFalse(target) => {
2183                    let v = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2184                    if !Self::to_bool(&v) {
2185                        pc = *target as usize;
2186                    } else {
2187                        pc += 1;
2188                    }
2189                }
2190
2191                Op::PopJumpIfTrue(target) => {
2192                    let v = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2193                    if Self::to_bool(&v) {
2194                        pc = *target as usize;
2195                    } else {
2196                        pc += 1;
2197                    }
2198                }
2199
2200                Op::Nop => {
2201                    pc += 1;
2202                }
2203
2204                // STRING CONCATENATION
2205                Op::Concat => {
2206                    let b = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2207                    let a = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2208                    let result = if a.is_null() || b.is_null() {
2209                        Value::Null(DataType::Text)
2210                    } else {
2211                        // Fast path: both are Text
2212                        match (&*a, &*b) {
2213                            (Value::Text(a_str), Value::Text(b_str)) => {
2214                                // Use optimized concat - handles inline and heap efficiently
2215                                Value::Text(SmartString::concat(a_str, b_str))
2216                            }
2217                            (Value::Text(a_str), _) => {
2218                                // Use Arc to avoid shrink_to_fit
2219                                use std::fmt::Write;
2220                                let mut s = String::with_capacity(a_str.len() + 32);
2221                                s.push_str(a_str);
2222                                let _ = write!(s, "{}", *b);
2223                                Value::Text(SmartString::from_string_shared(s))
2224                            }
2225                            (_, Value::Text(b_str)) => {
2226                                // Use Arc to avoid shrink_to_fit
2227                                use std::fmt::Write;
2228                                let mut s = String::with_capacity(32 + b_str.len());
2229                                let _ = write!(s, "{}", *a);
2230                                s.push_str(b_str);
2231                                Value::Text(SmartString::from_string_shared(s))
2232                            }
2233                            _ => {
2234                                // Use Arc to avoid shrink_to_fit
2235                                use std::fmt::Write;
2236                                let mut s = String::with_capacity(64);
2237                                let _ = write!(s, "{}{}", *a, *b);
2238                                Value::Text(SmartString::from_string_shared(s))
2239                            }
2240                        }
2241                    };
2242                    stack.push(Cow::Owned(result));
2243                    pc += 1;
2244                }
2245
2246                // Multi-value string concatenation (optimized for chained ||)
2247                Op::ConcatN(n) => {
2248                    let n = *n as usize;
2249                    let start = stack.len().saturating_sub(n);
2250
2251                    // Single pass: check for NULL and calculate total length
2252                    let mut total_len = 0usize;
2253                    let mut has_null = false;
2254                    let mut all_text = true;
2255
2256                    for v in &stack[start..] {
2257                        match &**v {
2258                            Value::Null(_) => {
2259                                has_null = true;
2260                                break;
2261                            }
2262                            Value::Text(s) => total_len += s.len(),
2263                            _ => {
2264                                all_text = false;
2265                                total_len += 32;
2266                            }
2267                        }
2268                    }
2269
2270                    if has_null {
2271                        stack.truncate(start);
2272                        stack.push(Cow::Owned(Value::Null(DataType::Text)));
2273                        pc += 1;
2274                        continue;
2275                    }
2276
2277                    // Build result - optimize for inline vs heap
2278                    let result = if all_text && total_len <= 15 {
2279                        // Fast path: build directly into inline SmartString (no heap allocation)
2280                        let mut data = [0u8; 15];
2281                        let mut pos = 0;
2282                        for v in stack.drain(start..) {
2283                            if let Value::Text(text) = &*v {
2284                                let bytes = text.as_bytes();
2285                                data[pos..pos + bytes.len()].copy_from_slice(bytes);
2286                                pos += bytes.len();
2287                            }
2288                        }
2289                        let text = std::str::from_utf8(&data[..total_len])
2290                            .expect("concatenated Text values remain valid UTF-8");
2291                        SmartString::new(text)
2292                    } else if all_text {
2293                        // Heap path: exact capacity, into_boxed_str is O(1) when len == capacity
2294                        let mut s = String::with_capacity(total_len);
2295                        for v in stack.drain(start..) {
2296                            if let Value::Text(text) = &*v {
2297                                s.push_str(text);
2298                            }
2299                        }
2300                        // len == capacity, so into_boxed_str is O(1)
2301                        SmartString::from_string(s)
2302                    } else {
2303                        // Mixed types: capacity is estimate, use Arc to avoid shrink_to_fit
2304                        let mut s = String::with_capacity(total_len);
2305                        for v in stack.drain(start..) {
2306                            match &*v {
2307                                Value::Text(text) => s.push_str(text),
2308                                other => {
2309                                    use std::fmt::Write;
2310                                    let _ = write!(s, "{}", other);
2311                                }
2312                            }
2313                        }
2314                        SmartString::from_string_shared(s)
2315                    };
2316                    stack.push(Cow::Owned(Value::Text(result)));
2317                    pc += 1;
2318                }
2319
2320                // CASE expression operations
2321                Op::CaseStart => {
2322                    // Marker only, no operation
2323                    pc += 1;
2324                }
2325
2326                Op::CaseWhen(next_branch) => {
2327                    let cond = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2328                    if !Self::to_bool(&cond) {
2329                        pc = *next_branch as usize;
2330                    } else {
2331                        pc += 1;
2332                    }
2333                }
2334
2335                Op::CaseThen(end_pos) => {
2336                    // Result is on stack, jump to end
2337                    pc = *end_pos as usize;
2338                }
2339
2340                Op::CaseElse => {
2341                    // Marker only
2342                    pc += 1;
2343                }
2344
2345                Op::CaseEnd => {
2346                    // Marker only
2347                    pc += 1;
2348                }
2349
2350                Op::CaseCompare => {
2351                    let when_val = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2352                    let case_val = stack.pop().unwrap_or(Cow::Borrowed(&NULL_VALUE));
2353                    let result = if !case_val.is_null() && !when_val.is_null() {
2354                        Value::Boolean(*case_val == *when_val)
2355                    } else {
2356                        Value::Boolean(false)
2357                    };
2358                    stack.push(Cow::Owned(result));
2359                    pc += 1;
2360                }
2361
2362                Op::Return => break,
2363                Op::ReturnTrue => {
2364                    return Ok(Value::Boolean(true));
2365                }
2366                Op::ReturnFalse => {
2367                    return Ok(Value::Boolean(false));
2368                }
2369                Op::ReturnNull(dt) => {
2370                    return Ok(Value::Null(*dt));
2371                }
2372
2373                // For any unhandled operation, fall back to the regular execute
2374                _ => {
2375                    return Err(radixdb_core::Error::internal(
2376                        "Cow VM support classification drifted from dispatch",
2377                    ));
2378                }
2379            }
2380        }
2381
2382        // Return top of stack or NULL
2383        Ok(stack
2384            .pop()
2385            .map(Cow::into_owned)
2386            .unwrap_or_else(Value::null_unknown))
2387    }
2388
2389    #[inline]
2390    fn cow_supports_op(op: &Op) -> bool {
2391        matches!(
2392            op,
2393            Op::LoadColumn(_)
2394                | Op::LoadColumn2(_)
2395                | Op::LoadConst(_)
2396                | Op::LoadParam(_)
2397                | Op::LoadNull(_)
2398                | Op::LoadAggregateResult(_)
2399                | Op::Eq
2400                | Op::Ne
2401                | Op::Lt
2402                | Op::Le
2403                | Op::Gt
2404                | Op::Ge
2405                | Op::IsNull
2406                | Op::IsNotNull
2407                | Op::And(_)
2408                | Op::Or(_)
2409                | Op::AndFinalize
2410                | Op::OrFinalize
2411                | Op::Not
2412                | Op::CallScalar { .. }
2413                | Op::CallStored { .. }
2414                | Op::GtColumnConst(_, _)
2415                | Op::LtColumnConst(_, _)
2416                | Op::EqColumnConst(_, _)
2417                | Op::Coalesce(_)
2418                | Op::NullIf
2419                | Op::Greatest(_)
2420                | Op::Least(_)
2421                | Op::JumpIfNotNull(_)
2422                | Op::Pop
2423                | Op::Jump(_)
2424                | Op::JumpIfTrue(_)
2425                | Op::JumpIfFalse(_)
2426                | Op::JumpIfNull(_)
2427                | Op::PopJumpIfFalse(_)
2428                | Op::PopJumpIfTrue(_)
2429                | Op::Nop
2430                | Op::Concat
2431                | Op::ConcatN(_)
2432                | Op::CaseStart
2433                | Op::CaseWhen(_)
2434                | Op::CaseThen(_)
2435                | Op::CaseElse
2436                | Op::CaseEnd
2437                | Op::CaseCompare
2438                | Op::Return
2439                | Op::ReturnTrue
2440                | Op::ReturnFalse
2441                | Op::ReturnNull(_)
2442        )
2443    }
2444
2445    /// Execute and return boolean result (for WHERE clauses)
2446    ///
2447    /// This method is optimized for common filter patterns, avoiding
2448    /// the full VM loop overhead for simple comparisons.
2449    #[inline]
2450    pub fn execute_bool(&mut self, program: &Program, ctx: &ExecuteContext) -> Result<bool> {
2451        self.execute_bool_checked(program, ctx)
2452    }
2453
2454    /// Like execute_bool but returns errors instead of swallowing them.
2455    ///
2456    /// Used by RowFilter::matches_checked to propagate VM errors (e.g. invalid
2457    /// REGEXP patterns) through the query result iterator.
2458    #[inline]
2459    pub fn execute_bool_checked(
2460        &mut self,
2461        program: &Program,
2462        ctx: &ExecuteContext,
2463    ) -> radixdb_core::Result<bool> {
2464        let ops = program.ops();
2465
2466        // Fast path: Single comparison + Return (most common filter)
2467        if ops.len() == 2 && matches!(&ops[1], Op::Return) && Self::is_fast_bool_op(&ops[0]) {
2468            return Self::eval_single_op_bool(&ops[0], ctx);
2469        }
2470
2471        // Fast path: Two comparisons with AND/OR
2472        if ops.len() == 5 {
2473            if Self::is_fast_bool_op(&ops[0])
2474                && Self::is_fast_bool_op(&ops[2])
2475                && matches!(
2476                    (&ops[1], &ops[3], &ops[4]),
2477                    (Op::And(_), Op::AndFinalize, Op::Return)
2478                )
2479            {
2480                let a = Self::eval_single_op_tribool(&ops[0], ctx)?;
2481                if a == Some(false) {
2482                    return Ok(false);
2483                }
2484                let b = Self::eval_single_op_tribool(&ops[2], ctx)?;
2485                return Ok(a == Some(true) && b == Some(true));
2486            }
2487            if Self::is_fast_bool_op(&ops[0])
2488                && Self::is_fast_bool_op(&ops[2])
2489                && matches!(
2490                    (&ops[1], &ops[3], &ops[4]),
2491                    (Op::Or(_), Op::OrFinalize, Op::Return)
2492                )
2493            {
2494                let a = Self::eval_single_op_tribool(&ops[0], ctx)?;
2495                if a == Some(true) {
2496                    return Ok(true);
2497                }
2498                let b = Self::eval_single_op_tribool(&ops[2], ctx)?;
2499                return Ok(a == Some(true) || b == Some(true));
2500            }
2501        }
2502
2503        // General path: full VM execution
2504        match self.execute_cow(program, ctx) {
2505            Ok(Value::Boolean(b)) => Ok(b),
2506            Ok(Value::Integer(i)) => Ok(i != 0),
2507            Ok(Value::Null(_)) => Ok(false),
2508            Ok(value) => Err(Error::Type(format!(
2509                "predicate expression produced {}, expected BOOLEAN, INTEGER, or NULL",
2510                value.data_type()
2511            ))),
2512            Err(e) => Err(e),
2513        }
2514    }
2515
2516    #[inline]
2517    fn is_fast_bool_op(op: &Op) -> bool {
2518        matches!(
2519            op,
2520            Op::GtColumnConst(_, _)
2521                | Op::LtColumnConst(_, _)
2522                | Op::GeColumnConst(_, _)
2523                | Op::LeColumnConst(_, _)
2524                | Op::EqColumnConst(_, _)
2525                | Op::NeColumnConst(_, _)
2526                | Op::IsNullColumn(_)
2527                | Op::IsNotNullColumn(_)
2528                | Op::BetweenColumnConst(_, _, _)
2529                | Op::InSetColumn(_, _, _)
2530                | Op::LoadConst(Value::Boolean(_) | Value::Integer(_) | Value::Null(_))
2531                | Op::LikeColumn(_, _, _)
2532        )
2533    }
2534
2535    /// Evaluate a single comparison op and return bool (for fast path)
2536    #[inline]
2537    fn eval_single_op_bool(op: &Op, ctx: &ExecuteContext) -> Result<bool> {
2538        Ok(match op {
2539            Op::GtColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2540                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2541                    ordering == std::cmp::Ordering::Greater
2542                })?
2543                .unwrap_or(false),
2544                None => false,
2545            },
2546            Op::LtColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2547                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2548                    ordering == std::cmp::Ordering::Less
2549                })?
2550                .unwrap_or(false),
2551                None => false,
2552            },
2553            Op::GeColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2554                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2555                    ordering != std::cmp::Ordering::Less
2556                })?
2557                .unwrap_or(false),
2558                None => false,
2559            },
2560            Op::LeColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2561                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2562                    ordering != std::cmp::Ordering::Greater
2563                })?
2564                .unwrap_or(false),
2565                None => false,
2566            },
2567            Op::EqColumnConst(idx, value) => match ctx.row.get(*idx as usize) {
2568                Some(column) => {
2569                    Self::sql_equality_tribool(ctx, column, value, false)?.unwrap_or(false)
2570                }
2571                None => false,
2572            },
2573            Op::NeColumnConst(idx, value) => match ctx.row.get(*idx as usize) {
2574                Some(column) => {
2575                    Self::sql_equality_tribool(ctx, column, value, true)?.unwrap_or(false)
2576                }
2577                None => false,
2578            },
2579            Op::IsNullColumn(idx) => ctx.row.get(*idx as usize).is_some_and(|v| v.is_null()),
2580            Op::IsNotNullColumn(idx) => ctx.row.get(*idx as usize).is_some_and(|v| !v.is_null()),
2581            Op::BetweenColumnConst(idx, low, high) => match ctx.row.get(*idx as usize) {
2582                Some(col_val) => {
2583                    Self::sql_between_tribool(ctx, col_val, low, high)?.unwrap_or(false)
2584                }
2585                _ => false,
2586            },
2587            Op::InSetColumn(idx, set, has_null) => {
2588                match ctx.row.get(*idx as usize) {
2589                    Some(v) if v.is_null() => false, // NULL IN set -> NULL -> false in bool context
2590                    Some(v) => {
2591                        let mut found = false;
2592                        for candidate in set.iter() {
2593                            if Self::sql_values_equal(ctx, v, candidate)? {
2594                                found = true;
2595                                break;
2596                            }
2597                        }
2598                        let _ = has_null;
2599                        found
2600                    }
2601                    None => false,
2602                }
2603            }
2604            // Handle LIKE pattern matching (e.g., fruit LIKE 'a%')
2605            Op::LikeColumn(idx, pattern, case_insensitive) => {
2606                match ctx.row.get(*idx as usize) {
2607                    Some(Value::Text(s)) => pattern.matches(s, *case_insensitive),
2608                    _ => false, // NULL or non-text -> false
2609                }
2610            }
2611            // For other ops, fall back to tribool and convert
2612            _ => Self::eval_single_op_tribool(op, ctx)? == Some(true),
2613        })
2614    }
2615
2616    /// Evaluate a single comparison op and return Option<bool> (tribool)
2617    /// None = NULL, Some(true) = true, Some(false) = false
2618    #[inline]
2619    fn eval_single_op_tribool(op: &Op, ctx: &ExecuteContext) -> Result<Option<bool>> {
2620        Ok(match op {
2621            Op::GtColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2622                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2623                    ordering == std::cmp::Ordering::Greater
2624                })?,
2625                None => None,
2626            },
2627            Op::LtColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2628                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2629                    ordering == std::cmp::Ordering::Less
2630                })?,
2631                None => None,
2632            },
2633            Op::GeColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2634                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2635                    ordering != std::cmp::Ordering::Less
2636                })?,
2637                None => None,
2638            },
2639            Op::LeColumnConst(idx, threshold) => match ctx.row.get(*idx as usize) {
2640                Some(value) => Self::sql_order_tribool(ctx, value, threshold, |ordering| {
2641                    ordering != std::cmp::Ordering::Greater
2642                })?,
2643                None => None,
2644            },
2645            Op::EqColumnConst(idx, value) => match ctx.row.get(*idx as usize) {
2646                Some(column) => Self::sql_equality_tribool(ctx, column, value, false)?,
2647                None => None,
2648            },
2649            Op::NeColumnConst(idx, value) => match ctx.row.get(*idx as usize) {
2650                Some(column) => Self::sql_equality_tribool(ctx, column, value, true)?,
2651                None => None,
2652            },
2653            Op::IsNullColumn(idx) => Some(ctx.row.get(*idx as usize).is_some_and(|v| v.is_null())),
2654            Op::IsNotNullColumn(idx) => {
2655                Some(ctx.row.get(*idx as usize).is_some_and(|v| !v.is_null()))
2656            }
2657            Op::BetweenColumnConst(idx, low, high) => match ctx.row.get(*idx as usize) {
2658                Some(col_val) => Self::sql_between_tribool(ctx, col_val, low, high)?,
2659                None => None,
2660            },
2661            Op::InSetColumn(idx, set, has_null) => {
2662                match ctx.row.get(*idx as usize) {
2663                    Some(v) if v.is_null() => None, // NULL IN set -> NULL
2664                    Some(v) => {
2665                        let mut found = false;
2666                        for candidate in set.iter() {
2667                            if Self::sql_values_equal(ctx, v, candidate)? {
2668                                found = true;
2669                                break;
2670                            }
2671                        }
2672                        if found {
2673                            Some(true)
2674                        } else if *has_null {
2675                            None
2676                        } else {
2677                            Some(false)
2678                        }
2679                    }
2680                    None => None,
2681                }
2682            }
2683            // Handle boolean constants (from processed subqueries like NOT EXISTS)
2684            Op::LoadConst(Value::Boolean(b)) => Some(*b),
2685            Op::LoadConst(Value::Integer(i)) => Some(*i != 0),
2686            Op::LoadConst(Value::Null(_)) => None,
2687            // Handle LIKE pattern matching (e.g., fruit LIKE 'a%')
2688            Op::LikeColumn(idx, pattern, case_insensitive) => match ctx.row.get(*idx as usize) {
2689                Some(Value::Text(s)) => Some(pattern.matches(s, *case_insensitive)),
2690                Some(Value::Null(_)) | None => None,
2691                _ => Some(false),
2692            },
2693            // For other comparisons, return None to fall back to full VM
2694            _ => None,
2695        })
2696    }
2697
2698    // =========================================================================
2699    // HELPER METHODS
2700    // =========================================================================
2701
2702    #[inline]
2703    fn to_bool(v: &Value) -> bool {
2704        match v {
2705            Value::Boolean(b) => *b,
2706            Value::Integer(i) => *i != 0,
2707            Value::Null(_) => false,
2708            _ => true, // Non-null values are truthy
2709        }
2710    }
2711
2712    #[inline]
2713    fn to_tribool(v: &Value) -> Option<bool> {
2714        match v {
2715            Value::Boolean(b) => Some(*b),
2716            Value::Integer(i) => Some(*i != 0),
2717            Value::Null(_) => None,
2718            _ => Some(true),
2719        }
2720    }
2721
2722    /// SQL equality uses plugin-owned semantics for external types. Structural
2723    /// `Value::Eq` is never a fallback for an external payload.
2724    #[inline]
2725    fn sql_values_equal(ctx: &ExecuteContext<'_>, a: &Value, b: &Value) -> Result<bool> {
2726        if a.is_external() || b.is_external() {
2727            let invoker = ctx.stored_function_invoker.ok_or_else(|| {
2728                Error::NotSupported(
2729                    "external equality is unavailable in this execution context".to_owned(),
2730                )
2731            })?;
2732            return invoker.external_equal(a, b);
2733        }
2734        match a
2735            .compare(b)
2736            .ok()
2737            .or_else(|| Self::sql_literal_ordering(a, b))
2738        {
2739            Some(std::cmp::Ordering::Equal) => Ok(true),
2740            Some(_) => Ok(false),
2741            None => Ok(a == b),
2742        }
2743    }
2744
2745    #[inline]
2746    fn sql_set_contains(
2747        ctx: &ExecuteContext<'_>,
2748        set: &radixdb_core::ValueSet,
2749        value: &Value,
2750    ) -> Result<bool> {
2751        if !value.is_external() {
2752            return Ok(set.contains(value));
2753        }
2754        for candidate in set.iter() {
2755            if Self::sql_values_equal(ctx, value, candidate)? {
2756                return Ok(true);
2757            }
2758        }
2759        Ok(false)
2760    }
2761
2762    /// Canonical SQL comparison result shared by the stack, fused and
2763    /// bool/tribool execution paths. NULL remains UNKNOWN; non-NULL numeric
2764    /// values use `Value::compare`, which does not round an i64 through f64.
2765    #[inline]
2766    fn sql_ordering(
2767        ctx: &ExecuteContext<'_>,
2768        a: &Value,
2769        b: &Value,
2770    ) -> Result<Option<std::cmp::Ordering>> {
2771        if a.is_null() || b.is_null() {
2772            return Ok(None);
2773        }
2774        if a.is_external() || b.is_external() {
2775            let invoker = ctx.stored_function_invoker.ok_or_else(|| {
2776                Error::NotSupported(
2777                    "external ordering is unavailable in this execution context".to_owned(),
2778                )
2779            })?;
2780            return invoker.external_compare(a, b).map(Some);
2781        }
2782        Ok(a.compare(b)
2783            .ok()
2784            .or_else(|| Self::sql_literal_ordering(a, b)))
2785    }
2786
2787    /// SQL literals may be compared with typed UUID/TIMESTAMP values before a
2788    /// schema-aware pushdown can bind them (TVFs and post-join residuals are
2789    /// the two important cases). Keep those explicit admissions here instead
2790    /// of making structural `Value` identity perform general string coercion.
2791    #[inline]
2792    fn sql_literal_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
2793        match (a, b) {
2794            (Value::Extension(_), Value::Text(text)) => a
2795                .as_uuid_bytes()
2796                .zip(radixdb_core::value::parse_uuid_str(text))
2797                .map(|(left, right)| left.cmp(&right)),
2798            (Value::Text(text), Value::Extension(_)) => radixdb_core::value::parse_uuid_str(text)
2799                .zip(b.as_uuid_bytes())
2800                .map(|(left, right)| left.cmp(&right)),
2801            (Value::Timestamp(left), Value::Text(text)) => {
2802                radixdb_core::value::parse_timestamp(text)
2803                    .ok()
2804                    .map(|right| left.cmp(&right))
2805            }
2806            (Value::Text(text), Value::Timestamp(right)) => {
2807                radixdb_core::value::parse_timestamp(text)
2808                    .ok()
2809                    .map(|left| left.cmp(right))
2810            }
2811            _ => None,
2812        }
2813    }
2814
2815    #[inline]
2816    fn sql_equality_result(
2817        ctx: &ExecuteContext<'_>,
2818        a: &Value,
2819        b: &Value,
2820        negated: bool,
2821    ) -> Result<Value> {
2822        if a.is_null() || b.is_null() {
2823            Ok(Value::Null(DataType::Boolean))
2824        } else {
2825            Ok(Value::Boolean(
2826                Self::sql_values_equal(ctx, a, b)? != negated,
2827            ))
2828        }
2829    }
2830
2831    #[inline]
2832    fn sql_order_result(
2833        ctx: &ExecuteContext<'_>,
2834        a: &Value,
2835        b: &Value,
2836        predicate: impl FnOnce(std::cmp::Ordering) -> bool,
2837    ) -> Result<Value> {
2838        match Self::sql_ordering(ctx, a, b)? {
2839            Some(ordering) => Ok(Value::Boolean(predicate(ordering))),
2840            None => Ok(Value::Null(DataType::Boolean)),
2841        }
2842    }
2843
2844    #[inline]
2845    fn sql_order_tribool(
2846        ctx: &ExecuteContext<'_>,
2847        a: &Value,
2848        b: &Value,
2849        predicate: impl FnOnce(std::cmp::Ordering) -> bool,
2850    ) -> Result<Option<bool>> {
2851        Ok(Self::sql_ordering(ctx, a, b)?.map(predicate))
2852    }
2853
2854    #[inline]
2855    fn sql_equality_tribool(
2856        ctx: &ExecuteContext<'_>,
2857        a: &Value,
2858        b: &Value,
2859        negated: bool,
2860    ) -> Result<Option<bool>> {
2861        if a.is_null() || b.is_null() {
2862            Ok(None)
2863        } else {
2864            Ok(Some(Self::sql_values_equal(ctx, a, b)? != negated))
2865        }
2866    }
2867
2868    #[inline]
2869    fn sql_between_result(
2870        ctx: &ExecuteContext<'_>,
2871        value: &Value,
2872        low: &Value,
2873        high: &Value,
2874        negated: bool,
2875    ) -> Result<Value> {
2876        let Some(ge_low) = Self::sql_order_tribool(ctx, value, low, |ordering| {
2877            ordering != std::cmp::Ordering::Less
2878        })?
2879        else {
2880            return Ok(Value::Null(DataType::Boolean));
2881        };
2882        let Some(le_high) = Self::sql_order_tribool(ctx, value, high, |ordering| {
2883            ordering != std::cmp::Ordering::Greater
2884        })?
2885        else {
2886            return Ok(Value::Null(DataType::Boolean));
2887        };
2888        Ok(Value::Boolean((ge_low && le_high) != negated))
2889    }
2890
2891    #[inline]
2892    fn sql_between_tribool(
2893        ctx: &ExecuteContext<'_>,
2894        value: &Value,
2895        low: &Value,
2896        high: &Value,
2897    ) -> Result<Option<bool>> {
2898        let Some(ge_low) = Self::sql_order_tribool(ctx, value, low, |ordering| {
2899            ordering != std::cmp::Ordering::Less
2900        })?
2901        else {
2902            return Ok(None);
2903        };
2904        let Some(le_high) = Self::sql_order_tribool(ctx, value, high, |ordering| {
2905            ordering != std::cmp::Ordering::Greater
2906        })?
2907        else {
2908            return Ok(None);
2909        };
2910        Ok(Some(ge_low && le_high))
2911    }
2912
2913    #[inline]
2914    fn date_add_days(days_since_epoch: i32, delta_days: i64) -> Result<Value> {
2915        let result = i64::from(days_since_epoch)
2916            .checked_add(delta_days)
2917            .and_then(|days| i32::try_from(days).ok())
2918            .ok_or_else(|| Error::Type("DATE arithmetic overflow".to_string()))?;
2919        Ok(Value::date(result))
2920    }
2921
2922    #[inline]
2923    fn arithmetic_op<FF>(a: &Value, b: &Value, int_op: ArithmeticOp, float_op: FF) -> Result<Value>
2924    where
2925        FF: Fn(f64, f64) -> f64,
2926    {
2927        match (a, b) {
2928            (Value::Integer(x), Value::Integer(y)) => {
2929                // Use checked operations to detect overflow and return an error
2930                let result = match int_op {
2931                    ArithmeticOp::Add => x.checked_add(*y),
2932                    ArithmeticOp::Sub => x.checked_sub(*y),
2933                    ArithmeticOp::Mul => x.checked_mul(*y),
2934                    ArithmeticOp::Div => {
2935                        if *y == 0 {
2936                            return Ok(Value::Null(DataType::Integer));
2937                        }
2938                        x.checked_div(*y)
2939                    }
2940                    ArithmeticOp::Mod => {
2941                        if *y == 0 {
2942                            return Ok(Value::Null(DataType::Integer));
2943                        }
2944                        x.checked_rem(*y)
2945                    }
2946                };
2947                match result {
2948                    Some(r) => Ok(Value::Integer(r)),
2949                    None => Err(radixdb_core::Error::Type(format!(
2950                        "Integer overflow in arithmetic operation: {} and {}",
2951                        x, y
2952                    ))),
2953                }
2954            }
2955            (Value::Float(x), Value::Float(y)) => Ok(Value::Float(float_op(*x, *y))),
2956            (Value::Integer(x), Value::Float(y)) => Ok(Value::Float(float_op(*x as f64, *y))),
2957            (Value::Float(x), Value::Integer(y)) => Ok(Value::Float(float_op(*x, *y as f64))),
2958            _ if a.is_null() || b.is_null() => Ok(Value::Null(DataType::Float)),
2959            _ => Ok(Value::Null(DataType::Null)),
2960        }
2961    }
2962
2963    #[inline]
2964    fn div_op(a: &Value, b: &Value) -> radixdb_core::Result<Value> {
2965        match (a, b) {
2966            (Value::Integer(x), Value::Integer(y)) if *y != 0 => x
2967                .checked_div(*y)
2968                .map(Value::Integer)
2969                .ok_or_else(|| radixdb_core::Error::Type("integer division overflow".to_string())),
2970            (Value::Float(x), Value::Float(y)) if *y != 0.0 => Ok(Value::Float(x / y)),
2971            (Value::Integer(x), Value::Float(y)) if *y != 0.0 => Ok(Value::Float(*x as f64 / y)),
2972            (Value::Float(x), Value::Integer(y)) if *y != 0 => Ok(Value::Float(x / *y as f64)),
2973            _ if a.is_null() || b.is_null() => Ok(Value::Null(DataType::Float)),
2974            _ => Ok(Value::Null(DataType::Null)),
2975        }
2976    }
2977
2978    #[inline]
2979    fn mod_op(a: &Value, b: &Value) -> radixdb_core::Result<Value> {
2980        match (a, b) {
2981            (Value::Integer(x), Value::Integer(y)) if *y != 0 => x
2982                .checked_rem(*y)
2983                .map(Value::Integer)
2984                .ok_or_else(|| radixdb_core::Error::Type("integer remainder overflow".to_string())),
2985            (Value::Float(x), Value::Float(y)) if *y != 0.0 => Ok(Value::Float(x % y)),
2986            (Value::Integer(x), Value::Float(y)) if *y != 0.0 => Ok(Value::Float(*x as f64 % y)),
2987            (Value::Float(x), Value::Integer(y)) if *y != 0 => Ok(Value::Float(x % *y as f64)),
2988            _ if a.is_null() || b.is_null() => Ok(Value::Null(DataType::Float)),
2989            _ => Ok(Value::Null(DataType::Null)),
2990        }
2991    }
2992
2993    /// JSON access helper
2994    /// If as_text is true, returns TEXT; otherwise returns JSON
2995    fn json_access(&self, json_val: &Value, key: &Value, as_text: bool) -> Value {
2996        use serde_json;
2997
2998        // Get the JSON string
2999        let json_str = match json_val {
3000            Value::Extension(data) if data.first() == Some(&(DataType::Json as u8)) => {
3001                std::str::from_utf8(&data[1..]).unwrap_or("")
3002            }
3003            Value::Text(s) => s.as_ref(),
3004            Value::Null(_) => {
3005                return Value::Null(if as_text {
3006                    DataType::Text
3007                } else {
3008                    DataType::Json
3009                })
3010            }
3011            _ => {
3012                return Value::Null(if as_text {
3013                    DataType::Text
3014                } else {
3015                    DataType::Json
3016                })
3017            }
3018        };
3019
3020        // Parse the JSON
3021        let parsed: serde_json::Value = match serde_json::from_str(json_str) {
3022            Ok(v) => v,
3023            Err(_) => {
3024                return Value::Null(if as_text {
3025                    DataType::Text
3026                } else {
3027                    DataType::Json
3028                })
3029            }
3030        };
3031
3032        // Access by key or index
3033        let result = match key {
3034            Value::Text(k) => parsed.get(k.as_str()),
3035            Value::Integer(i) => {
3036                if *i >= 0 {
3037                    parsed.get(*i as usize)
3038                } else {
3039                    None
3040                }
3041            }
3042            _ => None,
3043        };
3044
3045        match result {
3046            Some(v) => {
3047                if as_text {
3048                    // ->> returns text
3049                    match v {
3050                        serde_json::Value::String(s) => Value::Text(SmartString::new(s)),
3051                        serde_json::Value::Null => Value::Null(DataType::Text),
3052                        other => Value::Text(SmartString::from_string(other.to_string())),
3053                    }
3054                } else {
3055                    // -> returns JSON
3056                    Value::json(v.to_string())
3057                }
3058            }
3059            None => Value::Null(if as_text {
3060                DataType::Text
3061            } else {
3062                DataType::Json
3063            }),
3064        }
3065    }
3066
3067    /// Add or subtract interval from timestamp
3068    fn timestamp_add_days(timestamp: chrono::DateTime<chrono::Utc>, days: i64) -> Result<Value> {
3069        let duration = chrono::Duration::try_days(days)
3070            .ok_or_else(|| Error::Type("timestamp day interval overflow".to_string()))?;
3071        timestamp
3072            .checked_add_signed(duration)
3073            .map(Value::Timestamp)
3074            .ok_or_else(|| Error::Type("timestamp result is out of range".to_string()))
3075    }
3076
3077    fn timestamp_add_interval(&self, ts: &Value, interval: &Value, add: bool) -> Result<Value> {
3078        let timestamp = match ts {
3079            Value::Timestamp(t) => *t,
3080            Value::Null(_) => return Ok(Value::Null(DataType::Timestamp)),
3081            _ => return Ok(Value::Null(DataType::Timestamp)),
3082        };
3083
3084        let interval_str = match interval {
3085            Value::Text(s) => s.as_ref(),
3086            Value::Null(_) => return Ok(Value::Null(DataType::Timestamp)),
3087            _ => return Ok(Value::Null(DataType::Timestamp)),
3088        };
3089
3090        // Parse interval string
3091        // Formats: "1 day", "2 hours", "30 minutes", "1 year", "1 month", etc.
3092        match self.parse_interval(interval_str)? {
3093            IntervalValue::Duration(duration) => {
3094                let duration = if add {
3095                    duration
3096                } else {
3097                    duration
3098                        .checked_mul(-1)
3099                        .ok_or_else(|| Error::Type("fixed interval overflow".to_string()))?
3100                };
3101                timestamp
3102                    .checked_add_signed(duration)
3103                    .map(Value::Timestamp)
3104                    .ok_or_else(|| Error::Type("timestamp result is out of range".to_string()))
3105            }
3106            IntervalValue::Months(months) => {
3107                let months = if add {
3108                    months
3109                } else {
3110                    months
3111                        .checked_neg()
3112                        .ok_or_else(|| Error::Type("calendar interval overflow".to_string()))?
3113                };
3114                Self::calendar_add_months(timestamp, months)
3115                    .map(Value::Timestamp)
3116                    .ok_or_else(|| Error::Type("timestamp result is out of range".to_string()))
3117            }
3118        }
3119    }
3120
3121    /// Calendar-aware month addition preserving time-of-day and nanoseconds.
3122    fn calendar_add_months(
3123        ts: chrono::DateTime<chrono::Utc>,
3124        months: i64,
3125    ) -> Option<chrono::DateTime<chrono::Utc>> {
3126        use chrono::{Datelike, NaiveDate, Timelike};
3127
3128        let total_months = (ts.year() as i64)
3129            .checked_mul(12)?
3130            .checked_add(i64::from(ts.month()) - 1)?
3131            .checked_add(months)?;
3132        let new_year_i64 = total_months.div_euclid(12);
3133        let new_month = (total_months.rem_euclid(12) + 1) as u32;
3134
3135        let new_year = i32::try_from(new_year_i64).ok()?;
3136        if !(1..=9999).contains(&new_year) {
3137            return None;
3138        }
3139
3140        // Clamp day to valid range for the new month
3141        let max_day = match new_month {
3142            1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
3143            4 | 6 | 9 | 11 => 30,
3144            2 => {
3145                if (new_year % 4 == 0 && new_year % 100 != 0) || (new_year % 400 == 0) {
3146                    29
3147                } else {
3148                    28
3149                }
3150            }
3151            _ => 30,
3152        };
3153        let day = ts.day().min(max_day);
3154
3155        // Rebuild date preserving original time including nanoseconds
3156        let date = NaiveDate::from_ymd_opt(new_year, new_month, day)?;
3157        let time = ts.time();
3158        let naive = date.and_hms_nano_opt(
3159            time.hour(),
3160            time.minute(),
3161            time.second(),
3162            ts.timestamp_subsec_nanos(),
3163        )?;
3164        Some(chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(
3165            naive,
3166            chrono::Utc,
3167        ))
3168    }
3169
3170    /// Parsed interval: either a fixed duration or a calendar-relative month count.
3171    fn parse_interval(&self, s: &str) -> Result<IntervalValue> {
3172        let s = s.trim();
3173        let parts: Vec<&str> = s.split_whitespace().collect();
3174
3175        if parts.len() < 2 {
3176            // Try parsing as just a number (days)
3177            if let Ok(n) = s.parse::<i64>() {
3178                return chrono::Duration::try_days(n)
3179                    .map(IntervalValue::Duration)
3180                    .ok_or_else(|| Error::Type("interval is out of range".to_string()));
3181            }
3182            return Err(Error::Type(format!("invalid interval: {s}")));
3183        }
3184
3185        let value: i64 = parts[0]
3186            .parse()
3187            .map_err(|_| Error::Type(format!("invalid interval: {s}")))?;
3188        let unit = parts[1];
3189
3190        // Case-insensitive unit matching without allocation
3191        // Handle both singular and plural forms
3192        if unit.eq_ignore_ascii_case("year") || unit.eq_ignore_ascii_case("years") {
3193            value
3194                .checked_mul(12)
3195                .map(IntervalValue::Months)
3196                .ok_or_else(|| Error::Type("calendar interval overflow".to_string()))
3197        } else if unit.eq_ignore_ascii_case("month") || unit.eq_ignore_ascii_case("months") {
3198            Ok(IntervalValue::Months(value))
3199        } else if unit.eq_ignore_ascii_case("week") || unit.eq_ignore_ascii_case("weeks") {
3200            chrono::Duration::try_weeks(value)
3201                .map(IntervalValue::Duration)
3202                .ok_or_else(|| Error::Type("interval is out of range".to_string()))
3203        } else if unit.eq_ignore_ascii_case("day") || unit.eq_ignore_ascii_case("days") {
3204            chrono::Duration::try_days(value)
3205                .map(IntervalValue::Duration)
3206                .ok_or_else(|| Error::Type("interval is out of range".to_string()))
3207        } else if unit.eq_ignore_ascii_case("hour") || unit.eq_ignore_ascii_case("hours") {
3208            chrono::Duration::try_hours(value)
3209                .map(IntervalValue::Duration)
3210                .ok_or_else(|| Error::Type("interval is out of range".to_string()))
3211        } else if unit.eq_ignore_ascii_case("minute")
3212            || unit.eq_ignore_ascii_case("minutes")
3213            || unit.eq_ignore_ascii_case("min")
3214        {
3215            chrono::Duration::try_minutes(value)
3216                .map(IntervalValue::Duration)
3217                .ok_or_else(|| Error::Type("interval is out of range".to_string()))
3218        } else if unit.eq_ignore_ascii_case("second")
3219            || unit.eq_ignore_ascii_case("seconds")
3220            || unit.eq_ignore_ascii_case("sec")
3221        {
3222            chrono::Duration::try_seconds(value)
3223                .map(IntervalValue::Duration)
3224                .ok_or_else(|| Error::Type("interval is out of range".to_string()))
3225        } else if unit.eq_ignore_ascii_case("millisecond")
3226            || unit.eq_ignore_ascii_case("milliseconds")
3227            || unit.eq_ignore_ascii_case("ms")
3228        {
3229            Ok(IntervalValue::Duration(chrono::Duration::milliseconds(
3230                value,
3231            )))
3232        } else if unit.eq_ignore_ascii_case("microsecond")
3233            || unit.eq_ignore_ascii_case("microseconds")
3234            || unit.eq_ignore_ascii_case("us")
3235        {
3236            Ok(IntervalValue::Duration(chrono::Duration::microseconds(
3237                value,
3238            )))
3239        } else {
3240            Err(Error::Type(format!("invalid interval unit: {unit}")))
3241        }
3242    }
3243
3244    /// Format chrono Duration as interval string
3245    fn format_duration_as_interval(&self, duration: chrono::TimeDelta) -> String {
3246        let total_seconds = duration.num_seconds();
3247        let abs_seconds = total_seconds.abs();
3248
3249        let days = abs_seconds / 86400;
3250        let hours = (abs_seconds % 86400) / 3600;
3251        let minutes = (abs_seconds % 3600) / 60;
3252        let seconds = abs_seconds % 60;
3253
3254        let sign = if total_seconds < 0 { "-" } else { "" };
3255
3256        if days > 0 {
3257            format!(
3258                "{}{} days {:02}:{:02}:{:02}",
3259                sign, days, hours, minutes, seconds
3260            )
3261        } else {
3262            format!("{}{:02}:{:02}:{:02}", sign, hours, minutes, seconds)
3263        }
3264    }
3265}
3266
3267impl Default for ExprVM {
3268    fn default() -> Self {
3269        Self::new()
3270    }
3271}
3272
3273/// Extract raw LE f32 bytes from a vector Value, zero-copy for Extension.
3274/// Process a LIKE pattern with a custom escape character at runtime.
3275///
3276/// Converts escaped wildcards (e.g. `!%` with escape `!`) into the
3277/// default `\%` escape form that `CompiledPattern::compile` understands.
3278fn process_like_escape_runtime(pattern: &str, escape: char) -> String {
3279    let mut result = String::with_capacity(pattern.len());
3280    let mut chars = pattern.chars().peekable();
3281
3282    while let Some(c) = chars.next() {
3283        if c == escape {
3284            if let Some(&next) = chars.peek() {
3285                if next == '%' || next == '_' || next == escape {
3286                    // Convert to default escape form: \% or \_ or \\
3287                    result.push('\\');
3288                    result.push(chars.next().unwrap());
3289                } else {
3290                    result.push(c);
3291                }
3292            } else {
3293                result.push(c);
3294            }
3295        } else {
3296            result.push(c);
3297        }
3298    }
3299
3300    result
3301}
3302
3303/// For Text values, parses and writes into `buf` as fallback.
3304#[inline]
3305fn extract_vector_bytes<'a>(v: &'a Value, buf: &'a mut Vec<u8>) -> Option<&'a [u8]> {
3306    match v {
3307        Value::Extension(data) if data.first() == Some(&(DataType::Vector as u8)) => {
3308            Some(&data[1..])
3309        }
3310        Value::Text(s) => {
3311            let floats = radixdb_core::value::parse_vector_str(s.as_ref())?;
3312            buf.clear();
3313            buf.reserve(floats.len() * 4);
3314            for f in &floats {
3315                buf.extend_from_slice(&f.to_le_bytes());
3316            }
3317            Some(buf.as_slice())
3318        }
3319        _ => None,
3320    }
3321}
3322
3323#[cfg(test)]
3324mod tests;