27895806754

Committed 21 Jun 2026 06:20AM UTC coverage: 98.637% (-0.07%) from 98.711%

Build # 27895806754

Build Type

push

github

Committed by

web-flow

Commit Message

Convert core and native storages to sync execution (#1928)

GlueSQL's async support has mostly been driven by browser-facing storage constraints such as IndexedDB, rather than by the core execution model itself.

That async boundary ended up spreading through the planner, executor, storage traits, CLI, examples, tests, and native storage implementations. For Rust users and native storages, this added futures/stream plumbing and async dependencies even when the underlying work was synchronous or backed by synchronous APIs.

With the JavaScript and Python packages being split out of this workspace, this PR makes the Rust core synchronous again and keeps browser/package-specific async concerns out of the core storage contract.

Coverage Stats

2100 of 2147 new or added lines in 141 files covered. (97.81%)

18 existing lines in 9 files now uncovered.

44000 of 44608 relevant lines covered (98.64%)

77908.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.18

/core/src/executor/fetch.rs

use {
    super::{context::RowContext, evaluate::evaluate_stateless, filter::check_expr},
    crate::{
        ast::{ColumnDef, ColumnUniqueOption, Dictionary, ToSql, ToSqlUnquoted},
        data::{Key, Row, SCHEMALESS_DOC_COLUMN, Value},
        executor::{evaluate::evaluate, select::select},
        plan::{
            ExprPlan, IndexItemPlan, JoinPlan, ProjectionPlan, QueryPlan, SelectItemPlan,
            SetExprPlan, TableAliasPlan, TableFactorPlan, TableWithJoinsPlan, ValuesPlan,
        },
        result::Result,
        store::GStore,
    },
    serde::Serialize,
    std::{borrow::Cow, collections::BTreeMap, fmt::Debug, iter, rc::Rc},
    thiserror::Error as ThisError,
};

pub type KeyedRows<'a> = Box<dyn Iterator<Item = Result<(Key, Row)>> + 'a>;
pub type RelationRows<'a> = Box<dyn Iterator<Item = Result<Row>> + 'a>;

#[derive(ThisError, Serialize, Debug, PartialEq, Eq)]
pub enum FetchError {
    #[error("table not found: {0}")]
    TableNotFound(String),

    #[error("table alias not found: {0}")]
    TableAliasNotFound(String),

    #[error("SERIES has wrong size: {0}")]
    SeriesSizeWrong(i64),

    #[error("table '{0}' has {1} columns available but {2} column aliases specified")]
    TooManyColumnAliases(String, usize, usize),

    #[error("unreachable")]
    Unreachable,
}

pub fn fetch<'a, T: GStore>(
    storage: &'a T,
    table_name: &'a str,
    columns: Rc<[String]>,
    where_clause: Option<&'a ExprPlan>,
) -> Result<KeyedRows<'a>> {
    let rows = storage.scan_data(table_name)?.filter_map(move |row| {
        let (key, values) = match row {
            Ok(row) => row,
            Err(error) => return Some(Err(error)),
        };
        let row = Row {
            columns: Rc::clone(&columns),
            values,
        };

        match where_clause {
            Some(expr) => {
                let context = RowContext::new(table_name, Cow::Borrowed(&row), None);
                let context = Rc::new(context);
                match check_expr(storage, Some(&context), None, expr) {
                    Ok(true) => Some(Ok((key, row))),
                    Ok(false) => None,
                    Err(error) => Some(Err(error)),
                }
            }
            None => Some(Ok((key, row))),
        }
    });

    Ok(Box::new(rows))
}

