// Copyright 2026 TiKV Project Authors. Licensed under Apache-2.0.

use async_trait::async_trait;

use super::*;

pub(crate) fn test_encode_table_data_key(db_id: u64, table_id: u64, row_key: &[u8]) -> Vec<u8> {
    encode_table_data_key(db_id, table_id, row_key)
}

pub(crate) fn test_encode_snapshot_scoped_txn_raw_key(
    snapshot_start_key: &[u8],
    raw_key: &[u8],
) -> Vec<u8> {
    encode_snapshot_scoped_txn_raw_key(snapshot_start_key, raw_key)
}

pub(crate) fn test_encode_index_key(
    db_id: u64,
    table_id: u64,
    index_id: u64,
    values: Vec<TestStoredValue>,
    pk_bytes: Option<&[u8]>,
) -> Vec<u8> {
    let mut key = encode_index_key_prefix(db_id, table_id, index_id);
    for value in values {
        test_encode_index_value(value, &mut key);
    }
    if let Some(pk_bytes) = pk_bytes {
        key.push(INDEX_ENTRY_PK_SENTINEL[0]);
        key.extend_from_slice(pk_bytes);
    }
    key
}

pub(crate) fn test_encode_pk_bytes(values: Vec<TestStoredValue>) -> Vec<u8> {
    let mut out = Vec::new();
    for value in values {
        test_encode_index_value(value, &mut out);
    }
    out
}

pub(crate) fn test_prefix_end(prefix: &[u8]) -> Vec<u8> {
    encode_prefix_end(prefix)
}

#[allow(dead_code)]
#[derive(Debug, Clone)]
pub(crate) enum TestStoredValue {
    Null,
    Boolean(bool),
    Int32(i32),
    Int64(i64),
    Float64(f64),
    Timestamp(i64),
    Time(i64),
    Date(i32),
    Text(String),
    Bytes(Vec<u8>),
}

impl From<TestStoredValue> for StoredValue {
    fn from(value: TestStoredValue) -> Self {
        match value {
            TestStoredValue::Null => StoredValue::Null,
            TestStoredValue::Boolean(value) => StoredValue::Boolean(value),
            TestStoredValue::Int32(value) => StoredValue::Int32(value),
            TestStoredValue::Int64(value) => StoredValue::Int64(value),
            TestStoredValue::Float64(value) => StoredValue::Float64(value),
            TestStoredValue::Timestamp(value) => StoredValue::Timestamp(value),
            TestStoredValue::Time(value) => StoredValue::Time(value),
            TestStoredValue::Date(value) => StoredValue::Date(value),
            TestStoredValue::Text(value) => StoredValue::Text(value),
            TestStoredValue::Bytes(value) => StoredValue::Bytes(value),
        }
    }
}

pub(crate) fn test_encode_row(values: Vec<TestStoredValue>) -> Vec<u8> {
    let row = StoredRow {
        values: values.into_iter().map(StoredValue::from).collect(),
    };
    bincode::serialize(&row).expect("DB9 test row encoding should succeed")
}

fn test_encode_index_value(value: TestStoredValue, out: &mut Vec<u8>) {
    use tikv_util::codec::number::NumberEncoder;

    match value {
        TestStoredValue::Null => out.push(0x00),
        TestStoredValue::Boolean(value) => {
            out.push(0x01);
            out.push(u8::from(value));
        }
        TestStoredValue::Int32(value) => {
            out.push(0x01);
            out.extend_from_slice(&(((value as u32) ^ 0x8000_0000).to_be_bytes()));
        }
        TestStoredValue::Int64(value) => {
            out.push(0x01);
            out.encode_i64(value)
                .expect("DB9 test int64 encoding should succeed");
        }
        TestStoredValue::Float64(value) => {
            out.push(0x01);
            out.encode_f64(value)
                .expect("DB9 test float64 encoding should succeed");
        }
        TestStoredValue::Timestamp(_) => {
            panic!("DB9 test timestamp values are not supported in index keys");
        }
        TestStoredValue::Time(_) => {
            panic!("DB9 test time values are not supported in index keys");
        }
        TestStoredValue::Date(_) => {
            panic!("DB9 test date values are not supported in index keys");
        }
        TestStoredValue::Text(value) => {
            out.push(0x01);
            test_encode_db9_var_bytes_payload(value.as_bytes(), out);
        }
        TestStoredValue::Bytes(value) => {
            out.push(0x01);
            test_encode_db9_var_bytes_payload(&value, out);
        }
    }
}

fn test_encode_db9_var_bytes_payload(value: &[u8], out: &mut Vec<u8>) {
    out.push(u8::from(!value.is_empty()));
    if value.is_empty() {
        return;
    }

    let mut encoded = 0;
    for chunk in value.chunks(8) {
        out.extend_from_slice(chunk);
        if chunk.len() != 8 {
            out.resize(out.len() + (8 - chunk.len()), 0);
        }
        encoded += chunk.len();
        out.push(if encoded == value.len() {
            chunk.len() as u8
        } else {
            9
        });
    }
}

fn encode_prefix_end(prefix: &[u8]) -> Vec<u8> {
    let mut end = prefix.to_vec();
    for idx in (0..end.len()).rev() {
        if end[idx] != 0xFF {
            end[idx] = end[idx].wrapping_add(1);
            end.truncate(idx + 1);
            return end;
        }
    }

    let mut end = prefix.to_vec();
    end.push(0xFF);
    end
}

#[test]
fn snapshot_scoped_txn_raw_key_adds_keyspace_prefix_from_snapshot_bounds() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();
    let snapshot_start_key = ApiV2::get_txn_keyspace_prefix(42);

    let scoped = encode_snapshot_scoped_txn_raw_key(&snapshot_start_key, &raw_key);

    let mut expected = snapshot_start_key;
    expected.extend_from_slice(&raw_key);
    assert_eq!(scoped, expected);
}

#[test]
fn snapshot_scoped_txn_raw_key_leaves_unprefixed_snapshots_unchanged() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();

    let scoped = encode_snapshot_scoped_txn_raw_key(b"", &raw_key);

    assert_eq!(scoped, raw_key);
}

#[derive(Clone)]
struct FixtureRegionStorageAccessor {
    lookup_raw_key: Vec<u8>,
    region: kvproto::metapb::Region,
    original_region_id: Option<u64>,
    data: std::collections::BTreeMap<Key, Vec<u8>>,
}

#[async_trait]
impl RegionStorageAccessor for FixtureRegionStorageAccessor {
    type Storage = crate::storage::txn::FixtureStore;

    async fn find_region_by_key(&self, key: &[u8]) -> FindRegionResult {
        assert_eq!(key, Key::from_raw(&self.lookup_raw_key).as_encoded());
        FindRegionResult::with_found(self.region.clone(), StateRole::Leader)
    }

    async fn get_local_region_storage(
        &self,
        region: &kvproto::metapb::Region,
        key_ranges: &[KeyRange],
    ) -> tidb_query_common::storage::Result<Self::Storage> {
        assert_eq!(region.get_id(), self.region.get_id());
        assert_eq!(key_ranges.len(), 1);
        assert_eq!(key_ranges[0].get_start(), self.lookup_raw_key.as_slice());
        let mut end = self.lookup_raw_key.clone();
        convert_to_prefix_next(&mut end);
        assert_eq!(key_ranges[0].get_end(), end.as_slice());
        Ok(crate::storage::txn::FixtureStore::new(
            self.data
                .iter()
                .map(|(key, value)| (key.clone(), Ok(value.clone())))
                .collect(),
        ))
    }

    fn get_original_region_id(&self) -> Option<u64> {
        self.original_region_id
    }
}

#[test]
fn codec_version_zero_is_rejected() {
    let err = validate_codec_version(0).unwrap_err();
    assert!(
        err.to_string()
            .contains("unsupported DB9 codec_version 0, expected 2")
    );
    validate_codec_version(DB9_CODEC_VERSION).unwrap();
}

#[test]
fn row_fetch_uses_region_storage_accessor() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();
    let mut data = std::collections::BTreeMap::new();
    data.insert(Key::from_raw(&raw_key), b"row".to_vec());
    let accessor = FixtureRegionStorageAccessor {
        lookup_raw_key: raw_key.clone(),
        region: kvproto::metapb::Region {
            id: 7,
            ..Default::default()
        },
        original_region_id: Some(7),
        data,
    };

    let (snapshot_row_key, row_bytes) =
        futures::executor::block_on(fetch_row_from_region_storage(&accessor, &raw_key, &raw_key))
            .unwrap();

    assert_eq!(snapshot_row_key, raw_key);
    assert_eq!(row_bytes, Some(b"row".to_vec()));
}

#[test]
fn row_fetch_uses_scoped_key_for_region_lookup() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();
    let mut lookup_raw_key = ApiV2::get_txn_keyspace_prefix(42);
    lookup_raw_key.extend_from_slice(&raw_key);
    let mut data = std::collections::BTreeMap::new();
    data.insert(Key::from_raw(&raw_key), b"row".to_vec());
    let accessor = FixtureRegionStorageAccessor {
        lookup_raw_key: lookup_raw_key.clone(),
        region: kvproto::metapb::Region {
            id: 9,
            ..Default::default()
        },
        original_region_id: Some(7),
        data,
    };

    let (snapshot_row_key, row_bytes) = futures::executor::block_on(fetch_row_from_region_storage(
        &accessor,
        &raw_key,
        &lookup_raw_key,
    ))
    .unwrap();

    assert_eq!(snapshot_row_key, raw_key);
    assert_eq!(row_bytes, Some(b"row".to_vec()));
}

#[derive(Clone)]
struct PanicRegionStorageAccessor;

#[async_trait]
impl RegionStorageAccessor for PanicRegionStorageAccessor {
    type Storage = crate::storage::txn::FixtureStore;

    async fn find_region_by_key(&self, _key: &[u8]) -> FindRegionResult {
        panic!("unexpected region lookup for local row fetch hit");
    }

    async fn get_local_region_storage(
        &self,
        _region: &kvproto::metapb::Region,
        _key_ranges: &[KeyRange],
    ) -> tidb_query_common::storage::Result<Self::Storage> {
        panic!("unexpected region storage fetch for local row fetch hit");
    }
}

#[test]
fn row_fetch_prefers_local_store_before_region_storage() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();
    let mut data = std::collections::BTreeMap::new();
    data.insert(Key::from_raw(&raw_key), Ok(b"row".to_vec()));
    let mut store = crate::storage::txn::FixtureStore::new(data);

    let (snapshot_row_key, row_bytes) =
        futures::executor::block_on(fetch_row_from_store_or_region_storage(
            &mut store,
            Some(&PanicRegionStorageAccessor),
            &raw_key,
        ))
        .unwrap();

    assert_eq!(snapshot_row_key, raw_key);
    assert_eq!(row_bytes, Some(b"row".to_vec()));
}

#[test]
fn row_fetch_allows_cross_region_storage_accessor() {
    let raw_key = b"d_\0\0\0\0\0\0\0\x01_t_demo".to_vec();
    let mut data = std::collections::BTreeMap::new();
    data.insert(Key::from_raw(&raw_key), b"row".to_vec());
    let accessor = FixtureRegionStorageAccessor {
        lookup_raw_key: raw_key.clone(),
        region: kvproto::metapb::Region {
            id: 8,
            ..Default::default()
        },
        original_region_id: Some(7),
        data,
    };

    let (snapshot_row_key, row_bytes) =
        futures::executor::block_on(fetch_row_from_region_storage(&accessor, &raw_key, &raw_key))
            .unwrap();

    assert_eq!(snapshot_row_key, raw_key);
    assert_eq!(row_bytes, Some(b"row".to_vec()));
}

#[test]
fn response_size_tracker_matches_protobuf_compute_size() {
    let mut response = Db9SelectResponse::default();
    response.mut_stats().set_scanned_keys(3);
    response.mut_stats().set_scanned_rows(2);
    response.mut_stats().set_decoded_rows(2);
    response.mut_stats().set_returned_rows(2);
    response.mut_warnings().push("warn-1".to_owned());
    response.mut_warnings().push("warn-2".to_owned());

    let mut tracker = Db9ResponseSizeTracker::with_response_metadata(&response);

    let mut first_value = Db9Value::default();
    first_value.set_text_value("alpha".to_owned());
    let mut first_row = Db9Row::default();
    first_row.mut_values().push(first_value);
    tracker.add_row(&first_row);
    response.mut_rows().push(first_row);

    let mut second_value = Db9Value::default();
    second_value.set_int64_value(42);
    let mut second_row = Db9Row::default();
    second_row.mut_values().push(second_value);
    tracker.add_row(&second_row);
    response.mut_rows().push(second_row);

    let total_size = tracker.total_size(&response);
    assert_eq!(total_size, response.compute_size() as u64);
    assert!(response.compute_size() as u64 > total_size - 1);
}

#[test]
fn response_size_tracker_incrementally_accounts_for_warnings() {
    let response = Db9SelectResponse::default();
    let mut tracker = Db9ResponseSizeTracker::with_response_metadata(&response);
    tracker.add_warning("warn-1");
    tracker.add_warning("warn-2");

    let expected = db9_length_delimited_field_size("warn-1".len() as u32)
        .saturating_add(db9_length_delimited_field_size("warn-2".len() as u32));
    assert_eq!(tracker.warnings_size, expected);
}

#[test]
fn response_size_limit_error_uses_db9_cop_mapper_contract_message() {
    let err = ensure_db9_response_size_within_limit(1025, 1024).unwrap_err();
    assert_eq!(
        err.to_string(),
        "DB9 response size 1025 exceeds coprocessor max_resp_size 1024"
    );
}

fn test_db9_type(kind: Db9TypeKind) -> Db9Type {
    let mut data_type = Db9Type::default();
    data_type.set_kind(kind);
    data_type
}

fn test_db9_array_type(elem_kind: Db9TypeKind) -> Db9Type {
    let mut data_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    data_type.set_elem_type(test_db9_type(elem_kind));
    data_type
}

fn test_db9_array_column_expr(index: u32, elem_kind: Db9TypeKind) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_array_type(elem_kind));
    let mut column = kvproto::db9_coprocessor::Db9ColumnRefExpr::default();
    column.set_column_index(index);
    column.set_column_name(format!("c{index}"));
    expr.set_column_ref(column);
    expr
}

fn test_db9_array_func_expr(
    function_name: &str,
    return_type: Db9Type,
    args: Vec<Db9Expr>,
) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(return_type);
    expr.set_func_call(test_db9_func_call(function_name, args));
    expr
}

fn test_db9_text_const_expr(value: &str) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindText));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_text_value(value.to_owned());
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_bool_const_expr(value: bool) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindBoolean));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_bool_value(value);
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_int32_const_expr(value: i32) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindInt32));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_int32_value(value);
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_float64_const_expr(value: f64) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindFloat64));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_float64_value(value);
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_null_const_expr(kind: Db9TypeKind) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(kind));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_null_value(Db9Null::default());
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_timestamp_const_expr(value: i64) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindTimestamp));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_timestamp_value(value);
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_date_const_expr(value: i32) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindDate));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_int32_value(value);
    constant.set_value(wire);
    expr.set_constant(constant);
    expr
}

fn test_db9_func_call(function_name: &str, args: Vec<Db9Expr>) -> Db9FuncCallExpr {
    let mut func_call = Db9FuncCallExpr::default();
    func_call.set_function_name(function_name.to_owned());
    func_call.set_args(args.into());
    func_call
}

fn test_db9_func_expr(
    function_name: &str,
    return_kind: Db9TypeKind,
    args: Vec<Db9Expr>,
) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(return_kind));
    expr.set_func_call(test_db9_func_call(function_name, args));
    expr
}

fn test_db9_binary_expr(
    op: Db9BinaryOp,
    return_kind: Db9TypeKind,
    left: Db9Expr,
    right: Db9Expr,
) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(return_kind));
    let mut binary = kvproto::db9_coprocessor::Db9BinaryExpr::default();
    binary.set_left(left);
    binary.set_op(op);
    binary.set_right(right);
    expr.set_binary(binary);
    expr
}

fn test_db9_projection_request(expr: Db9Expr) -> Db9DagRequest {
    let mut request = Db9DagRequest::default();
    request.set_table(test_db9_table_for_expr(&expr));
    let mut projection = Db9NamedExpr::default();
    projection.set_output_name("out".to_owned());
    projection.set_expr(expr);
    request.mut_projections().push(projection);
    request
}

fn test_db9_table_for_expr(expr: &Db9Expr) -> Db9TableInfo {
    let mut columns = Vec::new();
    collect_test_db9_column_refs(expr, &mut columns);

    let mut table = Db9TableInfo::default();
    for (idx, column) in columns.into_iter().enumerate() {
        let (name, data_type) = column.unwrap_or_else(|| {
            (
                format!("c{idx}"),
                test_db9_type(Db9TypeKind::Db9TypeKindInt64),
            )
        });
        let mut column_info = kvproto::db9_coprocessor::Db9ColumnInfo::default();
        column_info.set_name(name);
        column_info.set_data_type(data_type);
        table.mut_columns().push(column_info);
    }
    table
}

fn collect_test_db9_column_refs(expr: &Db9Expr, columns: &mut Vec<Option<(String, Db9Type)>>) {
    match expr.expr.as_ref() {
        Some(Db9Expr_oneof_expr::ColumnRef(column_ref)) => {
            let index = column_ref.get_column_index() as usize;
            if columns.len() <= index {
                columns.resize_with(index + 1, || None);
            }
            columns[index].get_or_insert_with(|| {
                (
                    column_ref.get_column_name().to_owned(),
                    expr.get_return_type().clone(),
                )
            });
        }
        Some(Db9Expr_oneof_expr::Binary(binary)) => {
            collect_test_db9_column_refs(binary.get_left(), columns);
            collect_test_db9_column_refs(binary.get_right(), columns);
        }
        Some(Db9Expr_oneof_expr::Unary(unary)) => {
            collect_test_db9_column_refs(unary.get_operand(), columns);
        }
        Some(Db9Expr_oneof_expr::Cast(cast)) => {
            collect_test_db9_column_refs(cast.get_expr(), columns);
        }
        Some(Db9Expr_oneof_expr::FuncCall(func_call)) => {
            for arg in func_call.get_args() {
                collect_test_db9_column_refs(arg, columns);
            }
        }
        Some(Db9Expr_oneof_expr::IsNull(is_null)) => {
            collect_test_db9_column_refs(is_null.get_expr(), columns);
        }
        Some(Db9Expr_oneof_expr::Constant(_)) | None => {}
    }
}

