17021222400

Committed 17 Aug 2025 12:49PM UTC coverage: 97.834% (-0.2%) from 98.008%

Build # 17021222400

Build Type

push

github

Committed by

web-flow

Commit Message

Bump Rust toolchain to 1.88 (#1747)

Update format! macros to use direct variable interpolation instead of
positional arguments for improved readability and Rust 1.88 compliance.

- Fixed clippy warnings across core, cli, storages, and test packages
- Changed format!("{}", var) to format!("{var}") pattern
- Applied fixes consistently throughout the entire workspace
- Updated rust-toolchain.toml to use Rust 1.88

Signed-off-by: john <meenseek5929@naver.com>

Coverage Stats

92 of 122 new or added lines in 48 files covered. (75.41%)

60 existing lines in 20 files now uncovered.

37086 of 37907 relevant lines covered (97.83%)

127441.63 hits per line

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

98.27

/core/src/executor/fetch.rs

use {
    super::{context::RowContext, evaluate::evaluate_stateless, filter::check_expr},
    crate::{
        ast::{
            ColumnDef, ColumnUniqueOption, Dictionary, Expr, IndexItem, Join, Query, Select,
            SelectItem, SetExpr, TableAlias, TableFactor, TableWithJoins, ToSql, ToSqlUnquoted,
            Values,
        },
        data::{Key, Row, Value, get_alias, get_index},
        executor::{
            evaluate::{Evaluated, evaluate},
            select::select,
        },
        result::Result,
        store::{DataRow, GStore},
    },
    async_recursion::async_recursion,
    futures::{
        future,
        stream::{self, Stream, TryStreamExt},
    },
    serde::Serialize,
    std::{borrow::Cow, collections::BTreeMap, fmt::Debug, iter, sync::Arc},
    thiserror::Error as ThisError,
};

#[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 async fn fetch<'a, T: GStore>(
    storage: &'a T,
    table_name: &'a str,
    columns: Option<Arc<[String]>>,
    where_clause: Option<&'a Expr>,
) -> Result<impl Stream<Item = Result<(Key, Row)>> + 'a> {
    let columns = columns.unwrap_or_else(|| Arc::from([]));
    let rows = storage
        .scan_data(table_name)
        .await?
        .try_filter_map(move |(key, data_row)| {
            let row = match data_row {
                DataRow::Vec(values) => Row::Vec {
                    columns: Arc::clone(&columns),
                    values,
                },
                DataRow::Map(values) => Row::Map(values),
            };

            async move {
                let expr = match where_clause {
                    None => {
                        return Ok(Some((key, row)));
                    }
                    Some(expr) => expr,
                };

                let context = RowContext::new(table_name, Cow::Borrowed(&row), None);

                check_expr(storage, Some(Arc::new(context)), None, expr)
                    .await
                    .map(|pass| pass.then_some((key, row)))
            }
        });

    Ok(rows)
}

#[derive(futures_enum::Stream)]
pub enum Rows<I1, I2, I3, I4> {
    Derived(I1),
    Table(I2),
    Series(I3),
    Dictionary(I4),
}

