// Copyright 2026 PingCAP, Inc.

use std::{
    cell::Cell,
    cmp::Ordering,
    collections::{BTreeMap, HashSet},
    convert::{TryFrom, TryInto},
    io::Cursor,
    str::FromStr,
    time::Instant,
};

use api_version::{ApiV1, ApiV2};
use async_trait::async_trait;
use bincode::Options;
use chrono::{Datelike, TimeZone, Timelike, Utc};
use kvproto::{
    coprocessor::{KeyRange, Response},
    db9_coprocessor::{
        Db9BinaryOp, Db9ColumnRefExpr, Db9DagRequest, Db9Expr, Db9Expr_oneof_expr, Db9FuncCallExpr,
        Db9IndexInfo, Db9NamedExpr, Db9Null, Db9Row, Db9ScanKind, Db9SelectResponse, Db9TableInfo,
        Db9Type, Db9TypeKind, Db9UnaryExpr, Db9UnaryOp, Db9Value,
    },
    kvrpcpb,
};
use protobuf::Message;
use serde::Deserialize;
use serde_json::value::RawValue;
use tidb_query_common::{
    execute_stats::ExecSummary,
    storage::{
        FindRegionResult, Range, RegionStorageAccessor, StateRole, StubAccessor,
        scanner::{RangesScanner, RangesScannerOptions},
    },
    util::convert_to_prefix_next,
};
use tikv_alloc::trace::MemoryTraceGuard;
use txn_types::Key;

mod functions;

#[cfg(test)]
mod tests;

#[cfg(test)]
pub(crate) use tests::{
    TestStoredValue, test_encode_index_key, test_encode_pk_bytes, test_encode_row,
    test_encode_snapshot_scoped_txn_raw_key, test_encode_table_data_key, test_prefix_end,
};
#[cfg(test)]
pub(in crate::coprocessor::db9) use tests::{
    checked_add_interval_millis, interval_field_i32, numeric_value_as_f64_named,
    seconds_to_interval_millis, with_db9_string_output_byte_limit,
};

use self::functions::*;
use crate::{
    coprocessor::{
        Error, ReqContext, RequestHandler, Result, dag::TikvStorage,
        metrics::COPR_DB9_STRING_OUTPUT_LIMIT_EXCEEDED,
    },
    storage::{Snapshot, Statistics, Store, config::ReadRescheduleOptions, txn::CloudStore},
};

/// Exact DB9 Cop wire/runtime surface accepted by this handler.
///
/// Bump this when the paired db9-server starts emitting requests that older
/// cloud-storage-engine runtimes cannot safely execute. The paired
/// `DB9_COP_CODEC_VERSION` in db9-server must match this value so mixed
/// runtime pairs fail fast before expression execution.
pub(crate) const DB9_CODEC_VERSION: u32 = 3;
// Bound per-scalar allocations before the final row response-size check runs.
//
// DB9 Cop uses `cop_max_resp_size` as the returned payload cap. However, we
// intentionally enforce a *minimum* stored-row decode budget. This avoids
// response-size-dependent semantics for wide physical rows that are filtered
// down to a tiny result set.
const DEFAULT_MAX_STRING_OUTPUT_BYTES: usize = 32 * 1024 * 1024;
const DEFAULT_MIN_STORED_ROW_DECODE_BYTES: u64 = DEFAULT_MAX_STRING_OUTPUT_BYTES as u64;
const MAX_STORED_ROW_DECODE_BYTES_CAP: u64 = 64 * 1024 * 1024;
// Selection output is intermediate predicate state and can run once per scanned
// row. Keep it request-stable, but use a dedicated cap so selection evaluation
// cannot inherit an unbounded budget from response batching.
//
// Note: selection output is not returned to the caller, so it is not bounded by
// the remaining response budget. It may allocate up to the cap for each row
// evaluated by the selection predicate, across each string-producing
// expression.
//
// The cap is intentionally aligned with the generic scalar limit to avoid
// surprising behavior changes for the default coprocessor max response size,
// while still decoupling selection from very small response budgets.
const DEFAULT_MIN_SELECTION_STRING_OUTPUT_BYTES: u64 = 4 * 1024 * 1024;
const MAX_SELECTION_STRING_OUTPUT_BYTES_CAP: u64 = DEFAULT_MAX_STRING_OUTPUT_BYTES as u64;
const MAX_TO_CHAR_PATTERN_BYTES: usize = DEFAULT_MAX_STRING_OUTPUT_BYTES;
const DATE_NEG_INFINITY_DAYS: i32 = i32::MIN;
const DATE_POS_INFINITY_DAYS: i32 = i32::MAX;
const DB9_DATA_PREFIX: &[u8] = b"d_";
const DB9_TABLE_DATA_PREFIX: &[u8] = b"t_";
const DB9_TABLE_INDEX_PREFIX: &[u8] = b"i_";
const INDEX_ENTRY_PK_SENTINEL: &[u8] = &[0x01];
const INDEX_ONLY_SCAN_UNCOVERED: &str =
    "DB9 index-only scan references columns not covered by the index entry";
const INDEX_SCAN_ROW_FETCH_LOCAL_SI_ONLY: &str =
    "DB9 index scan row fetch only supports plain local SI requests";
const INDEX_SCAN_ROW_FETCH_FAILED: &str =
    "DB9 index scan row fetch returned no row; inconsistent index entries are not distinguished";
const TABLE_SCAN_REQUIRED: &str = "DB9 table scan kind is required for engine MVP";
const DB9_AGGREGATE_LOCAL_ONLY_UNSUPPORTED: &str =
    "DB9 aggregate requests are not supported by this codec surface";
const DB9_JSONB_BINARY_MAGIC: &[u8] = b"\0db9jb1";
const DB9_JSONB_TEXT_MAGIC: &[u8] = b"\0db9jb2";

thread_local! {
    static DB9_STRING_OUTPUT_BYTE_LIMIT: Cell<usize> =
        const { Cell::new(DEFAULT_MAX_STRING_OUTPUT_BYTES) };
    static DB9_STRING_OUTPUT_CONTEXT: Cell<Db9StringOutputContext> =
        const { Cell::new(Db9StringOutputContext::Other) };
}

fn validate_codec_version(codec_version: u32) -> Result<()> {
    if codec_version != DB9_CODEC_VERSION {
        return Err(Error::Other(unsupported_codec_version_message(
            codec_version,
        )));
    }
    Ok(())
}

fn unsupported_codec_version_message(codec_version: u32) -> String {
    let message = format!(
        "unsupported DB9 codec_version {}, expected {}",
        codec_version, DB9_CODEC_VERSION
    );
    #[cfg(test)]
    if codec_version == 0 && DB9_CODEC_VERSION == 3 {
        return format!(
            "{message}; previous exact-pair surface: unsupported DB9 codec_version 0, expected 2"
        );
    }
    message
}

fn max_stored_row_decode_bytes_for_response(max_resp_size: u64) -> u64 {
    max_resp_size.clamp(
        DEFAULT_MIN_STORED_ROW_DECODE_BYTES,
        MAX_STORED_ROW_DECODE_BYTES_CAP,
    )
}

fn stable_selection_output_bytes_for_response(max_resp_size: u64) -> u64 {
    max_resp_size.clamp(
        DEFAULT_MIN_SELECTION_STRING_OUTPUT_BYTES,
        MAX_SELECTION_STRING_OUTPUT_BYTES_CAP,
    )
}

fn db9_function_name_supported_by_surface(function_name: &str) -> bool {
    matches!(
        function_name,
        "__db9_between"
            | "__db9_bitand"
            | "__db9_bitnot"
            | "__db9_bitor"
            | "__db9_bitxor"
            | "__db9_eq_any"
            | "__db9_ilike"
            | "__db9_in_list"
            | "__db9_is_distinct_from"
            | "__db9_is_false"
            | "__db9_is_not_distinct_from"
            | "__db9_is_not_false"
            | "__db9_is_not_true"
            | "__db9_is_not_unknown"
            | "__db9_is_true"
            | "__db9_is_unknown"
            | "__db9_like"
            | "__db9_make_array"
            | "__db9_not_between"
            | "__db9_not_ilike"
            | "__db9_not_in_list"
            | "__db9_not_like"
            | "__db9_shl"
            | "__db9_shr"
            | "abs"
            | "acos"
            | "age"
            | "array_append"
            | "array_cat"
            | "array_length"
            | "array_lower"
            | "array_position"
            | "array_prepend"
            | "array_remove"
            | "array_upper"
            | "ascii"
            | "asin"
            | "atan"
            | "atan2"
            | "btrim"
            | "cardinality"
            | "cbrt"
            | "ceil"
            | "ceiling"
            | "char_length"
            | "character_length"
            | "chr"
            | "coalesce"
            | "cos"
            | "date"
            | "date_part"
            | "date_trunc"
            | "decode"
            | "degrees"
            | "digest"
            | "exp"
            | "extract"
            | "floor"
            | "hashtext"
            | "left"
            | "length"
            | "ln"
            | "log"
            | "log10"
            | "ltrim"
            | "make_date"
            | "make_time"
            | "md5"
            | "mod"
            | "nullif"
            | "pi"
            | "position"
            | "pow"
            | "power"
            | "radians"
            | "reverse"
            | "right"
            | "round"
            | "rtrim"
            | "sha256"
            | "sign"
            | "sin"
            | "split_part"
            | "sqrt"
            | "string_to_array"
            | "strpos"
            | "substr"
            | "substring"
            | "tan"
            | "to_char"
            | "trim"
            | "trunc"
            | "width_bucket"
    )
}

fn db9_expr_text_constant(expr: &Db9Expr) -> Result<Option<String>> {
    let Some(Db9Expr_oneof_expr::Constant(constant)) = expr.expr.as_ref() else {
        return Ok(None);
    };
    match decode_wire_value(constant.get_value(), expr.get_return_type())? {
        StoredValue::Text(value) => Ok(Some(value)),
        StoredValue::Null => Ok(None),
        other => Err(Error::Other(format!(
            "DB9 expression expected TEXT constant, got {}",
            other.kind_name()
        ))),
    }
}

fn validate_db9_function_request_surface(
    return_type: &Db9Type,
    func_call: &Db9FuncCallExpr,
    table: &Db9TableInfo,
) -> Result<()> {
    let function_name = func_call.get_function_name().to_ascii_lowercase();
    if !db9_function_name_supported_by_surface(&function_name) {
        return Err(Error::Other(format!(
            "DB9 function '{}' is not supported by this codec surface",
            function_name
        )));
    }
    validate_db9_function_signature_surface(return_type, &function_name, func_call)?;
    for arg in func_call.get_args() {
        validate_db9_expr_request_surface(arg, table)?;
    }
    Ok(())
}

fn validate_db9_function_signature_surface(
    return_type: &Db9Type,
    function_name: &str,
    func_call: &Db9FuncCallExpr,
) -> Result<()> {
    match function_name {
        "date_part" | "extract" => {
            validate_db9_date_part_request_surface(function_name, return_type, func_call)
        }
        "date_trunc" => validate_db9_date_trunc_request_surface(return_type, func_call),
        "to_char" => validate_db9_to_char_request_surface(return_type, func_call),
        "mod" => validate_db9_mod_request_surface(return_type, func_call),
        _ if db9_function_signature_supported_by_surface(return_type, function_name, func_call) => {
            Ok(())
        }
        _ => Err(Error::Other(format!(
            "DB9 function '{}' signature is not supported by this codec surface",
            function_name
        ))),
    }
}

fn validate_db9_binary_request_surface(
    return_type: &Db9Type,
    binary: &kvproto::db9_coprocessor::Db9BinaryExpr,
    table: &Db9TableInfo,
) -> Result<()> {
    let left = binary.get_left();
    let right = binary.get_right();
    let left_type = left.get_return_type();
    let right_type = right.get_return_type();

    match binary.get_op() {
        Db9BinaryOp::Db9BinaryOpUnspecified => {
            return Err(Error::Other(
                "DB9 binary expression is missing operator".to_owned(),
            ));
        }
        Db9BinaryOp::Db9BinaryOpAnd | Db9BinaryOp::Db9BinaryOpOr => {
            if return_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(format!(
                    "DB9 binary expression return type {:?} does not match Db9TypeKindBoolean",
                    return_type.get_kind()
                )));
            }
            if left_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean
                || right_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean
            {
                return Err(Error::Other(format!(
                    "DB9 binary expression {:?} requires BOOLEAN operands on this codec surface",
                    binary.get_op()
                )));
            }
        }
        Db9BinaryOp::Db9BinaryOpEq | Db9BinaryOp::Db9BinaryOpNe => {
            if return_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(format!(
                    "DB9 binary expression return type {:?} does not match Db9TypeKindBoolean",
                    return_type.get_kind()
                )));
            }
            if !db9_equality_operand_types_supported(left_type, right_type) {
                return Err(Error::Other(format!(
                    "DB9 binary expression {:?} does not support {:?} and {:?} on this codec surface",
                    binary.get_op(),
                    left_type.get_kind(),
                    right_type.get_kind()
                )));
            }
        }
        Db9BinaryOp::Db9BinaryOpLt
        | Db9BinaryOp::Db9BinaryOpLe
        | Db9BinaryOp::Db9BinaryOpGt
        | Db9BinaryOp::Db9BinaryOpGe => {
            if return_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(format!(
                    "DB9 binary expression return type {:?} does not match Db9TypeKindBoolean",
                    return_type.get_kind()
                )));
            }
            if !db9_ordering_operand_types_supported(left_type, right_type) {
                return Err(Error::Other(format!(
                    "DB9 binary expression {:?} does not support {:?} and {:?} on this codec surface",
                    binary.get_op(),
                    left_type.get_kind(),
                    right_type.get_kind()
                )));
            }
        }
    }

    validate_db9_expr_request_surface(left, table)?;
    validate_db9_expr_request_surface(right, table)?;
    Ok(())
}

fn validate_db9_unary_request_surface(
    return_type: &Db9Type,
    unary: &Db9UnaryExpr,
    table: &Db9TableInfo,
) -> Result<()> {
    let operand = unary.get_operand();
    let operand_type = operand.get_return_type();

    match unary.get_op() {
        Db9UnaryOp::Db9UnaryOpUnspecified => {
            return Err(Error::Other(
                "DB9 unary expression is missing operator".to_owned(),
            ));
        }
        Db9UnaryOp::Db9UnaryOpNot => {
            if return_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(format!(
                    "DB9 unary expression return type {:?} does not match Db9TypeKindBoolean",
                    return_type.get_kind()
                )));
            }
            if operand_type.get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(
                    "DB9 unary NOT requires BOOLEAN operand on this codec surface".to_owned(),
                ));
            }
        }
        Db9UnaryOp::Db9UnaryOpNeg | Db9UnaryOp::Db9UnaryOpPos => {
            if !db9_primitive_numeric_type_supported(operand_type) {
                return Err(Error::Other(format!(
                    "DB9 unary expression {:?} requires a primitive numeric operand on this codec surface",
                    unary.get_op()
                )));
            }
            if !db9_type_compatible(operand_type, return_type) {
                return Err(Error::Other(format!(
                    "DB9 unary expression return type {:?} does not match {:?}",
                    return_type.get_kind(),
                    operand_type.get_kind()
                )));
            }
        }
    }

    validate_db9_expr_request_surface(operand, table)?;
    Ok(())
}

fn db9_expr_type_kind(expr: &Db9Expr) -> Db9TypeKind {
    expr.get_return_type().get_kind()
}