#[test]
fn request_surface_rejects_leaf_type_confusion_before_execution() {
    let mut bad_constant = test_db9_text_const_expr("not-used");
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut wire = Db9Value::default();
    wire.set_int64_value(7);
    constant.set_value(wire);
    bad_constant.set_constant(constant);

    let err = validate_db9_request_expression_surface(&test_db9_projection_request(bad_constant))
        .unwrap_err();
    assert!(
        err.to_string()
            .contains("does not match declared return type"),
        "unexpected error: {}",
        err
    );

    let mut request =
        test_db9_projection_request(test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText));
    request.set_table(test_db9_table(&[Db9TypeKind::Db9TypeKindInt64]));
    let err = validate_db9_request_expression_surface(&request).unwrap_err();
    assert!(
        err.to_string().contains("does not match table column type"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_rejects_unsupported_output_types_before_execution() {
    let mut full_row = Db9DagRequest::default();
    full_row.set_table(test_db9_table(&[Db9TypeKind::Db9TypeKindTsvector]));
    let err = validate_db9_request_expression_surface(&full_row).unwrap_err();
    assert!(
        err.to_string().contains("not supported"),
        "unexpected error: {}",
        err
    );

    let mut expr = test_db9_text_const_expr("a");
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindTsquery));
    let err =
        validate_db9_request_expression_surface(&test_db9_projection_request(expr)).unwrap_err();
    assert!(
        err.to_string().contains("not supported"),
        "unexpected error: {}",
        err
    );

    for kind in [
        Db9TypeKind::Db9TypeKindUuid,
        Db9TypeKind::Db9TypeKindJson,
        Db9TypeKind::Db9TypeKindJsonb,
    ] {
        let mut full_row = Db9DagRequest::default();
        full_row.set_table(test_db9_table(&[kind]));
        let err = validate_db9_request_expression_surface(&full_row).unwrap_err();
        assert!(
            err.to_string().contains("response output type"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }

    let unsupported_projection =
        test_db9_projection_request(test_db9_column_expr(0, Db9TypeKind::Db9TypeKindUuid));
    let err = validate_db9_request_expression_surface(&unsupported_projection).unwrap_err();
    assert!(
        err.to_string().contains("response output type"),
        "unexpected error: {}",
        err
    );

    for kind in [
        Db9TypeKind::Db9TypeKindNumeric,
        Db9TypeKind::Db9TypeKindUuid,
        Db9TypeKind::Db9TypeKindJson,
        Db9TypeKind::Db9TypeKindJsonb,
    ] {
        let err = validate_db9_request_expression_surface(&test_db9_projection_request(
            test_db9_array_column_expr(0, kind),
        ))
        .unwrap_err();
        assert!(
            err.to_string().contains("response output type"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }
}

#[test]
fn request_surface_rejects_is_null_return_type_mismatch() {
    let mut is_null_expr = Db9Expr::default();
    is_null_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindText));
    let mut is_null = kvproto::db9_coprocessor::Db9IsNullExpr::default();
    is_null.set_expr(test_db9_text_const_expr("a"));
    is_null_expr.set_is_null(is_null);

    let err = validate_db9_request_expression_surface(&test_db9_projection_request(is_null_expr))
        .unwrap_err();
    assert!(
        err.to_string().contains("must return BOOLEAN"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_rejects_unpaired_functions_and_operators() {
    for (function_name, return_kind) in [
        ("lower", Db9TypeKind::Db9TypeKindText),
        ("upper", Db9TypeKind::Db9TypeKindText),
        ("concat", Db9TypeKind::Db9TypeKindText),
        ("concat_ws", Db9TypeKind::Db9TypeKindText),
        ("lpad", Db9TypeKind::Db9TypeKindText),
        ("format", Db9TypeKind::Db9TypeKindText),
        ("array_to_string", Db9TypeKind::Db9TypeKindText),
        ("encode", Db9TypeKind::Db9TypeKindText),
        ("json_build_object", Db9TypeKind::Db9TypeKindJson),
        ("jsonb_build_object", Db9TypeKind::Db9TypeKindJsonb),
        ("json_build_array", Db9TypeKind::Db9TypeKindJson),
        ("jsonb_build_array", Db9TypeKind::Db9TypeKindJsonb),
        ("__db9_array_contains", Db9TypeKind::Db9TypeKindBoolean),
        ("__db9_array_contained_by", Db9TypeKind::Db9TypeKindBoolean),
        ("__db9_array_overlap", Db9TypeKind::Db9TypeKindBoolean),
        ("regexp_replace", Db9TypeKind::Db9TypeKindText),
        ("regexp_match", Db9TypeKind::Db9TypeKindArray),
        ("regexp_matches", Db9TypeKind::Db9TypeKindArray),
        ("regexp_split_to_array", Db9TypeKind::Db9TypeKindArray),
        ("starts_with", Db9TypeKind::Db9TypeKindBoolean),
        ("to_hex", Db9TypeKind::Db9TypeKindText),
        ("random", Db9TypeKind::Db9TypeKindFloat64),
        ("clock_timestamp", Db9TypeKind::Db9TypeKindTimestamptz),
        ("gen_random_uuid", Db9TypeKind::Db9TypeKindUuid),
        ("uuid_generate_v4", Db9TypeKind::Db9TypeKindUuid),
        ("uuidv7", Db9TypeKind::Db9TypeKindUuid),
        ("db9_unknown_function", Db9TypeKind::Db9TypeKindText),
    ] {
        let return_type = match return_kind {
            Db9TypeKind::Db9TypeKindArray => test_db9_array_type(Db9TypeKind::Db9TypeKindText),
            _ => test_db9_type(return_kind),
        };
        let request = test_db9_projection_request(test_db9_array_func_expr(
            function_name,
            return_type,
            vec![],
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains(function_name),
            "unexpected error for {function_name}: {}",
            err
        );
        assert!(
            err.to_string()
                .contains("not supported by this codec surface"),
            "unexpected error for {function_name}: {}",
            err
        );
    }
}

#[test]
fn request_surface_rejects_malformed_supported_function_signatures() {
    for request in [
        test_db9_projection_request(test_db9_func_expr(
            "__db9_in_list",
            Db9TypeKind::Db9TypeKindBoolean,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt64)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "abs",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "sign",
            Db9TypeKind::Db9TypeKindInt32,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "sqrt",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindNumeric)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "cbrt",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindNumeric)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "length",
            Db9TypeKind::Db9TypeKindInt32,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindUserDefined)],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "width_bucket",
            Db9TypeKind::Db9TypeKindInt32,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindFloat64),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindFloat64),
                test_db9_column_expr(2, Db9TypeKind::Db9TypeKindFloat64),
                test_db9_column_expr(3, Db9TypeKind::Db9TypeKindInt64),
            ],
        )),
    ] {
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error: {}",
            err
        );
    }
}

#[test]
fn request_surface_rejects_local_only_array_carriers() {
    for function_name in ["array_position", "array_remove"] {
        let return_type = if function_name == "array_position" {
            test_db9_type(Db9TypeKind::Db9TypeKindInt32)
        } else {
            test_db9_array_type(Db9TypeKind::Db9TypeKindJson)
        };
        let request = test_db9_projection_request(test_db9_array_func_expr(
            function_name,
            return_type,
            vec![
                test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJson),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindJson),
            ],
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error for {function_name}: {}",
            err
        );
    }

    for function_name in ["array_position", "array_remove"] {
        let return_type = if function_name == "array_position" {
            test_db9_type(Db9TypeKind::Db9TypeKindInt32)
        } else {
            test_db9_array_type(Db9TypeKind::Db9TypeKindJsonb)
        };
        let request = test_db9_projection_request(test_db9_array_func_expr(
            function_name,
            return_type,
            vec![
                test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJsonb),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindJsonb),
            ],
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error for {function_name}: {}",
            err
        );
    }

    for (function_name, return_type, args) in [
        (
            "array_append",
            test_db9_array_type(Db9TypeKind::Db9TypeKindJson),
            vec![
                test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJson),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindJson),
            ],
        ),
        (
            "array_prepend",
            test_db9_array_type(Db9TypeKind::Db9TypeKindJson),
            vec![
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindJson),
                test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJson),
            ],
        ),
        (
            "array_cat",
            test_db9_array_type(Db9TypeKind::Db9TypeKindJson),
            vec![
                test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJson),
                test_db9_array_column_expr(1, Db9TypeKind::Db9TypeKindJson),
            ],
        ),
    ] {
        let request =
            test_db9_projection_request(test_db9_array_func_expr(function_name, return_type, args));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported")
                || err.to_string().contains("response output type"),
            "unexpected error for {function_name}: {}",
            err
        );
    }

    let array_to_string = test_db9_projection_request(test_db9_func_expr(
        "array_to_string",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindJson),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindText),
        ],
    ));
    let err = validate_db9_request_expression_surface(&array_to_string).unwrap_err();
    assert!(
        err.to_string()
            .contains("not supported by this codec surface"),
        "unexpected error: {}",
        err
    );

    for kind in [
        Db9TypeKind::Db9TypeKindNumeric,
        Db9TypeKind::Db9TypeKindDate,
        Db9TypeKind::Db9TypeKindTime,
        Db9TypeKind::Db9TypeKindUuid,
        Db9TypeKind::Db9TypeKindJsonb,
    ] {
        let equality = test_db9_projection_request(test_db9_binary_expr(
            Db9BinaryOp::Db9BinaryOpEq,
            Db9TypeKind::Db9TypeKindBoolean,
            test_db9_array_column_expr(0, kind),
            test_db9_array_column_expr(1, kind),
        ));
        let err = validate_db9_request_expression_surface(&equality).unwrap_err();
        assert!(
            err.to_string().contains("does not support"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }
}

#[test]
fn request_surface_keeps_array_cat_to_one_dimensional_contract() {
    let int_array = test_db9_array_type(Db9TypeKind::Db9TypeKindInt32);
    let mut int_matrix = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    int_matrix.set_elem_type(int_array.clone());

    let array_arg = test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindInt32);
    let mut matrix_arg = test_db9_column_expr(1, Db9TypeKind::Db9TypeKindArray);
    matrix_arg.set_return_type(int_matrix.clone());

    let one_dimensional = test_db9_projection_request(test_db9_array_func_expr(
        "array_cat",
        int_array.clone(),
        vec![array_arg.clone(), array_arg.clone()],
    ));
    validate_db9_request_expression_surface(&one_dimensional).unwrap();

    for args in [
        vec![array_arg.clone(), matrix_arg.clone()],
        vec![matrix_arg.clone(), array_arg.clone()],
        vec![matrix_arg.clone(), matrix_arg],
    ] {
        let request = test_db9_projection_request(test_db9_array_func_expr(
            "array_cat",
            int_matrix.clone(),
            args,
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error: {}",
            err
        );
    }
}

#[test]
fn request_surface_rejects_multidimensional_array_element_helpers() {
    let int_array = test_db9_array_type(Db9TypeKind::Db9TypeKindInt32);
    let mut int_matrix = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    int_matrix.set_elem_type(int_array.clone());

    let mut matrix_arg = test_db9_column_expr(0, Db9TypeKind::Db9TypeKindArray);
    matrix_arg.set_return_type(int_matrix.clone());
    let elem_arg = test_db9_array_column_expr(1, Db9TypeKind::Db9TypeKindInt32);

    for (function_name, return_type, args) in [
        (
            "array_position",
            test_db9_type(Db9TypeKind::Db9TypeKindInt32),
            vec![matrix_arg.clone(), elem_arg.clone()],
        ),
        (
            "array_remove",
            int_matrix.clone(),
            vec![matrix_arg.clone(), elem_arg.clone()],
        ),
        (
            "array_append",
            int_matrix.clone(),
            vec![matrix_arg.clone(), elem_arg.clone()],
        ),
        (
            "array_prepend",
            int_matrix.clone(),
            vec![elem_arg, matrix_arg],
        ),
    ] {
        let request =
            test_db9_projection_request(test_db9_array_func_expr(function_name, return_type, args));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error for {function_name}: {}",
            err
        );
    }
}

#[test]
fn request_surface_validates_sign_pg_return_contract() {
    let int_sign = test_db9_projection_request(test_db9_func_expr(
        "sign",
        Db9TypeKind::Db9TypeKindFloat64,
        vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32)],
    ));
    validate_db9_request_expression_surface(&int_sign).unwrap();

    let numeric_sign = test_db9_projection_request(test_db9_func_expr(
        "sign",
        Db9TypeKind::Db9TypeKindNumeric,
        vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindNumeric)],
    ));
    validate_db9_request_expression_surface(&numeric_sign).unwrap();
}

#[test]
fn request_surface_rejects_two_arg_round_trunc_float8_precision() {
    for function_name in ["round", "trunc"] {
        let request = test_db9_projection_request(test_db9_func_expr(
            function_name,
            Db9TypeKind::Db9TypeKindFloat64,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindFloat64),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt64),
            ],
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error for {function_name}: {}",
            err
        );
    }
}