pub async fn fetch_relation_rows<'a, T: GStore>(
    storage: &'a T,
    table_factor: &'a TableFactor,
    filter_context: &Option<Arc<RowContext<'a>>>,
) -> Result<impl Stream<Item = Result<Row>> + 'a> {
    let columns = Arc::from(
        fetch_relation_columns(storage, table_factor)
            .await?
            .unwrap_or_default(),
    );

    match table_factor {
        TableFactor::Derived { subquery, .. } => {
            let filter_context = filter_context.as_ref().map(Arc::clone);
            let rows =
                select(storage, subquery, filter_context)
                    .await?
                    .map_ok(move |row| match row {
                        Row::Vec { values, .. } => Row::Vec {
                            columns: Arc::clone(&columns),
                            values,
                        },
                        Row::Map(values) => Row::Map(values),
                    });

            Ok(Rows::Derived(rows))
        }
        TableFactor::Table { name, .. } => {
            let rows = {
                #[derive(futures_enum::Stream)]
                enum Rows<I1, I2, I3, I4> {
                    Indexed(I1),
                    PrimaryKey(I2),
                    PrimaryKeyEmpty(I3),
                    FullScan(I4),
                }

                match get_index(table_factor) {
                    Some(IndexItem::NonClustered {
                        name: index_name,
                        asc,
                        cmp_expr,
                    }) => {
                        let cmp_value = match cmp_expr {
                            Some((op, expr)) => {
                                let evaluated = evaluate(storage, None, None, expr).await?;

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

                        let rows = storage
                            .scan_indexed_data(name, index_name, *asc, cmp_value)
                            .await?
                            .map_ok(move |(_, data_row)| match data_row {
                                DataRow::Vec(values) => Row::Vec {
                                    columns: Arc::clone(&columns),
                                    values,
                                },
                                DataRow::Map(values) => Row::Map(values),
                            });

                        Rows::Indexed(rows)
                    }
                    Some(IndexItem::PrimaryKey(expr)) => {
                        let schema = storage
                            .fetch_schema(name)
                            .await?
                            .ok_or(FetchError::Unreachable)?;

                        let filter_context = filter_context.as_ref().map(Arc::clone);
                        let evaluated = evaluate(storage, filter_context, None, expr).await?;

                        let value = match evaluated {
                            Evaluated::Literal(literal) => {
                                let data_type = schema
                                    .column_defs
                                    .as_ref()
                                    .and_then(|column_defs| {
                                        column_defs.iter().find(|column_def| {
                                            column_def.unique.map(|u| u.is_primary) == Some(true)
                                        })
                                    })
                                    .map(|column_def| &column_def.data_type)
                                    .ok_or(FetchError::Unreachable)?;

                                Value::try_from_literal(data_type, &literal)
                            }
                            eval => eval.try_into(),
                        }?;
                        let key = Key::try_from(value)?;

                        match storage.fetch_data(name, &key).await? {
                            Some(data_row) => {
                                let row = match data_row {
                                    DataRow::Vec(values) => Row::Vec {
                                        columns: Arc::clone(&columns),
                                        values,
                                    },
                                    DataRow::Map(values) => Row::Map(values),
                                };

                                Rows::PrimaryKey(stream::once(future::ready(Ok(row))))
                            }
                            None => Rows::PrimaryKeyEmpty(stream::empty()),
                        }
                    }
                    _ => {
                        let rows = storage.scan_data(name).await?.map_ok(move |(_, data_row)| {
                            match data_row {
                                DataRow::Vec(values) => Row::Vec {
                                    columns: Arc::clone(&columns),
                                    values,
                                },
                                DataRow::Map(values) => Row::Map(values),
                            }
                        });

                        Rows::FullScan(rows)
                    }
                }
            };

            Ok(Rows::Table(rows))
        }
        TableFactor::Series { size, .. } => {
            let value: Value = evaluate_stateless(None, size).await?.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 = Arc::from(vec!["N".to_owned()]);
            let rows = (1..=size).map(move |v| {
                Ok(Row::Vec {
                    columns: Arc::clone(&columns),
                    values: vec![Value::I64(v)],
                })
            });

            Ok(Rows::Series(stream::iter(rows)))
        }
        TableFactor::Dictionary { dict, .. } => {
            let rows = {
                #[derive(futures_enum::Stream)]
                enum Rows<I1, I2, I3, I4> {
                    Tables(I1),
                    TableColumns(I2),
                    Indexes(I3),
                    Objects(I4),
                }

                match dict {
                    Dictionary::GlueObjects => {
                        let schemas = storage.fetch_all_schemas().await?;
                        let table_metas = storage
                            .scan_table_meta()
                            .await?
                            .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| Ok(Row::Map(hash_map)))
                        });

                        Rows::Objects(stream::iter(rows))
                    }
                    Dictionary::GlueTables => {
                        let schemas = storage.fetch_all_schemas().await?;
                        let rows = schemas.into_iter().map(move |schema| {
                            Ok(Row::Vec {
                                columns: Arc::clone(&columns),
                                values: vec![
                                    Value::Str(schema.table_name),
                                    schema.comment.map(Value::Str).unwrap_or(Value::Null),
                                ],
                            })
                        });

                        Rows::Tables(stream::iter(rows))
                    }
                    Dictionary::GlueTableColumns => {
                        let schemas = storage.fetch_all_schemas().await?;
                        let rows = schemas.into_iter().flat_map(move |schema| {
                            let columns = Arc::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(|unique| Value::Str(unique.to_sql()))
                                            .unwrap_or(Value::Null),
                                        column_def
                                            .default
                                            .map(|expr| Value::Str(expr.to_sql()))
                                            .unwrap_or(Value::Null),
                                        column_def.comment.map(Value::Str).unwrap_or(Value::Null),
                                    ];

                                    Ok(Row::Vec {
                                        columns: Arc::clone(&columns),
                                        values,
                                    })
                                })
                        });

                        Rows::TableColumns(stream::iter(rows))
                    }
                    Dictionary::GlueIndexes => {
                        let schemas = storage.fetch_all_schemas().await?;
                        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::Vec {
                                        columns: Arc::clone(&columns),
                                        values,
                                    };

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

                            let columns = Arc::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),
                                ];

                                Ok(Row::Vec {
                                    columns: Arc::clone(&columns),
                                    values,
                                })
                            });

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

                        Rows::Indexes(stream::iter(rows))
                    }
                }
            };

            Ok(Rows::Dictionary(rows))
        }
    }
}

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

    Ok(columns)
}

