refactor: table options

crates/catalog/src/session_catalog.rs

@@ -18,11 +18,7 @@ use protogen::metastore::types::catalog::{
     TableEntry,
     TunnelEntry,
 };
-use protogen::metastore::types::options::{
-    InternalColumnDefinition,
-    TableOptions,
-    TableOptionsInternal,
-};
+use protogen::metastore::types::options::{InternalColumnDefinition, TableOptionsInternal};
 use tracing::debug;
 
 use super::client::MetastoreClientHandle;
@@ -461,12 +457,13 @@ impl TempCatalog {
                     external: false,
                     is_temp: true,
                 },
-                options: TableOptions::Internal(TableOptionsInternal {
-                    columns: columns.to_owned(),
-                }),
+                options: TableOptionsInternal {
+                    columns: columns.clone(),
+                }
+                .into(),
                 tunnel_id: None,
                 access_mode: SourceAccessMode::ReadWrite,
-                columns: Some(columns.to_owned()),
+                columns: Some(columns),
             }
         })
     }
@@ -505,12 +502,13 @@ impl TempCatalog {
                     external: false,
                     is_temp: true,
                 },
-                options: TableOptions::Internal(TableOptionsInternal {
+                options: TableOptionsInternal {
                     columns: Vec::new(),
-                }),
+                }
+                .into(),
                 tunnel_id: None,
                 access_mode: SourceAccessMode::ReadWrite,
-                columns: Some(Vec::new()),
+                columns: None,
             });
         }
 

crates/datasources/src/bson/table.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use bson::RawDocumentBuf;
 use bytes::BytesMut;
-use datafusion::arrow::datatypes::{FieldRef, Schema};
+use datafusion::arrow::datatypes::Schema;
 use datafusion::datasource::streaming::StreamingTable;
 use datafusion::datasource::TableProvider;
 use datafusion::parquet::data_type::AsBytes;