pub fn fetch_relation_rows<'a, T: GStore>(
    storage: &'a T,
    table_factor: &'a TableFactorPlan,
    filter_context: Option<&Rc<RowContext<'a>>>,
) -> Result<RelationRows<'a>> {
    let columns = Rc::from(fetch_relation_columns(storage, table_factor)?);

    match table_factor {
        TableFactorPlan::Derived { subquery, .. } => {
            let filter_context = filter_context.map(Rc::clone);
            let rows = select(storage, subquery, filter_context)?.map(move |row| {
                let row = row?;
                Ok(Row {
                    columns: Rc::clone(&columns),
                    values: row.values,
                })
            });

            Ok(Box::new(rows))
        }
        TableFactorPlan::Table { name, .. } => {
            let rows = match table_factor.index() {
                Some(IndexItemPlan::NonClustered {
                    name: index_name,
                    asc,
                    cmp_expr,
                }) => {
                    let cmp_value = match cmp_expr {
                        Some((op, expr)) => {
                            let evaluated = evaluate(storage, None, None, expr)?;

                            Some((op, evaluated.try_into()?))
                        }
                        None => None,
                    };

                    let rows = storage
                        .scan_indexed_data(name, index_name, *asc, cmp_value)?
                        .map(move |row| {
                            let (_, values) = row?;
                            Ok(Row {
                                columns: Rc::clone(&columns),
                                values,
                            })
                        });
                    Box::new(rows) as Box<dyn Iterator<Item = Result<Row>> + 'a>
                }
                Some(IndexItemPlan::PrimaryKey(expr)) => {
                    let schema = storage.fetch_schema(name)?.ok_or(FetchError::Unreachable)?;

                    let evaluated = evaluate(storage, filter_context, None, expr)?;

                    let column_def = schema
                        .column_defs
                        .as_ref()
                        .and_then(|column_defs| {
                            column_defs
                                .iter()
                                .find(|column_def| column_def.unique.is_some_and(|u| u.is_primary))
                        })
                        .ok_or(FetchError::Unreachable)?;

                    let value =
                        evaluated.try_into_value(&column_def.data_type, column_def.nullable)?;
                    let key = Key::try_from(value)?;

                    match storage.fetch_data(name, &key)? {
                        Some(values) => Box::new(iter::once(Ok(Row {
                            columns: Rc::clone(&columns),
                            values,
                        })))
                            as Box<dyn Iterator<Item = Result<Row>> + 'a>,
                        None => {
                            Box::new(iter::empty()) as Box<dyn Iterator<Item = Result<Row>> + 'a>
                        }
                    }
                }
                _ => {
                    let rows = storage.scan_data(name)?.map(move |row| {
                        let (_, values) = row?;
                        Ok(Row {
                            columns: Rc::clone(&columns),
                            values,
                        })
                    });
                    Box::new(rows) as Box<dyn Iterator<Item = Result<Row>> + 'a>
                }
            };

            Ok(rows)
        }
        TableFactorPlan::Series { size, .. } => {
            let value: Value = evaluate_stateless(None, size)?.try_into()?;
            let size: i64 = value.try_into()?;
            let size = match size {
                n if n >= 0 => size,
                n => return Err(FetchError::SeriesSizeWrong(n).into()),
            };

            let columns = Rc::from(vec!["N".to_owned()]);
            let rows = (1..=size).map(move |v| {
                Ok(Row {
                    columns: Rc::clone(&columns),
                    values: vec![Value::I64(v)],
                })
            });

            Ok(Box::new(rows))
        }
        TableFactorPlan::Dictionary { dict, .. } => {
            let rows = match dict {
                Dictionary::GlueObjects => {
                    let schemas = storage.fetch_all_schemas()?;
                    let table_metas = storage
                        .scan_table_meta()?
                        .collect::<Result<BTreeMap<_, _>>>()?;
                    let rows = schemas.into_iter().flat_map(move |schema| {
                        let meta = table_metas
                            .iter()
                            .find_map(|(table_name, hash_map)| {
                                (table_name == &schema.table_name).then(|| hash_map.clone())
                            })
                            .unwrap_or_default();

                        let table_rows = BTreeMap::from([
                            ("OBJECT_NAME".to_owned(), Value::Str(schema.table_name)),
                            ("OBJECT_TYPE".to_owned(), Value::Str("TABLE".to_owned())),
                        ])
                        .into_iter()
                        .chain(meta)
                        .collect::<BTreeMap<_, _>>();

                        let index_rows = schema.indexes.into_iter().map(|index| {
                            BTreeMap::from([
                                ("OBJECT_NAME".to_owned(), Value::Str(index.name)),
                                ("OBJECT_TYPE".to_owned(), Value::Str("INDEX".to_owned())),
                            ])
                        });

                        iter::once(table_rows).chain(index_rows).map(|hash_map| {
                            let (columns, values): (Vec<_>, Vec<_>) = hash_map.into_iter().unzip();
                            Row {
                                columns: columns.into(),
                                values,
                            }
                        })
                    });

                    Box::new(rows.map(Ok)) as Box<dyn Iterator<Item = Result<Row>> + 'a>
                }
                Dictionary::GlueTables => {
                    let schemas = storage.fetch_all_schemas()?;
                    let rows = schemas.into_iter().map(move |schema| Row {
                        columns: Rc::clone(&columns),
                        values: vec![
                            Value::Str(schema.table_name),
                            schema.comment.map_or(Value::Null, Value::Str),
                        ],
                    });

                    Box::new(rows.map(Ok))
                }
                Dictionary::GlueTableColumns => {
                    let schemas = storage.fetch_all_schemas()?;
                    let rows = schemas.into_iter().flat_map(move |schema| {
                        let columns = Rc::clone(&columns);
                        let table_name = schema.table_name;

                        schema
                            .column_defs
                            .unwrap_or_default()
                            .into_iter()
                            .enumerate()
                            .map(move |(index, column_def)| {
                                let values = vec![
                                    Value::Str(table_name.clone()),
                                    Value::Str(column_def.name),
                                    Value::I64(index as i64 + 1),
                                    Value::Bool(column_def.nullable),
                                    column_def
                                        .unique
                                        .map_or(Value::Null, |unique| Value::Str(unique.to_sql())),
                                    column_def
                                        .default
                                        .map_or(Value::Null, |expr| Value::Str(expr.to_sql())),
                                    column_def.comment.map_or(Value::Null, Value::Str),
                                ];

                                Row {
                                    columns: Rc::clone(&columns),
                                    values,
                                }
                            })
                    });

                    Box::new(rows.map(Ok))
                }
                Dictionary::GlueIndexes => {
                    let schemas = storage.fetch_all_schemas()?;
                    let rows = schemas.into_iter().flat_map(move |schema| {
                        let column_defs = schema.column_defs.unwrap_or_default();
                        let primary_column = column_defs.iter().find_map(|column_def| {
                            let ColumnDef { name, unique, .. } = column_def;

                            (unique == &Some(ColumnUniqueOption { is_primary: true }))
                                .then_some(name)
                        });

                        let clustered = match primary_column {
                            Some(column_name) => {
                                let values = vec![
                                    Value::Str(schema.table_name.clone()),
                                    Value::Str("PRIMARY".to_owned()),
                                    Value::Str("BOTH".to_owned()),
                                    Value::Str(column_name.to_owned()),
                                    Value::Bool(true),
                                ];

                                let row = Row {
                                    columns: Rc::clone(&columns),
                                    values,
                                };

                                vec![row]
                            }
                            None => Vec::new(),
                        };

                        let columns = Rc::clone(&columns);
                        let non_clustered = schema.indexes.into_iter().map(move |index| {
                            let values = vec![
                                Value::Str(schema.table_name.clone()),
                                Value::Str(index.name),
                                Value::Str(index.order.to_string()),
                                Value::Str(index.expr.to_sql_unquoted()),
                                Value::Bool(false),
                            ];

                            Row {
                                columns: Rc::clone(&columns),
                                values,
                            }
                        });

                        clustered.into_iter().chain(non_clustered)
                    });

                    Box::new(rows.map(Ok))
                }
            };

            Ok(rows)
        }
    }
}