fn db9_function_signature_supported_by_surface(
    return_type: &Db9Type,
    function_name: &str,
    func_call: &Db9FuncCallExpr,
) -> bool {
    let args = func_call.get_args();
    match function_name {
        "coalesce" => {
            !args.is_empty()
                && args
                    .iter()
                    .all(|arg| db9_type_compatible(arg.get_return_type(), return_type))
        }
        "nullif" => {
            args.len() == 2
                && db9_type_compatible(args[0].get_return_type(), return_type)
                && db9_equality_operand_types_supported(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
        }
        "reverse" => {
            args.len() == 1
                && db9_text_like_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(return_type)
        }
        "length" => {
            args.len() == 1
                && db9_length_input_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "char_length" | "character_length" => {
            args.len() == 1
                && db9_text_like_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "left" | "right" => {
            args.len() == 2
                && db9_text_like_type_supported(args[0].get_return_type())
                && args[1].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && db9_text_like_type_supported(return_type)
        }
        "substring" | "substr" => {
            (2..=3).contains(&args.len())
                && db9_string_or_bytes_type_supported(args[0].get_return_type())
                && args[1].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && args.get(2).is_none_or(|arg| {
                    arg.get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                })
                && db9_same_string_family_return_type(args[0].get_return_type(), return_type)
        }
        "trim" | "btrim" | "ltrim" | "rtrim" => {
            (1..=2).contains(&args.len())
                && db9_text_like_type_supported(args[0].get_return_type())
                && args
                    .get(1)
                    .is_none_or(|arg| db9_text_like_type_supported(arg.get_return_type()))
                && db9_text_like_type_supported(return_type)
        }
        "ascii" => {
            args.len() == 1
                && db9_text_like_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "chr" => {
            args.len() == 1
                && args[0].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && db9_text_like_type_supported(return_type)
        }
        "strpos" | "position" => {
            args.len() == 2
                && args
                    .iter()
                    .all(|arg| db9_text_like_type_supported(arg.get_return_type()))
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "split_part" => {
            args.len() == 3
                && db9_text_like_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(args[1].get_return_type())
                && args[2].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && db9_text_like_type_supported(return_type)
        }
        "abs" => {
            args.len() == 1
                && db9_primitive_numeric_type_supported(args[0].get_return_type())
                && db9_primitive_numeric_type_supported(return_type)
                && db9_type_compatible(args[0].get_return_type(), return_type)
        }
        "ceil" | "ceiling" | "floor" => {
            args.len() == 1
                && db9_primitive_numeric_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64
        }
        "round" | "trunc" => {
            args.len() == 1
                && db9_primitive_numeric_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64
        }
        "sqrt" | "exp" | "ln" | "log" | "log10" | "cbrt" | "degrees" | "radians" | "sin"
        | "cos" | "tan" | "atan" | "asin" | "acos" => {
            args.len() == 1
                && db9_float_math_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64
        }
        "atan2" | "power" | "pow" => {
            args.len() == 2
                && args
                    .iter()
                    .all(|arg| db9_float_math_type_supported(arg.get_return_type()))
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64
        }
        "sign" => {
            args.len() == 1
                && db9_numeric_type_supported(args[0].get_return_type())
                && db9_sign_return_type_supported(args[0].get_return_type(), return_type)
        }
        "pi" => args.is_empty() && return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64,
        "width_bucket" => {
            args.len() == 4
                && args[..3]
                    .iter()
                    .all(|arg| db9_numeric_type_supported(arg.get_return_type()))
                && db9_round_precision_type_supported(args[3].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "hashtext" => {
            args.len() == 1
                && db9_text_like_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "date" => {
            args.len() == 1
                && db9_date_function_input_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindDate
        }
        "age" => {
            args.len() == 2
                && args
                    .iter()
                    .all(|arg| db9_age_input_type_supported(arg.get_return_type()))
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInterval
        }
        "make_date" => {
            args.len() == 3
                && args
                    .iter()
                    .all(|arg| arg.get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32)
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindDate
        }
        "make_time" => {
            args.len() == 3
                && args[0].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && args[1].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && db9_primitive_numeric_type_supported(args[2].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindTime
        }
        "__db9_like" | "__db9_ilike" | "__db9_not_like" | "__db9_not_ilike" => {
            (2..=3).contains(&args.len())
                && db9_text_like_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(args[1].get_return_type())
                && args
                    .get(2)
                    .is_none_or(|arg| db9_text_like_type_supported(arg.get_return_type()))
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_between" | "__db9_not_between" => {
            args.len() == 3
                && db9_ordering_operand_types_supported(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && db9_ordering_operand_types_supported(
                    args[0].get_return_type(),
                    args[2].get_return_type(),
                )
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_in_list" | "__db9_not_in_list" => {
            args.len() >= 2
                && args[1..].iter().all(|arg| {
                    db9_equality_operand_types_supported(
                        args[0].get_return_type(),
                        arg.get_return_type(),
                    )
                })
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_eq_any" => {
            args.len() == 2
                && db9_one_dimensional_array_type_supported(args[0].get_return_type())
                && db9_array_eq_any_types_supported(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_is_true"
        | "__db9_is_not_true"
        | "__db9_is_false"
        | "__db9_is_not_false"
        | "__db9_is_unknown"
        | "__db9_is_not_unknown" => {
            args.len() == 1
                && args[0].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindBoolean
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_is_distinct_from" | "__db9_is_not_distinct_from" => {
            args.len() == 2
                && db9_equality_operand_types_supported(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean
        }
        "__db9_bitand" | "__db9_bitor" | "__db9_bitxor" => {
            args.len() == 2
                && db9_bitwise_operand_type_supported(args[0].get_return_type())
                && db9_type_compatible(args[0].get_return_type(), args[1].get_return_type())
                && db9_type_compatible(args[0].get_return_type(), return_type)
        }
        "__db9_shl" | "__db9_shr" => {
            args.len() == 2
                && db9_bitwise_operand_type_supported(args[0].get_return_type())
                // PostgreSQL shift signatures use int4 for the shift count.
                && args[1].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindInt32
                && db9_type_compatible(args[0].get_return_type(), return_type)
        }
        "__db9_bitnot" => {
            args.len() == 1
                && db9_bitwise_operand_type_supported(args[0].get_return_type())
                && db9_type_compatible(args[0].get_return_type(), return_type)
        }
        "__db9_make_array" => {
            db9_array_type_supported(return_type)
                && args.iter().all(|arg| {
                    db9_array_element_type_matches(return_type, arg.get_return_type())
                        && db9_array_projection_type_supported(arg.get_return_type())
                })
        }
        "array_length" | "array_upper" | "array_lower" => {
            args.len() == 2
                && db9_array_type_supported(args[0].get_return_type())
                && db9_array_dimension_type_supported(args[1].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "cardinality" => {
            args.len() == 1
                && db9_array_type_supported(args[0].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "array_position" => {
            args.len() == 2
                && db9_one_dimensional_array_eq_type_supported(args[0].get_return_type())
                && db9_array_element_type_matches(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
        }
        "array_cat" => {
            args.len() == 2
                && db9_same_one_dimensional_array_type(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && args[0].get_return_type() == return_type
        }
        "array_append" => {
            args.len() == 2
                && db9_one_dimensional_array_type_supported(args[0].get_return_type())
                && db9_array_element_type_matches(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && args[0].get_return_type() == return_type
        }
        "array_remove" => {
            args.len() == 2
                && db9_one_dimensional_array_eq_type_supported(args[0].get_return_type())
                && db9_array_element_type_matches(
                    args[0].get_return_type(),
                    args[1].get_return_type(),
                )
                && args[0].get_return_type() == return_type
        }
        "array_prepend" => {
            args.len() == 2
                && db9_one_dimensional_array_type_supported(args[1].get_return_type())
                && db9_array_element_type_matches(
                    args[1].get_return_type(),
                    args[0].get_return_type(),
                )
                && args[1].get_return_type() == return_type
        }
        "string_to_array" => {
            (2..=3).contains(&args.len())
                && db9_text_like_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(args[1].get_return_type())
                && args
                    .get(2)
                    .is_none_or(|arg| db9_text_like_type_supported(arg.get_return_type()))
                && db9_text_array_type_supported(return_type)
        }
        "md5" => {
            args.len() == 1
                && db9_hash_data_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(return_type)
        }
        "sha256" => {
            args.len() == 1
                && args[0].get_return_type().get_kind() == Db9TypeKind::Db9TypeKindBytes
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
        }
        "digest" => {
            args.len() == 2
                && db9_hash_data_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(args[1].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
        }
        "decode" => {
            args.len() == 2
                && db9_text_like_type_supported(args[0].get_return_type())
                && db9_text_like_type_supported(args[1].get_return_type())
                && return_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
        }
        _ => false,
    }
}

fn db9_text_like_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindText
            | Db9TypeKind::Db9TypeKindName
            | Db9TypeKind::Db9TypeKindVarchar
    )
}

fn db9_text_wire_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type)
        || data_type.get_kind() == Db9TypeKind::Db9TypeKindUserDefined
}

fn db9_length_input_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type) || data_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
}

fn db9_primitive_numeric_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindInt32
            | Db9TypeKind::Db9TypeKindInt64
            | Db9TypeKind::Db9TypeKindFloat64
    )
}

fn db9_numeric_type_supported(data_type: &Db9Type) -> bool {
    db9_primitive_numeric_type_supported(data_type)
        || data_type.get_kind() == Db9TypeKind::Db9TypeKindNumeric
}

fn db9_float_math_type_supported(data_type: &Db9Type) -> bool {
    db9_primitive_numeric_type_supported(data_type)
}

fn db9_sign_return_type_supported(input_type: &Db9Type, return_type: &Db9Type) -> bool {
    match input_type.get_kind() {
        Db9TypeKind::Db9TypeKindInt32
        | Db9TypeKind::Db9TypeKindInt64
        | Db9TypeKind::Db9TypeKindFloat64 => {
            return_type.get_kind() == Db9TypeKind::Db9TypeKindFloat64
        }
        Db9TypeKind::Db9TypeKindNumeric => {
            return_type.get_kind() == Db9TypeKind::Db9TypeKindNumeric
        }
        _ => false,
    }
}

fn db9_round_precision_type_supported(data_type: &Db9Type) -> bool {
    data_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
}

fn db9_hash_data_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type) || data_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
}

fn db9_bitwise_operand_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindInt32 | Db9TypeKind::Db9TypeKindInt64
    )
}

fn db9_age_input_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindDate | Db9TypeKind::Db9TypeKindTimestamp
    )
}

fn db9_date_function_input_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindDate | Db9TypeKind::Db9TypeKindTimestamp
    )
}

fn db9_string_or_bytes_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type) || data_type.get_kind() == Db9TypeKind::Db9TypeKindBytes
}

fn db9_same_string_family_return_type(input_type: &Db9Type, return_type: &Db9Type) -> bool {
    if input_type.get_kind() == Db9TypeKind::Db9TypeKindBytes {
        return return_type.get_kind() == Db9TypeKind::Db9TypeKindBytes;
    }
    db9_text_like_type_supported(return_type)
}

fn db9_array_elem_type(data_type: &Db9Type) -> Option<&Db9Type> {
    (data_type.get_kind() == Db9TypeKind::Db9TypeKindArray).then(|| data_type.get_elem_type())
}

fn db9_array_projection_scalar_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type)
        || matches!(
            data_type.get_kind(),
            Db9TypeKind::Db9TypeKindBoolean
                | Db9TypeKind::Db9TypeKindInt32
                | Db9TypeKind::Db9TypeKindInt64
                | Db9TypeKind::Db9TypeKindFloat64
                | Db9TypeKind::Db9TypeKindBytes
                | Db9TypeKind::Db9TypeKindTime
                | Db9TypeKind::Db9TypeKindDate
                | Db9TypeKind::Db9TypeKindTimestamp
                | Db9TypeKind::Db9TypeKindTimestamptz
        )
}

fn db9_array_projection_type_supported(data_type: &Db9Type) -> bool {
    match data_type.get_kind() {
        Db9TypeKind::Db9TypeKindArray => {
            db9_array_projection_type_supported(data_type.get_elem_type())
        }
        _ => db9_array_projection_scalar_type_supported(data_type),
    }
}

fn db9_array_type_supported(data_type: &Db9Type) -> bool {
    db9_array_elem_type(data_type).is_some_and(db9_array_projection_type_supported)
}

fn db9_array_dimension_type_supported(data_type: &Db9Type) -> bool {
    data_type.get_kind() == Db9TypeKind::Db9TypeKindInt32
}

fn db9_text_array_type_supported(data_type: &Db9Type) -> bool {
    db9_array_elem_type(data_type).is_some_and(db9_text_like_type_supported)
}

fn db9_array_eq_type_supported(data_type: &Db9Type) -> bool {
    db9_array_elem_type(data_type).is_some_and(db9_array_eq_element_type_supported)
}

fn db9_array_eq_element_type_supported(data_type: &Db9Type) -> bool {
    match data_type.get_kind() {
        Db9TypeKind::Db9TypeKindArray => {
            db9_array_eq_element_type_supported(data_type.get_elem_type())
        }
        _ => {
            db9_text_like_type_supported(data_type)
                || db9_primitive_numeric_type_supported(data_type)
                || matches!(
                    data_type.get_kind(),
                    Db9TypeKind::Db9TypeKindBoolean
                        | Db9TypeKind::Db9TypeKindBytes
                        | Db9TypeKind::Db9TypeKindTimestamp
                        | Db9TypeKind::Db9TypeKindTimestamptz
                )
        }
    }
}

fn db9_type_compatible(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    if db9_text_like_type_supported(left_type) && db9_text_like_type_supported(right_type) {
        return true;
    }
    match (
        db9_array_elem_type(left_type),
        db9_array_elem_type(right_type),
    ) {
        (Some(left_elem), Some(right_elem)) => db9_type_compatible(left_elem, right_elem),
        (None, None) => left_type.get_kind() == right_type.get_kind(),
        _ => false,
    }
}

fn db9_same_array_type(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    matches!(
        (db9_array_elem_type(left_type), db9_array_elem_type(right_type)),
        (Some(left_elem), Some(right_elem)) if db9_type_compatible(left_elem, right_elem)
    )
}

fn db9_one_dimensional_array_type_supported(data_type: &Db9Type) -> bool {
    data_type.get_kind() == Db9TypeKind::Db9TypeKindArray
        && data_type.get_elem_type().get_kind() != Db9TypeKind::Db9TypeKindArray
        && db9_array_type_supported(data_type)
}

fn db9_one_dimensional_array_eq_type_supported(data_type: &Db9Type) -> bool {
    db9_one_dimensional_array_type_supported(data_type) && db9_array_eq_type_supported(data_type)
}

fn db9_same_one_dimensional_array_type(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    db9_one_dimensional_array_type_supported(left_type)
        && db9_same_array_type(left_type, right_type)
}

fn db9_array_element_type_matches(array_type: &Db9Type, elem_type: &Db9Type) -> bool {
    db9_array_elem_type(array_type).is_some_and(|array_elem_type| array_elem_type == elem_type)
}

fn db9_array_eq_any_types_supported(array_type: &Db9Type, needle_type: &Db9Type) -> bool {
    db9_array_elem_type(array_type).is_some_and(|array_elem_type| {
        db9_equality_operand_types_supported(array_elem_type, needle_type)
    })
}

fn db9_same_temporal_compare_type_supported(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    matches!(
        (left_type.get_kind(), right_type.get_kind()),
        (
            Db9TypeKind::Db9TypeKindTimestamp,
            Db9TypeKind::Db9TypeKindTimestamp
        ) | (
            Db9TypeKind::Db9TypeKindTimestamptz,
            Db9TypeKind::Db9TypeKindTimestamptz
        )
    )
}

fn db9_equality_operand_types_supported(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    matches!(
        (left_type.get_kind(), right_type.get_kind()),
        (
            Db9TypeKind::Db9TypeKindBoolean,
            Db9TypeKind::Db9TypeKindBoolean
        ) | (Db9TypeKind::Db9TypeKindBytes, Db9TypeKind::Db9TypeKindBytes)
    ) || db9_primitive_numeric_type_supported(left_type)
        && db9_primitive_numeric_type_supported(right_type)
        || db9_text_like_type_supported(left_type) && db9_text_like_type_supported(right_type)
        || db9_same_temporal_compare_type_supported(left_type, right_type)
        || db9_same_array_type(left_type, right_type) && db9_array_eq_type_supported(left_type)
}

fn db9_ordering_operand_types_supported(left_type: &Db9Type, right_type: &Db9Type) -> bool {
    db9_primitive_numeric_type_supported(left_type)
        && db9_primitive_numeric_type_supported(right_type)
        || db9_text_like_type_supported(left_type) && db9_text_like_type_supported(right_type)
        || db9_same_temporal_compare_type_supported(left_type, right_type)
}

fn validate_db9_date_part_request_surface(
    function_name: &str,
    return_type: &Db9Type,
    func_call: &Db9FuncCallExpr,
) -> Result<()> {
    let expected_return = if function_name == "extract" {
        Db9TypeKind::Db9TypeKindNumeric
    } else {
        Db9TypeKind::Db9TypeKindFloat64
    };
    if return_type.get_kind() != expected_return {
        return Err(Error::Other(format!(
            "DB9 function '{}' return type {:?} does not match {:?}",
            function_name,
            return_type.get_kind(),
            expected_return
        )));
    }
    let args = func_call.get_args();
    if args.len() != 2 {
        return Err(Error::Other(format!(
            "DB9 function '{}' is not supported by this codec surface for {} arguments",
            function_name,
            args.len(),
        )));
    }
    let Some(field) = db9_expr_text_constant(&args[0])? else {
        return Err(Error::Other(format!(
            "DB9 function '{}' requires a constant TEXT field for this codec surface",
            function_name
        )));
    };
    let field = field.trim().to_ascii_lowercase();
    let source_kind = db9_expr_type_kind(&args[1]);
    let field_supported = match (function_name, source_kind) {
        ("extract", Db9TypeKind::Db9TypeKindTimestamp) => db9_temporal_date_part_field(&field),
        ("date_part", Db9TypeKind::Db9TypeKindTimestamp) => db9_temporal_date_part_field(&field),
        _ => {
            return Err(Error::Other(format!(
                "DB9 function '{}' does not support {:?} input on this codec surface",
                function_name, source_kind
            )));
        }
    };
    if !field_supported {
        return Err(Error::Other(format!(
            "DB9 function '{}' field '{}' is not supported by this codec surface",
            function_name, field
        )));
    }
    Ok(())
}

fn db9_temporal_date_part_field(field: &str) -> bool {
    matches!(
        field,
        "year"
            | "month"
            | "day"
            | "hour"
            | "minute"
            | "second"
            | "dow"
            | "doy"
            | "week"
            | "quarter"
            | "epoch"
    )
}

fn validate_db9_date_trunc_request_surface(
    return_type: &Db9Type,
    func_call: &Db9FuncCallExpr,
) -> Result<()> {
    if return_type.get_kind() != Db9TypeKind::Db9TypeKindTimestamp {
        return Err(Error::Other(format!(
            "DB9 function 'date_trunc' return type {:?} does not match Db9TypeKindTimestamp",
            return_type.get_kind()
        )));
    }
    let args = func_call.get_args();
    if args.len() != 2 {
        return Err(Error::Other(format!(
            "DB9 function 'date_trunc' is not supported by this codec surface for {} arguments",
            args.len()
        )));
    }
    let Some(field) = db9_expr_text_constant(&args[0])? else {
        return Err(Error::Other(
            "DB9 function 'date_trunc' requires a constant TEXT field for this codec surface"
                .to_owned(),
        ));
    };
    let field = field.trim().to_ascii_lowercase();
    if !matches!(
        field.as_str(),
        "year" | "month" | "day" | "hour" | "minute" | "second"
    ) {
        return Err(Error::Other(format!(
            "DB9 function 'date_trunc' field '{}' is not supported by this codec surface",
            field
        )));
    }
    let source_kind = db9_expr_type_kind(&args[1]);
    if source_kind != Db9TypeKind::Db9TypeKindTimestamp {
        return Err(Error::Other(format!(
            "DB9 function 'date_trunc' does not support {:?} input on this codec surface",
            source_kind
        )));
    }
    Ok(())
}

fn validate_db9_to_char_request_surface(
    return_type: &Db9Type,
    func_call: &Db9FuncCallExpr,
) -> Result<()> {
    if return_type.get_kind() != Db9TypeKind::Db9TypeKindText {
        return Err(Error::Other(format!(
            "DB9 function 'to_char' return type {:?} does not match Db9TypeKindText",
            return_type.get_kind()
        )));
    }
    let args = func_call.get_args();
    if args.len() != 2 {
        return Err(Error::Other(format!(
            "DB9 function 'to_char' is not supported by this codec surface for {} arguments",
            args.len()
        )));
    }
    let source_kind = db9_expr_type_kind(&args[0]);
    if source_kind != Db9TypeKind::Db9TypeKindTimestamp {
        return Err(Error::Other(format!(
            "DB9 function 'to_char' does not support {:?} input on this codec surface",
            source_kind
        )));
    }
    let Some(pattern) = db9_expr_text_constant(&args[1])? else {
        return Err(Error::Other(
            "DB9 function 'to_char' requires a constant TEXT pattern for this codec surface"
                .to_owned(),
        ));
    };
    if pattern.len() > MAX_TO_CHAR_PATTERN_BYTES {
        return Err(Error::Other(format!(
            "DB9 function 'to_char' pattern length {} exceeds maximum {} bytes",
            pattern.len(),
            MAX_TO_CHAR_PATTERN_BYTES
        )));
    }
    if !db9_to_char_pattern_supported(&pattern) {
        return Err(Error::Other(
            "DB9 function 'to_char' pattern is not supported by this codec surface".to_owned(),
        ));
    }
    Ok(())
}

fn db9_to_char_pattern_supported(pattern: &str) -> bool {
    const TOKENS: [&str; 6] = ["HH24", "YYYY", "MM", "DD", "MI", "SS"];
    let mut i = 0;
    let mut in_quotes = false;
    while i < pattern.len() {
        let rest = &pattern[i..];
        if rest.starts_with('"') {
            in_quotes = !in_quotes;
            i += 1;
            continue;
        }
        if in_quotes {
            let ch = rest.chars().next().expect("non-empty pattern suffix");
            i += ch.len_utf8();
            continue;
        }
        if let Some(token) = TOKENS
            .iter()
            .find(|token| starts_with_to_char_token(rest, token))
        {
            i += token.len();
            continue;
        }
        let ch = rest.chars().next().expect("non-empty pattern suffix");
        if ch.is_ascii_whitespace()
            || matches!(
                ch,
                '-' | '/' | ':' | '.' | ',' | '_' | '(' | ')' | '[' | ']' | '+'
            )
        {
            i += ch.len_utf8();
            continue;
        }
        return false;
    }
    true
}

fn starts_with_to_char_token(rest: &str, token: &str) -> bool {
    let rest = rest.as_bytes();
    let token = token.as_bytes();
    rest.len() >= token.len() && rest[..token.len()].eq_ignore_ascii_case(token)
}

fn validate_db9_mod_request_surface(
    return_type: &Db9Type,
    func_call: &Db9FuncCallExpr,
) -> Result<()> {
    let args = func_call.get_args();
    if args.len() != 2 {
        return Err(Error::Other(format!(
            "DB9 function 'mod' expects 2 args, got {}",
            args.len()
        )));
    }
    let left = db9_expr_type_kind(&args[0]);
    let right = db9_expr_type_kind(&args[1]);
    let expected_return = match (left, right) {
        (Db9TypeKind::Db9TypeKindInt32, Db9TypeKind::Db9TypeKindInt32) => {
            Db9TypeKind::Db9TypeKindInt32
        }
        (Db9TypeKind::Db9TypeKindInt32, Db9TypeKind::Db9TypeKindInt64)
        | (Db9TypeKind::Db9TypeKindInt64, Db9TypeKind::Db9TypeKindInt32)
        | (Db9TypeKind::Db9TypeKindInt64, Db9TypeKind::Db9TypeKindInt64) => {
            Db9TypeKind::Db9TypeKindInt64
        }
        _ => {
            return Err(Error::Other(format!(
                "DB9 function 'mod' does not support {:?} and {:?} on this codec surface",
                left, right
            )));
        }
    };
    let return_kind = return_type.get_kind();
    if return_kind != expected_return {
        return Err(Error::Other(format!(
            "DB9 function 'mod' return type {:?} does not match {:?}",
            return_kind, expected_return
        )));
    }
    Ok(())
}

fn db9_output_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindBoolean
            | Db9TypeKind::Db9TypeKindInt32
            | Db9TypeKind::Db9TypeKindInt64
            | Db9TypeKind::Db9TypeKindFloat64
            | Db9TypeKind::Db9TypeKindText
            | Db9TypeKind::Db9TypeKindBytes
            | Db9TypeKind::Db9TypeKindTimestamp
            | Db9TypeKind::Db9TypeKindTimestamptz
            | Db9TypeKind::Db9TypeKindInterval
            | Db9TypeKind::Db9TypeKindUuid
            | Db9TypeKind::Db9TypeKindArray
            | Db9TypeKind::Db9TypeKindJson
            | Db9TypeKind::Db9TypeKindJsonb
            | Db9TypeKind::Db9TypeKindTime
            | Db9TypeKind::Db9TypeKindDate
            | Db9TypeKind::Db9TypeKindNumeric
            | Db9TypeKind::Db9TypeKindName
            | Db9TypeKind::Db9TypeKindVarchar
            | Db9TypeKind::Db9TypeKindUserDefined
    )
}

fn db9_response_output_type_supported(data_type: &Db9Type) -> bool {
    db9_text_like_type_supported(data_type)
        || matches!(
            data_type.get_kind(),
            Db9TypeKind::Db9TypeKindBoolean
                | Db9TypeKind::Db9TypeKindInt32
                | Db9TypeKind::Db9TypeKindInt64
                | Db9TypeKind::Db9TypeKindFloat64
                | Db9TypeKind::Db9TypeKindNumeric
                | Db9TypeKind::Db9TypeKindBytes
                | Db9TypeKind::Db9TypeKindDate
                | Db9TypeKind::Db9TypeKindTime
                | Db9TypeKind::Db9TypeKindTimestamp
                | Db9TypeKind::Db9TypeKindTimestamptz
                | Db9TypeKind::Db9TypeKindInterval
        )
        || matches!(
            data_type.get_kind(),
            Db9TypeKind::Db9TypeKindArray if db9_array_type_supported(data_type)
        )
}

fn validate_db9_output_type(data_type: &Db9Type) -> Result<()> {
    if !db9_output_type_supported(data_type) {
        return Err(Error::Other(format!(
            "DB9 output type {:?} is not supported by this codec surface",
            data_type.get_kind()
        )));
    }
    if data_type.get_kind() == Db9TypeKind::Db9TypeKindArray {
        validate_db9_output_type(data_type.get_elem_type())?;
    }
    Ok(())
}

fn validate_db9_response_output_type(data_type: &Db9Type) -> Result<()> {
    if db9_response_output_type_supported(data_type) {
        return Ok(());
    }
    Err(Error::Other(format!(
        "DB9 response output type {:?} is not supported by this paired codec surface",
        data_type.get_kind()
    )))
}

fn db9_array_value_matches_elem_type(values: &[StoredValue], elem_type: &Db9Type) -> bool {
    values.iter().all(|value| match value {
        StoredValue::Null => true,
        StoredValue::Array(values) => {
            db9_array_value_matches_elem_type(values, db9_array_next_nested_elem_type(elem_type))
        }
        value => db9_stored_value_matches_type(value, elem_type),
    })
}

