Slim data-lake: replace TableManager/FileManager/DataLake with native SessionContext

beacon-core/src/query/compiler.rs

@@ -4,7 +4,6 @@
 //! straight to DataFusion's SQL parser in `Runtime::plan_client_query`; only the
 //! JSON form is "compiled".
 
-use beacon_data_lake::{FileManager, TableManager};
 use datafusion::{logical_expr::LogicalPlan, prelude::SessionContext};
 
 use crate::query::QueryBody;
@@ -13,8 +12,6 @@ use crate::query::QueryBody;
 pub async fn compile_json_query(
     query_body: QueryBody,
     session: &SessionContext,
-    table_manager: &TableManager,
-    file_manager: &FileManager,
 ) -> anyhow::Result<LogicalPlan> {
     // The runtime config is published as a SessionConfig extension; fall back to
     // defaults if absent (e.g. a bare session in a unit test).
@@ -39,10 +36,10 @@ pub async fn compile_json_query(
             all_columns.extend(select_cols);
         }
 
-        from.init_builder(session, table_manager, file_manager, Some(&all_columns))
+        from.init_builder(session, Some(&all_columns))
             .await?
     } else {
-        from.init_builder(session, table_manager, file_manager, None)
+        from.init_builder(session, None)
             .await?
     };
 

beacon-core/src/query/from.rs

@@ -4,7 +4,6 @@
 
 use std::sync::Arc;
 
-use beacon_data_lake::{FileManager, TableManager};
 use beacon_datafusion_ext::file_collection::FileCollection;
 use beacon_arrow_odv::datafusion::OdvFormat;
 use beacon_arrow_csv::datafusion::CsvFormat;
@@ -42,22 +41,18 @@ impl From {
     ///
     /// # Arguments
     /// * `session_context` - The DataFusion session context.
-    /// * `table_manager` - Table manager used to resolve table providers.
-    /// * `file_manager` - File manager used to resolve listing paths.
     ///
     /// # Returns
     /// * `LogicalPlanBuilder` for the specified source.
     pub async fn init_builder(
         &self,
         session_context: &SessionContext,
-        table_manager: &TableManager,
-        file_manager: &FileManager,
         projection: Option<&Vec<String>>,
     ) -> datafusion::error::Result<LogicalPlanBuilder> {
         match self {
             From::Table(name) => {
-                // Use a registered table.
-                if let Some(mut table) = table_manager.table_provider(name) {
+                // Use a registered table from the catalog.
+                if let Ok(mut table) = session_context.table_provider(name.as_str()).await {
                     if let (Some(projection), Some(file_collection)) =
                         (projection, table.as_any().downcast_ref::<FileCollection>())
                     {
@@ -75,9 +70,7 @@ impl From {
             }
             From::Format { format } => {
                 // Use a file format as a table source.
-                let table_source = format
-                    .as_table_source(session_context, file_manager)
-                    .await?;
+                let table_source = format.as_table_source(session_context).await?;
                 Ok(LogicalPlanBuilder::scan("tmp", table_source, None)?)
             }
         }
@@ -121,17 +114,15 @@ impl FromFormat {
     ///
     /// # Arguments
     /// * `session_context` - The DataFusion session context.
-    /// * `file_manager` - File manager used for file path resolution.
     ///
     /// # Returns
     /// * `Arc<dyn TableSource>` for the specified format.
     pub async fn as_table_source(
         &self,
         session_context: &SessionContext,
-        file_manager: &FileManager,
     ) -> datafusion::error::Result<Arc<dyn TableSource>> {
         let file_format = self.file_format(session_context).await?;
-        let urls = self.listing_table_urls(file_manager)?;
+        let urls = self.listing_table_urls()?;
 
         // Create a FileCollection as the table provider.
         let table =
@@ -167,11 +158,8 @@ impl FromFormat {
         }
     }
 
-    /// Resolves file paths to [`ListingTableUrl`]s using the data lake.
-    fn listing_table_urls(
-        &self,
-        file_manager: &FileManager,
-    ) -> datafusion::error::Result<Vec<ListingTableUrl>> {
+    /// Resolves file paths to [`ListingTableUrl`]s under the datasets store.
+    fn listing_table_urls(&self) -> datafusion::error::Result<Vec<ListingTableUrl>> {
         let paths = match self {
             FromFormat::Csv { paths, .. }
             | FromFormat::Parquet { paths }
@@ -185,7 +173,7 @@ impl FromFormat {
 
         let mut listing_table_urls = Vec::with_capacity(paths.len());
         for path in paths {
-            let url = file_manager.try_create_listing_url(path.to_string())?;
+            let url = beacon_data_lake::create_listing_url(path.to_string())?;
             listing_table_urls.push(url);
         }
         Ok(listing_table_urls)

beacon-core/src/query/output.rs

@@ -8,7 +8,6 @@ use std::sync::Arc;
 use beacon_arrow_netcdf::datafusion::{options::NetcdfOptions, NetCDFFormatFactory, NetcdfConfig};
 use beacon_arrow_odv::datafusion::OdvFileFormatFactory;
 use beacon_arrow_odv::writer::OdvOptions;
-use beacon_data_lake::FileManager;
 use beacon_arrow_csv::datafusion::CsvFormatFactory;
 use beacon_arrow_geoparquet::datafusion::{GeoParquetFormatFactory, GeoParquetOptions};
 use beacon_arrow_ipc::datafusion::ArrowFormatFactory;
@@ -33,16 +32,16 @@ impl Output {
     /// Parses the logical plan and prepares an output file in the specified format.
     ///
     /// # Arguments
-    /// * `_session_context` - DataFusion session context (unused).
-    /// * `file_manager` - FileManager instance for temporary file creation.
+    /// * `session_context` - DataFusion session context (provides the datasets store).
+    /// * `tmp_dir` - Directory the temporary output file is created in (the tmp store root).
     /// * `input_plan` - The logical plan to export.
     ///
     /// # Returns
     /// Tuple of the new logical plan and the output file wrapper.
     pub async fn parse(
         &self,
         session_context: &SessionContext,
-        file_manager: &FileManager,
+        tmp_dir: &std::path::Path,
         input_plan: LogicalPlan,
     ) -> datafusion::error::Result<(LogicalPlan, QueryOutputFile)> {
         let datasets_store = session_context
@@ -55,7 +54,7 @@ impl Output {
                 )
             })?;
         let file_type = self.format.file_type(datasets_store).await;
-        let temp_output = file_manager.try_create_temp_output_file(".tmp");
+        let temp_output = beacon_data_lake::create_temp_output_file(tmp_dir, ".tmp");
         let plan = LogicalPlanBuilder::copy_to(
             input_plan,
             temp_output.output_url(),