pub fn fetch_columns<T: GStore>(storage: &T, table_name: &str) -> Result<Vec<String>> {
    let columns = storage
        .fetch_schema(table_name)?
        .ok_or_else(|| FetchError::TableNotFound(table_name.to_owned()))?
        .column_defs
        .map_or_else(
            || vec![SCHEMALESS_DOC_COLUMN.to_owned()],
            |column_defs| {
                column_defs
                    .into_iter()
                    .map(|column_def| column_def.name)
                    .collect()
            },
        );

    Ok(columns)
}

pub fn fetch_relation_columns<T>(storage: &T, table_factor: &TableFactorPlan) -> Result<Vec<String>>
where
    T: GStore,
{
    match table_factor {
        TableFactorPlan::Table { name, alias, .. } => {
            let columns = fetch_columns(storage, name)?;
            match alias {
                None => Ok(columns),
                Some(alias) if alias.columns.len() > columns.len() => {
                    Err(FetchError::TooManyColumnAliases(
                        name.clone(),
                        columns.len(),
                        alias.columns.len(),
                    )
                    .into())
                }
                Some(alias) => Ok(alias
                    .columns
                    .iter()
                    .cloned()
                    .chain(columns[alias.columns.len()..columns.len()].to_vec())
                    .collect()),
            }
        }
        TableFactorPlan::Series { .. } => Ok(vec!["N".to_owned()]),
        TableFactorPlan::Dictionary { dict, .. } => Ok(match dict {
            Dictionary::GlueObjects => vec![
                "OBJECT_NAME".to_owned(),
                "OBJECT_TYPE".to_owned(),
                "CREATED".to_owned(),
            ],
            Dictionary::GlueTables => vec!["TABLE_NAME".to_owned(), "COMMENT".to_owned()],
            Dictionary::GlueTableColumns => vec![
                "TABLE_NAME".to_owned(),
                "COLUMN_NAME".to_owned(),
                "COLUMN_ID".to_owned(),
                "NULLABLE".to_owned(),
                "KEY".to_owned(),
                "DEFAULT".to_owned(),
                "COMMENT".to_owned(),
            ],
            Dictionary::GlueIndexes => vec![
                "TABLE_NAME".to_owned(),
                "INDEX_NAME".to_owned(),
                "ORDER".to_owned(),
                "EXPRESSION".to_owned(),
                "UNIQUENESS".to_owned(),
            ],
        }),
        TableFactorPlan::Derived {
            subquery: QueryPlan { body, .. },
            alias:
                TableAliasPlan {
                    columns: alias_columns,
                    name,
                },
        } => match body {
            SetExprPlan::Select(statement) => {
                let crate::plan::SelectPlan {
                    from:
                        TableWithJoinsPlan {
                            relation, joins, ..
                        },
                    projection,
                    ..
                } = statement.as_ref();

                let labels = fetch_labels(storage, relation, joins, projection)?;
                if alias_columns.is_empty() {
                    Ok(labels)
                } else if alias_columns.len() > labels.len() {
                    Err(FetchError::TooManyColumnAliases(
                        name.clone(),
                        labels.len(),
                        alias_columns.len(),
                    )
                    .into())
                } else {
                    Ok(alias_columns
                        .iter()
                        .cloned()
                        .chain(labels[alias_columns.len()..labels.len()].to_vec())
                        .collect())
                }
            }
            SetExprPlan::Values(ValuesPlan(values_list)) => {
                let total_len = values_list[0].len();
                let alias_len = alias_columns.len();
                if alias_len > total_len {
                    return Err(FetchError::TooManyColumnAliases(
                        name.into(),
                        total_len,
                        alias_len,
                    )
                    .into());
                }
                let labels = (alias_len + 1..=total_len).map(|i| format!("column{i}"));
                let labels = alias_columns
                    .iter()
                    .cloned()
                    .chain(labels)
                    .collect::<Vec<_>>();

                Ok(labels)
            }
        },
    }
}