#[test]
fn request_surface_rejects_cast_expressions() {
    let request = test_db9_projection_request(test_db9_cast_expr(
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt64),
        Db9TypeKind::Db9TypeKindText,
    ));
    let err = validate_db9_request_expression_surface(&request).unwrap_err();
    assert!(
        err.to_string().contains("CAST expressions"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_rejects_aggregate_requests() {
    let mut request = Db9DagRequest::default();
    request.set_aggregate(kvproto::db9_coprocessor::Db9AggregatePlan::default());
    let err = validate_db9_request_expression_surface(&request).unwrap_err();
    assert!(
        err.to_string()
            .contains("aggregate requests are not supported"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_validates_binary_expression_signatures() {
    let valid_and = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpAnd,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindBoolean),
        test_db9_bool_const_expr(true),
    ));
    validate_db9_request_expression_surface(&valid_and).unwrap();

    let valid_text_equality = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpEq,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
        test_db9_text_const_expr("alpha"),
    ));
    validate_db9_request_expression_surface(&valid_text_equality).unwrap();

    let valid_numeric_ordering = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpLt,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
        test_db9_float64_const_expr(1.0),
    ));
    validate_db9_request_expression_surface(&valid_numeric_ordering).unwrap();

    for kind in [
        Db9TypeKind::Db9TypeKindDate,
        Db9TypeKind::Db9TypeKindTime,
        Db9TypeKind::Db9TypeKindUuid,
    ] {
        let request = test_db9_projection_request(test_db9_binary_expr(
            Db9BinaryOp::Db9BinaryOpEq,
            Db9TypeKind::Db9TypeKindBoolean,
            test_db9_column_expr(0, kind),
            test_db9_column_expr(1, kind),
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("does not support"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }

    for kind in [Db9TypeKind::Db9TypeKindDate, Db9TypeKind::Db9TypeKindTime] {
        let request = test_db9_projection_request(test_db9_binary_expr(
            Db9BinaryOp::Db9BinaryOpLt,
            Db9TypeKind::Db9TypeKindBoolean,
            test_db9_column_expr(0, kind),
            test_db9_column_expr(1, kind),
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("does not support"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }

    let int_and = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpAnd,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
        test_db9_bool_const_expr(true),
    ));
    let err = validate_db9_request_expression_surface(&int_and).unwrap_err();
    assert!(
        err.to_string().contains("requires BOOLEAN operands"),
        "unexpected error: {}",
        err
    );

    let wrong_return = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpEq,
        Db9TypeKind::Db9TypeKindInt32,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
        test_db9_int32_const_expr(1),
    ));
    let err = validate_db9_request_expression_surface(&wrong_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    for kind in [Db9TypeKind::Db9TypeKindJson, Db9TypeKind::Db9TypeKindJsonb] {
        let request = test_db9_projection_request(test_db9_binary_expr(
            Db9BinaryOp::Db9BinaryOpEq,
            Db9TypeKind::Db9TypeKindBoolean,
            test_db9_column_expr(0, kind),
            test_db9_column_expr(1, kind),
        ));
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("does not support"),
            "unexpected error for {kind:?}: {}",
            err
        );
    }

    let json_array_type = test_db9_array_type(Db9TypeKind::Db9TypeKindJson);
    let mut left_json_array = test_db9_column_expr(0, Db9TypeKind::Db9TypeKindArray);
    left_json_array.set_return_type(json_array_type.clone());
    let mut right_json_array = test_db9_column_expr(1, Db9TypeKind::Db9TypeKindArray);
    right_json_array.set_return_type(json_array_type);
    let json_array_equality = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpEq,
        Db9TypeKind::Db9TypeKindBoolean,
        left_json_array,
        right_json_array,
    ));
    let err = validate_db9_request_expression_surface(&json_array_equality).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let bool_ordering = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpLt,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindBoolean),
        test_db9_bool_const_expr(true),
    ));
    let err = validate_db9_request_expression_surface(&bool_ordering).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let array_ordering = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpLt,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_array_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
        test_db9_array_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
    ));
    let err = validate_db9_request_expression_surface(&array_ordering).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let missing_operator = test_db9_projection_request(test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpUnspecified,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindBoolean),
        test_db9_bool_const_expr(true),
    ));
    let err = validate_db9_request_expression_surface(&missing_operator).unwrap_err();
    assert!(
        err.to_string().contains("missing operator"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_validates_unary_operand_types() {
    let valid = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNot,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_bool_const_expr(true),
    ));
    validate_db9_request_expression_surface(&valid).unwrap();

    let valid_neg = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNeg,
        Db9TypeKind::Db9TypeKindInt32,
        test_db9_int32_const_expr(1),
    ));
    validate_db9_request_expression_surface(&valid_neg).unwrap();

    let valid_pos = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpPos,
        Db9TypeKind::Db9TypeKindFloat64,
        test_db9_float64_const_expr(1.0),
    ));
    validate_db9_request_expression_surface(&valid_pos).unwrap();

    let invalid_not = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNot,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_int32_const_expr(1),
    ));
    let err = validate_db9_request_expression_surface(&invalid_not).unwrap_err();
    assert!(
        err.to_string().contains("requires BOOLEAN operand"),
        "unexpected error: {}",
        err
    );

    let invalid_not_return = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNot,
        Db9TypeKind::Db9TypeKindInt32,
        test_db9_bool_const_expr(true),
    ));
    let err = validate_db9_request_expression_surface(&invalid_not_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    let invalid_neg = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNeg,
        Db9TypeKind::Db9TypeKindInt32,
        test_db9_text_const_expr("x"),
    ));
    let err = validate_db9_request_expression_surface(&invalid_neg).unwrap_err();
    assert!(
        err.to_string().contains("primitive numeric operand"),
        "unexpected error: {}",
        err
    );

    let invalid_neg_return = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpNeg,
        Db9TypeKind::Db9TypeKindFloat64,
        test_db9_int32_const_expr(1),
    ));
    let err = validate_db9_request_expression_surface(&invalid_neg_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    let missing_operator = test_db9_projection_request(test_db9_unary_expr(
        Db9UnaryOp::Db9UnaryOpUnspecified,
        Db9TypeKind::Db9TypeKindInt32,
        test_db9_int32_const_expr(1),
    ));
    let err = validate_db9_request_expression_surface(&missing_operator).unwrap_err();
    assert!(
        err.to_string().contains("missing operator"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_validates_temporal_pushdown_contract() {
    for request in [
        test_db9_projection_request(test_db9_func_expr(
            "extract",
            Db9TypeKind::Db9TypeKindNumeric,
            vec![
                test_db9_text_const_expr("year"),
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "date_part",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![
                test_db9_text_const_expr("dow"),
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "date_trunc",
            Db9TypeKind::Db9TypeKindTimestamp,
            vec![
                test_db9_text_const_expr(" hour "),
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "to_char",
            Db9TypeKind::Db9TypeKindText,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
                test_db9_text_const_expr("YYYY-MM-DD HH24:MI:SS"),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "to_char",
            Db9TypeKind::Db9TypeKindText,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
                test_db9_text_const_expr(r#"YYYY-MM-DD"T"HH24:MI:SS"Z""#),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "age",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
                test_db9_timestamp_const_expr(1_704_067_200_000),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "date",
            Db9TypeKind::Db9TypeKindDate,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp)],
        )),
    ] {
        validate_db9_request_expression_surface(&request).unwrap();
    }

    let bad_return = test_db9_projection_request(test_db9_func_expr(
        "date_part",
        Db9TypeKind::Db9TypeKindInt64,
        vec![
            test_db9_text_const_expr("epoch"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInterval),
        ],
    ));
    let err = validate_db9_request_expression_surface(&bad_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    let interval_date_part = test_db9_projection_request(test_db9_func_expr(
        "date_part",
        Db9TypeKind::Db9TypeKindFloat64,
        vec![
            test_db9_text_const_expr("epoch"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInterval),
        ],
    ));
    let err = validate_db9_request_expression_surface(&interval_date_part).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let timestamptz_date_part = test_db9_projection_request(test_db9_func_expr(
        "date_part",
        Db9TypeKind::Db9TypeKindFloat64,
        vec![
            test_db9_text_const_expr("epoch"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamptz),
        ],
    ));
    let err = validate_db9_request_expression_surface(&timestamptz_date_part).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let timestamptz_date = test_db9_projection_request(test_db9_func_expr(
        "date",
        Db9TypeKind::Db9TypeKindDate,
        vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamptz)],
    ));
    let err = validate_db9_request_expression_surface(&timestamptz_date).unwrap_err();
    assert!(
        err.to_string().contains("signature is not supported"),
        "unexpected error: {}",
        err
    );

    let text_date = test_db9_projection_request(test_db9_func_expr(
        "date",
        Db9TypeKind::Db9TypeKindDate,
        vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText)],
    ));
    let err = validate_db9_request_expression_surface(&text_date).unwrap_err();
    assert!(
        err.to_string().contains("signature is not supported"),
        "unexpected error: {}",
        err
    );

    for request in [
        test_db9_projection_request(test_db9_func_expr(
            "age",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamptz),
                test_db9_timestamp_const_expr(1_704_067_200_000),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "age",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindText),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "age",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindTimestamptz),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "age",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamptz),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindTimestamp),
            ],
        )),
    ] {
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string().contains("signature is not supported"),
            "unexpected error: {}",
            err
        );
    }

    let make_time_numeric = test_db9_projection_request(test_db9_func_expr(
        "make_time",
        Db9TypeKind::Db9TypeKindTime,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
            test_db9_column_expr(2, Db9TypeKind::Db9TypeKindNumeric),
        ],
    ));
    let err = validate_db9_request_expression_surface(&make_time_numeric).unwrap_err();
    assert!(
        err.to_string().contains("signature is not supported"),
        "unexpected error: {}",
        err
    );

    for request in [
        test_db9_projection_request(test_db9_func_expr(
            "make_timestamp",
            Db9TypeKind::Db9TypeKindTimestamp,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(2, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(3, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(4, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(5, Db9TypeKind::Db9TypeKindNumeric),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "make_interval",
            Db9TypeKind::Db9TypeKindInterval,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(2, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(3, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(4, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(5, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(6, Db9TypeKind::Db9TypeKindNumeric),
            ],
        )),
        test_db9_projection_request(test_db9_func_expr(
            "to_timestamp",
            Db9TypeKind::Db9TypeKindTimestamptz,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindNumeric)],
        )),
    ] {
        let err = validate_db9_request_expression_surface(&request).unwrap_err();
        assert!(
            err.to_string()
                .contains("is not supported by this codec surface"),
            "unexpected error: {}",
            err
        );
    }

    let unsupported_extract_interval = test_db9_projection_request(test_db9_func_expr(
        "extract",
        Db9TypeKind::Db9TypeKindNumeric,
        vec![
            test_db9_text_const_expr("epoch"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInterval),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_extract_interval).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let unsupported_date_part_field = test_db9_projection_request(test_db9_func_expr(
        "date_part",
        Db9TypeKind::Db9TypeKindFloat64,
        vec![
            test_db9_text_const_expr("timezone"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_date_part_field).unwrap_err();
    assert!(
        err.to_string().contains("field 'timezone'"),
        "unexpected error: {}",
        err
    );

    let unsupported_date_trunc_field = test_db9_projection_request(test_db9_func_expr(
        "date_trunc",
        Db9TypeKind::Db9TypeKindTimestamp,
        vec![
            test_db9_text_const_expr("week"),
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_date_trunc_field).unwrap_err();
    assert!(
        err.to_string().contains("field 'week'"),
        "unexpected error: {}",
        err
    );

    let unsupported_to_char_pattern = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_text_const_expr("YYYY Mon DD"),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_to_char_pattern).unwrap_err();
    assert!(
        err.to_string().contains("pattern"),
        "unexpected error: {}",
        err
    );

    let lowercase_to_char_pattern = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_text_const_expr("yyyy-mm-dd hh24:mi:ss"),
        ],
    ));
    validate_db9_request_expression_surface(&lowercase_to_char_pattern)
        .expect("lowercase to_char pattern should be accepted");

    let unterminated_to_char_literal = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_text_const_expr("YYYY\"x"),
        ],
    ));
    validate_db9_request_expression_surface(&unterminated_to_char_literal)
        .expect("unterminated to_char quoted literal should be accepted");

    let unsupported_to_char_bare_t = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_text_const_expr("YYYY-MM-DDTHH24:MI:SS"),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_to_char_bare_t).unwrap_err();
    assert!(
        err.to_string().contains("pattern"),
        "unexpected error: {}",
        err
    );

    let unsupported_to_char_source = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindDate),
            test_db9_text_const_expr("YYYY-MM-DD"),
        ],
    ));
    let err = validate_db9_request_expression_surface(&unsupported_to_char_source).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let oversized_to_char_pattern = test_db9_projection_request(test_db9_func_expr(
        "to_char",
        Db9TypeKind::Db9TypeKindText,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_text_const_expr(&"YYYY".repeat(MAX_TO_CHAR_PATTERN_BYTES / 4 + 1)),
        ],
    ));
    let err = validate_db9_request_expression_surface(&oversized_to_char_pattern).unwrap_err();
    assert!(
        err.to_string().contains("pattern length"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn request_surface_validates_mod_signatures() {
    let mixed_width = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindInt64,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt64),
        ],
    ));
    validate_db9_request_expression_surface(&mixed_width).unwrap();

    let mixed_width_wrong_return = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindInt32,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt64),
        ],
    ));
    let err = validate_db9_request_expression_surface(&mixed_width_wrong_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    let numeric = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindNumeric,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindNumeric),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindNumeric),
        ],
    ));
    let err = validate_db9_request_expression_surface(&numeric).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let float64 = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindFloat64,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindFloat64),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindFloat64),
        ],
    ));
    let err = validate_db9_request_expression_surface(&float64).unwrap_err();
    assert!(
        err.to_string().contains("does not support"),
        "unexpected error: {}",
        err
    );

    let wrong_return = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindInt64,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
        ],
    ));
    let err = validate_db9_request_expression_surface(&wrong_return).unwrap_err();
    assert!(
        err.to_string().contains("return type"),
        "unexpected error: {}",
        err
    );

    let supported_int64 = test_db9_projection_request(test_db9_func_expr(
        "mod",
        Db9TypeKind::Db9TypeKindInt64,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt64),
            test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt64),
        ],
    ));
    validate_db9_request_expression_surface(&supported_int64).unwrap();
}

#[test]
fn request_surface_validates_shift_signatures() {
    for function_name in ["__db9_shl", "__db9_shr"] {
        let supported_int32 = test_db9_projection_request(test_db9_func_expr(
            function_name,
            Db9TypeKind::Db9TypeKindInt32,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
            ],
        ));
        validate_db9_request_expression_surface(&supported_int32).unwrap();

        let supported_int64 = test_db9_projection_request(test_db9_func_expr(
            function_name,
            Db9TypeKind::Db9TypeKindInt64,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt64),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
            ],
        ));
        validate_db9_request_expression_surface(&supported_int64).unwrap();

        for (return_kind, left_kind, right_kind) in [
            (
                Db9TypeKind::Db9TypeKindInt32,
                Db9TypeKind::Db9TypeKindInt32,
                Db9TypeKind::Db9TypeKindInt64,
            ),
            (
                Db9TypeKind::Db9TypeKindInt64,
                Db9TypeKind::Db9TypeKindInt64,
                Db9TypeKind::Db9TypeKindInt64,
            ),
        ] {
            let unsupported_count = test_db9_projection_request(test_db9_func_expr(
                function_name,
                return_kind,
                vec![
                    test_db9_column_expr(0, left_kind),
                    test_db9_column_expr(1, right_kind),
                ],
            ));
            let err = validate_db9_request_expression_surface(&unsupported_count).unwrap_err();
            assert!(
                err.to_string().contains("signature"),
                "unexpected error: {}",
                err
            );
        }

        let wrong_return = test_db9_projection_request(test_db9_func_expr(
            function_name,
            Db9TypeKind::Db9TypeKindInt32,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt64),
                test_db9_column_expr(1, Db9TypeKind::Db9TypeKindInt32),
            ],
        ));
        let err = validate_db9_request_expression_surface(&wrong_return).unwrap_err();
        assert!(
            err.to_string().contains("signature"),
            "unexpected error: {}",
            err
        );
    }
}

fn test_db9_column_expr(index: u32, kind: Db9TypeKind) -> Db9Expr {
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(kind));
    let mut column = kvproto::db9_coprocessor::Db9ColumnRefExpr::default();
    column.set_column_index(index);
    column.set_column_name(format!("c{index}"));
    expr.set_column_ref(column);
    expr
}

fn test_db9_cast_expr(expr: Db9Expr, target_kind: Db9TypeKind) -> Db9Expr {
    let mut out = Db9Expr::default();
    out.set_return_type(test_db9_type(target_kind));
    let mut cast = kvproto::db9_coprocessor::Db9CastExpr::default();
    cast.set_expr(expr);
    cast.set_target_type(test_db9_type(target_kind));
    out.set_cast(cast);
    out
}

fn test_db9_unary_expr(op: Db9UnaryOp, return_kind: Db9TypeKind, operand: Db9Expr) -> Db9Expr {
    let mut out = Db9Expr::default();
    out.set_return_type(test_db9_type(return_kind));
    let mut unary = Db9UnaryExpr::default();
    unary.set_op(op);
    unary.set_operand(operand);
    out.set_unary(unary);
    out
}

fn test_db9_table(columns: &[Db9TypeKind]) -> Db9TableInfo {
    let mut table = Db9TableInfo::default();
    for (idx, kind) in columns.iter().enumerate() {
        let mut column = kvproto::db9_coprocessor::Db9ColumnInfo::default();
        column.set_name(format!("c{idx}"));
        column.set_data_type(test_db9_type(*kind));
        table.mut_columns().push(column);
    }
    table
}

fn test_db9_named_table(columns: &[(&str, Db9TypeKind)]) -> Db9TableInfo {
    let mut table = Db9TableInfo::default();
    for (name, kind) in columns.iter() {
        let mut column = kvproto::db9_coprocessor::Db9ColumnInfo::default();
        column.set_name((*name).to_owned());
        column.set_data_type(test_db9_type(*kind));
        table.mut_columns().push(column);
    }
    table
}

#[test]
fn encode_response_row_allows_short_physical_rows() {
    let row = StoredRow {
        values: vec![StoredValue::Int64(7)],
    };
    let encoded = encode_response_row(
        &test_db9_table(&[Db9TypeKind::Db9TypeKindInt64, Db9TypeKind::Db9TypeKindText]),
        row,
    )
    .unwrap();

    assert_eq!(encoded.get_values().len(), 2);
    assert_eq!(encoded.get_values()[0].get_int64_value(), 7);
    assert!(encoded.get_values()[1].has_null_value());
}

#[test]
fn encode_response_row_ignores_stale_trailing_values_from_old_wide_rows() {
    let row = StoredRow {
        values: vec![
            StoredValue::Int64(7),
            StoredValue::Text("alpha".to_owned()),
            StoredValue::Boolean(true),
        ],
    };
    let encoded = encode_response_row(
        &test_db9_table(&[Db9TypeKind::Db9TypeKindInt64, Db9TypeKind::Db9TypeKindText]),
        row,
    )
    .unwrap();

    assert_eq!(encoded.get_values().len(), 2);
    assert_eq!(encoded.get_values()[0].get_int64_value(), 7);
    assert_eq!(encoded.get_values()[1].get_text_value(), "alpha");
}

#[test]
fn encode_response_value_supports_timestamp_and_timestamptz() {
    let timestamp = encode_response_value(
        StoredValue::Timestamp(1_700_000_000_000),
        &test_db9_type(Db9TypeKind::Db9TypeKindTimestamp),
    )
    .unwrap();
    assert!(timestamp.has_timestamp_value());
    assert_eq!(timestamp.get_timestamp_value(), 1_700_000_000_000);

    let timestamptz = encode_response_value(
        StoredValue::Timestamp(1_700_000_000_001),
        &test_db9_type(Db9TypeKind::Db9TypeKindTimestamptz),
    )
    .unwrap();
    assert!(timestamptz.has_timestamptz_value());
    assert_eq!(timestamptz.get_timestamptz_value(), 1_700_000_000_001);

    let time = encode_response_value(
        StoredValue::Time(3_723_004_005),
        &test_db9_type(Db9TypeKind::Db9TypeKindTime),
    )
    .unwrap();
    assert!(time.has_int64_value());
    assert_eq!(time.get_int64_value(), 3_723_004_005);

    let date = encode_response_value(
        StoredValue::Date(19_724),
        &test_db9_type(Db9TypeKind::Db9TypeKindDate),
    )
    .unwrap();
    assert!(date.has_int32_value());
    assert_eq!(date.get_int32_value(), 19_724);

    let interval = encode_response_value(
        StoredValue::Interval(StoredIntervalValue {
            months: 2,
            millis: 90_061_000,
        }),
        &test_db9_type(Db9TypeKind::Db9TypeKindInterval),
    )
    .unwrap();
    assert!(interval.has_interval_value());
    assert_eq!(interval.get_interval_value().get_months(), 2);
    assert_eq!(interval.get_interval_value().get_millis(), 90_061_000);
}

#[test]
fn age_projection_encodes_interval_value() {
    let request = test_db9_projection_request(test_db9_func_expr(
        "age",
        Db9TypeKind::Db9TypeKindInterval,
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindTimestamp),
            test_db9_timestamp_const_expr(1_704_067_200_000),
        ],
    ));
    validate_db9_request_expression_surface(&request).unwrap();

    let encoded = project_row(
        &request,
        &StoredRow {
            values: vec![StoredValue::Timestamp(1_704_153_600_000)],
        },
    )
    .unwrap();

    assert_eq!(encoded.get_values().len(), 1);
    assert!(encoded.get_values()[0].has_interval_value());
    assert_eq!(encoded.get_values()[0].get_interval_value().get_months(), 0);
    assert_eq!(
        encoded.get_values()[0].get_interval_value().get_millis(),
        86_400_000
    );
}

#[test]
fn encode_response_value_supports_user_defined_text_values() {
    let encoded = encode_response_value(
        StoredValue::Text("udt-payload".to_owned()),
        &test_db9_type(Db9TypeKind::Db9TypeKindUserDefined),
    )
    .unwrap();
    assert!(encoded.has_text_value());
    assert_eq!(encoded.get_text_value(), "udt-payload");
}

