uqa-engine 0.1.11

Engine: schema-aware table store, catalog restore, transactions
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

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

//! Physical tuple-successor tracking for rows that move between partitions.

use super::{
    epoch_is_after, lock_strengths_conflict, table_hash, CommittedRowChange,
    CommittedRowChangeKind, CrossAttachment, DocId, LockStrength, RowChangeBaseline, RowLockKey,
    RowLockManager, SQLError,
};

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum PhysicalRowChangeTarget {
    Unchanged,
    Present { table_hash: u64, doc_id: DocId },
    Deleted,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum LocalPhysicalRowChangeTarget {
    Unchanged,
    Present(RowLockKey),
    Deleted,
}

impl RowLockManager {
    /// Follow committed rewrites while retaining the physical relation as well as the document id. Partition movement changes both halves of the tuple identity, so DML rechecks must not collapse the successor back onto the source leaf.
    pub(crate) fn physical_row_successor_after(
        &self,
        table: &str,
        doc_id: DocId,
        baseline: RowChangeBaseline,
    ) -> Result<PhysicalRowChangeTarget, SQLError> {
        let key = RowLockKey {
            table: self.table_key(table),
            doc_id,
        };
        if let Some(CrossAttachment::Active(coordinator)) = self.cross.as_ref() {
            return coordinator
                .physical_change_target_after(
                    table_hash(table.as_bytes()),
                    doc_id,
                    baseline.cross_sequence,
                    LockStrength::ForUpdate,
                )
                .map_err(SQLError::Internal);
        }
        Ok(
            match resolve_local_physical_change_target(
                &self.state.lock().changes,
                key,
                baseline.epoch,
                LockStrength::ForUpdate,
            ) {
                LocalPhysicalRowChangeTarget::Unchanged => PhysicalRowChangeTarget::Unchanged,
                LocalPhysicalRowChangeTarget::Deleted => PhysicalRowChangeTarget::Deleted,
                LocalPhysicalRowChangeTarget::Present(target) => PhysicalRowChangeTarget::Present {
                    table_hash: table_hash(&self.relation_bytes(target.table)),
                    doc_id: target.doc_id,
                },
            },
        )
    }
}

pub(super) fn resolve_local_physical_change_target(
    changes: &[CommittedRowChange],
    key: RowLockKey,
    baseline: u64,
    wanted: LockStrength,
) -> LocalPhysicalRowChangeTarget {
    let mut current = key;
    let mut requires_recheck = false;
    for change in changes {
        if !epoch_is_after(change.epoch, baseline) || change.key != current {
            continue;
        }
        match change.kind {
            CommittedRowChangeKind::Update => {
                requires_recheck |= lock_strengths_conflict(change.strength, wanted);
            }
            CommittedRowChangeKind::Delete => {
                if lock_strengths_conflict(change.strength, wanted) {
                    return LocalPhysicalRowChangeTarget::Deleted;
                }
            }
            CommittedRowChangeKind::Rewrite(successor) => {
                if lock_strengths_conflict(change.strength, wanted) {
                    requires_recheck = true;
                    current = successor;
                }
            }
        }
    }
    if requires_recheck {
        LocalPhysicalRowChangeTarget::Present(current)
    } else {
        LocalPhysicalRowChangeTarget::Unchanged
    }
}