linkrs-query 0.1.0

Query engine, parser, and executor for Linkrs
Documentation 
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

use crate::planning::plan::factorization::{FactorizationError, FactorizedSchema};

use crate::planning::plan::logical::logical_nodes::wco_intersect::LogicalWcoIntersectNode;

pub(super) fn union_minus(
    child_schemas: &[FactorizedSchema],
) -> Result<FactorizedSchema, FactorizationError> {
    // Set-operation union keeps both sides' expression ids in scope (each
    // side has distinct ids for the same output columns), so unlike
    // Ladybug's first-child-only sink rebuild this merges both children and
    // then flattens: dropping either side's ids would break downstream
    // scope resolution.
    if child_schemas.len() >= 2 {
        let left = &child_schemas[0];
        let right = &child_schemas[1];
        let mut merged = left.clone();
        let mapping = merged.merge_groups_from(right);
        for (expr_id, gpos) in right.expression_to_group_iter() {
            let new_pos = mapping.get(gpos).copied().unwrap_or(*gpos);
            merged.insert_to_scope_may_repeat(expr_id.clone(), new_pos)?;
        }
        merged.flatten_all()?;
        merged.validate_at_most_one_unflat()?;
        Ok(merged)
    } else {
        Ok(child_schemas.first().cloned().unwrap_or_default())
    }
}

pub(super) fn intersect(
    child_schemas: &[FactorizedSchema],
) -> Result<FactorizedSchema, FactorizationError> {
    if child_schemas.len() > 2 {
        let mut schema = child_schemas[0].clone();
        if schema.has_unflat_group() {
            schema.flatten_all()?;
        }
        let out_pos = schema.create_group();
        for build_schema in &child_schemas[1..] {
            for expr in build_schema.expressions_in_scope() {
                if !schema.is_expression_in_scope(expr) {
                    schema.insert_to_group_and_scope(expr.clone(), out_pos)?;
                }
            }
        }
        schema.validate_at_most_one_unflat()?;
        Ok(schema)
    } else if child_schemas.len() >= 2 {
        let left = &child_schemas[0];
        let right = &child_schemas[1];
        let mut merged = left.clone();
        let mapping = merged.merge_groups_from(right);
        for (expr_id, gpos) in right.expression_to_group_iter() {
            let new_pos = mapping.get(gpos).copied().unwrap_or(*gpos);
            merged.insert_to_scope_may_repeat(expr_id.clone(), new_pos)?;
        }
        merged.flatten_all()?;
        merged.validate_at_most_one_unflat()?;
        Ok(merged)
    } else {
        Ok(child_schemas.first().cloned().unwrap_or_default())
    }
}

pub(super) fn wco_intersect(
    n: &LogicalWcoIntersectNode,
    child_schemas: &[FactorizedSchema],
) -> Result<FactorizedSchema, FactorizationError> {
    let mut schema = child_schemas.first().cloned().unwrap_or_default();
    if schema.has_unflat_group() {
        schema.flatten_all()?;
    }
    let out_pos = schema.create_group();
    let intersect_id = n.intersect_key().id().clone();
    if schema.is_expression_in_scope(&intersect_id) {
        if let Some(pos) = schema.get_group_pos(&intersect_id) {
            schema.flatten_group(pos)?;
        }
    } else if let Some(name) = n.intersect_key().as_variable() {
        schema.insert_to_group_and_scope_with_name(intersect_id.clone(), Some(name), out_pos)?;
    } else {
        schema.insert_to_group_and_scope(intersect_id.clone(), out_pos)?;
    }
    for (build_idx, build_schema) in child_schemas.iter().skip(1).enumerate() {
        let bound_id = n.bound_keys().get(build_idx).map(|k| k.id().clone());
        for expr in build_schema.expressions_in_scope() {
            if *expr == intersect_id {
                continue;
            }
            if let Some(bound) = &bound_id {
                if expr == bound {
                    continue;
                }
            }
            if !schema.is_expression_in_scope(expr) {
                schema.insert_to_group_and_scope(expr.clone(), out_pos)?;
            }
        }
        for (name, _) in build_schema.expression_name_to_group_iter() {
            if let Some(var) = n.intersect_key().as_variable() {
                if name.as_str() == var.as_str() {
                    continue;
                }
            }
            if schema.get_group_pos_by_name_opt(name).is_none() {
                schema.insert_name_for_group(name.clone(), out_pos)?;
            }
        }
    }
    schema.validate_at_most_one_unflat()?;
    Ok(schema)
}