#[test]
fn encode_response_value_rejects_declared_type_mismatches() {
    let int_as_text = encode_response_value(
        StoredValue::Int64(7),
        &test_db9_type(Db9TypeKind::Db9TypeKindText),
    )
    .unwrap_err();
    assert!(
        int_as_text.to_string().contains("cannot be encoded as"),
        "unexpected error: {}",
        int_as_text
    );

    let bytes_as_text = encode_response_value(
        StoredValue::Bytes(b"abc".to_vec()),
        &test_db9_type(Db9TypeKind::Db9TypeKindText),
    )
    .unwrap_err();
    assert!(
        bytes_as_text.to_string().contains("cannot be encoded as"),
        "unexpected error: {}",
        bytes_as_text
    );

    let mut int_array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    int_array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindInt64));
    let text_as_int_array = encode_response_value(
        StoredValue::Array(vec![StoredValue::Text("x".to_owned())]),
        &int_array_type,
    )
    .unwrap_err();
    assert!(
        text_as_int_array
            .to_string()
            .contains("cannot be encoded as"),
        "unexpected error: {}",
        text_as_int_array
    );

    let nested_text_as_int_array = encode_response_value(
        StoredValue::Array(vec![StoredValue::Array(vec![StoredValue::Text(
            "x".to_owned(),
        )])]),
        &int_array_type,
    )
    .unwrap_err();
    assert!(
        nested_text_as_int_array
            .to_string()
            .contains("cannot be encoded as"),
        "unexpected error: {}",
        nested_text_as_int_array
    );
}

#[test]
fn encode_response_value_preserves_multidimensional_array_shape_for_flat_declared_type() {
    let mut array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindInt32));

    let encoded = encode_response_value(
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
        ]),
        &array_type,
    )
    .unwrap();

    assert!(encoded.has_text_value());
    assert_eq!(encoded.get_text_value(), "{{1,2},{3,4}}");
}

#[test]
fn decode_stored_row_enforces_byte_limit_before_allocating() {
    let row = StoredRow {
        values: vec![StoredValue::Text("bounded".to_owned())],
    };
    let encoded = bincode::serialize(&row).unwrap();
    let err = decode_stored_row(&encoded, 1).unwrap_err();

    assert!(
        err.to_string().contains("stored row payload size")
            && err.to_string().contains("exceeds decode limit"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn decode_stored_row_accepts_trailing_bytes_compat_fixture() {
    #[rustfmt::skip]
    let fixture = [
        // StoredRow { values: vec![Int32(7), Text("old")] }
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
        0x02, 0x00, 0x00, 0x00,
        0x07, 0x00, 0x00, 0x00,
        0x05, 0x00, 0x00, 0x00,
        0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
        b'o', b'l', b'd',
        // Future row-format fields may be appended after StoredRow::values.
        0xdb, 0x09,
    ];

    let row = decode_stored_row(&fixture, fixture.len() as u64).unwrap();

    assert_eq!(
        row.values,
        vec![StoredValue::Int32(7), StoredValue::Text("old".to_owned())]
    );
}

#[test]
fn stored_row_decode_limit_honors_response_budget() {
    assert_eq!(
        max_stored_row_decode_bytes_for_response(1024),
        DEFAULT_MIN_STORED_ROW_DECODE_BYTES
    );
    assert_eq!(
        max_stored_row_decode_bytes_for_response(MAX_STORED_ROW_DECODE_BYTES_CAP + 1),
        MAX_STORED_ROW_DECODE_BYTES_CAP
    );
}

#[test]
fn decode_stored_row_limits_declared_nested_lengths() {
    let mut malicious = Vec::new();
    malicious.extend_from_slice(&1_u64.to_le_bytes());
    malicious.extend_from_slice(&5_u32.to_le_bytes());
    malicious.extend_from_slice(&u64::MAX.to_le_bytes());

    let err = decode_stored_row(&malicious, 1024).unwrap_err();

    assert!(
        err.to_string().contains("stored row payload")
            && err.to_string().contains("exceeds decode limit"),
        "unexpected error: {}",
        err
    );
}

#[test]
fn encode_response_value_preserves_multidimensional_array_shape_for_nested_declared_type() {
    let mut inner_array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    inner_array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindInt32));

    let mut outer_array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    outer_array_type.set_elem_type(inner_array_type);

    let encoded = encode_response_value(
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
        ]),
        &outer_array_type,
    )
    .unwrap();

    assert!(encoded.has_text_value());
    assert_eq!(encoded.get_text_value(), "{{1,2},{3,4}}");
}

#[test]
fn decode_wire_value_supports_temporal_types() {
    let timestamp_type = test_db9_type(Db9TypeKind::Db9TypeKindTimestamp);
    let mut timestamp = Db9Value::default();
    timestamp.set_timestamp_value(1_700_000_000_000);
    assert_eq!(
        decode_wire_value(&timestamp, &timestamp_type).unwrap(),
        StoredValue::Timestamp(1_700_000_000_000)
    );

    let timestamptz_type = test_db9_type(Db9TypeKind::Db9TypeKindTimestamptz);
    let mut timestamptz = Db9Value::default();
    timestamptz.set_timestamptz_value(1_700_000_000_001);
    assert_eq!(
        decode_wire_value(&timestamptz, &timestamptz_type).unwrap(),
        StoredValue::Timestamp(1_700_000_000_001)
    );

    let time_type = test_db9_type(Db9TypeKind::Db9TypeKindTime);
    let mut time = Db9Value::default();
    time.set_int64_value(3_723_004_005);
    assert_eq!(
        decode_wire_value(&time, &time_type).unwrap(),
        StoredValue::Time(3_723_004_005)
    );

    let date_type = test_db9_type(Db9TypeKind::Db9TypeKindDate);
    let mut date = Db9Value::default();
    date.set_int32_value(19_724);
    assert_eq!(
        decode_wire_value(&date, &date_type).unwrap(),
        StoredValue::Date(19_724)
    );

    let interval_type = test_db9_type(Db9TypeKind::Db9TypeKindInterval);
    let mut interval = Db9Value::default();
    let mut interval_value = kvproto::db9_coprocessor::Db9IntervalValue::default();
    interval_value.set_months(2);
    interval_value.set_millis(90_061_000);
    interval.set_interval_value(interval_value);
    assert_eq!(
        decode_wire_value(&interval, &interval_type).unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: 2,
            millis: 90_061_000,
        })
    );
}

#[test]
fn decode_wire_value_supports_numeric_text_constants() {
    let numeric_type = test_db9_type(Db9TypeKind::Db9TypeKindNumeric);
    let mut numeric = Db9Value::default();
    numeric.set_text_value("5.5".to_owned());
    assert_eq!(
        decode_wire_value(&numeric, &numeric_type).unwrap(),
        StoredValue::Numeric(StoredDecimal {
            lo: 55,
            mid: 0,
            hi: 0,
            negative: false,
            scale: 1,
        })
    );
}

#[test]
fn decode_wire_value_supports_numeric_infinity_text_constants() {
    let numeric_type = test_db9_type(Db9TypeKind::Db9TypeKindNumeric);

    let mut positive = Db9Value::default();
    positive.set_text_value("Infinity".to_owned());
    assert_eq!(
        decode_wire_value(&positive, &numeric_type).unwrap(),
        StoredValue::Float64(f64::INFINITY)
    );

    let mut negative = Db9Value::default();
    negative.set_text_value("-Infinity".to_owned());
    assert_eq!(
        decode_wire_value(&negative, &numeric_type).unwrap(),
        StoredValue::Float64(f64::NEG_INFINITY)
    );
}

#[test]
fn parse_db9_array_text_items_preserves_quoted_empty_final_element() {
    assert_eq!(
        parse_db9_array_text_items("{\"\"}").unwrap(),
        vec![(String::new(), true)]
    );
}

#[test]
fn parse_db9_array_text_items_preserves_nested_array_shape() {
    assert_eq!(
        parse_db9_array_text_items("{{1,2},{3,4}}").unwrap(),
        vec![("{1,2}".to_owned(), false), ("{3,4}".to_owned(), false),]
    );
}

#[test]
fn decode_wire_value_preserves_quoted_empty_array_text_elements() {
    let mut array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindText));

    let mut array_wire = Db9Value::default();
    array_wire.set_text_value("{\"\"}".to_owned());
    assert_eq!(
        decode_wire_value(&array_wire, &array_type).unwrap(),
        StoredValue::Array(vec![StoredValue::Text(String::new())])
    );
}

#[test]
fn decode_wire_value_preserves_nested_array_text_elements() {
    let mut inner_array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    inner_array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindInt32));
    let mut outer_array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    outer_array_type.set_elem_type(inner_array_type);

    let mut array_wire = Db9Value::default();
    array_wire.set_text_value("{{1,2},{3,4}}".to_owned());
    assert_eq!(
        decode_wire_value(&array_wire, &outer_array_type).unwrap(),
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
        ])
    );
}

#[test]
fn decode_wire_value_rejects_excessively_nested_array_text_elements() {
    let mut nested_type = test_db9_type(Db9TypeKind::Db9TypeKindInt32);
    for _ in 0..=MAX_ARRAY_RECURSION_DEPTH {
        let mut outer = test_db9_type(Db9TypeKind::Db9TypeKindArray);
        outer.set_elem_type(nested_type);
        nested_type = outer;
    }

    let mut literal = "1".to_owned();
    for _ in 0..=MAX_ARRAY_RECURSION_DEPTH {
        literal = format!("{{{literal}}}");
    }

    let mut array_wire = Db9Value::default();
    array_wire.set_text_value(literal);
    let err = decode_wire_value(&array_wire, &nested_type).unwrap_err();
    assert_eq!(err.to_string(), "array value is too deep");
}

#[test]
fn decode_wire_value_preserves_quoted_whitespace_array_text_elements() {
    let mut array_type = test_db9_type(Db9TypeKind::Db9TypeKindArray);
    array_type.set_elem_type(test_db9_type(Db9TypeKind::Db9TypeKindText));

    let mut array_wire = Db9Value::default();
    array_wire.set_text_value("{\"  hi  \"}".to_owned());
    assert_eq!(
        decode_wire_value(&array_wire, &array_type).unwrap(),
        StoredValue::Array(vec![StoredValue::Text("  hi  ".to_owned())])
    );
}