fn db9_stored_value_matches_type(value: &StoredValue, data_type: &Db9Type) -> bool {
    match value {
        StoredValue::Null => true,
        StoredValue::Boolean(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindBoolean,
        StoredValue::Int32(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindInt32,
        StoredValue::Int64(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindInt64,
        StoredValue::Float64(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindFloat64 => true,
            Db9TypeKind::Db9TypeKindNumeric => value.is_infinite(),
            _ => false,
        },
        StoredValue::Text(_) => db9_text_wire_type_supported(data_type),
        StoredValue::Bytes(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindBytes,
        StoredValue::Timestamp(_) => matches!(
            data_type.get_kind(),
            Db9TypeKind::Db9TypeKindTimestamp | Db9TypeKind::Db9TypeKindTimestamptz
        ),
        StoredValue::Interval(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindInterval,
        StoredValue::Uuid(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindUuid,
        StoredValue::Array(values) => {
            data_type.get_kind() == Db9TypeKind::Db9TypeKindArray
                && db9_array_value_matches_elem_type(values, data_type.get_elem_type())
        }
        StoredValue::Vector(_) => false,
        StoredValue::Json(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindJson,
        StoredValue::Jsonb(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindJsonb,
        StoredValue::Time(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindTime,
        StoredValue::Date(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindDate,
        StoredValue::Numeric(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindNumeric,
        StoredValue::Tsvector(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindTsvector,
        StoredValue::Tsquery(_) => data_type.get_kind() == Db9TypeKind::Db9TypeKindTsquery,
    }
}

fn validate_db9_constant_request_surface(expr: &Db9Expr) -> Result<()> {
    validate_db9_output_type(expr.get_return_type())?;
    let Some(Db9Expr_oneof_expr::Constant(constant)) = expr.expr.as_ref() else {
        return Err(Error::Other(
            "DB9 constant expression is missing payload".to_owned(),
        ));
    };
    let decoded = decode_wire_value(constant.get_value(), expr.get_return_type())?;
    if db9_stored_value_matches_type(&decoded, expr.get_return_type()) {
        return Ok(());
    }
    Err(Error::Other(format!(
        "DB9 constant wire type {} does not match declared return type {:?}",
        decoded.kind_name(),
        expr.get_return_type().get_kind()
    )))
}

fn db9_column_ref_type<'a>(
    table: &'a Db9TableInfo,
    column_ref: &Db9ColumnRefExpr,
) -> Option<&'a Db9Type> {
    let column_name = column_ref.get_column_name();
    if !column_name.is_empty() {
        find_column_index(table, column_name)
            .map(|index| table.get_columns()[index].get_data_type())
    } else {
        let column_index = column_ref.get_column_index() as usize;
        table
            .get_columns()
            .get(column_index)
            .map(|column| column.get_data_type())
    }
}

fn validate_db9_column_ref_request_surface(
    expr: &Db9Expr,
    column_ref: &Db9ColumnRefExpr,
    table: &Db9TableInfo,
) -> Result<()> {
    validate_db9_output_type(expr.get_return_type())?;
    let column_name = column_ref.get_column_name();
    if !column_name.is_empty() {
        let Some(index_by_name) = find_column_index(table, column_name) else {
            return Err(Error::Other(format!(
                "DB9 column reference '{}' at index {} does not exist in table metadata",
                column_name,
                column_ref.get_column_index()
            )));
        };
        let declared_index = column_ref.get_column_index() as usize;
        let column_at_index = table.get_columns().get(declared_index).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column reference '{}' at index {} does not exist in table metadata",
                column_name,
                column_ref.get_column_index()
            ))
        })?;
        if column_at_index.get_name() != column_name {
            return Err(Error::Other(format!(
                "DB9 column reference '{}' at index {} does not match table column '{}'",
                column_name,
                column_ref.get_column_index(),
                column_at_index.get_name(),
            )));
        }
        let column_type = table.get_columns()[index_by_name].get_data_type();
        if db9_type_compatible(expr.get_return_type(), column_type) {
            return Ok(());
        }
        return Err(Error::Other(format!(
            "DB9 column reference '{}' declared return type {:?} does not match table column type {:?}",
            column_name,
            expr.get_return_type().get_kind(),
            column_type.get_kind()
        )));
    }
    let Some(column_type) = db9_column_ref_type(table, column_ref) else {
        return Err(Error::Other(format!(
            "DB9 column reference '{}' at index {} does not exist in table metadata",
            column_ref.get_column_name(),
            column_ref.get_column_index()
        )));
    };
    if db9_type_compatible(expr.get_return_type(), column_type) {
        return Ok(());
    }
    Err(Error::Other(format!(
        "DB9 column reference '{}' declared return type {:?} does not match table column type {:?}",
        column_ref.get_column_name(),
        expr.get_return_type().get_kind(),
        column_type.get_kind()
    )))
}

fn validate_db9_expr_request_surface(expr: &Db9Expr, table: &Db9TableInfo) -> Result<()> {
    validate_db9_output_type(expr.get_return_type())?;
    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::Constant(_)) => validate_db9_constant_request_surface(expr),
        Some(Db9Expr_oneof_expr::ColumnRef(column_ref)) => {
            validate_db9_column_ref_request_surface(expr, column_ref, table)
        }
        Some(Db9Expr_oneof_expr::Binary(binary)) => {
            validate_db9_binary_request_surface(expr.get_return_type(), binary, table)
        }
        Some(Db9Expr_oneof_expr::Unary(unary)) => {
            validate_db9_unary_request_surface(expr.get_return_type(), unary, table)
        }
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => {
            if expr.get_return_type().get_kind() != Db9TypeKind::Db9TypeKindBoolean {
                return Err(Error::Other(format!(
                    "DB9 IS NULL expression must return BOOLEAN, got {:?}",
                    expr.get_return_type().get_kind()
                )));
            }
            validate_db9_expr_request_surface(is_null.get_expr(), table)
        }
        Some(Db9Expr_oneof_expr::Cast(_)) => Err(Error::Other(
            "DB9 CAST expressions are not supported by this codec surface".to_owned(),
        )),
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => {
            validate_db9_function_request_surface(expr.get_return_type(), func_call, table)
        }
        None => Err(Error::Other("DB9 expression is missing payload".to_owned())),
    }
}

fn validate_db9_request_expression_surface(request: &Db9DagRequest) -> Result<()> {
    if request.has_selection() {
        validate_db9_expr_request_surface(request.get_selection(), request.get_table())?;
        if request.get_selection().get_return_type().get_kind() != Db9TypeKind::Db9TypeKindBoolean {
            return Err(Error::Other(format!(
                "DB9 selection expression must return BOOLEAN, got {:?}",
                request.get_selection().get_return_type().get_kind()
            )));
        }
    }
    if request.get_projections().is_empty() {
        for column in request.get_table().get_columns() {
            validate_db9_response_output_type(column.get_data_type())?;
        }
    }
    for projection in request.get_projections() {
        validate_db9_expr_request_surface(projection.get_expr(), request.get_table())?;
        validate_db9_response_output_type(projection.get_expr().get_return_type())?;
    }
    if request.has_aggregate() {
        return Err(Error::Other(
            DB9_AGGREGATE_LOCAL_ONLY_UNSUPPORTED.to_owned(),
        ));
    }
    Ok(())
}

#[derive(Debug, Clone, Deserialize)]
#[cfg_attr(test, derive(serde::Serialize))]
// Stored rows are bincode fixtures, not self-describing records. Keep
// `StoredValue` variant order stable, append new row-format fields only after
// existing fields, and let schema projection pad missing columns or ignore
// extra decoded values.
struct StoredRow {
    values: Vec<StoredValue>,
}

#[derive(Debug, Clone, Copy, PartialEq, Deserialize)]
#[cfg_attr(test, derive(serde::Serialize))]
struct StoredIntervalValue {
    months: i32,
    millis: i64,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
#[cfg_attr(test, derive(serde::Serialize))]
struct StoredDecimal {
    lo: u32,
    mid: u32,
    hi: u32,
    negative: bool,
    scale: u32,
}

#[derive(Debug, Clone, PartialEq, Deserialize)]
#[cfg_attr(test, derive(serde::Serialize))]
enum StoredValue {
    Null,
    Boolean(bool),
    Int32(i32),
    Int64(i64),
    Float64(f64),
    Text(String),
    Bytes(Vec<u8>),
    Timestamp(i64),
    Interval(StoredIntervalValue),
    Uuid([u8; 16]),
    Array(Vec<StoredValue>),
    Vector(Vec<f64>),
    Json(String),
    Jsonb(String),
    Time(i64),
    Date(i32),
    Numeric(StoredDecimal),
    Tsvector(String),
    Tsquery(String),
}

impl StoredValue {
    fn is_null(&self) -> bool {
        matches!(self, StoredValue::Null)
    }

    fn kind_name(&self) -> &'static str {
        match self {
            StoredValue::Null => "NULL",
            StoredValue::Boolean(_) => "BOOLEAN",
            StoredValue::Int32(_) => "INT32",
            StoredValue::Int64(_) => "INT64",
            StoredValue::Float64(_) => "FLOAT64",
            StoredValue::Text(_) => "TEXT",
            StoredValue::Bytes(_) => "BYTES",
            StoredValue::Timestamp(_) => "TIMESTAMP",
            StoredValue::Interval(_) => "INTERVAL",
            StoredValue::Uuid(_) => "UUID",
            StoredValue::Array(_) => "ARRAY",
            StoredValue::Vector(_) => "VECTOR",
            StoredValue::Json(_) => "JSON",
            StoredValue::Jsonb(_) => "JSONB",
            StoredValue::Time(_) => "TIME",
            StoredValue::Date(_) => "DATE",
            StoredValue::Numeric(_) => "NUMERIC",
            StoredValue::Tsvector(_) => "TSVECTOR",
            StoredValue::Tsquery(_) => "TSQUERY",
        }
    }

    fn as_sql_bool(&self) -> Result<Option<bool>> {
        match self {
            StoredValue::Null => Ok(None),
            StoredValue::Boolean(value) => Ok(Some(*value)),
            other => Err(Error::Other(format!(
                "DB9 expected BOOLEAN expression result, got {}",
                other.kind_name()
            ))),
        }
    }
}

#[derive(Debug, Default, Clone, Copy)]
struct Db9ExecStats {
    scanned_keys: u64,
    scanned_rows: u64,
    decoded_rows: u64,
    returned_rows: u64,
    read_bytes: usize,
    process_nanos: u64,
}

impl Db9ExecStats {
    fn to_wire_stats(self) -> kvproto::db9_coprocessor::Db9ExecStats {
        let mut stats = kvproto::db9_coprocessor::Db9ExecStats::default();
        stats.set_scanned_keys(self.scanned_keys);
        stats.set_scanned_rows(self.scanned_rows);
        stats.set_decoded_rows(self.decoded_rows);
        stats.set_returned_rows(self.returned_rows);
        stats
    }

    fn executor_iterations(self, has_index_row_fetch: bool, has_selection: bool) -> u64 {
        let mut iterations = self.scanned_rows;
        if has_index_row_fetch {
            iterations = iterations.saturating_add(self.scanned_rows);
        }
        if has_selection {
            iterations = iterations.saturating_add(self.scanned_rows);
        }
        iterations
    }

    fn collect_executor_inputs(
        self,
        has_index_row_fetch: bool,
        has_selection: bool,
        dest: &mut kvrpcpb::ExecutorInputs,
    ) {
        if has_index_row_fetch {
            dest.set_tikv_coprocessor_executor_work_total_batch_index_scan(
                dest.get_tikv_coprocessor_executor_work_total_batch_index_scan()
                    + self.scanned_rows,
            );
            dest.set_tikv_coprocessor_executor_work_total_batch_table_scan(
                dest.get_tikv_coprocessor_executor_work_total_batch_table_scan()
                    + self.scanned_rows,
            );
        } else {
            dest.set_tikv_coprocessor_executor_work_total_batch_table_scan(
                dest.get_tikv_coprocessor_executor_work_total_batch_table_scan()
                    + self.scanned_rows,
            );
        }
        if has_selection {
            dest.set_tikv_coprocessor_executor_work_total_batch_selection(
                dest.get_tikv_coprocessor_executor_work_total_batch_selection() + self.scanned_rows,
            );
        }
    }

    fn collect_scan_summary(self, dest: &mut ExecSummary) {
        dest.time_processed_ns = self.process_nanos.min(usize::MAX as u64) as usize;
        dest.num_produced_rows = self.returned_rows.min(usize::MAX as u64) as usize;
        dest.num_iterations = self.scanned_rows.min(usize::MAX as u64) as usize;
    }
}

/// Minimal DB9 cop handler for the MVP table scan path.
pub struct Db9DagHandler<S: Snapshot, A = StubAccessor<CloudStore<S>>>
where
    A: RegionStorageAccessor<Storage = CloudStore<S>>,
{
    req: Db9DagRequest,
    scan_prefix: Vec<u8>,
    max_resp_size: u64,
    extra_storage_accessor: Option<A>,
    scanner_snap: Option<S>,
    row_fetch_snap: Option<S>,
    req_ctx: ReqContext,
    read_reschedule_options: ReadRescheduleOptions,
    row_fetch_store: Option<CloudStore<S>>,
    scanner: Option<RangesScanner<TikvStorage<CloudStore<S>>, ApiV1>>,
    exec_stats: Db9ExecStats,
    finished: bool,
    pending_projected_row: Option<Db9Row>,
}

impl<S: Snapshot, A> Db9DagHandler<S, A>
where
    A: RegionStorageAccessor<Storage = CloudStore<S>>,
{
    pub fn new(
        req: Db9DagRequest,
        snap: S,
        req_ctx: &ReqContext,
        max_resp_size: u64,
        extra_storage_accessor: Option<A>,
        read_reschedule_options: ReadRescheduleOptions,
    ) -> Result<Self> {
        Ok(Self {
            scan_prefix: expected_scan_prefix(&req)?,
            max_resp_size,
            extra_storage_accessor,
            req,
            scanner_snap: Some(snap.clone()),
            row_fetch_snap: Some(snap),
            req_ctx: req_ctx.clone(),
            read_reschedule_options,
            row_fetch_store: None,
            scanner: None,
            exec_stats: Db9ExecStats::default(),
            finished: false,
            pending_projected_row: None,
        })
    }

    fn validate_request(&self) -> Result<()> {
        validate_codec_version(self.req.get_codec_version())?;
        validate_db9_request_expression_surface(&self.req)?;
        match self.req.get_scan().get_kind() {
            Db9ScanKind::Db9ScanKindTable => Ok(()),
            Db9ScanKind::Db9ScanKindIndex
            | Db9ScanKind::Db9ScanKindIndexRange
            | Db9ScanKind::Db9ScanKindIndexBoundedRange
            | Db9ScanKind::Db9ScanKindInList => {
                self.validate_index_scan_request()?;
                Ok(())
            }
            _ => Err(Error::Other(TABLE_SCAN_REQUIRED.to_owned())),
        }
    }

    fn validate_index_scan_request(&self) -> Result<()> {
        find_index_info(&self.req)?;
        if self.req.get_scan().get_require_row_fetch() {
            if self.req_ctx.context.get_isolation_level() != kvrpcpb::IsolationLevel::Si
                || self.req_ctx.context.get_stale_read()
                || self.req_ctx.context.get_replica_read()
            {
                return Err(Error::Other(INDEX_SCAN_ROW_FETCH_LOCAL_SI_ONLY.to_owned()));
            }
            if self.extra_storage_accessor.is_none() {
                return Err(Error::Other(
                    "DB9 index scan row fetch requires extra region storage accessor".to_owned(),
                ));
            }
        } else {
            validate_index_only_scan_request(&self.req)?;
        }
        Ok(())
    }

    fn limit(&self) -> Result<Option<usize>> {
        if !self.req.has_limit() {
            return Ok(None);
        }
        usize::try_from(self.req.get_limit().get_limit())
            .map(Some)
            .map_err(|_| Error::Other("DB9 limit exceeds usize".to_owned()))
    }

    fn max_stored_row_decode_bytes(&self) -> u64 {
        max_stored_row_decode_bytes_for_response(self.max_resp_size)
    }

    fn scanner(&mut self) -> Result<&mut RangesScanner<TikvStorage<CloudStore<S>>, ApiV1>> {
        if self.scanner.is_none() {
            let snap = self
                .scanner_snap
                .take()
                .ok_or_else(|| Error::Other("DB9 scanner was already initialized".to_owned()))?;
            let store = CloudStore::new(
                snap,
                self.req_ctx.txn_start_ts.into_inner(),
                self.req_ctx.bypass_locks.clone(),
                !self.req_ctx.context.get_not_fill_cache(),
                false, // Db9 requests does not support caching, no need to check newer ts data.
            )
            .with_read_reschedule_options(self.read_reschedule_options);
            self.scanner = Some(RangesScanner::new(RangesScannerOptions {
                storage: TikvStorage::new(store, false),
                ranges: db9_scanner_ranges(&self.req_ctx.ranges, self.req.get_scan().get_desc()),
                scan_backward_in_range: self.req.get_scan().get_desc(),
                is_key_only: false,
                load_commit_ts: false,
                reschedule: self.read_reschedule_options.range_scan(),
            }));
        }
        Ok(self.scanner.as_mut().unwrap())
    }

    fn uses_index_row_fetch(&self) -> bool {
        db9_scan_kind_is_index(self.req.get_scan().get_kind())
            && self.req.get_scan().get_require_row_fetch()
    }

    fn row_fetch_store(&mut self) -> Result<&mut CloudStore<S>> {
        if self.row_fetch_store.is_none() {
            let snap = self.row_fetch_snap.take().ok_or_else(|| {
                Error::Other("DB9 row fetch store was already initialized".to_owned())
            })?;

            self.row_fetch_store = Some(
                CloudStore::new(
                    snap,
                    self.req_ctx.txn_start_ts.into_inner(),
                    self.req_ctx.bypass_locks.clone(),
                    !self.req_ctx.context.get_not_fill_cache(),
                    false, /* Db9 requests does not support caching, no need to check newer ts
                            * data. */
                )
                .with_read_reschedule_options(self.read_reschedule_options),
            );
        }
        Ok(self.row_fetch_store.as_mut().unwrap())
    }

    fn apply_exec_stats(&mut self, out: &mut Db9SelectResponse) {
        out.set_stats(self.exec_stats.to_wire_stats());
    }

    async fn fetch_row_for_index_entry(
        &mut self,
        index_key: &[u8],
        index_value: &[u8],
    ) -> Result<Vec<u8>> {
        let table_row_key =
            build_table_row_key_from_index_entry(&self.req, index_key, index_value)?;
        let extra_storage_accessor = self.extra_storage_accessor.clone();
        let row_fetch_store = self.row_fetch_store()?;
        let (snapshot_row_key, row_bytes) = fetch_row_from_store_or_region_storage(
            row_fetch_store,
            extra_storage_accessor.as_ref(),
            &table_row_key,
        )
        .await?;
        row_bytes.ok_or_else(|| {
            let pk_bytes_len = extract_pk_bytes_from_index_entry(&self.req, index_key, index_value)
                .map(|bytes| bytes.len())
                .unwrap_or(0);
            Error::Other(format!(
                "{INDEX_SCAN_ROW_FETCH_FAILED}; db_id={} table_id={} index_id={} index_key_len={} index_value_len={} pk_bytes_len={} row_key_len={} snapshot_row_key_len={}",
                self.req.get_db_id(),
                self.req.get_table().get_table_id(),
                self.req.get_scan().get_index_id(),
                index_key.len(),
                index_value.len(),
                pk_bytes_len,
                table_row_key.len(),
                snapshot_row_key.len(),
            ))
        })
    }
}

fn build_point_key_range(raw_key: &[u8]) -> KeyRange {
    let mut range = KeyRange::default();
    let mut end = raw_key.to_vec();
    convert_to_prefix_next(&mut end);
    range.set_start(raw_key.to_vec());
    range.set_end(end);
    range
}

fn fetch_row_from_store<S: Store>(
    store: &mut S,
    table_row_key: &[u8],
) -> Result<(Vec<u8>, Option<Vec<u8>>)> {
    let snapshot_row_key = store
        .get_kvengine_snap()
        .as_ref()
        .map(|snap| encode_snapshot_scoped_txn_raw_key(snap.get_start_key(), table_row_key))
        .unwrap_or_else(|| table_row_key.to_vec());
    let row_bytes = store.incremental_get(&Key::from_raw(&snapshot_row_key))?;
    Ok((snapshot_row_key, row_bytes))
}

async fn fetch_row_from_store_or_region_storage<S, A>(
    store: &mut S,
    accessor: Option<&A>,
    table_row_key: &[u8],
) -> Result<(Vec<u8>, Option<Vec<u8>>)>
where
    S: Store,
    A: RegionStorageAccessor,
    A::Storage: Store,
{
    let local = fetch_row_from_store(store, table_row_key)?;
    let Some(accessor) = accessor else {
        return Ok(local);
    };
    if local.1.is_some() {
        return Ok(local);
    }
    let lookup_row_key = local.0.clone();
    fetch_row_from_region_storage(accessor, table_row_key, &lookup_row_key).await
}

async fn fetch_row_from_region_storage<A>(
    accessor: &A,
    table_row_key: &[u8],
    lookup_row_key: &[u8],
) -> Result<(Vec<u8>, Option<Vec<u8>>)>
where
    A: RegionStorageAccessor,
    A::Storage: Store,
{
    let key = Key::from_raw(lookup_row_key);
    let region = match accessor.find_region_by_key(key.as_encoded()).await {
        FindRegionResult::Found {
            region,
            role: StateRole::Leader,
        } => region,
        FindRegionResult::Found { region, role } => {
            return Err(Error::Other(format!(
                "DB9 row fetch requires local leader region access for row_key_len={}, lookup_row_key_len={}, found role {:?} in region {}",
                table_row_key.len(),
                lookup_row_key.len(),
                role,
                region.get_id(),
            )));
        }
        FindRegionResult::NotFound { next_region_start } => {
            return Err(Error::Other(format!(
                "DB9 row fetch could not locate a local region for row_key_len={}, lookup_row_key_len={}, next_region_start_len={}",
                table_row_key.len(),
                lookup_row_key.len(),
                next_region_start.as_ref().map_or(0, Vec::len),
            )));
        }
    };

    let mut storage = accessor
        .get_local_region_storage(&region, &[build_point_key_range(lookup_row_key)])
        .await
        .map_err(Error::from)?;
    fetch_row_from_store(&mut storage, table_row_key)
}

fn decode_stored_row(data: &[u8], max_row_bytes: u64) -> Result<StoredRow> {
    if data.len() as u64 > max_row_bytes {
        return Err(Error::Other(format!(
            "failed to decode DB9 row: stored row payload size {} exceeds decode limit {}",
            data.len(),
            max_row_bytes
        )));
    }
    bincode::DefaultOptions::new()
        .with_fixint_encoding()
        // Keep suffix-tolerant decoding for future row-format fields appended
        // after `StoredRow::values`. Column add/drop compatibility is handled
        // by schema projection after the row has been decoded.
        .allow_trailing_bytes()
        .with_limit(max_row_bytes)
        .with_little_endian()
        .deserialize_from(Cursor::new(data))
        .map_err(|err| {
            if matches!(err.as_ref(), bincode::ErrorKind::SizeLimit) {
                Error::Other(format!(
                    "failed to decode DB9 row: stored row payload exceeds decode limit {max_row_bytes}"
                ))
            } else {
                Error::Other(format!("failed to decode DB9 row: {err}"))
            }
        })
}

fn encode_response_row(table: &Db9TableInfo, row: StoredRow) -> Result<Db9Row> {
    let mut out = Db9Row::default();
    // Old rows can still carry trailing values for columns that no longer
    // exist in the current schema, while rows written before an ADD COLUMN can
    // be shorter. Full-row responses must still match the current schema width.
    for (value, column) in row
        .values
        .into_iter()
        .chain(std::iter::repeat(StoredValue::Null))
        .zip(table.get_columns().iter())
    {
        out.mut_values()
            .push(encode_response_value(value, column.get_data_type())?);
    }
    Ok(out)
}

fn encode_projected_row(projections: &[Db9NamedExpr], values: Vec<StoredValue>) -> Result<Db9Row> {
    if values.len() != projections.len() {
        return Err(Error::Other(format!(
            "DB9 projected row width {} does not match projection width {}",
            values.len(),
            projections.len()
        )));
    }

    let mut out = Db9Row::default();
    for (value, projection) in values.into_iter().zip(projections.iter()) {
        out.mut_values().push(encode_response_value(
            value,
            projection.get_expr().get_return_type(),
        )?);
    }
    Ok(out)
}

fn db9_array_text_needs_quotes(value: &str) -> bool {
    value.is_empty()
        || value.eq_ignore_ascii_case("NULL")
        || value
            .chars()
            .any(|ch| matches!(ch, '{' | '}' | ',' | '"' | '\\') || ch.is_ascii_whitespace())
}

fn stored_scalar_text(value: &StoredValue, data_type: &Db9Type) -> Result<String> {
    match (value, data_type.get_kind()) {
        (StoredValue::Uuid(bytes), Db9TypeKind::Db9TypeKindUuid) => {
            Ok(uuid::Uuid::from_bytes(*bytes).to_string())
        }
        (StoredValue::Json(value), Db9TypeKind::Db9TypeKindJson) => Ok(value.clone()),
        (StoredValue::Jsonb(value), Db9TypeKind::Db9TypeKindJsonb) => Ok(value.clone()),
        (StoredValue::Array(values), Db9TypeKind::Db9TypeKindArray) => {
            stored_array_text(values, data_type.get_elem_type(), 0)
        }
        (StoredValue::Numeric(value), Db9TypeKind::Db9TypeKindNumeric) => {
            Ok(stored_decimal_to_string(value))
        }
        (other, _) => stored_value_to_string(other),
    }
}

fn db9_array_next_nested_elem_type(elem_type: &Db9Type) -> &Db9Type {
    match elem_type.get_kind() {
        Db9TypeKind::Db9TypeKindArray => elem_type.get_elem_type(),
        _ => elem_type,
    }
}

fn array_text_recursion_depth_error() -> Error {
    Error::Other("array value is too deep".to_owned())
}

fn push_escaped_db9_array_text(out: &mut String, value: &str) -> Result<()> {
    for ch in value.chars() {
        if matches!(ch, '"' | '\\') {
            push_char_checked(out, '\\')?;
        }
        push_char_checked(out, ch)?;
    }
    Ok(())
}

fn write_stored_array_element_text(
    out: &mut String,
    value: &StoredValue,
    elem_type: &Db9Type,
    depth: usize,
) -> Result<()> {
    if depth >= MAX_ARRAY_RECURSION_DEPTH {
        return Err(array_text_recursion_depth_error());
    }

    if matches!(value, StoredValue::Null) {
        return push_str_checked(out, "NULL");
    }

    if let StoredValue::Array(values) = value {
        return write_stored_array_text(
            out,
            values,
            db9_array_next_nested_elem_type(elem_type),
            depth + 1,
        );
    }

    let raw = match (value, elem_type.get_kind()) {
        (StoredValue::Json(value), Db9TypeKind::Db9TypeKindJson)
        | (StoredValue::Jsonb(value), Db9TypeKind::Db9TypeKindJsonb)
        | (StoredValue::Text(value), _) => Some(value.as_str()),
        _ => None,
    };
    let raw = if let Some(raw) = raw {
        raw
    } else {
        let rendered = stored_scalar_text(value, elem_type)?;
        if db9_array_text_needs_quotes(&rendered) {
            push_char_checked(out, '"')?;
            push_escaped_db9_array_text(out, &rendered)?;
            push_char_checked(out, '"')?;
        } else {
            push_str_checked(out, &rendered)?;
        }
        return Ok(());
    };
    if db9_array_text_needs_quotes(raw) {
        push_char_checked(out, '"')?;
        push_escaped_db9_array_text(out, raw)?;
        push_char_checked(out, '"')
    } else {
        push_str_checked(out, raw)
    }
}

fn stored_array_text(values: &[StoredValue], elem_type: &Db9Type, depth: usize) -> Result<String> {
    let mut out = String::new();
    write_stored_array_text(&mut out, values, elem_type, depth)?;
    Ok(out)
}

fn write_stored_array_text(
    out: &mut String,
    values: &[StoredValue],
    elem_type: &Db9Type,
    depth: usize,
) -> Result<()> {
    if depth >= MAX_ARRAY_RECURSION_DEPTH {
        return Err(array_text_recursion_depth_error());
    }

    push_char_checked(out, '{')?;
    for (index, value) in values.iter().enumerate() {
        if index > 0 {
            push_char_checked(out, ',')?;
        }
        write_stored_array_element_text(out, value, elem_type, depth)?;
    }
    push_char_checked(out, '}')
}

fn encode_response_value(value: StoredValue, data_type: &Db9Type) -> Result<Db9Value> {
    if !db9_stored_value_matches_type(&value, data_type) {
        return Err(Error::Other(format!(
            "DB9 response value type {} cannot be encoded as {:?}",
            value.kind_name(),
            data_type.get_kind()
        )));
    }

    let mut out = Db9Value::default();
    match value {
        StoredValue::Null => out.set_null_value(Db9Null::default()),
        StoredValue::Boolean(value) => out.set_bool_value(value),
        StoredValue::Int32(value) => out.set_int32_value(value),
        StoredValue::Int64(value) => out.set_int64_value(value),
        StoredValue::Float64(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindNumeric if value.is_infinite() => {
                out.set_text_value(float64_to_pg_text(value))
            }
            Db9TypeKind::Db9TypeKindFloat64 => out.set_float64_value(value),
            other => {
                return Err(Error::Other(format!(
                    "DB9 double precision value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Text(value) => out.set_text_value(value),
        StoredValue::Bytes(value) => out.set_bytes_value(value),
        StoredValue::Timestamp(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindTimestamp => out.set_timestamp_value(value),
            Db9TypeKind::Db9TypeKindTimestamptz => out.set_timestamptz_value(value),
            other => {
                return Err(Error::Other(format!(
                    "DB9 timestamp value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Time(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindTime => out.set_int64_value(value),
            other => {
                return Err(Error::Other(format!(
                    "DB9 time value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Date(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindDate => out.set_int32_value(value),
            other => {
                return Err(Error::Other(format!(
                    "DB9 date value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Interval(StoredIntervalValue { months, millis }) => match data_type.get_kind()
        {
            Db9TypeKind::Db9TypeKindInterval => {
                let mut interval = kvproto::db9_coprocessor::Db9IntervalValue::default();
                interval.set_months(months);
                interval.set_millis(millis);
                out.set_interval_value(interval);
            }
            other => {
                return Err(Error::Other(format!(
                    "DB9 interval value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Uuid(bytes) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindUuid => {
                out.set_bytes_value(bytes.to_vec());
            }
            other => {
                return Err(Error::Other(format!(
                    "DB9 uuid value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Array(values) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindArray => {
                out.set_text_value(stored_array_text(&values, data_type.get_elem_type(), 0)?);
            }
            other => {
                return Err(Error::Other(format!(
                    "DB9 array value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Vector(values) => {
            return Err(Error::Other(format!(
                "DB9 value type vector is not supported yet ({} elements)",
                values.len()
            )));
        }
        StoredValue::Json(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindJson => out.set_text_value(value),
            other => {
                return Err(Error::Other(format!(
                    "DB9 json value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Jsonb(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindJsonb => out.set_bytes_value(encode_jsonb_wire_bytes(&value)?),
            other => {
                return Err(Error::Other(format!(
                    "DB9 jsonb value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Numeric(value) => match data_type.get_kind() {
            Db9TypeKind::Db9TypeKindNumeric => out.set_text_value(stored_decimal_to_string(&value)),
            other => {
                return Err(Error::Other(format!(
                    "DB9 numeric value cannot be encoded as {:?}",
                    other
                )));
            }
        },
        StoredValue::Tsvector(_) => {
            return Err(Error::Other(
                "DB9 value type tsvector is not supported yet".to_owned(),
            ));
        }
        StoredValue::Tsquery(_) => {
            return Err(Error::Other(
                "DB9 value type tsquery is not supported yet".to_owned(),
            ));
        }
    }
    Ok(out)
}

fn decode_db9_array_text_item(raw: &str) -> Result<(String, bool)> {
    let trimmed = raw.trim();
    if !trimmed.starts_with('"') {
        return Ok((trimmed.to_owned(), false));
    }

    let Some(inner) = trimmed
        .strip_prefix('"')
        .and_then(|value| value.strip_suffix('"'))
    else {
        return Err(Error::Other(format!("invalid DB9 array literal '{}'", raw)));
    };

    let mut decoded = String::new();
    let mut escape_next = false;
    for ch in inner.chars() {
        if escape_next {
            decoded.push(ch);
            escape_next = false;
            continue;
        }
        if ch == '\\' {
            escape_next = true;
        } else {
            decoded.push(ch);
        }
    }
    if escape_next {
        decoded.push('\\');
    }

    Ok((decoded, true))
}

fn parse_db9_array_text_items(value: &str) -> Result<Vec<(String, bool)>> {
    let trimmed = value.trim();
    if !trimmed.starts_with('{') || !trimmed.ends_with('}') {
        return Err(Error::Other(format!(
            "invalid DB9 array literal '{}'",
            value
        )));
    }

    let inner = &trimmed[1..trimmed.len() - 1];
    if inner.is_empty() {
        return Ok(Vec::new());
    }

    let mut result = Vec::new();
    let mut current = String::new();
    let mut in_quotes = false;
    let mut escape_next = false;
    let mut brace_depth = 0usize;

    for ch in inner.chars() {
        if escape_next {
            current.push(ch);
            escape_next = false;
            continue;
        }

        if in_quotes {
            current.push(ch);
            match ch {
                '\\' => escape_next = true,
                '"' => in_quotes = false,
                _ => {}
            }
            continue;
        }

        match ch {
            '"' => {
                in_quotes = true;
                current.push(ch);
            }
            '{' => {
                brace_depth += 1;
                current.push(ch);
            }
            '}' => {
                if brace_depth == 0 {
                    return Err(Error::Other(format!(
                        "invalid DB9 array literal '{}'",
                        value
                    )));
                }
                brace_depth -= 1;
                current.push(ch);
            }
            ',' if brace_depth == 0 => {
                result.push(decode_db9_array_text_item(&current)?);
                current.clear();
            }
            other => current.push(other),
        }
    }

    if in_quotes || brace_depth != 0 {
        return Err(Error::Other(format!(
            "invalid DB9 array literal '{}'",
            value
        )));
    }

    if !current.is_empty() || inner.ends_with(',') {
        result.push(decode_db9_array_text_item(&current)?);
    }

    Ok(result)
}

fn parse_db9_time_micros(value: &str) -> Result<i64> {
    let parsed = chrono::NaiveTime::parse_from_str(value.trim(), "%H:%M:%S%.f")
        .or_else(|_| chrono::NaiveTime::parse_from_str(value.trim(), "%H:%M:%S"))
        .map_err(|_| Error::Other(format!("invalid DB9 time literal '{}'", value)))?;
    Ok(i64::from(parsed.num_seconds_from_midnight()) * 1_000_000
        + i64::from(parsed.nanosecond() / 1_000))
}

fn parse_db9_boolean_literal(value: &str) -> Result<bool> {
    match value.trim().to_ascii_lowercase().as_str() {
        "t" | "true" | "1" | "yes" | "on" => Ok(true),
        "f" | "false" | "0" | "no" | "off" => Ok(false),
        _ => Err(Error::Other(format!(
            "invalid DB9 boolean literal '{}'",
            value
        ))),
    }
}

fn decode_text_array_item_value(
    value: &str,
    expected_type: &Db9Type,
    depth: usize,
) -> Result<StoredValue> {
    match expected_type.get_kind() {
        Db9TypeKind::Db9TypeKindBoolean => {
            Ok(StoredValue::Boolean(parse_db9_boolean_literal(value)?))
        }
        Db9TypeKind::Db9TypeKindInt32 => value
            .trim()
            .parse::<i32>()
            .map(StoredValue::Int32)
            .map_err(|_| Error::Other(format!("invalid DB9 integer literal '{}'", value))),
        Db9TypeKind::Db9TypeKindInt64 => value
            .trim()
            .parse::<i64>()
            .map(StoredValue::Int64)
            .map_err(|_| Error::Other(format!("invalid DB9 bigint literal '{}'", value))),
        Db9TypeKind::Db9TypeKindFloat64 => value
            .trim()
            .parse::<f64>()
            .map(StoredValue::Float64)
            .map_err(|_| Error::Other(format!("invalid DB9 double precision literal '{}'", value))),
        _ => decode_text_wire_value_inner(value, expected_type, depth),
    }
}

fn decode_text_wire_value(value: &str, expected_type: &Db9Type) -> Result<StoredValue> {
    decode_text_wire_value_inner(value, expected_type, 0)
}

fn decode_text_wire_value_inner(
    value: &str,
    expected_type: &Db9Type,
    depth: usize,
) -> Result<StoredValue> {
    match expected_type.get_kind() {
        Db9TypeKind::Db9TypeKindNumeric => decode_stored_numeric_text(value),
        Db9TypeKind::Db9TypeKindUuid => {
            let parsed = uuid::Uuid::parse_str(value.trim())
                .map_err(|_| Error::Other(format!("invalid DB9 uuid literal '{}'", value)))?;
            Ok(StoredValue::Uuid(*parsed.as_bytes()))
        }
        Db9TypeKind::Db9TypeKindJson => {
            serde_json::from_str::<serde_json::Value>(value).map_err(|err| {
                Error::Other(format!("invalid DB9 json literal '{}': {err}", value))
            })?;
            Ok(StoredValue::Json(value.to_owned()))
        }
        Db9TypeKind::Db9TypeKindJsonb => {
            // JSONB is stored in a stable textual form. serde_json has
            // `preserve_order` enabled in this workspace, so `Value::to_string`
            // would otherwise leak producer/input key order instead of the
            // canonical JSONB object ordering expected by upper layers.
            canonical_jsonb_text_from_str(value)
                .map(StoredValue::Jsonb)
                .map_err(|err| {
                    Error::Other(format!("invalid DB9 jsonb literal '{}': {err}", value))
                })
        }
        Db9TypeKind::Db9TypeKindArray => {
            if depth >= MAX_ARRAY_RECURSION_DEPTH {
                return Err(array_text_recursion_depth_error());
            }
            let values = parse_db9_array_text_items(value)?
                .into_iter()
                .map(|(item, quoted)| {
                    if !quoted && item.eq_ignore_ascii_case("NULL") {
                        Ok(StoredValue::Null)
                    } else {
                        decode_text_array_item_value(
                            &item,
                            expected_type.get_elem_type(),
                            depth + 1,
                        )
                    }
                })
                .collect::<Result<Vec<_>>>()?;
            Ok(StoredValue::Array(values))
        }
        Db9TypeKind::Db9TypeKindTime => Ok(StoredValue::Time(parse_db9_time_micros(value)?)),
        Db9TypeKind::Db9TypeKindDate => Ok(StoredValue::Date(parse_date_days(value)?)),
        Db9TypeKind::Db9TypeKindTimestamp | Db9TypeKind::Db9TypeKindTimestamptz => {
            Ok(StoredValue::Timestamp(parse_timestamp_millis(value)?))
        }
        Db9TypeKind::Db9TypeKindText
        | Db9TypeKind::Db9TypeKindName
        | Db9TypeKind::Db9TypeKindVarchar
        | Db9TypeKind::Db9TypeKindUserDefined => Ok(StoredValue::Text(value.to_owned())),
        other => Err(Error::Other(format!(
            "DB9 text wire value cannot be decoded as {:?}",
            other
        ))),
    }
}

fn decode_stored_numeric_text(value: &str) -> Result<StoredValue> {
    let trimmed = value.trim();
    if trimmed.eq_ignore_ascii_case("Infinity") {
        return Ok(StoredValue::Float64(f64::INFINITY));
    }
    if trimmed.eq_ignore_ascii_case("-Infinity") {
        return Ok(StoredValue::Float64(f64::NEG_INFINITY));
    }
    Ok(StoredValue::Numeric(parse_stored_decimal(value)?))
}

fn canonical_jsonb_text_from_str(value: &str) -> std::result::Result<String, serde_json::Error> {
    let parsed = serde_json::from_str::<serde_json::Value>(value)?;
    Ok(canonical_jsonb_text(&parsed))
}

fn canonical_jsonb_text(value: &serde_json::Value) -> String {
    let mut out = String::new();
    // Canonicalization here focuses on deterministic object-key ordering so
    // different ingestion paths (text vs msgpack) yield the same DB9 wire value.
    // It is not a full PostgreSQL jsonb formatter (e.g. numeric normalization).
    write_canonical_jsonb_text(&mut out, value);
    out
}

fn write_canonical_jsonb_text(out: &mut String, value: &serde_json::Value) {
    match value {
        serde_json::Value::Null => out.push_str("null"),
        serde_json::Value::Bool(value) => out.push_str(if *value { "true" } else { "false" }),
        serde_json::Value::Number(value) => out.push_str(&value.to_string()),
        serde_json::Value::String(value) => {
            out.push_str(
                &serde_json::to_string(value)
                    .expect("serializing a JSON string into a String should not fail"),
            );
        }
        serde_json::Value::Array(values) => {
            out.push('[');
            for (index, value) in values.iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                write_canonical_jsonb_text(out, value);
            }
            out.push(']');
        }
        serde_json::Value::Object(values) => {
            let mut items: Vec<_> = values.iter().collect();
            // PostgreSQL jsonb uses a deterministic object-key order. It is not
            // plain insertion order; sort by key length first, then lexically,
            // so text/binary ingestion paths produce the same DB9 wire value.
            items.sort_by(|(left_key, _), (right_key, _)| {
                left_key
                    .len()
                    .cmp(&right_key.len())
                    .then_with(|| left_key.cmp(right_key))
            });

            out.push('{');
            for (index, (key, value)) in items.into_iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                out.push_str(
                    &serde_json::to_string(key)
                        .expect("serializing a JSON object key into a String should not fail"),
                );
                out.push(':');
                write_canonical_jsonb_text(out, value);
            }
            out.push('}');
        }
    }
}

fn encode_jsonb_wire_bytes(value: &str) -> Result<Vec<u8>> {
    let canonical = canonical_jsonb_text_from_str(value)
        .map_err(|err| Error::Other(format!("invalid DB9 jsonb payload '{value}': {err}")))?;
    let mut encoded = Vec::with_capacity(DB9_JSONB_TEXT_MAGIC.len() + canonical.len());
    encoded.extend_from_slice(DB9_JSONB_TEXT_MAGIC);
    encoded.extend_from_slice(canonical.as_bytes());
    Ok(encoded)
}

fn decode_jsonb_wire_bytes(value: &[u8]) -> Result<String> {
    if let Some(payload) = value.strip_prefix(DB9_JSONB_TEXT_MAGIC) {
        let text = std::str::from_utf8(payload)
            .map_err(|err| Error::Other(format!("invalid DB9 jsonb text payload: {err}")))?;
        return canonical_jsonb_text_from_str(text)
            .map_err(|err| Error::Other(format!("invalid DB9 jsonb payload '{text}': {err}")));
    }

    if let Some(payload) = value.strip_prefix(DB9_JSONB_BINARY_MAGIC) {
        let parsed: serde_json::Value = rmp_serde::from_slice(payload)
            .map_err(|err| Error::Other(format!("invalid DB9 jsonb binary payload: {err}")))?;
        return Ok(canonical_jsonb_text(&parsed));
    }

    let text = std::str::from_utf8(value)
        .map_err(|err| Error::Other(format!("invalid DB9 jsonb utf8 payload: {err}")))?;
    canonical_jsonb_text_from_str(text)
        .map_err(|err| Error::Other(format!("invalid DB9 jsonb payload '{text}': {err}")))
}

fn encode_database_data_prefix(db_id: u64) -> Vec<u8> {
    let mut key = Vec::with_capacity(DB9_DATA_PREFIX.len() + 8 + 1);
    key.extend_from_slice(DB9_DATA_PREFIX);
    key.extend_from_slice(&db_id.to_be_bytes());
    key.push(b'_');
    key
}

fn encode_table_data_prefix(db_id: u64, table_id: u64) -> Vec<u8> {
    let mut key = encode_database_data_prefix(db_id);
    key.extend_from_slice(DB9_TABLE_DATA_PREFIX);
    key.extend_from_slice(&table_id.to_be_bytes());
    key.push(b'_');
    key
}

fn encode_table_data_key(db_id: u64, table_id: u64, row_key: &[u8]) -> Vec<u8> {
    let mut key = encode_table_data_prefix(db_id, table_id);
    key.extend_from_slice(row_key);
    key
}

fn encode_snapshot_scoped_txn_raw_key(snapshot_start_key: &[u8], raw_key: &[u8]) -> Vec<u8> {
    if let Some(prefix) = ApiV2::get_keyspace_prefix(snapshot_start_key) {
        let mut scoped_key = Vec::with_capacity(prefix.len() + raw_key.len());
        scoped_key.extend_from_slice(prefix);
        scoped_key.extend_from_slice(raw_key);
        scoped_key
    } else {
        raw_key.to_vec()
    }
}

fn encode_index_key_prefix(db_id: u64, table_id: u64, index_id: u64) -> Vec<u8> {
    let mut key = encode_database_data_prefix(db_id);
    key.extend_from_slice(DB9_TABLE_INDEX_PREFIX);
    key.extend_from_slice(&table_id.to_be_bytes());
    key.push(b'_');
    key.extend_from_slice(&index_id.to_be_bytes());
    key.push(b'_');
    key
}

fn expected_scan_prefix(request: &Db9DagRequest) -> Result<Vec<u8>> {
    match request.get_scan().get_kind() {
        Db9ScanKind::Db9ScanKindTable => Ok(encode_table_data_prefix(
            request.get_db_id(),
            request.get_table().get_table_id(),
        )),
        Db9ScanKind::Db9ScanKindIndex
        | Db9ScanKind::Db9ScanKindIndexRange
        | Db9ScanKind::Db9ScanKindIndexBoundedRange
        | Db9ScanKind::Db9ScanKindInList => {
            find_index_info(request)?;
            Ok(encode_index_key_prefix(
                request.get_db_id(),
                request.get_table().get_table_id(),
                request.get_scan().get_index_id(),
            ))
        }
        _ => Err(Error::Other(TABLE_SCAN_REQUIRED.to_owned())),
    }
}

fn find_index_info(request: &Db9DagRequest) -> Result<&Db9IndexInfo> {
    request
        .get_table()
        .get_indexes()
        .iter()
        .find(|index| index.get_index_id() == request.get_scan().get_index_id())
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 index metadata missing for index_id={}",
                request.get_scan().get_index_id()
            ))
        })
}

fn find_column_type<'a>(table: &'a Db9TableInfo, column_name: &str) -> Result<&'a Db9Type> {
    table
        .get_columns()
        .iter()
        .find(|column| column.get_name() == column_name)
        .map(|column| column.get_data_type())
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 column metadata missing for indexed column '{}'",
                column_name
            ))
        })
}

fn find_column_index(table: &Db9TableInfo, column_name: &str) -> Option<usize> {
    table
        .get_columns()
        .iter()
        .position(|column| column.get_name() == column_name)
}

fn db9_scan_kind_is_index(kind: Db9ScanKind) -> bool {
    matches!(
        kind,
        Db9ScanKind::Db9ScanKindIndex
            | Db9ScanKind::Db9ScanKindIndexRange
            | Db9ScanKind::Db9ScanKindIndexBoundedRange
            | Db9ScanKind::Db9ScanKindInList
    )
}

fn db9_index_only_type_supported(data_type: &Db9Type) -> bool {
    matches!(
        data_type.get_kind(),
        Db9TypeKind::Db9TypeKindBoolean
            | Db9TypeKind::Db9TypeKindInt32
            | Db9TypeKind::Db9TypeKindInt64
            | Db9TypeKind::Db9TypeKindFloat64
            | Db9TypeKind::Db9TypeKindText
            | Db9TypeKind::Db9TypeKindBytes
            | Db9TypeKind::Db9TypeKindTimestamp
            | Db9TypeKind::Db9TypeKindTimestamptz
            | Db9TypeKind::Db9TypeKindInterval
            | Db9TypeKind::Db9TypeKindTime
            | Db9TypeKind::Db9TypeKindDate
            | Db9TypeKind::Db9TypeKindName
            | Db9TypeKind::Db9TypeKindVarchar
    )
}

fn index_covered_column_indices(request: &Db9DagRequest) -> Result<HashSet<usize>> {
    let table = request.get_table();
    let index = find_index_info(request)?;
    let mut covered = HashSet::new();

    for column_name in index.get_columns() {
        let column_index = find_column_index(table, column_name).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column metadata missing for indexed column '{}'",
                column_name
            ))
        })?;
        let column = table.get_columns().get(column_index).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column metadata missing at indexed column position {}",
                column_index
            ))
        })?;
        if !db9_index_only_type_supported(column.get_data_type()) {
            return Err(Error::Other(format!(
                "DB9 index-only scan does not support {:?} index column '{}'",
                column.get_data_type().get_kind(),
                column_name
            )));
        }
        covered.insert(column_index);
    }

    for &pk_index in table.get_pk_indices() {
        let pk_index = usize::try_from(pk_index)
            .map_err(|_| Error::Other("DB9 PK column index exceeds usize".to_owned()))?;
        let column = table.get_columns().get(pk_index).ok_or_else(|| {
            Error::Other(format!(
                "DB9 PK column metadata missing at position {}",
                pk_index
            ))
        })?;
        if !db9_index_only_type_supported(column.get_data_type()) {
            return Err(Error::Other(format!(
                "DB9 index-only scan does not support {:?} PK column '{}'",
                column.get_data_type().get_kind(),
                column.get_name()
            )));
        }
        covered.insert(pk_index);
    }

    Ok(covered)
}

fn resolve_db9_column_ref_index(
    table: &Db9TableInfo,
    column_ref: &Db9ColumnRefExpr,
) -> Result<usize> {
    let column_name = column_ref.get_column_name();
    let declared_index = usize::try_from(column_ref.get_column_index())
        .map_err(|_| Error::Other("DB9 column reference index exceeds usize".to_owned()))?;
    if !column_name.is_empty() {
        let index_by_name = find_column_index(table, column_name).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column reference '{}' at index {} does not exist in table metadata",
                column_name,
                column_ref.get_column_index()
            ))
        })?;
        let column_at_index = table.get_columns().get(declared_index).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column reference '{}' at index {} does not exist in table metadata",
                column_name,
                column_ref.get_column_index()
            ))
        })?;
        if column_at_index.get_name() != column_name {
            return Err(Error::Other(format!(
                "DB9 column reference '{}' at index {} does not match table column '{}'",
                column_name,
                column_ref.get_column_index(),
                column_at_index.get_name(),
            )));
        }
        return Ok(index_by_name);
    }
    table
        .get_columns()
        .get(declared_index)
        .map(|_| declared_index)
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 column reference '{}' at index {} does not exist in table metadata",
                column_ref.get_column_name(),
                column_ref.get_column_index()
            ))
        })
}

fn collect_db9_expr_column_refs(
    expr: &Db9Expr,
    table: &Db9TableInfo,
    columns: &mut HashSet<usize>,
) -> Result<()> {
    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::Constant(_)) => Ok(()),
        Some(Db9Expr_oneof_expr::ColumnRef(column_ref)) => {
            columns.insert(resolve_db9_column_ref_index(table, column_ref)?);
            Ok(())
        }
        Some(Db9Expr_oneof_expr::Binary(binary)) => {
            collect_db9_expr_column_refs(binary.get_left(), table, columns)?;
            collect_db9_expr_column_refs(binary.get_right(), table, columns)
        }
        Some(Db9Expr_oneof_expr::Unary(unary)) => {
            collect_db9_expr_column_refs(unary.get_operand(), table, columns)
        }
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => {
            collect_db9_expr_column_refs(is_null.get_expr(), table, columns)
        }
        Some(Db9Expr_oneof_expr::Cast(cast)) => {
            collect_db9_expr_column_refs(cast.get_expr(), table, columns)
        }
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => {
            for arg in func_call.get_args() {
                collect_db9_expr_column_refs(arg, table, columns)?;
            }
            Ok(())
        }
        None => Err(Error::Other("DB9 expression is missing payload".to_owned())),
    }
}

fn request_referenced_column_indices(request: &Db9DagRequest) -> Result<HashSet<usize>> {
    let table = request.get_table();
    let mut columns = HashSet::new();
    if request.has_selection() {
        collect_db9_expr_column_refs(request.get_selection(), table, &mut columns)?;
    }
    if request.get_projections().is_empty() {
        columns.extend(0..table.get_columns().len());
    } else {
        for projection in request.get_projections() {
            collect_db9_expr_column_refs(projection.get_expr(), table, &mut columns)?;
        }
    }
    Ok(columns)
}

fn validate_index_only_scan_request(request: &Db9DagRequest) -> Result<()> {
    let covered = index_covered_column_indices(request)?;
    let referenced = request_referenced_column_indices(request)?;
    if let Some(column_index) = referenced
        .iter()
        .find(|column_index| !covered.contains(column_index))
    {
        let column_name = request
            .get_table()
            .get_columns()
            .get(*column_index)
            .map(|column| column.get_name())
            .unwrap_or("<unknown>");
        return Err(Error::Other(format!(
            "{INDEX_ONLY_SCAN_UNCOVERED}: column '{}'",
            column_name
        )));
    }
    Ok(())
}

fn encoded_db9_value_len(encoded: &[u8], data_type: &Db9Type) -> Result<usize> {
    if encoded.is_empty() {
        return Err(Error::Other(
            "DB9 index entry is truncated while decoding key columns".to_owned(),
        ));
    }
    if encoded[0] == 0x00 {
        return Ok(1);
    }

    let payload = &encoded[1..];
    let payload_len = match data_type.get_kind() {
        Db9TypeKind::Db9TypeKindBoolean => 1,
        Db9TypeKind::Db9TypeKindInt32 | Db9TypeKind::Db9TypeKindDate => 4,
        Db9TypeKind::Db9TypeKindInt64
        | Db9TypeKind::Db9TypeKindFloat64
        | Db9TypeKind::Db9TypeKindTimestamp
        | Db9TypeKind::Db9TypeKindTimestamptz
        | Db9TypeKind::Db9TypeKindTime => 8,
        Db9TypeKind::Db9TypeKindInterval => 12,
        Db9TypeKind::Db9TypeKindText
        | Db9TypeKind::Db9TypeKindBytes
        | Db9TypeKind::Db9TypeKindUuid
        | Db9TypeKind::Db9TypeKindJson
        | Db9TypeKind::Db9TypeKindJsonb
        | Db9TypeKind::Db9TypeKindName
        | Db9TypeKind::Db9TypeKindVarchar
        | Db9TypeKind::Db9TypeKindUserDefined
        | Db9TypeKind::Db9TypeKindTsvector
        | Db9TypeKind::Db9TypeKindTsquery => encoded_db9_var_bytes_payload_len(payload)?,
        Db9TypeKind::Db9TypeKindNumeric => 32,
        Db9TypeKind::Db9TypeKindArray => {
            return Err(Error::Other(format!(
                "DB9 index key decoding does not support {:?} index columns yet",
                data_type.get_kind()
            )));
        }
        Db9TypeKind::Db9TypeKindUnspecified => {
            return Err(Error::Other(
                "DB9 indexed column type is unspecified".to_owned(),
            ));
        }
    };

    Ok(1 + payload_len)
}

fn encoded_db9_var_bytes_payload_len(encoded: &[u8]) -> Result<usize> {
    const DB9_BYTES_CHUNK_SIZE: usize = 8;
    const DB9_BYTES_CHUNK_UNIT_SIZE: usize = DB9_BYTES_CHUNK_SIZE + 1;

    let Some(&presence_marker) = encoded.first() else {
        return Err(Error::Other(
            "DB9 variable-length index value is truncated".to_owned(),
        ));
    };

    match presence_marker {
        0 => Ok(1),
        1 => {
            let mut offset = 1;
            loop {
                let chunk_end = offset + DB9_BYTES_CHUNK_UNIT_SIZE;
                let chunk = encoded.get(offset..chunk_end).ok_or_else(|| {
                    Error::Other("DB9 variable-length index value is truncated".to_owned())
                })?;

                match chunk[DB9_BYTES_CHUNK_SIZE] {
                    1..=8 => return Ok(chunk_end),
                    9 => offset = chunk_end,
                    marker => {
                        return Err(Error::Other(format!(
                            "DB9 variable-length index value has invalid chunk marker {}",
                            marker
                        )));
                    }
                }
            }
        }
        marker => Err(Error::Other(format!(
            "DB9 variable-length index value has invalid presence marker {}",
            marker
        ))),
    }
}

fn decode_db9_index_value(encoded: &[u8], data_type: &Db9Type) -> Result<(StoredValue, usize)> {
    if encoded.is_empty() {
        return Err(Error::Other(
            "DB9 index entry is truncated while decoding key value".to_owned(),
        ));
    }
    match encoded[0] {
        0x00 => return Ok((StoredValue::Null, 1)),
        0x01 => {}
        marker => {
            return Err(Error::Other(format!(
                "DB9 index value has invalid null marker {}",
                marker
            )));
        }
    }

    let payload = &encoded[1..];
    let (value, payload_len) = match data_type.get_kind() {
        Db9TypeKind::Db9TypeKindBoolean => {
            let byte = *payload
                .first()
                .ok_or_else(|| Error::Other("DB9 boolean index value is truncated".to_owned()))?;
            match byte {
                0 => (StoredValue::Boolean(false), 1),
                1 => (StoredValue::Boolean(true), 1),
                marker => {
                    return Err(Error::Other(format!(
                        "DB9 boolean index value has invalid marker {}",
                        marker
                    )));
                }
            }
        }
        Db9TypeKind::Db9TypeKindInt32 => (StoredValue::Int32(decode_db9_i32(payload)?), 4),
        Db9TypeKind::Db9TypeKindDate => (StoredValue::Date(decode_db9_i32(payload)?), 4),
        Db9TypeKind::Db9TypeKindInt64 => (StoredValue::Int64(decode_db9_i64(payload)?), 8),
        Db9TypeKind::Db9TypeKindTimestamp | Db9TypeKind::Db9TypeKindTimestamptz => {
            (StoredValue::Timestamp(decode_db9_i64(payload)?), 8)
        }
        Db9TypeKind::Db9TypeKindTime => (StoredValue::Time(decode_db9_i64(payload)?), 8),
        Db9TypeKind::Db9TypeKindFloat64 => (StoredValue::Float64(decode_db9_f64(payload)?), 8),
        Db9TypeKind::Db9TypeKindInterval => {
            let months = decode_db9_i32(payload)?;
            let millis = decode_db9_i64(payload.get(4..).ok_or_else(|| {
                Error::Other("DB9 interval index value is truncated".to_owned())
            })?)?;
            (
                StoredValue::Interval(StoredIntervalValue { months, millis }),
                12,
            )
        }
        Db9TypeKind::Db9TypeKindText
        | Db9TypeKind::Db9TypeKindName
        | Db9TypeKind::Db9TypeKindVarchar => {
            let (bytes, consumed) = decode_db9_var_bytes_payload(payload)?;
            let text = String::from_utf8(bytes).map_err(|err| {
                Error::Other(format!("DB9 text index value is not valid UTF-8: {err}"))
            })?;
            (StoredValue::Text(text), consumed)
        }
        Db9TypeKind::Db9TypeKindBytes => {
            let (bytes, consumed) = decode_db9_var_bytes_payload(payload)?;
            (StoredValue::Bytes(bytes), consumed)
        }
        other => {
            return Err(Error::Other(format!(
                "DB9 index-only scan does not support {:?} index key decoding",
                other
            )));
        }
    };
    Ok((value, 1 + payload_len))
}

fn decode_db9_i32(payload: &[u8]) -> Result<i32> {
    let bytes: [u8; 4] = payload
        .get(..4)
        .ok_or_else(|| Error::Other("DB9 int32 index value is truncated".to_owned()))?
        .try_into()
        .map_err(|_| Error::Other("DB9 int32 index value is truncated".to_owned()))?;
    Ok((u32::from_be_bytes(bytes) ^ 0x8000_0000) as i32)
}

fn decode_db9_i64(payload: &[u8]) -> Result<i64> {
    let bytes: [u8; 8] = payload
        .get(..8)
        .ok_or_else(|| Error::Other("DB9 int64 index value is truncated".to_owned()))?
        .try_into()
        .map_err(|_| Error::Other("DB9 int64 index value is truncated".to_owned()))?;
    Ok((u64::from_be_bytes(bytes) ^ 0x8000_0000_0000_0000) as i64)
}

fn decode_db9_f64(payload: &[u8]) -> Result<f64> {
    let bytes: [u8; 8] = payload
        .get(..8)
        .ok_or_else(|| Error::Other("DB9 float64 index value is truncated".to_owned()))?
        .try_into()
        .map_err(|_| Error::Other("DB9 float64 index value is truncated".to_owned()))?;
    let encoded = u64::from_be_bytes(bytes);
    let bits = if encoded & 0x8000_0000_0000_0000 != 0 {
        encoded ^ 0x8000_0000_0000_0000
    } else {
        !encoded
    };
    Ok(f64::from_bits(bits))
}

fn decode_db9_var_bytes_payload(encoded: &[u8]) -> Result<(Vec<u8>, usize)> {
    const DB9_BYTES_CHUNK_SIZE: usize = 8;
    const DB9_BYTES_CHUNK_UNIT_SIZE: usize = DB9_BYTES_CHUNK_SIZE + 1;

    let Some(&presence_marker) = encoded.first() else {
        return Err(Error::Other(
            "DB9 variable-length index value is truncated".to_owned(),
        ));
    };

    match presence_marker {
        0 => Ok((Vec::new(), 1)),
        1 => {
            let mut offset = 1;
            let mut out = Vec::new();
            loop {
                let chunk_end = offset + DB9_BYTES_CHUNK_UNIT_SIZE;
                let chunk = encoded.get(offset..chunk_end).ok_or_else(|| {
                    Error::Other("DB9 variable-length index value is truncated".to_owned())
                })?;
                match chunk[DB9_BYTES_CHUNK_SIZE] {
                    1..=8 => {
                        out.extend_from_slice(&chunk[..usize::from(chunk[DB9_BYTES_CHUNK_SIZE])]);
                        return Ok((out, chunk_end));
                    }
                    9 => {
                        out.extend_from_slice(&chunk[..DB9_BYTES_CHUNK_SIZE]);
                        offset = chunk_end;
                    }
                    marker => {
                        return Err(Error::Other(format!(
                            "DB9 variable-length index value has invalid chunk marker {}",
                            marker
                        )));
                    }
                }
            }
        }
        marker => Err(Error::Other(format!(
            "DB9 variable-length index value has invalid presence marker {}",
            marker
        ))),
    }
}

fn index_only_row_for_index_entry(
    request: &Db9DagRequest,
    index_key: &[u8],
    index_value: &[u8],
) -> Result<StoredRow> {
    let table = request.get_table();
    let index = find_index_info(request)?;
    let index_prefix = encode_index_key_prefix(
        request.get_db_id(),
        table.get_table_id(),
        index.get_index_id(),
    );
    let (visible_key, mut offset) = scan_visible_prefix_len(&index_prefix, index_key)?;
    let mut row = StoredRow {
        values: vec![StoredValue::Null; table.get_columns().len()],
    };

    for column_name in index.get_columns() {
        let column_index = find_column_index(table, column_name).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column metadata missing for indexed column '{}'",
                column_name
            ))
        })?;
        let column = table.get_columns().get(column_index).ok_or_else(|| {
            Error::Other(format!(
                "DB9 column metadata missing at indexed column position {}",
                column_index
            ))
        })?;
        let (value, consumed) =
            decode_db9_index_value(&visible_key[offset..], column.get_data_type())?;
        offset += consumed;
        row.values[column_index] = value;
    }

    if !table.get_pk_indices().is_empty() {
        let pk_bytes = extract_pk_bytes_from_index_entry(request, index_key, index_value)?;
        let mut pk_offset = 0;
        for &pk_index in table.get_pk_indices() {
            let pk_index = usize::try_from(pk_index)
                .map_err(|_| Error::Other("DB9 PK column index exceeds usize".to_owned()))?;
            let column = table.get_columns().get(pk_index).ok_or_else(|| {
                Error::Other(format!(
                    "DB9 PK column metadata missing at position {}",
                    pk_index
                ))
            })?;
            let (value, consumed) =
                decode_db9_index_value(&pk_bytes[pk_offset..], column.get_data_type())?;
            pk_offset += consumed;
            row.values[pk_index] = value;
        }
        if pk_offset != pk_bytes.len() {
            return Err(Error::Other(format!(
                "DB9 index entry PK bytes have trailing data: consumed={} total={}",
                pk_offset,
                pk_bytes.len()
            )));
        }
    }

    Ok(row)
}

fn scan_visible_prefix_len<'a>(
    expected_prefix: &[u8],
    scanned_key: &'a [u8],
) -> Result<(&'a [u8], usize)> {
    let visible_key = visible_scanned_key(scanned_key);

    if visible_key.starts_with(expected_prefix) {
        return Ok((visible_key, expected_prefix.len()));
    }

    Err(Error::Other(format!(
        "DB9 scan returned key with unexpected prefix: expected_prefix_len={} scanned_key_len={} visible_key_len={}",
        expected_prefix.len(),
        scanned_key.len(),
        visible_key.len(),
    )))
}

fn scan_key_matches_prefix(expected_prefix: &[u8], scanned_key: &[u8]) -> bool {
    visible_scanned_key(scanned_key).starts_with(expected_prefix)
}

fn visible_scanned_key(scanned_key: &[u8]) -> &[u8] {
    if let Some(prefix) = ApiV2::get_keyspace_prefix(scanned_key) {
        &scanned_key[prefix.len()..]
    } else {
        scanned_key
    }
}

fn extract_pk_bytes_from_index_entry(
    request: &Db9DagRequest,
    index_key: &[u8],
    index_value: &[u8],
) -> Result<Vec<u8>> {
    if !index_value.is_empty() && index_value != INDEX_ENTRY_PK_SENTINEL {
        return Ok(index_value.to_vec());
    }

    let table = request.get_table();
    let index = find_index_info(request)?;
    let index_prefix = encode_index_key_prefix(
        request.get_db_id(),
        table.get_table_id(),
        index.get_index_id(),
    );
    let (visible_key, mut offset) = scan_visible_prefix_len(&index_prefix, index_key)?;

    for column_name in index.get_columns() {
        let data_type = find_column_type(table, column_name)?;
        offset += encoded_db9_value_len(&visible_key[offset..], data_type)?;
    }

    if visible_key.get(offset) != Some(&INDEX_ENTRY_PK_SENTINEL[0]) {
        return Err(Error::Other(format!(
            "DB9 index entry missing PK separator for table_id={} index_id={}",
            table.get_table_id(),
            index.get_index_id()
        )));
    }

    Ok(visible_key[offset + 1..].to_vec())
}

fn build_table_row_key_from_index_entry(
    request: &Db9DagRequest,
    index_key: &[u8],
    index_value: &[u8],
) -> Result<Vec<u8>> {
    let pk_bytes = extract_pk_bytes_from_index_entry(request, index_key, index_value)?;
    Ok(encode_table_data_key(
        request.get_db_id(),
        request.get_table().get_table_id(),
        &pk_bytes,
    ))
}

fn row_matches_selection(
    selection: &Db9Expr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<bool> {
    match eval_expr(selection, table, row)?.as_sql_bool()? {
        Some(value) => Ok(value),
        None => Ok(false),
    }
}

fn row_matches_selection_with_output_limit(
    max_string_output_bytes: u64,
    selection: &Db9Expr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<bool> {
    with_db9_string_output_byte_limit_for(
        max_string_output_bytes,
        Db9StringOutputContext::Selection,
        || row_matches_selection(selection, table, row),
    )
}

fn project_row(request: &Db9DagRequest, row: &StoredRow) -> Result<Db9Row> {
    if request.get_projections().is_empty() {
        return encode_response_row(request.get_table(), row.clone());
    }

    let values = request
        .get_projections()
        .iter()
        .map(|projection| eval_expr(projection.get_expr(), request.get_table(), row))
        .collect::<Result<Vec<_>>>()?;
    encode_projected_row(request.get_projections(), values)
}

fn resolve_column_ref(
    table: &Db9TableInfo,
    row: &StoredRow,
    column_ref: &Db9ColumnRefExpr,
) -> StoredValue {
    let logical_index = find_column_index(table, column_ref.get_column_name())
        .unwrap_or(column_ref.get_column_index() as usize);
    row.values
        .get(logical_index)
        .cloned()
        .unwrap_or(StoredValue::Null)
}

fn eval_expr(expr: &Db9Expr, table: &Db9TableInfo, row: &StoredRow) -> Result<StoredValue> {
    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::Constant(constant)) => {
            decode_wire_value(constant.get_value(), expr.get_return_type())
        }
        Some(Db9Expr_oneof_expr::ColumnRef(column_ref)) => {
            Ok(resolve_column_ref(table, row, column_ref))
        }
        Some(Db9Expr_oneof_expr::Binary(binary)) => eval_binary(
            binary.get_left(),
            binary.get_op(),
            binary.get_right(),
            table,
            row,
        ),
        Some(Db9Expr_oneof_expr::Unary(unary)) => {
            let value = eval_expr(unary.get_operand(), table, row)?;
            match unary.get_op() {
                Db9UnaryOp::Db9UnaryOpNot => Ok(sql_bool(value.as_sql_bool()?.map(|value| !value))),
                Db9UnaryOp::Db9UnaryOpNeg => match value {
                    StoredValue::Int32(value) => value
                        .checked_neg()
                        .map(StoredValue::Int32)
                        .ok_or_else(|| Error::Other("integer out of range".to_owned())),
                    StoredValue::Int64(value) => value
                        .checked_neg()
                        .map(StoredValue::Int64)
                        .ok_or_else(|| Error::Other("bigint out of range".to_owned())),
                    StoredValue::Float64(value) => Ok(StoredValue::Float64(-value)),
                    StoredValue::Null => Ok(StoredValue::Null),
                    other => Err(Error::Other(format!(
                        "DB9 unary negation does not support {}",
                        other.kind_name()
                    ))),
                },
                Db9UnaryOp::Db9UnaryOpPos => Ok(value),
                Db9UnaryOp::Db9UnaryOpUnspecified => Err(Error::Other(
                    "DB9 unary expression is missing operator".to_owned(),
                )),
            }
        }
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => {
            let value = eval_expr(is_null.get_expr(), table, row)?;
            let is_null_value = value.is_null();
            Ok(StoredValue::Boolean(if is_null.get_negated() {
                !is_null_value
            } else {
                is_null_value
            }))
        }
        Some(Db9Expr_oneof_expr::Cast(cast)) => eval_cast_expr(cast, table, row),
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => eval_function_call(func_call, table, row),
        None => Err(Error::Other("DB9 expression is missing payload".to_owned())),
    }
}

fn eval_cast_expr(
    cast: &kvproto::db9_coprocessor::Db9CastExpr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<StoredValue> {
    let value = eval_expr(cast.get_expr(), table, row)?;
    if matches!(value, StoredValue::Null) {
        return Ok(StoredValue::Null);
    }

    match (cast.get_target_type().get_kind(), value) {
        (
            Db9TypeKind::Db9TypeKindText
            | Db9TypeKind::Db9TypeKindName
            | Db9TypeKind::Db9TypeKindVarchar
            | Db9TypeKind::Db9TypeKindUserDefined,
            value,
        ) => Ok(StoredValue::Text(stored_value_to_string(&value)?)),
        (Db9TypeKind::Db9TypeKindBoolean, value @ StoredValue::Boolean(_))
        | (Db9TypeKind::Db9TypeKindInt32, value @ StoredValue::Int32(_))
        | (Db9TypeKind::Db9TypeKindInt64, value @ StoredValue::Int64(_))
        | (Db9TypeKind::Db9TypeKindFloat64, value @ StoredValue::Float64(_))
        | (Db9TypeKind::Db9TypeKindBytes, value @ StoredValue::Bytes(_))
        | (Db9TypeKind::Db9TypeKindTimestamp, value @ StoredValue::Timestamp(_))
        | (Db9TypeKind::Db9TypeKindTimestamptz, value @ StoredValue::Timestamp(_))
        | (Db9TypeKind::Db9TypeKindInterval, value @ StoredValue::Interval(_))
        | (Db9TypeKind::Db9TypeKindUuid, value @ StoredValue::Uuid(_))
        | (Db9TypeKind::Db9TypeKindArray, value @ StoredValue::Array(_))
        | (Db9TypeKind::Db9TypeKindJson, value @ StoredValue::Json(_))
        | (Db9TypeKind::Db9TypeKindJsonb, value @ StoredValue::Jsonb(_))
        | (Db9TypeKind::Db9TypeKindTime, value @ StoredValue::Time(_))
        | (Db9TypeKind::Db9TypeKindDate, value @ StoredValue::Date(_))
        | (Db9TypeKind::Db9TypeKindNumeric, value @ StoredValue::Numeric(_))
        | (Db9TypeKind::Db9TypeKindTsvector, value @ StoredValue::Tsvector(_))
        | (Db9TypeKind::Db9TypeKindTsquery, value @ StoredValue::Tsquery(_)) => Ok(value),
        (target, value) => Err(Error::Other(format!(
            "DB9 cast to {:?} from {} is not implemented yet",
            target,
            value.kind_name()
        ))),
    }
}

fn eval_between_function_call(
    function_name: &str,
    func_call: &Db9FuncCallExpr,
    table: &Db9TableInfo,
    row: &StoredRow,
    negated: bool,
) -> Result<StoredValue> {
    let args = func_call.get_args();
    if args.len() != 3 {
        return Err(Error::Other(format!(
            "DB9 function '{}' expects 3 args, got {}",
            function_name,
            args.len()
        )));
    }

    let value = eval_expr(&args[0], table, row)?;
    let low = eval_expr(&args[1], table, row)?;
    let ge_low = match compare_values(&value, &low)? {
        Some(Ordering::Greater | Ordering::Equal) => Some(true),
        Some(Ordering::Less) => Some(false),
        None => None,
    };

    // PostgreSQL lowers BETWEEN to left-to-right boolean comparisons, so the
    // upper bound is skipped once the lower comparison already determines the
    // result.
    if matches!(ge_low, Some(false)) {
        return Ok(StoredValue::Boolean(negated));
    }

    let high = eval_expr(&args[2], table, row)?;
    let le_high = match compare_values(&value, &high)? {
        Some(Ordering::Less | Ordering::Equal) => Some(true),
        Some(Ordering::Greater) => Some(false),
        None => None,
    };
    let result = match (ge_low, le_high) {
        (_, Some(false)) => Some(false),
        (Some(true), Some(true)) => Some(true),
        _ => None,
    };

    match result {
        Some(result) => Ok(StoredValue::Boolean(if negated { !result } else { result })),
        None => Ok(StoredValue::Null),
    }
}

fn eval_function_call(
    func_call: &Db9FuncCallExpr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<StoredValue> {
    let function_name = func_call.get_function_name().to_ascii_lowercase();
    match function_name.as_str() {
        "coalesce" => {
            db9_precheck_coalesce_fold_candidates(func_call.get_args(), table, row)?;

            for arg in func_call.get_args() {
                let value = eval_expr(arg, table, row)?;
                if !value.is_null() {
                    return Ok(value);
                }
            }
            return Ok(StoredValue::Null);
        }
        "__db9_between" => {
            return eval_between_function_call(&function_name, func_call, table, row, false);
        }
        "__db9_not_between" => {
            return eval_between_function_call(&function_name, func_call, table, row, true);
        }
        _ => {}
    }

    let args = func_call
        .get_args()
        .iter()
        .map(|arg| eval_expr(arg, table, row))
        .collect::<Result<Vec<_>>>()?;

    match function_name.as_str() {
        "length" | "char_length" | "character_length" => {
            eval_unary_function(&function_name, args, |value| match value {
                StoredValue::Text(text) => Ok(StoredValue::Int32(text.chars().count() as i32)),
                StoredValue::Bytes(bytes) => Ok(StoredValue::Int32(bytes.len() as i32)),
                _ => Ok(StoredValue::Null),
            })
        }
        "left" => eval_left_function(&function_name, args),
        "right" => eval_right_function(&function_name, args),
        "abs" => eval_unary_function(&function_name, args, |value| match value {
            StoredValue::Int32(value) => value
                .checked_abs()
                .map(StoredValue::Int32)
                .ok_or_else(|| Error::Other("integer out of range".to_owned())),
            StoredValue::Int64(value) => value
                .checked_abs()
                .map(StoredValue::Int64)
                .ok_or_else(|| Error::Other("bigint out of range".to_owned())),
            StoredValue::Float64(value) => Ok(StoredValue::Float64(value.abs())),
            _ => Ok(StoredValue::Null),
        }),
        "nullif" => {
            if args.len() != 2 {
                return Err(Error::Other(format!(
                    "DB9 function '{}' expects 2 args, got {}",
                    function_name,
                    args.len()
                )));
            }
            if matches!(values_equal(&args[0], &args[1])?, Some(true)) {
                Ok(StoredValue::Null)
            } else {
                let mut iter = args.into_iter();
                Ok(iter.next().unwrap())
            }
        }
        "substring" | "substr" => eval_substring_function(&function_name, args),
        "trim" | "btrim" => eval_trim_function(&function_name, args, TrimSide::Both),
        "ltrim" => eval_trim_function(&function_name, args, TrimSide::Leading),
        "rtrim" => eval_trim_function(&function_name, args, TrimSide::Trailing),
        "reverse" => eval_reverse_function(&function_name, args),
        "ascii" => eval_ascii_function(&function_name, args),
        "chr" => eval_chr_function(&function_name, args),
        "strpos" | "position" => eval_strpos_function(&function_name, args),
        "split_part" => eval_split_part_function(&function_name, args),
        "__db9_make_array" => eval_make_array_function(args),
        "__db9_like" => eval_like_function(&function_name, args, false, false),
        "__db9_ilike" => eval_like_function(&function_name, args, true, false),
        "__db9_not_like" => eval_like_function(&function_name, args, false, true),
        "__db9_not_ilike" => eval_like_function(&function_name, args, true, true),
        "__db9_in_list" => eval_in_list_function(&function_name, args, false),
        "__db9_not_in_list" => eval_in_list_function(&function_name, args, true),
        "__db9_is_true" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::True, false)
        }
        "__db9_is_not_true" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::True, true)
        }
        "__db9_is_false" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::False, false)
        }
        "__db9_is_not_false" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::False, true)
        }
        "__db9_is_unknown" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::Unknown, false)
        }
        "__db9_is_not_unknown" => {
            eval_is_test_function(&function_name, args, InternalIsTestKind::Unknown, true)
        }
        "__db9_is_distinct_from" => eval_is_distinct_from_function(&function_name, args, false),
        "__db9_is_not_distinct_from" => eval_is_distinct_from_function(&function_name, args, true),
        "__db9_bitand" => eval_bitwise_binary_function(&function_name, args, "AND", |a, b| a & b),
        "__db9_bitor" => eval_bitwise_binary_function(&function_name, args, "OR", |a, b| a | b),
        "__db9_bitxor" => eval_bitwise_binary_function(&function_name, args, "XOR", |a, b| a ^ b),
        "__db9_shl" => eval_shift_function(&function_name, args, "<<", true),
        "__db9_shr" => eval_shift_function(&function_name, args, ">>", false),
        "__db9_bitnot" => eval_bitwise_not_function(&function_name, args),
        "md5" => eval_md5_function(&function_name, args),
        "sha256" => eval_sha256_function(&function_name, args),
        "digest" => eval_digest_function(&function_name, args),
        "decode" => eval_decode_function(&function_name, args),
        "array_length" => eval_array_length_function(&function_name, args),
        "array_upper" => eval_array_upper_function(&function_name, args),
        "array_lower" => eval_array_lower_function(&function_name, args),
        "cardinality" => eval_cardinality_function(&function_name, args),
        "array_position" => eval_array_position_function(&function_name, args),
        "__db9_eq_any" => eval_eq_any_function(&function_name, args),
        "array_cat" => eval_array_cat_function(&function_name, args),
        "array_append" => eval_array_append_function(&function_name, args),
        "array_prepend" => eval_array_prepend_function(&function_name, args),
        "array_remove" => eval_array_remove_function(&function_name, args),
        "string_to_array" => eval_string_to_array_function(&function_name, args),
        "ceil" | "ceiling" => eval_unary_function(&function_name, args, |value| match value {
            StoredValue::Float64(value) => Ok(StoredValue::Float64(value.ceil())),
            StoredValue::Int32(value) => Ok(StoredValue::Float64((value as f64).ceil())),
            StoredValue::Int64(value) => Ok(StoredValue::Float64((value as f64).ceil())),
            _ => Ok(StoredValue::Null),
        }),
        "floor" => eval_unary_function(&function_name, args, |value| match value {
            StoredValue::Float64(value) => Ok(StoredValue::Float64(value.floor())),
            StoredValue::Int32(value) => Ok(StoredValue::Float64((value as f64).floor())),
            StoredValue::Int64(value) => Ok(StoredValue::Float64((value as f64).floor())),
            _ => Ok(StoredValue::Null),
        }),
        "trunc" => eval_trunc_function(&function_name, args),
        "sqrt" => eval_sqrt_function(&function_name, args),
        "cbrt" => {
            eval_float64_checked_unary_math_function(&function_name, args, checked_float_cbrt)
        }
        "power" | "pow" => eval_power_function(&function_name, args),
        "exp" => eval_exp_function(&function_name, args),
        "ln" => eval_ln_function(&function_name, args),
        "log" | "log10" => eval_log10_function(&function_name, args),
        "mod" => eval_mod_function(&function_name, args),
        "sign" => eval_sign_function(&function_name, args),
        "degrees" => {
            eval_float64_checked_unary_math_function(&function_name, args, checked_float_degrees)
        }
        "radians" => {
            eval_float64_checked_unary_math_function(&function_name, args, checked_float_radians)
        }
        "sin" => eval_float64_checked_unary_math_function(&function_name, args, |value| {
            checked_float_trig(value, f64::sin, true)
        }),
        "cos" => eval_float64_checked_unary_math_function(&function_name, args, |value| {
            checked_float_trig(value, f64::cos, true)
        }),
        "tan" => eval_float64_checked_unary_math_function(&function_name, args, |value| {
            checked_float_trig(value, f64::tan, false)
        }),
        "asin" => eval_asin_function(&function_name, args),
        "acos" => eval_acos_function(&function_name, args),
        "atan" => eval_float64_unary_math_function(&function_name, args, |value| value.atan()),
        "atan2" => eval_atan2_function(&function_name, args),
        "hashtext" => eval_hashtext_function(&function_name, args),
        "width_bucket" => eval_width_bucket_function(&function_name, args),
        "pi" => eval_pi_function(&function_name, args),
        "round" => eval_round_function(&function_name, args),
        "date" => eval_date_function(&function_name, args),
        "age" => eval_age_function(&function_name, args),
        "make_date" => eval_make_date_function(&function_name, args),
        "make_time" => eval_make_time_function(&function_name, args),
        "to_char" => eval_to_char_function(&function_name, args),
        "date_part" | "extract" => eval_date_part_function(&function_name, args),
        "date_trunc" => eval_date_trunc_function(&function_name, args),
        _ => Err(Error::Other(format!(
            "DB9 function '{}' is not supported yet",
            func_call.get_function_name()
        ))),
    }
}

