skade-flight/src/views.rs
//! **Session-scoped views** — `CREATE [OR REPLACE] [TEMP] VIEW` / `DROP VIEW`
//! that live in this server and never in the warehouse.
//!
//! # Why a read-only door holds views at all
//!
//! MEASURED (t14s, 2026-09-10, the 22-query TPC-H run over Flight SQL): 21/22.
//! Q15 failed because its spec form is three statements — `CREATE VIEW
//! revenue0 AS …`, a `SELECT` over the view, `DROP VIEW revenue0` — and the
//! first one came back as
//!
//! ```text
//! Internal: datafusion: External error: Unexpected => register_table does not
//! support tables with data.
//! ```
//!
//! That text is iceberg-datafusion refusing to put a view INTO THE CATALOG,
//! which is right: the warehouse is read-only from here. But "read-only" and
//! "cannot hold a view" are not the same fact. A view is a named query. It
//! writes nothing, and a client that is handed the spec's Q15 verbatim (a
//! benchmark harness, a BI tool's saved query) has no way to know a CTE
//! rewrite exists — the refusal did not say so, and its status was INTERNAL,
//! which reads as "the server broke", not "you may not".
//!
//! # The decision: engine-scoped, never persisted, re-planned on refresh
//!
//! Flight SQL carries no session on the wire. One bearer admits every client
//! of this process, and a `CommandStatementQuery` is two RPCs (`GetFlightInfo`,
//! `DoGet`) that need not even share a connection. So the honest scope for a
//! view is **the server process**: every client of this server sees it, it is
//! never written to the Iceberg catalog, and it dies with the process. That is
//! what "temp" means here; `TEMPORARY` is accepted and means the same thing.
//!
//! The views live OUTSIDE the DataFusion session, in a [`ViewStore`] the engine
//! owns, because the session is rebuilt every `refresh_after` (5 s) to see new
//! commits. A view registered in the session would vanish inside the window a
//! Q15 needs to run its three statements. Instead the engine's catalog is an
//! [`OverlayCatalog`]: the Iceberg provider underneath, the store's views laid
//! over its default schema. On every rebuild the definitions are planned again
//! against the fresh session, so a view keeps reading the CURRENT snapshot of
//! the tables it names.
//!
//! # What is and is not allowed, and where it is decided
//!
//! * `CREATE [OR REPLACE] [TEMP|TEMPORARY] VIEW name AS <select>` — allowed;
//! `name` unqualified or in the default schema; it may not shadow a
//! warehouse table; without `OR REPLACE` an existing view is an error.
//! * `DROP VIEW [IF EXISTS] name` — allowed; a warehouse table is not a view.
//! * Every other DDL, every DML, `COPY`, `SET` — refused with [`refusal`], which
//! names the rewrite: `WITH … AS (…)` or a session view. `INVALID_ARGUMENT`,
//! because it is the client's statement that is out of scope, not the server.
//!
//! DDL is applied on the EXECUTION RPC only (`DoGet` of the statement, or
//! `DoPut` of a `CommandStatementUpdate`) — never on `GetFlightInfo`, which
//! only describes. A driver that prepares, describes and then executes runs
//! the DDL once.
//!
//! `GetTables` does not advertise views: a BI navigator that saved a session
//! view in a report would find it gone after a restart, which is worse than
//! never having been offered it. Views are for the query text that created
//! them.
use std::collections::BTreeMap;
use std::sync::{Arc, RwLock};
use async_trait::async_trait;
use datafusion::catalog::{CatalogProvider, SchemaProvider, TableProvider};
use datafusion::datasource::view::ViewTable;
use datafusion::error::DataFusionError;
use datafusion::logical_expr::{DdlStatement, LogicalPlan};
use datafusion::prelude::SessionContext;
use datafusion::common::TableReference;
use crate::error::{Error, Result};
/// One session view: the statement that made it (re-planned on every session
/// rebuild) and the provider the current session resolves it to.
#[derive(Clone)]
struct View {
/// The full `CREATE … VIEW … AS …` text, as DataFusion's planner kept it.
definition: String,
table: Arc<dyn TableProvider>,
}
/// The engine's session views, by bare name. Shared by the engine (which
/// applies DDL and re-plans on rebuild) and every [`OverlaySchema`] the
/// sessions resolve names through.
#[derive(Default)]
pub struct ViewStore {
views: RwLock<BTreeMap<String, View>>,
}
impl ViewStore {
fn read(&self) -> std::sync::RwLockReadGuard<'_, BTreeMap<String, View>> {
self.views.read().unwrap_or_else(|p| p.into_inner())
}
fn write(&self) -> std::sync::RwLockWriteGuard<'_, BTreeMap<String, View>> {
self.views.write().unwrap_or_else(|p| p.into_inner())
}
/// The view's provider for the current session, if `name` is a view.
pub fn table(&self, name: &str) -> Option<Arc<dyn TableProvider>> {
self.read().get(name).map(|v| Arc::clone(&v.table))
}
pub fn contains(&self, name: &str) -> bool {
self.read().contains_key(name)
}
/// Every view name, sorted.
pub fn names(&self) -> Vec<String> {
self.read().keys().cloned().collect()
}
/// The `CREATE VIEW` text of every view, sorted by name — what a rebuild
/// plans again.
pub fn definitions(&self) -> Vec<(String, String)> {
self.read().iter().map(|(n, v)| (n.clone(), v.definition.clone())).collect()
}
fn insert(&self, name: String, definition: String, table: Arc<dyn TableProvider>) {
self.write().insert(name, View { definition, table });
}
fn remove(&self, name: &str) -> bool {
self.write().remove(name).is_some()
}
}
/// A view-aware `CREATE`/`DROP` the engine will apply, cut out of a planned
/// statement by [`classify`].
#[derive(Debug, Clone)]
pub enum ViewDdl {
Create {
/// Bare view name (qualification already checked against the default schema).
name: String,
or_replace: bool,
/// The full statement text, kept for re-planning after a session rebuild.
definition: String,
/// The view's query, planned against the session that classified it.
input: Arc<LogicalPlan>,
},
Drop {
name: String,
if_exists: bool,
},
}
/// What a statement IS, once planned: a query to run, a session-view DDL to
/// apply, or something this door refuses.
pub enum Classified {
Query(LogicalPlan),
View(ViewDdl),
}
/// Split a planned statement into what the engine does with it.
///
/// `default_schema` is the one schema a view may live in (unqualified names
/// resolve there); a view named in any other schema or catalog is refused, so
/// no client can be led to think a view landed in `skade.tpch`.
pub fn classify(plan: LogicalPlan, catalog: &str, default_schema: &str) -> Result<Classified> {
match plan {
LogicalPlan::Ddl(DdlStatement::CreateView(cv)) => {
let name = bare_view_name(&cv.name, catalog, default_schema)?;
let definition = cv.definition.unwrap_or_default();
Ok(Classified::View(ViewDdl::Create { name, or_replace: cv.or_replace, definition, input: cv.input }))
}
LogicalPlan::Ddl(DdlStatement::DropView(dv)) => {
let name = bare_view_name(&dv.name, catalog, default_schema)?;
Ok(Classified::View(ViewDdl::Drop { name, if_exists: dv.if_exists }))
}
LogicalPlan::Ddl(ddl) => Err(Error::Sql(refusal(&ddl_kind(&ddl)))),
LogicalPlan::Dml(dml) => Err(Error::Sql(refusal(&format!("{}", dml.op)))),
LogicalPlan::Copy(_) => Err(Error::Sql(refusal("COPY"))),
LogicalPlan::Statement(st) => Err(Error::Sql(refusal(st.name()))),
other => Ok(Classified::Query(other)),
}
}
fn ddl_kind(ddl: &DdlStatement) -> String {
match ddl {
DdlStatement::CreateExternalTable(_) => "CREATE EXTERNAL TABLE".into(),
DdlStatement::CreateMemoryTable(_) => "CREATE TABLE".into(),
DdlStatement::CreateCatalogSchema(_) => "CREATE SCHEMA".into(),
DdlStatement::CreateCatalog(_) => "CREATE DATABASE".into(),
DdlStatement::CreateIndex(_) => "CREATE INDEX".into(),
DdlStatement::CreateFunction(_) => "CREATE FUNCTION".into(),
DdlStatement::DropTable(_) => "DROP TABLE".into(),
DdlStatement::DropCatalogSchema(_) => "DROP SCHEMA".into(),
DdlStatement::DropFunction(_) => "DROP FUNCTION".into(),
other => other.name().to_string(),
}
}
/// **The refusal a write gets** — it says what this door is, and what IS
/// allowed, so the reader has the rewrite in hand instead of "forbidden".
///
/// Pure, so the wording is testable: every refusal names the CTE form and the
/// session view, because those are the two ways a Q15-shaped statement is
/// honoured here.
pub fn refusal(what: &str) -> String {
format!(
"skade-flight is read-only over Flight SQL: {what} is refused, the warehouse is never \
written from here. What IS honoured: any SELECT, including WITH <name> AS (<select>) \
common table expressions; and session-scoped views — CREATE [OR REPLACE] [TEMP] VIEW \
<name> AS <select> / DROP VIEW [IF EXISTS] <name> — which live in this server's \
session, never in the catalog, and end with the process"
)
}
/// The bare name of a view that lives in the default schema, or a refusal
/// naming the schema it was asked for.
fn bare_view_name(r: &TableReference, catalog: &str, default_schema: &str) -> Result<String> {
let ok_schema = |s: &str| s == default_schema;
let ok_catalog = |c: &str| c == catalog;
match r {
TableReference::Bare { table } => Ok(table.to_string()),
TableReference::Partial { schema, table } if ok_schema(schema) => Ok(table.to_string()),
TableReference::Full { catalog: c, schema, table } if ok_catalog(c) && ok_schema(schema) => Ok(table.to_string()),
other => Err(Error::Sql(format!(
"a session view lives in the default schema `{catalog}.{default_schema}` only; `{other}` \
names another — use the bare name, or `{default_schema}.<name>`"
))),
}
}
/// The catalog the engine mounts: the Iceberg provider, with the store's views
/// laid over its default schema.
pub struct OverlayCatalog {
inner: Arc<dyn CatalogProvider>,
default_schema: String,
views: Arc<ViewStore>,
}
impl OverlayCatalog {
pub fn new(inner: Arc<dyn CatalogProvider>, default_schema: &str, views: Arc<ViewStore>) -> Self {
Self { inner, default_schema: default_schema.to_string(), views }
}
}
impl std::fmt::Debug for OverlayCatalog {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OverlayCatalog").field("default_schema", &self.default_schema).finish_non_exhaustive()
}
}
impl CatalogProvider for OverlayCatalog {
fn schema_names(&self) -> Vec<String> {
self.inner.schema_names()
}
fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
let inner = self.inner.schema(name)?;
if name == self.default_schema {
Some(Arc::new(OverlaySchema { inner, views: Arc::clone(&self.views) }))
} else {
Some(inner)
}
}
}
/// The default schema with the session views laid over the Iceberg tables.
/// A view resolves first — but [`Engine`](crate::Engine) refuses to CREATE one
/// over an existing table, so nothing is ever shadowed.
struct OverlaySchema {
inner: Arc<dyn SchemaProvider>,
views: Arc<ViewStore>,
}
impl std::fmt::Debug for OverlaySchema {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OverlaySchema").finish_non_exhaustive()
}
}
#[async_trait]
impl SchemaProvider for OverlaySchema {
fn table_names(&self) -> Vec<String> {
let mut names = self.inner.table_names();
names.extend(self.views.names());
names
}
async fn table(&self, name: &str) -> std::result::Result<Option<Arc<dyn TableProvider>>, DataFusionError> {
if let Some(v) = self.views.table(name) {
return Ok(Some(v));
}
self.inner.table(name).await
}
fn table_exist(&self, name: &str) -> bool {
self.views.contains(name) || self.inner.table_exist(name)
}
}
/// Apply a view DDL to `store`, resolving names through `ctx` (the CURRENT
/// session) so a view can neither shadow a warehouse table nor be dropped
/// when the name is a table.
pub async fn apply(store: &ViewStore, ctx: &SessionContext, catalog: &str, default_schema: &str, ddl: ViewDdl) -> Result<()> {
let is_table = |name: &str| -> bool {
ctx.catalog(catalog)
.and_then(|c| c.schema(default_schema))
.map(|s| !store.contains(name) && s.table_exist(name))
.unwrap_or(false)
};
match ddl {
ViewDdl::Create { name, or_replace, definition, input } => {
if is_table(&name) {
return Err(Error::Sql(format!(
"`{name}` is a table in the warehouse; a session view cannot shadow it — pick another name"
)));
}
if store.contains(&name) && !or_replace {
return Err(Error::Sql(format!("view `{name}` already exists — CREATE OR REPLACE VIEW replaces it")));
}
let table = view_table(ctx, Arc::unwrap_or_clone(input), &definition)?;
store.insert(name, definition, table);
Ok(())
}
ViewDdl::Drop { name, if_exists } => {
if store.remove(&name) || if_exists {
return Ok(());
}
if is_table(&name) {
return Err(Error::Sql(format!("`{name}` is a table in the warehouse, not a session view; the warehouse is read-only from here")));
}
Err(Error::Sql(format!("no such view `{name}`")))
}
}
}
/// Build the provider for a view's planned query — with the same type
/// coercion DataFusion's own `create_view` applies, so a `CAST`-needing
/// projection behaves identically to a session-registered view.
fn view_table(ctx: &SessionContext, input: LogicalPlan, definition: &str) -> Result<Arc<dyn TableProvider>> {
use datafusion::common::config::ConfigOptions;
use datafusion::optimizer::Analyzer;
let _ = ctx;
// The same rule set the session runs (TIME literals lowered BEFORE
// coercion), so a view body behaves exactly like the same text in a query.
let coerced = Analyzer::with_rules(crate::time_literal::analyzer_rules())
.execute_and_check(input, &ConfigOptions::default(), |_, _| {})
.map_err(Error::from_df)?;
Ok(Arc::new(ViewTable::new(coerced, Some(definition.to_string()))))
}
/// After a session rebuild: plan every stored definition again against the
/// fresh `ctx`, so each view reads the current snapshot. A definition that no
/// longer plans (its table was dropped) is removed, and the reason is returned
/// so the caller can log it — one line per lost view, never a silent hole.
pub async fn replan(store: &ViewStore, ctx: &SessionContext, catalog: &str, default_schema: &str) -> Vec<String> {
let mut lost = Vec::new();
for (name, definition) in store.definitions() {
let planned = ctx.state().create_logical_plan(&definition).await.map_err(Error::from_df);
let table = match planned.and_then(|p| classify(p, catalog, default_schema)) {
Ok(Classified::View(ViewDdl::Create { input, .. })) => view_table(ctx, Arc::unwrap_or_clone(input), &definition),
Ok(_) => Err(Error::Sql(format!("stored definition of `{name}` no longer plans as a view"))),
Err(e) => Err(e),
};
match table {
Ok(t) => store.insert(name, definition, t),
Err(e) => {
store.remove(&name);
lost.push(format!("session view `{name}` dropped on refresh — its definition no longer plans: {e}"));
}
}
}
lost
}
#[cfg(test)]
mod tests {
use super::*;
/// **Every refusal hands the reader the rewrite.** RED direction: drop
/// either clause from [`refusal`] and the matching assertion fails.
#[test]
fn the_refusal_names_what_is_honoured_not_only_what_is_not() {
let m = refusal("DELETE");
assert!(m.starts_with("skade-flight is read-only"), "{m}");
assert!(m.contains("DELETE is refused"), "names the statement: {m}");
assert!(m.contains("WITH <name> AS (<select>)"), "the CTE rewrite: {m}");
assert!(m.contains("CREATE [OR REPLACE] [TEMP] VIEW"), "the session view: {m}");
assert!(m.contains("never in the catalog"), "and its scope: {m}");
}
/// **A view name is bare, or in the default schema — never elsewhere.**
#[test]
fn a_view_lives_in_the_default_schema_only() {
let ok = |r: TableReference| bare_view_name(&r, "skade", "main").unwrap();
assert_eq!(ok(TableReference::bare("v")), "v");
assert_eq!(ok(TableReference::partial("main", "v")), "v");
assert_eq!(ok(TableReference::full("skade", "main", "v")), "v");
for bad in [TableReference::partial("tpch", "v"), TableReference::full("other", "main", "v")] {
let e = bare_view_name(&bad, "skade", "main").unwrap_err().to_string();
assert!(e.contains("default schema `skade.main` only"), "{e}");
}
}
/// **Classification splits the three fates, and refuses by name.**
#[tokio::test]
async fn classify_splits_query_view_ddl_and_refusal() {
let ctx = SessionContext::new();
let plan = |sql: &str| {
let ctx = ctx.clone();
let sql = sql.to_string();
async move { ctx.state().create_logical_plan(&sql).await.unwrap() }
};
assert!(matches!(classify(plan("SELECT 1").await, "skade", "main"), Ok(Classified::Query(_))));
match classify(plan("CREATE OR REPLACE TEMP VIEW v AS SELECT 1 AS x").await, "skade", "main") {
Ok(Classified::View(ViewDdl::Create { name, or_replace, definition, .. })) => {
assert_eq!(name, "v");
assert!(or_replace);
assert!(definition.to_uppercase().contains("CREATE"), "the statement text is kept: {definition}");
}
_ => panic!("CREATE VIEW must classify as a view DDL"),
}
assert!(matches!(
classify(plan("DROP VIEW IF EXISTS v").await, "skade", "main"),
Ok(Classified::View(ViewDdl::Drop { if_exists: true, .. }))
));
for sql in ["CREATE TABLE t (x INT)", "DROP TABLE t", "CREATE SCHEMA s", "SET datafusion.execution.batch_size = 1"] {
let e = classify(plan(sql).await, "skade", "main").err().map(|e| e.to_string()).unwrap_or_default();
assert!(e.contains("read-only"), "{sql}: {e}");
}
}
}