feat(session): Replaces direct usage of DaftCatalog with Session (#3794)

daft/catalog/__init__.py

@@ -164,8 +164,8 @@ class Identifier(Sequence):
     """A reference (path) to a catalog object.
 
     Example:
-    >>> id1 = Identifier("a", "b")
-    >>> id2 = Identifier.parse("a.b")
+    >>> id = Identifier("a", "b")
+    >>> assert len(id) == 2
     """
 
     _identifier: native_catalog.PyIdentifier
@@ -174,8 +174,8 @@ def __init__(self, *parts: str):
         """Creates an Identifier from its parts.
 
         Example:
-        >>> id = Identifier("schema", "table")
-        >>> id  # Identifier('schema.table')
+        >>> Identifier("schema", "table")
+        >>> #
 
         Returns:
             Identifier: A new identifier.
@@ -185,18 +185,19 @@ def __init__(self, *parts: str):
         self._identifier = native_catalog.PyIdentifier(parts[:-1], parts[-1])
 
     @staticmethod
-    def parse(input: str) -> Identifier:
-        """Parses an Identifier from an SQL string.
+    def from_sql(input: str, normalize: bool = False) -> Identifier:
+        """Parses an Identifier from an SQL string, normalizing to lowercase if specified.
 
         Example:
-        >>> id = Identifier.parse("schema.table")
-        >>> assert len(id) == 2
+        >>> Identifier.from_sql("schema.table") == Identifier("schema", "table")
+        >>> Identifier.from_sql('"a.b"') == Identifier('"a.b."')
+        >>> Identifier.from_sql('ABC."xYz"', normalize=True) == Identifier("abc", "xYz")
 
         Returns:
             Identifier: A new identifier.
         """
         i = Identifier.__new__(Identifier)
-        i._identifier = native_catalog.PyIdentifier.parse(input)
+        i._identifier = native_catalog.PyIdentifier.from_sql(input, normalize)
         return i
 
     def __eq__(self, other: object) -> bool:

daft/daft/catalog.pyi

@@ -8,7 +8,7 @@ if TYPE_CHECKING:
 class PyIdentifier:
     def __init__(self, namespace: tuple[str, ...], name: str): ...
     @staticmethod
-    def parse(input: str): ...
+    def from_sql(input: str, normalize: bool): ...
     def eq(self, other: PyIdentifier) -> bool: ...
     def getitem(self, index: int) -> str: ...
     def __len__(self) -> int: ...

src/daft-catalog/src/catalog.rs

@@ -0,0 +1,43 @@
+use std::{collections::HashMap, sync::Arc};
+
+use crate::DataCatalog;
+
+/// Catalogs is a collection of referenceable catalogs (glorified map).
+///
+/// Notes:
+///  - This is intentionally lightweight and everything is exact-case.
+///  - All APIs are non-fallible because callers determine what is an error.
+///  - It does not necessarily have map semantics.
+///  - Callers are responsible for case-normalization hence String, &str.
+///  - Intentionally using String and &str rather than Into and AsRef.
+#[derive(Debug)]
+pub struct Catalogs(HashMap<String, Arc<dyn DataCatalog>>);
+
+impl Catalogs {
+    /// Creates an empty catalogs collection
+    pub fn empty() -> Catalogs {
+        Self(HashMap::new())
+    }
+
+    /// Attaches a catalog to this catalog collection.
+    pub fn attach(&mut self, name: String, catalog: Arc<dyn DataCatalog>) {
+        self.0.insert(name, catalog);
+    }
+
+    /// Detaches a catalog from this catalog collection.
+    pub fn detach(&mut self, name: &str) {
+        self.0.remove(name);
+    }
+
+    /// Get the catalog by name.
+    pub fn get(&self, name: &str) -> Option<Arc<dyn DataCatalog>> {
+        self.0.get(name).map(Arc::clone)
+    }
+
+    /// Returns true iff a catalog with the given name exists (exact-case).
+    pub fn exists(&self, name: &str) -> bool {
+        self.0.contains_key(name)
+    }
+}
+
+// TODO Catalog trait for implementations.

src/daft-catalog/src/error.rs

@@ -1,9 +1,14 @@
 use snafu::Snafu;
 
+use crate::Identifier;
+
+/// Catalog Result
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
+/// Catalog Error
 #[derive(Debug, Snafu)]
 pub enum Error {
+    // TODO remove me
     #[snafu(display(
         "Failed to find specified table identifier {} in the requested catalog {}",
         catalog_name,
@@ -14,10 +19,13 @@ pub enum Error {
         table_id: String,
     },
 
-    #[snafu(display("Catalog not found: {}", name))]
-    CatalogNotFound { name: String },
+    #[snafu(display("{type_} with name {ident} already exists!"))]
+    ObjectAlreadyExists { type_: String, ident: String },
+
+    #[snafu(display("{type_} with name {ident} not found!"))]
+    ObjectNotFound { type_: String, ident: String },
 
-    #[snafu(display("Invalid identifier `{input}`."))]
+    #[snafu(display("Invalid identifier {input}!"))]
     InvalidIdentifier { input: String },
 
     #[snafu(display("{message}"))]
@@ -33,21 +41,33 @@ pub enum Error {
 
 impl Error {
     #[inline]
-    pub fn unsupported(message: String) -> Error {
-        Error::Unsupported { message }
+    pub fn unsupported<S: Into<String>>(message: S) -> Error {
+        Error::Unsupported {
+            message: message.into(),
+        }
+    }
+
+    #[inline]
+    pub fn obj_already_exists<S: Into<String>>(type_: S, ident: &Identifier) -> Error {
+        Error::ObjectAlreadyExists {
+            type_: type_.into(),
+            ident: ident.to_string(),
+        }
+    }
+
+    // Consider typed arguments vs strings for consistent formatting.
+    #[inline]
+    pub fn obj_not_found<S: Into<String>>(typ_: S, ident: &Identifier) -> Error {
+        Error::ObjectNotFound {
+            type_: typ_.into(),
+            ident: ident.to_string(),
+        }
     }
 }
 
 impl From<Error> for common_error::DaftError {
     fn from(err: Error) -> Self {
-        match &err {
-            Error::TableNotFound { .. }
-            | Error::CatalogNotFound { .. }
-            | Error::InvalidIdentifier { .. }
-            | Error::Unsupported { .. } => common_error::DaftError::CatalogError(err.to_string()),
-            #[cfg(feature = "python")]
-            Error::PythonError { .. } => common_error::DaftError::CatalogError(err.to_string()),
-        }
+        common_error::DaftError::CatalogError(err.to_string())
     }
 }
 

src/daft-catalog/src/identifier.rs

@@ -1,11 +1,11 @@
-use std::{fmt::Display, iter::once, vec::IntoIter};
+use std::fmt::Display;
 
 use crate::error::{Error, Result};
 
-/// An object's namespace (location).
+/// A reference qualifier.
 pub type Namespace = Vec<String>;
 
-/// A reference (path) to some catalog object.
+/// A reference to a catalog object.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Identifier {
     pub namespace: Namespace,
@@ -18,13 +18,21 @@ impl Identifier {
         Self { namespace, name }
     }
 
-    /// Returns true if this is a qualified identifier e.g. has a namespace.
+    /// Returns a new simple identifier (no namespace)
+    pub fn simple<T: Into<String>>(name: T) -> Self {
+        Self {
+            namespace: vec![],
+            name: name.into(),
+        }
+    }
+
+    /// Returns true if this is a qualified identifier.
     pub fn has_namespace(&self) -> bool {
         !self.namespace.is_empty()
     }
 
     /// Parses an identifier using sqlparser to validate the input.
-    pub fn parse(input: &str) -> Result<Identifier> {
+    pub fn from_sql(input: &str, normalize: bool) -> Result<Identifier> {
         // TODO daft should define its own identifier domain.
         use sqlparser::{dialect::PostgreSqlDialect, parser::Parser};
         let err = Error::InvalidIdentifier {
@@ -33,58 +41,59 @@ impl Identifier {
         let Ok(mut parser) = Parser::new(&PostgreSqlDialect {}).try_with_sql(input) else {
             return Err(err);
         };
-        let Ok(parts) = parser.parse_multipart_identifier() else {
+        let Ok(idents) = parser.parse_multipart_identifier() else {
             return Err(err);
         };
-        if parts.is_empty() {
+        if idents.is_empty() {
             return Err(err);
         }
-        // TODO Identifier normalization is omitted until further discussion.
-        let mut parts = parts
-            .iter()
-            .map(|part| part.value.to_string())
-            .collect::<Vec<String>>();
-        let name = parts.pop().unwrap();
-        let namespace = parts;
+        // Convert sqlparser identifiers applying normalization if given.
+        let mut names: Vec<String> = vec![];
+        for ident in idents {
+            if normalize && ident.quote_style.is_none() {
+                names.push(ident.value.to_lowercase());
+            } else {
+                names.push(ident.value);
+            }
+        }
+        let name = names.pop().unwrap();
+        let namespace = names;
         Ok(Identifier::new(namespace, name))
     }
 }
 
 impl Display for Identifier {
-    /// Joins the identifier to a dot-delimited path.
+    /// Returns the identifier as a dot-delimited string with double-quoted parts.
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         if self.namespace.is_empty() {
-            f.write_str(&self.name)
+            write!(f, "{}", escape_double_quotes(&self.name))
         } else {
-            let prefix = self.namespace.join(".");
-            let string = format!("{}.{}", prefix, self.name);
-            f.write_str(&string)
+            let namespace = self
+                .namespace
+                .iter()
+                .map(|n| escape_double_quotes(n))
+                .collect::<Vec<String>>()
+                .join(".");
+            let name = escape_double_quotes(&self.name);
+            write!(f, "{}.{}", namespace, name)
         }
     }
 }
 
-impl IntoIterator for Identifier {
-    type Item = String;
-    type IntoIter = IntoIter<Self::Item>;
-
-    fn into_iter(self) -> Self::IntoIter {
-        self.namespace
-            .into_iter()
-            .chain(once(self.name))
-            .collect::<Vec<String>>()
-            .into_iter()
+impl From<String> for Identifier {
+    /// Returns an unqualified delimited identifier.
+    fn from(name: String) -> Self {
+        Self::simple(name)
     }
 }
 
-impl<'a> IntoIterator for &'a Identifier {
-    type Item = &'a String;
-    type IntoIter = IntoIter<Self::Item>;
-
-    fn into_iter(self) -> Self::IntoIter {
-        self.namespace
-            .iter()
-            .chain(once(&self.name))
-            .collect::<Vec<&String>>()
-            .into_iter()
+impl From<&str> for Identifier {
+    /// Returns an unqualified delmited identifier.
+    fn from(name: &str) -> Self {
+        Self::simple(name.to_string())
     }
 }
+
+fn escape_double_quotes(text: &str) -> String {
+    text.replace('"', "\"\"")
+}

src/daft-catalog/src/lib.rs

@@ -1,8 +1,8 @@
+pub mod catalog;
 mod data_catalog;
 mod data_catalog_table;
 pub mod error;
-pub mod identifier;
-
+mod identifier;
 // Export public-facing traits
 use std::{collections::HashMap, default, sync::Arc};
 
@@ -14,6 +14,7 @@ pub use data_catalog_table::DataCatalogTable;
 pub mod python;
 
 use error::{Error, Result};
+pub use identifier::*;
 
 pub mod global_catalog {
     use std::sync::{Arc, RwLock};
@@ -43,7 +44,6 @@ pub mod global_catalog {
             .unregister_catalog(name)
     }
 }
-use identifier::Identifier;
 
 /// Name of the default catalog
 static DEFAULT_CATALOG_NAME: &str = "default";
@@ -103,13 +103,8 @@ impl DaftCatalog {
         name: &str,
         view: impl Into<LogicalPlanBuilder>,
     ) -> Result<()> {
-        let identifier = Identifier::parse(name)?;
-        if identifier.has_namespace() {
-            return Err(Error::Unsupported {
-                message: format!("Qualified identifiers are not yet supported. Instead use a single identifier, or wrap your table name in quotes such as `\"{}\"`", name),
-            });
-        }
-        self.named_tables.insert(identifier.name, view.into());
+        // TODO this API is being removed, for now preserve the exact name as if it were delimited.
+        self.named_tables.insert(name.into(), view.into());
         Ok(())
     }
 
@@ -159,6 +154,7 @@ impl DaftCatalog {
             table_id: searched_table_name.to_string(),
         })
     }
+
     /// Copy from another catalog, using tables from other in case of conflict
     pub fn copy_from(&mut self, other: &Self) {
         for (name, plan) in &other.named_tables {
@@ -168,6 +164,13 @@ impl DaftCatalog {
             self.data_catalogs.insert(name.clone(), catalog.clone());
         }
     }
+
+    /// TODO remove py register and read methods are moved to session
+    /// I cannot remove DaftMetaCatalog until I invert the dependency
+    /// so that the current register_ methods use the session rather than the catalog.
+    pub fn into_catalog_map(self) -> HashMap<String, Arc<dyn DataCatalog>> {
+        self.data_catalogs
+    }
 }
 
 #[cfg(test)]
@@ -208,11 +211,6 @@ mod tests {
 
         // Register a table
         assert!(catalog.register_table("test_table", plan.clone()).is_ok());
-
-        // Try to register a table with invalid name
-        assert!(catalog
-            .register_table("invalid name", plan.clone())
-            .is_err());
     }
 
     #[test]