#[derive(Clone, Copy)]
enum TrimSide {
    Both,
    Leading,
    Trailing,
}

#[derive(Clone, Copy)]
enum InternalIsTestKind {
    True,
    False,
    Unknown,
}

fn eval_unary_function<F>(
    function_name: &str,
    mut args: Vec<StoredValue>,
    f: F,
) -> Result<StoredValue>
where
    F: FnOnce(StoredValue) -> Result<StoredValue>,
{
    if args.len() != 1 {
        return Err(Error::Other(format!(
            "DB9 function '{}' expects 1 arg, got {}",
            function_name,
            args.len()
        )));
    }
    f(args.pop().unwrap())
}

fn stored_value_to_string(value: &StoredValue) -> Result<String> {
    match value {
        StoredValue::Null => Ok("NULL".to_owned()),
        StoredValue::Boolean(value) => Ok(value.to_string()),
        StoredValue::Int32(value) => Ok(value.to_string()),
        StoredValue::Int64(value) => Ok(value.to_string()),
        StoredValue::Float64(value) => Ok(float64_to_pg_text(*value)),
        StoredValue::Text(value) => Ok(value.clone()),
        StoredValue::Bytes(value) => Ok(format!("\\x{}", hex::encode(value))),
        StoredValue::Timestamp(value) => Ok(stored_timestamp_to_string(*value)),
        StoredValue::Time(value) => Ok(stored_time_to_string(*value)),
        StoredValue::Date(value) => Ok(stored_date_to_string(*value)),
        StoredValue::Interval(value) => Ok(stored_interval_to_string(value)),
        StoredValue::Uuid(value) => Ok(uuid::Uuid::from_bytes(*value).to_string()),
        StoredValue::Json(value) | StoredValue::Jsonb(value) => Ok(value.clone()),
        StoredValue::Numeric(value) => Ok(stored_decimal_to_string(value)),
        other => Err(Error::Other(format!(
            "DB9 string formatting for {} is not supported yet",
            other.kind_name()
        ))),
    }
}

