uqa-execution 0.4.5

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
126
127
128
129
130
131
132
133
134

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

//! Rebuild admission retains the exact census source and its original temporary allowance.

use uqa_storage::{
    diskann_index::{
        build::DiskANNTemporaryBudget,
        format::DiskANNGeneration,
        maintenance::{
            DiskANNChangeStatistics, DiskANNMaintenanceSource, DiskANNStatisticsCursor,
            DiskANNStatisticsRequest,
        },
        DiskANNIndexOptions,
    },
    read_control::StorageReadControl,
    StorageBackendError, StorageBackendResult,
};

/// Soft triggers for one index's current uncovered documents or logical raw vector bytes. Neither limit is a write quota or a physical-retention limit.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DiskANNRebuildPolicy {
    documents: u64,
    vector_bytes: u64,
}

impl Default for DiskANNRebuildPolicy {
    fn default() -> Self {
        Self {
            documents: 1024,
            vector_bytes: 64 << 20,
        }
    }
}

impl DiskANNRebuildPolicy {
    pub fn new(documents: u64, vector_bytes: u64) -> StorageBackendResult<Self> {
        if documents == 0 || vector_bytes == 0 {
            return Err(StorageBackendError::Other(
                "DiskANN rebuild thresholds must be positive".into(),
            ));
        }
        Ok(Self {
            documents,
            vector_bytes,
        })
    }

    pub fn documents(self) -> u64 {
        self.documents
    }
    pub fn vector_bytes(self) -> u64 {
        self.vector_bytes
    }

    pub(super) fn admits(self, changes: DiskANNChangeStatistics) -> bool {
        changes.documents >= self.documents || changes.vector_bytes >= self.vector_bytes
    }
}

/// Exact counts for the most recently completed index capture; later commits may already have changed its live corpus.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DiskANNMaintenanceCensus {
    pub generation: DiskANNGeneration,
    pub changes: DiskANNChangeStatistics,
}

#[derive(Clone)]
pub(super) struct Configuration {
    pub policy: DiskANNRebuildPolicy,
    pub temporary: DiskANNTemporaryBudget,
}

pub(super) struct Capture {
    pub source: Box<dyn DiskANNMaintenanceSource>,
    pub options: DiskANNIndexOptions,
    pub configuration: Configuration,
    pub ready: bool,
    after: Option<DiskANNStatisticsCursor>,
    generation: Option<DiskANNGeneration>,
    total: DiskANNChangeStatistics,
}

impl Capture {
    pub fn new(
        source: Box<dyn DiskANNMaintenanceSource>,
        options: DiskANNIndexOptions,
        configuration: Configuration,
    ) -> Self {
        Self {
            source,
            options,
            configuration,
            ready: false,
            after: None,
            generation: None,
            total: DiskANNChangeStatistics::default(),
        }
    }

    pub fn step(
        &mut self,
        control: &StorageReadControl,
    ) -> StorageBackendResult<Option<DiskANNMaintenanceCensus>> {
        let page = self.source.statistics(
            DiskANNStatisticsRequest {
                after: self.after,
                max_records: 64,
            },
            control,
        )?;
        if page.examined > 64
            || (page.next.is_some() && (page.examined == 0 || page.next == self.after))
            || self
                .generation
                .is_some_and(|generation| generation != page.generation)
        {
            return Err(StorageBackendError::Other(
                "invalid DiskANN maintenance census page".into(),
            ));
        }
        control.check()?;
        self.total = self.total.checked_add(page.outstanding)?;
        self.generation = Some(page.generation);
        self.after = page.next;
        Ok(page.next.is_none().then_some(DiskANNMaintenanceCensus {
            generation: page.generation,
            changes: self.total,
        }))
    }
}