#[test]
fn uuid_and_jsonb_wire_values_use_bytes_payloads() {
    let uuid = encode_response_value(
        StoredValue::Uuid([0x33; 16]),
        &test_db9_type(Db9TypeKind::Db9TypeKindUuid),
    )
    .unwrap();
    assert!(uuid.has_bytes_value());
    assert_eq!(uuid.get_bytes_value(), [0x33; 16]);

    let jsonb = encode_response_value(
        StoredValue::Jsonb("{\"a\":1}".to_owned()),
        &test_db9_type(Db9TypeKind::Db9TypeKindJsonb),
    )
    .unwrap();
    assert!(jsonb.has_bytes_value());
    assert!(jsonb.get_bytes_value().starts_with(DB9_JSONB_TEXT_MAGIC));
    assert_eq!(
        decode_jsonb_wire_bytes(jsonb.get_bytes_value()).unwrap(),
        "{\"a\":1}"
    );

    let jsonb_numeric = encode_response_value(
        StoredValue::Jsonb(r#"{"n":9007199254740993.123456789}"#.to_owned()),
        &test_db9_type(Db9TypeKind::Db9TypeKindJsonb),
    )
    .unwrap();
    assert!(
        jsonb_numeric
            .get_bytes_value()
            .starts_with(DB9_JSONB_TEXT_MAGIC)
    );
    assert_eq!(
        decode_jsonb_wire_bytes(jsonb_numeric.get_bytes_value()).unwrap(),
        r#"{"n":9007199254740993.123456789}"#
    );

    let mut uuid_wire = Db9Value::default();
    uuid_wire.set_bytes_value([0x44; 16].to_vec());
    assert_eq!(
        decode_wire_value(&uuid_wire, &test_db9_type(Db9TypeKind::Db9TypeKindUuid)).unwrap(),
        StoredValue::Uuid([0x44; 16])
    );

    let mut jsonb_wire = Db9Value::default();
    jsonb_wire.set_bytes_value(encode_jsonb_wire_bytes("{\"b\":2,\"a\":1}").unwrap());
    assert_eq!(
        decode_wire_value(&jsonb_wire, &test_db9_type(Db9TypeKind::Db9TypeKindJsonb)).unwrap(),
        StoredValue::Jsonb("{\"a\":1,\"b\":2}".to_owned())
    );
}

#[test]
fn decode_wire_value_accepts_legacy_utf8_jsonb_bytes() {
    let mut jsonb_wire = Db9Value::default();
    jsonb_wire.set_bytes_value(b"{\"b\":2,\"a\":1}".to_vec());
    assert_eq!(
        decode_wire_value(&jsonb_wire, &test_db9_type(Db9TypeKind::Db9TypeKindJsonb)).unwrap(),
        StoredValue::Jsonb("{\"a\":1,\"b\":2}".to_owned())
    );
}

#[test]
fn jsonb_wire_values_use_stable_canonical_key_order() {
    let mut jsonb_wire = Db9Value::default();
    jsonb_wire.set_text_value(r#"{"aa":{"bb":2,"a":1},"b":3,"a":4}"#.to_owned());

    assert_eq!(
        decode_wire_value(&jsonb_wire, &test_db9_type(Db9TypeKind::Db9TypeKindJsonb)).unwrap(),
        StoredValue::Jsonb(r#"{"a":4,"b":3,"aa":{"a":1,"bb":2}}"#.to_owned())
    );

    assert_eq!(
        decode_jsonb_wire_bytes(
            &encode_jsonb_wire_bytes(r#"{"aa":{"bb":2,"a":1},"b":3,"a":4}"#).unwrap()
        )
        .unwrap(),
        r#"{"a":4,"b":3,"aa":{"a":1,"bb":2}}"#
    );
}

#[test]
fn encoding_and_hashing_functions_match_expected_outputs() {
    assert_eq!(
        eval_md5_function("md5", vec![StoredValue::Text("hello".to_owned())]).unwrap(),
        StoredValue::Text("5d41402abc4b2a76b9719d911017c592".to_owned())
    );
    assert_eq!(
        eval_md5_function("md5", vec![StoredValue::Null]).unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_sha256_function("sha256", vec![StoredValue::Bytes(b"hello".to_vec())]).unwrap(),
        StoredValue::Bytes(
            hex::decode("2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824")
                .unwrap()
        )
    );
    assert_eq!(
        eval_sha256_function("sha256", vec![StoredValue::Null]).unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_digest_function(
            "digest",
            vec![
                StoredValue::Text("hello".to_owned()),
                StoredValue::Text("sha256".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(
            hex::decode("2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824")
                .unwrap()
        )
    );
}

#[test]
fn decode_function_supports_base64_hex_and_escape() {
    assert_eq!(
        eval_decode_function(
            "decode",
            vec![
                StoredValue::Text("aGVsbG8=".to_owned()),
                StoredValue::Text("base64".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(b"hello".to_vec())
    );
    assert_eq!(
        eval_decode_function(
            "decode",
            vec![
                StoredValue::Text("deadbeef".to_owned()),
                StoredValue::Text("hex".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(vec![0xde, 0xad, 0xbe, 0xef])
    );
    assert_eq!(
        eval_decode_function(
            "decode",
            vec![
                StoredValue::Text("AB CD".to_owned()),
                StoredValue::Text("hex".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(vec![0xab, 0xcd])
    );
    assert_eq!(
        eval_decode_function(
            "decode",
            vec![
                StoredValue::Text(" 61\n62\t63\r".to_owned()),
                StoredValue::Text("hex".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(b"abc".to_vec())
    );
    assert_eq!(
        eval_decode_function(
            "decode",
            vec![
                StoredValue::Text("a\\\\\\377".to_owned()),
                StoredValue::Text("escape".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Bytes(vec![b'a', b'\\', 0xff])
    );
}

#[test]
fn stored_value_to_string_uses_pg_hex_bytea_text() {
    assert_eq!(
        stored_value_to_string(&StoredValue::Bytes(vec![0xde, 0xad, 0xbe, 0xef])).unwrap(),
        "\\xdeadbeef"
    );
}

#[test]
fn stored_value_to_string_formats_temporal_values_as_sql_text() {
    assert_eq!(
        stored_value_to_string(&StoredValue::Timestamp(1_700_000_000_000)).unwrap(),
        "2023-11-14 22:13:20"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Timestamp(1_700_000_000_123)).unwrap(),
        "2023-11-14 22:13:20.123000"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Time(3_723_004_005)).unwrap(),
        "01:02:03.004005"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Date(19_724)).unwrap(),
        "2024-01-02"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Date(DATE_POS_INFINITY_DAYS)).unwrap(),
        "infinity"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Date(DATE_NEG_INFINITY_DAYS)).unwrap(),
        "-infinity"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Interval(StoredIntervalValue {
            months: 14,
            millis: 90_061_000,
        }))
        .unwrap(),
        "1 year 2 mons 1 day 01:01:01"
    );
}

#[test]
fn stored_value_to_string_formats_float_special_values_as_pg_text() {
    assert_eq!(
        stored_value_to_string(&StoredValue::Float64(f64::INFINITY)).unwrap(),
        "Infinity"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Float64(f64::NEG_INFINITY)).unwrap(),
        "-Infinity"
    );
    assert_eq!(
        stored_value_to_string(&StoredValue::Float64(f64::NAN)).unwrap(),
        "NaN"
    );
}

#[test]
fn operator_internal_functions_match_pg_semantics() {
    assert_eq!(
        eval_like_function(
            "__db9_like",
            vec![
                StoredValue::Text("A_20".to_owned()),
                StoredValue::Text("A\\_20".to_owned()),
            ],
            false,
            false,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_like_function(
            "__db9_ilike",
            vec![
                StoredValue::Text("A%Twenty".to_owned()),
                StoredValue::Text("a\\%twenty".to_owned()),
            ],
            true,
            false,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_like_function(
            "__db9_like",
            vec![
                StoredValue::Text("A_20".to_owned()),
                StoredValue::Text("A\\_20".to_owned()),
                StoredValue::Text(String::new()),
            ],
            false,
            false,
        )
        .unwrap(),
        StoredValue::Boolean(false)
    );
    assert_eq!(
        eval_like_function(
            "__db9_ilike",
            vec![
                StoredValue::Text("Hello".to_owned()),
                StoredValue::Text("he%".to_owned()),
            ],
            true,
            false,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_between_function(
            "__db9_between",
            vec![
                StoredValue::Int32(20),
                StoredValue::Int32(10),
                StoredValue::Int32(30),
            ],
            false,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_in_list_function(
            "__db9_not_in_list",
            vec![
                StoredValue::Int32(20),
                StoredValue::Int32(10),
                StoredValue::Null,
            ],
            true,
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_is_test_function(
            "__db9_is_unknown",
            vec![StoredValue::Null],
            InternalIsTestKind::Unknown,
            false,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_is_distinct_from_function(
            "__db9_is_not_distinct_from",
            vec![StoredValue::Null, StoredValue::Null],
            true,
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
}

#[test]
fn bitwise_internal_functions_match_db9_semantics() {
    assert_eq!(
        eval_bitwise_binary_function(
            "__db9_bitand",
            vec![StoredValue::Int32(6), StoredValue::Int32(3)],
            "AND",
            |a, b| a & b,
        )
        .unwrap(),
        StoredValue::Int32(2)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int64(3), StoredValue::Int32(2)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int64(12)
    );
    assert_eq!(
        eval_bitwise_not_function("__db9_bitnot", vec![StoredValue::Int32(1)]).unwrap(),
        StoredValue::Int32(!1)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int32(1), StoredValue::Int32(31)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int32(1_i32 << 31)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int64(1), StoredValue::Int32(63)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int64(1_i64 << 63)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int32(1), StoredValue::Int32(32)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int32(1)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int32(1), StoredValue::Int32(33)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int32(2)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shr",
            vec![StoredValue::Int32(1), StoredValue::Int32(32)],
            ">>",
            false,
        )
        .unwrap(),
        StoredValue::Int32(1)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int64(1), StoredValue::Int32(-1)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int64(i64::MIN)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shl",
            vec![StoredValue::Int32(1), StoredValue::Int32(-1)],
            "<<",
            true,
        )
        .unwrap(),
        StoredValue::Int32(i32::MIN)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shr",
            vec![StoredValue::Int32(1), StoredValue::Int32(-1)],
            ">>",
            false,
        )
        .unwrap(),
        StoredValue::Int32(0)
    );
    assert_eq!(
        eval_shift_function(
            "__db9_shr",
            vec![StoredValue::Int64(1), StoredValue::Int32(64)],
            ">>",
            false,
        )
        .unwrap(),
        StoredValue::Int64(1)
    );
}

#[test]
fn array_functions_match_pg_edge_cases() {
    assert_eq!(
        eval_array_cat_function("array_cat", vec![StoredValue::Null, StoredValue::Null]).unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![StoredValue::Array(vec![]), StoredValue::Null],
        )
        .unwrap(),
        StoredValue::Array(vec![])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Null,
                StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![StoredValue::Int32(1)]),
                StoredValue::Int32(3),
            ],
        )
        .unwrap_err()
        .to_string(),
        "function array_cat(integer[], integer) does not exist"
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![StoredValue::Null]),
                StoredValue::Int32(3)
            ],
        )
        .unwrap_err()
        .to_string(),
        "function array_cat(text[], integer) does not exist"
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![StoredValue::Int32(1)]),
                StoredValue::Array(vec![StoredValue::Int64(2)]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int64(2)])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                    StoredValue::Array(vec![StoredValue::Int32(5), StoredValue::Int32(6)]),
                ]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
            StoredValue::Array(vec![StoredValue::Int32(5), StoredValue::Int32(6)]),
        ])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                    StoredValue::Array(vec![StoredValue::Int32(5), StoredValue::Int32(6)]),
                ]),
                StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
            StoredValue::Array(vec![StoredValue::Int32(5), StoredValue::Int32(6)]),
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
        ])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![StoredValue::Int32(1)]),
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3)]),
                ]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1)]),
            StoredValue::Array(vec![StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3)]),
        ])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![]),
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
            StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
        ])
    );
    assert_eq!(
        eval_array_cat_function(
            "array_cat",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
                StoredValue::Array(vec![
                    StoredValue::Array(vec![
                        StoredValue::Int32(5),
                        StoredValue::Int32(6),
                        StoredValue::Int32(7)
                    ]),
                    StoredValue::Array(vec![
                        StoredValue::Int32(8),
                        StoredValue::Int32(9),
                        StoredValue::Int32(10)
                    ]),
                ]),
            ],
        )
        .unwrap_err()
        .to_string(),
        "cannot concatenate incompatible arrays"
    );
    assert_eq!(
        eval_array_position_function(
            "array_position",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
                StoredValue::Int32(1),
            ],
        )
        .unwrap_err()
        .to_string(),
        "searching for elements in multidimensional arrays is not supported"
    );
    assert_eq!(
        eval_array_remove_function(
            "array_remove",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
                StoredValue::Int32(1),
            ],
        )
        .unwrap_err()
        .to_string(),
        "removing elements from multidimensional arrays is not supported"
    );
    assert_eq!(
        eval_array_append_function(
            "array_append",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
                StoredValue::Int32(5),
            ],
        )
        .unwrap_err()
        .to_string(),
        "argument must be empty or one-dimensional array"
    );
    assert_eq!(
        eval_array_prepend_function(
            "array_prepend",
            vec![
                StoredValue::Int32(5),
                StoredValue::Array(vec![
                    StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
                    StoredValue::Array(vec![StoredValue::Int32(3), StoredValue::Int32(4)]),
                ]),
            ],
        )
        .unwrap_err()
        .to_string(),
        "argument must be empty or one-dimensional array"
    );
    assert_eq!(
        eval_array_position_function(
            "array_position",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Int32(1),
                    StoredValue::Null,
                    StoredValue::Int32(2),
                ]),
                StoredValue::Null,
            ],
        )
        .unwrap(),
        StoredValue::Int32(2)
    );
    assert_eq!(
        eval_eq_any_function(
            "__db9_eq_any",
            vec![
                StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Null]),
                StoredValue::Int32(1),
            ],
        )
        .unwrap(),
        StoredValue::Boolean(true)
    );
    assert_eq!(
        eval_eq_any_function(
            "__db9_eq_any",
            vec![
                StoredValue::Array(vec![StoredValue::Null, StoredValue::Int32(2)]),
                StoredValue::Int32(1),
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_eq_any_function(
            "__db9_eq_any",
            vec![
                StoredValue::Array(vec![StoredValue::Null]),
                StoredValue::Null,
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_eq_any_function(
            "__db9_eq_any",
            vec![StoredValue::Array(vec![]), StoredValue::Null],
        )
        .unwrap(),
        StoredValue::Boolean(false)
    );
    assert_eq!(
        eval_array_remove_function(
            "array_remove",
            vec![
                StoredValue::Array(vec![
                    StoredValue::Int32(1),
                    StoredValue::Null,
                    StoredValue::Int32(2),
                    StoredValue::Null,
                ]),
                StoredValue::Null,
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Text(String::new()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Text("abc".to_owned())])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![StoredValue::Text("abc".to_owned()), StoredValue::Null],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Text("a".to_owned()),
            StoredValue::Text("b".to_owned()),
            StoredValue::Text("c".to_owned()),
        ])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text(String::new()),
                StoredValue::Text(",".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text(String::new()),
                StoredValue::Text(",".to_owned()),
                StoredValue::Text(String::new()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Null])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text(String::new()),
                StoredValue::Text(String::new()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![StoredValue::Text(String::new()), StoredValue::Null],
        )
        .unwrap(),
        StoredValue::Array(vec![])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Null,
                StoredValue::Text("b".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![
            StoredValue::Text("a".to_owned()),
            StoredValue::Null,
            StoredValue::Text("c".to_owned()),
        ])
    );
    assert_eq!(
        eval_string_to_array_function(
            "string_to_array",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Text(String::new()),
                StoredValue::Text("abc".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Array(vec![StoredValue::Null])
    );
}

#[test]
fn temporal_values_support_equality_and_ordering() {
    assert_eq!(
        values_equal(
            &StoredValue::Timestamp(1_700_000_000_000),
            &StoredValue::Timestamp(1_700_000_000_000),
        )
        .unwrap(),
        Some(true)
    );
    assert_eq!(
        compare_values(
            &StoredValue::Timestamp(1_700_000_000_000),
            &StoredValue::Timestamp(1_700_000_000_001),
        )
        .unwrap(),
        Some(Ordering::Less)
    );
    assert_eq!(
        values_equal(&StoredValue::Date(19_724), &StoredValue::Date(19_724)).unwrap(),
        Some(true)
    );
    assert_eq!(
        compare_values(&StoredValue::Date(19_724), &StoredValue::Date(19_725)).unwrap(),
        Some(Ordering::Less)
    );
    assert_eq!(
        values_equal(&StoredValue::Time(10), &StoredValue::Time(10)).unwrap(),
        Some(true)
    );
    assert_eq!(
        compare_values(&StoredValue::Time(10), &StoredValue::Time(11)).unwrap(),
        Some(Ordering::Less)
    );
}

#[test]
fn array_ordering_treats_null_elements_as_greater_like_pg() {
    assert_eq!(
        compare_values(
            &StoredValue::Array(vec![StoredValue::Null]),
            &StoredValue::Array(vec![StoredValue::Int32(1)]),
        )
        .unwrap(),
        Some(Ordering::Greater)
    );
    assert_eq!(
        compare_values(
            &StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Null]),
            &StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Int32(2)]),
        )
        .unwrap(),
        Some(Ordering::Greater)
    );
    assert_eq!(
        compare_values(
            &StoredValue::Array(vec![StoredValue::Int32(1)]),
            &StoredValue::Array(vec![StoredValue::Int32(1), StoredValue::Null]),
        )
        .unwrap(),
        Some(Ordering::Less)
    );
}

#[test]
fn jsonb_equality_uses_pg_semantics_for_object_order_and_numbers() {
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb(r#"{"a":1,"b":2}"#.to_owned()),
            &StoredValue::Jsonb(r#"{"b":2,"a":1}"#.to_owned()),
        )
        .unwrap(),
        Some(true)
    );
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb(r#"{"a":[1,{"b":1.0}]}"#.to_owned()),
            &StoredValue::Jsonb(r#"{"a":[1,{"b":1}]}"#.to_owned()),
        )
        .unwrap(),
        Some(true)
    );
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb("18446744073709551616".to_owned()),
            &StoredValue::Jsonb("18446744073709551615".to_owned()),
        )
        .unwrap(),
        Some(false)
    );
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb("18446744073709551616".to_owned()),
            &StoredValue::Jsonb("18446744073709551617".to_owned()),
        )
        .unwrap(),
        Some(false)
    );
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb("9007199254740992.0000000000000000001".to_owned()),
            &StoredValue::Jsonb("9007199254740992.0000000000000000002".to_owned()),
        )
        .unwrap(),
        Some(false)
    );
    assert_eq!(
        values_equal(
            &StoredValue::Jsonb(r#"{"a":[1.0,{"b":2}]}"#.to_owned()),
            &StoredValue::Jsonb(r#"{"a":[1,{"b":2.0}]}"#.to_owned()),
        )
        .unwrap(),
        Some(true)
    );

    let mut deep_json = "1".to_owned();
    for index in (0..=MAX_ARRAY_RECURSION_DEPTH).rev() {
        deep_json = format!("{{\"k{index}\":{deep_json}}}");
    }
    let err = jsonb_values_equal(&deep_json, &deep_json).unwrap_err();
    assert_eq!(err.to_string(), "json value is too deep");
}

#[test]
fn string_functions_propagate_null_args() {
    assert_eq!(
        eval_trim_function(
            "btrim",
            vec![StoredValue::Text("  Bravo  ".to_owned()), StoredValue::Null],
            TrimSide::Both,
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_strpos_function(
            "strpos",
            vec![StoredValue::Text("Bravo".to_owned()), StoredValue::Null],
        )
        .unwrap(),
        StoredValue::Null
    );
}

#[test]
fn round_and_trunc_follow_supported_pg_float8_surface() {
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Float64(2.5)]).unwrap(),
        StoredValue::Float64(2.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Float64(3.5)]).unwrap(),
        StoredValue::Float64(4.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Float64(-2.5)]).unwrap(),
        StoredValue::Float64(-2.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Float64(-3.5)]).unwrap(),
        StoredValue::Float64(-4.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Int32(20)]).unwrap(),
        StoredValue::Float64(20.0)
    );
    assert_eq!(
        eval_trunc_function("trunc", vec![StoredValue::Float64(20.6)]).unwrap(),
        StoredValue::Float64(20.0)
    );
    assert_eq!(
        eval_trunc_function("trunc", vec![StoredValue::Int64(20)]).unwrap(),
        StoredValue::Float64(20.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Float64(20.6), StoredValue::Null])
            .unwrap_err()
            .to_string(),
        "DB9 function 'round' expects 1 args, got 2"
    );
}