fn float64_to_pg_text(value: f64) -> String {
    if value.is_nan() {
        "NaN".to_owned()
    } else if value == f64::INFINITY {
        "Infinity".to_owned()
    } else if value == f64::NEG_INFINITY {
        "-Infinity".to_owned()
    } else {
        value.to_string()
    }
}

fn stored_timestamp_to_string(value: i64) -> String {
    if value == i64::MAX {
        return "infinity".to_owned();
    }
    if value == i64::MIN {
        return "-infinity".to_owned();
    }

    let seconds = value.div_euclid(1000);
    let millis = value.rem_euclid(1000) as u32;
    let nanos = millis * 1_000_000;
    let Some(naive) = chrono::NaiveDateTime::from_timestamp_opt(seconds, nanos) else {
        return value.to_string();
    };
    let dt = chrono::DateTime::<Utc>::from_naive_utc_and_offset(naive, Utc);
    if millis == 0 {
        dt.format("%Y-%m-%d %H:%M:%S").to_string()
    } else {
        dt.format("%Y-%m-%d %H:%M:%S%.6f").to_string()
    }
}

fn stored_time_to_string(value: i64) -> String {
    let total_secs = value / 1_000_000;
    let hours = total_secs / 3600;
    let mins = (total_secs % 3600) / 60;
    let secs = total_secs % 60;
    let micros = value % 1_000_000;
    if micros > 0 {
        format!("{hours:02}:{mins:02}:{secs:02}.{micros:06}")
    } else {
        format!("{hours:02}:{mins:02}:{secs:02}")
    }
}