@@ -20,12 +20,12 @@ use crate::object_store::{ObjStoreAccess, ObjStoreAccessor};
 pub async fn bson_streaming_table(
     store_access: Arc<dyn ObjStoreAccess>,
     source_url: DatasourceUrl,
-    fields: Option<Vec<FieldRef>>,
+    schema: Option<Schema>,
     schema_inference_sample_size: Option<i64>,
 ) -> Result<Arc<dyn TableProvider>, BsonError> {
     // TODO: set a maximum (1024?) or have an adaptive mode
     // (at least n but stop after n the same) or skip documents
-    let sample_size = if fields.is_some() {
+    let sample_size = if schema.is_some() {
         0
     } else {
         schema_inference_sample_size.unwrap_or(100)
@@ -93,8 +93,8 @@ pub async fn bson_streaming_table(
     let mut streams = Vec::<Arc<(dyn PartitionStream + 'static)>>::with_capacity(readers.len() + 1);
 
     // get the schema; if provided as an argument, just use that, otherwise, sample.
-    let schema = if let Some(fields) = fields {
-        Arc::new(Schema::new(fields))
+    let schema = if let Some(schema) = schema {
+        Arc::new(schema)
     } else {
         // iterate through the readers and build up a sample of the first <n>
         // documents to be used to infer the schema.

crates/datasources/src/debug/errors.rs

@@ -8,4 +8,5 @@ pub enum DebugError {
     InvalidTunnel(String),
 }
 
+
 pub type Result<T, E = DebugError> = std::result::Result<T, E>;

crates/datasources/src/lib.rs

@@ -1,4 +1,5 @@
 //! Data source implementations.
+
 pub mod bigquery;
 pub mod bson;
 pub mod cassandra;

crates/datasources/src/native/access.rs

@@ -30,7 +30,7 @@ use object_store::prefix::PrefixStore;
 use object_store::ObjectStore;
 use object_store_util::shared::SharedObjectStore;
 use protogen::metastore::types::catalog::TableEntry;
-use protogen::metastore::types::options::{TableOptions, TableOptionsInternal};
+use protogen::metastore::types::options::{TableOptionsInternal, TableOptionsV0};
 use serde_json::{json, Value};
 use url::Url;
 use uuid::Uuid;
@@ -262,12 +262,11 @@ impl NativeTableStorage {
         Ok(x.next().await.is_some())
     }
 
-    fn opts_from_ent(table: &TableEntry) -> Result<&TableOptionsInternal> {
-        let opts = match &table.options {
-            TableOptions::Internal(opts) => opts,
-            _ => return Err(NativeError::NotNative(table.clone())),
-        };
-        Ok(opts)
+    fn opts_from_ent(table: &TableEntry) -> Result<TableOptionsInternal> {
+        match table.options {
+            TableOptionsV0::Internal(ref opts) => Ok(opts.clone()),
+            _ => Err(NativeError::NotNative(table.clone())),
+        }
     }
 
     fn create_delta_store_for_table(&self, table: &TableEntry) -> Arc<dyn LogStore> {
@@ -498,11 +497,7 @@ mod tests {
     use deltalake::protocol::SaveMode;
     use object_store_util::conf::StorageConfig;
     use protogen::metastore::types::catalog::{EntryMeta, EntryType, SourceAccessMode, TableEntry};
-    use protogen::metastore::types::options::{
-        InternalColumnDefinition,
-        TableOptions,
-        TableOptionsInternal,
-    };
+    use protogen::metastore::types::options::{InternalColumnDefinition, TableOptionsInternal};
     use tempfile::tempdir;
     use url::Url;
     use uuid::Uuid;
@@ -533,13 +528,14 @@ mod tests {
                 external: false,
                 is_temp: false,
             },
-            options: TableOptions::Internal(TableOptionsInternal {
+            options: TableOptionsInternal {
                 columns: vec![InternalColumnDefinition {
                     name: "id".to_string(),
                     nullable: true,
                     arrow_type: DataType::Int32,
                 }],
-            }),
+            }
+            .into(),
             tunnel_id: None,
             access_mode: SourceAccessMode::ReadOnly,
             columns: None,

crates/datasources/src/object_store/mod.rs

@@ -25,7 +25,7 @@ use futures::StreamExt;
 use glob::{MatchOptions, Pattern};
 use object_store::path::Path as ObjectStorePath;
 use object_store::{ObjectMeta, ObjectStore};
-use protogen::metastore::types::options::{TableOptions, TableOptionsObjectStore};
+use protogen::metastore::types::options::{TableOptionsObjectStore, TableOptionsV0};
 
 use self::azure::AzureStoreAccess;
 use self::glob_util::{get_resolved_patterns, ResolvedPattern};
@@ -380,49 +380,49 @@ pub fn file_type_from_path(path: &ObjectStorePath) -> Result<FileType> {
 
 pub fn init_session_registry<'a>(
     runtime: &RuntimeEnv,
-    entries: impl Iterator<Item = &'a TableOptions>,
+    entries: impl Iterator<Item = &'a TableOptionsV0>,
 ) -> Result<()> {
     for opts in entries {
         let access: Arc<dyn ObjStoreAccess> = match opts {
-            TableOptions::Local(_) => Arc::new(LocalStoreAccess),
+            TableOptionsV0::Local(_) => Arc::new(LocalStoreAccess),
             // TODO: Consider consolidating Gcs, S3 and Delta and Iceberg `TableOptions` and
             // `ObjStoreAccess` since they largely overlap
-            TableOptions::Gcs(opts) => Arc::new(GcsStoreAccess {
+            TableOptionsV0::Gcs(opts) => Arc::new(GcsStoreAccess {
                 bucket: opts.bucket.clone(),
                 service_account_key: opts.service_account_key.clone(),
                 opts: HashMap::new(),
             }),
-            TableOptions::S3(opts) => Arc::new(S3StoreAccess {
+            TableOptionsV0::S3(opts) => Arc::new(S3StoreAccess {
                 bucket: opts.bucket.clone(),
                 region: Some(opts.region.clone()),
                 access_key_id: opts.access_key_id.clone(),
                 secret_access_key: opts.secret_access_key.clone(),
                 opts: HashMap::new(),
             }),
-            TableOptions::Azure(TableOptionsObjectStore {
+            TableOptionsV0::Azure(TableOptionsObjectStore {
                 location,
                 storage_options,
                 ..
             }) => {
                 let uri = DatasourceUrl::try_new(location)?;
                 Arc::new(AzureStoreAccess::try_from_uri(&uri, storage_options)?)
             }
-            TableOptions::Delta(TableOptionsObjectStore {
+            TableOptionsV0::Delta(TableOptionsObjectStore {
                 location,
                 storage_options,
                 ..
             })
-            | TableOptions::Iceberg(TableOptionsObjectStore {
+            | TableOptionsV0::Iceberg(TableOptionsObjectStore {
                 location,
                 storage_options,
                 ..
             })
-            | TableOptions::Lance(TableOptionsObjectStore {
+            | TableOptionsV0::Lance(TableOptionsObjectStore {
                 location,
                 storage_options,
                 ..
             })
-            | TableOptions::Bson(TableOptionsObjectStore {
+            | TableOptionsV0::Bson(TableOptionsObjectStore {
                 location,
                 storage_options,
                 ..
@@ -434,18 +434,18 @@ pub fn init_session_registry<'a>(
             // Continue on all others. Explicitly mentioning all the left
             // over options so we don't forget adding object stores that are
             // supported in the future (like azure).
-            TableOptions::Internal(_)
-            | TableOptions::Debug(_)
-            | TableOptions::Postgres(_)
-            | TableOptions::BigQuery(_)
-            | TableOptions::Mysql(_)
-            | TableOptions::MongoDb(_)
-            | TableOptions::Snowflake(_)
-            | TableOptions::SqlServer(_)
-            | TableOptions::Clickhouse(_)
-            | TableOptions::Cassandra(_)
-            | TableOptions::Excel(_)
-            | TableOptions::Sqlite(_) => continue,
+            TableOptionsV0::Internal(_)
+            | TableOptionsV0::Debug(_)
+            | TableOptionsV0::Postgres(_)
+            | TableOptionsV0::BigQuery(_)
+            | TableOptionsV0::Mysql(_)
+            | TableOptionsV0::MongoDb(_)
+            | TableOptionsV0::Snowflake(_)
+            | TableOptionsV0::SqlServer(_)
+            | TableOptionsV0::Clickhouse(_)
+            | TableOptionsV0::Cassandra(_)
+            | TableOptionsV0::Excel(_)
+            | TableOptionsV0::Sqlite(_) => continue,
         };
 
         let base_url = access.base_url()?;

crates/glaredb/tests/catalog_compat.rs

@@ -0,0 +1,37 @@
+mod setup;
+use crate::setup::make_cli;
+
+
+#[test]
+/// Assert that we can still load the old catalog without panicking
+fn test_catalog_backwards_compat() {
+    let mut cmd = make_cli();
+    let pwd = std::env::current_dir().unwrap();
+    let root_dir = pwd.parent().unwrap().parent().unwrap();
+    let old_catalog = root_dir.join("testdata/catalog_compat/v0");
+    cmd.args(["-l", old_catalog.to_str().unwrap()])
+        .assert()
+        .success();
+}
+
+
+#[test]
+/// Make sure that we can read the table options from the old catalog
+/// The v0 catalog has a table created from the following SQL:
+/// ```sql
+/// CREATE EXTERNAL TABLE debug_table
+/// FROM debug OPTIONS (
+///     table_type 'never_ending'
+/// );
+/// ```
+fn test_catalog_backwards_compat_tbl_options() {
+    let mut cmd = make_cli();
+    let pwd = std::env::current_dir().unwrap();
+    let root_dir = pwd.parent().unwrap().parent().unwrap();
+    let old_catalog = root_dir.join("testdata/catalog_compat/v0");
+
+    let query = "SELECT * FROM debug_table LIMIT 1";
+    cmd.args(["-l", old_catalog.to_str().unwrap(), "-q", query])
+        .assert()
+        .success();
+}

crates/metastore/src/client.rs

@@ -359,7 +359,6 @@ impl StatefulWorker {
                         .map_err(CatalogError::from),
                     Err(e) => Err(CatalogError::new(e.to_string())),
                 };
-
                 let result = match result {
                     Ok(resp) => {
                         let resp = resp.into_inner();