#[test]
fn datetime_functions_evaluate_timestamp_inputs() {
    let timestamp = 1_704_198_896_000;

    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("day".to_owned()),
                StoredValue::Timestamp(timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Float64(2.0)
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text(" day ".to_owned()),
                StoredValue::Timestamp(timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Float64(2.0)
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("second".to_owned()),
                StoredValue::Timestamp(1_704_198_896_789),
            ],
        )
        .unwrap(),
        StoredValue::Float64(56.789)
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("second".to_owned()),
                StoredValue::Timestamp(1_704_198_896_789),
            ],
        )
        .unwrap(),
        StoredValue::Numeric(parse_stored_decimal("56.789000").unwrap())
    );
    let one_bc_timestamp = chrono::NaiveDate::from_ymd_opt(0, 1, 1)
        .unwrap()
        .and_hms_opt(0, 0, 0)
        .unwrap()
        .and_utc()
        .timestamp_millis();
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("year".to_owned()),
                StoredValue::Timestamp(one_bc_timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Float64(-1.0)
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("year".to_owned()),
                StoredValue::Timestamp(one_bc_timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Numeric(parse_stored_decimal("-1").unwrap())
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("year".to_owned()),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Float64(f64::INFINITY)
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("epoch".to_owned()),
                StoredValue::Timestamp(i64::MIN),
            ],
        )
        .unwrap(),
        StoredValue::Float64(f64::NEG_INFINITY)
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("month".to_owned()),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("year".to_owned()),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Float64(f64::INFINITY)
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("epoch".to_owned()),
                StoredValue::Timestamp(i64::MIN),
            ],
        )
        .unwrap(),
        StoredValue::Float64(f64::NEG_INFINITY)
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("quarter".to_owned()),
                StoredValue::Timestamp(i64::MIN),
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_date_trunc_function(
            "date_trunc",
            vec![
                StoredValue::Text("hour".to_owned()),
                StoredValue::Timestamp(timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(1_704_196_800_000)
    );
    assert_eq!(
        eval_date_trunc_function(
            "date_trunc",
            vec![
                StoredValue::Text(" hour ".to_owned()),
                StoredValue::Timestamp(timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(1_704_196_800_000)
    );
    assert_eq!(
        eval_date_trunc_function(
            "date_trunc",
            vec![
                StoredValue::Text("day".to_owned()),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(i64::MAX)
    );
    assert_eq!(
        eval_date_trunc_function(
            "date_trunc",
            vec![
                StoredValue::Text("second".to_owned()),
                StoredValue::Timestamp(i64::MIN),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(i64::MIN)
    );
    assert_eq!(
        eval_date_function("date", vec![StoredValue::Timestamp(timestamp)]).unwrap(),
        StoredValue::Date(19_724)
    );
    assert_eq!(
        eval_date_function("date", vec![StoredValue::Timestamp(i64::MAX)]).unwrap(),
        StoredValue::Date(DATE_POS_INFINITY_DAYS)
    );
    assert_eq!(
        eval_date_function("date", vec![StoredValue::Timestamp(i64::MIN)]).unwrap(),
        StoredValue::Date(DATE_NEG_INFINITY_DAYS)
    );
    assert_eq!(
        eval_date_function("date", vec![StoredValue::Text("2024-01-02".to_owned())]).unwrap(),
        StoredValue::Date(19_724)
    );
    assert_eq!(
        eval_make_date_function(
            "make_date",
            vec![
                StoredValue::Int32(2024),
                StoredValue::Int32(1),
                StoredValue::Int32(2),
            ],
        )
        .unwrap(),
        StoredValue::Date(19_724)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(8),
                StoredValue::Int32(15),
                StoredValue::Float64(23.5),
            ],
        )
        .unwrap(),
        StoredValue::Time(29_723_500_000)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(60.0),
            ],
        )
        .unwrap(),
        StoredValue::Time(60_000_000)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(59.9999995),
            ],
        )
        .unwrap(),
        StoredValue::Time(60_000_000)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0000005),
            ],
        )
        .unwrap(),
        StoredValue::Time(0)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0000025),
            ],
        )
        .unwrap(),
        StoredValue::Time(2)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(23),
                StoredValue::Int32(59),
                StoredValue::Float64(60.0),
            ],
        )
        .unwrap(),
        StoredValue::Time(86_400_000_000)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(24),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0),
            ],
        )
        .unwrap(),
        StoredValue::Time(86_400_000_000)
    );
    assert_eq!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(-0.0000005),
            ],
        )
        .unwrap(),
        StoredValue::Time(0)
    );
    eval_make_time_function(
        "make_time",
        vec![
            StoredValue::Int32(0),
            StoredValue::Int32(0),
            StoredValue::Float64(-0.0000006),
        ],
    )
    .unwrap_err();
    eval_make_time_function(
        "make_time",
        vec![
            StoredValue::Int32(24),
            StoredValue::Int32(0),
            StoredValue::Float64(0.000001),
        ],
    )
    .unwrap_err();
    eval_make_time_function(
        "make_time",
        vec![
            StoredValue::Int32(0),
            StoredValue::Int32(60),
            StoredValue::Float64(0.0),
        ],
    )
    .unwrap_err();
    assert!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(25),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0),
            ],
        )
        .unwrap_err()
        .to_string()
        .contains("field value out of range")
    );
    assert!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(-1.0),
            ],
        )
        .unwrap_err()
        .to_string()
        .contains("field value out of range")
    );
    assert!(
        eval_make_time_function(
            "make_time",
            vec![
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(f64::NAN),
            ],
        )
        .unwrap_err()
        .to_string()
        .contains("field value out of range")
    );
    let expected_timestamp = chrono::NaiveDate::from_ymd_opt(2024, 1, 2)
        .unwrap()
        .and_hms_micro_opt(8, 15, 23, 500_000)
        .unwrap()
        .and_utc()
        .timestamp_millis();
    assert_eq!(
        eval_make_timestamp_function(
            "make_timestamp",
            vec![
                StoredValue::Int32(2024),
                StoredValue::Int32(1),
                StoredValue::Int32(2),
                StoredValue::Int32(8),
                StoredValue::Int32(15),
                StoredValue::Float64(23.5),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(expected_timestamp)
    );
    let carried_timestamp = chrono::NaiveDate::from_ymd_opt(2024, 1, 2)
        .unwrap()
        .and_hms_micro_opt(0, 0, 0, 0)
        .unwrap()
        .and_utc()
        .timestamp_millis();
    assert_eq!(
        eval_make_timestamp_function(
            "make_timestamp",
            vec![
                StoredValue::Int32(2024),
                StoredValue::Int32(1),
                StoredValue::Int32(1),
                StoredValue::Int32(23),
                StoredValue::Int32(59),
                StoredValue::Float64(59.9999995),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(carried_timestamp)
    );
    assert_eq!(
        eval_make_timestamp_function(
            "make_timestamp",
            vec![
                StoredValue::Int32(2024),
                StoredValue::Int32(1),
                StoredValue::Int32(1),
                StoredValue::Int32(24),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(carried_timestamp)
    );
    eval_make_timestamp_function(
        "make_timestamp",
        vec![
            StoredValue::Int32(2024),
            StoredValue::Int32(1),
            StoredValue::Int32(1),
            StoredValue::Int32(24),
            StoredValue::Int32(0),
            StoredValue::Float64(0.000001),
        ],
    )
    .unwrap_err();
    assert!(
        eval_make_timestamp_function(
            "make_timestamp",
            vec![
                StoredValue::Int32(2024),
                StoredValue::Int32(1),
                StoredValue::Int32(1),
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(f64::NAN),
            ],
        )
        .unwrap_err()
        .to_string()
        .contains("field value out of range")
    );
    eval_make_timestamp_function(
        "make_timestamp",
        vec![
            StoredValue::Int32(2024),
            StoredValue::Int32(1),
            StoredValue::Int32(1),
            StoredValue::Int32(0),
            StoredValue::Int32(60),
            StoredValue::Float64(0.0),
        ],
    )
    .unwrap_err();
    let expected_timestamptz = Utc
        .timestamp_opt(1_704_187_323, 500_000_000)
        .single()
        .unwrap()
        .timestamp_millis();
    assert_eq!(
        eval_to_timestamp_function("to_timestamp", vec![StoredValue::Float64(1_704_187_323.5)],)
            .unwrap(),
        StoredValue::Timestamp(expected_timestamptz)
    );
    assert_eq!(
        eval_to_timestamp_function("to_timestamp", vec![StoredValue::Float64(f64::INFINITY)])
            .unwrap(),
        StoredValue::Timestamp(i64::MAX)
    );
    assert_eq!(stored_timestamp_to_string(i64::MAX), "infinity");
    assert_eq!(
        eval_to_timestamp_function(
            "to_timestamp",
            vec![StoredValue::Float64(f64::NEG_INFINITY)]
        )
        .unwrap(),
        StoredValue::Timestamp(i64::MIN)
    );
    assert_eq!(stored_timestamp_to_string(i64::MIN), "-infinity");
    assert!(
        eval_to_timestamp_function("to_timestamp", vec![StoredValue::Float64(f64::NAN)])
            .unwrap_err()
            .to_string()
            .contains("timestamp cannot be NaN")
    );
    assert!(
        eval_to_timestamp_function("to_timestamp", vec![StoredValue::Float64(1.0e20)])
            .unwrap_err()
            .to_string()
            .contains("timestamp out of range")
    );
    assert_eq!(
        eval_make_interval_function(
            "make_interval",
            vec![
                StoredValue::Int32(1),
                StoredValue::Int32(2),
                StoredValue::Int32(0),
                StoredValue::Int32(3),
                StoredValue::Int32(4),
                StoredValue::Int32(5),
                StoredValue::Float64(6.5),
            ],
        )
        .unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: 14,
            millis: 273_906_500,
        })
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("epoch".to_owned()),
                StoredValue::Interval(StoredIntervalValue {
                    months: 0,
                    millis: 273_906_500,
                }),
            ],
        )
        .unwrap(),
        StoredValue::Float64(273_906.5)
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("second".to_owned()),
                StoredValue::Interval(StoredIntervalValue {
                    months: 0,
                    millis: 273_906_500,
                }),
            ],
        )
        .unwrap(),
        StoredValue::Float64(6.5)
    );
    assert_eq!(
        eval_date_part_function(
            "extract",
            vec![
                StoredValue::Text("second".to_owned()),
                StoredValue::Interval(StoredIntervalValue {
                    months: 0,
                    millis: 273_906_500,
                }),
            ],
        )
        .unwrap(),
        StoredValue::Numeric(parse_stored_decimal("6.500000").unwrap())
    );
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("epoch".to_owned()),
                StoredValue::Interval(StoredIntervalValue {
                    months: 12,
                    millis: 0,
                }),
            ],
        )
        .unwrap(),
        StoredValue::Float64(31_557_600.0)
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text("2024-01-01 00:00:00".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: 0,
            millis: 131_696_000,
        })
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Timestamp(i64::MAX),
                StoredValue::Timestamp(timestamp),
            ],
        )
        .unwrap(),
        StoredValue::Interval(stored_interval_infinity(false))
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Interval(stored_interval_infinity(true))
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Timestamp(i64::MIN),
                StoredValue::Timestamp(i64::MAX),
            ],
        )
        .unwrap(),
        StoredValue::Interval(stored_interval_infinity(true))
    );
    let err = eval_age_function(
        "age",
        vec![
            StoredValue::Timestamp(i64::MAX),
            StoredValue::Timestamp(i64::MAX),
        ],
    )
    .unwrap_err();
    assert!(err.to_string().contains("interval is out of range"));
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text("YYYY-MM-DD HH24:MI:SS".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02 12:34:56".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text("yyyy-mm-dd hh24:mi:ss".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02 12:34:56".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Date(19_724),
                StoredValue::Text("YYYY-MM-DD".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02".to_owned())
    );
    let one_bc_timestamp = chrono::NaiveDate::from_ymd_opt(0, 1, 1)
        .unwrap()
        .and_hms_opt(0, 0, 0)
        .unwrap()
        .and_utc()
        .timestamp_millis();
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(one_bc_timestamp),
                StoredValue::Text("YYYY-MM-DD".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("0001-01-01".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(i64::MAX),
                StoredValue::Text("YYYY".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("infinity".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(i64::MIN),
                StoredValue::Text("YYYY".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("-infinity".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Date(DATE_POS_INFINITY_DAYS),
                StoredValue::Text("YYYY".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("infinity".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Date(DATE_NEG_INFINITY_DAYS),
                StoredValue::Text("YYYY".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("-infinity".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Text("2024-01-02 08:15:23".to_owned()),
                StoredValue::Text("YYYY-MM-DD HH24:MI:SS".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02 08:15:23".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text(r#"YYYY-MM-DD"T"HH24:MI:SS"Z""#.to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02T12:34:56Z".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text(r#""YYYY" YYYY"#.to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("YYYY 2024".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text(r#"YYYY"-"MM"-"DD"#.to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024-01-02".to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text(r#"YYYY"ab""cd"MM"#.to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text(r#"2024ab"cd01"#.to_owned())
    );
    assert_eq!(
        eval_to_char_function(
            "to_char",
            vec![
                StoredValue::Timestamp(timestamp),
                StoredValue::Text("YYYY\"x".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Text("2024x".to_owned())
    );
}

#[test]
fn datetime_functions_reject_invalid_finite_timestamp_values() {
    let invalid_timestamp = i64::MAX - 1;

    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("year".to_owned()),
                StoredValue::Timestamp(invalid_timestamp),
            ],
        )
        .unwrap_err()
        .to_string(),
        "invalid DB9 timestamp value"
    );
    assert_eq!(
        eval_date_trunc_function(
            "date_trunc",
            vec![
                StoredValue::Text("day".to_owned()),
                StoredValue::Timestamp(invalid_timestamp),
            ],
        )
        .unwrap_err()
        .to_string(),
        "invalid DB9 timestamp value"
    );
    assert_eq!(
        eval_date_function("date", vec![StoredValue::Timestamp(invalid_timestamp)],)
            .unwrap_err()
            .to_string(),
        "invalid DB9 timestamp value"
    );
}

#[test]
fn make_date_and_timestamp_use_postgresql_sql_year_semantics() {
    let year_zero_date_err = eval_make_date_function(
        "make_date",
        vec![
            StoredValue::Int32(0),
            StoredValue::Int32(1),
            StoredValue::Int32(1),
        ],
    )
    .unwrap_err()
    .to_string();
    assert!(year_zero_date_err.contains("date field value out of range"));

    let year_zero_timestamp_err = eval_make_timestamp_function(
        "make_timestamp",
        vec![
            StoredValue::Int32(0),
            StoredValue::Int32(1),
            StoredValue::Int32(1),
            StoredValue::Int32(0),
            StoredValue::Int32(0),
            StoredValue::Float64(0.0),
        ],
    )
    .unwrap_err()
    .to_string();
    assert!(year_zero_timestamp_err.contains("date field value out of range"));

    let one_bc_date = chrono::NaiveDate::from_ymd_opt(0, 1, 1).unwrap();
    assert_eq!(
        eval_make_date_function(
            "make_date",
            vec![
                StoredValue::Int32(-1),
                StoredValue::Int32(1),
                StoredValue::Int32(1),
            ],
        )
        .unwrap(),
        StoredValue::Date(naive_date_to_days(one_bc_date).unwrap())
    );

    let forty_four_bc_date = chrono::NaiveDate::from_ymd_opt(-43, 3, 15).unwrap();
    assert_eq!(
        eval_make_date_function(
            "make_date",
            vec![
                StoredValue::Int32(-44),
                StoredValue::Int32(3),
                StoredValue::Int32(15),
            ],
        )
        .unwrap(),
        StoredValue::Date(naive_date_to_days(forty_four_bc_date).unwrap())
    );

    let one_bc_timestamp = one_bc_date
        .and_hms_micro_opt(0, 0, 0, 0)
        .unwrap()
        .and_utc()
        .timestamp_millis();
    assert_eq!(
        eval_make_timestamp_function(
            "make_timestamp",
            vec![
                StoredValue::Int32(-1),
                StoredValue::Int32(1),
                StoredValue::Int32(1),
                StoredValue::Int32(0),
                StoredValue::Int32(0),
                StoredValue::Float64(0.0),
            ],
        )
        .unwrap(),
        StoredValue::Timestamp(one_bc_timestamp)
    );

    let invalid_date_err = eval_make_timestamp_function(
        "make_timestamp",
        vec![
            StoredValue::Int32(2024),
            StoredValue::Int32(2),
            StoredValue::Int32(30),
            StoredValue::Int32(0),
            StoredValue::Int32(0),
            StoredValue::Float64(0.0),
        ],
    )
    .unwrap_err()
    .to_string();
    assert!(invalid_date_err.contains("date field value out of range"));

    let invalid_time_err = eval_make_timestamp_function(
        "make_timestamp",
        vec![
            StoredValue::Int32(2024),
            StoredValue::Int32(1),
            StoredValue::Int32(1),
            StoredValue::Int32(24),
            StoredValue::Int32(0),
            StoredValue::Float64(0.000001),
        ],
    )
    .unwrap_err()
    .to_string();
    assert!(invalid_time_err.contains("time field value out of range"));
}

#[test]
fn interval_epoch_uses_postgresql_symbolic_month_semantics() {
    fn interval_epoch_seconds(months: i32, millis: i64) -> f64 {
        match eval_date_part_function(
            "date_part",
            vec![
                StoredValue::Text("epoch".to_owned()),
                StoredValue::Interval(StoredIntervalValue { months, millis }),
            ],
        )
        .unwrap()
        {
            StoredValue::Float64(value) => value,
            other => panic!("expected epoch to be float64, got {:?}", other),
        }
    }

    assert_eq!(interval_epoch_seconds(1, 0), 2_592_000.0);
    assert_eq!(interval_epoch_seconds(11, 0), 28_512_000.0);
    assert_eq!(interval_epoch_seconds(12, 0), 31_557_600.0);
    assert_eq!(interval_epoch_seconds(13, 0), 34_149_600.0);
    assert_eq!(interval_epoch_seconds(-1, 0), -2_592_000.0);
    assert_eq!(interval_epoch_seconds(-13, 0), -34_149_600.0);

    let mixed_interval = eval_make_interval_function(
        "make_interval",
        vec![
            StoredValue::Int32(2),
            StoredValue::Int32(3),
            StoredValue::Int32(0),
            StoredValue::Int32(1),
            StoredValue::Int32(2),
            StoredValue::Int32(3),
            StoredValue::Float64(4.0),
        ],
    )
    .unwrap();
    assert_eq!(
        eval_date_part_function(
            "date_part",
            vec![StoredValue::Text("epoch".to_owned()), mixed_interval],
        )
        .unwrap(),
        StoredValue::Float64(70_984_984.0)
    );
}

#[test]
fn cast_expr_supports_text_and_identity_targets() {
    let table = test_db9_table(&[
        Db9TypeKind::Db9TypeKindInt32,
        Db9TypeKind::Db9TypeKindFloat64,
    ]);
    let row = StoredRow {
        values: vec![StoredValue::Int32(42), StoredValue::Float64(183_845.5)],
    };
    let cast_to_text = test_db9_cast_expr(
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindInt32),
        Db9TypeKind::Db9TypeKindText,
    );
    assert_eq!(
        eval_expr(&cast_to_text, &table, &row).unwrap(),
        StoredValue::Text("42".to_owned())
    );

    let identity_cast = test_db9_cast_expr(
        test_db9_column_expr(1, Db9TypeKind::Db9TypeKindFloat64),
        Db9TypeKind::Db9TypeKindFloat64,
    );
    assert_eq!(
        eval_expr(&identity_cast, &table, &row).unwrap(),
        StoredValue::Float64(183_845.5)
    );
}

#[test]
fn interval_datetime_functions_cover_defaults_nulls_and_reverse_order() {
    assert_eq!(
        eval_make_interval_function("make_interval", vec![]).unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: 0,
            millis: 0,
        })
    );
    assert_eq!(
        eval_make_interval_function(
            "make_interval",
            vec![
                StoredValue::Int32(1),
                StoredValue::Int32(2),
                StoredValue::Int32(0),
                StoredValue::Int32(3),
                StoredValue::Int32(4),
                StoredValue::Int32(5),
                StoredValue::Null,
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Text("2024-01-01 00:00:00".to_owned()),
                StoredValue::Text("2024-02-01 00:00:00".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: -1,
            millis: 0,
        })
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Null,
                StoredValue::Text("2024-02-01 00:00:00".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_age_function(
            "age",
            vec![
                StoredValue::Text("2024-05-01 08:00:00".to_owned()),
                StoredValue::Text("2024-03-15 12:34:56.789".to_owned()),
            ],
        )
        .unwrap(),
        StoredValue::Interval(StoredIntervalValue {
            months: 1,
            millis: 1_452_303_211,
        })
    );
}

#[test]
fn function_call_aliases_dispatch_to_existing_scalar_evaluators() {
    let table = test_db9_table(&[]);
    let row = StoredRow { values: vec![] };
    let timestamp = 1_704_198_896_000;

    assert_eq!(
        eval_function_call(
            &test_db9_func_call(
                "substr",
                vec![
                    test_db9_text_const_expr("  Bravo  "),
                    test_db9_int32_const_expr(3),
                    test_db9_int32_const_expr(5),
                ],
            ),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Text("Bravo".to_owned())
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call("trim", vec![test_db9_text_const_expr("  Bravo  ")]),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Text("Bravo".to_owned())
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call(
                "position",
                vec![
                    test_db9_text_const_expr("  Bravo  "),
                    test_db9_text_const_expr("Bravo"),
                ],
            ),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Int32(3)
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call("ceiling", vec![test_db9_float64_const_expr(-20.6)]),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Float64(-20.0)
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call(
                "extract",
                vec![
                    test_db9_text_const_expr("day"),
                    test_db9_timestamp_const_expr(timestamp),
                ],
            ),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Numeric(parse_stored_decimal("2").unwrap())
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call("date", vec![test_db9_date_const_expr(19_725)]),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Date(19_725)
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call(
                "make_date",
                vec![
                    test_db9_int32_const_expr(2024),
                    test_db9_int32_const_expr(1),
                    test_db9_int32_const_expr(2),
                ],
            ),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Date(19_724)
    );
    assert_eq!(
        eval_function_call(
            &test_db9_func_call(
                "make_time",
                vec![
                    test_db9_int32_const_expr(8),
                    test_db9_int32_const_expr(15),
                    test_db9_float64_const_expr(23.5),
                ],
            ),
            &table,
            &row,
        )
        .unwrap(),
        StoredValue::Time(29_723_500_000)
    );
}

#[test]
fn local_only_temporal_builders_are_not_production_dispatched() {
    let table = test_db9_table(&[]);
    let row = StoredRow { values: vec![] };

    for func_call in [
        test_db9_func_call(
            "make_timestamp",
            vec![
                test_db9_int32_const_expr(2024),
                test_db9_int32_const_expr(1),
                test_db9_int32_const_expr(2),
                test_db9_int32_const_expr(8),
                test_db9_int32_const_expr(15),
                test_db9_float64_const_expr(23.5),
            ],
        ),
        test_db9_func_call("make_interval", vec![]),
        test_db9_func_call(
            "to_timestamp",
            vec![test_db9_float64_const_expr(1_704_187_323.5)],
        ),
    ] {
        let err = eval_function_call(&func_call, &table, &row).unwrap_err();
        assert!(
            err.to_string().contains("is not supported yet"),
            "unexpected error: {}",
            err
        );
    }
}

#[test]
fn extended_string_functions_match_server_semantics() {
    assert_eq!(
        eval_split_part_function(
            "split_part",
            vec![
                StoredValue::Text("a-Bravo-c".to_owned()),
                StoredValue::Text("-".to_owned()),
                StoredValue::Int32(2),
            ],
        )
        .unwrap(),
        StoredValue::Text("Bravo".to_owned())
    );
    assert_eq!(
        eval_split_part_function(
            "split_part",
            vec![
                StoredValue::Text("a,b,c,d".to_owned()),
                StoredValue::Text(",".to_owned()),
                StoredValue::Int64(-2),
            ],
        )
        .unwrap(),
        StoredValue::Text("c".to_owned())
    );
    assert_eq!(
        eval_split_part_function(
            "split_part",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Text(String::new()),
                StoredValue::Int32(1),
            ],
        )
        .unwrap(),
        StoredValue::Text("abc".to_owned())
    );
    assert_eq!(
        eval_split_part_function(
            "split_part",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Text(String::new()),
                StoredValue::Int32(2),
            ],
        )
        .unwrap(),
        StoredValue::Text(String::new())
    );
    assert_eq!(
        eval_split_part_function(
            "split_part",
            vec![
                StoredValue::Text("abc".to_owned()),
                StoredValue::Text(String::new()),
                StoredValue::Int32(-1),
            ],
        )
        .unwrap(),
        StoredValue::Text("abc".to_owned())
    );
    let err = eval_split_part_function(
        "split_part",
        vec![
            StoredValue::Text("a,b,c".to_owned()),
            StoredValue::Text(",".to_owned()),
            StoredValue::Int32(0),
        ],
    )
    .unwrap_err();
    assert!(err.to_string().contains("field position must not be zero"));
}

#[test]
fn extended_math_functions_match_server_semantics() {
    assert_eq!(
        eval_trunc_function("trunc", vec![StoredValue::Float64(8.75)]).unwrap(),
        StoredValue::Float64(8.0)
    );
    assert_eq!(
        eval_round_function("round", vec![StoredValue::Int32(8)]).unwrap(),
        StoredValue::Float64(8.0)
    );
    assert_eq!(
        eval_mod_function("mod", vec![StoredValue::Int64(27), StoredValue::Int64(4)]).unwrap(),
        StoredValue::Int64(3)
    );
    assert_eq!(
        eval_mod_function(
            "mod",
            vec![StoredValue::Int32(i32::MIN), StoredValue::Int32(-1)],
        )
        .unwrap(),
        StoredValue::Int32(0)
    );
    assert_eq!(
        eval_mod_function(
            "mod",
            vec![StoredValue::Int64(i64::MIN), StoredValue::Int64(-1)],
        )
        .unwrap(),
        StoredValue::Int64(0)
    );
    assert_eq!(
        eval_mod_function(
            "mod",
            vec![StoredValue::Int64(i64::MIN), StoredValue::Int32(-1)],
        )
        .unwrap(),
        StoredValue::Int64(0)
    );
    let err =
        eval_mod_function("mod", vec![StoredValue::Int64(5), StoredValue::Int64(0)]).unwrap_err();
    assert!(err.to_string().contains("division by zero"));
    let float_mod = eval_mod_function(
        "mod",
        vec![StoredValue::Float64(5.0), StoredValue::Float64(2.0)],
    )
    .unwrap_err();
    assert!(float_mod.to_string().contains("does not support"));
    assert_eq!(
        eval_mod_function("mod", vec![StoredValue::Null, StoredValue::Int64(3)]).unwrap(),
        StoredValue::Null
    );
    assert_eq!(
        eval_mod_function("mod", vec![StoredValue::Int32(7), StoredValue::Int64(3)]).unwrap(),
        StoredValue::Int64(1)
    );
    assert_eq!(
        eval_mod_function("mod", vec![StoredValue::Int64(7), StoredValue::Int32(3)]).unwrap(),
        StoredValue::Int64(1)
    );
    let mixed_zero =
        eval_mod_function("mod", vec![StoredValue::Int32(7), StoredValue::Int64(0)]).unwrap_err();
    assert!(mixed_zero.to_string().contains("division by zero"));
    assert_eq!(
        eval_mod_function(
            "mod",
            vec![StoredValue::Int64(i64::MIN), StoredValue::Int32(-1)],
        )
        .unwrap(),
        StoredValue::Int64(0)
    );
    let numeric_mod = eval_mod_function(
        "mod",
        vec![
            StoredValue::Numeric(StoredDecimal {
                lo: 7,
                mid: 0,
                hi: 0,
                negative: false,
                scale: 0,
            }),
            StoredValue::Numeric(StoredDecimal {
                lo: 3,
                mid: 0,
                hi: 0,
                negative: false,
                scale: 0,
            }),
        ],
    )
    .unwrap_err();
    assert!(numeric_mod.to_string().contains("does not support"));
    assert_eq!(
        eval_sign_function("sign", vec![StoredValue::Int32(-7)]).unwrap(),
        StoredValue::Float64(-1.0)
    );
    assert_eq!(
        eval_sign_function("sign", vec![StoredValue::Int64(0)]).unwrap(),
        StoredValue::Float64(0.0)
    );
    assert_eq!(
        eval_sign_function("sign", vec![StoredValue::Float64(2.5)]).unwrap(),
        StoredValue::Float64(1.0)
    );
    assert_eq!(
        eval_pi_function("pi", vec![]).unwrap(),
        StoredValue::Float64(std::f64::consts::PI)
    );

    let sqrt = eval_float64_unary_math_function("sqrt", vec![StoredValue::Int32(16)], |value| {
        value.sqrt()
    })
    .unwrap();
    let cbrt = eval_float64_unary_math_function("cbrt", vec![StoredValue::Int32(8)], pg_like_cbrt)
        .unwrap();
    let power =
        eval_power_function("power", vec![StoredValue::Int32(2), StoredValue::Int32(3)]).unwrap();
    let exp =
        eval_float64_unary_math_function("exp", vec![StoredValue::Int32(0)], |value| value.exp())
            .unwrap();
    let ln =
        eval_float64_unary_math_function("ln", vec![StoredValue::Int32(1)], |value| value.ln())
            .unwrap();
    let log10 = eval_float64_unary_math_function("log10", vec![StoredValue::Int32(100)], |value| {
        value.log10()
    })
    .unwrap();
    let degrees = eval_float64_unary_math_function(
        "degrees",
        vec![StoredValue::Float64(std::f64::consts::PI)],
        |value| value.to_degrees(),
    )
    .unwrap();
    let radians =
        eval_float64_unary_math_function("radians", vec![StoredValue::Int32(180)], |value| {
            value.to_radians()
        })
        .unwrap();
    let asin = eval_float64_unary_math_function("asin", vec![StoredValue::Float64(0.0)], |value| {
        value.asin()
    })
    .unwrap();
    let acos = eval_float64_unary_math_function("acos", vec![StoredValue::Float64(1.0)], |value| {
        value.acos()
    })
    .unwrap();
    let atan = eval_float64_unary_math_function("atan", vec![StoredValue::Float64(0.0)], |value| {
        value.atan()
    })
    .unwrap();
    let atan2 = eval_atan2_function(
        "atan2",
        vec![StoredValue::Float64(1.0), StoredValue::Float64(1.0)],
    )
    .unwrap();
    let hashtext =
        eval_hashtext_function("hashtext", vec![StoredValue::Text("hello".to_owned())]).unwrap();
    let width_bucket = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(5.0),
            StoredValue::Float64(0.0),
            StoredValue::Float64(10.0),
            StoredValue::Int32(5),
        ],
    )
    .unwrap();

    let assert_float = |value: StoredValue, expected: f64| match value {
        StoredValue::Float64(value) => assert!((value - expected).abs() < 1e-12),
        other => panic!("expected FLOAT64 result, got {:?}", other),
    };

    assert_float(sqrt, 4.0);
    assert_float(cbrt, 2.0);
    assert_float(power, 8.0);
    assert_float(exp, 1.0);
    assert_float(ln, 0.0);
    assert_float(log10, 2.0);
    assert_float(degrees, 180.0);
    assert_float(radians, std::f64::consts::PI);
    assert_float(asin, 0.0);
    assert_float(acos, 0.0);
    assert_float(atan, 0.0);
    assert_float(atan2, std::f64::consts::FRAC_PI_4);
    assert_eq!(hashtext, StoredValue::Int32(-1_870_292_951));
    assert_eq!(width_bucket, StoredValue::Int32(3));
    assert_eq!(
        eval_width_bucket_function(
            "width_bucket",
            vec![
                StoredValue::Float64(f64::INFINITY),
                StoredValue::Float64(0.0),
                StoredValue::Float64(10.0),
                StoredValue::Int32(4),
            ],
        )
        .unwrap(),
        StoredValue::Int32(5)
    );
    assert_eq!(
        eval_width_bucket_function(
            "width_bucket",
            vec![
                StoredValue::Float64(f64::NEG_INFINITY),
                StoredValue::Float64(0.0),
                StoredValue::Float64(10.0),
                StoredValue::Int32(4),
            ],
        )
        .unwrap(),
        StoredValue::Int32(0)
    );
    let width_bucket_overflow = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(10.0),
            StoredValue::Float64(0.0),
            StoredValue::Float64(1.0),
            StoredValue::Int32(i32::MAX),
        ],
    )
    .unwrap_err();
    assert!(
        width_bucket_overflow
            .to_string()
            .contains("integer out of range")
    );
    assert_eq!(
        eval_width_bucket_function(
            "width_bucket",
            vec![
                StoredValue::Numeric(parse_stored_decimal("9007199254740993").unwrap()),
                StoredValue::Numeric(parse_stored_decimal("9007199254740992").unwrap()),
                StoredValue::Numeric(parse_stored_decimal("9007199254740994").unwrap()),
                StoredValue::Int32(2),
            ],
        )
        .unwrap(),
        StoredValue::Int32(2)
    );
}

#[test]
fn math_domain_errors_match_db9_local_contract() {
    let sqrt_err = eval_sqrt_function("sqrt", vec![StoredValue::Float64(-1.0)]).unwrap_err();
    assert!(
        sqrt_err
            .to_string()
            .contains("cannot take square root of a negative number")
    );

    let ln_err = eval_ln_function("ln", vec![StoredValue::Float64(-1.0)]).unwrap_err();
    assert!(
        ln_err
            .to_string()
            .contains("cannot take logarithm of a negative number")
    );

    let ln_zero_err = eval_ln_function("ln", vec![StoredValue::Float64(0.0)]).unwrap_err();
    assert!(
        ln_zero_err
            .to_string()
            .contains("cannot take logarithm of zero")
    );

    let log10_err = eval_log10_function("log10", vec![StoredValue::Float64(0.0)]).unwrap_err();
    assert!(
        log10_err
            .to_string()
            .contains("cannot take logarithm of zero")
    );

    let asin_err = eval_asin_function("asin", vec![StoredValue::Float64(2.0)]).unwrap_err();
    assert!(asin_err.to_string().contains("input is out of range"));
    let asin_inf_err =
        eval_asin_function("asin", vec![StoredValue::Float64(f64::INFINITY)]).unwrap_err();
    assert!(asin_inf_err.to_string().contains("input is out of range"));
    match eval_asin_function("asin", vec![StoredValue::Float64(f64::NAN)]).unwrap() {
        StoredValue::Float64(value) => assert!(value.is_nan()),
        other => panic!("expected NaN float64 result, got {:?}", other),
    }

    let acos_err = eval_acos_function("acos", vec![StoredValue::Float64(-2.0)]).unwrap_err();
    assert!(acos_err.to_string().contains("input is out of range"));
    let acos_neg_inf_err =
        eval_acos_function("acos", vec![StoredValue::Float64(f64::NEG_INFINITY)]).unwrap_err();
    assert!(
        acos_neg_inf_err
            .to_string()
            .contains("input is out of range")
    );

    let exp_numeric_err = eval_exp_function(
        "exp",
        vec![StoredValue::Numeric(parse_stored_decimal("1000").unwrap())],
    )
    .unwrap_err();
    assert!(
        exp_numeric_err
            .to_string()
            .contains("value overflows numeric format")
    );

    let exp_float_err = eval_exp_function("exp", vec![StoredValue::Float64(1000.0)]).unwrap_err();
    assert!(
        exp_float_err
            .to_string()
            .contains("value overflows numeric format")
    );
    assert_eq!(
        eval_exp_function("exp", vec![StoredValue::Float64(f64::INFINITY)]).unwrap(),
        StoredValue::Float64(f64::INFINITY)
    );

    let power_domain_err = eval_power_function(
        "power",
        vec![StoredValue::Float64(-1.0), StoredValue::Float64(0.5)],
    )
    .unwrap_err();
    assert!(
        power_domain_err
            .to_string()
            .contains("a negative number raised to a non-integer power yields a complex result")
    );
    let power_negative_infinity_domain_err = eval_power_function(
        "power",
        vec![
            StoredValue::Float64(f64::NEG_INFINITY),
            StoredValue::Float64(0.5),
        ],
    )
    .unwrap_err();
    assert!(
        power_negative_infinity_domain_err
            .to_string()
            .contains("a negative number raised to a non-integer power yields a complex result")
    );
    let power_zero_negative_err = eval_power_function(
        "power",
        vec![StoredValue::Float64(0.0), StoredValue::Float64(-1.0)],
    )
    .unwrap_err();
    assert!(
        power_zero_negative_err
            .to_string()
            .contains("zero raised to a negative power is undefined")
    );

    let power_overflow_err = eval_power_function(
        "power",
        vec![StoredValue::Float64(10.0), StoredValue::Float64(400.0)],
    )
    .unwrap_err();
    assert!(
        power_overflow_err
            .to_string()
            .contains("value overflows numeric format")
    );
    assert_eq!(
        eval_power_function(
            "power",
            vec![
                StoredValue::Float64(f64::INFINITY),
                StoredValue::Float64(2.0)
            ],
        )
        .unwrap(),
        StoredValue::Float64(f64::INFINITY)
    );

    let power_ok = eval_power_function(
        "power",
        vec![StoredValue::Float64(-2.0), StoredValue::Float64(3.0)],
    )
    .unwrap();
    match power_ok {
        StoredValue::Float64(value) => assert!((value + 8.0).abs() < 1e-12),
        other => panic!("expected float64 power result, got {:?}", other),
    }

    let width_bucket_nan_err = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(f64::NAN),
            StoredValue::Float64(0.0),
            StoredValue::Float64(10.0),
            StoredValue::Int32(4),
        ],
    )
    .unwrap_err();
    assert!(
        width_bucket_nan_err
            .to_string()
            .contains("operand, lower bound, and upper bound cannot be NaN")
    );

    let width_bucket_infinite_bound_err = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(1.0),
            StoredValue::Float64(f64::NEG_INFINITY),
            StoredValue::Float64(10.0),
            StoredValue::Int32(4),
        ],
    )
    .unwrap_err();
    assert!(
        width_bucket_infinite_bound_err
            .to_string()
            .contains("lower and upper bounds must be finite")
    );

    let width_bucket_count_err = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(1.0),
            StoredValue::Float64(0.0),
            StoredValue::Float64(10.0),
            StoredValue::Int32(0),
        ],
    )
    .unwrap_err();
    assert!(
        width_bucket_count_err
            .to_string()
            .contains("count must be greater than zero")
    );

    let width_bucket_equal_bound_err = eval_width_bucket_function(
        "width_bucket",
        vec![
            StoredValue::Float64(1.0),
            StoredValue::Float64(10.0),
            StoredValue::Float64(10.0),
            StoredValue::Int32(4),
        ],
    )
    .unwrap_err();
    assert!(
        width_bucket_equal_bound_err
            .to_string()
            .contains("lower bound cannot equal upper bound")
    );
}

