use std::{collections::BTreeMap, sync::Arc};
use crate::{
catalog::{Catalog, projection::ProjectionCache},
error::Result,
store::{
proto::{
ColumnValue, DataFileValue, DeleteFileValue, FileColumnStatsValue, GcFileValue,
HeadValue, MacroValue, MappingValue, OptionScopeValue, PartitionValue, SchemaValue,
SnapshotValue, SortValue, TableColumnStatsValue, TableStatsValue, TableValue, TagValue,
ViewValue,
},
read::{
EntityRecord, read_head, scan_current_entities, scan_history_entities,
scan_schema_versions, scan_snapshots,
},
},
};
async fn session_head(session: &crate::store::handle::ReadSession) -> Result<Option<HeadValue>> {
read_head(session.handle()).await
}
fn projections_read(catalog: &Catalog) -> std::sync::RwLockReadGuard<'_, ProjectionCache> {
catalog
.projections()
.read()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn projections_write(catalog: &Catalog) -> std::sync::RwLockWriteGuard<'_, ProjectionCache> {
catalog
.projections()
.write()
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
#[doc(hidden)]
pub mod index;
#[doc(hidden)]
pub mod inline;
#[doc(hidden)]
pub mod staged;
#[doc(hidden)]
pub use crate::store::proto::{
ColumnValue as ColumnRecord, DataFileValue as DataFileRecord,
DeleteFileValue as DeleteFileRecord, FileColumnStatsValue as FileColumnStatsRecord,
GcFileValue as GcFileRecord, MacroValue as MacroRecord, MappingValue as MappingRecord,
PartitionValue as PartitionRecord, SchemaValue as SchemaRecord,
SnapshotValue as SnapshotRecord, SortValue as SortRecord,
TableColumnStatsValue as TableColumnStatsRecord, TableStatsValue as TableStatsRecord,
TableValue as TableRecord, ViewValue as ViewRecord,
};
async fn all_entities(catalog: &Catalog) -> Result<Arc<Vec<EntityRecord>>> {
let session = catalog.begin_read().await?;
let head = session_head(&session).await?;
let cache_at = match head {
Some(head) => {
if let Some(records) = projections_read(catalog).entities_at(&head) {
session.finish();
return Ok(records);
}
Some(head)
}
None => None,
};
let handle = session.handle();
let scanned = crate::store::read::consistent(handle, || async move {
let mut records = scan_current_entities(handle).await?;
records.extend(scan_history_entities(handle).await?);
Ok(records)
})
.await;
session.finish();
let records = Arc::new(scanned?);
if let Some(head) = cache_at {
projections_write(catalog).install_entities(head, records.as_ref().clone());
}
Ok(records)
}
async fn dump_entities<T>(
catalog: &Catalog,
extract: impl Fn(&EntityRecord) -> Option<T>,
) -> Result<Vec<T>> {
let records = all_entities(catalog).await?;
Ok(records.iter().filter_map(extract).collect())
}
async fn dump_current_entities<T>(
catalog: &Catalog,
extract: impl Fn(&EntityRecord) -> Option<T>,
) -> Result<Vec<T>> {
if catalog.maintains_projections() {
return dump_entities(catalog, extract).await;
}
let session = catalog.begin_read().await?;
let current = scan_current_entities(session.handle()).await;
session.finish();
Ok(current?.iter().filter_map(extract).collect())
}
#[doc(hidden)]
pub async fn dump_schemas(catalog: &Catalog) -> Result<Vec<SchemaValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Schema(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_tables(catalog: &Catalog) -> Result<Vec<TableValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Table(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_views(catalog: &Catalog) -> Result<Vec<ViewValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::View(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_macros(catalog: &Catalog) -> Result<Vec<MacroValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Macro(m) => Some(m.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_mappings(catalog: &Catalog) -> Result<Vec<MappingValue>> {
dump_current_entities(catalog, |r| match r {
EntityRecord::Mapping(m) => Some(m.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_columns(catalog: &Catalog) -> Result<Vec<ColumnValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Column(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_data_files(catalog: &Catalog) -> Result<Vec<DataFileValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::File(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_delete_files(catalog: &Catalog) -> Result<Vec<DeleteFileValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::DeleteFile(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_partition_info(catalog: &Catalog) -> Result<Vec<PartitionValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Partition(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_sort_info(catalog: &Catalog) -> Result<Vec<SortValue>> {
dump_entities(catalog, |r| match r {
EntityRecord::Sort(v) => Some(v.clone()),
_ => None,
})
.await
}
async fn dump_projected_current<T: Clone>(
catalog: &Catalog,
read: impl Fn(&ProjectionCache, &HeadValue) -> Option<Vec<T>>,
install: impl Fn(&mut ProjectionCache, HeadValue, Vec<T>),
extract: impl Fn(EntityRecord) -> Option<T>,
) -> Result<Vec<T>> {
let session = catalog.begin_read().await?;
let head = session_head(&session).await?;
let cache_at = match head {
Some(head) => {
if let Some(rows) = read(&projections_read(catalog), &head) {
session.finish();
return Ok(rows);
}
Some(head)
}
None => None,
};
let current = scan_current_entities(session.handle()).await;
session.finish();
let rows: Vec<T> = current?.into_iter().filter_map(extract).collect();
if let Some(head) = cache_at {
install(&mut projections_write(catalog), head, rows.clone());
}
Ok(rows)
}
#[doc(hidden)]
pub async fn dump_table_stats(catalog: &Catalog) -> Result<Vec<TableStatsValue>> {
dump_projected_current(
catalog,
ProjectionCache::table_stats_at,
ProjectionCache::install_table_stats,
|r| match r {
EntityRecord::TableStats(v) => Some(v),
_ => None,
},
)
.await
}
#[doc(hidden)]
pub async fn dump_table_column_stats(catalog: &Catalog) -> Result<Vec<TableColumnStatsValue>> {
dump_projected_current(
catalog,
ProjectionCache::table_column_stats_at,
ProjectionCache::install_table_column_stats,
|r| match r {
EntityRecord::TableColumnStats(v) => Some(v),
_ => None,
},
)
.await
}
#[doc(hidden)]
pub async fn dump_file_column_stats(catalog: &Catalog) -> Result<Vec<FileColumnStatsValue>> {
dump_current_entities(catalog, |r| match r {
EntityRecord::FileColumnStats(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
pub async fn dump_snapshots(catalog: &Catalog) -> Result<Vec<SnapshotValue>> {
let session = catalog.begin_read().await?;
let head = session_head(&session).await?;
if let Some(head) = head {
if let Some(rows) = projections_read(catalog).snapshots_at(&head) {
session.finish();
return Ok(rows);
}
let result = scan_snapshots(session.handle()).await;
session.finish();
let rows = result?;
projections_write(catalog).install_snapshots(head, rows.clone());
return Ok(rows);
}
let result = scan_snapshots(session.handle()).await;
session.finish();
result
}
#[doc(hidden)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SchemaVersionRow {
pub begin_snapshot: u64,
pub schema_version: u64,
pub table_id: u64,
}
#[doc(hidden)]
pub async fn dump_schema_versions(catalog: &Catalog) -> Result<Vec<SchemaVersionRow>> {
let session = catalog.begin_read().await?;
let records = scan_schema_versions(session.handle()).await;
session.finish();
Ok(schema_version_rows_from(
records?,
&dump_snapshots(catalog).await?,
&dump_data_files(catalog).await?,
))
}
#[doc(hidden)]
#[must_use]
pub fn schema_version_rows_from(
records: Vec<(u64, u64, u64)>,
snapshots: &[SnapshotValue],
data_files: &[DataFileValue],
) -> Vec<SchemaVersionRow> {
let mut rows: BTreeMap<(u64, u64), u64> = records
.into_iter()
.map(|(table_id, begin_snapshot, schema_version)| {
((table_id, begin_snapshot), schema_version)
})
.collect();
for snapshot in snapshots {
for table_id in &snapshot.schema_changed_table_ids {
rows.entry((*table_id, snapshot.snapshot_id))
.or_insert(snapshot.schema_version);
}
}
rows.extend(schema_version_floors(&rows, snapshots, data_files));
rows.into_iter()
.map(
|((table_id, begin_snapshot), schema_version)| SchemaVersionRow {
begin_snapshot,
schema_version,
table_id,
},
)
.collect()
}
fn schema_version_floors(
rows: &BTreeMap<(u64, u64), u64>,
snapshots: &[SnapshotValue],
data_files: &[DataFileValue],
) -> BTreeMap<(u64, u64), u64> {
let mut oldest_file: BTreeMap<u64, u64> = BTreeMap::new();
for file in data_files {
oldest_file
.entry(file.table_id)
.and_modify(|oldest| *oldest = (*oldest).min(file.begin_snapshot))
.or_insert(file.begin_snapshot);
}
let uncovered: Vec<(u64, u64)> = oldest_file
.into_iter()
.filter(|&(table_id, file_begin_snapshot)| {
rows.range((table_id, 0)..=(table_id, file_begin_snapshot))
.next()
.is_none()
})
.collect();
if uncovered.is_empty() {
return BTreeMap::new();
}
let current_schema_version = snapshots
.last()
.map_or(0, |snapshot| snapshot.schema_version);
uncovered
.into_iter()
.map(|(table_id, file_begin_snapshot)| {
let schema_version = rows
.range((table_id, 0)..=(table_id, u64::MAX))
.next()
.map_or(current_schema_version, |(_, version)| *version);
((table_id, file_begin_snapshot), schema_version)
})
.collect()
}
#[doc(hidden)]
pub async fn dump_scheduled_deletions(catalog: &Catalog) -> Result<Vec<GcFileValue>> {
dump_current_entities(catalog, |r| match r {
EntityRecord::GcFile(v) => Some(v.clone()),
_ => None,
})
.await
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TagRow {
pub object_id: u64,
pub begin_snapshot: u64,
pub end_snapshot: Option<u64>,
pub key: String,
pub value: String,
}
#[doc(hidden)]
pub async fn dump_tags(catalog: &Catalog) -> Result<Vec<TagRow>> {
Ok(tag_rows_from(
dump_current_entities(catalog, |r| match r {
EntityRecord::Tag(v) => Some(v.clone()),
_ => None,
})
.await?,
))
}
#[doc(hidden)]
#[must_use]
pub fn tag_rows_from(containers: Vec<TagValue>) -> Vec<TagRow> {
containers
.into_iter()
.flat_map(|container| {
let object_id = container.object_id;
container.entries.into_iter().map(move |e| TagRow {
object_id,
begin_snapshot: e.begin_snapshot,
end_snapshot: e.end_snapshot,
key: e.key,
value: e.value,
})
})
.collect()
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OptionRow {
pub key: String,
pub value: String,
pub scope: Option<String>,
pub scope_id: Option<u64>,
}
#[doc(hidden)]
pub async fn dump_options(catalog: &Catalog) -> Result<Vec<OptionRow>> {
Ok(option_rows_from(
dump_current_entities(catalog, |r| match r {
EntityRecord::Option {
scope_kind,
scope_id,
value,
} => Some((*scope_kind, *scope_id, value.clone())),
_ => None,
})
.await?,
))
}
#[doc(hidden)]
#[must_use]
pub fn option_rows_from(scopes: Vec<(u64, u64, OptionScopeValue)>) -> Vec<OptionRow> {
let mut rows: Vec<OptionRow> = scopes
.into_iter()
.flat_map(|(scope_kind, scope_id, value)| {
let scope = match scope_kind {
1 => Some("schema".to_string()),
2 => Some("table".to_string()),
_ => None,
};
let scope_id = scope.as_ref().map(|_| scope_id);
value
.options
.into_iter()
.map(move |(key, value)| OptionRow {
key,
value,
scope: scope.clone(),
scope_id,
})
})
.collect();
rows.sort_by(|a, b| (&a.scope, a.scope_id, &a.key).cmp(&(&b.scope, b.scope_id, &b.key)));
rows
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ColumnTagRow {
pub table_id: u64,
pub column_id: u64,
pub begin_snapshot: u64,
pub end_snapshot: Option<u64>,
pub key: String,
pub value: String,
}
#[doc(hidden)]
#[must_use]
pub fn column_tag_rows_from(columns: &[ColumnValue]) -> Vec<ColumnTagRow> {
let mut latest: std::collections::BTreeMap<(u64, u64), &ColumnValue> =
std::collections::BTreeMap::new();
for column in columns {
let entry = latest
.entry((column.table_id, column.column_id))
.or_insert(column);
let newer = match (column.end_snapshot, entry.end_snapshot) {
(None, _) => true,
(Some(_), None) => false,
(Some(a), Some(b)) => a > b,
};
if newer {
*entry = column;
}
}
latest
.into_values()
.flat_map(|column| {
column.tags.iter().map(|t| ColumnTagRow {
table_id: column.table_id,
column_id: column.column_id,
begin_snapshot: t.begin_snapshot,
end_snapshot: t.end_snapshot,
key: t.key.clone(),
value: t.value.clone(),
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_column_tags(catalog: &Catalog) -> Result<Vec<ColumnTagRow>> {
Ok(column_tag_rows_from(&dump_columns(catalog).await?))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MacroImplRow {
pub macro_id: u64,
pub impl_id: u64,
pub dialect: String,
pub sql: String,
pub macro_type: String,
}
#[doc(hidden)]
#[must_use]
pub fn macro_impl_rows_from(parents: Vec<MacroValue>) -> Vec<MacroImplRow> {
parents
.into_iter()
.flat_map(|value| {
let macro_id = value.macro_id;
value
.implementations
.into_iter()
.map(move |implementation| MacroImplRow {
macro_id,
impl_id: implementation.impl_id,
dialect: implementation.dialect,
sql: implementation.sql,
macro_type: implementation.macro_type,
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_macro_impl_rows(catalog: &Catalog) -> Result<Vec<MacroImplRow>> {
Ok(macro_impl_rows_from(dump_macros(catalog).await?))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MacroParameterRow {
pub macro_id: u64,
pub impl_id: u64,
pub column_id: u64,
pub parameter_name: String,
pub parameter_type: String,
pub default_value: Option<String>,
pub default_value_type: String,
}
#[doc(hidden)]
#[must_use]
pub fn macro_parameter_rows_from(parents: Vec<MacroValue>) -> Vec<MacroParameterRow> {
parents
.into_iter()
.flat_map(|value| {
let macro_id = value.macro_id;
value.implementations.into_iter().flat_map(move |i| {
let impl_id = i.impl_id;
i.parameters.into_iter().map(move |p| MacroParameterRow {
macro_id,
impl_id,
column_id: p.column_id,
parameter_name: p.parameter_name,
parameter_type: p.parameter_type,
default_value: p.default_value,
default_value_type: p.default_value_type,
})
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_macro_parameter_rows(catalog: &Catalog) -> Result<Vec<MacroParameterRow>> {
Ok(macro_parameter_rows_from(dump_macros(catalog).await?))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NameMappingRow {
pub mapping_id: u64,
pub column_id: u64,
pub source_name: String,
pub target_field_id: u64,
pub parent_column: Option<u64>,
pub is_partition: bool,
}
#[doc(hidden)]
#[must_use]
pub fn name_mapping_rows_from(parents: Vec<MappingValue>) -> Vec<NameMappingRow> {
parents
.into_iter()
.flat_map(|value| {
let mapping_id = value.mapping_id;
value
.name_mappings
.into_iter()
.map(move |row| NameMappingRow {
mapping_id,
column_id: row.column_id,
source_name: row.source_name,
target_field_id: row.target_field_id,
parent_column: row.parent_column,
is_partition: row.is_partition,
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_name_mapping_rows(catalog: &Catalog) -> Result<Vec<NameMappingRow>> {
Ok(name_mapping_rows_from(dump_mappings(catalog).await?))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionColumnRow {
pub partition_id: u64,
pub table_id: u64,
pub partition_key_index: u64,
pub column_id: u64,
pub transform: String,
}
#[doc(hidden)]
#[must_use]
pub fn partition_column_rows_from(parents: Vec<PartitionValue>) -> Vec<PartitionColumnRow> {
parents
.into_iter()
.flat_map(|spec| {
let (partition_id, table_id) = (spec.partition_id, spec.table_id);
spec.columns
.into_iter()
.map(move |column| PartitionColumnRow {
partition_id,
table_id,
partition_key_index: column.partition_key_index,
column_id: column.column_id,
transform: column.transform,
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_partition_column_rows(catalog: &Catalog) -> Result<Vec<PartitionColumnRow>> {
Ok(partition_column_rows_from(
dump_partition_info(catalog).await?,
))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FilePartitionValueRow {
pub data_file_id: u64,
pub table_id: u64,
pub partition_key_index: u64,
pub partition_value: String,
}
#[doc(hidden)]
#[must_use]
pub fn file_partition_value_rows_from(parents: Vec<DataFileValue>) -> Vec<FilePartitionValueRow> {
parents
.into_iter()
.flat_map(|file| {
let (data_file_id, table_id) = (file.data_file_id, file.table_id);
file.partition_values
.into_iter()
.map(move |value| FilePartitionValueRow {
data_file_id,
table_id,
partition_key_index: value.partition_key_index,
partition_value: value.partition_value,
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_file_partition_value_rows(
catalog: &Catalog,
) -> Result<Vec<FilePartitionValueRow>> {
Ok(file_partition_value_rows_from(
dump_data_files(catalog).await?,
))
}
#[doc(hidden)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SortExpressionRow {
pub sort_id: u64,
pub table_id: u64,
pub sort_key_index: u64,
pub expression: String,
pub dialect: String,
pub sort_direction: String,
pub null_order: String,
}
#[doc(hidden)]
#[must_use]
pub fn sort_expression_rows_from(parents: Vec<SortValue>) -> Vec<SortExpressionRow> {
parents
.into_iter()
.flat_map(|spec| {
let (sort_id, table_id) = (spec.sort_id, spec.table_id);
spec.expressions
.into_iter()
.map(move |expression| SortExpressionRow {
sort_id,
table_id,
sort_key_index: expression.sort_key_index,
expression: expression.expression,
dialect: expression.dialect,
sort_direction: expression.sort_direction,
null_order: expression.null_order,
})
})
.collect()
}
#[doc(hidden)]
pub async fn dump_sort_expression_rows(catalog: &Catalog) -> Result<Vec<SortExpressionRow>> {
Ok(sort_expression_rows_from(dump_sort_info(catalog).await?))
}
#[cfg(test)]
mod tests {
use object_store::memory::InMemory;
use super::*;
use crate::catalog::{
Catalog, CatalogOptions, ColumnDef, ColumnStats, DataFile, DeleteFile, FileColumnStats,
MacroImplementationDef, MacroParameterDef,
};
async fn seed() -> Catalog {
let catalog = Catalog::open(Arc::new(InMemory::new()), CatalogOptions::default())
.await
.unwrap();
catalog
.commit(|tx| {
let schema = tx.create_schema("sales")?;
let table = tx.create_table(
schema,
"orders",
&[
ColumnDef {
name: "id".into(),
column_type: "BIGINT".into(),
nulls_allowed: false,
default_value: None,
children: Vec::new(),
},
ColumnDef {
name: "amount".into(),
column_type: "DOUBLE".into(),
nulls_allowed: true,
default_value: None,
children: Vec::new(),
},
],
)?;
let column = tx.columns_of(table)[0].id;
let file = tx.register_data_file(
table,
DataFile {
path: "orders/data-1.parquet".into(),
path_is_relative: true,
file_format: "parquet".into(),
record_count: 10,
file_size_bytes: 1024,
footer_size: 64,
encryption_key: None,
partition_values: vec![],
column_stats: vec![FileColumnStats {
column_id: column,
column_size_bytes: 100,
value_count: 10,
null_count: 0,
min_value: Some("1".into()),
max_value: Some("10".into()),
contains_nan: None,
extra_stats: None,
}],
},
&[],
)?;
tx.register_delete_file(
table,
DeleteFile {
data_file_id: file,
path: "orders/delete-1.parquet".into(),
path_is_relative: true,
format: "parquet".into(),
delete_count: 2,
file_size_bytes: 128,
footer_size: 32,
encryption_key: None,
},
&[],
)?;
tx.update_column_stats(
table,
column,
ColumnStats {
contains_null: Some(false),
contains_nan: None,
min_value: Some("1".into()),
max_value: Some("10".into()),
extra_stats: None,
},
)?;
tx.create_view(schema, "orders_v", "duckdb", "select * from orders")?;
Ok(())
})
.await
.unwrap();
catalog
.commit(|tx| {
let table = tx.tables_in(tx.schemas()[1].id)[0].id;
tx.rename_table(table, "orders2")
})
.await
.unwrap();
catalog
}
#[tokio::test]
async fn unversioned_dumps_serve_live_rows_on_a_history_bearing_catalog() {
let catalog = seed().await;
let stats = dump_table_stats(&catalog).await.unwrap();
assert_eq!(stats.len(), 1);
assert_eq!(stats[0].record_count, 10);
let column_stats = dump_table_column_stats(&catalog).await.unwrap();
assert_eq!(column_stats.len(), 1);
assert_eq!(column_stats[0].min_value.as_deref(), Some("1"));
let file_stats = dump_file_column_stats(&catalog).await.unwrap();
assert_eq!(file_stats.len(), 1);
let deletions = dump_scheduled_deletions(&catalog).await.unwrap();
assert!(deletions.is_empty());
}
async fn plant_migration_marker(catalog: &Catalog) {
use crate::store::{
key::{Key, SysKey},
proto::MigrationValue,
value::encode_value,
};
let tx = catalog.begin_write_tx().await.unwrap();
tx.put(
Key::Sys(SysKey::Migration).encode(),
encode_value(&MigrationValue {
from_format: 1,
to_format: 2,
cursor: Vec::new(),
}),
)
.unwrap();
tx.commit_with_options(&crate::transaction::commit::durable())
.await
.unwrap();
}
fn refuses<T>(outcome: &Result<T>) -> bool {
matches!(outcome, Err(crate::error::Error::Migration(_)))
}
#[tokio::test]
async fn a_planted_marker_refuses_every_read_seam() {
let catalog = seed().await;
plant_migration_marker(&catalog).await;
assert!(refuses(&catalog.snapshot().await), "snapshot");
assert!(refuses(&dump_schemas(&catalog).await), "dump_schemas");
assert!(refuses(&dump_mappings(&catalog).await), "dump_mappings");
assert!(
refuses(&dump_table_stats(&catalog).await),
"dump_table_stats"
);
assert!(refuses(&dump_snapshots(&catalog).await), "dump_snapshots");
assert!(
refuses(&inline::scan_inline(&catalog, 1, inline::InlineScanKind::Table, 1, 0).await),
"scan_inline"
);
assert!(
refuses(&inline::inline_schemas(&catalog, 1).await),
"inline_schemas"
);
assert!(
refuses(&inline::inline_registered_tables(&catalog).await),
"inline_registered_tables"
);
assert!(
refuses(&inline::inline_file_delete_table_exists(&catalog, 1).await),
"inline_file_delete_table_exists"
);
assert!(
refuses(&inline::inline_file_deletes(&catalog, 1).await),
"inline_file_deletes"
);
}
async fn expire_snapshots_below_head(catalog: &Catalog, records: bool) {
use crate::transaction::staged::{Cell, RowOperation, TableKind};
let head = catalog.snapshot().await.unwrap().current_snapshot().id;
let mut tx = staged::staged_begin(catalog, None, String::new())
.await
.unwrap();
for snapshot in dump_snapshots(catalog).await.unwrap() {
if snapshot.snapshot_id == head.get() {
continue;
}
tx.stage(RowOperation::Delete {
table: TableKind::Snapshot,
cells: vec![Cell::U64(snapshot.snapshot_id)],
});
}
if records {
for row in dump_schema_versions(catalog).await.unwrap() {
tx.stage(RowOperation::Delete {
table: TableKind::SchemaVersions,
cells: vec![
Cell::U64(row.begin_snapshot),
Cell::U64(row.schema_version),
Cell::U64(row.table_id),
],
});
}
}
tx.commit().await.unwrap();
}
#[tokio::test]
async fn schema_version_rows_outlive_the_snapshots_that_wrote_them() {
let catalog = seed().await;
let before = dump_schema_versions(&catalog).await.unwrap();
assert!(
before
.iter()
.any(|row| row.begin_snapshot == 1 && row.schema_version == 1),
"the creating commit's rows must be recorded: {before:?}"
);
expire_snapshots_below_head(&catalog, false).await;
assert_eq!(dump_snapshots(&catalog).await.unwrap().len(), 1);
assert_eq!(dump_schema_versions(&catalog).await.unwrap(), before);
}
#[tokio::test]
async fn a_store_whose_schema_version_rows_are_gone_gets_a_floor_row() {
let catalog = seed().await;
let file = dump_data_files(&catalog).await.unwrap()[0].clone();
assert_eq!(file.begin_snapshot, 1);
expire_snapshots_below_head(&catalog, true).await;
assert_eq!(
dump_schema_versions(&catalog).await.unwrap(),
vec![
SchemaVersionRow {
begin_snapshot: file.begin_snapshot,
schema_version: 2,
table_id: file.table_id,
},
SchemaVersionRow {
begin_snapshot: 2,
schema_version: 2,
table_id: file.table_id,
},
]
);
}
#[tokio::test]
async fn dump_schemas_returns_bootstrap_and_seeded_schemas_with_no_history() {
let catalog = seed().await;
let rows = dump_schemas(&catalog).await.unwrap();
assert_eq!(rows.len(), 2);
assert!(rows.iter().all(|r| r.end_snapshot.is_none()));
let names: Vec<&str> = rows.iter().map(|r| r.schema_name.as_str()).collect();
assert_eq!(names, vec!["main", "sales"]);
}
#[tokio::test]
async fn dump_tables_returns_both_versions_of_a_renamed_table() {
let catalog = seed().await;
let rows = dump_tables(&catalog).await.unwrap();
assert_eq!(
rows.len(),
2,
"rename must yield exactly one current + one history row"
);
let ended = rows.iter().find(|r| r.end_snapshot.is_some()).unwrap();
let live = rows.iter().find(|r| r.end_snapshot.is_none()).unwrap();
assert_eq!(ended.table_name, "orders");
assert_eq!(live.table_name, "orders2");
assert_eq!(ended.table_id, live.table_id);
assert_eq!(ended.table_uuid, live.table_uuid);
assert_eq!(ended.end_snapshot, Some(live.begin_snapshot));
assert!(live.begin_snapshot > ended.begin_snapshot);
}
#[tokio::test]
async fn dump_columns_and_views_are_row_faithful() {
let catalog = seed().await;
let columns = dump_columns(&catalog).await.unwrap();
assert_eq!(columns.len(), 2);
assert!(columns.iter().all(|c| c.end_snapshot.is_none()));
let mut names: Vec<&str> = columns.iter().map(|c| c.column_name.as_str()).collect();
names.sort_unstable();
assert_eq!(names, vec!["amount", "id"]);
let views = dump_views(&catalog).await.unwrap();
assert_eq!(views.len(), 1);
assert_eq!(views[0].sql, "select * from orders");
assert!(views[0].end_snapshot.is_none());
}
#[tokio::test]
async fn dump_mappings_serves_embedded_rows() {
use crate::transaction::staged::{Cell, RowOperation, StagedTransaction, TableKind};
let catalog = Catalog::open(Arc::new(InMemory::new()), CatalogOptions::default())
.await
.unwrap();
let db_tx = catalog.begin_write_tx().await.unwrap();
let mut tx = StagedTransaction::begin_detached(db_tx);
tx.stage(RowOperation::Insert {
table: TableKind::ColumnMapping,
cells: vec![Cell::U64(21), Cell::U64(1), Cell::Str("map_by_name".into())],
});
tx.stage(RowOperation::Insert {
table: TableKind::NameMapping,
cells: vec![
Cell::U64(21),
Cell::U64(0),
Cell::Str("id".into()),
Cell::U64(1),
Cell::Null,
Cell::Bool(false),
],
});
tx.stage(RowOperation::Insert {
table: TableKind::Snapshot,
cells: vec![
Cell::U64(1),
Cell::I64(1),
Cell::U64(1),
Cell::U64(11),
Cell::U64(22),
],
});
tx.stage(RowOperation::Insert {
table: TableKind::SnapshotChanges,
cells: vec![
Cell::U64(1),
Cell::Str("inserted_into_table:1".into()),
Cell::Null,
Cell::Null,
Cell::Null,
],
});
tx.commit().await.unwrap();
let mappings = dump_mappings(&catalog).await.unwrap();
assert_eq!(mappings.len(), 1);
assert_eq!(mappings[0].mapping_id, 21);
assert_eq!(mappings[0].table_id, 1);
assert_eq!(mappings[0].map_type, "map_by_name");
assert_eq!(mappings[0].name_mappings.len(), 1);
assert_eq!(mappings[0].name_mappings[0].source_name, "id");
}
#[tokio::test]
async fn dump_macros_serves_children_current_and_history() {
let catalog = Catalog::open(Arc::new(InMemory::new()), CatalogOptions::default())
.await
.unwrap();
catalog
.commit(|tx| {
let schema = tx.create_schema("s")?;
tx.create_macro(
schema,
"add",
&[
MacroImplementationDef {
dialect: "duckdb".into(),
sql: "(a + 1)".into(),
macro_type: "scalar".into(),
parameters: vec![MacroParameterDef {
name: "a".into(),
parameter_type: "unknown".into(),
default_value: None,
default_value_type: "unknown".into(),
}],
},
MacroImplementationDef {
dialect: "duckdb".into(),
sql: "(a + b)".into(),
macro_type: "scalar".into(),
parameters: vec![
MacroParameterDef {
name: "a".into(),
parameter_type: "unknown".into(),
default_value: None,
default_value_type: "unknown".into(),
},
MacroParameterDef {
name: "b".into(),
parameter_type: "unknown".into(),
default_value: Some("5".into()),
default_value_type: "int32".into(),
},
],
},
],
)?;
Ok(())
})
.await
.unwrap();
let head = catalog.snapshot().await.unwrap();
let schema = head.schema_by_name("s").unwrap();
let created = head.macro_by_name(schema.id, "add").unwrap();
catalog
.commit(move |tx| tx.drop_macro(created.id))
.await
.unwrap();
let macros = dump_macros(&catalog).await.unwrap();
assert_eq!(macros.len(), 1);
let ended = ¯os[0];
assert!(ended.end_snapshot.is_some());
assert_eq!(ended.implementations.len(), 2);
assert_eq!(ended.implementations[0].impl_id, 0);
assert_eq!(ended.implementations[1].parameters[1].parameter_name, "b");
assert_eq!(
ended.implementations[1].parameters[1]
.default_value
.as_deref(),
Some("5")
);
}
#[tokio::test]
async fn dump_data_and_delete_files_carry_registration_values_verbatim() {
let catalog = seed().await;
let files = dump_data_files(&catalog).await.unwrap();
assert_eq!(files.len(), 1);
assert_eq!(files[0].path, "orders/data-1.parquet");
assert_eq!(files[0].record_count, 10);
assert_eq!(files[0].row_id_start, Some(0));
assert!(files[0].end_snapshot.is_none());
let deletes = dump_delete_files(&catalog).await.unwrap();
assert_eq!(deletes.len(), 1);
assert_eq!(deletes[0].path, "orders/delete-1.parquet");
assert_eq!(deletes[0].data_file_id, files[0].data_file_id);
assert_eq!(deletes[0].delete_count, 2);
}
#[tokio::test]
async fn dump_statistics_kinds_are_unversioned_single_rows() {
let catalog = seed().await;
let table_rows = dump_table_stats(&catalog).await.unwrap();
assert_eq!(table_rows.len(), 1);
assert_eq!(table_rows[0].record_count, 10);
let table_col_rows = dump_table_column_stats(&catalog).await.unwrap();
assert_eq!(table_col_rows.len(), 1);
assert_eq!(table_col_rows[0].contains_null, Some(false));
let file_col_rows = dump_file_column_stats(&catalog).await.unwrap();
assert_eq!(file_col_rows.len(), 1);
assert_eq!(file_col_rows[0].min_value.as_deref(), Some("1"));
assert_eq!(file_col_rows[0].max_value.as_deref(), Some("10"));
}
#[tokio::test]
async fn dump_snapshots_returns_every_committed_snapshot_in_order() {
let catalog = seed().await;
let rows = dump_snapshots(&catalog).await.unwrap();
assert_eq!(rows.len(), 3);
let ids: Vec<u64> = rows.iter().map(|r| r.snapshot_id).collect();
assert_eq!(ids, vec![0, 1, 2]);
assert_eq!(rows[0].changes_made, "created_schema:\"main\"");
assert!(!rows[1].changes_made.is_empty());
assert!(!rows[2].changes_made.is_empty());
}
}