indexlake 0.4.0

An experimental table format with extensible index support.
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
113
114
115
116
117
118
119
120
121
122
123

use std::collections::{HashMap, HashSet};

use uuid::Uuid;

use crate::catalog::{RowValidity, TransactionHelper, inline_row_table_name};
use crate::expr::Expr;
use crate::{ILError, ILResult};

impl TransactionHelper {
    pub(crate) async fn update_namespace_name(
        &mut self,
        namespace_id: &Uuid,
        namespace_name: &str,
    ) -> ILResult<usize> {
        self.transaction
            .execute(&format!(
                "UPDATE indexlake_namespace SET namespace_name = {} WHERE namespace_id = {}",
                self.catalog.sql_string_literal(namespace_name),
                self.catalog.sql_uuid_literal(namespace_id),
            ))
            .await
    }

    pub(crate) async fn update_table_name(
        &mut self,
        table_id: &Uuid,
        table_name: &str,
    ) -> ILResult<usize> {
        self.transaction
            .execute(&format!(
                "UPDATE indexlake_table SET table_name = {} WHERE table_id = {}",
                self.catalog.sql_string_literal(table_name),
                self.catalog.sql_uuid_literal(table_id),
            ))
            .await
    }

    pub(crate) async fn update_field_name(
        &mut self,
        field_id: &Uuid,
        field_name: &str,
    ) -> ILResult<usize> {
        self.transaction
            .execute(&format!(
                "UPDATE indexlake_field SET field_name = {} WHERE field_id = {}",
                self.catalog.sql_string_literal(field_name),
                self.catalog.sql_uuid_literal(field_id),
            ))
            .await
    }

    pub(crate) async fn update_inline_rows(
        &mut self,
        table_id: &Uuid,
        set_map: &HashMap<String, Expr>,
        condition: &Expr,
    ) -> ILResult<usize> {
        let mut set_strs = Vec::new();
        for (field_name, new_value) in set_map {
            set_strs.push(format!(
                "{} = {}",
                self.catalog.sql_identifier(field_name),
                self.catalog.unparse_expr(new_value)?,
            ));
        }

        self.transaction
            .execute(&format!(
                "UPDATE {} SET {} WHERE {}",
                inline_row_table_name(table_id),
                set_strs.join(", "),
                self.catalog.unparse_expr(condition)?
            ))
            .await
    }

    pub(crate) async fn update_data_file_validity(
        &mut self,
        data_file_id: &Uuid,
        old_validity: &RowValidity,
        new_validity: &RowValidity,
    ) -> ILResult<usize> {
        let update_count = self.transaction
            .execute(&format!(
                "UPDATE indexlake_data_file SET validity = {} WHERE data_file_id = {} AND validity = {}",
                self.catalog.sql_binary_literal(new_validity.bytes()),
                self.catalog.sql_uuid_literal(data_file_id),
                self.catalog.sql_binary_literal(old_validity.bytes()),
            ))
            .await?;
        if update_count != 1 {
            return Err(ILError::internal(
                "Failed to update data file validity due to concurrency conflict",
            ));
        }
        Ok(update_count)
    }

    pub(crate) async fn update_data_file_rows_as_invalid(
        &mut self,
        data_file_id: &Uuid,
        mut validity: RowValidity,
        sorted_row_ids: &[Uuid],
        invalid_row_ids: &HashSet<Uuid>,
    ) -> ILResult<usize> {
        debug_assert!(sorted_row_ids.is_sorted());

        if invalid_row_ids.is_empty() {
            return Ok(1);
        }

        let old_validity = validity.clone();

        for invalid_row_id in invalid_row_ids {
            if let Ok(idx) = sorted_row_ids.binary_search(invalid_row_id) {
                validity.set(idx, false);
            }
        }

        self.update_data_file_validity(data_file_id, &old_validity, &validity)
            .await
    }
}