#[test]
fn between_uses_three_valued_logic() {
    assert_eq!(
        eval_between_function(
            "__db9_between",
            vec![
                StoredValue::Int32(5),
                StoredValue::Null,
                StoredValue::Int32(3)
            ],
            false,
        )
        .unwrap(),
        StoredValue::Boolean(false)
    );
    assert_eq!(
        eval_between_function(
            "__db9_between",
            vec![
                StoredValue::Int32(5),
                StoredValue::Null,
                StoredValue::Int32(10)
            ],
            false,
        )
        .unwrap(),
        StoredValue::Null
    );
}

#[test]
fn schema_evolved_column_refs_evaluate_to_null() {
    let table = test_db9_named_table(&[
        ("id", Db9TypeKind::Db9TypeKindInt64),
        ("name", Db9TypeKind::Db9TypeKindText),
    ]);
    let row = StoredRow {
        values: vec![StoredValue::Int64(7)],
    };
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindText));
    let mut column_ref = kvproto::db9_coprocessor::Db9ColumnRefExpr::default();
    column_ref.set_column_index(1);
    column_ref.set_column_name("name".to_owned());
    expr.set_column_ref(column_ref);

    assert_eq!(eval_expr(&expr, &table, &row).unwrap(), StoredValue::Null);
}

#[test]
fn schema_evolved_missing_columns_work_in_filters() {
    let table = test_db9_named_table(&[
        ("id", Db9TypeKind::Db9TypeKindInt64),
        ("name", Db9TypeKind::Db9TypeKindText),
    ]);
    let row = StoredRow {
        values: vec![StoredValue::Int64(7)],
    };
    let mut column_expr = Db9Expr::default();
    column_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindText));
    let mut column_ref = kvproto::db9_coprocessor::Db9ColumnRefExpr::default();
    column_ref.set_column_index(1);
    column_ref.set_column_name("name".to_owned());
    column_expr.set_column_ref(column_ref);

    let mut filter = Db9Expr::default();
    filter.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindBoolean));
    let mut is_null = kvproto::db9_coprocessor::Db9IsNullExpr::default();
    is_null.set_expr(column_expr);
    filter.set_is_null(is_null);

    assert!(row_matches_selection(&filter, &table, &row).unwrap());
}

#[test]
fn selection_eval_uses_stable_request_output_limit() {
    let table = test_db9_table(&[Db9TypeKind::Db9TypeKindText]);
    let row = StoredRow {
        values: vec![StoredValue::Text("abcdef".to_owned())],
    };
    let selection = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpEq,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_func_expr(
            "left",
            Db9TypeKind::Db9TypeKindText,
            vec![
                test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
                test_db9_int32_const_expr(5),
            ],
        ),
        test_db9_text_const_expr("abcde"),
    );

    let err =
        with_db9_string_output_byte_limit(4, || row_matches_selection(&selection, &table, &row))
            .unwrap_err();
    assert_eq!(err.to_string(), "requested length too large");

    assert!(
        row_matches_selection_with_output_limit(
            stable_selection_output_bytes_for_response(4),
            &selection,
            &table,
            &row,
        )
        .expect("selection should use the full request-scoped limit")
    );
}