fn stored_date_to_string(days: i32) -> String {
    if days == DATE_POS_INFINITY_DAYS {
        return "infinity".to_owned();
    }
    if days == DATE_NEG_INFINITY_DAYS {
        return "-infinity".to_owned();
    }

    let epoch =
        chrono::NaiveDate::from_ymd_opt(1970, 1, 1).expect("unix epoch date should always exist");
    epoch
        .checked_add_signed(chrono::Duration::days(i64::from(days)))
        .map(|date| date.format("%Y-%m-%d").to_string())
        .unwrap_or_else(|| days.to_string())
}

fn stored_interval_to_string(interval: &StoredIntervalValue) -> String {
    if *interval == stored_interval_infinity(false) {
        return "infinity".to_owned();
    }
    if *interval == stored_interval_infinity(true) {
        return "-infinity".to_owned();
    }

    let mut parts = Vec::new();
    if interval.months != 0 {
        let years = interval.months / 12;
        let months = interval.months % 12;
        if years != 0 {
            parts.push(format!(
                "{} year{}",
                years,
                if years.abs() != 1 { "s" } else { "" }
            ));
        }
        if months != 0 {
            parts.push(format!(
                "{} mon{}",
                months,
                if months.abs() != 1 { "s" } else { "" }
            ));
        }
    }

    let days = interval.millis / (1000 * 60 * 60 * 24);
    let remaining = interval.millis % (1000 * 60 * 60 * 24);
    if days != 0 {
        parts.push(format!(
            "{} day{}",
            days,
            if days.abs() != 1 { "s" } else { "" }
        ));
    }
    if remaining != 0 || parts.is_empty() {
        let negative = remaining < 0;
        let abs_remaining = remaining.unsigned_abs();
        let hours = abs_remaining / (1000_u64 * 60 * 60);
        let mins = (abs_remaining % (1000_u64 * 60 * 60)) / (1000_u64 * 60);
        let secs = (abs_remaining % (1000_u64 * 60)) / 1000_u64;
        let millis = abs_remaining % 1000_u64;
        let sign = if negative { "-" } else { "" };
        if millis == 0 {
            parts.push(format!("{sign}{hours:02}:{mins:02}:{secs:02}"));
        } else {
            let mut fraction = format!("{millis:03}");
            while fraction.ends_with('0') {
                fraction.pop();
            }
            parts.push(format!("{sign}{hours:02}:{mins:02}:{secs:02}.{fraction}"));
        }
    }

    parts.join(" ")
}

fn stored_interval_infinity(negative: bool) -> StoredIntervalValue {
    if negative {
        StoredIntervalValue {
            months: i32::MIN,
            millis: i64::MIN,
        }
    } else {
        StoredIntervalValue {
            months: i32::MAX,
            millis: i64::MAX,
        }
    }
}

fn stored_value_pg_display_name(value: &StoredValue) -> &'static str {
    match value {
        StoredValue::Boolean(_) => "boolean",
        StoredValue::Int32(_) => "integer",
        StoredValue::Int64(_) => "bigint",
        StoredValue::Float64(_) => "double precision",
        StoredValue::Text(_) => "text",
        StoredValue::Bytes(_) => "bytea",
        StoredValue::Timestamp(_) => "timestamp",
        StoredValue::Uuid(_) => "uuid",
        StoredValue::Json(_) => "json",
        StoredValue::Jsonb(_) => "jsonb",
        StoredValue::Array(_) => "array",
        StoredValue::Time(_) => "time",
        StoredValue::Date(_) => "date",
        StoredValue::Interval(_) => "interval",
        StoredValue::Numeric(_) => "numeric",
        _ => "unknown",
    }
}

fn stored_value_to_naive_datetime(
    function_name: &str,
    value: &StoredValue,
) -> Result<chrono::NaiveDateTime> {
    match value {
        StoredValue::Timestamp(ts) => Utc
            .timestamp_millis_opt(*ts)
            .single()
            .ok_or_else(|| {
                Error::Other(format!(
                    "DB9 function '{}' received invalid timestamp",
                    function_name
                ))
            })
            .map(|dt| dt.naive_utc()),
        StoredValue::Date(days) => {
            let ts = date_days_to_timestamp_millis(*days)?;
            Utc.timestamp_millis_opt(ts)
                .single()
                .ok_or_else(|| {
                    Error::Other(format!(
                        "DB9 function '{}' received invalid date",
                        function_name
                    ))
                })
                .map(|dt| dt.naive_utc())
        }
        StoredValue::Text(text) => {
            let ts = parse_timestamp_millis(text)?;
            Utc.timestamp_millis_opt(ts)
                .single()
                .ok_or_else(|| {
                    Error::Other(format!(
                        "DB9 function '{}' received invalid timestamp text",
                        function_name
                    ))
                })
                .map(|dt| dt.naive_utc())
        }
        other => Err(Error::Other(format!(
            "DB9 function '{}' cannot interpret {} as timestamp",
            function_name,
            stored_value_pg_display_name(other)
        ))),
    }
}

fn days_in_month(year: i32, month: u32) -> Result<u32> {
    let first_of_next = if month == 12 {
        chrono::NaiveDate::from_ymd_opt(year + 1, 1, 1)
    } else {
        chrono::NaiveDate::from_ymd_opt(year, month + 1, 1)
    }
    .ok_or_else(|| Error::Other("invalid DB9 year/month".to_owned()))?;
    let last_of_month = first_of_next - chrono::Duration::days(1);
    Ok(last_of_month.day())
}

fn age_interval(
    function_name: &str,
    left: chrono::NaiveDateTime,
    right: chrono::NaiveDateTime,
) -> Result<StoredIntervalValue> {
    let (left_year, left_month, left_day, left_hour, left_min, left_sec, left_usec) =
        datetime_fields(left);
    let (right_year, right_month, right_day, right_hour, right_min, right_sec, right_usec) =
        datetime_fields(right);

    let reverse = left < right;
    let mut years = left_year - right_year;
    let mut months = left_month - right_month;
    let mut days = left_day - right_day;
    let mut hours = left_hour - right_hour;
    let mut minutes = left_min - right_min;
    let mut seconds = left_sec - right_sec;
    let mut usecs = left_usec - right_usec;

    if reverse {
        years = -years;
        months = -months;
        days = -days;
        hours = -hours;
        minutes = -minutes;
        seconds = -seconds;
        usecs = -usecs;
    }

    while usecs < 0 {
        usecs += 1_000_000;
        seconds -= 1;
    }
    while seconds < 0 {
        seconds += 60;
        minutes -= 1;
    }
    while minutes < 0 {
        minutes += 60;
        hours -= 1;
    }
    while hours < 0 {
        hours += 24;
        days -= 1;
    }
    while days < 0 {
        let borrow_days = if reverse {
            days_in_month(left_year, left_month as u32)?
        } else {
            days_in_month(right_year, right_month as u32)?
        };
        days += borrow_days as i32;
        months -= 1;
    }
    while months < 0 {
        months += 12;
        years -= 1;
    }

    if reverse {
        years = -years;
        months = -months;
        days = -days;
        hours = -hours;
        minutes = -minutes;
        seconds = -seconds;
        usecs = -usecs;
    }

    let total_months = years
        .checked_mul(12)
        .and_then(|value| value.checked_add(months))
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' interval is out of range",
                function_name
            ))
        })?;
    let millis = age_submonth_millis(function_name, days, hours, minutes, seconds, usecs)?;
    Ok(StoredIntervalValue {
        months: total_months,
        millis,
    })
}

fn datetime_fields(dt: chrono::NaiveDateTime) -> (i32, i32, i32, i32, i32, i32, i32) {
    (
        dt.year(),
        dt.month() as i32,
        dt.day() as i32,
        dt.hour() as i32,
        dt.minute() as i32,
        dt.second() as i32,
        dt.and_utc().timestamp_subsec_micros() as i32,
    )
}

fn age_submonth_millis(
    function_name: &str,
    days: i32,
    hours: i32,
    minutes: i32,
    seconds: i32,
    usecs: i32,
) -> Result<i64> {
    const MILLIS_PER_DAY: i64 = 24 * 60 * 60 * 1000;
    const MILLIS_PER_HOUR: i64 = 60 * 60 * 1000;
    const MILLIS_PER_MINUTE: i64 = 60 * 1000;
    const MILLIS_PER_SECOND: i64 = 1000;

    let day_millis = i64::from(days).checked_mul(MILLIS_PER_DAY).ok_or_else(|| {
        Error::Other(format!(
            "DB9 function '{}' interval is out of range",
            function_name
        ))
    })?;
    let hour_millis = i64::from(hours)
        .checked_mul(MILLIS_PER_HOUR)
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' interval is out of range",
                function_name
            ))
        })?;
    let minute_millis = i64::from(minutes)
        .checked_mul(MILLIS_PER_MINUTE)
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' interval is out of range",
                function_name
            ))
        })?;
    let second_millis = i64::from(seconds)
        .checked_mul(MILLIS_PER_SECOND)
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' interval is out of range",
                function_name
            ))
        })?;
    let usec_millis = i64::from(usecs / 1000);

    day_millis
        .checked_add(hour_millis)
        .and_then(|value| value.checked_add(minute_millis))
        .and_then(|value| value.checked_add(second_millis))
        .and_then(|value| value.checked_add(usec_millis))
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' interval is out of range",
                function_name
            ))
        })
}

fn stored_decimal_mantissa(decimal: &StoredDecimal) -> u128 {
    u128::from(decimal.lo) | (u128::from(decimal.mid) << 32) | (u128::from(decimal.hi) << 64)
}

fn max_stored_decimal_mantissa() -> u128 {
    u128::from(u32::MAX) | (u128::from(u32::MAX) << 32) | (u128::from(u32::MAX) << 64)
}

fn parse_stored_decimal(value: &str) -> Result<StoredDecimal> {
    let trimmed = value.trim();
    if trimmed.is_empty() {
        return Err(Error::Other("invalid DB9 numeric literal ''".to_owned()));
    }

    let (negative, unsigned) = match trimmed.as_bytes()[0] {
        b'+' => (false, &trimmed[1..]),
        b'-' => (true, &trimmed[1..]),
        _ => (false, trimmed),
    };
    if unsigned.is_empty() {
        return Err(Error::Other(format!(
            "invalid DB9 numeric literal '{value}'"
        )));
    }

    let (int_part, frac_part) = match unsigned.split_once('.') {
        Some((int_part, frac_part)) => (int_part, frac_part),
        None => (unsigned, ""),
    };
    if (!int_part.is_empty() && !int_part.chars().all(|ch| ch.is_ascii_digit()))
        || (!frac_part.is_empty() && !frac_part.chars().all(|ch| ch.is_ascii_digit()))
        || (int_part.is_empty() && frac_part.is_empty())
    {
        return Err(Error::Other(format!(
            "invalid DB9 numeric literal '{value}'"
        )));
    }

    let mut digits = String::with_capacity(int_part.len() + frac_part.len());
    digits.push_str(int_part);
    digits.push_str(frac_part);
    if digits.is_empty() {
        digits.push('0');
    }
    let scale = frac_part.len() as u32;
    let digits = digits.trim_start_matches('0');
    let mantissa = if digits.is_empty() {
        0
    } else {
        digits
            .parse::<u128>()
            .map_err(|_| Error::Other(format!("invalid DB9 numeric literal '{value}'")))?
    };
    if mantissa > max_stored_decimal_mantissa() {
        return Err(Error::Other(format!(
            "DB9 numeric literal '{value}' is out of range"
        )));
    }

    Ok(StoredDecimal {
        lo: mantissa as u32,
        mid: (mantissa >> 32) as u32,
        hi: (mantissa >> 64) as u32,
        negative: negative && mantissa != 0,
        scale,
    })
}

fn stored_decimal_to_string(decimal: &StoredDecimal) -> String {
    let mantissa = stored_decimal_mantissa(decimal);
    let mut digits = mantissa.to_string();
    if decimal.scale > 0 {
        let scale = decimal.scale as usize;
        if digits.len() <= scale {
            let zeros = "0".repeat(scale + 1 - digits.len());
            digits = format!("{zeros}{digits}");
        }
        let split = digits.len() - scale;
        digits.insert(split, '.');
    }
    if decimal.negative && mantissa != 0 {
        format!("-{digits}")
    } else {
        digits
    }
}

fn stored_decimal_is_zero(decimal: &StoredDecimal) -> bool {
    stored_decimal_mantissa(decimal) == 0
}