fn fetch_join_columns<'a, T: GStore>(
    storage: &T,
    joins: &'a [JoinPlan],
) -> Result<Vec<(&'a str, Vec<String>)>> {
    let mut all_columns = Vec::with_capacity(joins.len());
    for join in joins {
        let columns = fetch_relation_columns(storage, &join.relation)?;
        let alias = join.relation.alias_name();
        all_columns.push((alias, columns));
    }
    Ok(all_columns)
}

pub fn fetch_labels<T: GStore>(
    storage: &T,
    relation: &TableFactorPlan,
    joins: &[JoinPlan],
    projection: &ProjectionPlan,
) -> Result<Vec<String>> {
    let table_alias = relation.alias_name();
    let columns = fetch_relation_columns(storage, relation)?;
    let join_columns = fetch_join_columns(storage, joins)?;

    match projection {
        ProjectionPlan::SchemalessMap => Ok(vec![SCHEMALESS_DOC_COLUMN.to_owned()]),
        ProjectionPlan::SelectItems(projection) => projection
            .iter()
            .flat_map(|item| match item {
                SelectItemPlan::Wildcard => {
                    let columns = columns.iter().cloned();
                    let join_columns = join_columns.iter().flat_map(|(_, columns)| columns.clone());

                    columns.chain(join_columns).map(Ok).collect()
                }
                SelectItemPlan::QualifiedWildcard(target_table_alias) => {
                    if table_alias == target_table_alias {
                        return columns.iter().cloned().map(Ok).collect();
                    }

                    let labels = join_columns
                        .iter()
                        .find(|(table_alias, _)| table_alias == target_table_alias)
                        .map(|(_, columns)| columns.clone());

                    match labels {
                        Some(columns) => columns.into_iter().map(Ok).collect(),
                        None => {
                            vec![Err(FetchError::TableAliasNotFound(
                                target_table_alias.to_owned(),
                            )
                            .into())]
                        }
                    }
                }
                SelectItemPlan::Expr { label, .. } => vec![Ok(label.to_owned())],
            })
            .collect::<Result<_>>(),
    }
}