#[async_recursion]
pub async fn fetch_relation_columns<T>(
    storage: &T,
    table_factor: &TableFactor,
) -> Result<Option<Vec<String>>>
where
    T: GStore,
{
    match table_factor {
        TableFactor::Table { name, alias, .. } => {
            let columns = fetch_columns(storage, name).await?;
            match (columns, alias) {
                (columns, None) => Ok(columns),
                (None, Some(_)) => Ok(None),
                (Some(columns), Some(alias)) if alias.columns.len() > columns.len() => {
                    Err(FetchError::TooManyColumnAliases(
                        name.to_string(),
                        columns.len(),
                        alias.columns.len(),
                    )
                    .into())
                }
                (Some(columns), Some(alias)) => Ok(Some(
                    alias
                        .columns
                        .iter()
                        .cloned()
                        .chain(columns[alias.columns.len()..columns.len()].to_vec())
                        .collect(),
                )),
            }
        }
        TableFactor::Series { .. } => Ok(Some(vec!["N".to_owned()])),
        TableFactor::Dictionary { dict, .. } => Ok(Some(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(),
            ],
        })),
        TableFactor::Derived {
            subquery: Query { body, .. },
            alias:
                TableAlias {
                    columns: alias_columns,
                    name,
                },
        } => match body {
            SetExpr::Select(statement) => {
                let Select {
                    from:
                        TableWithJoins {
                            relation, joins, ..
                        },
                    projection,
                    ..
                } = statement.as_ref();

                let labels = fetch_labels(storage, relation, joins, projection).await?;
                match labels {
                    None => Ok(None),
                    Some(labels) if alias_columns.is_empty() => Ok(Some(labels)),
                    Some(labels) if alias_columns.len() > labels.len() => {
                        Err(FetchError::TooManyColumnAliases(
                            name.to_string(),
                            labels.len(),
                            alias_columns.len(),
                        )
                        .into())
                    }
                    Some(labels) => Ok(Some(
                        alias_columns
                            .iter()
                            .cloned()
                            .chain(labels[alias_columns.len()..labels.len()].to_vec())
                            .collect(),
                    )),
                }
            }
            SetExpr::Values(Values(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(Some(labels))
            }
        },
    }
}

async fn fetch_join_columns<'a, T: GStore>(
    storage: &T,
    joins: &'a [Join],
) -> Result<Option<Vec<(&'a String, Vec<String>)>>> {
    let mut all_columns = Vec::with_capacity(joins.len());
    for join in joins {
        if let Some(columns) = fetch_relation_columns(storage, &join.relation).await? {
            let alias = get_alias(&join.relation);
            all_columns.push((alias, columns));
        } else {
            return Ok(None);
        }
    }
    Ok(Some(all_columns))
}

pub async fn fetch_labels<T: GStore>(
    storage: &T,
    relation: &TableFactor,
    joins: &[Join],
    projection: &[SelectItem],
) -> Result<Option<Vec<String>>> {
    let table_alias = get_alias(relation);
    let columns = fetch_relation_columns(storage, relation).await?;
    let join_columns = fetch_join_columns(storage, joins).await?;

    if (columns.is_none() || join_columns.is_none())
        && projection.iter().any(|item| {
            matches!(
                item,
                SelectItem::Wildcard | SelectItem::QualifiedWildcard(_)
            )
        })
    {
        return Ok(None);
    }

    let columns = columns.unwrap_or_default();
    let join_columns = join_columns.unwrap_or_default();

    projection
        .iter()
        .flat_map(|item| match item {
            SelectItem::Wildcard => {
                let columns = columns.iter().cloned();
                let join_columns = join_columns.iter().flat_map(|(_, columns)| columns.clone());

                columns.chain(join_columns).map(Ok).collect()
            }
            SelectItem::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())]
                    }
                }
            }
            SelectItem::Expr { label, .. } => vec![Ok(label.to_owned())],
        })
        .collect::<Result<_>>()
        .map(Some)
}

