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

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

//! Read generation and writable services for mutation command orchestration.
use super::MutationExecutionContext;
use crate::mutation::insert::source::InsertSourceContext;
use crate::query::statement::context::{QueryContext, SnapshotSource};
use uqa_sql::SQLError;

pub trait MutationSnapshots<S: Clone + 'static>: SnapshotSource<S> {
    fn with_snapshot(
        &self,
        snapshot: &S,
        operation: &mut dyn ScopedMutationOperation<S>,
    ) -> Result<(), SQLError>;
}
pub trait ScopedMutationOperation<S: Clone + 'static> {
    fn run(&mut self, context: &MutationStatementContext<'_, S>) -> Result<(), SQLError>;
}
#[derive(Clone)]
pub struct MutationStatementContext<'a, S: Clone + 'static> {
    pub query: QueryContext<'a, S>,
    pub mutation: MutationExecutionContext<'a, S>,
    pub snapshots: &'a dyn MutationSnapshots<S>,
}
impl<S: Clone + 'static> Copy for MutationStatementContext<'_, S> {}

impl<'a, S: Clone + 'static> MutationStatementContext<'a, S> {
    pub fn insert_source(&self) -> InsertSourceContext<'a, S> {
        InsertSourceContext {
            rows: self.mutation.preparation,
            identities: self.mutation.identities,
            runtime: self.query.source.relational.runtime,
        }
    }
}

pub fn with_mutation_snapshot<S: Clone + 'static, T>(
    snapshots: &dyn MutationSnapshots<S>,
    snapshot: &S,
    action: impl for<'scope> FnOnce(&MutationStatementContext<'scope, S>) -> Result<T, SQLError>,
) -> Result<T, SQLError> {
    struct Operation<F, T> {
        action: Option<F>,
        result: Option<T>,
    }
    impl<S: Clone + 'static, F, T> ScopedMutationOperation<S> for Operation<F, T>
    where
        F: for<'scope> FnOnce(&MutationStatementContext<'scope, S>) -> Result<T, SQLError>,
    {
        fn run(&mut self, context: &MutationStatementContext<'_, S>) -> Result<(), SQLError> {
            let action = self.action.take().ok_or_else(|| {
                SQLError::Internal("statement snapshot operation was invoked more than once".into())
            })?;
            self.result = Some(action(context)?);
            Ok(())
        }
    }
    let mut operation = Operation {
        action: Some(action),
        result: None,
    };
    snapshots.with_snapshot(snapshot, &mut operation)?;
    operation
        .result
        .ok_or_else(|| SQLError::Internal("statement snapshot operation was not invoked".into()))
}