use crate::Result;
use crate::error::{TypeChangeOperation, TypeChangeWriteMode};
use crate::maintenance::{
CleanupCriteria, ExpireCriteria, ExpiredSnapshot, ScheduledFile, format_sql_timestamp,
};
use crate::metadata_provider::block_on;
use crate::metadata_writer::{
ColumnDef, ColumnStat, CommitIds, DataFileInfo, DeleteFileEntry, DeleteFileInfo,
MetadataWriter, WriteMode, WriteSetupResult, columns_differ, validate_delete_entries,
validate_name,
};
use sqlx::Row;
use sqlx::sqlite::{SqlitePool, SqlitePoolOptions};
const DEFAULT_MAX_CONNECTIONS: u32 = 5;
fn id_list(ids: &[i64]) -> String {
ids.iter()
.map(|id| id.to_string())
.collect::<Vec<_>>()
.join(", ")
}
const SQL_CREATE_SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS ducklake_metadata (
key VARCHAR NOT NULL,
value VARCHAR NOT NULL,
scope VARCHAR
);
-- `schema_version` is the per-catalog monotonic schema counter from the DuckLake
-- spec (`ducklake_metadata_manager.cpp:232`): it bumps on a schema change (DDL)
-- and is carried forward unchanged on a data write, exactly mirroring upstream's
-- `if (SchemaChangesMade()) schema_version++` (`ducklake_transaction_state.cpp:1826`)
-- and the validated Postgres writer here. Upstream also stores `next_catalog_id` /
-- `next_file_id` on this row as ITS id allocators; we deliberately omit them (as the
-- Postgres writer does) because this library allocates ids from its own counters
-- (the `next_column_id` metadata row + autoincrement PKs), never from the snapshot.
CREATE TABLE IF NOT EXISTS ducklake_snapshot (
snapshot_id INTEGER PRIMARY KEY,
snapshot_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
schema_version INTEGER NOT NULL DEFAULT 0
);
-- Per-table schema-change ledger (DuckLake spec, `ducklake_metadata_manager.cpp:281`):
-- one row per (table that changed schema, snapshot at which it changed). Written on
-- every DDL that leaves the table live (create, column add/remove/reorder, type
-- promotion); NOT written for a drop (the table has no live schema afterward). Rows
-- are never tombstoned (no `end_snapshot`); they go unreferenced once the table is
-- fully expired and are reclaimed by vacuum — same as upstream `DropTables`.
CREATE TABLE IF NOT EXISTS ducklake_schema_versions (
begin_snapshot INTEGER NOT NULL,
schema_version INTEGER NOT NULL,
table_id INTEGER NOT NULL,
UNIQUE (table_id, begin_snapshot)
);
CREATE TABLE IF NOT EXISTS ducklake_schema (
schema_id INTEGER PRIMARY KEY,
schema_name VARCHAR NOT NULL,
path VARCHAR NOT NULL DEFAULT '',
path_is_relative BOOLEAN NOT NULL DEFAULT 1,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER
);
CREATE TABLE IF NOT EXISTS ducklake_table (
table_id INTEGER PRIMARY KEY,
schema_id INTEGER NOT NULL,
table_name VARCHAR NOT NULL,
path VARCHAR NOT NULL DEFAULT '',
path_is_relative BOOLEAN NOT NULL DEFAULT 1,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER
);
-- Faithful match of upstream `duckdb/ducklake` `ducklake_column`
-- (`ducklake_metadata_manager.cpp:258`): a BARE table — no PRIMARY KEY, no
-- NOT NULL, no DEFAULT — in upstream's exact column order. `column_id` is
-- deliberately NOT a single-row PK: a column is versioned by
-- `[begin_snapshot, end_snapshot)` and a stable-id type promotion writes a
-- SECOND row with the same `column_id`, which a single-row PK would forbid.
-- One-live-row / non-null / ownership invariants are enforced in the writer
-- code + tests, exactly as upstream guarantees them (its table has no such
-- constraints). The four `*default*` columns + `parent_column` are projected by
-- DuckDB when it reads catalogs we produce; we leave them NULL (no nested-type
-- or column-default writes yet). See docs/column-id-versioning-design.md §4.1.
CREATE TABLE IF NOT EXISTS ducklake_column (
column_id BIGINT,
begin_snapshot BIGINT,
end_snapshot BIGINT,
table_id BIGINT,
column_order BIGINT,
column_name VARCHAR,
column_type VARCHAR,
initial_default VARCHAR,
default_value VARCHAR,
nulls_allowed BOOLEAN,
parent_column BIGINT,
default_value_type VARCHAR,
default_value_dialect VARCHAR
);
-- `partial_max` (DuckLake v1.0) marks a *partial data file* produced by
-- `merge_adjacent_files`: it is the maximum origin snapshot id among the file's
-- merged rows, whose per-row origin is stored in the embedded
-- `_ducklake_internal_snapshot_id` parquet column. NULL for ordinary files
-- (single-origin appends, and `rewrite_data_files` outputs).
CREATE TABLE IF NOT EXISTS ducklake_data_file (
data_file_id INTEGER PRIMARY KEY,
table_id INTEGER NOT NULL,
path VARCHAR NOT NULL,
path_is_relative BOOLEAN NOT NULL DEFAULT 1,
file_size_bytes INTEGER NOT NULL,
footer_size INTEGER,
encryption_key VARCHAR,
record_count INTEGER,
row_id_start INTEGER,
mapping_id INTEGER,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER,
partial_max INTEGER
);
-- Per-snapshot change ledger (DuckLake spec). One row per snapshot recording
-- what the commit did, as a comma-separated `changes_made` string (e.g.
-- `compacted_table:<table_id>` for a compaction). Written by the compaction
-- commit so DuckDB and other spec readers can attribute the snapshot; other
-- commit paths in this crate do not populate it yet.
CREATE TABLE IF NOT EXISTS ducklake_snapshot_changes (
snapshot_id INTEGER PRIMARY KEY,
changes_made VARCHAR NOT NULL,
author VARCHAR,
commit_message VARCHAR,
commit_extra_info VARCHAR
);
-- Per-table row-lineage counter (DuckLake spec). `next_row_id` is the
-- monotonic rowid allocator: a new data file gets its `row_id_start` from
-- the current value, then we advance by `record_count` in the same
-- transaction. `record_count` and `file_size_bytes` mirror the currently-
-- visible totals so DuckDB's `ducklake_table_info` aggregate sees correct
-- numbers for tables we wrote.
CREATE TABLE IF NOT EXISTS ducklake_table_stats (
table_id INTEGER PRIMARY KEY,
record_count INTEGER NOT NULL DEFAULT 0,
next_row_id INTEGER NOT NULL DEFAULT 0,
file_size_bytes INTEGER NOT NULL DEFAULT 0
);
-- Per-file, per-column statistics (DuckLake spec zone maps). Powers file-level
-- pruning: min/max are the DuckDB-canonical VARCHAR encoding of the bounds,
-- value_count is the non-null count, null_count the null count. Column set
-- mirrors the official extension exactly (no PK, matching upstream). A brand-
-- new table for pre-existing catalogs: `CREATE TABLE IF NOT EXISTS` (re-run by
-- initialize_schema on every open) creates it, which is also what lets DuckDB
-- attach a catalog we wrote (its global-stats query joins the two stats tables).
CREATE TABLE IF NOT EXISTS ducklake_file_column_stats (
data_file_id INTEGER NOT NULL,
table_id INTEGER NOT NULL,
column_id INTEGER NOT NULL,
column_size_bytes INTEGER,
value_count INTEGER,
null_count INTEGER,
min_value VARCHAR,
max_value VARCHAR,
contains_nan BOOLEAN,
extra_stats VARCHAR
);
-- Table-wide per-column roll-up (DuckLake spec). Feeds the query optimizer
-- (cardinality/column bounds), not file pruning. `contains_null` is a boolean
-- (any null across the table), not a count; min/max widen across all files.
CREATE TABLE IF NOT EXISTS ducklake_table_column_stats (
table_id INTEGER NOT NULL,
column_id INTEGER NOT NULL,
contains_null BOOLEAN,
contains_nan BOOLEAN,
min_value VARCHAR,
max_value VARCHAR,
extra_stats VARCHAR
);
CREATE TABLE IF NOT EXISTS ducklake_delete_file (
delete_file_id INTEGER PRIMARY KEY,
data_file_id INTEGER NOT NULL,
table_id INTEGER NOT NULL,
path VARCHAR NOT NULL,
path_is_relative BOOLEAN NOT NULL DEFAULT 1,
file_size_bytes INTEGER NOT NULL,
footer_size INTEGER,
encryption_key VARCHAR,
delete_count INTEGER,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER
);
-- Files queued for physical deletion by the two-phase vacuum (DuckLake spec).
-- `expire_snapshots` GCs unreachable catalog rows and records the orphaned
-- physical paths here; `cleanup_old_files` deletes the objects and removes
-- these rows. `path` is stored relative to the catalog `data_path` root
-- (i.e. already resolved through schema/table) so cleanup needs only a
-- single-level join with `data_path`. Mirrors the upstream
-- `ducklake_files_scheduled_for_deletion` table.
CREATE TABLE IF NOT EXISTS ducklake_files_scheduled_for_deletion (
data_file_id INTEGER NOT NULL,
path VARCHAR NOT NULL,
path_is_relative BOOLEAN NOT NULL DEFAULT 1,
schedule_start TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
"#;
#[derive(Debug, Clone)]
pub struct SqliteMetadataWriter {
pool: SqlitePool,
}
impl SqliteMetadataWriter {
pub async fn new(connection_string: &str) -> Result<Self> {
Self::with_max_connections(connection_string, DEFAULT_MAX_CONNECTIONS).await
}
pub async fn with_max_connections(
connection_string: &str,
max_connections: u32,
) -> Result<Self> {
let pool = SqlitePoolOptions::new()
.max_connections(max_connections)
.connect(connection_string)
.await?;
Ok(Self {
pool,
})
}
pub async fn new_with_init(connection_string: &str) -> Result<Self> {
let writer = Self::new(connection_string).await?;
writer.initialize_schema()?;
Ok(writer)
}
pub fn drop_table(&self, schema_name: &str, table_name: &str) -> Result<bool> {
validate_name(schema_name, "Schema")?;
validate_name(table_name, "Table")?;
block_on(async {
let mut tx = self.pool.begin().await?;
let (drop_snapshot, _schema_version) = insert_snapshot(&mut tx).await?;
let table_id: i64 = match sqlx::query(
"SELECT t.table_id FROM ducklake_table t
JOIN ducklake_schema s ON s.schema_id = t.schema_id
WHERE s.schema_name = ? AND s.end_snapshot IS NULL
AND t.table_name = ? AND t.end_snapshot IS NULL",
)
.bind(schema_name)
.bind(table_name)
.fetch_optional(&mut *tx)
.await?
{
Some(r) => r.try_get(0)?,
None => return Ok(false),
};
bump_schema_version(&mut tx, drop_snapshot).await?;
for child in
["ducklake_table", "ducklake_column", "ducklake_data_file", "ducklake_delete_file"]
{
sqlx::query(&format!(
"UPDATE {child} SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL"
))
.bind(drop_snapshot)
.bind(table_id)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(true)
})
}
pub fn expire_snapshots(&self, criteria: ExpireCriteria) -> Result<Vec<ExpiredSnapshot>> {
block_on(async {
let mut tx = self.pool.begin().await?;
let most_recent: Option<i64> =
sqlx::query("SELECT MAX(snapshot_id) FROM ducklake_snapshot")
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
let most_recent = match most_recent {
Some(id) => id,
None => {
tx.commit().await?;
return Ok(Vec::new());
},
};
let candidates: Vec<ExpiredSnapshot> = match &criteria {
ExpireCriteria::Versions(versions) => {
let ids: Vec<i64> = versions
.iter()
.copied()
.filter(|&v| v != most_recent)
.collect();
if ids.is_empty() {
tx.commit().await?;
return Ok(Vec::new());
}
let rows = sqlx::query(&format!(
"SELECT snapshot_id, snapshot_time FROM ducklake_snapshot
WHERE snapshot_id IN ({}) ORDER BY snapshot_id",
id_list(&ids)
))
.fetch_all(&mut *tx)
.await?;
rows_to_snapshots(rows)?
},
ExpireCriteria::OlderThan(ts) => {
let rows = sqlx::query(
"SELECT snapshot_id, snapshot_time FROM ducklake_snapshot
WHERE snapshot_id != ? AND snapshot_time < ? ORDER BY snapshot_id",
)
.bind(most_recent)
.bind(format_sql_timestamp(ts))
.fetch_all(&mut *tx)
.await?;
rows_to_snapshots(rows)?
},
};
if candidates.is_empty() {
tx.commit().await?;
return Ok(Vec::new());
}
let expire_ids: Vec<i64> = candidates.iter().map(|s| s.snapshot_id).collect();
sqlx::query(&format!(
"DELETE FROM ducklake_snapshot WHERE snapshot_id IN ({})",
id_list(&expire_ids)
))
.execute(&mut *tx)
.await?;
let dead_tables: Vec<i64> = sqlx::query(
"SELECT t.table_id FROM ducklake_table t
WHERE t.end_snapshot IS NOT NULL AND NOT EXISTS (
SELECT 1 FROM ducklake_snapshot
WHERE snapshot_id >= t.begin_snapshot AND snapshot_id < t.end_snapshot)
AND NOT EXISTS (
SELECT 1 FROM ducklake_table t2
WHERE t2.table_id = t.table_id
AND (t2.end_snapshot IS NULL OR EXISTS (
SELECT 1 FROM ducklake_snapshot
WHERE snapshot_id >= t2.begin_snapshot
AND snapshot_id < t2.end_snapshot)))",
)
.fetch_all(&mut *tx)
.await?
.into_iter()
.map(|r| r.try_get::<i64, _>(0))
.collect::<std::result::Result<_, _>>()?;
let dead_table_filter = if dead_tables.is_empty() {
"0".to_string()
} else {
format!("df.table_id IN ({})", id_list(&dead_tables))
};
let dead_data_files = sqlx::query(&format!(
"SELECT df.data_file_id, {RESOLVED_PATH} AS resolved_path, {REL_FLAG} AS rel
FROM ducklake_data_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
WHERE ({dead_table_filter}) OR (df.end_snapshot IS NOT NULL AND NOT EXISTS (
SELECT 1 FROM ducklake_snapshot
WHERE snapshot_id >= df.begin_snapshot AND snapshot_id < df.end_snapshot))"
))
.fetch_all(&mut *tx)
.await?;
let data_file_ids = schedule_files(&mut tx, dead_data_files).await?;
if !data_file_ids.is_empty() {
sqlx::query(&format!(
"DELETE FROM ducklake_data_file WHERE data_file_id IN ({})",
id_list(&data_file_ids)
))
.execute(&mut *tx)
.await?;
}
let dead_data_filter = if data_file_ids.is_empty() {
"0".to_string()
} else {
format!("df.data_file_id IN ({})", id_list(&data_file_ids))
};
let dead_delete_table_filter = if dead_tables.is_empty() {
"0".to_string()
} else {
format!("df.table_id IN ({})", id_list(&dead_tables))
};
let dead_delete_files = sqlx::query(&format!(
"SELECT df.delete_file_id, {RESOLVED_PATH} AS resolved_path, {REL_FLAG} AS rel
FROM ducklake_delete_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
WHERE ({dead_data_filter}) OR ({dead_delete_table_filter})
OR (df.end_snapshot IS NOT NULL AND NOT EXISTS (
SELECT 1 FROM ducklake_snapshot
WHERE snapshot_id >= df.begin_snapshot AND snapshot_id < df.end_snapshot))"
))
.fetch_all(&mut *tx)
.await?;
let delete_file_ids = schedule_files(&mut tx, dead_delete_files).await?;
if !delete_file_ids.is_empty() {
sqlx::query(&format!(
"DELETE FROM ducklake_delete_file WHERE delete_file_id IN ({})",
id_list(&delete_file_ids)
))
.execute(&mut *tx)
.await?;
}
if !dead_tables.is_empty() {
let dead = id_list(&dead_tables);
for table in [
"ducklake_table",
"ducklake_table_stats",
"ducklake_column",
"ducklake_schema_versions",
] {
sqlx::query(&format!("DELETE FROM {table} WHERE table_id IN ({dead})"))
.execute(&mut *tx)
.await?;
}
}
sqlx::query(
"DELETE FROM ducklake_schema
WHERE end_snapshot IS NOT NULL AND NOT EXISTS (
SELECT 1 FROM ducklake_snapshot
WHERE snapshot_id >= ducklake_schema.begin_snapshot
AND snapshot_id < ducklake_schema.end_snapshot)",
)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(candidates)
})
}
pub(crate) fn list_scheduled_for_deletion(
&self,
criteria: &CleanupCriteria,
) -> Result<Vec<ScheduledFile>> {
block_on(async {
let rows = match criteria {
CleanupCriteria::All => {
sqlx::query(
"SELECT data_file_id, path, path_is_relative
FROM ducklake_files_scheduled_for_deletion",
)
.fetch_all(&self.pool)
.await?
},
CleanupCriteria::OlderThan(ts) => {
sqlx::query(
"SELECT data_file_id, path, path_is_relative
FROM ducklake_files_scheduled_for_deletion
WHERE schedule_start < ?",
)
.bind(format_sql_timestamp(ts))
.fetch_all(&self.pool)
.await?
},
};
rows.into_iter()
.map(|r| {
Ok(ScheduledFile {
data_file_id: r.try_get(0)?,
path: r.try_get(1)?,
path_is_relative: r.try_get::<i64, _>(2)? != 0,
})
})
.collect()
})
}
pub(crate) fn remove_scheduled(&self, ids: &[i64]) -> Result<()> {
if ids.is_empty() {
return Ok(());
}
block_on(async {
sqlx::query(&format!(
"DELETE FROM ducklake_files_scheduled_for_deletion WHERE data_file_id IN ({})",
id_list(ids)
))
.execute(&self.pool)
.await?;
Ok(())
})
}
pub(crate) fn list_referenced_paths(&self) -> Result<Vec<(String, bool)>> {
block_on(async {
let q = format!(
"SELECT {RESOLVED_PATH} AS p, {REL_FLAG} AS rel
FROM ducklake_data_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
UNION ALL
SELECT {RESOLVED_PATH} AS p, {REL_FLAG} AS rel
FROM ducklake_delete_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
UNION ALL
SELECT path AS p, CAST(path_is_relative AS INTEGER) AS rel
FROM ducklake_files_scheduled_for_deletion"
);
let rows = sqlx::query(&q).fetch_all(&self.pool).await?;
rows.into_iter()
.map(|r| {
let p: String = r.try_get(0)?;
let rel: i64 = r.try_get(1)?;
Ok((p, rel != 0))
})
.collect()
})
}
}
const RESOLVED_PATH: &str = "CASE
WHEN NOT df.path_is_relative THEN df.path
WHEN NOT t.path_is_relative THEN t.path || '/' || df.path
ELSE s.path || '/' || t.path || '/' || df.path
END";
const REL_FLAG: &str =
"(CASE WHEN df.path_is_relative AND t.path_is_relative AND s.path_is_relative
THEN 1 ELSE 0 END)";
fn rows_to_snapshots(rows: Vec<sqlx::sqlite::SqliteRow>) -> Result<Vec<ExpiredSnapshot>> {
rows.into_iter()
.map(|r| {
Ok(ExpiredSnapshot {
snapshot_id: r.try_get(0)?,
snapshot_time: r.try_get(1)?,
})
})
.collect()
}
async fn schedule_files(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
rows: Vec<sqlx::sqlite::SqliteRow>,
) -> Result<Vec<i64>> {
let mut ids = Vec::with_capacity(rows.len());
for row in rows {
let id: i64 = row.try_get(0)?;
let path: String = row.try_get(1)?;
let rel: i64 = row.try_get(2)?;
sqlx::query(
"INSERT INTO ducklake_files_scheduled_for_deletion
(data_file_id, path, path_is_relative, schedule_start)
VALUES (?, ?, ?, CURRENT_TIMESTAMP)",
)
.bind(id)
.bind(&path)
.bind(rel)
.execute(&mut **tx)
.await?;
ids.push(id);
}
Ok(ids)
}
async fn reserve_ids(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
key: &str,
n: i64,
) -> Result<i64> {
let last: i64 = sqlx::query(
"UPDATE ducklake_metadata
SET value = CAST(CAST(value AS INTEGER) + ? AS TEXT)
WHERE key = ? AND scope IS NULL
RETURNING CAST(value AS INTEGER)",
)
.bind(n)
.bind(key)
.fetch_one(&mut **tx)
.await?
.try_get(0)?;
Ok(last)
}
async fn migrate_ducklake_column_drop_pk(pool: &SqlitePool) -> Result<()> {
let info = sqlx::query("SELECT name, pk FROM pragma_table_info('ducklake_column')")
.fetch_all(pool)
.await?;
let mut legacy_cols: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut column_id_is_pk = false;
for row in &info {
let name: String = row.try_get("name")?;
let pk: i64 = row.try_get("pk")?;
if name == "column_id" && pk != 0 {
column_id_is_pk = true;
}
legacy_cols.insert(name);
}
if !column_id_is_pk {
return Ok(());
}
const TARGET: &[&str] = &[
"column_id",
"begin_snapshot",
"end_snapshot",
"table_id",
"column_order",
"column_name",
"column_type",
"initial_default",
"default_value",
"nulls_allowed",
"parent_column",
"default_value_type",
"default_value_dialect",
];
let select_list = TARGET
.iter()
.map(|c| {
if legacy_cols.contains(*c) {
(*c).to_string()
} else {
"NULL".to_string()
}
})
.collect::<Vec<_>>()
.join(", ");
let insert_list = TARGET.join(", ");
let mut tx = pool.begin().await?;
sqlx::query(
"CREATE TABLE ducklake_column__migrate (
column_id BIGINT,
begin_snapshot BIGINT,
end_snapshot BIGINT,
table_id BIGINT,
column_order BIGINT,
column_name VARCHAR,
column_type VARCHAR,
initial_default VARCHAR,
default_value VARCHAR,
nulls_allowed BOOLEAN,
parent_column BIGINT,
default_value_type VARCHAR,
default_value_dialect VARCHAR
)",
)
.execute(&mut *tx)
.await?;
sqlx::query(&format!(
"INSERT INTO ducklake_column__migrate ({insert_list}) \
SELECT {select_list} FROM ducklake_column"
))
.execute(&mut *tx)
.await?;
sqlx::query("DROP TABLE ducklake_column")
.execute(&mut *tx)
.await?;
sqlx::query("ALTER TABLE ducklake_column__migrate RENAME TO ducklake_column")
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
async fn migrate_add_schema_version(pool: &SqlitePool) -> Result<()> {
let info = sqlx::query("SELECT name FROM pragma_table_info('ducklake_snapshot')")
.fetch_all(pool)
.await?;
let mut has_schema_version = false;
for row in &info {
let name: String = row.try_get("name")?;
if name == "schema_version" {
has_schema_version = true;
}
}
if !has_schema_version {
sqlx::query(
"ALTER TABLE ducklake_snapshot ADD COLUMN schema_version INTEGER NOT NULL DEFAULT 0",
)
.execute(pool)
.await?;
}
Ok(())
}
async fn migrate_add_partial_max(pool: &SqlitePool) -> Result<()> {
let info = sqlx::query("SELECT name FROM pragma_table_info('ducklake_data_file')")
.fetch_all(pool)
.await?;
let mut has_partial_max = false;
for row in &info {
let name: String = row.try_get("name")?;
if name == "partial_max" {
has_partial_max = true;
}
}
if !has_partial_max {
sqlx::query("ALTER TABLE ducklake_data_file ADD COLUMN partial_max INTEGER")
.execute(pool)
.await?;
}
Ok(())
}
async fn detect_replace_conflict(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table_id: i64,
base_snapshot: i64,
) -> Result<()> {
let conflict: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE table_id = ? AND (begin_snapshot > ? OR end_snapshot > ?)
LIMIT 1",
)
.bind(table_id)
.bind(base_snapshot)
.bind(base_snapshot)
.fetch_optional(&mut **tx)
.await?;
if conflict.is_some() {
return Err(crate::DuckLakeError::Conflict(format!(
"Replace on table {table_id} conflicts with a concurrent write committed since \
snapshot {base_snapshot}; aborting (retry the write against the new generation)"
)));
}
Ok(())
}
async fn retire_prior_generation(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table_id: i64,
snapshot_id: i64,
) -> Result<()> {
sqlx::query(
"UPDATE ducklake_data_file SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL AND begin_snapshot < ?",
)
.bind(snapshot_id)
.bind(table_id)
.bind(snapshot_id)
.execute(&mut **tx)
.await?;
sqlx::query(
"UPDATE ducklake_table_stats SET record_count = 0, file_size_bytes = 0 WHERE table_id = ?",
)
.bind(table_id)
.execute(&mut **tx)
.await?;
Ok(())
}
async fn insert_snapshot(tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>) -> Result<(i64, i64)> {
let row = sqlx::query(
"INSERT INTO ducklake_snapshot (snapshot_id, snapshot_time, schema_version)
SELECT COALESCE(MAX(snapshot_id), 0) + 1, CURRENT_TIMESTAMP,
COALESCE(MAX(schema_version), 0)
FROM ducklake_snapshot
RETURNING snapshot_id, schema_version",
)
.fetch_one(&mut **tx)
.await?;
Ok((row.try_get(0)?, row.try_get(1)?))
}
async fn bump_schema_version(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
snapshot_id: i64,
) -> Result<i64> {
let prev_max: i64 = sqlx::query_scalar(
"SELECT COALESCE(MAX(schema_version), 0) FROM ducklake_snapshot WHERE snapshot_id <> ?",
)
.bind(snapshot_id)
.fetch_one(&mut **tx)
.await?;
let new_version = prev_max + 1;
sqlx::query("UPDATE ducklake_snapshot SET schema_version = ? WHERE snapshot_id = ?")
.bind(new_version)
.bind(snapshot_id)
.execute(&mut **tx)
.await?;
Ok(new_version)
}
async fn record_schema_version(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
snapshot_id: i64,
schema_version: i64,
table_id: i64,
) -> Result<()> {
sqlx::query(
"INSERT INTO ducklake_schema_versions (begin_snapshot, schema_version, table_id)
VALUES (?, ?, ?)",
)
.bind(snapshot_id)
.bind(schema_version)
.bind(table_id)
.execute(&mut **tx)
.await?;
Ok(())
}
async fn insert_file_column_stats(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table_id: i64,
data_file_id: i64,
column_stats: &[ColumnStat],
) -> Result<()> {
for stat in column_stats {
sqlx::query(
"INSERT INTO ducklake_file_column_stats
(data_file_id, table_id, column_id, column_size_bytes,
value_count, null_count, min_value, max_value, contains_nan, extra_stats)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)",
)
.bind(data_file_id)
.bind(table_id)
.bind(stat.column_id)
.bind(stat.column_size_bytes)
.bind(stat.value_count)
.bind(stat.null_count)
.bind(stat.min_value.as_deref())
.bind(stat.max_value.as_deref())
.bind(stat.contains_nan)
.execute(&mut **tx)
.await?;
}
Ok(())
}
async fn recompute_table_column_stats(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table_id: i64,
columns: &[ColumnDef],
column_ids: &[i64],
) -> Result<()> {
use crate::stats_encode::{FileColumnStat, aggregate_global_column_stats};
let live_file_count: i64 = sqlx::query(
"SELECT COUNT(*) FROM ducklake_data_file WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(table_id)
.fetch_one(&mut **tx)
.await?
.try_get(0)?;
let mut per_file: Vec<FileColumnStat> = Vec::new();
for row in sqlx::query(
"SELECT s.column_id, s.min_value, s.max_value, s.null_count, s.contains_nan
FROM ducklake_file_column_stats s
JOIN ducklake_data_file d ON d.data_file_id = s.data_file_id
WHERE d.table_id = ? AND d.end_snapshot IS NULL",
)
.bind(table_id)
.fetch_all(&mut **tx)
.await?
{
per_file.push(FileColumnStat {
column_id: row.try_get(0)?,
min_value: row.try_get(1)?,
max_value: row.try_get(2)?,
null_count: row.try_get(3)?,
contains_nan: row.try_get(4)?,
});
}
let numeric_of = |column_id: i64| -> bool {
column_ids
.iter()
.position(|id| *id == column_id)
.and_then(|i| columns.get(i))
.map(|c| crate::stats_encode::is_numeric_ducklake_type(c.ducklake_type()))
.unwrap_or(false)
};
let globals = aggregate_global_column_stats(&per_file, live_file_count, numeric_of);
sqlx::query("DELETE FROM ducklake_table_column_stats WHERE table_id = ?")
.bind(table_id)
.execute(&mut **tx)
.await?;
for g in globals {
sqlx::query(
"INSERT INTO ducklake_table_column_stats
(table_id, column_id, contains_null, contains_nan, min_value, max_value, extra_stats)
VALUES (?, ?, ?, ?, ?, ?, NULL)",
)
.bind(table_id)
.bind(g.column_id)
.bind(g.contains_null)
.bind(g.contains_nan)
.bind(g.min_value)
.bind(g.max_value)
.execute(&mut **tx)
.await?;
}
Ok(())
}
async fn finalize_snapshot(
tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
table_id: i64,
columns: &[ColumnDef],
column_ids: &[i64],
mode: WriteMode,
base_snapshot: i64,
) -> Result<i64> {
let (snapshot_id, mut schema_version) = insert_snapshot(tx).await?;
use std::collections::{HashMap, HashSet};
let current = sqlx::query(
"SELECT column_name, column_type, column_order, nulls_allowed
FROM ducklake_column
WHERE table_id = ? AND end_snapshot IS NULL
ORDER BY column_order",
)
.bind(table_id)
.fetch_all(&mut **tx)
.await?;
let mut existing: Vec<(String, String, bool)> = Vec::with_capacity(current.len());
for row in ¤t {
let name: String = row.try_get("column_name")?;
let ty: String = row.try_get("column_type")?;
let nullable: bool = row
.try_get::<Option<bool>, _>("nulls_allowed")?
.unwrap_or(true);
existing.push((name, ty, nullable));
}
let is_ddl = existing.is_empty() || columns_differ(&existing, columns);
if is_ddl {
schema_version = bump_schema_version(tx, snapshot_id).await?;
}
let new_names: HashSet<&str> = columns.iter().map(|c| c.name.as_str()).collect();
let mut current_by_name: HashMap<String, (i64, bool)> = HashMap::new();
for row in ¤t {
let name: String = row.try_get("column_name")?;
let order: i64 = row.try_get("column_order")?;
let nullable: bool = row
.try_get::<Option<bool>, _>("nulls_allowed")?
.unwrap_or(true);
if !new_names.contains(name.as_str()) {
sqlx::query(
"UPDATE ducklake_column SET end_snapshot = ?
WHERE table_id = ? AND column_name = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(table_id)
.bind(&name)
.execute(&mut **tx)
.await?;
}
current_by_name.insert(name, (order, nullable));
}
for (order, (col, column_id)) in columns.iter().zip(column_ids.iter()).enumerate() {
match current_by_name.get(&col.name) {
Some(&(cur_order, cur_nullable)) => {
if cur_order != order as i64 || cur_nullable != col.is_nullable {
sqlx::query(
"UPDATE ducklake_column SET column_order = ?, nulls_allowed = ?
WHERE table_id = ? AND column_name = ? AND end_snapshot IS NULL",
)
.bind(order as i64)
.bind(col.is_nullable)
.bind(table_id)
.bind(&col.name)
.execute(&mut **tx)
.await?;
}
},
None => {
sqlx::query(
"INSERT INTO ducklake_column
(column_id, table_id, column_name, column_type, column_order, nulls_allowed, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.bind(column_id)
.bind(table_id)
.bind(&col.name)
.bind(&col.ducklake_type)
.bind(order as i64)
.bind(col.is_nullable)
.bind(snapshot_id)
.execute(&mut **tx)
.await?;
},
}
}
if mode == WriteMode::Replace {
detect_replace_conflict(tx, table_id, base_snapshot).await?;
sqlx::query(
"INSERT OR IGNORE INTO ducklake_table_stats
(table_id, record_count, next_row_id, file_size_bytes)
VALUES (?, 0, 0, 0)",
)
.bind(table_id)
.execute(&mut **tx)
.await?;
retire_prior_generation(tx, table_id, snapshot_id).await?;
}
if is_ddl {
record_schema_version(tx, snapshot_id, schema_version, table_id).await?;
}
Ok(snapshot_id)
}
impl MetadataWriter for SqliteMetadataWriter {
fn supports_update(&self) -> bool {
true
}
fn create_snapshot(&self) -> Result<i64> {
block_on(async {
let mut tx = self.pool.begin().await?;
let (snapshot_id, _schema_version) = insert_snapshot(&mut tx).await?;
tx.commit().await?;
Ok(snapshot_id)
})
}
fn get_or_create_schema(
&self,
name: &str,
path: Option<&str>,
snapshot_id: i64,
) -> Result<(i64, bool)> {
validate_name(name, "Schema")?;
block_on(async {
let existing = sqlx::query(
"SELECT schema_id FROM ducklake_schema
WHERE schema_name = ? AND end_snapshot IS NULL",
)
.bind(name)
.fetch_optional(&self.pool)
.await?;
if let Some(row) = existing {
return Ok((row.try_get(0)?, false));
}
let schema_path = path.unwrap_or(name);
let row = sqlx::query(
"INSERT INTO ducklake_schema (schema_name, path, path_is_relative, begin_snapshot)
VALUES (?, ?, 1, ?) RETURNING schema_id",
)
.bind(name)
.bind(schema_path)
.bind(snapshot_id)
.fetch_one(&self.pool)
.await?;
Ok((row.try_get(0)?, true))
})
}
fn get_or_create_table(
&self,
schema_id: i64,
name: &str,
path: Option<&str>,
snapshot_id: i64,
) -> Result<(i64, bool)> {
validate_name(name, "Table")?;
block_on(async {
let existing = sqlx::query(
"SELECT table_id FROM ducklake_table
WHERE schema_id = ? AND table_name = ? AND end_snapshot IS NULL",
)
.bind(schema_id)
.bind(name)
.fetch_optional(&self.pool)
.await?;
if let Some(row) = existing {
return Ok((row.try_get(0)?, false));
}
let table_path = path.unwrap_or(name);
let row = sqlx::query(
"INSERT INTO ducklake_table (schema_id, table_name, path, path_is_relative, begin_snapshot)
VALUES (?, ?, ?, 1, ?) RETURNING table_id",
)
.bind(schema_id)
.bind(name)
.bind(table_path)
.bind(snapshot_id)
.fetch_one(&self.pool)
.await?;
Ok((row.try_get(0)?, true))
})
}
fn promote_column_type(
&self,
table_id: i64,
column_name: &str,
new_ducklake_type: &str,
) -> Result<i64> {
crate::types::ducklake_to_arrow_type(new_ducklake_type)?;
block_on(async {
let mut tx = self.pool.begin().await?;
let (new_snapshot, _carried) = insert_snapshot(&mut tx).await?;
let row = sqlx::query(
"SELECT column_id, column_type, column_order, nulls_allowed
FROM ducklake_column
WHERE table_id = ? AND column_name = ? AND end_snapshot IS NULL",
)
.bind(table_id)
.bind(column_name)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| {
crate::DuckLakeError::InvalidConfig(format!(
"promote_column_type: no live column '{column_name}' in table {table_id}"
))
})?;
let column_id: i64 = row.try_get("column_id")?;
let cur_type: String = row.try_get("column_type")?;
let column_order: i64 = row.try_get("column_order")?;
let nulls_allowed: bool = row
.try_get::<Option<bool>, _>("nulls_allowed")?
.unwrap_or(true);
if crate::types::types_equal_canonical(&cur_type, new_ducklake_type) {
return Err(crate::DuckLakeError::InvalidConfig(format!(
"promote_column_type: column '{column_name}' is already type '{cur_type}' (no change)"
)));
}
if !crate::types::is_promotable(&cur_type, new_ducklake_type) {
return Err(crate::DuckLakeError::UnsupportedTypeChange {
operation: TypeChangeOperation::PromoteColumnType,
column: column_name.to_string(),
from: cur_type,
to: new_ducklake_type.to_string(),
});
}
let new_schema_version = bump_schema_version(&mut tx, new_snapshot).await?;
sqlx::query(
"UPDATE ducklake_column SET end_snapshot = ?
WHERE table_id = ? AND column_id = ? AND end_snapshot IS NULL",
)
.bind(new_snapshot)
.bind(table_id)
.bind(column_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"INSERT INTO ducklake_column
(column_id, begin_snapshot, end_snapshot, table_id, column_order, column_name, column_type, nulls_allowed)
VALUES (?, ?, NULL, ?, ?, ?, ?, ?)",
)
.bind(column_id)
.bind(new_snapshot)
.bind(table_id)
.bind(column_order)
.bind(column_name)
.bind(new_ducklake_type)
.bind(nulls_allowed)
.execute(&mut *tx)
.await?;
record_schema_version(&mut tx, new_snapshot, new_schema_version, table_id).await?;
tx.commit().await?;
Ok(new_snapshot)
})
}
fn set_columns(
&self,
table_id: i64,
columns: &[ColumnDef],
snapshot_id: i64,
) -> Result<Vec<i64>> {
if columns.is_empty() {
return Err(crate::DuckLakeError::InvalidConfig(
"Table must have at least one column".to_string(),
));
}
block_on(async {
let mut tx = self.pool.begin().await?;
sqlx::query(
"UPDATE ducklake_column SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(table_id)
.execute(&mut *tx)
.await?;
let n = columns.len() as i64;
let last_column_id = reserve_ids(&mut tx, "next_column_id", n).await?;
let first_column_id = last_column_id - n + 1;
let mut column_ids = Vec::with_capacity(columns.len());
for (order, col) in columns.iter().enumerate() {
let column_id = first_column_id + order as i64;
sqlx::query(
"INSERT INTO ducklake_column (column_id, table_id, column_name, column_type, column_order, nulls_allowed, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.bind(column_id)
.bind(table_id)
.bind(&col.name)
.bind(&col.ducklake_type)
.bind(order as i64)
.bind(col.is_nullable)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
column_ids.push(column_id);
}
tx.commit().await?;
Ok(column_ids)
})
}
fn register_data_file(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
_snapshot_id: i64,
file: &DataFileInfo,
mode: WriteMode,
base_snapshot: i64,
columns: &[ColumnDef],
column_ids: &[i64],
) -> Result<CommitIds> {
block_on(async {
let mut tx = self.pool.begin().await?;
let snapshot_id =
finalize_snapshot(&mut tx, table_id, columns, column_ids, mode, base_snapshot)
.await?;
sqlx::query(
"INSERT OR IGNORE INTO ducklake_table_stats
(table_id, record_count, next_row_id, file_size_bytes)
VALUES (?, 0, 0, 0)",
)
.bind(table_id)
.execute(&mut *tx)
.await?;
let stats_row =
sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
let row_id_start: i64 = stats_row.try_get(0)?;
sqlx::query(
"INSERT INTO ducklake_data_file
(table_id, path, path_is_relative, file_size_bytes,
footer_size, record_count, row_id_start, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(table_id)
.bind(&file.path)
.bind(file.path_is_relative)
.bind(file.file_size_bytes)
.bind(file.footer_size)
.bind(file.record_count)
.bind(row_id_start)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
let data_file_id: i64 = sqlx::query("SELECT last_insert_rowid()")
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats).await?;
recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;
sqlx::query(
"UPDATE ducklake_table_stats
SET next_row_id = next_row_id + ?,
record_count = record_count + ?,
file_size_bytes = file_size_bytes + ?
WHERE table_id = ?",
)
.bind(file.record_count)
.bind(file.record_count)
.bind(file.file_size_bytes)
.bind(table_id)
.execute(&mut *tx)
.await?;
let schema_id: i64 =
sqlx::query("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
#[allow(clippy::too_many_arguments)]
fn set_delete_file(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
_snapshot_id: i64,
data_file_id: i64,
expected_prev_delete_file: Option<i64>,
base_snapshot: i64,
delete: &DeleteFileInfo,
) -> Result<CommitIds> {
block_on(async {
let mut tx = self.pool.begin().await?;
let (snapshot_id, _schema_version) = insert_snapshot(&mut tx).await?;
let target_live: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(data_file_id)
.fetch_optional(&mut *tx)
.await?;
if target_live.is_none() {
return Err(crate::DuckLakeError::Conflict(format!(
"delete targets data file {data_file_id}, which was retired by a \
concurrent write since snapshot {base_snapshot}; retry against the \
new generation"
)));
}
let current_prev: Option<i64> = sqlx::query_scalar(
"SELECT delete_file_id FROM ducklake_delete_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(data_file_id)
.fetch_optional(&mut *tx)
.await?;
if current_prev != expected_prev_delete_file {
return Err(crate::DuckLakeError::Conflict(format!(
"delete on data file {data_file_id} conflicts with a concurrent delete \
(expected live delete file {expected_prev_delete_file:?}, found \
{current_prev:?}); retry against the new generation"
)));
}
if let Some(prev) = expected_prev_delete_file {
sqlx::query(
"UPDATE ducklake_delete_file SET end_snapshot = ?
WHERE delete_file_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(prev)
.execute(&mut *tx)
.await?;
}
sqlx::query(
"INSERT INTO ducklake_delete_file
(data_file_id, table_id, path, path_is_relative, file_size_bytes,
footer_size, delete_count, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(data_file_id)
.bind(table_id)
.bind(&delete.path)
.bind(delete.path_is_relative)
.bind(delete.file_size_bytes)
.bind(delete.footer_size)
.bind(delete.delete_count)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
let schema_id: i64 =
sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
#[allow(clippy::too_many_arguments)]
fn register_data_file_with_deletes(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
_snapshot_id: i64,
file: &DataFileInfo,
deletes: &[DeleteFileEntry],
mode: WriteMode,
base_snapshot: i64,
columns: &[ColumnDef],
column_ids: &[i64],
) -> Result<CommitIds> {
validate_delete_entries(mode, deletes)?;
block_on(async {
let mut tx = self.pool.begin().await?;
let snapshot_id =
finalize_snapshot(&mut tx, table_id, columns, column_ids, mode, base_snapshot)
.await?;
sqlx::query(
"INSERT OR IGNORE INTO ducklake_table_stats
(table_id, record_count, next_row_id, file_size_bytes)
VALUES (?, 0, 0, 0)",
)
.bind(table_id)
.execute(&mut *tx)
.await?;
let stats_row =
sqlx::query("SELECT next_row_id FROM ducklake_table_stats WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
let row_id_start: i64 = stats_row.try_get(0)?;
sqlx::query(
"INSERT INTO ducklake_data_file
(table_id, path, path_is_relative, file_size_bytes,
footer_size, record_count, row_id_start, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(table_id)
.bind(&file.path)
.bind(file.path_is_relative)
.bind(file.file_size_bytes)
.bind(file.footer_size)
.bind(file.record_count)
.bind(row_id_start)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
let data_file_id: i64 = sqlx::query("SELECT last_insert_rowid()")
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
insert_file_column_stats(&mut tx, table_id, data_file_id, &file.column_stats).await?;
recompute_table_column_stats(&mut tx, table_id, columns, column_ids).await?;
sqlx::query(
"UPDATE ducklake_table_stats
SET next_row_id = next_row_id + ?,
record_count = record_count + ?,
file_size_bytes = file_size_bytes + ?
WHERE table_id = ?",
)
.bind(file.record_count)
.bind(file.record_count)
.bind(file.file_size_bytes)
.bind(table_id)
.execute(&mut *tx)
.await?;
for entry in deletes {
let target_live: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(entry.data_file_id)
.fetch_optional(&mut *tx)
.await?;
if target_live.is_none() {
return Err(crate::DuckLakeError::Conflict(format!(
"UPDATE/DELETE on data file {} could not commit: the file is no longer \
live as of the catalog's current head (retired since snapshot \
{base_snapshot}). This happens when another writer committed a \
Replace/compaction, OR when an earlier write in THIS session already \
advanced the catalog (the catalog pins its snapshot at creation and does \
not refresh). Re-open the catalog at the latest snapshot and retry.",
entry.data_file_id
)));
}
let current_prev: Option<i64> = sqlx::query_scalar(
"SELECT delete_file_id FROM ducklake_delete_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(entry.data_file_id)
.fetch_optional(&mut *tx)
.await?;
if current_prev != entry.expected_prev_delete_file {
return Err(crate::DuckLakeError::Conflict(format!(
"UPDATE/DELETE on data file {} could not commit: its live delete file \
changed from {:?} to {current_prev:?} since snapshot {base_snapshot}. \
Another writer committed a delete on this file, OR an earlier \
UPDATE/DELETE in THIS session did (the catalog pins its snapshot at \
creation and does not refresh). Re-open the catalog at the latest \
snapshot and retry.",
entry.data_file_id, entry.expected_prev_delete_file
)));
}
if let Some(prev) = entry.expected_prev_delete_file {
sqlx::query(
"UPDATE ducklake_delete_file SET end_snapshot = ?
WHERE delete_file_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(prev)
.execute(&mut *tx)
.await?;
}
sqlx::query(
"INSERT INTO ducklake_delete_file
(data_file_id, table_id, path, path_is_relative, file_size_bytes,
footer_size, delete_count, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(entry.data_file_id)
.bind(table_id)
.bind(&entry.delete.path)
.bind(entry.delete.path_is_relative)
.bind(entry.delete.file_size_bytes)
.bind(entry.delete.footer_size)
.bind(entry.delete.delete_count)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
}
let schema_id: i64 =
sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
fn commit_positional_deletes(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
base_snapshot: i64,
deletes: &[DeleteFileEntry],
) -> Result<CommitIds> {
if deletes.is_empty() {
return Err(crate::DuckLakeError::InvalidConfig(
"commit_positional_deletes requires at least one delete entry".to_string(),
));
}
validate_delete_entries(WriteMode::Append, deletes)?;
block_on(async {
let mut tx = self.pool.begin().await?;
let (snapshot_id, _schema_version) = insert_snapshot(&mut tx).await?;
for entry in deletes {
let target_live: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(entry.data_file_id)
.fetch_optional(&mut *tx)
.await?;
if target_live.is_none() {
return Err(crate::DuckLakeError::Conflict(format!(
"DELETE on data file {} could not commit: the file is no longer live as \
of the catalog's current head (retired since snapshot {base_snapshot}). \
This happens when another writer committed a Replace/compaction, OR when \
an earlier write in THIS session already advanced the catalog (the \
catalog pins its snapshot at creation and does not refresh). Re-open the \
catalog at the latest snapshot and retry.",
entry.data_file_id
)));
}
let current_prev: Option<i64> = sqlx::query_scalar(
"SELECT delete_file_id FROM ducklake_delete_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(entry.data_file_id)
.fetch_optional(&mut *tx)
.await?;
if current_prev != entry.expected_prev_delete_file {
return Err(crate::DuckLakeError::Conflict(format!(
"DELETE on data file {} could not commit: its live delete file changed \
from {:?} to {current_prev:?} since snapshot {base_snapshot}. Another \
writer committed a delete on this file, OR an earlier DELETE in THIS \
session did (the catalog pins its snapshot at creation and does not \
refresh). Re-open the catalog at the latest snapshot and retry.",
entry.data_file_id, entry.expected_prev_delete_file
)));
}
if let Some(prev) = entry.expected_prev_delete_file {
sqlx::query(
"UPDATE ducklake_delete_file SET end_snapshot = ?
WHERE delete_file_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(prev)
.execute(&mut *tx)
.await?;
}
sqlx::query(
"INSERT INTO ducklake_delete_file
(data_file_id, table_id, path, path_is_relative, file_size_bytes,
footer_size, delete_count, begin_snapshot)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(entry.data_file_id)
.bind(table_id)
.bind(&entry.delete.path)
.bind(entry.delete.path_is_relative)
.bind(entry.delete.file_size_bytes)
.bind(entry.delete.footer_size)
.bind(entry.delete.delete_count)
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
}
let schema_id: i64 =
sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
fn commit_compaction(
&self,
table_id: i64,
base_snapshot: i64,
sources: &[crate::metadata_writer::CompactionSourceFile],
outputs: &[crate::metadata_writer::CompactionOutputFile],
retirement: crate::metadata_writer::SourceRetirement,
) -> Result<CommitIds> {
use crate::metadata_writer::SourceRetirement;
if sources.is_empty() {
return Err(crate::DuckLakeError::InvalidConfig(
"commit_compaction requires at least one source file".to_string(),
));
}
block_on(async {
let mut tx = self.pool.begin().await?;
let (snapshot_id, _schema_version) = insert_snapshot(&mut tx).await?;
for src in sources {
let target_live: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE data_file_id = ? AND table_id = ? AND end_snapshot IS NULL",
)
.bind(src.data_file_id)
.bind(table_id)
.fetch_optional(&mut *tx)
.await?;
if target_live.is_none() {
return Err(crate::DuckLakeError::Conflict(format!(
"compaction of table {table_id} could not commit: source data file {} is \
no longer live (retired by a concurrent Replace/compaction since \
snapshot {base_snapshot}). Re-open the catalog at the latest snapshot \
and re-plan.",
src.data_file_id
)));
}
let current_delete: Option<i64> = sqlx::query_scalar(
"SELECT delete_file_id FROM ducklake_delete_file
WHERE data_file_id = ? AND end_snapshot IS NULL",
)
.bind(src.data_file_id)
.fetch_optional(&mut *tx)
.await?;
if current_delete != src.delete_file_id {
return Err(crate::DuckLakeError::Conflict(format!(
"compaction of table {table_id} could not commit: the live delete file of \
source data file {} changed from {:?} to {current_delete:?} since \
snapshot {base_snapshot} (a concurrent DELETE/UPDATE). Re-open the \
catalog at the latest snapshot and re-plan.",
src.data_file_id, src.delete_file_id
)));
}
}
let source_data_ids: Vec<i64> = sources.iter().map(|s| s.data_file_id).collect();
match retirement {
SourceRetirement::Remove => {
let dead_data = sqlx::query(&format!(
"SELECT df.data_file_id, {RESOLVED_PATH} AS resolved_path, {REL_FLAG} AS rel
FROM ducklake_data_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
WHERE df.data_file_id IN ({})",
id_list(&source_data_ids)
))
.fetch_all(&mut *tx)
.await?;
schedule_files(&mut tx, dead_data).await?;
let dead_del = sqlx::query(&format!(
"SELECT df.delete_file_id, {RESOLVED_PATH} AS resolved_path, {REL_FLAG} AS rel
FROM ducklake_delete_file df
JOIN ducklake_table t ON t.table_id = df.table_id
JOIN ducklake_schema s ON s.schema_id = t.schema_id
WHERE df.data_file_id IN ({})",
id_list(&source_data_ids)
))
.fetch_all(&mut *tx)
.await?;
schedule_files(&mut tx, dead_del).await?;
sqlx::query(&format!(
"DELETE FROM ducklake_delete_file WHERE data_file_id IN ({})",
id_list(&source_data_ids)
))
.execute(&mut *tx)
.await?;
sqlx::query(&format!(
"DELETE FROM ducklake_data_file WHERE data_file_id IN ({})",
id_list(&source_data_ids)
))
.execute(&mut *tx)
.await?;
sqlx::query(&format!(
"DELETE FROM ducklake_file_column_stats WHERE data_file_id IN ({})",
id_list(&source_data_ids)
))
.execute(&mut *tx)
.await?;
},
SourceRetirement::Retire => {
sqlx::query(&format!(
"UPDATE ducklake_data_file SET end_snapshot = ?
WHERE data_file_id IN ({}) AND end_snapshot IS NULL",
id_list(&source_data_ids)
))
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
sqlx::query(&format!(
"UPDATE ducklake_delete_file SET end_snapshot = ?
WHERE data_file_id IN ({}) AND end_snapshot IS NULL",
id_list(&source_data_ids)
))
.bind(snapshot_id)
.execute(&mut *tx)
.await?;
},
}
for out in outputs {
let begin = out.begin_snapshot.unwrap_or(snapshot_id);
sqlx::query(
"INSERT INTO ducklake_data_file
(table_id, path, path_is_relative, file_size_bytes,
footer_size, record_count, row_id_start, begin_snapshot, partial_max)
VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?)",
)
.bind(table_id)
.bind(&out.file.path)
.bind(out.file.path_is_relative)
.bind(out.file.file_size_bytes)
.bind(out.file.footer_size)
.bind(out.file.record_count)
.bind(begin)
.bind(out.partial_max)
.execute(&mut *tx)
.await?;
let data_file_id: i64 = sqlx::query("SELECT last_insert_rowid()")
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
insert_file_column_stats(&mut tx, table_id, data_file_id, &out.file.column_stats)
.await?;
}
sqlx::query(
"INSERT OR IGNORE INTO ducklake_table_stats
(table_id, record_count, next_row_id, file_size_bytes)
VALUES (?, 0, 0, 0)",
)
.bind(table_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"UPDATE ducklake_table_stats SET
record_count = (SELECT COALESCE(SUM(record_count), 0)
FROM ducklake_data_file
WHERE table_id = ? AND end_snapshot IS NULL),
file_size_bytes = (SELECT COALESCE(SUM(file_size_bytes), 0)
FROM ducklake_data_file
WHERE table_id = ? AND end_snapshot IS NULL)
WHERE table_id = ?",
)
.bind(table_id)
.bind(table_id)
.bind(table_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"INSERT INTO ducklake_snapshot_changes
(snapshot_id, changes_made, commit_message)
VALUES (?, ?, ?)",
)
.bind(snapshot_id)
.bind(format!("compacted_table:{table_id}"))
.bind("datafusion compaction")
.execute(&mut *tx)
.await?;
let schema_id: i64 =
sqlx::query_scalar("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
fn commit_truncate(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
_base_snapshot: i64,
) -> Result<u64> {
block_on(async {
let mut tx = self.pool.begin().await?;
let (snapshot_id, _schema_version) = insert_snapshot(&mut tx).await?;
let has_live_data: Option<i64> = sqlx::query_scalar(
"SELECT 1 FROM ducklake_data_file
WHERE table_id = ? AND end_snapshot IS NULL LIMIT 1",
)
.bind(table_id)
.fetch_optional(&mut *tx)
.await?;
if has_live_data.is_none() {
return Ok(0);
}
let gross: Option<i64> = sqlx::query_scalar(
"SELECT COALESCE(record_count, 0) FROM ducklake_table_stats WHERE table_id = ?",
)
.bind(table_id)
.fetch_optional(&mut *tx)
.await?;
let deleted: i64 = sqlx::query_scalar(
"SELECT COALESCE(SUM(delete_count), 0) FROM ducklake_delete_file
WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(table_id)
.fetch_one(&mut *tx)
.await?;
let live_rows = (gross.unwrap_or(0) - deleted).max(0) as u64;
sqlx::query(
"UPDATE ducklake_data_file SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(table_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"UPDATE ducklake_delete_file SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(table_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"UPDATE ducklake_table_stats SET record_count = 0, file_size_bytes = 0
WHERE table_id = ?",
)
.bind(table_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(live_rows)
})
}
fn publish_snapshot(
&self,
table_id: i64,
_schema_name: &str,
_table_name: &str,
_snapshot_id: i64,
mode: WriteMode,
base_snapshot: i64,
columns: &[ColumnDef],
column_ids: &[i64],
) -> Result<CommitIds> {
block_on(async {
let mut tx = self.pool.begin().await?;
let snapshot_id =
finalize_snapshot(&mut tx, table_id, columns, column_ids, mode, base_snapshot)
.await?;
let schema_id: i64 =
sqlx::query("SELECT schema_id FROM ducklake_table WHERE table_id = ?")
.bind(table_id)
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
tx.commit().await?;
Ok(CommitIds {
snapshot_id,
schema_id,
table_id,
})
})
}
fn end_table_files(&self, table_id: i64, snapshot_id: i64) -> Result<u64> {
block_on(async {
let mut tx = self.pool.begin().await?;
let result = sqlx::query(
"UPDATE ducklake_data_file SET end_snapshot = ?
WHERE table_id = ? AND end_snapshot IS NULL",
)
.bind(snapshot_id)
.bind(table_id)
.execute(&mut *tx)
.await?;
sqlx::query(
"UPDATE ducklake_table_stats
SET record_count = 0, file_size_bytes = 0
WHERE table_id = ?",
)
.bind(table_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(result.rows_affected())
})
}
fn get_data_path(&self) -> Result<String> {
block_on(async {
let row =
sqlx::query("SELECT value FROM ducklake_metadata WHERE key = ? AND scope IS NULL")
.bind("data_path")
.fetch_optional(&self.pool)
.await?;
match row {
Some(r) => Ok(r.try_get(0)?),
None => Err(crate::error::DuckLakeError::InvalidConfig(
"Missing required catalog metadata: 'data_path' not configured.".to_string(),
)),
}
})
}
fn set_data_path(&self, path: &str) -> Result<()> {
block_on(async {
sqlx::query("DELETE FROM ducklake_metadata WHERE key = 'data_path' AND scope IS NULL")
.execute(&self.pool)
.await?;
sqlx::query(
"INSERT INTO ducklake_metadata (key, value, scope)
VALUES ('data_path', ?, NULL)",
)
.bind(path)
.execute(&self.pool)
.await?;
Ok(())
})
}
fn initialize_schema(&self) -> Result<()> {
block_on(async {
sqlx::query(SQL_CREATE_SCHEMA).execute(&self.pool).await?;
migrate_ducklake_column_drop_pk(&self.pool).await?;
migrate_add_schema_version(&self.pool).await?;
migrate_add_partial_max(&self.pool).await?;
sqlx::query(
"INSERT INTO ducklake_metadata (key, value, scope)
SELECT 'next_column_id',
CAST(COALESCE((SELECT MAX(column_id) FROM ducklake_column), 0) AS TEXT),
NULL
WHERE NOT EXISTS (
SELECT 1 FROM ducklake_metadata WHERE key = 'next_column_id' AND scope IS NULL
)",
)
.execute(&self.pool)
.await?;
Ok(())
})
}
fn begin_write_transaction(
&self,
schema_name: &str,
table_name: &str,
columns: &[ColumnDef],
mode: WriteMode,
) -> Result<WriteSetupResult> {
validate_name(schema_name, "Schema")?;
validate_name(table_name, "Table")?;
if columns.is_empty() {
return Err(crate::DuckLakeError::InvalidConfig(
"Table must have at least one column".to_string(),
));
}
block_on(async {
let mut tx = self.pool.begin().await?;
let n = columns.len() as i64;
let last_column_id = reserve_ids(&mut tx, "next_column_id", n).await?;
let fresh_ids: Vec<i64> = ((last_column_id - n + 1)..=last_column_id).collect();
let base_snapshot_id: i64 =
sqlx::query("SELECT COALESCE(MAX(snapshot_id), 0) FROM ducklake_snapshot")
.fetch_one(&mut *tx)
.await?
.try_get(0)?;
let snapshot_id: i64 = base_snapshot_id + 1;
let schema_id: i64 = {
let existing = sqlx::query(
"SELECT schema_id FROM ducklake_schema
WHERE schema_name = ? AND end_snapshot IS NULL",
)
.bind(schema_name)
.fetch_optional(&mut *tx)
.await?;
if let Some(row) = existing {
row.try_get(0)?
} else {
let row = sqlx::query(
"INSERT INTO ducklake_schema (schema_name, path, path_is_relative, begin_snapshot)
VALUES (?, ?, 1, ?) RETURNING schema_id",
)
.bind(schema_name)
.bind(schema_name)
.bind(snapshot_id)
.fetch_one(&mut *tx)
.await?;
row.try_get(0)?
}
};
let table_id: i64 = {
let existing = sqlx::query(
"SELECT table_id FROM ducklake_table
WHERE schema_id = ? AND table_name = ? AND end_snapshot IS NULL",
)
.bind(schema_id)
.bind(table_name)
.fetch_optional(&mut *tx)
.await?;
if let Some(row) = existing {
row.try_get(0)?
} else {
let row = sqlx::query(
"INSERT INTO ducklake_table (schema_id, table_name, path, path_is_relative, begin_snapshot)
VALUES (?, ?, ?, 1, ?) RETURNING table_id",
)
.bind(schema_id)
.bind(table_name)
.bind(table_name)
.bind(snapshot_id)
.fetch_one(&mut *tx)
.await?;
row.try_get(0)?
}
};
let rows = sqlx::query(
"SELECT column_name, column_type, nulls_allowed, column_id
FROM ducklake_column
WHERE table_id = ? AND end_snapshot IS NULL
ORDER BY column_order",
)
.bind(table_id)
.fetch_all(&mut *tx)
.await?;
let mut existing_columns: Vec<(String, String, bool)> = Vec::with_capacity(rows.len());
let mut existing_ids: std::collections::HashMap<String, i64> =
std::collections::HashMap::new();
for row in rows {
let name: String = row.try_get(0)?;
let col_type: String = row.try_get(1)?;
let nullable: bool = row.try_get::<Option<bool>, _>(2)?.unwrap_or(true);
let cid: i64 = row.try_get(3)?;
existing_ids.insert(name.clone(), cid);
existing_columns.push((name, col_type, nullable));
}
if !existing_columns.is_empty() {
use std::collections::HashMap;
let existing_map: HashMap<&str, (&str, bool)> = existing_columns
.iter()
.map(|(name, col_type, nullable)| {
(name.as_str(), (col_type.as_str(), *nullable))
})
.collect();
for new_col in columns.iter() {
if let Some((existing_type, _existing_nullable)) =
existing_map.get(new_col.name.as_str())
{
if !crate::types::types_equal_canonical(
existing_type,
&new_col.ducklake_type,
) {
return Err(crate::error::DuckLakeError::UnsupportedTypeChange {
operation: TypeChangeOperation::DataWrite {
mode: match mode {
WriteMode::Replace => TypeChangeWriteMode::Replace,
WriteMode::Append => TypeChangeWriteMode::Append,
},
},
column: new_col.name.clone(),
from: (*existing_type).to_string(),
to: new_col.ducklake_type.clone(),
});
}
} else if mode == WriteMode::Append && !new_col.is_nullable {
return Err(crate::error::DuckLakeError::InvalidConfig(format!(
"Schema evolution error: new column '{}' must be nullable. Adding non-nullable columns is not allowed.",
new_col.name
)));
}
}
}
let column_ids: Vec<i64> = columns
.iter()
.zip(fresh_ids.iter())
.map(|(col, &fresh)| existing_ids.get(&col.name).copied().unwrap_or(fresh))
.collect();
tx.commit().await?;
Ok(WriteSetupResult {
snapshot_id,
base_snapshot_id,
schema_id,
table_id,
column_ids,
})
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
async fn create_test_writer() -> (SqliteMetadataWriter, TempDir) {
let temp_dir = TempDir::new().unwrap();
let db_path = temp_dir.path().join("test.db");
let conn_str = format!("sqlite:{}?mode=rwc", db_path.display());
let writer = SqliteMetadataWriter::new_with_init(&conn_str)
.await
.unwrap();
(writer, temp_dir)
}
#[tokio::test(flavor = "multi_thread")]
async fn migrate_legacy_column_pk_to_bare_shape() {
let temp_dir = TempDir::new().unwrap();
let db_path = temp_dir.path().join("legacy.db");
let conn_str = format!("sqlite:{}?mode=rwc", db_path.display());
let pool = SqlitePool::connect(&conn_str).await.unwrap();
sqlx::query(
"CREATE TABLE ducklake_column (
column_id INTEGER PRIMARY KEY,
table_id INTEGER NOT NULL,
column_name VARCHAR NOT NULL,
column_type VARCHAR NOT NULL,
column_order INTEGER NOT NULL,
nulls_allowed BOOLEAN DEFAULT 1,
initial_default VARCHAR,
default_value VARCHAR,
parent_column INTEGER,
default_value_type VARCHAR,
default_value_dialect VARCHAR,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER
)",
)
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO ducklake_column
(column_id, table_id, column_name, column_type, column_order, nulls_allowed, begin_snapshot)
VALUES (5, 1, 'id', 'int32', 0, 1, 1)",
)
.execute(&pool)
.await
.unwrap();
let dup = sqlx::query(
"INSERT INTO ducklake_column
(column_id, table_id, column_name, column_type, column_order, nulls_allowed, begin_snapshot)
VALUES (5, 1, 'id', 'int64', 0, 1, 2)",
)
.execute(&pool)
.await;
assert!(
dup.is_err(),
"legacy single-row PK must reject a duplicate column_id"
);
migrate_ducklake_column_drop_pk(&pool).await.unwrap();
let row = sqlx::query(
"SELECT column_id, column_name, column_type FROM ducklake_column WHERE begin_snapshot = 1",
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(row.try_get::<i64, _>("column_id").unwrap(), 5);
assert_eq!(row.try_get::<String, _>("column_name").unwrap(), "id");
assert_eq!(row.try_get::<String, _>("column_type").unwrap(), "int32");
sqlx::query(
"INSERT INTO ducklake_column
(column_id, begin_snapshot, end_snapshot, table_id, column_order, column_name, column_type, nulls_allowed)
VALUES (5, 2, NULL, 1, 0, 'id', 'int64', 1)",
)
.execute(&pool)
.await
.unwrap();
let cnt: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM ducklake_column WHERE column_id = 5")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
cnt, 2,
"two version rows sharing a column_id must coexist post-migration"
);
migrate_ducklake_column_drop_pk(&pool).await.unwrap();
let cnt2: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ducklake_column")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(cnt2, 2, "migration must be idempotent");
}
#[tokio::test(flavor = "multi_thread")]
async fn migrate_add_schema_version_to_legacy_snapshot() {
let temp_dir = TempDir::new().unwrap();
let db_path = temp_dir.path().join("legacy.db");
let conn_str = format!("sqlite:{}?mode=rwc", db_path.display());
let pool = SqlitePool::connect(&conn_str).await.unwrap();
sqlx::query(
"CREATE TABLE ducklake_snapshot (
snapshot_id INTEGER PRIMARY KEY,
snapshot_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)",
)
.execute(&pool)
.await
.unwrap();
sqlx::query("INSERT INTO ducklake_snapshot (snapshot_id) VALUES (1), (2)")
.execute(&pool)
.await
.unwrap();
migrate_add_schema_version(&pool).await.unwrap();
let has_col: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM pragma_table_info('ducklake_snapshot') WHERE name = 'schema_version'",
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(has_col, 1, "schema_version column must be added");
let max_v: i64 =
sqlx::query_scalar("SELECT COALESCE(MAX(schema_version), 0) FROM ducklake_snapshot")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
max_v, 0,
"pre-existing snapshots backfill to schema_version 0"
);
let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ducklake_snapshot")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(rows, 2, "no snapshot rows lost in the migration");
migrate_add_schema_version(&pool).await.unwrap();
let rows2: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ducklake_snapshot")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(rows2, 2, "migration must be idempotent");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_create_snapshot() {
let (writer, _temp) = create_test_writer().await;
let snap1 = writer.create_snapshot().unwrap();
assert_eq!(snap1, 1);
let snap2 = writer.create_snapshot().unwrap();
assert_eq!(snap2, 2);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_get_or_create_schema() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = writer.create_snapshot().unwrap();
let (schema_id, created) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
assert!(created);
assert_eq!(schema_id, 1);
let (schema_id2, created2) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
assert!(!created2);
assert_eq!(schema_id2, 1);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_get_or_create_table() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = writer.create_snapshot().unwrap();
let (schema_id, _) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
let (table_id, created) = writer
.get_or_create_table(schema_id, "users", None, snapshot_id)
.unwrap();
assert!(created);
assert_eq!(table_id, 1);
let (table_id2, created2) = writer
.get_or_create_table(schema_id, "users", None, snapshot_id)
.unwrap();
assert!(!created2);
assert_eq!(table_id2, 1);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_set_columns() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = writer.create_snapshot().unwrap();
let (schema_id, _) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "users", None, snapshot_id)
.unwrap();
let columns = vec![
ColumnDef::new("id", "int64", false).unwrap(),
ColumnDef::new("name", "varchar", true).unwrap(),
];
let column_ids = writer.set_columns(table_id, &columns, snapshot_id).unwrap();
assert_eq!(column_ids.len(), 2);
assert_eq!(column_ids[0], 1);
assert_eq!(column_ids[1], 2);
}
#[tokio::test(flavor = "multi_thread")]
async fn promote_leaves_two_versioned_column_rows_same_id() {
let (writer, _temp) = create_test_writer().await;
let snap1 = writer.create_snapshot().unwrap();
let (schema_id, _) = writer.get_or_create_schema("main", None, snap1).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap1)
.unwrap();
writer
.set_columns(
table_id,
&[ColumnDef::new("id", "int32", false).unwrap()],
snap1,
)
.unwrap();
let snap2 = writer.promote_column_type(table_id, "id", "int64").unwrap();
assert!(snap2 > snap1, "promote creates a newer snapshot");
let rows = sqlx::query(
"SELECT column_id, column_type, begin_snapshot, end_snapshot
FROM ducklake_column
WHERE table_id = ? AND column_name = 'id'
ORDER BY begin_snapshot",
)
.bind(table_id)
.fetch_all(&writer.pool)
.await
.unwrap();
assert_eq!(
rows.len(),
2,
"promote must leave TWO versioned rows for the column"
);
let cid0: i64 = rows[0].try_get("column_id").unwrap();
let type0: String = rows[0].try_get("column_type").unwrap();
let end0: Option<i64> = rows[0].try_get("end_snapshot").unwrap();
let cid1: i64 = rows[1].try_get("column_id").unwrap();
let type1: String = rows[1].try_get("column_type").unwrap();
let begin1: i64 = rows[1].try_get("begin_snapshot").unwrap();
let end1: Option<i64> = rows[1].try_get("end_snapshot").unwrap();
assert_eq!(
cid0, cid1,
"both versions share the SAME column_id (stable field-id)"
);
assert_eq!(type0, "int32", "old version retains its int32 type");
assert_eq!(
end0,
Some(snap2),
"old version retired at the promote snapshot"
);
assert_eq!(type1, "int64", "new version carries the widened int64 type");
assert_eq!(begin1, snap2, "new version begins at the promote snapshot");
assert_eq!(end1, None, "new version is the live one");
let live: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ducklake_column
WHERE table_id = ? AND column_name = 'id' AND end_snapshot IS NULL",
)
.bind(table_id)
.fetch_one(&writer.pool)
.await
.unwrap();
assert_eq!(
live, 1,
"exactly one live version per field-id after promote"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn phase_d_legacy_catalog_with_data_upgrades_and_reads() {
use crate::{DuckLakeCatalog, DuckLakeTableWriter, SqliteMetadataProvider};
use arrow::array::{Array, Int32Array, Int64Array};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datafusion::prelude::SessionContext;
use object_store::local::LocalFileSystem;
let temp = TempDir::new().unwrap();
let db_path = temp.path().join("test.db");
let data_path = temp.path().join("data");
std::fs::create_dir_all(&data_path).unwrap();
let conn_str = format!("sqlite:{}?mode=rwc", db_path.display());
{
let writer = SqliteMetadataWriter::new_with_init(&conn_str)
.await
.unwrap();
writer.set_data_path(data_path.to_str().unwrap()).unwrap();
let store: std::sync::Arc<dyn object_store::ObjectStore> =
std::sync::Arc::new(LocalFileSystem::new());
let schema =
std::sync::Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
let batch = RecordBatch::try_new(
schema,
vec![std::sync::Arc::new(Int32Array::from(vec![1, 2, 3]))],
)
.unwrap();
DuckLakeTableWriter::new(std::sync::Arc::new(writer), store)
.unwrap()
.write_table("main", "t", &[batch])
.await
.unwrap();
}
{
let pool = SqlitePool::connect(&conn_str).await.unwrap();
sqlx::query("ALTER TABLE ducklake_column RENAME TO ducklake_column__tmp")
.execute(&pool)
.await
.unwrap();
sqlx::query(
"CREATE TABLE ducklake_column (
column_id INTEGER PRIMARY KEY,
table_id INTEGER NOT NULL,
column_name VARCHAR NOT NULL,
column_type VARCHAR NOT NULL,
column_order INTEGER NOT NULL,
nulls_allowed BOOLEAN DEFAULT 1,
initial_default VARCHAR,
default_value VARCHAR,
parent_column INTEGER,
default_value_type VARCHAR,
default_value_dialect VARCHAR,
begin_snapshot INTEGER NOT NULL,
end_snapshot INTEGER
)",
)
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO ducklake_column
(column_id, table_id, column_name, column_type, column_order,
nulls_allowed, begin_snapshot, end_snapshot)
SELECT column_id, table_id, column_name, column_type, column_order,
nulls_allowed, begin_snapshot, end_snapshot
FROM ducklake_column__tmp",
)
.execute(&pool)
.await
.unwrap();
sqlx::query("DROP TABLE ducklake_column__tmp")
.execute(&pool)
.await
.unwrap();
let is_pk: i64 = sqlx::query_scalar(
"SELECT pk FROM pragma_table_info('ducklake_column') WHERE name = 'column_id'",
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(is_pk, 1, "downgrade produced the legacy single-row PK");
pool.close().await;
}
let writer = SqliteMetadataWriter::new_with_init(&conn_str)
.await
.unwrap();
let is_pk: i64 = sqlx::query_scalar(
"SELECT pk FROM pragma_table_info('ducklake_column') WHERE name = 'column_id'",
)
.fetch_one(&writer.pool)
.await
.unwrap();
assert_eq!(is_pk, 0, "re-open migrated the legacy PK away");
let read_conn = format!("sqlite:{}", db_path.display());
let provider = SqliteMetadataProvider::new(&read_conn).await.unwrap();
let catalog = DuckLakeCatalog::new(provider).unwrap();
let ctx = SessionContext::new();
ctx.register_catalog("test", std::sync::Arc::new(catalog));
let batches = ctx
.sql("SELECT id FROM test.main.t ORDER BY id")
.await
.unwrap()
.collect()
.await
.unwrap();
let ids = batches[0]
.column(0)
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
assert_eq!(
ids.values(),
&[1, 2, 3],
"old-version data reads intact after upgrade"
);
let table_id: i64 =
sqlx::query_scalar("SELECT table_id FROM ducklake_table WHERE table_name = 't'")
.fetch_one(&writer.pool)
.await
.unwrap();
writer.promote_column_type(table_id, "id", "int64").unwrap();
let provider2 = SqliteMetadataProvider::new(&read_conn).await.unwrap();
let catalog2 = DuckLakeCatalog::new(provider2).unwrap();
let ctx2 = SessionContext::new();
ctx2.register_catalog("test", std::sync::Arc::new(catalog2));
let batches2 = ctx2
.sql("SELECT id FROM test.main.t ORDER BY id")
.await
.unwrap()
.collect()
.await
.unwrap();
assert_eq!(batches2[0].schema().field(0).data_type(), &DataType::Int64);
let ids2 = batches2[0]
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert_eq!(
ids2.values(),
&[1, 2, 3],
"post-upgrade promote widens + reads intact"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_register_data_file() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = reserve_snapshot(&writer).await;
let (schema_id, _) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "users", None, snapshot_id)
.unwrap();
let file = DataFileInfo::new("data.parquet", 1024, 100).with_footer_size(256);
let committed = writer
.register_data_file(
table_id,
"main",
"users",
snapshot_id,
&file,
WriteMode::Append,
0,
&[],
&[],
)
.unwrap();
assert_eq!(committed.snapshot_id, 1);
}
async fn reserve_snapshot(writer: &SqliteMetadataWriter) -> i64 {
sqlx::query("SELECT COALESCE(MAX(snapshot_id), 0) + 1 FROM ducklake_snapshot")
.fetch_one(&writer.pool)
.await
.unwrap()
.try_get(0)
.unwrap()
}
async fn read_row_id_start(writer: &SqliteMetadataWriter, path: &str) -> Option<i64> {
let row = sqlx::query("SELECT row_id_start FROM ducklake_data_file WHERE path = ?")
.bind(path)
.fetch_one(&writer.pool)
.await
.unwrap();
row.try_get(0).ok()
}
async fn read_table_stats(writer: &SqliteMetadataWriter, table_id: i64) -> (i64, i64, i64) {
let row = sqlx::query(
"SELECT record_count, next_row_id, file_size_bytes
FROM ducklake_table_stats WHERE table_id = ?",
)
.bind(table_id)
.fetch_one(&writer.pool)
.await
.unwrap();
(
row.try_get(0).unwrap(),
row.try_get(1).unwrap(),
row.try_get(2).unwrap(),
)
}
#[allow(clippy::type_complexity)]
async fn read_file_column_stats(
writer: &SqliteMetadataWriter,
path: &str,
) -> Vec<(
i64,
Option<String>,
Option<String>,
Option<i64>,
Option<i64>,
)> {
sqlx::query(
"SELECT s.column_id, s.min_value, s.max_value, s.null_count, s.value_count
FROM ducklake_file_column_stats s
JOIN ducklake_data_file d ON d.data_file_id = s.data_file_id
WHERE d.path = ?
ORDER BY s.column_id",
)
.bind(path)
.fetch_all(&writer.pool)
.await
.unwrap()
.into_iter()
.map(|row| {
(
row.try_get(0).unwrap(),
row.try_get(1).unwrap(),
row.try_get(2).unwrap(),
row.try_get(3).unwrap(),
row.try_get(4).unwrap(),
)
})
.collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn register_data_file_persists_column_stats() {
let (writer, _temp) = create_test_writer().await;
let snap = reserve_snapshot(&writer).await;
let (schema_id, _) = writer.get_or_create_schema("main", None, snap).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap)
.unwrap();
let file = DataFileInfo::new("a.parquet", 100, 3).with_column_stats(vec![
ColumnStat {
column_id: 1,
min_value: Some("1".to_string()),
max_value: Some("3".to_string()),
null_count: Some(0),
value_count: Some(3),
contains_nan: None,
column_size_bytes: None,
},
ColumnStat {
column_id: 2,
min_value: Some("Alice".to_string()),
max_value: Some("Bob".to_string()),
null_count: Some(1),
value_count: Some(2),
contains_nan: None,
column_size_bytes: None,
},
]);
writer
.register_data_file(
table_id,
"main",
"t",
snap,
&file,
WriteMode::Append,
0,
&[],
&[],
)
.unwrap();
let rows = read_file_column_stats(&writer, "a.parquet").await;
assert_eq!(
rows,
vec![
(
1,
Some("1".to_string()),
Some("3".to_string()),
Some(0),
Some(3)
),
(
2,
Some("Alice".to_string()),
Some("Bob".to_string()),
Some(1),
Some(2)
),
]
);
}
async fn read_table_column_stats(
writer: &SqliteMetadataWriter,
table_id: i64,
) -> Vec<(i64, Option<bool>, Option<String>, Option<String>)> {
sqlx::query(
"SELECT column_id, contains_null, min_value, max_value
FROM ducklake_table_column_stats WHERE table_id = ? ORDER BY column_id",
)
.bind(table_id)
.fetch_all(&writer.pool)
.await
.unwrap()
.into_iter()
.map(|row| {
(
row.try_get(0).unwrap(),
row.try_get(1).unwrap(),
row.try_get(2).unwrap(),
row.try_get(3).unwrap(),
)
})
.collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn table_column_stats_widen_numerically_across_files() {
let (writer, _temp) = create_test_writer().await;
let snap1 = reserve_snapshot(&writer).await;
let (schema_id, _) = writer.get_or_create_schema("main", None, snap1).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap1)
.unwrap();
let columns = vec![ColumnDef::new("n", "int32", true).unwrap()];
let column_ids = vec![1_i64];
let stat = |min: &str, max: &str, nulls: i64, vals: i64| {
vec![ColumnStat {
column_id: 1,
min_value: Some(min.to_string()),
max_value: Some(max.to_string()),
null_count: Some(nulls),
value_count: Some(vals),
contains_nan: None,
column_size_bytes: None,
}]
};
writer
.register_data_file(
table_id,
"main",
"t",
snap1,
&DataFileInfo::new("a.parquet", 100, 3).with_column_stats(stat("9", "9", 0, 3)),
WriteMode::Append,
0,
&columns,
&column_ids,
)
.unwrap();
let snap2 = reserve_snapshot(&writer).await;
writer
.register_data_file(
table_id,
"main",
"t",
snap2,
&DataFileInfo::new("b.parquet", 100, 2).with_column_stats(stat("10", "10", 1, 1)),
WriteMode::Append,
0,
&columns,
&column_ids,
)
.unwrap();
assert_eq!(read_file_column_stats(&writer, "a.parquet").await.len(), 1);
assert_eq!(read_file_column_stats(&writer, "b.parquet").await.len(), 1);
assert_eq!(
read_table_column_stats(&writer, table_id).await,
vec![(1, Some(true), Some("9".to_string()), Some("10".to_string()))]
);
}
#[tokio::test(flavor = "multi_thread")]
async fn table_column_stats_null_when_a_live_file_lacks_stats() {
let (writer, _temp) = create_test_writer().await;
let snap1 = reserve_snapshot(&writer).await;
let (schema_id, _) = writer.get_or_create_schema("main", None, snap1).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap1)
.unwrap();
let columns = vec![ColumnDef::new("n", "int32", true).unwrap()];
let column_ids = vec![1_i64];
writer
.register_data_file(
table_id,
"main",
"t",
snap1,
&DataFileInfo::new("a.parquet", 100, 3).with_column_stats(vec![ColumnStat {
column_id: 1,
min_value: Some("1".to_string()),
max_value: Some("1".to_string()),
null_count: Some(0),
value_count: Some(3),
contains_nan: None,
column_size_bytes: None,
}]),
WriteMode::Append,
0,
&columns,
&column_ids,
)
.unwrap();
let snap2 = reserve_snapshot(&writer).await;
writer
.register_data_file(
table_id,
"main",
"t",
snap2,
&DataFileInfo::new("b.parquet", 100, 2),
WriteMode::Append,
0,
&columns,
&column_ids,
)
.unwrap();
assert_eq!(
read_table_column_stats(&writer, table_id).await,
vec![(1, None, None, None)],
"incomplete coverage (a statless live file) must NULL the roll-up"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn row_id_start_advances_across_inserts() {
let (writer, _temp) = create_test_writer().await;
let snap1 = reserve_snapshot(&writer).await;
let (schema_id, _) = writer.get_or_create_schema("main", None, snap1).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap1)
.unwrap();
writer
.register_data_file(
table_id,
"main",
"t",
snap1,
&DataFileInfo::new("a.parquet", 100, 3),
WriteMode::Append,
0, &[],
&[],
)
.unwrap();
let snap2 = reserve_snapshot(&writer).await;
writer
.register_data_file(
table_id,
"main",
"t",
snap2,
&DataFileInfo::new("b.parquet", 250, 7),
WriteMode::Append,
0, &[],
&[],
)
.unwrap();
assert_eq!(read_row_id_start(&writer, "a.parquet").await, Some(0));
assert_eq!(read_row_id_start(&writer, "b.parquet").await, Some(3));
let (records, next, bytes) = read_table_stats(&writer, table_id).await;
assert_eq!(records, 10, "record_count = 3 + 7");
assert_eq!(next, 10, "next_row_id advances by sum of record_counts");
assert_eq!(bytes, 350, "file_size_bytes accumulates");
}
#[tokio::test(flavor = "multi_thread")]
async fn end_table_files_preserves_next_row_id() {
let (writer, _temp) = create_test_writer().await;
let snap1 = reserve_snapshot(&writer).await;
let (schema_id, _) = writer.get_or_create_schema("main", None, snap1).unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "t", None, snap1)
.unwrap();
writer
.register_data_file(
table_id,
"main",
"t",
snap1,
&DataFileInfo::new("a.parquet", 100, 5),
WriteMode::Append,
0, &[],
&[],
)
.unwrap();
let snap2 = writer.create_snapshot().unwrap();
writer.end_table_files(table_id, snap2).unwrap();
let (records, next, bytes) = read_table_stats(&writer, table_id).await;
assert_eq!(records, 0, "record_count cleared after end_table_files");
assert_eq!(next, 5, "next_row_id preserved (monotonic across lifetime)");
assert_eq!(bytes, 0, "file_size_bytes cleared");
let snap3 = reserve_snapshot(&writer).await;
writer
.register_data_file(
table_id,
"main",
"t",
snap3,
&DataFileInfo::new("b.parquet", 200, 2),
WriteMode::Append,
0, &[],
&[],
)
.unwrap();
assert_eq!(
read_row_id_start(&writer, "b.parquet").await,
Some(5),
"post-replace files must start at the preserved counter, not 0",
);
}
#[tokio::test(flavor = "multi_thread")]
async fn row_id_start_works_when_stats_row_missing() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = reserve_snapshot(&writer).await;
let (schema_id, _) = writer
.get_or_create_schema("main", None, snapshot_id)
.unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "legacy", None, snapshot_id)
.unwrap();
sqlx::query("DELETE FROM ducklake_table_stats WHERE table_id = ?")
.bind(table_id)
.execute(&writer.pool)
.await
.unwrap();
writer
.register_data_file(
table_id,
"main",
"legacy",
snapshot_id,
&DataFileInfo::new("a.parquet", 50, 4),
WriteMode::Append,
0, &[],
&[],
)
.unwrap();
assert_eq!(read_row_id_start(&writer, "a.parquet").await, Some(0));
let (records, next, _) = read_table_stats(&writer, table_id).await;
assert_eq!(records, 4);
assert_eq!(next, 4);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_end_table_files() {
let (writer, _temp) = create_test_writer().await;
let snapshot1 = reserve_snapshot(&writer).await;
let (schema_id, _) = writer
.get_or_create_schema("main", None, snapshot1)
.unwrap();
let (table_id, _) = writer
.get_or_create_table(schema_id, "users", None, snapshot1)
.unwrap();
let file = DataFileInfo::new("data1.parquet", 1024, 100);
writer
.register_data_file(
table_id,
"main",
"users",
snapshot1,
&file,
WriteMode::Append,
0,
&[],
&[],
)
.unwrap();
let snapshot2 = writer.create_snapshot().unwrap();
let ended = writer.end_table_files(table_id, snapshot2).unwrap();
assert_eq!(ended, 1);
let ended2 = writer.end_table_files(table_id, snapshot2).unwrap();
assert_eq!(ended2, 0);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_data_path() {
let (writer, _temp) = create_test_writer().await;
writer.set_data_path("/data/path").unwrap();
let path = writer.get_data_path().unwrap();
assert_eq!(path, "/data/path");
writer.set_data_path("/new/path").unwrap();
let path2 = writer.get_data_path().unwrap();
assert_eq!(path2, "/new/path");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_get_or_create_schema_empty_name_rejected() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = writer.create_snapshot().unwrap();
let result = writer.get_or_create_schema("", None, snapshot_id);
assert!(result.is_err());
let err_msg = format!("{}", result.unwrap_err());
assert!(err_msg.contains("empty"), "Expected 'empty' in: {err_msg}");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_get_or_create_schema_control_char_rejected() {
let (writer, _temp) = create_test_writer().await;
let snapshot_id = writer.create_snapshot().unwrap();
let result = writer.get_or_create_schema("bad\0schema", None, snapshot_id);
assert!(result.is_err());
let err_msg = format!("{}", result.unwrap_err());
assert!(
err_msg.contains("control character"),
"Expected 'control character' in: {err_msg}"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn test_begin_write_transaction_empty_schema_name_rejected() {
let (writer, _temp) = create_test_writer().await;
let columns = vec![ColumnDef::new("id", "int64", false).unwrap()];
let result = writer.begin_write_transaction("", "table", &columns, WriteMode::Replace);
assert!(result.is_err());
let err_msg = format!("{}", result.unwrap_err());
assert!(err_msg.contains("empty"), "Expected 'empty' in: {err_msg}");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_begin_write_transaction_empty_table_name_rejected() {
let (writer, _temp) = create_test_writer().await;
let columns = vec![ColumnDef::new("id", "int64", false).unwrap()];
let result = writer.begin_write_transaction("main", "", &columns, WriteMode::Replace);
assert!(result.is_err());
let err_msg = format!("{}", result.unwrap_err());
assert!(err_msg.contains("empty"), "Expected 'empty' in: {err_msg}");
}
#[tokio::test(flavor = "multi_thread")]
async fn test_begin_write_transaction_control_char_names_rejected() {
let (writer, _temp) = create_test_writer().await;
let columns = vec![ColumnDef::new("id", "int64", false).unwrap()];
let result =
writer.begin_write_transaction("bad\nschema", "table", &columns, WriteMode::Replace);
assert!(result.is_err());
let result =
writer.begin_write_transaction("main", "bad\ttable", &columns, WriteMode::Replace);
assert!(result.is_err());
}
}