refactor: table options

crates/metastore/Cargo.toml

@@ -11,7 +11,6 @@ ioutil = { path = "../ioutil" }
 logutil = { path = "../logutil" }
 protogen = { path = "../protogen" }
 sqlbuiltins = { path = "../sqlbuiltins" }
-datasources = { path = "../datasources" }
 object_store_util = { path = "../object_store_util" }
 pgrepr = { path = "../pgrepr" }
 async-trait = { workspace = true }

crates/protogen/proto/metastore/catalog.proto

@@ -39,6 +39,12 @@ import "common/arrow.proto";
 
 // The state of the catalog at some version.
 message CatalogState {
+  // The version of the actual catalog implementation.
+  // This is incremented when there are incompatible changes to the physical
+  // catalog layout.
+  // Purposely set to a high number to avoid accidental collisions
+  optional uint32 catalog_version = 2047;
+
   // Version of this catalog. Increments on every mutation.
   uint64 version = 1;
 
@@ -50,12 +56,6 @@ message CatalogState {
   // Metadata for the deployment.
   DeploymentMetadata deployment = 3;
   // next: 4
-
-  // The version of the actual catalog implementation.
-  // This is incremented when there are incompatible changes to the physical
-  // catalog layout.
-  // Purposely set to a high number to avoid accidental collisions
-  optional uint32 catalog_version = 2047;
 }
 
 // Metadata for the deployment.

crates/protogen/proto/metastore/options.proto

@@ -122,6 +122,7 @@ message StorageOptions {
 message TableOptionsV1 {
   string name = 1;
   bytes options = 2;
+  uint32 version = 3;
 }
 
 // Table options

crates/protogen/src/metastore/types/options.rs

@@ -69,7 +69,6 @@ impl TryFrom<options::InternalColumnDefinition> for InternalColumnDefinition {
     fn try_from(value: options::InternalColumnDefinition) -> Result<Self, Self::Error> {
         let arrow_type: DataType = value.arrow_type.as_ref().required("arrow_type")?;
 
-
         Ok(InternalColumnDefinition {
             name: value.name,
             nullable: value.nullable,
@@ -592,7 +591,6 @@ impl StorageOptions {
     }
 }
 
-
 impl TryFrom<options::StorageOptions> for StorageOptions {
     type Error = ProtoConvError;
     fn try_from(value: options::StorageOptions) -> Result<Self, Self::Error> {
@@ -610,13 +608,14 @@ impl From<StorageOptions> for options::StorageOptions {
 pub struct TableOptionsV1 {
     pub name: String,
     pub options: Vec<u8>,
+    pub version: u32,
 }
 
 pub trait TableOptionsImpl: Serialize + DeserializeOwned {
     const NAME: &'static str;
+    const VERSION: u32 = 1;
 }
 
-
 impl Display for TableOptionsV1 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.name)
@@ -634,6 +633,7 @@ impl TableOptionsV1 {
         TableOptionsV1 {
             name: S::NAME.to_string(),
             options,
+            version: S::VERSION,
         }
     }
 
@@ -667,7 +667,6 @@ where
     }
 }
 
-
 impl TryFrom<options::TableOptionsV1> for TableOptionsV1 {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsV1) -> Result<Self, Self::Error> {
@@ -677,6 +676,7 @@ impl TryFrom<options::TableOptionsV1> for TableOptionsV1 {
         Ok(TableOptionsV1 {
             options: values,
             name,
+            version: value.version,
         })
     }
 }
@@ -686,11 +686,11 @@ impl From<TableOptionsV1> for options::TableOptionsV1 {
         options::TableOptionsV1 {
             name: value.name,
             options: value.options,
+            version: value.version,
         }
     }
 }
 
-
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum TableOptionsV0 {
     Debug(TableOptionsDebug),
@@ -800,7 +800,6 @@ impl TryFrom<&TableOptionsV1> for TableOptionsV0 {
     fn try_from(value: &TableOptionsV1) -> Result<Self, Self::Error> {
         let _v: serde_json::Value = serde_json::from_slice(&value.options).unwrap();
 
-
         match value.name.as_ref() {
             TableOptionsObjectStore::NAME => {
                 let obj_store: TableOptionsObjectStore = value.extract()?;
@@ -1031,7 +1030,6 @@ impl TableOptionsImpl for TableOptionsInternal {
     const NAME: &'static str = "internal";
 }
 
-
 impl From<DFSchemaRef> for TableOptionsInternal {
     fn from(value: DFSchemaRef) -> Self {
         TableOptionsInternal {
@@ -1103,7 +1101,6 @@ impl TableOptionsImpl for TableOptionsPostgres {
     const NAME: &'static str = "postgres";
 }
 
-
 impl TryFrom<options::TableOptionsPostgres> for TableOptionsPostgres {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsPostgres) -> Result<Self, Self::Error> {
@@ -1143,7 +1140,6 @@ impl TableOptionsImpl for TableOptionsBigQuery {
     const NAME: &'static str = "bigquery";
 }
 
-
 impl TryFrom<options::TableOptionsBigQuery> for TableOptionsBigQuery {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsBigQuery) -> Result<Self, Self::Error> {
@@ -1184,7 +1180,6 @@ impl TableOptionsImpl for TableOptionsMysql {
     const NAME: &'static str = "mysql";
 }
 
-
 impl TryFrom<options::TableOptionsMysql> for TableOptionsMysql {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsMysql) -> Result<Self, Self::Error> {
@@ -1223,7 +1218,6 @@ impl TableOptionsImpl for TableOptionsLocal {
     const NAME: &'static str = "local";
 }
 
-
 impl TryFrom<options::TableOptionsLocal> for TableOptionsLocal {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsLocal) -> Result<Self, Self::Error> {
@@ -1355,7 +1349,6 @@ impl TableOptionsImpl for TableOptionsMongoDb {
     const NAME: &'static str = "mongo";
 }
 
-
 impl TryFrom<options::TableOptionsMongo> for TableOptionsMongoDb {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsMongo) -> Result<Self, Self::Error> {
@@ -1397,7 +1390,6 @@ impl TableOptionsImpl for TableOptionsExcel {
     const NAME: &'static str = "excel";
 }
 
-
 impl TryFrom<options::TableOptionsExcel> for TableOptionsExcel {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsExcel) -> Result<Self, Self::Error> {
@@ -1442,7 +1434,6 @@ impl TableOptionsImpl for TableOptionsSqlServer {
     const NAME: &'static str = "sql_server";
 }
 
-
 impl TryFrom<options::TableOptionsSqlServer> for TableOptionsSqlServer {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsSqlServer) -> Result<Self, Self::Error> {
@@ -1562,7 +1553,6 @@ impl TableOptionsImpl for TableOptionsSqlite {
     const NAME: &'static str = "sqlite";
 }
 
-
 impl TryFrom<options::TableOptionsSqlite> for TableOptionsSqlite {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsSqlite) -> Result<Self, Self::Error> {
@@ -1644,12 +1634,10 @@ pub struct TableOptionsObjectStore {
     pub schema_sample_size: Option<i64>,
 }
 
-
 impl TableOptionsImpl for TableOptionsObjectStore {
     const NAME: &'static str = "object_store";
 }
 
-
 impl TryFrom<options::TableOptionsObjectStore> for TableOptionsObjectStore {
     type Error = ProtoConvError;
     fn try_from(value: options::TableOptionsObjectStore) -> Result<Self, Self::Error> {

crates/sqlexec/src/dispatch/external.rs

@@ -77,7 +77,6 @@ use sqlbuiltins::functions::FunctionRegistry;
 use super::{DispatchError, Result};
 
 /// Dispatch to external tables and databases.
-// TODO: add a `DatasourceRegistry` to the `ExternalDispatcher` to allow for dynamic datasources.
 pub struct ExternalDispatcher<'a> {
     catalog: &'a SessionCatalog,
     // TODO: Remove need for this.

crates/sqlexec/src/planner/session_planner.rs

@@ -966,16 +966,22 @@ impl<'a> SessionPlanner<'a> {
                     .into_iter()
                     .map(|coll| -> Result<Field, PlanError> {
                         // check if there is a NOT NULL constraint
-                        let is_nullable = !coll
+                        let has_not_null_constraint = coll
                             .options
                             .into_iter()
                             .any(|k| matches!(k.option, ColumnOption::NotNull));
 
-                        Ok(Field::new(
-                            coll.name.to_string(),
-                            convert_data_type(&coll.data_type)?,
-                            is_nullable,
-                        ))
+                        if has_not_null_constraint {
+                            Err(PlanError::String(
+                                "'NOT NULL' constraint is not supported".to_string(),
+                            ))
+                        } else {
+                            Ok(Field::new(
+                                coll.name.to_string(),
+                                convert_data_type(&coll.data_type)?,
+                                true,
+                            ))
+                        }
                     })
                     .collect::<Result<Vec<_>, PlanError>>()?;
                 Ok::<_, PlanError>(Schema::new(fields))