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

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

//! Positional row construction for catalog records and operator-join tuples.

use super::TableFunctionRows;
use uqa_core::Value;
use uqa_sql::{semantics::doc_id_value, SQLError};

pub(super) fn singleton_record_rows(
    value: Value,
    default_columns: &[&str],
    column_aliases: &[String],
) -> Result<TableFunctionRows, SQLError> {
    let columns = default_columns
        .iter()
        .enumerate()
        .map(|(index, column)| {
            column_aliases
                .get(index)
                .cloned()
                .unwrap_or_else(|| (*column).into())
        })
        .collect::<Vec<_>>();
    let values = match value {
        Value::Null => vec![Value::Null; columns.len()],
        Value::Record(fields) if fields.len() == columns.len() => {
            fields.into_iter().map(|(_, value)| value).collect()
        }
        Value::Record(fields) => {
            return Err(SQLError::Internal(format!(
                "sequence introspection returned {} fields for {} columns",
                fields.len(),
                columns.len()
            )));
        }
        value => {
            return Err(SQLError::Internal(format!(
                "sequence introspection returned non-record value {value:?}"
            )));
        }
    };
    Ok(TableFunctionRows::materialized(columns, vec![values]))
}

pub(super) fn operator_join_rows(
    tuples: uqa_core::GeneralizedPostingList,
    _alias: Option<&str>,
    column_aliases: &[String],
) -> Result<TableFunctionRows, SQLError> {
    let left_column = column_aliases
        .first()
        .cloned()
        .unwrap_or_else(|| "left_doc_id".into());
    let right_column = column_aliases
        .get(1)
        .cloned()
        .unwrap_or_else(|| "right_doc_id".into());
    let score_column = column_aliases
        .get(2)
        .cloned()
        .unwrap_or_else(|| "_score".into());
    let mut rows = Vec::with_capacity(tuples.len());
    for tuple in tuples.entries() {
        let [left_doc_id, right_doc_id] = tuple.doc_ids.as_slice() else {
            return Err(SQLError::Internal(format!(
                "operator join produced a {}-element tuple; SQL table joins require pairs",
                tuple.doc_ids.len()
            )));
        };
        rows.push(vec![
            doc_id_value(*left_doc_id)?,
            doc_id_value(*right_doc_id)?,
            tuple
                .payload
                .fields
                .get("_score")
                .cloned()
                .unwrap_or(Value::Null),
        ]);
    }
    Ok(TableFunctionRows::materialized(
        vec![left_column, right_column, score_column],
        rows,
    ))
}