1	use {
2	super::{context::RowContext, evaluate::evaluate_stateless, filter::check_expr},
3	crate::{
4	ast::{
5	ColumnDef, ColumnUniqueOption, Dictionary, Expr, IndexItem, Join, Query, Select,
6	SelectItem, SetExpr, TableAlias, TableFactor, TableWithJoins, ToSql, ToSqlUnquoted,
7	Values,
8	},
9	data::{Key, Row, Value, get_alias, get_index},
10	executor::{
11	evaluate::{Evaluated, evaluate},
12	select::select,
13	},
14	result::Result,
15	store::{DataRow, GStore},
16	},
17	async_recursion::async_recursion,
18	futures::{
19	future,
20	stream::{self, Stream, TryStreamExt},
21	},
22	serde::Serialize,
23	std::{borrow::Cow, collections::BTreeMap, fmt::Debug, iter, sync::Arc},
24	thiserror::Error as ThisError,
25	};
26
27	#[derive(ThisError, Serialize, Debug, PartialEq, Eq)]
28	pub enum FetchError {
29	#[error("table not found: {0}")]
30	TableNotFound(String),
31
32	#[error("table alias not found: {0}")]
33	TableAliasNotFound(String),
34
35	#[error("SERIES has wrong size: {0}")]
36	SeriesSizeWrong(i64),
37
38	#[error("table '{0}' has {1} columns available but {2} column aliases specified")]
39	TooManyColumnAliases(String, usize, usize),
40
41	#[error("unreachable")]
42	Unreachable,
43	}
44
45	pub async fn fetch<'a, T: GStore>(	4,172✔
46	storage: &'a T,	4,172✔
47	table_name: &'a str,	4,172✔
48	columns: Option<Arc<[String]>>,	4,172✔
49	where_clause: Option<&'a Expr>,	4,172✔
50	) -> Result<impl Stream<Item = Result<(Key, Row)>> + 'a> {	4,172✔
51	let columns = columns.unwrap_or_else(\|\| Arc::from([]));	4,172✔
52	let rows = storage	4,172✔
53	.scan_data(table_name)	4,172✔
54	.await?	4,172✔
55	.try_filter_map(move \|(key, data_row)\| {	11,432✔
56	let row = match data_row {	11,432✔
57	DataRow::Vec(values) => Row::Vec {	11,032✔
58	columns: Arc::clone(&columns),	11,032✔
59	values,	11,032✔
60	},	11,032✔
61	DataRow::Map(values) => Row::Map(values),	400✔
62	};
63
64	async move {	11,432✔
65	let expr = match where_clause {	11,432✔
66	None => {
67	return Ok(Some((key, row)));	5,252✔
68	}
69	Some(expr) => expr,	6,180✔
70	};
71
72	let context = RowContext::new(table_name, Cow::Borrowed(&row), None);	6,180✔
73
74	check_expr(storage, Some(Arc::new(context)), None, expr)	6,180✔
75	.await
76	.map(\|pass\| pass.then_some((key, row)))	6,180✔
77	}	11,432✔
78	});	11,432✔
79
80	Ok(rows)	4,172✔
81	}	4,172✔
82
83	#[derive(futures_enum::Stream)]
84	pub enum Rows<I1, I2, I3, I4> {
85	Derived(I1),
86	Table(I2),
87	Series(I3),
88	Dictionary(I4),
89	}
90
91	pub async fn fetch_relation_rows<'a, T: GStore>(	100,090✔
92	storage: &'a T,	100,090✔
93	table_factor: &'a TableFactor,	100,090✔
94	filter_context: &Option<Arc<RowContext<'a>>>,	100,090✔
95	) -> Result<impl Stream<Item = Result<Row>> + 'a> {	100,090✔
96	let columns = Arc::from(	99,322✔
97	fetch_relation_columns(storage, table_factor)	100,090✔
98	.await?	100,090✔
99	.unwrap_or_default(),	99,322✔
100	);
101
102	match table_factor {	99,322✔
103	TableFactor::Derived { subquery, .. } => {	3,080✔
104	let filter_context = filter_context.as_ref().map(Arc::clone);	3,080✔
105	let rows =	3,080✔
106	select(storage, subquery, filter_context)	3,080✔
107	.await?	3,080✔
108	.map_ok(move \|row\| match row {	8,736✔
109	Row::Vec { values, .. } => Row::Vec {	8,736✔
110	columns: Arc::clone(&columns),	8,736✔
111	values,	8,736✔
112	},	8,736✔
113	Row::Map(values) => Row::Map(values),	×
114	});	8,736✔
115
116	Ok(Rows::Derived(rows))	3,080✔
117	}
118	TableFactor::Table { name, .. } => {	78,100✔
119	let rows = {	78,084✔
120	#[derive(futures_enum::Stream)]
121	enum Rows<I1, I2, I3, I4> {
122	Indexed(I1),
123	PrimaryKey(I2),
124	PrimaryKeyEmpty(I3),
125	FullScan(I4),
126	}
127
128	match get_index(table_factor) {	78,100✔
129	Some(IndexItem::NonClustered {
130	name: index_name,	392✔
131	asc,	392✔
132	cmp_expr,	392✔
133	}) => {
134	let cmp_value = match cmp_expr {	392✔
135	Some((op, expr)) => {	364✔
136	let evaluated = evaluate(storage, None, None, expr).await?;	364✔
137
138	Some((op, evaluated.try_into()?))	364✔
139	}
140	None => None,	28✔
141	};
142
143	let rows = storage	392✔
144	.scan_indexed_data(name, index_name, *asc, cmp_value)	392✔
145	.await?	392✔
146	.map_ok(move \|(_, data_row)\| match data_row {	908✔
147	DataRow::Vec(values) => Row::Vec {	908✔
148	columns: Arc::clone(&columns),	908✔
149	values,	908✔
150	},	908✔
151	DataRow::Map(values) => Row::Map(values),	×
152	});	908✔
153
154	Rows::Indexed(rows)	392✔
155	}
156	Some(IndexItem::PrimaryKey(expr)) => {	224✔
157	let schema = storage	224✔
158	.fetch_schema(name)	224✔
159	.await?	224✔
160	.ok_or(FetchError::Unreachable)?;	224✔
161
162	let filter_context = filter_context.as_ref().map(Arc::clone);	224✔
163	let evaluated = evaluate(storage, filter_context, None, expr).await?;	224✔
164
165	let value = match evaluated {	224✔
166	Evaluated::Literal(literal) => {	224✔
167	let data_type = schema	224✔
168	.column_defs	224✔
169	.as_ref()	224✔
170	.and_then(\|column_defs\| {	224✔
171	column_defs.iter().find(\|column_def\| {	224✔
172	column_def.unique.map(\|u\| u.is_primary) == Some(true)	224✔
173	})	224✔
174	})	224✔
175	.map(\|column_def\| &column_def.data_type)	224✔
176	.ok_or(FetchError::Unreachable)?;	224✔
177
178	Value::try_from_literal(data_type, &literal)	224✔
179	}
180	eval => eval.try_into(),	×
181	}?;	×
182	let key = Key::try_from(value)?;	224✔
183
184	match storage.fetch_data(name, &key).await? {	224✔
185	Some(data_row) => {	224✔
186	let row = match data_row {	224✔
187	DataRow::Vec(values) => Row::Vec {	224✔
188	columns: Arc::clone(&columns),	224✔
189	values,	224✔
190	},	224✔
191	DataRow::Map(values) => Row::Map(values),	×
192	};
193
194	Rows::PrimaryKey(stream::once(future::ready(Ok(row))))	224✔
195	}
196	None => Rows::PrimaryKeyEmpty(stream::empty()),	×
197	}
198	}
199	_ => {
200	let rows = storage.scan_data(name).await?.map_ok(move \|(_, data_row)\| {	313,860✔
201	match data_row {	313,836✔
202	DataRow::Vec(values) => Row::Vec {	312,212✔
203	columns: Arc::clone(&columns),	312,212✔
204	values,	312,212✔
205	},	312,212✔
206	DataRow::Map(values) => Row::Map(values),	1,624✔
207	}
208	});	313,836✔
209
210	Rows::FullScan(rows)	77,468✔
211	}
212	}
213	};
214
215	Ok(Rows::Table(rows))	78,084✔
216	}
217	TableFactor::Series { size, .. } => {	17,690✔
218	let value: Value = evaluate_stateless(None, size).await?.try_into()?;	17,690✔
219	let size: i64 = value.try_into()?;	17,690✔
220	let size = match size {	17,690✔
221	n if n >= 0 => size,	17,690✔
222	n => return Err(FetchError::SeriesSizeWrong(n).into()),	56✔
223	};
224
225	let columns = Arc::from(vec!["N".to_owned()]);	17,634✔
226	let rows = (1..=size).map(move \|v\| {	19,762✔
227	Ok(Row::Vec {	19,762✔
228	columns: Arc::clone(&columns),	19,762✔
229	values: vec![Value::I64(v)],	19,762✔
230	})	19,762✔
231	});	19,762✔
232
233	Ok(Rows::Series(stream::iter(rows)))	17,634✔
234	}
235	TableFactor::Dictionary { dict, .. } => {	452✔
236	let rows = {	452✔
237	#[derive(futures_enum::Stream)]
238	enum Rows<I1, I2, I3, I4> {
239	Tables(I1),
240	TableColumns(I2),
241	Indexes(I3),
242	Objects(I4),
243	}
244
245	match dict {	452✔
246	Dictionary::GlueObjects => {
247	let schemas = storage.fetch_all_schemas().await?;	48✔
248	let table_metas = storage	48✔
249	.scan_table_meta()	48✔
250	.await?	48✔
251	.collect::<Result<BTreeMap<_, _>>>()?;	48✔
252	let rows = schemas.into_iter().flat_map(move \|schema\| {	104✔
253	let meta = table_metas	96✔
254	.iter()	96✔
255	.find_map(\|(table_name, hash_map)\| {	96✔
256	(table_name == &schema.table_name).then(\|\| hash_map.clone())	8✔
257	})	8✔
258	.unwrap_or_default();	96✔
259
260	let table_rows = BTreeMap::from([	96✔
261	("OBJECT_NAME".to_owned(), Value::Str(schema.table_name)),	96✔
262	("OBJECT_TYPE".to_owned(), Value::Str("TABLE".to_owned())),	96✔
263	])	96✔
264	.into_iter()	96✔
265	.chain(meta)	96✔
266	.collect::<BTreeMap<_, _>>();	96✔
267
268	let index_rows = schema.indexes.into_iter().map(\|index\| {	100✔
269	BTreeMap::from([	36✔
270	("OBJECT_NAME".to_owned(), Value::Str(index.name)),	36✔
271	("OBJECT_TYPE".to_owned(), Value::Str("INDEX".to_owned())),	36✔
272	])	36✔
273	});	36✔
274
275	iter::once(table_rows)	96✔
276	.chain(index_rows)	96✔
277	.map(\|hash_map\| Ok(Row::Map(hash_map)))	132✔
278	});	96✔
279
280	Rows::Objects(stream::iter(rows))	48✔
281	}
282	Dictionary::GlueTables => {
283	let schemas = storage.fetch_all_schemas().await?;	332✔
284	let rows = schemas.into_iter().map(move \|schema\| {	564✔
285	Ok(Row::Vec {	552✔
286	columns: Arc::clone(&columns),	552✔
287	values: vec![	552✔
288	Value::Str(schema.table_name),	552✔
289	schema.comment.map(Value::Str).unwrap_or(Value::Null),	552✔
290	],	552✔
291	})	552✔
292	});	552✔
293
294	Rows::Tables(stream::iter(rows))	332✔
295	}
296	Dictionary::GlueTableColumns => {
297	let schemas = storage.fetch_all_schemas().await?;	56✔
298	let rows = schemas.into_iter().flat_map(move \|schema\| {	168✔
299	let columns = Arc::clone(&columns);	168✔
300	let table_name = schema.table_name;	168✔
301
302	schema	168✔
303	.column_defs	168✔
304	.unwrap_or_default()	168✔
305	.into_iter()	168✔
306	.enumerate()	168✔
307	.map(move \|(index, column_def)\| {	336✔
308	let values = vec![	336✔
309	Value::Str(table_name.clone()),	336✔
310	Value::Str(column_def.name),	336✔
311	Value::I64(index as i64 + 1),	336✔
312	Value::Bool(column_def.nullable),	336✔
313	column_def	336✔
314	.unique	336✔
315	.map(\|unique\| Value::Str(unique.to_sql()))	336✔
316	.unwrap_or(Value::Null),	336✔
317	column_def	336✔
318	.default	336✔
319	.map(\|expr\| Value::Str(expr.to_sql()))	336✔
320	.unwrap_or(Value::Null),	336✔
321	column_def.comment.map(Value::Str).unwrap_or(Value::Null),	336✔
322	];
323
324	Ok(Row::Vec {	336✔
325	columns: Arc::clone(&columns),	336✔
326	values,	336✔
327	})	336✔
328	})	336✔
329	});	168✔
330
331	Rows::TableColumns(stream::iter(rows))	56✔
332	}
333	Dictionary::GlueIndexes => {
334	let schemas = storage.fetch_all_schemas().await?;	16✔
335	let rows = schemas.into_iter().flat_map(move \|schema\| {	20✔
336	let column_defs = schema.column_defs.unwrap_or_default();	20✔
337	let primary_column = column_defs.iter().find_map(\|column_def\| {	44✔
338	let ColumnDef { name, unique, .. } = column_def;	44✔
339
340	(unique == &Some(ColumnUniqueOption { is_primary: true }))	44✔
341	.then_some(name)	44✔
342	});	44✔
343
344	let clustered = match primary_column {	20✔
345	Some(column_name) => {	4✔
346	let values = vec![	4✔
347	Value::Str(schema.table_name.clone()),	4✔
348	Value::Str("PRIMARY".to_owned()),	4✔
349	Value::Str("BOTH".to_owned()),	4✔
350	Value::Str(column_name.to_owned()),	4✔
351	Value::Bool(true),	4✔
352	];
353
354	let row = Row::Vec {	4✔
355	columns: Arc::clone(&columns),	4✔
356	values,	4✔
357	};	4✔
358
359	vec![Ok(row)]	4✔
360	}
361	None => Vec::new(),	16✔
362	};
363
364	let columns = Arc::clone(&columns);	20✔
365	let non_clustered = schema.indexes.into_iter().map(move \|index\| {	44✔
366	let values = vec![	44✔
367	Value::Str(schema.table_name.clone()),	44✔
368	Value::Str(index.name),	44✔
369	Value::Str(index.order.to_string()),	44✔
370	Value::Str(index.expr.to_sql_unquoted()),	44✔
371	Value::Bool(false),	44✔
372	];
373
374	Ok(Row::Vec {	44✔
375	columns: Arc::clone(&columns),	44✔
376	values,	44✔
377	})	44✔
378	});	44✔
379
380	clustered.into_iter().chain(non_clustered)	20✔
381	});	20✔
382
383	Rows::Indexes(stream::iter(rows))	16✔
384	}
385	}
386	};
387
388	Ok(Rows::Dictionary(rows))	452✔
389	}
390	}
391	}	100,090✔
392
393	pub async fn fetch_columns<T: GStore>(	166,636✔
394	storage: &T,	166,636✔
395	table_name: &str,	166,636✔
396	) -> Result<Option<Vec<String>>> {	166,636✔
397	let columns = storage	166,636✔
398	.fetch_schema(table_name)	166,636✔
399	.await?	166,636✔
400	.ok_or_else(\|\| FetchError::TableNotFound(table_name.to_owned()))?	166,636✔
401	.column_defs
402	.map(\|column_defs\| {	166,036✔
403	column_defs	163,512✔
404	.into_iter()	163,512✔
405	.map(\|column_def\| column_def.name)	163,512✔
406	.collect()	163,512✔
407	});	163,512✔
408
409	Ok(columns)	166,036✔
410	}	166,636✔
411
412	#[async_recursion]
413	pub async fn fetch_relation_columns<T>(
414	storage: &T,
415	table_factor: &TableFactor,
416	) -> Result<Option<Vec<String>>>
417	where
418	T: GStore,
419	{	415,880✔
420	match table_factor {	415,880✔
421	TableFactor::Table { name, alias, .. } => {	164,712✔
422	let columns = fetch_columns(storage, name).await?;	164,712✔
423	match (columns, alias) {	164,112✔
424	(columns, None) => Ok(columns),	141,240✔
UNCOV 425	(None, Some(_)) => Ok(None),	×
426	(Some(columns), Some(alias)) if alias.columns.len() > columns.len() => {	22,872✔
427	Err(FetchError::TooManyColumnAliases(	56✔
428	name.to_string(),	56✔
429	columns.len(),	56✔
430	alias.columns.len(),	56✔
431	)	56✔
432	.into())	56✔
433	}
434	(Some(columns), Some(alias)) => Ok(Some(	22,816✔
435	alias	22,816✔
436	.columns	22,816✔
437	.iter()	22,816✔
438	.cloned()	22,816✔
439	.chain(columns[alias.columns.len()..columns.len()].to_vec())	22,816✔
440	.collect(),	22,816✔
441	)),	22,816✔
442	}
443	}
444	TableFactor::Series { .. } => Ok(Some(vec!["N".to_owned()])),	35,436✔
445	TableFactor::Dictionary { dict, .. } => Ok(Some(match dict {	904✔
446	Dictionary::GlueObjects => vec![	96✔
447	"OBJECT_NAME".to_owned(),	96✔
448	"OBJECT_TYPE".to_owned(),	96✔
449	"CREATED".to_owned(),	96✔
450	],
451	Dictionary::GlueTables => vec!["TABLE_NAME".to_owned(), "COMMENT".to_owned()],	664✔
452	Dictionary::GlueTableColumns => vec![	112✔
453	"TABLE_NAME".to_owned(),	112✔
454	"COLUMN_NAME".to_owned(),	112✔
455	"COLUMN_ID".to_owned(),	112✔
456	"NULLABLE".to_owned(),	112✔
457	"KEY".to_owned(),	112✔
458	"DEFAULT".to_owned(),	112✔
459	"COMMENT".to_owned(),	112✔
460	],
461	Dictionary::GlueIndexes => vec![	32✔
462	"TABLE_NAME".to_owned(),	32✔
463	"INDEX_NAME".to_owned(),	32✔
464	"ORDER".to_owned(),	32✔
465	"EXPRESSION".to_owned(),	32✔
466	"UNIQUENESS".to_owned(),	32✔
467	],
468	})),
469	TableFactor::Derived {
470	subquery: Query { body, .. },	6,888✔
471	alias:
472	TableAlias {
473	columns: alias_columns,	6,888✔
474	name,	6,888✔
475	},
476	} => match body {	6,888✔
477	SetExpr::Select(statement) => {	5,824✔
478	let Select {
479	from:
480	TableWithJoins {
481	relation, joins, ..	5,824✔
482	},
483	projection,	5,824✔
484	..
485	} = statement.as_ref();	5,824✔
486
487	let labels = fetch_labels(storage, relation, joins, projection).await?;	5,824✔
488	match labels {	392✔
489	None => Ok(None),	112✔
490	Some(labels) if alias_columns.is_empty() => Ok(Some(labels)),	5,712✔
491	Some(labels) if alias_columns.len() > labels.len() => {	392✔
492	Err(FetchError::TooManyColumnAliases(	56✔
493	name.to_string(),	56✔
494	labels.len(),	56✔
495	alias_columns.len(),	56✔
496	)	56✔
497	.into())	56✔
498	}
499	Some(labels) => Ok(Some(	336✔
500	alias_columns	336✔
501	.iter()	336✔
502	.cloned()	336✔
503	.chain(labels[alias_columns.len()..labels.len()].to_vec())	336✔
504	.collect(),	336✔
505	)),	336✔
506	}
507	}
508	SetExpr::Values(Values(values_list)) => {	1,064✔
509	let total_len = values_list[0].len();	1,064✔
510	let alias_len = alias_columns.len();	1,064✔
511	if alias_len > total_len {	1,064✔
512	return Err(FetchError::TooManyColumnAliases(	56✔
513	name.into(),	56✔
514	total_len,	56✔
515	alias_len,	56✔
516	)	56✔
517	.into());	56✔
518	}	1,008✔
519	let labels = (alias_len + 1..=total_len).map(\|i\| format!("column{i}"));	1,456✔
520	let labels = alias_columns	1,008✔
521	.iter()	1,008✔
522	.cloned()	1,008✔
523	.chain(labels)	1,008✔
524	.collect::<Vec<_>>();	1,008✔
525
526	Ok(Some(labels))	1,008✔
527	}
528	},
529	}
530	}	207,940✔
531
532	async fn fetch_join_columns<'a, T: GStore>(	96,130✔
533	storage: &T,	96,130✔
534	joins: &'a [Join],	96,130✔
535	) -> Result<Option<Vec<(&'a String, Vec<String>)>>> {	96,130✔
536	let mut all_columns = Vec::with_capacity(joins.len());	96,130✔
537	for join in joins {	102,214✔
538	if let Some(columns) = fetch_relation_columns(storage, &join.relation).await? {	6,140✔
539	let alias = get_alias(&join.relation);	6,084✔
540	all_columns.push((alias, columns));	6,084✔
541	} else {	6,084✔
542	return Ok(None);	56✔
543	}
544	}
545	Ok(Some(all_columns))	96,074✔
546	}	96,130✔
547
548	pub async fn fetch_labels<T: GStore>(	96,130✔
549	storage: &T,	96,130✔
550	relation: &TableFactor,	96,130✔
551	joins: &[Join],	96,130✔
552	projection: &[SelectItem],	96,130✔
553	) -> Result<Option<Vec<String>>> {	96,130✔
554	let table_alias = get_alias(relation);	96,130✔
555	let columns = fetch_relation_columns(storage, relation).await?;	96,130✔
556	let join_columns = fetch_join_columns(storage, joins).await?;	96,130✔
557
558	if (columns.is_none() \|\| join_columns.is_none())	96,130✔
559	&& projection.iter().any(\|item\| {	1,852✔
560	matches!(	1,152✔
561	item,	1,852✔
562	SelectItem::Wildcard \| SelectItem::QualifiedWildcard(_)
563	)
564	})	1,852✔
565	{
566	return Ok(None);	700✔
567	}	95,430✔
568
569	let columns = columns.unwrap_or_default();	95,430✔
570	let join_columns = join_columns.unwrap_or_default();	95,430✔
571
572	projection	95,430✔
573	.iter()	95,430✔
574	.flat_map(\|item\| match item {	120,034✔
575	SelectItem::Wildcard => {
576	let columns = columns.iter().cloned();	20,256✔
577	let join_columns = join_columns.iter().flat_map(\|(_, columns)\| columns.clone());	20,256✔
578
579	columns.chain(join_columns).map(Ok).collect()	20,256✔
580	}
581	SelectItem::QualifiedWildcard(target_table_alias) => {	1,068✔
582	if table_alias == target_table_alias {	1,068✔
583	return columns.iter().cloned().map(Ok).collect();	788✔
584	}	280✔
585
586	let labels = join_columns	280✔
587	.iter()	280✔
588	.find(\|(table_alias, _)\| table_alias == &target_table_alias)	280✔
589	.map(\|(_, columns)\| columns.clone());	280✔
590
591	match labels {	280✔
592	Some(columns) => columns.into_iter().map(Ok).collect(),	224✔
593	None => {
594	vec![Err(FetchError::TableAliasNotFound(	56✔
595	target_table_alias.to_owned(),	56✔
596	)	56✔
597	.into())]	56✔
598	}
599	}
600	}
601	SelectItem::Expr { label, .. } => vec![Ok(label.to_owned())],	98,710✔
602	})	120,034✔
603	.collect::<Result<_>>()	95,430✔
604	.map(Some)	95,430✔
605	}	96,130✔

gluesql / gluesql / 17021222400

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