uqa-execution 0.4.6

Volcano physical operators with row-batch pipelines
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

use super::*;
use std::sync::Arc;

fn fixture() -> (TableSchema, TableConstraintSet, KeyConstraintNames) {
    let key: TableKeyConstraint = serde_json::from_value(serde_json::json!({
        "name": "original", "columns": ["v"], "kind": "Unique",
        "catalog_identity": {"object_id": vec![3; 16], "oid": 16384}
    }))
    .unwrap();
    let constraints = TableConstraintSet {
        key_constraints: vec![key],
        ..Default::default()
    };
    let schema = TableSchema {
        relation: RelationIdentity::new("public", "t"),
        security: uqa_storage::RelationSecurityRow::legacy("uqa"),
        object_id: [1; 16],
        storage_generation: [2; 16],
        analyzer_json: "{}".into(),
        fts_fields: Vec::new(),
        vector_fields: Vec::new(),
        columns_json: "[]".into(),
        constraints_json: encode(&constraints).unwrap(),
    };
    let names = KeyConstraintNames {
        names: Some(BTreeMap::from([(
            [3; 16],
            KeyName {
                table: schema.relation.clone(),
                table_object_id: schema.object_id,
                name: "current".into(),
            },
        )])),
    };
    (schema, constraints, names)
}

#[test]
fn structural_storage_rehydrates_the_current_owned_name_without_changing_identity() {
    let (schema, mut expected, names) = fixture();
    let stored: TableConstraintSet = serde_json::from_str(&schema.constraints_json).unwrap();
    assert!(stored.key_constraints[0].name.is_none());
    expected.key_constraints[0].name = Some("current".into());
    assert_eq!(
        serde_json::to_value(names.decode(&schema).unwrap()).unwrap(),
        serde_json::to_value(expected).unwrap(),
    );
}

#[test]
fn missing_recreated_or_redundant_name_owners_are_rejected() {
    let (schema, constraints, mut names) = fixture();
    let mut recreated = schema.clone();
    recreated.object_id = [4; 16];
    assert!(names.decode(&recreated).is_err());
    let mut duplicated = schema.clone();
    duplicated.constraints_json = serde_json::to_string(&constraints).unwrap();
    assert!(names.decode(&duplicated).is_err());
    names.names.as_mut().unwrap().clear();
    assert!(names.decode(&schema).is_err());
}

#[test]
fn a_legacy_declaration_keeps_its_name_until_registry_conversion() {
    let (mut schema, constraints, _) = fixture();
    let names = KeyConstraintNames { names: None };
    schema.constraints_json = names.encode(&constraints).unwrap();
    assert_eq!(
        serde_json::to_value(names.decode(&schema).unwrap()).unwrap(),
        serde_json::to_value(constraints).unwrap(),
    );
}

#[test]
fn candidate_names_follow_catalog_identity_without_replacing_key_structure() {
    let (_, mut candidate, _) = fixture();
    let mut current_key = candidate.key_constraints[0].clone();
    current_key.name = Some("renamed".into());
    candidate.hierarchy.partition_inherited_key_constraints = candidate.key_constraints.clone();
    let mut added = candidate.key_constraints[0].clone();
    added.catalog_identity.as_mut().unwrap().object_id = [4; 16];
    added.name = Some("new_key".into());
    candidate.key_constraints.push(added.clone());
    candidate.key_constraints[0].columns = vec!["new_column".into()];
    let current = crate::catalog::CatalogTableSnapshot {
        object_id: [1; 16],
        security: std::sync::Arc::new(crate::catalog::security::BoundTableSecurity::owner(
            uqa_sql::catalog::roles::RoleIdentity::BOOTSTRAP,
        )),
        columns: Arc::default(),
        columns_declared: true,
        checks: Arc::default(),
        foreign_keys: Arc::default(),
        keys: vec![current_key].into(),
        hierarchy: Arc::default(),
        persistence: uqa_sql::ast::RelationPersistence::Permanent,
    };
    rebind_current_key_names(
        candidate.key_constraints.iter_mut().chain(
            candidate
                .hierarchy
                .partition_inherited_key_constraints
                .iter_mut(),
        ),
        &current,
    );
    assert_eq!(
        candidate.key_constraints[0].name.as_deref(),
        Some("renamed")
    );
    assert_eq!(candidate.key_constraints[0].columns, ["new_column"]);
    assert_eq!(candidate.key_constraints[1], added);
    assert_eq!(
        candidate.hierarchy.partition_inherited_key_constraints[0]
            .name
            .as_deref(),
        Some("renamed")
    );
}