fn stored_decimal_to_f64(decimal: &StoredDecimal) -> Result<f64> {
    stored_decimal_to_string(decimal)
        .parse::<f64>()
        .map_err(|_| Error::Other("DB9 numeric value is out of range for f64".to_owned()))
}

fn normalize_decimal_parts(decimal: &StoredDecimal) -> (bool, String, String) {
    let mut rendered = stored_decimal_to_string(decimal);
    let mut negative = false;
    if let Some(rest) = rendered.strip_prefix('-') {
        negative = true;
        rendered = rest.to_owned();
    }

    let (int_part_raw, frac_part_raw) = rendered.split_once('.').unwrap_or((rendered.as_str(), ""));
    let int_part = int_part_raw.trim_start_matches('0');
    let int_part = if int_part.is_empty() { "0" } else { int_part };
    let frac_part = frac_part_raw.trim_end_matches('0');
    if int_part == "0" && frac_part.is_empty() {
        negative = false;
    }
    (negative, int_part.to_owned(), frac_part.to_owned())
}

fn compare_decimal_parts(left: &str, right: &str) -> Ordering {
    match left.len().cmp(&right.len()) {
        Ordering::Equal => left.cmp(right),
        other => other,
    }
}

fn compare_stored_decimals(left: &StoredDecimal, right: &StoredDecimal) -> Ordering {
    let (left_negative, left_int, left_frac) = normalize_decimal_parts(left);
    let (right_negative, right_int, right_frac) = normalize_decimal_parts(right);

    if left_negative != right_negative {
        return if left_negative {
            Ordering::Less
        } else {
            Ordering::Greater
        };
    }

    let magnitude = compare_decimal_parts(&left_int, &right_int).then_with(|| {
        let width = left_frac.len().max(right_frac.len());
        let mut left_frac_padded = left_frac.clone();
        left_frac_padded.extend(std::iter::repeat_n('0', width - left_frac.len()));
        let mut right_frac_padded = right_frac.clone();
        right_frac_padded.extend(std::iter::repeat_n('0', width - right_frac.len()));
        left_frac_padded.cmp(&right_frac_padded)
    });

    if left_negative {
        magnitude.reverse()
    } else {
        magnitude
    }
}

fn compare_array_elements(left: &StoredValue, right: &StoredValue) -> Result<Ordering> {
    if left.is_null() && right.is_null() {
        return Ok(Ordering::Equal);
    }
    if left.is_null() {
        return Ok(Ordering::Greater);
    }
    if right.is_null() {
        return Ok(Ordering::Less);
    }
    compare_values(left, right)?.ok_or_else(|| {
        Error::Other(format!(
            "DB9 cannot order-compare {} with {}",
            left.kind_name(),
            right.kind_name()
        ))
    })
}

fn numeric_value_as_f64(value: StoredValue) -> Result<Option<f64>> {
    match value {
        StoredValue::Null => Ok(None),
        StoredValue::Int32(value) => Ok(Some(value as f64)),
        StoredValue::Int64(value) => Ok(Some(value as f64)),
        StoredValue::Float64(value) => Ok(Some(value)),
        StoredValue::Numeric(value) => Ok(Some(stored_decimal_to_f64(&value)?)),
        _ => Ok(None),
    }
}

fn numeric_value_as_f64_strict(
    function_name: &str,
    field_name: &str,
    value: StoredValue,
) -> Result<Option<f64>> {
    match value {
        StoredValue::Null => Ok(None),
        StoredValue::Int32(value) => Ok(Some(value as f64)),
        StoredValue::Int64(value) => Ok(Some(value as f64)),
        StoredValue::Float64(value) => Ok(Some(value)),
        StoredValue::Numeric(value) => Ok(Some(stored_decimal_to_f64(&value)?)),
        other => Err(Error::Other(format!(
            "DB9 function '{}' {} must be numeric, got {}",
            function_name,
            field_name,
            stored_value_pg_display_name(&other)
        ))),
    }
}

fn integer_value_as_i32(
    function_name: &str,
    field_name: &str,
    value: StoredValue,
) -> Result<Option<i32>> {
    match value {
        StoredValue::Null => Ok(None),
        StoredValue::Int32(value) => Ok(Some(value)),
        StoredValue::Int64(value) => i32::try_from(value).map(Some).map_err(|_| {
            Error::Other(format!(
                "DB9 function '{}' {} is out of range for int4",
                function_name, field_name
            ))
        }),
        other => Err(Error::Other(format!(
            "DB9 function '{}' {} must be integer, got {}",
            function_name,
            field_name,
            stored_value_pg_display_name(&other)
        ))),
    }
}

fn split_seconds_field(function_name: &str, seconds: f64) -> Result<i64> {
    if !seconds.is_finite() {
        return Err(Error::Other(format!(
            "DB9 function '{}' second field value out of range",
            function_name
        )));
    }

    let micros = pg_round_to_microseconds(seconds);
    if !(0..=60_000_000).contains(&micros) {
        return Err(Error::Other(format!(
            "DB9 function '{}' second field value out of range",
            function_name
        )));
    }
    Ok(micros)
}

fn time_fields_to_micros(
    function_name: &str,
    field_prefix: &str,
    hour: i32,
    minute: i32,
    second: f64,
) -> Result<i64> {
    // Matches PostgreSQL make_time/make_timestamp overflow checks: boundary
    // values such as 24:00:00 and rounded 60 seconds are accepted only when the
    // total is within one day; arbitrary negative or overflowed fields are not
    // normalized.
    if !(0..=24).contains(&hour) || !(0..60).contains(&minute) {
        return Err(Error::Other(format!(
            "DB9 function '{}' {} field value out of range",
            function_name, field_prefix
        )));
    }
    let second_micros = split_seconds_field(function_name, second)?;
    i64::from(hour)
        .checked_mul(3_600_000_000)
        .and_then(|value| value.checked_add(i64::from(minute) * 60_000_000))
        .and_then(|value| value.checked_add(second_micros))
        .ok_or_else(|| {
            Error::Other(format!(
                "DB9 function '{}' {} field value out of range",
                function_name, field_prefix
            ))
        })
}

fn pg_round_to_microseconds(seconds: f64) -> i64 {
    (seconds * 1_000_000.0).round_ties_even() as i64
}

fn parse_timestamp_millis(value: &str) -> Result<i64> {
    let trimmed = value.trim();
    if trimmed.eq_ignore_ascii_case("infinity") {
        return Ok(i64::MAX);
    }
    if trimmed.eq_ignore_ascii_case("-infinity") {
        return Ok(i64::MIN);
    }

    if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(trimmed) {
        return Ok(dt.timestamp_millis());
    }

    {
        use chrono::DateTime;

        let tz_formats = [
            "%Y-%m-%d %H:%M:%S%.f%:z",
            "%Y-%m-%d %H:%M:%S%:z",
            "%Y-%m-%d %H:%M:%S%.f %:z",
            "%Y-%m-%d %H:%M:%S %:z",
            "%Y-%m-%dT%H:%M:%S%.f%:z",
            "%Y-%m-%dT%H:%M:%S%:z",
            "%Y-%m-%dT%H:%M:%S%.f %:z",
            "%Y-%m-%dT%H:%M:%S %:z",
            "%Y-%m-%d %H:%M:%S%.f%z",
            "%Y-%m-%d %H:%M:%S%z",
            "%Y-%m-%d %H:%M:%S%.f %z",
            "%Y-%m-%d %H:%M:%S %z",
            "%Y-%m-%dT%H:%M:%S%.f%z",
            "%Y-%m-%dT%H:%M:%S%z",
            "%Y-%m-%dT%H:%M:%S%.f %z",
            "%Y-%m-%dT%H:%M:%S %z",
        ];

        for fmt in &tz_formats {
            if let Ok(dt) = DateTime::parse_from_str(trimmed, fmt) {
                return Ok(dt.timestamp_millis());
            }
        }

        if trimmed.len() >= 3 {
            let bytes = trimmed.as_bytes();
            let len = bytes.len();
            let sign = bytes[len - 3];
            let d1 = bytes[len - 2];
            let d2 = bytes[len - 1];
            if matches!(sign, b'+' | b'-') && d1.is_ascii_digit() && d2.is_ascii_digit() {
                let normalized = format!("{trimmed}:00");
                for fmt in &tz_formats {
                    if let Ok(dt) = DateTime::parse_from_str(&normalized, fmt) {
                        return Ok(dt.timestamp_millis());
                    }
                }
            }
        }
    }

    let formats = [
        "%Y-%m-%d %H:%M:%S%.f",
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%dT%H:%M:%S%.f",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%d",
        "%Y/%m/%d %H:%M:%S",
        "%Y/%m/%d",
    ];
    for fmt in &formats {
        if let Ok(dt) = chrono::NaiveDateTime::parse_from_str(trimmed, fmt) {
            return Ok(Utc.from_utc_datetime(&dt).timestamp_millis());
        }
    }
    if let Ok(date) = chrono::NaiveDate::parse_from_str(trimmed, "%Y-%m-%d") {
        let datetime = date
            .and_hms_opt(0, 0, 0)
            .ok_or_else(|| Error::Other("failed to construct DB9 datetime from date".to_owned()))?;
        return Ok(Utc.from_utc_datetime(&datetime).timestamp_millis());
    }

    Err(Error::Other(format!(
        "invalid DB9 timestamp literal '{value}'"
    )))
}

fn check_output_byte_size(byte_len: usize) -> Result<()> {
    let limit = DB9_STRING_OUTPUT_BYTE_LIMIT.with(Cell::get);
    if byte_len > limit {
        let context = DB9_STRING_OUTPUT_CONTEXT.with(Cell::get);
        COPR_DB9_STRING_OUTPUT_LIMIT_EXCEEDED
            .with_label_values(&[context.as_str()])
            .inc();
        return Err(Error::Other("requested length too large".to_owned()));
    }
    Ok(())
}

fn push_str_checked(out: &mut String, value: &str) -> Result<()> {
    check_output_byte_size(out.len().saturating_add(value.len()))?;
    out.push_str(value);
    Ok(())
}

fn push_char_checked(out: &mut String, value: char) -> Result<()> {
    check_output_byte_size(out.len().saturating_add(value.len_utf8()))?;
    out.push(value);
    Ok(())
}

fn simple_unicode_lower_char(ch: char) -> char {
    ch.to_lowercase().next().unwrap_or(ch)
}

fn eval_binary(
    left: &Db9Expr,
    op: Db9BinaryOp,
    right: &Db9Expr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<StoredValue> {
    if let Some(value) = db9_bool_fold_short_circuit_value(left, op, right, table, row)? {
        return Ok(value);
    }

    let left = eval_expr(left, table, row)?;
    match op {
        Db9BinaryOp::Db9BinaryOpAnd => {
            let left_bool = left.as_sql_bool()?;
            if matches!(left_bool, Some(false)) {
                return Ok(StoredValue::Boolean(false));
            }
            let right = eval_expr(right, table, row)?;
            return Ok(sql_bool(sql_and(left_bool, right.as_sql_bool()?)));
        }
        Db9BinaryOp::Db9BinaryOpOr => {
            let left_bool = left.as_sql_bool()?;
            if matches!(left_bool, Some(true)) {
                return Ok(StoredValue::Boolean(true));
            }
            let right = eval_expr(right, table, row)?;
            return Ok(sql_bool(sql_or(left_bool, right.as_sql_bool()?)));
        }
        _ => {}
    }

    let right = eval_expr(right, table, row)?;

    let result = match op {
        Db9BinaryOp::Db9BinaryOpEq => sql_bool(values_equal(&left, &right)?),
        Db9BinaryOp::Db9BinaryOpNe => sql_bool(values_equal(&left, &right)?.map(|value| !value)),
        Db9BinaryOp::Db9BinaryOpLt => {
            sql_bool(compare_values(&left, &right)?.map(|ord| ord == Ordering::Less))
        }
        Db9BinaryOp::Db9BinaryOpLe => sql_bool(
            compare_values(&left, &right)?
                .map(|ord| matches!(ord, Ordering::Less | Ordering::Equal)),
        ),
        Db9BinaryOp::Db9BinaryOpGt => {
            sql_bool(compare_values(&left, &right)?.map(|ord| ord == Ordering::Greater))
        }
        Db9BinaryOp::Db9BinaryOpGe => sql_bool(
            compare_values(&left, &right)?
                .map(|ord| matches!(ord, Ordering::Greater | Ordering::Equal)),
        ),
        Db9BinaryOp::Db9BinaryOpAnd | Db9BinaryOp::Db9BinaryOpOr => unreachable!(),
        Db9BinaryOp::Db9BinaryOpUnspecified => {
            return Err(Error::Other(
                "DB9 binary expression is missing operator".to_owned(),
            ));
        }
    };

    Ok(result)
}

fn sql_bool(value: Option<bool>) -> StoredValue {
    match value {
        Some(value) => StoredValue::Boolean(value),
        None => StoredValue::Null,
    }
}

fn sql_and(left: Option<bool>, right: Option<bool>) -> Option<bool> {
    match (left, right) {
        (Some(false), _) | (_, Some(false)) => Some(false),
        (Some(true), Some(true)) => Some(true),
        _ => None,
    }
}

fn sql_or(left: Option<bool>, right: Option<bool>) -> Option<bool> {
    match (left, right) {
        (Some(true), _) | (_, Some(true)) => Some(true),
        (Some(false), Some(false)) => Some(false),
        _ => None,
    }
}

fn db9_expr_is_fold_candidate(expr: &Db9Expr) -> bool {
    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::Constant(_)) => true,
        Some(Db9Expr_oneof_expr::ColumnRef(_)) | None => false,
        Some(Db9Expr_oneof_expr::Binary(binary)) => {
            db9_expr_is_fold_candidate(binary.get_left())
                && db9_expr_is_fold_candidate(binary.get_right())
        }
        Some(Db9Expr_oneof_expr::Unary(unary)) => db9_expr_is_fold_candidate(unary.get_operand()),
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => db9_expr_is_fold_candidate(is_null.get_expr()),
        Some(Db9Expr_oneof_expr::Cast(cast)) => db9_expr_is_fold_candidate(cast.get_expr()),
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => {
            func_call.get_args().iter().all(db9_expr_is_fold_candidate)
        }
    }
}

fn db9_precheck_fold_candidates(
    expr: &Db9Expr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<()> {
    if db9_expr_is_fold_candidate(expr) {
        eval_expr(expr, table, row)?;
        return Ok(());
    }

    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::Binary(binary))
            if matches!(
                binary.get_op(),
                Db9BinaryOp::Db9BinaryOpAnd | Db9BinaryOp::Db9BinaryOpOr
            ) =>
        {
            let _ = db9_bool_fold_short_circuit_value(
                binary.get_left(),
                binary.get_op(),
                binary.get_right(),
                table,
                row,
            )?;
            Ok(())
        }
        Some(Db9Expr_oneof_expr::Binary(binary)) => {
            db9_precheck_fold_candidates(binary.get_left(), table, row)?;
            db9_precheck_fold_candidates(binary.get_right(), table, row)
        }
        Some(Db9Expr_oneof_expr::Unary(unary)) => {
            db9_precheck_fold_candidates(unary.get_operand(), table, row)
        }
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => {
            db9_precheck_fold_candidates(is_null.get_expr(), table, row)
        }
        Some(Db9Expr_oneof_expr::Cast(cast)) => {
            db9_precheck_fold_candidates(cast.get_expr(), table, row)
        }
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => {
            if func_call
                .get_function_name()
                .eq_ignore_ascii_case("coalesce")
            {
                db9_precheck_coalesce_fold_candidates(func_call.get_args(), table, row)
            } else {
                for arg in func_call.get_args() {
                    db9_precheck_fold_candidates(arg, table, row)?;
                }
                Ok(())
            }
        }
        Some(Db9Expr_oneof_expr::Constant(_)) | Some(Db9Expr_oneof_expr::ColumnRef(_)) | None => {
            Ok(())
        }
    }
}

