kip-sql 0.0.1-alpha.8

build the SQL layer of KipDB database
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

use crate::execution::codegen::CodeGenerator;
use crate::execution::volcano::dql::aggregate::hash_agg::HashAggStatus;
use crate::execution::ExecutorError;
use crate::expression::ScalarExpression;
use crate::impl_from_lua;
use crate::planner::operator::aggregate::AggregateOperator;
use crate::types::tuple::Tuple;
use mlua::prelude::{LuaResult, LuaValue};
use mlua::{FromLua, Lua, UserData, UserDataMethods, Value};
use std::mem;

pub struct HashAgg {
    id: i64,
    agg_calls: Option<Vec<ScalarExpression>>,
    groupby_exprs: Option<Vec<ScalarExpression>>,
    is_produced: bool,
}

impl From<(AggregateOperator, i64)> for HashAgg {
    fn from(
        (
            AggregateOperator {
                agg_calls,
                groupby_exprs,
                ..
            },
            id,
        ): (AggregateOperator, i64),
    ) -> Self {
        HashAgg {
            id,
            agg_calls: Some(agg_calls),
            groupby_exprs: Some(groupby_exprs),
            is_produced: false,
        }
    }
}

impl UserData for HashAggStatus {
    fn add_methods<'lua, M: UserDataMethods<'lua, Self>>(methods: &mut M) {
        methods.add_method_mut("update", |_, agg_status, tuple: Tuple| {
            agg_status.update(tuple).unwrap();

            Ok(())
        });
        methods.add_method_mut("to_tuples", |_, agg_status, ()| {
            Ok(agg_status.to_tuples().unwrap())
        });
    }
}

impl_from_lua!(HashAggStatus);

impl CodeGenerator for HashAgg {
    fn produce(&mut self, lua: &Lua, script: &mut String) -> Result<(), ExecutorError> {
        if let (Some(agg_calls), Some(groupby_exprs)) =
            (self.agg_calls.take(), self.groupby_exprs.take())
        {
            let env = format!("hash_agg_{}", self.id);
            lua.globals()
                .set(env.as_str(), HashAggStatus::new(agg_calls, groupby_exprs))?;

            script.push_str(
                format!(
                    r#"
            for _, tuple in ipairs(results) do
                {}:update(tuple)
            end

            results = {{}}

            for index, tuple in ipairs({}:to_tuples()) do
                index = index - 1
            "#,
                    env, env
                )
                .as_str(),
            );

            self.is_produced = true;
        }

        Ok(())
    }

    fn consume(&mut self, _: &Lua, script: &mut String) -> Result<(), ExecutorError> {
        if mem::replace(&mut self.is_produced, false) {
            script.push_str(
                r#"
                table.insert(results, tuple)
                ::continue::
            end
            "#,
            );
        }

        Ok(())
    }
}