Implement UDF linking (#78)

## Summary of changes

This PR implements linking of UDFs by passing in a table of function
pointers into `compile_hir`.

This PR also adds the `Catalog` field to both the rule engine and the
optimizer as a trait object (`Arc<dyn Catalog>`). I strongly considered
making this a generic parameter, but the problem with that is then the
`C: Catalog` type parameter would need to be added to literally every
structures and function, and I that would significantly reduce
readability. The performance is essentially negligible because it is
unlikely the compiler could make significant optimizations based on the
implementation of the `Catalog`.

Ideally I would add tests that test the functionality of the catalog as
well but I don't have time (and also I don't have any reason to think
that it wouldn't work now). Once I get an interface that the optimizer
needs from the catalog from @SarveshOO7 and @yliang412, I can add those
tests easily.

Author: connortsui20

Cargo.lock

@@ -18,6 +18,7 @@ colored = "3.0.0"
 enum_dispatch = "0.3.13"
 futures = "0.3.31"
 iceberg = { version = "0.4.0", default-features = false }
+iceberg-catalog-memory = "0.4.0"
 ordered-float = "5.0.0"
 tokio = { version = "1.44.2", features = ["macros", "rt"] }
 trait-variant = "0.1.2"

optd/Cargo.toml

@@ -1,20 +1,27 @@
 use super::{Catalog as OptdCatalog, CatalogError};
 use async_trait::async_trait;
-use iceberg::{Catalog as IcebergCatalog, NamespaceIdent, TableIdent, table::Table};
-use std::collections::HashMap;
+use iceberg::{
+    Catalog as IcebergCatalog, NamespaceIdent, TableIdent, io::FileIOBuilder, table::Table,
+};
+use iceberg_catalog_memory::MemoryCatalog;
+use std::{collections::HashMap, sync::Arc};
 
 /// The default namespace for the Iceberg catalog.
 ///
 /// TODO(connor): For now, keep everything in the default namespace for simplicity.
 static DEFAULT_NAMESPACE: &str = "default";
 
-/// A wrapper around an arbitrary Iceberg catalog.
-#[derive(Debug)]
-pub struct OptdIcebergCatalog<C: IcebergCatalog>(C);
+/// A wrapper around an arbitrary Iceberg Catalog.
+#[derive(Debug, Clone)]
+pub struct OptdIcebergCatalog<C: IcebergCatalog>(Arc<C>);
 
 impl<C: IcebergCatalog> OptdIcebergCatalog<C> {
     /// Creates a new catalog.
     pub fn new(catalog: C) -> Self {
+        Self(Arc::new(catalog))
+    }
+
+    pub fn new_from_arc(catalog: Arc<C>) -> Self {
         Self(catalog)
     }
 
@@ -23,14 +30,23 @@ impl<C: IcebergCatalog> OptdIcebergCatalog<C> {
         let namespace_ident = NamespaceIdent::new(DEFAULT_NAMESPACE.to_string());
         let table_ident = TableIdent::new(namespace_ident, table_name.to_string());
 
-        // TODO(connor): FIX ERROR HANDLING.
         self.0
             .load_table(&table_ident)
             .await
             .map_err(|e| CatalogError::Unknown(e.to_string()))
     }
 }
 
+/// Creates a basic in-memory catalog. Used for testing.
+pub fn memory_catalog() -> OptdIcebergCatalog<MemoryCatalog> {
+    let file_io = FileIOBuilder::new("memory")
+        .build()
+        .expect("unable to create file");
+    let catalog = Arc::new(MemoryCatalog::new(file_io, Some("mock".to_string())));
+
+    OptdIcebergCatalog(catalog)
+}
+
 #[async_trait]
 impl<C: IcebergCatalog> OptdCatalog for OptdIcebergCatalog<C> {
     async fn get_table_properties(

optd/src/catalog/iceberg.rs

@@ -269,7 +269,7 @@ impl<M: Memoize> Optimizer<M> {
         expression_id: LogicalExpressionId,
         job_id: JobId,
     ) -> Result<(), Error> {
-        let engine = Engine::new(self.hir_context.clone());
+        let engine = Engine::new(self.hir_context.clone(), self.catalog.clone());
 
         let plan: PartialLogicalPlan = self
             .memo
@@ -311,7 +311,7 @@ impl<M: Memoize> Optimizer<M> {
         group_id: GroupId,
         job_id: JobId,
     ) -> Result<(), Error> {
-        let engine = Engine::new(self.hir_context.clone());
+        let engine = Engine::new(self.hir_context.clone(), self.catalog.clone());
 
         let plan: PartialLogicalPlan = self
             .memo
@@ -350,7 +350,7 @@ impl<M: Memoize> Optimizer<M> {
         goal_id: GoalId,
         job_id: JobId,
     ) -> Result<(), Error> {
-        let engine = Engine::new(self.hir_context.clone());
+        let engine = Engine::new(self.hir_context.clone(), self.catalog.clone());
 
         let Goal(_, physical_props) = self.memo.materialize_goal(goal_id).await?;
         let plan = self
@@ -389,7 +389,7 @@ impl<M: Memoize> Optimizer<M> {
         expression_id: PhysicalExpressionId,
         job_id: JobId,
     ) -> Result<(), Error> {
-        let engine = Engine::new(self.hir_context.clone());
+        let engine = Engine::new(self.hir_context.clone(), self.catalog.clone());
 
         let plan = self.egest_partial_plan(expression_id).await?;