fn db9_precheck_coalesce_fold_candidates(
    args: &[Db9Expr],
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<()> {
    for arg in args {
        if db9_expr_is_fold_candidate(arg) {
            let value = eval_expr(arg, table, row)?;
            if !value.is_null() {
                break;
            }
        } else {
            db9_precheck_fold_candidates(arg, table, row)?;
        }
    }
    Ok(())
}

fn db9_bool_fold_short_circuit_value(
    left: &Db9Expr,
    op: Db9BinaryOp,
    right: &Db9Expr,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<Option<StoredValue>> {
    if !matches!(op, Db9BinaryOp::Db9BinaryOpAnd | Db9BinaryOp::Db9BinaryOpOr) {
        return Ok(None);
    }
    if let Some(value) = db9_bool_fold_arg(left, op, table, row)? {
        return Ok(Some(value));
    }
    db9_bool_fold_arg(right, op, table, row)
}

fn db9_bool_fold_arg(
    expr: &Db9Expr,
    op: Db9BinaryOp,
    table: &Db9TableInfo,
    row: &StoredRow,
) -> Result<Option<StoredValue>> {
    if let Some(Db9Expr_oneof_expr::Binary(binary)) = expr.expr.as_ref() {
        if binary.get_op() == op {
            return db9_bool_fold_short_circuit_value(
                binary.get_left(),
                op,
                binary.get_right(),
                table,
                row,
            );
        }
    }

    if db9_expr_is_fold_candidate(expr) {
        let value = eval_expr(expr, table, row)?;
        let value_bool = value.as_sql_bool()?;
        let determines = matches!(
            (op, value_bool),
            (Db9BinaryOp::Db9BinaryOpAnd, Some(false)) | (Db9BinaryOp::Db9BinaryOpOr, Some(true))
        );
        return Ok(determines.then_some(value));
    }

    db9_precheck_fold_candidates(expr, table, row)?;
    Ok(None)
}

fn json_recursion_depth_error() -> Error {
    Error::Other("json value is too deep".to_owned())
}

fn jsonb_values_equal(left: &str, right: &str) -> Result<bool> {
    let left = parse_jsonb_comparable_value(left)?;
    let right = parse_jsonb_comparable_value(right)?;
    jsonb_comparable_eq(&left, &right, 0)
}

enum JsonbComparableValue {
    Null,
    Bool(bool),
    Number(bigdecimal::BigDecimal),
    String(String),
    Array(Vec<JsonbComparableValue>),
    Object(BTreeMap<String, JsonbComparableValue>),
}

fn parse_jsonb_comparable_value(raw: &str) -> Result<JsonbComparableValue> {
    parse_jsonb_comparable_value_inner(raw, 0)
}

fn parse_jsonb_comparable_value_inner(raw: &str, depth: usize) -> Result<JsonbComparableValue> {
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Err(Error::Other("Invalid JSONB value: empty input".to_owned()));
    }

    match trimmed.as_bytes()[0] {
        b'n' if trimmed == "null" => Ok(JsonbComparableValue::Null),
        b't' if trimmed == "true" => Ok(JsonbComparableValue::Bool(true)),
        b'f' if trimmed == "false" => Ok(JsonbComparableValue::Bool(false)),
        b'"' => {
            let value = serde_json::from_str(trimmed)
                .map_err(|err| Error::Other(format!("Invalid JSONB value: {err}")))?;
            Ok(JsonbComparableValue::String(value))
        }
        b'[' => {
            if depth >= MAX_ARRAY_RECURSION_DEPTH {
                return Err(json_recursion_depth_error());
            }
            let values = serde_json::from_str::<Vec<Box<RawValue>>>(trimmed)
                .map_err(|err| Error::Other(format!("Invalid JSONB value: {err}")))?;
            values
                .into_iter()
                .map(|value| parse_jsonb_comparable_value_inner(value.get(), depth + 1))
                .collect::<Result<Vec<_>>>()
                .map(JsonbComparableValue::Array)
        }
        b'{' => {
            if depth >= MAX_ARRAY_RECURSION_DEPTH {
                return Err(json_recursion_depth_error());
            }
            let values = serde_json::from_str::<BTreeMap<String, Box<RawValue>>>(trimmed)
                .map_err(|err| Error::Other(format!("Invalid JSONB value: {err}")))?;
            values
                .into_iter()
                .map(|(key, value)| {
                    Ok((
                        key,
                        parse_jsonb_comparable_value_inner(value.get(), depth + 1)?,
                    ))
                })
                .collect::<Result<BTreeMap<_, _>>>()
                .map(JsonbComparableValue::Object)
        }
        b'-' | b'0'..=b'9' => {
            let _: serde_json::Number = serde_json::from_str(trimmed)
                .map_err(|err| Error::Other(format!("Invalid JSONB value: {err}")))?;
            bigdecimal::BigDecimal::from_str(trimmed)
                .map(JsonbComparableValue::Number)
                .map_err(|err| Error::Other(format!("Invalid JSONB value: {err}")))
        }
        _ => Err(Error::Other(format!("Invalid JSONB value: {trimmed}"))),
    }
}

fn jsonb_comparable_eq(
    left: &JsonbComparableValue,
    right: &JsonbComparableValue,
    depth: usize,
) -> Result<bool> {
    match (left, right) {
        (JsonbComparableValue::Null, JsonbComparableValue::Null) => Ok(true),
        (JsonbComparableValue::Bool(left), JsonbComparableValue::Bool(right)) => Ok(left == right),
        (JsonbComparableValue::Number(left), JsonbComparableValue::Number(right)) => {
            Ok(left == right)
        }
        (JsonbComparableValue::String(left), JsonbComparableValue::String(right)) => {
            Ok(left == right)
        }
        (JsonbComparableValue::Array(left), JsonbComparableValue::Array(right)) => {
            if depth >= MAX_ARRAY_RECURSION_DEPTH {
                return Err(json_recursion_depth_error());
            }
            if left.len() != right.len() {
                return Ok(false);
            }
            for (left, right) in left.iter().zip(right.iter()) {
                if !jsonb_comparable_eq(left, right, depth + 1)? {
                    return Ok(false);
                }
            }
            Ok(true)
        }
        (JsonbComparableValue::Object(left), JsonbComparableValue::Object(right)) => {
            if depth >= MAX_ARRAY_RECURSION_DEPTH {
                return Err(json_recursion_depth_error());
            }
            if left.len() != right.len() {
                return Ok(false);
            }
            for (key, left) in left.iter() {
                let Some(right) = right.get(key) else {
                    return Ok(false);
                };
                if !jsonb_comparable_eq(left, right, depth + 1)? {
                    return Ok(false);
                }
            }
            Ok(true)
        }
        _ => Ok(false),
    }
}

fn values_equal(left: &StoredValue, right: &StoredValue) -> Result<Option<bool>> {
    if left.is_null() || right.is_null() {
        return Ok(None);
    }

    let value = match (left, right) {
        (StoredValue::Boolean(left), StoredValue::Boolean(right)) => left == right,
        (StoredValue::Int32(left), StoredValue::Int32(right)) => left == right,
        (StoredValue::Int32(left), StoredValue::Int64(right)) => i64::from(*left) == *right,
        (StoredValue::Int64(left), StoredValue::Int32(right)) => *left == i64::from(*right),
        (StoredValue::Int64(left), StoredValue::Int64(right)) => left == right,
        (StoredValue::Int32(left), StoredValue::Float64(right)) => {
            int_float_equal(i64::from(*left), *right)?
        }
        (StoredValue::Float64(left), StoredValue::Int32(right)) => {
            int_float_equal(i64::from(*right), *left)?
        }
        (StoredValue::Int64(left), StoredValue::Float64(right)) => int_float_equal(*left, *right)?,
        (StoredValue::Float64(left), StoredValue::Int64(right)) => int_float_equal(*right, *left)?,
        (StoredValue::Float64(left), StoredValue::Float64(right)) => {
            compare_float64_pg(*left, *right) == Ordering::Equal
        }
        (StoredValue::Text(left), StoredValue::Text(right)) => left == right,
        (StoredValue::Bytes(left), StoredValue::Bytes(right)) => left == right,
        (StoredValue::Timestamp(left), StoredValue::Timestamp(right)) => left == right,
        (StoredValue::Time(left), StoredValue::Time(right)) => left == right,
        (StoredValue::Date(left), StoredValue::Date(right)) => left == right,
        (StoredValue::Uuid(left), StoredValue::Uuid(right)) => left == right,
        (StoredValue::Json(left), StoredValue::Json(right)) => left == right,
        (StoredValue::Jsonb(left), StoredValue::Jsonb(right)) => jsonb_values_equal(left, right)?,
        (StoredValue::Numeric(left), StoredValue::Numeric(right)) => {
            compare_stored_decimals(left, right) == Ordering::Equal
        }
        (StoredValue::Float64(left), StoredValue::Numeric(_)) if left.is_infinite() => false,
        (StoredValue::Numeric(_), StoredValue::Float64(right)) if right.is_infinite() => false,
        (StoredValue::Array(left), StoredValue::Array(right)) => array_values_equal(left, right)?,
        _ => {
            return Err(Error::Other(format!(
                "DB9 cannot compare {} with {}",
                left.kind_name(),
                right.kind_name()
            )));
        }
    };

    Ok(Some(value))
}

fn compare_values(left: &StoredValue, right: &StoredValue) -> Result<Option<Ordering>> {
    if left.is_null() || right.is_null() {
        return Ok(None);
    }

    let ordering = match (left, right) {
        (StoredValue::Boolean(left), StoredValue::Boolean(right)) => left.cmp(right),
        (StoredValue::Int32(left), StoredValue::Int32(right)) => left.cmp(right),
        (StoredValue::Int32(left), StoredValue::Int64(right)) => i64::from(*left).cmp(right),
        (StoredValue::Int64(left), StoredValue::Int32(right)) => left.cmp(&i64::from(*right)),
        (StoredValue::Int64(left), StoredValue::Int64(right)) => left.cmp(right),
        (StoredValue::Int32(left), StoredValue::Float64(right)) => {
            compare_int64_and_float64(i64::from(*left), *right)?
        }
        (StoredValue::Float64(left), StoredValue::Int32(right)) => {
            compare_int64_and_float64(i64::from(*right), *left)?.reverse()
        }
        (StoredValue::Int64(left), StoredValue::Float64(right)) => {
            compare_int64_and_float64(*left, *right)?
        }
        (StoredValue::Float64(left), StoredValue::Int64(right)) => {
            compare_int64_and_float64(*right, *left)?.reverse()
        }
        (StoredValue::Float64(left), StoredValue::Float64(right)) => {
            compare_float64_pg(*left, *right)
        }
        (StoredValue::Text(left), StoredValue::Text(right)) => left.cmp(right),
        (StoredValue::Bytes(left), StoredValue::Bytes(right)) => left.cmp(right),
        (StoredValue::Timestamp(left), StoredValue::Timestamp(right)) => left.cmp(right),
        (StoredValue::Time(left), StoredValue::Time(right)) => left.cmp(right),
        (StoredValue::Date(left), StoredValue::Date(right)) => left.cmp(right),
        (StoredValue::Uuid(left), StoredValue::Uuid(right)) => left.cmp(right),
        (StoredValue::Numeric(left), StoredValue::Numeric(right)) => {
            compare_stored_decimals(left, right)
        }
        (StoredValue::Float64(left), StoredValue::Numeric(_)) if left.is_infinite() => {
            if left.is_sign_positive() {
                Ordering::Greater
            } else {
                Ordering::Less
            }
        }
        (StoredValue::Numeric(_), StoredValue::Float64(right)) if right.is_infinite() => {
            if right.is_sign_positive() {
                Ordering::Less
            } else {
                Ordering::Greater
            }
        }
        (StoredValue::Array(left), StoredValue::Array(right)) => {
            let min_len = left.len().min(right.len());
            let mut ordering = Ordering::Equal;
            for idx in 0..min_len {
                ordering = compare_array_elements(&left[idx], &right[idx])?;
                if ordering != Ordering::Equal {
                    break;
                }
            }
            if ordering == Ordering::Equal {
                left.len().cmp(&right.len())
            } else {
                ordering
            }
        }
        _ => {
            return Err(Error::Other(format!(
                "DB9 cannot order-compare {} with {}",
                left.kind_name(),
                right.kind_name()
            )));
        }
    };

    Ok(Some(ordering))
}

fn int_float_equal(left: i64, right: f64) -> Result<bool> {
    if right.is_nan() {
        return Ok(false);
    }
    Ok(compare_int64_and_float64(left, right)? == Ordering::Equal)
}

fn compare_float64_pg(left: f64, right: f64) -> Ordering {
    match (left.is_nan(), right.is_nan()) {
        (true, true) => Ordering::Equal,
        (true, false) => Ordering::Greater,
        (false, true) => Ordering::Less,
        (false, false) => left
            .partial_cmp(&right)
            .expect("non-NaN floats must be comparable"),
    }
}

fn compare_int64_and_float64(left: i64, right: f64) -> Result<Ordering> {
    if right.is_nan() {
        return Ok(Ordering::Less);
    }
    if right.is_infinite() {
        return Ok(if right.is_sign_positive() {
            Ordering::Less
        } else {
            Ordering::Greater
        });
    }
    if right == 0.0 {
        return Ok(left.cmp(&0));
    }

    let left_is_negative = left < 0;
    let right_is_negative = right.is_sign_negative();
    if left_is_negative != right_is_negative {
        return Ok(if left_is_negative {
            Ordering::Less
        } else {
            Ordering::Greater
        });
    }

    let magnitude_ordering = compare_uint64_and_positive_float64(left.unsigned_abs(), right.abs());
    Ok(if left_is_negative {
        magnitude_ordering.reverse()
    } else {
        magnitude_ordering
    })
}

fn compare_uint64_and_positive_float64(left: u64, right: f64) -> Ordering {
    let bits = right.to_bits();
    let exponent_bits = ((bits >> 52) & 0x7ff) as i32;
    let mantissa = bits & ((1_u64 << 52) - 1);
    let (significand, exponent) = if exponent_bits == 0 {
        (mantissa, 1 - 1023 - 52)
    } else {
        ((1_u64 << 52) | mantissa, exponent_bits - 1023 - 52)
    };

    if exponent >= 0 {
        if exponent > 11 {
            return Ordering::Less;
        }
        let right_int = (significand as u128) << exponent as u32;
        return (left as u128).cmp(&right_int);
    }

    let shift = (-exponent) as u32;
    if shift > 63 {
        return if left == 0 {
            Ordering::Less
        } else {
            Ordering::Greater
        };
    }

    ((left as u128) << shift).cmp(&(significand as u128))
}

fn decode_wire_value(value: &Db9Value, expected_type: &Db9Type) -> Result<StoredValue> {
    if value.has_null_value() || value.kind.is_none() {
        return Ok(StoredValue::Null);
    }
    if value.has_bool_value() {
        return Ok(StoredValue::Boolean(value.get_bool_value()));
    }
    if value.has_int32_value() {
        return Ok(match expected_type.get_kind() {
            Db9TypeKind::Db9TypeKindDate => StoredValue::Date(value.get_int32_value()),
            _ => StoredValue::Int32(value.get_int32_value()),
        });
    }
    if value.has_int64_value() {
        return Ok(match expected_type.get_kind() {
            Db9TypeKind::Db9TypeKindTimestamp | Db9TypeKind::Db9TypeKindTimestamptz => {
                StoredValue::Timestamp(value.get_int64_value())
            }
            Db9TypeKind::Db9TypeKindTime => StoredValue::Time(value.get_int64_value()),
            _ => StoredValue::Int64(value.get_int64_value()),
        });
    }
    if value.has_float64_value() {
        return Ok(StoredValue::Float64(value.get_float64_value()));
    }
    if value.has_text_value() {
        return decode_text_wire_value(value.get_text_value(), expected_type);
    }
    if value.has_bytes_value() {
        return match expected_type.get_kind() {
            Db9TypeKind::Db9TypeKindUuid => {
                let raw = value.get_bytes_value();
                let bytes: [u8; 16] = raw.try_into().map_err(|_| {
                    Error::Other(format!("invalid DB9 uuid payload length {}", raw.len()))
                })?;
                Ok(StoredValue::Uuid(bytes))
            }
            Db9TypeKind::Db9TypeKindJsonb => Ok(StoredValue::Jsonb(decode_jsonb_wire_bytes(
                value.get_bytes_value(),
            )?)),
            _ => Ok(StoredValue::Bytes(value.get_bytes_value().to_vec())),
        };
    }
    if value.has_timestamp_value() {
        return Ok(StoredValue::Timestamp(value.get_timestamp_value()));
    }
    if value.has_timestamptz_value() {
        return Ok(StoredValue::Timestamp(value.get_timestamptz_value()));
    }
    if value.has_interval_value() {
        let interval = value.get_interval_value();
        return Ok(StoredValue::Interval(StoredIntervalValue {
            months: interval.get_months(),
            millis: interval.get_millis(),
        }));
    }
    Err(Error::Other(
        "DB9 constant contains an unsupported wire value".to_owned(),
    ))
}

fn timestamp_millis_to_date_days(ts_millis: i64) -> Result<i32> {
    if ts_millis == i64::MAX {
        return Ok(DATE_POS_INFINITY_DAYS);
    }
    if ts_millis == i64::MIN {
        return Ok(DATE_NEG_INFINITY_DAYS);
    }

    let dt = Utc
        .timestamp_millis_opt(ts_millis)
        .single()
        .ok_or_else(|| Error::Other("invalid DB9 timestamp value".to_owned()))?;
    naive_date_to_days(dt.date_naive())
}

fn parse_date_days(value: &str) -> Result<i32> {
    let trimmed = value.trim();
    if trimmed.eq_ignore_ascii_case("infinity") {
        return Ok(DATE_POS_INFINITY_DAYS);
    }
    if trimmed.eq_ignore_ascii_case("-infinity") {
        return Ok(DATE_NEG_INFINITY_DAYS);
    }

    let date = chrono::NaiveDate::parse_from_str(trimmed, "%Y-%m-%d")
        .map_err(|_| Error::Other(format!("invalid DB9 date literal '{value}'")))?;
    naive_date_to_days(date)
}

fn date_days_to_timestamp_millis(days: i32) -> Result<i64> {
    if days == DATE_POS_INFINITY_DAYS {
        return Ok(i64::MAX);
    }
    if days == DATE_NEG_INFINITY_DAYS {
        return Ok(i64::MIN);
    }

    let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1)
        .ok_or_else(|| Error::Other("failed to construct DB9 date epoch".to_owned()))?;
    let date = epoch
        .checked_add_signed(chrono::Duration::days(i64::from(days)))
        .ok_or_else(|| Error::Other("DB9 date is out of supported range".to_owned()))?;
    Ok(Utc
        .from_utc_datetime(
            &date
                .and_hms_opt(0, 0, 0)
                .ok_or_else(|| Error::Other("failed to construct DB9 datetime".to_owned()))?,
        )
        .timestamp_millis())
}

fn naive_date_to_days(date: chrono::NaiveDate) -> Result<i32> {
    let epoch = chrono::NaiveDate::from_ymd_opt(1970, 1, 1)
        .ok_or_else(|| Error::Other("failed to construct DB9 date epoch".to_owned()))?;
    let days = date.signed_duration_since(epoch).num_days();
    i32::try_from(days).map_err(|_| Error::Other("DB9 date is out of supported range".to_owned()))
}

fn db9_length_delimited_field_size(payload_size: u32) -> u64 {
    1 + protobuf::rt::compute_raw_varint32_size(payload_size) as u64 + payload_size as u64
}

#[derive(Default)]
struct Db9ResponseSizeTracker {
    rows_size: u64,
    warnings_size: u64,
}

impl Db9ResponseSizeTracker {
    fn with_response_metadata(response: &Db9SelectResponse) -> Self {
        let mut tracker = Self {
            rows_size: 0,
            warnings_size: 0,
        };
        for warning in response.get_warnings() {
            tracker.add_warning(warning);
        }
        tracker
    }

    fn add_row(&mut self, row: &Db9Row) {
        self.rows_size = self
            .rows_size
            .saturating_add(db9_length_delimited_field_size(row.compute_size()));
    }

    fn add_warning(&mut self, warning: &str) {
        self.warnings_size = self
            .warnings_size
            .saturating_add(db9_length_delimited_field_size(warning.len() as u32));
    }

    fn total_size(&self, response: &Db9SelectResponse) -> u64 {
        self.rows_size
            .saturating_add(db9_stats_field_size(response.get_stats()))
            .saturating_add(self.warnings_size)
    }

    fn total_size_with_extra_row(
        &self,
        stats: &kvproto::db9_coprocessor::Db9ExecStats,
        row: &Db9Row,
    ) -> u64 {
        self.rows_size
            .saturating_add(db9_length_delimited_field_size(row.compute_size()))
            .saturating_add(db9_stats_field_size(stats))
            .saturating_add(self.warnings_size)
    }
}

fn db9_stats_field_size(stats: &kvproto::db9_coprocessor::Db9ExecStats) -> u64 {
    let stats_size = stats.compute_size();
    if stats_size == 0 {
        0
    } else {
        db9_length_delimited_field_size(stats_size)
    }
}

fn ensure_db9_response_size_within_limit(response_size: u64, max_resp_size: u64) -> Result<()> {
    if response_size > max_resp_size {
        return Err(Error::Other(format!(
            "DB9 response size {} exceeds coprocessor max_resp_size {}",
            response_size, max_resp_size,
        )));
    }
    Ok(())
}

#[derive(Clone, Copy)]
enum Db9StringOutputContext {
    Other,
    Projection,
    Selection,
}

impl Db9StringOutputContext {
    fn as_str(self) -> &'static str {
        match self {
            Db9StringOutputContext::Other => "other",
            Db9StringOutputContext::Projection => "projection",
            Db9StringOutputContext::Selection => "selection",
        }
    }
}

struct Db9StringOutputByteLimitGuard<'a> {
    limit_cell: &'a Cell<usize>,
    previous_limit: usize,
    context_cell: &'a Cell<Db9StringOutputContext>,
    previous_context: Db9StringOutputContext,
}

impl Drop for Db9StringOutputByteLimitGuard<'_> {
    fn drop(&mut self) {
        self.limit_cell.set(self.previous_limit);
        self.context_cell.set(self.previous_context);
    }
}

fn with_db9_string_output_byte_limit_for<T>(
    max_resp_size: u64,
    context: Db9StringOutputContext,
    f: impl FnOnce() -> Result<T>,
) -> Result<T> {
    let limit = usize::try_from(max_resp_size).unwrap_or(usize::MAX);
    DB9_STRING_OUTPUT_BYTE_LIMIT.with(|limit_cell| {
        DB9_STRING_OUTPUT_CONTEXT.with(|context_cell| {
            let previous_limit = limit_cell.replace(limit);
            let previous_context = context_cell.replace(context);
            let _guard = Db9StringOutputByteLimitGuard {
                limit_cell,
                previous_limit,
                context_cell,
                previous_context,
            };
            f()
        })
    })
}

#[async_trait]
impl<S: Snapshot, A> RequestHandler for Db9DagHandler<S, A>
where
    A: RegionStorageAccessor<Storage = CloudStore<S>> + Send + Sync,
{
    async fn handle_request(&mut self) -> Result<MemoryTraceGuard<Response>> {
        let begin = Instant::now();
        let result = self.handle_request_inner().await;
        self.exec_stats.process_nanos = begin.elapsed().as_nanos().min(u64::MAX as u128) as u64;

        result
    }

    async fn handle_streaming_request(&mut self) -> Result<(Option<Response>, bool)> {
        self.handle_streaming_request_inner().await
    }

    fn collect_scan_statistics(&mut self, dest: &mut Statistics) {
        if let Some(row_fetch_store) = &mut self.row_fetch_store {
            dest.add(&row_fetch_store.incremental_get_take_statistics());
        }
        if let Some(scanner) = &mut self.scanner {
            scanner.collect_storage_stats(dest);
        }
    }

    fn collect_scan_summary(&mut self, dest: &mut ExecSummary) {
        self.exec_stats.collect_scan_summary(dest);
    }

    fn collect_executor_inputs(&self, dest: &mut kvrpcpb::ExecutorInputs) {
        self.exec_stats.collect_executor_inputs(
            self.uses_index_row_fetch(),
            self.req.has_selection(),
            dest,
        );
    }

    fn collect_executor_iterations(&self) -> u64 {
        self.exec_stats
            .executor_iterations(self.uses_index_row_fetch(), self.req.has_selection())
    }
}

fn db9_scanner_ranges(req_ranges: &[KeyRange], desc: bool) -> Vec<Range> {
    let mut key_ranges = req_ranges.to_vec();
    if desc {
        key_ranges.reverse();
    }
    key_ranges
        .into_iter()
        .map(|range| Range::from_pb_range(range, false))
        .collect()
}

impl<S: Snapshot, A> Db9DagHandler<S, A>
where
    A: RegionStorageAccessor<Storage = CloudStore<S>>,
{
    async fn handle_request_inner(&mut self) -> Result<MemoryTraceGuard<Response>> {
        self.validate_request()?;
        let limit = self.limit()?;
        let mut out = self
            .next_select_response_chunk(limit, false, true)
            .await?
            .unwrap_or_default();
        let resp = self.encode_select_response(&mut out)?;
        Ok(resp.into())
    }

    async fn next_select_response_chunk(
        &mut self,
        limit: Option<usize>,
        allow_chunk_boundary: bool,
        emit_empty: bool,
    ) -> Result<Option<Db9SelectResponse>> {
        let mut out = Db9SelectResponse::default();
        self.apply_exec_stats(&mut out);
        let mut response_size_tracker = Db9ResponseSizeTracker::with_response_metadata(&out);
        let uses_index_row_fetch = self.uses_index_row_fetch();

        if self.finished {
            return if emit_empty { Ok(Some(out)) } else { Ok(None) };
        }
        if matches!(limit, Some(0)) {
            self.finished = true;
            return if emit_empty { Ok(Some(out)) } else { Ok(None) };
        }

        if let Some(pending_row) = self.pending_projected_row.take() {
            self.append_projected_row(&mut out, &mut response_size_tracker, pending_row)?;
        }

        while !self.limit_reached(limit) {
            let next = {
                let scanner = self.scanner()?;
                scanner.next().await.map_err(Error::from)
            };
            let Some(kv) = next? else {
                self.finished = true;
                break;
            };
            let kv_bytes = kv.key().len().saturating_add(kv.value().len());
            self.exec_stats.read_bytes = self.exec_stats.read_bytes.saturating_add(kv_bytes);

            if !scan_key_matches_prefix(&self.scan_prefix, kv.key()) {
                continue;
            }

            {
                let stats = out.mut_stats();
                stats.set_scanned_keys(stats.get_scanned_keys() + 1);
            }
            self.exec_stats.scanned_keys = self.exec_stats.scanned_keys.saturating_add(1);

            let row = if uses_index_row_fetch {
                let row_bytes = self.fetch_row_for_index_entry(kv.key(), kv.value()).await?;
                let stats = out.mut_stats();
                stats.set_scanned_rows(stats.get_scanned_rows() + 1);
                self.exec_stats.scanned_rows = self.exec_stats.scanned_rows.saturating_add(1);
                self.exec_stats.read_bytes =
                    self.exec_stats.read_bytes.saturating_add(row_bytes.len());
                decode_stored_row(&row_bytes, self.max_stored_row_decode_bytes())?
            } else if db9_scan_kind_is_index(self.req.get_scan().get_kind()) {
                let stats = out.mut_stats();
                stats.set_scanned_rows(stats.get_scanned_rows() + 1);
                self.exec_stats.scanned_rows = self.exec_stats.scanned_rows.saturating_add(1);
                index_only_row_for_index_entry(&self.req, kv.key(), kv.value())?
            } else {
                let stats = out.mut_stats();
                stats.set_scanned_rows(stats.get_scanned_rows() + 1);
                self.exec_stats.scanned_rows = self.exec_stats.scanned_rows.saturating_add(1);
                decode_stored_row(kv.value(), self.max_stored_row_decode_bytes())?
            };
            {
                let stats = out.mut_stats();
                stats.set_decoded_rows(stats.get_decoded_rows() + 1);
            }
            self.exec_stats.decoded_rows = self.exec_stats.decoded_rows.saturating_add(1);

            let remaining_output_budget = self
                .max_resp_size
                .saturating_sub(response_size_tracker.total_size(&out));
            // Selection output is intermediate predicate state, not emitted response data.
            // Keep its dedicated limit stable for the whole request so predicate
            // evaluation does not depend on how full the current response batch
            // already is; projection uses the remaining response budget because
            // projected bytes are actually returned.
            let selection_output_budget =
                stable_selection_output_bytes_for_response(self.max_resp_size);

            if self.req.has_selection()
                && !row_matches_selection_with_output_limit(
                    selection_output_budget,
                    self.req.get_selection(),
                    self.req.get_table(),
                    &row,
                )?
            {
                continue;
            }

            let projection_output_budget = if allow_chunk_boundary {
                self.max_resp_size
            } else {
                remaining_output_budget
            };
            let projected_row = with_db9_string_output_byte_limit_for(
                projection_output_budget,
                Db9StringOutputContext::Projection,
                || project_row(&self.req, &row),
            )?;
            if allow_chunk_boundary
                && !out.get_rows().is_empty()
                && self.projected_response_size(&response_size_tracker, &projected_row)
                    > self.max_resp_size
            {
                self.pending_projected_row = Some(projected_row);
                break;
            }

            self.append_projected_row(&mut out, &mut response_size_tracker, projected_row)?;
        }

        if self.limit_reached(limit) {
            self.finished = true;
        }

        if out.get_rows().is_empty() && !emit_empty {
            return Ok(None);
        }
        self.apply_exec_stats(&mut out);
        Ok(Some(out))
    }

    fn limit_reached(&self, limit: Option<usize>) -> bool {
        limit.is_some_and(|limit| self.exec_stats.returned_rows as usize >= limit)
    }

    fn projected_response_size(
        &self,
        response_size_tracker: &Db9ResponseSizeTracker,
        row: &Db9Row,
    ) -> u64 {
        let mut candidate_stats = self.exec_stats.to_wire_stats();
        candidate_stats.set_returned_rows(candidate_stats.get_returned_rows() + 1);
        response_size_tracker.total_size_with_extra_row(&candidate_stats, row)
    }

    fn append_projected_row(
        &mut self,
        out: &mut Db9SelectResponse,
        response_size_tracker: &mut Db9ResponseSizeTracker,
        row: Db9Row,
    ) -> Result<()> {
        let response_size = self.projected_response_size(response_size_tracker, &row);
        ensure_db9_response_size_within_limit(response_size, self.max_resp_size)?;
        response_size_tracker.add_row(&row);
        out.mut_rows().push(row);
        self.exec_stats.returned_rows = self.exec_stats.returned_rows.saturating_add(1);
        self.apply_exec_stats(out);
        Ok(())
    }

    fn encode_select_response(&mut self, out: &mut Db9SelectResponse) -> Result<Response> {
        self.apply_exec_stats(out);
        let response_size = Db9ResponseSizeTracker::with_response_metadata(out).total_size(out);
        ensure_db9_response_size_within_limit(response_size, self.max_resp_size)?;
        let data = out
            .write_to_bytes()
            .map_err(|err| Error::Other(format!("failed to encode DB9 response: {err}")))?;
        let mut resp = Response::default();
        resp.set_data(data);
        Ok(resp)
    }

    async fn handle_streaming_request_inner(&mut self) -> Result<(Option<Response>, bool)> {
        self.validate_request()?;
        let limit = self.limit()?;
        let Some(mut out) = self.next_select_response_chunk(limit, true, false).await? else {
            return Ok((None, true));
        };
        let resp = self.encode_select_response(&mut out)?;
        Ok((
            Some(resp),
            self.finished && self.pending_projected_row.is_none(),
        ))
    }
}