1	use {
2	super::{context::RowContext, evaluate::evaluate_stateless, filter::check_expr},
3	crate::{
4	ast::{ColumnDef, ColumnUniqueOption, Dictionary, ToSql, ToSqlUnquoted},
5	data::{Key, Row, SCHEMALESS_DOC_COLUMN, Value},
6	executor::{evaluate::evaluate, select::select},
7	plan::{
8	ExprPlan, IndexItemPlan, JoinPlan, ProjectionPlan, QueryPlan, SelectItemPlan,
9	SetExprPlan, TableAliasPlan, TableFactorPlan, TableWithJoinsPlan, ValuesPlan,
10	},
11	result::Result,
12	store::GStore,
13	},
14	serde::Serialize,
15	std::{borrow::Cow, collections::BTreeMap, fmt::Debug, iter, rc::Rc},
16	thiserror::Error as ThisError,
17	};
18
19	pub type KeyedRows<'a> = Box<dyn Iterator<Item = Result<(Key, Row)>> + 'a>;
20	pub type RelationRows<'a> = Box<dyn Iterator<Item = Result<Row>> + 'a>;
21
22	#[derive(ThisError, Serialize, Debug, PartialEq, Eq)]
23	pub enum FetchError {
24	#[error("table not found: {0}")]
25	TableNotFound(String),
26
27	#[error("table alias not found: {0}")]
28	TableAliasNotFound(String),
29
30	#[error("SERIES has wrong size: {0}")]
31	SeriesSizeWrong(i64),
32
33	#[error("table '{0}' has {1} columns available but {2} column aliases specified")]
34	TooManyColumnAliases(String, usize, usize),
35
36	#[error("unreachable")]
37	Unreachable,
38	}
39
40	pub fn fetch<'a, T: GStore>(	1,072✔
41	storage: &'a T,	1,072✔
42	table_name: &'a str,	1,072✔
43	columns: Rc<[String]>,	1,072✔
44	where_clause: Option<&'a ExprPlan>,	1,072✔
45	) -> Result<KeyedRows<'a>> {	1,072✔
46	let rows = storage.scan_data(table_name)?.filter_map(move \|row\| {	2,929✔
47	let (key, values) = match row {	2,929✔
48	Ok(row) => row,	2,929✔
NEW 49	Err(error) => return Some(Err(error)),	×
50	};
51	let row = Row {	2,929✔
52	columns: Rc::clone(&columns),	2,929✔
53	values,	2,929✔
54	};	2,929✔
55
56	match where_clause {	2,929✔
57	Some(expr) => {	1,615✔
58	let context = RowContext::new(table_name, Cow::Borrowed(&row), None);	1,615✔
59	let context = Rc::new(context);	1,615✔
60	match check_expr(storage, Some(&context), None, expr) {	1,615✔
61	Ok(true) => Some(Ok((key, row))),	686✔
62	Ok(false) => None,	929✔
NEW 63	Err(error) => Some(Err(error)),	×
64	}
65	}
66	None => Some(Ok((key, row))),	1,314✔
67	}
68	});	2,929✔
69
70	Ok(Box::new(rows))	1,072✔
71	}	1,072✔
72
73	pub fn fetch_relation_rows<'a, T: GStore>(	26,027✔
74	storage: &'a T,	26,027✔
75	table_factor: &'a TableFactorPlan,	26,027✔
76	filter_context: Option<&Rc<RowContext<'a>>>,	26,027✔
77	) -> Result<RelationRows<'a>> {	26,027✔
78	let columns = Rc::from(fetch_relation_columns(storage, table_factor)?);	26,027✔
79
80	match table_factor {	25,835✔
81	TableFactorPlan::Derived { subquery, .. } => {	770✔
82	let filter_context = filter_context.map(Rc::clone);	770✔
83	let rows = select(storage, subquery, filter_context)?.map(move \|row\| {	2,184✔
84	let row = row?;	2,184✔
85	Ok(Row {	2,184✔
86	columns: Rc::clone(&columns),	2,184✔
87	values: row.values,	2,184✔
88	})	2,184✔
89	});	2,184✔
90
91	Ok(Box::new(rows))	770✔
92	}
93	TableFactorPlan::Table { name, .. } => {	20,343✔
94	let rows = match table_factor.index() {	20,343✔
95	Some(IndexItemPlan::NonClustered {
96	name: index_name,	98✔
97	asc,	98✔
98	cmp_expr,	98✔
99	}) => {
100	let cmp_value = match cmp_expr {	98✔
101	Some((op, expr)) => {	91✔
102	let evaluated = evaluate(storage, None, None, expr)?;	91✔
103
104	Some((op, evaluated.try_into()?))	91✔
105	}
106	None => None,	7✔
107	};
108
109	let rows = storage	98✔
110	.scan_indexed_data(name, index_name, *asc, cmp_value)?	98✔
111	.map(move \|row\| {	227✔
112	let (_, values) = row?;	227✔
113	Ok(Row {	227✔
114	columns: Rc::clone(&columns),	227✔
115	values,	227✔
116	})	227✔
117	});	227✔
118	Box::new(rows) as Box<dyn Iterator<Item = Result<Row>> + 'a>	98✔
119	}
120	Some(IndexItemPlan::PrimaryKey(expr)) => {	56✔
121	let schema = storage.fetch_schema(name)?.ok_or(FetchError::Unreachable)?;	56✔
122
123	let evaluated = evaluate(storage, filter_context, None, expr)?;	56✔
124
125	let column_def = schema	56✔
126	.column_defs	56✔
127	.as_ref()	56✔
128	.and_then(\|column_defs\| {	56✔
129	column_defs	56✔
130	.iter()	56✔
131	.find(\|column_def\| column_def.unique.is_some_and(\|u\| u.is_primary))	56✔
132	})	56✔
133	.ok_or(FetchError::Unreachable)?;	56✔
134
135	let value =	56✔
136	evaluated.try_into_value(&column_def.data_type, column_def.nullable)?;	56✔
137	let key = Key::try_from(value)?;	56✔
138
139	match storage.fetch_data(name, &key)? {	56✔
140	Some(values) => Box::new(iter::once(Ok(Row {	56✔
141	columns: Rc::clone(&columns),	56✔
142	values,	56✔
143	})))	56✔
144	as Box<dyn Iterator<Item = Result<Row>> + 'a>,	56✔
145	None => {
NEW 146	Box::new(iter::empty()) as Box<dyn Iterator<Item = Result<Row>> + 'a>	×
147	}
148	}
149	}
150	_ => {
151	let rows = storage.scan_data(name)?.map(move \|row\| {	79,659✔
152	let (_, values) = row?;	79,654✔
153	Ok(Row {	79,653✔
154	columns: Rc::clone(&columns),	79,653✔
155	values,	79,653✔
156	})	79,653✔
157	});	79,654✔
158	Box::new(rows) as Box<dyn Iterator<Item = Result<Row>> + 'a>	20,185✔
159	}
160	};
161
162	Ok(rows)	20,339✔
163	}
164	TableFactorPlan::Series { size, .. } => {	4,606✔
165	let value: Value = evaluate_stateless(None, size)?.try_into()?;	4,606✔
166	let size: i64 = value.try_into()?;	4,606✔
167	let size = match size {	4,606✔
168	n if n >= 0 => size,	4,606✔
169	n => return Err(FetchError::SeriesSizeWrong(n).into()),	14✔
170	};
171
172	let columns = Rc::from(vec!["N".to_owned()]);	4,592✔
173	let rows = (1..=size).map(move \|v\| {	5,274✔
174	Ok(Row {	5,274✔
175	columns: Rc::clone(&columns),	5,274✔
176	values: vec![Value::I64(v)],	5,274✔
177	})	5,274✔
178	});	5,274✔
179
180	Ok(Box::new(rows))	4,592✔
181	}
182	TableFactorPlan::Dictionary { dict, .. } => {	116✔
183	let rows = match dict {	116✔
184	Dictionary::GlueObjects => {
185	let schemas = storage.fetch_all_schemas()?;	15✔
186	let table_metas = storage	15✔
187	.scan_table_meta()?	15✔
188	.collect::<Result<BTreeMap<_, _>>>()?;	15✔
189	let rows = schemas.into_iter().flat_map(move \|schema\| {	45✔
190	let meta = table_metas	43✔
191	.iter()	43✔
192	.find_map(\|(table_name, hash_map)\| {	43✔
193	(table_name == &schema.table_name).then(\|\| hash_map.clone())	2✔
194	})	2✔
195	.unwrap_or_default();	43✔
196
197	let table_rows = BTreeMap::from([	43✔
198	("OBJECT_NAME".to_owned(), Value::Str(schema.table_name)),	43✔
199	("OBJECT_TYPE".to_owned(), Value::Str("TABLE".to_owned())),	43✔
200	])	43✔
201	.into_iter()	43✔
202	.chain(meta)	43✔
203	.collect::<BTreeMap<_, _>>();	43✔
204
205	let index_rows = schema.indexes.into_iter().map(\|index\| {	44✔
206	BTreeMap::from([	12✔
207	("OBJECT_NAME".to_owned(), Value::Str(index.name)),	12✔
208	("OBJECT_TYPE".to_owned(), Value::Str("INDEX".to_owned())),	12✔
209	])	12✔
210	});	12✔
211
212	iter::once(table_rows).chain(index_rows).map(\|hash_map\| {	55✔
213	let (columns, values): (Vec<_>, Vec<_>) = hash_map.into_iter().unzip();	55✔
214	Row {	55✔
215	columns: columns.into(),	55✔
216	values,	55✔
217	}	55✔
218	})	55✔
219	});	43✔
220
221	Box::new(rows.map(Ok)) as Box<dyn Iterator<Item = Result<Row>> + 'a>	15✔
222	}
223	Dictionary::GlueTables => {
224	let schemas = storage.fetch_all_schemas()?;	83✔
225	let rows = schemas.into_iter().map(move \|schema\| Row {	83✔
226	columns: Rc::clone(&columns),	138✔
227	values: vec![	138✔
228	Value::Str(schema.table_name),	138✔
229	schema.comment.map_or(Value::Null, Value::Str),	138✔
230	],
231	});	138✔
232
233	Box::new(rows.map(Ok))	83✔
234	}
235	Dictionary::GlueTableColumns => {
236	let schemas = storage.fetch_all_schemas()?;	14✔
237	let rows = schemas.into_iter().flat_map(move \|schema\| {	42✔
238	let columns = Rc::clone(&columns);	42✔
239	let table_name = schema.table_name;	42✔
240
241	schema	42✔
242	.column_defs	42✔
243	.unwrap_or_default()	42✔
244	.into_iter()	42✔
245	.enumerate()	42✔
246	.map(move \|(index, column_def)\| {	84✔
247	let values = vec![	84✔
248	Value::Str(table_name.clone()),	84✔
249	Value::Str(column_def.name),	84✔
250	Value::I64(index as i64 + 1),	84✔
251	Value::Bool(column_def.nullable),	84✔
252	column_def	84✔
253	.unique	84✔
254	.map_or(Value::Null, \|unique\| Value::Str(unique.to_sql())),	84✔
255	column_def	84✔
256	.default	84✔
257	.map_or(Value::Null, \|expr\| Value::Str(expr.to_sql())),	84✔
258	column_def.comment.map_or(Value::Null, Value::Str),	84✔
259	];
260
261	Row {	84✔
262	columns: Rc::clone(&columns),	84✔
263	values,	84✔
264	}	84✔
265	})	84✔
266	});	42✔
267
268	Box::new(rows.map(Ok))	14✔
269	}
270	Dictionary::GlueIndexes => {
271	let schemas = storage.fetch_all_schemas()?;	4✔
272	let rows = schemas.into_iter().flat_map(move \|schema\| {	5✔
273	let column_defs = schema.column_defs.unwrap_or_default();	5✔
274	let primary_column = column_defs.iter().find_map(\|column_def\| {	11✔
275	let ColumnDef { name, unique, .. } = column_def;	11✔
276
277	(unique == &Some(ColumnUniqueOption { is_primary: true }))	11✔
278	.then_some(name)	11✔
279	});	11✔
280
281	let clustered = match primary_column {	5✔
282	Some(column_name) => {	1✔
283	let values = vec![	1✔
284	Value::Str(schema.table_name.clone()),	1✔
285	Value::Str("PRIMARY".to_owned()),	1✔
286	Value::Str("BOTH".to_owned()),	1✔
287	Value::Str(column_name.to_owned()),	1✔
288	Value::Bool(true),	1✔
289	];
290
291	let row = Row {	1✔
292	columns: Rc::clone(&columns),	1✔
293	values,	1✔
294	};	1✔
295
296	vec![row]	1✔
297	}
298	None => Vec::new(),	4✔
299	};
300
301	let columns = Rc::clone(&columns);	5✔
302	let non_clustered = schema.indexes.into_iter().map(move \|index\| {	11✔
303	let values = vec![	11✔
304	Value::Str(schema.table_name.clone()),	11✔
305	Value::Str(index.name),	11✔
306	Value::Str(index.order.to_string()),	11✔
307	Value::Str(index.expr.to_sql_unquoted()),	11✔
308	Value::Bool(false),	11✔
309	];
310
311	Row {	11✔
312	columns: Rc::clone(&columns),	11✔
313	values,	11✔
314	}	11✔
315	});	11✔
316
317	clustered.into_iter().chain(non_clustered)	5✔
318	});	5✔
319
320	Box::new(rows.map(Ok))	4✔
321	}
322	};
323
324	Ok(rows)	116✔
325	}
326	}
327	}	26,027✔
328
329	pub fn fetch_columns<T: GStore>(storage: &T, table_name: &str) -> Result<Vec<String>> {	43,564✔
330	let columns = storage	43,564✔
331	.fetch_schema(table_name)?	43,564✔
332	.ok_or_else(\|\| FetchError::TableNotFound(table_name.to_owned()))?	43,564✔
333	.column_defs
334	.map_or_else(	43,414✔
335	\|\| vec![SCHEMALESS_DOC_COLUMN.to_owned()],	999✔
336	\|column_defs\| {	42,415✔
337	column_defs	42,415✔
338	.into_iter()	42,415✔
339	.map(\|column_def\| column_def.name)	42,415✔
340	.collect()	42,415✔
341	},	42,415✔
342	);
343
344	Ok(columns)	43,414✔
345	}	43,564✔
346
347	pub fn fetch_relation_columns<T>(storage: &T, table_factor: &TableFactorPlan) -> Result<Vec<String>>	54,221✔
348	where	54,221✔
349	T: GStore,	54,221✔
350	{
351	match table_factor {	54,221✔
352	TableFactorPlan::Table { name, alias, .. } => {	43,041✔
353	let columns = fetch_columns(storage, name)?;	43,041✔
354	match alias {	5,718✔
355	None => Ok(columns),	37,173✔
356	Some(alias) if alias.columns.len() > columns.len() => {	5,718✔
357	Err(FetchError::TooManyColumnAliases(	14✔
358	name.clone(),	14✔
359	columns.len(),	14✔
360	alias.columns.len(),	14✔
361	)	14✔
362	.into())	14✔
363	}
364	Some(alias) => Ok(alias	5,704✔
365	.columns	5,704✔
366	.iter()	5,704✔
367	.cloned()	5,704✔
368	.chain(columns[alias.columns.len()..columns.len()].to_vec())	5,704✔
369	.collect()),	5,704✔
370	}
371	}
372	TableFactorPlan::Series { .. } => Ok(vec!["N".to_owned()]),	9,226✔
373	TableFactorPlan::Dictionary { dict, .. } => Ok(match dict {	232✔
374	Dictionary::GlueObjects => vec![	30✔
375	"OBJECT_NAME".to_owned(),	30✔
376	"OBJECT_TYPE".to_owned(),	30✔
377	"CREATED".to_owned(),	30✔
378	],
379	Dictionary::GlueTables => vec!["TABLE_NAME".to_owned(), "COMMENT".to_owned()],	166✔
380	Dictionary::GlueTableColumns => vec![	28✔
381	"TABLE_NAME".to_owned(),	28✔
382	"COLUMN_NAME".to_owned(),	28✔
383	"COLUMN_ID".to_owned(),	28✔
384	"NULLABLE".to_owned(),	28✔
385	"KEY".to_owned(),	28✔
386	"DEFAULT".to_owned(),	28✔
387	"COMMENT".to_owned(),	28✔
388	],
389	Dictionary::GlueIndexes => vec![	8✔
390	"TABLE_NAME".to_owned(),	8✔
391	"INDEX_NAME".to_owned(),	8✔
392	"ORDER".to_owned(),	8✔
393	"EXPRESSION".to_owned(),	8✔
394	"UNIQUENESS".to_owned(),	8✔
395	],
396	}),
397	TableFactorPlan::Derived {
398	subquery: QueryPlan { body, .. },	1,722✔
399	alias:
400	TableAliasPlan {
401	columns: alias_columns,	1,722✔
402	name,	1,722✔
403	},
404	} => match body {	1,722✔
405	SetExprPlan::Select(statement) => {	1,456✔
406	let crate::plan::SelectPlan {
407	from:
408	TableWithJoinsPlan {
409	relation, joins, ..	1,456✔
410	},
411	projection,	1,456✔
412	..
413	} = statement.as_ref();	1,456✔
414
415	let labels = fetch_labels(storage, relation, joins, projection)?;	1,456✔
416	if alias_columns.is_empty() {	1,456✔
417	Ok(labels)	1,358✔
418	} else if alias_columns.len() > labels.len() {	98✔
419	Err(FetchError::TooManyColumnAliases(	14✔
420	name.clone(),	14✔
421	labels.len(),	14✔
422	alias_columns.len(),	14✔
423	)	14✔
424	.into())	14✔
425	} else {
426	Ok(alias_columns	84✔
427	.iter()	84✔
428	.cloned()	84✔
429	.chain(labels[alias_columns.len()..labels.len()].to_vec())	84✔
430	.collect())	84✔
431	}
432	}
433	SetExprPlan::Values(ValuesPlan(values_list)) => {	266✔
434	let total_len = values_list[0].len();	266✔
435	let alias_len = alias_columns.len();	266✔
436	if alias_len > total_len {	266✔
437	return Err(FetchError::TooManyColumnAliases(	14✔
438	name.into(),	14✔
439	total_len,	14✔
440	alias_len,	14✔
441	)	14✔
442	.into());	14✔
443	}	252✔
444	let labels = (alias_len + 1..=total_len).map(\|i\| format!("column{i}"));	364✔
445	let labels = alias_columns	252✔
446	.iter()	252✔
447	.cloned()	252✔
448	.chain(labels)	252✔
449	.collect::<Vec<_>>();	252✔
450
451	Ok(labels)	252✔
452	}
453	},
454	}
455	}	54,221✔
456
457	fn fetch_join_columns<'a, T: GStore>(	25,146✔
458	storage: &T,	25,146✔
459	joins: &'a [JoinPlan],	25,146✔
460	) -> Result<Vec<(&'a str, Vec<String>)>> {	25,146✔
461	let mut all_columns = Vec::with_capacity(joins.len());	25,146✔
462	for join in joins {	25,146✔
463	let columns = fetch_relation_columns(storage, &join.relation)?;	1,594✔
464	let alias = join.relation.alias_name();	1,594✔
465	all_columns.push((alias, columns));	1,594✔
466	}
467	Ok(all_columns)	25,146✔
468	}	25,146✔
469
470	pub fn fetch_labels<T: GStore>(	25,146✔
471	storage: &T,	25,146✔
472	relation: &TableFactorPlan,	25,146✔
473	joins: &[JoinPlan],	25,146✔
474	projection: &ProjectionPlan,	25,146✔
475	) -> Result<Vec<String>> {	25,146✔
476	let table_alias = relation.alias_name();	25,146✔
477	let columns = fetch_relation_columns(storage, relation)?;	25,146✔
478	let join_columns = fetch_join_columns(storage, joins)?;	25,146✔
479
480	match projection {	25,146✔
481	ProjectionPlan::SchemalessMap => Ok(vec![SCHEMALESS_DOC_COLUMN.to_owned()]),	194✔
482	ProjectionPlan::SelectItems(projection) => projection	24,952✔
483	.iter()	24,952✔
484	.flat_map(\|item\| match item {	31,170✔
485	SelectItemPlan::Wildcard => {
486	let columns = columns.iter().cloned();	5,151✔
487	let join_columns = join_columns.iter().flat_map(\|(_, columns)\| columns.clone());	5,151✔
488
489	columns.chain(join_columns).map(Ok).collect()	5,151✔
490	}
491	SelectItemPlan::QualifiedWildcard(target_table_alias) => {	267✔
492	if table_alias == target_table_alias {	267✔
493	return columns.iter().cloned().map(Ok).collect();	197✔
494	}	70✔
495
496	let labels = join_columns	70✔
497	.iter()	70✔
498	.find(\|(table_alias, _)\| table_alias == target_table_alias)	70✔
499	.map(\|(_, columns)\| columns.clone());	70✔
500
501	match labels {	70✔
502	Some(columns) => columns.into_iter().map(Ok).collect(),	56✔
503	None => {
504	vec![Err(FetchError::TableAliasNotFound(	14✔
505	target_table_alias.to_owned(),	14✔
506	)	14✔
507	.into())]	14✔
508	}
509	}
510	}
511	SelectItemPlan::Expr { label, .. } => vec![Ok(label.to_owned())],	25,752✔
512	})	31,170✔
513	.collect::<Result<_>>(),	24,952✔
514	}
515	}	25,146✔

gluesql / gluesql / 27895806754

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous