fix: refuse a CREATE on a name a table already holds

CHANGELOG.md

@@ -317,6 +317,31 @@ tag. Releases before 2.0.0 are recorded in the
 
 ### Fixed
 
+- **The default table takes the name you configured.** At startup Beacon registers an empty
+  stand-in table, so a JSON query without a `from` field reports no missing table. The stand-in
+  ignored `BEACON_DEFAULT_TABLE` and always took the literal name `default`. A server started with
+  `BEACON_DEFAULT_TABLE=observations` therefore held a table called `default`, left `observations`
+  missing, and answered a `from`-less query with a missing-table error. Startup now registers the
+  stand-in under the configured name. The rest of the rule is unchanged and now under test: Beacon
+  fills that name only when the name is free, so your own table under it survives a restart, and a
+  `CREATE` on a name a table holds still fails. That last error now names the stand-in and tells
+  you to run `DROP TABLE` first, because a table nobody made is a confusing thing to collide with.
+  See [Configuration](docs/docs/2.0.0-rc5/server/configuration.md#the-default-table).
+- **`CREATE EXTERNAL TABLE` and `CREATE VIEW` no longer discard the table under the name.** Both
+  registered straight over whatever held the name. `CREATE EXTERNAL TABLE obs STORED AS CSV
+  LOCATION 'other/'` therefore repointed an existing `obs` with no warning, and a second
+  `CREATE VIEW v` swapped the view a report depended on. Only `CREATE TABLE` and
+  `CREATE MATERIALIZED VIEW` checked first, so the same typo either failed or destroyed a table
+  depending on which statement carried it. The admin API said as much and did not do it: the
+  `if_not_exists` field of `POST /api/admin/external-tables` is documented as skipping "instead of
+  erroring", and nothing ever errored. Both statements now refuse a name that a table or a view
+  holds, as `CREATE TABLE` does. The two modifiers the SQL reference already documented now do the
+  work: `CREATE EXTERNAL TABLE IF NOT EXISTS` keeps the existing table and reports success, which
+  is what that field always promised, and `CREATE OR REPLACE EXTERNAL TABLE` overwrites it.
+  `CREATE OR REPLACE VIEW` swaps a view. Neither guard reaches the paths that replace a provider on
+  purpose: a materialized-view `REFRESH`, an `ALTER TABLE`, and a crawler that re-registers a table
+  it owns all register directly and are unchanged. A script that relied on a bare re-`CREATE` to
+  repoint a table needs `OR REPLACE` added, or a `DROP TABLE` in front of it.
 - **A long query no longer makes the API unreachable.** The HTTP API, Flight SQL and every query
   shared one Tokio runtime of `BEACON_WORKER_THREADS` threads. A scan holds a thread until a
   partition yields, and one query starts as many partitions as the machine has cores, so a long

beacon-db/beacon-core/src/schema_persistence/default_table.rs

@@ -0,0 +1,99 @@
+//! The stand-in for the configured default table.
+//!
+//! `sql.default_table` names the table a JSON query without a `from` resolves
+//! against. Beacon registers an empty stand-in under that name so such a query
+//! plans on a fresh database, and drops the stand-in as soon as a real table
+//! takes the name.
+//!
+//! A `CREATE` statement on that name fails, as it does on any other table that
+//! exists. Run `DROP TABLE` first to take the name.
+//!
+//! The stand-in is a distinct type, not a bare
+//! [`EmptyTable`](datafusion::datasource::empty::EmptyTable), so the `CREATE`
+//! paths recognize it and say that in the error.
+
+use std::{any::Any, sync::Arc};
+
+use arrow::datatypes::{Schema, SchemaRef};
+use datafusion::{
+    catalog::{Session, TableProvider},
+    datasource::{empty::EmptyTable, TableType},
+    error::Result as DataFusionResult,
+    logical_expr::Expr,
+    physical_plan::ExecutionPlan,
+    prelude::SessionContext,
+    sql::TableReference,
+};
+
+/// An empty, column-less table that holds the configured default-table name
+/// until a real table takes it.
+#[derive(Debug)]
+pub struct DefaultTablePlaceholder {
+    inner: EmptyTable,
+}
+
+impl DefaultTablePlaceholder {
+    pub fn new() -> Self {
+        Self {
+            inner: EmptyTable::new(Arc::new(Schema::empty())),
+        }
+    }
+}
+
+impl Default for DefaultTablePlaceholder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait::async_trait]
+impl TableProvider for DefaultTablePlaceholder {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        self.inner.schema()
+    }
+
+    fn table_type(&self) -> TableType {
+        self.inner.table_type()
+    }
+
+    async fn scan(
+        &self,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        filters: &[Expr],
+        limit: Option<usize>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        self.inner.scan(state, projection, filters, limit).await
+    }
+}
+
+/// True when `name` holds nothing but the default-table stand-in.
+pub async fn holds_placeholder(session_ctx: &SessionContext, name: TableReference) -> bool {
+    match session_ctx.table_provider(name).await {
+        Ok(provider) => provider.as_any().is::<DefaultTablePlaceholder>(),
+        Err(_) => false,
+    }
+}
+
+/// The reason a `CREATE` statement refuses `name`.
+///
+/// A name the stand-in holds looks free to the user, because no one made that
+/// table. The message therefore names the stand-in and says how to free the name.
+pub async fn already_exists_error(
+    session_ctx: &SessionContext,
+    name: TableReference,
+    subject: &str,
+) -> anyhow::Error {
+    let table = name.table().to_string();
+    if holds_placeholder(session_ctx, name).await {
+        return anyhow::anyhow!(
+            "{subject} '{table}' already exists. Beacon creates it at startup as the default \
+             table. Run `DROP TABLE \"{table}\"` first to take the name."
+        );
+    }
+    anyhow::anyhow!("{subject} '{table}' already exists")
+}

beacon-db/beacon-core/src/schema_persistence/mod.rs

@@ -3,6 +3,8 @@
 //! Beacon's catalog is the source of truth for which tables exist. This module owns:
 //! - [`PersistentSchemaProvider`] — the `beacon.public` schema provider that persists
 //!   a table's definition on registration and removes it on deregistration;
+//! - [`DefaultTablePlaceholder`] — the empty stand-in that holds the configured
+//!   default-table name until a real table takes it;
 //! - [`SchemaPersistenceService`] — the durable `db://<name>/table.json` read/write path;
 //! - [`init_tables`] — startup recovery that rebuilds every provider from those files;
 //! - the private `loading`/`ordering` helpers `init_tables` drives.
@@ -12,11 +14,13 @@ use std::{collections::HashMap, sync::Arc};
 use beacon_datafusion_ext::table_ext::TableDefinition;
 use datafusion::{execution::object_store::ObjectStoreUrl, prelude::SessionContext};
 
+pub mod default_table;
 mod loading;
 mod ordering;
 pub mod provider;
 pub mod service;
 
+pub use default_table::DefaultTablePlaceholder;
 pub use provider::PersistentSchemaProvider;
 pub use service::{definition_from_provider, SchemaPersistenceService};
 
@@ -27,7 +31,8 @@ pub use service::{definition_from_provider, SchemaPersistenceService};
 /// definitions, then views in dependency order), builds each provider against
 /// the live session — so a view's defining query resolves the tables already
 /// registered ahead of it — and inserts it into `schema` without re-persisting.
-/// Finishes by ensuring the empty `default` table exists.
+/// Finishes by registering the empty stand-in for the configured default table,
+/// but only when no loaded table already holds that name.
 ///
 /// `tables_store_url` is the store the persisted `<name>/table.json` definitions
 /// are read from (the caller supplies it — the runtime uses its tables store).
@@ -70,6 +75,7 @@ pub async fn init_tables(
         }
     }
 
-    schema.ensure_default_table();
+    let default_table = crate::settings::SqlSettings::from_session(session_ctx).default_table;
+    schema.ensure_default_table(&default_table);
     Ok(())
 }

beacon-db/beacon-core/src/schema_persistence/provider.rs

@@ -12,17 +12,16 @@ use std::{
     sync::{Arc, Weak},
 };
 
-use arrow::datatypes::Schema;
 use datafusion::{
     catalog::{MemorySchemaProvider, SchemaProvider, TableProvider},
-    datasource::empty::EmptyTable,
     error::DataFusionError,
     execution::object_store::ObjectStoreUrl,
     prelude::SessionContext,
 };
 
 use beacon_datafusion_ext::table_ext::INTERNAL_TABLE_PREFIX;
 
+use super::default_table::DefaultTablePlaceholder;
 use super::service::SchemaPersistenceService;
 
 /// Schema provider for `beacon.public` that persists table definitions on
@@ -127,16 +126,18 @@ impl PersistentSchemaProvider {
         self.inner.register_table(name, table)
     }
 
-    /// Register the in-memory `default` table backed by an empty provider.
+    /// Register the in-memory stand-in for `name`, the configured default table.
     ///
-    /// The default table is not persisted; it is recreated on every startup so
-    /// queries against the configured default table always resolve.
-    pub fn ensure_default_table(&self) {
-        if self.inner.table_exist("default") {
+    /// A no-op when `name` already holds a table, so a table the user created
+    /// under that name keeps the name. The stand-in itself is not persisted; it
+    /// is recreated on every startup, so a query against the default table
+    /// always plans.
+    pub fn ensure_default_table(&self, name: &str) {
+        if self.inner.table_exist(name) {
             return;
         }
-        let provider: Arc<dyn TableProvider> = Arc::new(EmptyTable::new(Arc::new(Schema::empty())));
-        let _ = self.inner.register_table("default".to_string(), provider);
+        let provider: Arc<dyn TableProvider> = Arc::new(DefaultTablePlaceholder::new());
+        let _ = self.inner.register_table(name.to_string(), provider);
     }
 }
 
@@ -312,8 +313,8 @@ mod tests {
         let (provider, _ctx, _store) = fixture();
         assert!(!provider.table_exist("default"));
 
-        provider.ensure_default_table();
-        provider.ensure_default_table();
+        provider.ensure_default_table("default");
+        provider.ensure_default_table("default");
 
         assert!(provider.table_exist("default"));
         assert_eq!(
@@ -326,4 +327,38 @@ mod tests {
             "the default table should be registered exactly once"
         );
     }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn ensure_default_table_uses_the_configured_name() {
+        let (provider, _ctx, _store) = fixture();
+
+        provider.ensure_default_table("observations");
+
+        assert!(provider.table_exist("observations"));
+        assert!(
+            !provider.table_exist("default"),
+            "the stand-in takes the configured name, not the literal 'default'"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn ensure_default_table_keeps_an_existing_table() {
+        let (provider, ctx, _store) = fixture();
+        provider
+            .register_table("default".to_string(), view(&ctx, "SELECT 1 AS x").await)
+            .expect("registration should succeed");
+
+        provider.ensure_default_table("default");
+
+        let table = provider
+            .table("default")
+            .await
+            .expect("lookup should succeed")
+            .expect("table should be present");
+        assert_eq!(
+            table.schema().field(0).name(),
+            "x",
+            "the user's table keeps the name; the stand-in must not replace it"
+        );
+    }
 }

beacon-db/beacon-core/src/statement_plan/actions.rs

@@ -218,10 +218,26 @@ pub(crate) async fn show_secrets(
 
 /// Register a `CREATE EXTERNAL TABLE` via the listing-table factory and persist
 /// it to the catalog.
+///
+/// Errors if a table with `cmd.name` already exists. `IF NOT EXISTS` keeps the
+/// existing table instead, and `OR REPLACE` overwrites it. The guard covers every
+/// `STORED AS` variant below, so no variant replaces a table the user did not drop.
 pub(crate) async fn create_external_table(
     session: &Arc<SessionContext>,
     cmd: &CreateExternalTable,
 ) -> anyhow::Result<()> {
+    if !cmd.or_replace && session.table_exist(cmd.name.clone())? {
+        if cmd.if_not_exists {
+            return Ok(());
+        }
+        return Err(crate::schema_persistence::default_table::already_exists_error(
+            session,
+            cmd.name.clone(),
+            "Table",
+        )
+        .await);
+    }
+
     // `STORED AS REMOTE` registers a federated table pointing at another Beacon
     // instance, rather than a listing table over the datasets store.
     if cmd.file_type.eq_ignore_ascii_case("REMOTE") {
@@ -491,12 +507,25 @@ async fn create_sql_db_table(
 }
 
 /// Register a `CREATE VIEW` as a `ViewTable` (re-plans its query on each scan).
-pub(crate) fn create_view(
+///
+/// Errors if a table or view with `name` already exists. `CREATE OR REPLACE VIEW`
+/// (`or_replace`) swaps it instead.
+pub(crate) async fn create_view(
     session: &Arc<SessionContext>,
     name: &TableReference,
     input: &LogicalPlan,
     definition: &Option<String>,
+    or_replace: bool,
 ) -> anyhow::Result<()> {
+    if !or_replace && session.table_exist(name.clone())? {
+        return Err(crate::schema_persistence::default_table::already_exists_error(
+            session,
+            name.clone(),
+            "View",
+        )
+        .await);
+    }
+
     let table = ViewTable::new(input.clone(), definition.clone());
     session.register_table(name.clone(), Arc::new(table))?;
     Ok(())
@@ -519,7 +548,12 @@ pub(crate) async fn create_table(
         if if_not_exists {
             return Ok(None);
         }
-        return Err(anyhow::anyhow!("Table '{table_name}' already exists"));
+        return Err(crate::schema_persistence::default_table::already_exists_error(
+            session,
+            name.clone(),
+            "Table",
+        )
+        .await);
     }
 
     let arrow_schema = child.schema();

beacon-db/beacon-core/src/statement_plan/materialized_view.rs

@@ -44,7 +44,12 @@ pub(crate) async fn create_materialized_view(
     let table_ref = crate::table_name::table_reference(name);
 
     if session_ctx.table_exist(table_ref.clone())? {
-        return Err(anyhow::anyhow!("Materialized view '{name}' already exists"));
+        return Err(crate::schema_persistence::default_table::already_exists_error(
+            session_ctx,
+            table_ref.clone(),
+            "Materialized view",
+        )
+        .await);
     }
 
     // Execute the defining query and persist its result as a single Parquet file

beacon-db/beacon-core/src/statement_plan/physical.rs

@@ -650,6 +650,7 @@ pub(crate) struct CreateViewExec {
     name: TableReference,
     input: LogicalPlan,
     definition: Option<String>,
+    or_replace: bool,
     session: SessionCell,
     cache: Arc<PlanProperties>,
 }
@@ -659,12 +660,14 @@ impl CreateViewExec {
         name: TableReference,
         input: LogicalPlan,
         definition: Option<String>,
+        or_replace: bool,
         session: SessionCell,
     ) -> Self {
         Self {
             name,
             input,
             definition,
+            or_replace,
             session,
             cache: Arc::new(side_effect_properties()),
         }
@@ -679,8 +682,11 @@ side_effect_exec!(CreateViewExec, "CreateViewExec", |exec: &CreateViewExec| {
     let name = exec.name.clone();
     let input = exec.input.clone();
     let definition = exec.definition.clone();
+    let or_replace = exec.or_replace;
     Ok(side_effect_stream(async move {
-        actions::create_view(&session, &name, &input, &definition).map_err(to_df_err)
+        actions::create_view(&session, &name, &input, &definition, or_replace)
+            .await
+            .map_err(to_df_err)
     }))
 });
 

beacon-db/beacon-core/src/statement_plan/query_planner.rs

@@ -74,6 +74,7 @@ impl QueryPlanner for BeaconQueryPlanner {
                     view.name.clone(),
                     view.input.as_ref().clone(),
                     view.definition.clone(),
+                    view.or_replace,
                     session,
                 )))
             }