uqa-execution 0.3.0

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

//
// 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 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 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
    }
}