#[test]
fn selection_output_budget_is_capped_for_oversized_max_resp_size() {
    assert_eq!(
        stable_selection_output_bytes_for_response(1),
        DEFAULT_MIN_SELECTION_STRING_OUTPUT_BYTES
    );
    assert_eq!(
        stable_selection_output_bytes_for_response(u64::MAX),
        MAX_SELECTION_STRING_OUTPUT_BYTES_CAP
    );
}

#[test]
fn scan_prefix_mismatch_error_redacts_raw_keys() {
    let expected_prefix = b"expected_prefix";
    let scanned_key = b"tenant_secret_scanned_key";
    let err = scan_visible_prefix_len(expected_prefix, scanned_key).unwrap_err();
    let message = err.to_string();

    assert!(message.contains("expected_prefix_len=15"));
    assert!(message.contains("scanned_key_len=25"));
    assert!(!message.contains(&hex::encode(expected_prefix)));
    assert!(!message.contains(&hex::encode(scanned_key)));
}

#[test]
fn column_refs_follow_logical_table_mapping_before_wire_ordinal() {
    let table = test_db9_named_table(&[
        ("id", Db9TypeKind::Db9TypeKindInt64),
        ("name", Db9TypeKind::Db9TypeKindText),
    ]);
    let row = StoredRow {
        values: vec![StoredValue::Int64(7), StoredValue::Text("alice".to_owned())],
    };
    let mut expr = Db9Expr::default();
    expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindText));
    let mut column_ref = kvproto::db9_coprocessor::Db9ColumnRefExpr::default();
    column_ref.set_column_index(0);
    column_ref.set_column_name("name".to_owned());
    expr.set_column_ref(column_ref);

    assert_eq!(
        eval_expr(&expr, &table, &row).unwrap(),
        StoredValue::Text("alice".to_owned())
    );
}

#[test]
fn abs_errors_on_min_signed_values() {
    let row = StoredRow { values: vec![] };
    let table = test_db9_table(&[]);

    let mut int32_call = Db9FuncCallExpr::default();
    int32_call.set_function_name("abs".to_owned());
    int32_call
        .mut_args()
        .push(test_db9_int32_const_expr(i32::MIN));
    assert_eq!(
        eval_function_call(&int32_call, &table, &row)
            .unwrap_err()
            .to_string(),
        "integer out of range"
    );

    let mut int64_call = Db9FuncCallExpr::default();
    int64_call.set_function_name("abs".to_owned());
    let mut int64_arg = Db9Expr::default();
    int64_arg.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindInt64));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut value = Db9Value::default();
    value.set_int64_value(i64::MIN);
    constant.set_value(value);
    int64_arg.set_constant(constant);
    int64_call.mut_args().push(int64_arg);
    assert_eq!(
        eval_function_call(&int64_call, &table, &row)
            .unwrap_err()
            .to_string(),
        "bigint out of range"
    );
}

#[test]
fn special_forms_short_circuit_guarded_domain_errors() {
    let row = StoredRow { values: vec![] };
    let table = test_db9_table(&[]);

    let coalesce = test_db9_func_call(
        "coalesce",
        vec![
            test_db9_text_const_expr("safe"),
            test_db9_cast_expr(
                test_db9_func_expr(
                    "sqrt",
                    Db9TypeKind::Db9TypeKindFloat64,
                    vec![test_db9_float64_const_expr(-1.0)],
                ),
                Db9TypeKind::Db9TypeKindText,
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&coalesce, &table, &row).unwrap(),
        StoredValue::Text("safe".to_owned())
    );

    let row_with_text = StoredRow {
        values: vec![StoredValue::Text("safe".to_owned())],
    };
    let table_with_text = test_db9_table(&[Db9TypeKind::Db9TypeKindText]);
    let coalesce_with_runtime_guard = test_db9_func_call(
        "coalesce",
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
            test_db9_func_expr(
                "chr",
                Db9TypeKind::Db9TypeKindText,
                vec![test_db9_int32_const_expr(-1)],
            ),
        ],
    );
    let err = eval_function_call(
        &coalesce_with_runtime_guard,
        &table_with_text,
        &row_with_text,
    )
    .unwrap_err();
    assert_eq!(err.to_string(), "character number must be positive");

    let coalesce_with_nested_fold_error = test_db9_func_call(
        "coalesce",
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
            test_db9_func_expr(
                "left",
                Db9TypeKind::Db9TypeKindText,
                vec![
                    test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
                    test_db9_func_expr(
                        "abs",
                        Db9TypeKind::Db9TypeKindInt32,
                        vec![test_db9_int32_const_expr(i32::MIN)],
                    ),
                ],
            ),
        ],
    );
    let err = eval_function_call(
        &coalesce_with_nested_fold_error,
        &table_with_text,
        &row_with_text,
    )
    .unwrap_err();
    assert_eq!(err.to_string(), "integer out of range");

    let in_list_with_runtime_match = test_db9_func_call(
        "__db9_in_list",
        vec![
            test_db9_column_expr(0, Db9TypeKind::Db9TypeKindText),
            test_db9_text_const_expr("safe"),
            test_db9_func_expr(
                "chr",
                Db9TypeKind::Db9TypeKindText,
                vec![test_db9_int32_const_expr(-1)],
            ),
        ],
    );
    let err = eval_function_call(
        &in_list_with_runtime_match,
        &table_with_text,
        &row_with_text,
    )
    .unwrap_err();
    assert_eq!(err.to_string(), "character number must be positive");

    fn sqrt_domain_error_predicate() -> Db9Expr {
        let mut binary = kvproto::db9_coprocessor::Db9BinaryExpr::default();
        binary.set_left(test_db9_func_expr(
            "sqrt",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![test_db9_float64_const_expr(-1.0)],
        ));
        binary.set_op(Db9BinaryOp::Db9BinaryOpGt);
        binary.set_right(test_db9_float64_const_expr(0.0));

        let mut expr = Db9Expr::default();
        expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindBoolean));
        expr.set_binary(binary);
        expr
    }

    let mut and_binary = kvproto::db9_coprocessor::Db9BinaryExpr::default();
    and_binary.set_left(test_db9_bool_const_expr(false));
    and_binary.set_op(Db9BinaryOp::Db9BinaryOpAnd);
    and_binary.set_right(sqrt_domain_error_predicate());
    let mut and_expr = Db9Expr::default();
    and_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindBoolean));
    and_expr.set_binary(and_binary);
    assert_eq!(
        eval_expr(&and_expr, &table, &row).unwrap(),
        StoredValue::Boolean(false)
    );

    let mut or_binary = kvproto::db9_coprocessor::Db9BinaryExpr::default();
    or_binary.set_left(test_db9_bool_const_expr(true));
    or_binary.set_op(Db9BinaryOp::Db9BinaryOpOr);
    or_binary.set_right(sqrt_domain_error_predicate());
    let mut or_expr = Db9Expr::default();
    or_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindBoolean));
    or_expr.set_binary(or_binary);
    assert_eq!(
        eval_expr(&or_expr, &table, &row).unwrap(),
        StoredValue::Boolean(true)
    );

    let bool_table = test_db9_table(&[Db9TypeKind::Db9TypeKindBoolean]);
    let runtime_and = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpAnd,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindBoolean),
        sqrt_domain_error_predicate(),
    );
    let err = eval_expr(
        &runtime_and,
        &bool_table,
        &StoredRow {
            values: vec![StoredValue::Boolean(false)],
        },
    )
    .unwrap_err();
    assert_eq!(
        err.to_string(),
        "cannot take square root of a negative number"
    );

    let runtime_or = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpOr,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_column_expr(0, Db9TypeKind::Db9TypeKindBoolean),
        sqrt_domain_error_predicate(),
    );
    let err = eval_expr(
        &runtime_or,
        &bool_table,
        &StoredRow {
            values: vec![StoredValue::Boolean(true)],
        },
    )
    .unwrap_err();
    assert_eq!(
        err.to_string(),
        "cannot take square root of a negative number"
    );

    let between_false = test_db9_func_call(
        "__db9_between",
        vec![
            test_db9_int32_const_expr(1),
            test_db9_int32_const_expr(10),
            test_db9_func_expr(
                "sqrt",
                Db9TypeKind::Db9TypeKindFloat64,
                vec![test_db9_float64_const_expr(-1.0)],
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&between_false, &table, &row).unwrap(),
        StoredValue::Boolean(false)
    );

    let not_between_true = test_db9_func_call(
        "__db9_not_between",
        vec![
            test_db9_int32_const_expr(1),
            test_db9_int32_const_expr(10),
            test_db9_func_expr(
                "sqrt",
                Db9TypeKind::Db9TypeKindFloat64,
                vec![test_db9_float64_const_expr(-1.0)],
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&not_between_true, &table, &row).unwrap(),
        StoredValue::Boolean(true)
    );

    let between_high_error = test_db9_func_call(
        "__db9_between",
        vec![
            test_db9_int32_const_expr(20),
            test_db9_int32_const_expr(10),
            test_db9_func_expr(
                "sqrt",
                Db9TypeKind::Db9TypeKindFloat64,
                vec![test_db9_float64_const_expr(-1.0)],
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&between_high_error, &table, &row)
            .unwrap_err()
            .to_string(),
        "cannot take square root of a negative number"
    );

    let null_low_high_error = test_db9_func_call(
        "__db9_between",
        vec![
            test_db9_int32_const_expr(1),
            test_db9_null_const_expr(Db9TypeKind::Db9TypeKindInt32),
            test_db9_func_expr(
                "sqrt",
                Db9TypeKind::Db9TypeKindFloat64,
                vec![test_db9_float64_const_expr(-1.0)],
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&null_low_high_error, &table, &row)
            .unwrap_err()
            .to_string(),
        "cannot take square root of a negative number"
    );

    let null_value_high_error = test_db9_func_call(
        "__db9_between",
        vec![
            test_db9_null_const_expr(Db9TypeKind::Db9TypeKindInt32),
            test_db9_int32_const_expr(10),
            test_db9_func_expr(
                "sqrt",
                Db9TypeKind::Db9TypeKindFloat64,
                vec![test_db9_float64_const_expr(-1.0)],
            ),
        ],
    );
    assert_eq!(
        eval_function_call(&null_value_high_error, &table, &row)
            .unwrap_err()
            .to_string(),
        "cannot take square root of a negative number"
    );

    let float_table = test_db9_table(&[Db9TypeKind::Db9TypeKindFloat64]);
    let negative_float_row = StoredRow {
        values: vec![StoredValue::Float64(-1.0)],
    };
    let runtime_sqrt_predicate = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpGt,
        Db9TypeKind::Db9TypeKindBoolean,
        test_db9_func_expr(
            "sqrt",
            Db9TypeKind::Db9TypeKindFloat64,
            vec![test_db9_column_expr(0, Db9TypeKind::Db9TypeKindFloat64)],
        ),
        test_db9_float64_const_expr(0.0),
    );
    let and_false = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpAnd,
        Db9TypeKind::Db9TypeKindBoolean,
        runtime_sqrt_predicate.clone(),
        test_db9_bool_const_expr(false),
    );
    assert_eq!(
        eval_expr(&and_false, &float_table, &negative_float_row).unwrap(),
        StoredValue::Boolean(false)
    );
    let or_true = test_db9_binary_expr(
        Db9BinaryOp::Db9BinaryOpOr,
        Db9TypeKind::Db9TypeKindBoolean,
        runtime_sqrt_predicate,
        test_db9_bool_const_expr(true),
    );
    assert_eq!(
        eval_expr(&or_true, &float_table, &negative_float_row).unwrap(),
        StoredValue::Boolean(true)
    );
}

#[test]
fn unary_minus_errors_on_min_signed_values() {
    let row = StoredRow { values: vec![] };
    let table = test_db9_table(&[]);

    let mut int32_expr = Db9Expr::default();
    int32_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindInt32));
    let mut int32_unary = kvproto::db9_coprocessor::Db9UnaryExpr::default();
    int32_unary.set_op(Db9UnaryOp::Db9UnaryOpNeg);
    int32_unary.set_operand(test_db9_int32_const_expr(i32::MIN));
    int32_expr.set_unary(int32_unary);
    assert_eq!(
        eval_expr(&int32_expr, &table, &row)
            .unwrap_err()
            .to_string(),
        "integer out of range"
    );

    let mut int64_expr = Db9Expr::default();
    int64_expr.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindInt64));
    let mut int64_unary = kvproto::db9_coprocessor::Db9UnaryExpr::default();
    int64_unary.set_op(Db9UnaryOp::Db9UnaryOpNeg);
    let mut int64_operand = Db9Expr::default();
    int64_operand.set_return_type(test_db9_type(Db9TypeKind::Db9TypeKindInt64));
    let mut constant = kvproto::db9_coprocessor::Db9ConstantExpr::default();
    let mut value = Db9Value::default();
    value.set_int64_value(i64::MIN);
    constant.set_value(value);
    int64_operand.set_constant(constant);
    int64_unary.set_operand(int64_operand);
    int64_expr.set_unary(int64_unary);
    assert_eq!(
        eval_expr(&int64_expr, &table, &row)
            .unwrap_err()
            .to_string(),
        "bigint out of range"
    );
}

#[test]
fn int_float_comparisons_remain_exact_above_f64_precision_limit() {
    let large = 9_007_199_254_740_993_i64;
    let rounded = 9_007_199_254_740_992_f64;

    assert_eq!(
        values_equal(&StoredValue::Int64(large), &StoredValue::Float64(rounded)).unwrap(),
        Some(false)
    );
    assert_eq!(
        compare_values(&StoredValue::Int64(large), &StoredValue::Float64(rounded)).unwrap(),
        Some(Ordering::Greater)
    );
    assert_eq!(
        compare_values(&StoredValue::Float64(rounded), &StoredValue::Int64(large)).unwrap(),
        Some(Ordering::Less)
    );
}

#[test]
fn float_nan_comparisons_match_pg_like_semantics() {
    let nan1 = f64::from_bits(0x7ff8_0000_0000_0001);
    let nan2 = f64::from_bits(0x7ff8_0000_0000_0002);

    assert_eq!(
        values_equal(&StoredValue::Float64(nan1), &StoredValue::Float64(nan2)).unwrap(),
        Some(true)
    );
    assert_eq!(
        compare_values(&StoredValue::Float64(nan1), &StoredValue::Float64(nan2)).unwrap(),
        Some(Ordering::Equal)
    );
    assert_eq!(
        compare_values(&StoredValue::Float64(1.0), &StoredValue::Float64(nan1)).unwrap(),
        Some(Ordering::Less)
    );
    assert_eq!(
        compare_values(&StoredValue::Float64(nan1), &StoredValue::Float64(1.0)).unwrap(),
        Some(Ordering::Greater)
    );
    assert_eq!(
        compare_values(&StoredValue::Int64(1), &StoredValue::Float64(nan1)).unwrap(),
        Some(Ordering::Less)
    );
    assert_eq!(
        compare_values(&StoredValue::Float64(nan1), &StoredValue::Int64(1)).unwrap(),
        Some(Ordering::Greater)
    );
}

pub(in crate::coprocessor::db9) fn numeric_value_as_f64_named(
    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)
        ))),
    }
}

pub(in crate::coprocessor::db9) fn interval_field_i32(
    function_name: &str,
    field_name: &str,
    value: Option<&StoredValue>,
) -> Result<Option<i32>> {
    match value {
        None => Ok(Some(0)),
        Some(value) => integer_value_as_i32(function_name, field_name, value.clone()),
    }
}

pub(in crate::coprocessor::db9) fn checked_add_interval_millis(
    function_name: &str,
    total: i64,
    units: i64,
    unit_millis: i64,
    field_name: &str,
) -> Result<i64> {
    let delta = units.checked_mul(unit_millis).ok_or_else(|| {
        Error::Other(format!(
            "DB9 function '{}' {} overflowed",
            function_name, field_name
        ))
    })?;
    total.checked_add(delta).ok_or_else(|| {
        Error::Other(format!(
            "DB9 function '{}' {} overflowed",
            function_name, field_name
        ))
    })
}

pub(in crate::coprocessor::db9) fn seconds_to_interval_millis(
    function_name: &str,
    seconds: f64,
) -> Result<i64> {
    let millis = (seconds * 1000.0).trunc();
    if millis < i64::MIN as f64 || millis > i64::MAX as f64 {
        return Err(Error::Other(format!(
            "DB9 function '{}' secs overflowed",
            function_name
        )));
    }
    Ok(millis as i64)
}

#[test]
fn string_output_limit_uses_request_scope_and_restores_default() {
    let oversized = DEFAULT_MAX_STRING_OUTPUT_BYTES + 1;
    assert!(check_output_byte_size(oversized).is_err());

    with_db9_string_output_byte_limit(oversized as u64, || check_output_byte_size(oversized))
        .unwrap();

    assert!(check_output_byte_size(oversized).is_err());
}

#[test]
fn string_output_limit_restores_after_unwind() {
    let result = std::panic::catch_unwind(|| {
        let _ = with_db9_string_output_byte_limit(1, || -> Result<()> {
            panic!("intentional string output limit unwind test")
        });
    });

    assert!(result.is_err());
    check_output_byte_size(2).unwrap();
}

#[test]
fn string_output_context_restores_after_unwind() {
    assert!(
        matches!(
            DB9_STRING_OUTPUT_CONTEXT.with(Cell::get),
            Db9StringOutputContext::Other
        ),
        "unexpected initial context"
    );

    let result = std::panic::catch_unwind(|| {
        let _ = with_db9_string_output_byte_limit_for(
            1,
            Db9StringOutputContext::Selection,
            || -> Result<()> { panic!("intentional string output context unwind test") },
        );
    });

    assert!(result.is_err());
    assert!(matches!(
        DB9_STRING_OUTPUT_CONTEXT.with(Cell::get),
        Db9StringOutputContext::Other
    ));
}

pub(in crate::coprocessor::db9) fn with_db9_string_output_byte_limit<T>(
    max_resp_size: u64,
    f: impl FnOnce() -> Result<T>,
) -> Result<T> {
    with_db9_string_output_byte_limit_for(max_resp_size, Db9StringOutputContext::Other, f)
}