[POC] route Engine implementation through a PlanExecutor

kernel/src/engine/mod.rs

@@ -39,6 +39,8 @@ pub use self::arrow_utils::{parse_json, to_json_bytes};
 
 #[cfg(feature = "default-engine-base")]
 pub mod default;
+#[cfg(feature = "default-engine-base")]
+pub mod plan;
 
 #[cfg(test)]
 pub(crate) mod sync;

kernel/src/engine/plan/engine.rs

@@ -0,0 +1,86 @@
+//! A plan-based [`Engine`] implementation.
+//!
+//! [`PlanBasedEngine`] delegates operations to a [`PlanExecutor`] when possible, otherwise
+//! falling back to using DefaultEngine implementations.
+
+use std::fmt;
+use std::sync::Arc;
+
+use super::storage::PlanBasedStorageHandler;
+use crate::engine::arrow_expression::ArrowEvaluationHandler;
+use crate::engine::default::executor::TaskExecutor;
+use crate::engine::default::json::DefaultJsonHandler;
+use crate::engine::default::parquet::DefaultParquetHandler;
+use crate::object_store::DynObjectStore;
+use crate::plan::PlanExecutor;
+use crate::{Engine, EvaluationHandler, JsonHandler, ParquetHandler, StorageHandler};
+
+/// An [`Engine`] that routes storage operations through a [`PlanExecutor`].
+///
+/// The storage handler converts [`StorageHandler`] calls into [`DeclarativePlanNode`]s and
+/// delegates them to the plan executor. JSON, Parquet, and evaluation handlers use the same
+/// default implementations as [`DefaultEngine`].
+///
+/// [`DeclarativePlanNode`]: crate::plan::DeclarativePlanNode
+/// [`DefaultEngine`]: crate::engine::default::DefaultEngine
+pub struct PlanBasedEngine<E: TaskExecutor> {
+    executor: Arc<dyn PlanExecutor>,
+    storage: Arc<PlanBasedStorageHandler>,
+    json: Arc<DefaultJsonHandler<E>>,
+    parquet: Arc<DefaultParquetHandler<E>>,
+    evaluation: Arc<ArrowEvaluationHandler>,
+}
+
+impl<E: TaskExecutor> fmt::Debug for PlanBasedEngine<E> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("PlanBasedEngine")
+            .field("storage", &self.storage)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<E: TaskExecutor> PlanBasedEngine<E> {
+    /// Create a new `PlanBasedEngine`.
+    ///
+    /// Storage operations are delegated to `plan_executor`. The JSON, Parquet, and evaluation
+    /// handlers are constructed from the given `object_store` and `task_executor`, identically
+    /// to [`DefaultEngine`](crate::engine::default::DefaultEngine).
+    pub fn new(
+        object_store: Arc<DynObjectStore>,
+        task_executor: Arc<E>,
+        plan_executor: Arc<dyn PlanExecutor>,
+    ) -> Self {
+        Self {
+            storage: Arc::new(PlanBasedStorageHandler::new(plan_executor.clone())),
+            json: Arc::new(DefaultJsonHandler::new(
+                object_store.clone(),
+                task_executor.clone(),
+            )),
+            parquet: Arc::new(DefaultParquetHandler::new(object_store, task_executor)),
+            executor: plan_executor,
+            evaluation: Arc::new(ArrowEvaluationHandler {}),
+        }
+    }
+}
+
+impl<E: TaskExecutor> Engine for PlanBasedEngine<E> {
+    fn evaluation_handler(&self) -> Arc<dyn EvaluationHandler> {
+        self.evaluation.clone()
+    }
+
+    fn storage_handler(&self) -> Arc<dyn StorageHandler> {
+        self.storage.clone()
+    }
+
+    fn json_handler(&self) -> Arc<dyn JsonHandler> {
+        self.json.clone()
+    }
+
+    fn parquet_handler(&self) -> Arc<dyn ParquetHandler> {
+        self.parquet.clone()
+    }
+
+    fn plan_executor(&self) -> Arc<dyn PlanExecutor> {
+        self.executor.clone()
+    }
+}

kernel/src/engine/plan/mod.rs

@@ -0,0 +1,20 @@
+//! Plan-based engine implementation.
+//!
+//! This module contains an implementation of the [`Engine`](crate::Engine) trait that is
+//! backed by a [`PlanExecutor`](crate::plan::PlanExecutor). The engine delegates handler
+//! operations (storage, JSON, parquet, evaluation) to declarative plan execution, rather than
+//! implementing each handler independently.
+//!
+//! [`PlanBasedEngine`] routes storage operations through a
+//! [`PlanExecutor`](crate::plan::PlanExecutor) via [`PlanBasedStorageHandler`], while using the
+//! default Arrow-based handlers for JSON, Parquet, and expression evaluation. [`NaivePlanExecutor`]
+//! provides a concrete executor
+//! backed by [`ObjectStoreStorageHandler`](crate::engine::default::filesystem::ObjectStoreStorageHandler).
+
+pub mod engine;
+pub mod naive;
+pub mod storage;
+
+pub use engine::PlanBasedEngine;
+pub use naive::NaivePlanExecutor;
+pub use storage::PlanBasedStorageHandler;

kernel/src/engine/plan/naive.rs

@@ -0,0 +1,118 @@
+//! A naive [`PlanExecutor`] implementation backed by [`ObjectStoreStorageHandler`].
+//!
+//! [`NaivePlanExecutor`] interprets each [`DeclarativePlanNode`] variant by delegating to the
+//! default engine's existing handler implementations and converting results into columnar
+//! [`EngineData`] batches.
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+
+use crate::arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+use crate::arrow::datatypes::{
+    DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema,
+};
+use crate::engine::arrow_data::ArrowEngineData;
+use crate::engine::default::executor::TaskExecutor;
+use crate::engine::default::filesystem::ObjectStoreStorageHandler;
+use crate::object_store::DynObjectStore;
+use crate::plan::{DeclarativePlanNode, PlanExecutor, PlanResult, FILE_META_SCHEMA};
+use crate::{DeltaResult, EngineData, Error, FileMeta, FileSlice, StorageHandler as _};
+
+/// A naive [`PlanExecutor`] that delegates to [`ObjectStoreStorageHandler`] and converts
+/// results into [`ArrowEngineData`] batches.
+#[derive(Debug)]
+pub struct NaivePlanExecutor<E: TaskExecutor> {
+    storage: ObjectStoreStorageHandler<E>,
+}
+
+impl<E: TaskExecutor> NaivePlanExecutor<E> {
+    /// Create a new `NaivePlanExecutor` backed by the given object store and task executor.
+    pub fn new(store: Arc<DynObjectStore>, task_executor: Arc<E>) -> Self {
+        Self {
+            storage: ObjectStoreStorageHandler::new(store, task_executor),
+        }
+    }
+}
+
+impl<E: TaskExecutor> PlanExecutor for NaivePlanExecutor<E> {
+    fn execute_plan(&self, plan: DeclarativePlanNode) -> DeltaResult<PlanResult> {
+        match plan {
+            DeclarativePlanNode::FileListing { url } => self.execute_file_listing(&url),
+            DeclarativePlanNode::ReadBytes { files } => self.execute_read_bytes(files),
+            DeclarativePlanNode::WriteBytes {
+                url,
+                data,
+                overwrite,
+            } => self.execute_write_bytes(&url, data, overwrite),
+            DeclarativePlanNode::HeadFile { url } => self.execute_head_file(&url),
+        }
+    }
+}
+
+impl<E: TaskExecutor> NaivePlanExecutor<E> {
+    fn execute_file_listing(&self, url: &url::Url) -> DeltaResult<PlanResult> {
+        let file_metas: Vec<FileMeta> = self
+            .storage
+            .list_from(url)?
+            .collect::<DeltaResult<Vec<_>>>()?;
+        let batch = file_metas_to_engine_data(&file_metas)?;
+        let iter: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>>> =
+            Box::new(std::iter::once(Ok(batch)));
+        Ok(PlanResult::Data(iter))
+    }
+
+    fn execute_read_bytes(&self, files: Vec<FileSlice>) -> DeltaResult<PlanResult> {
+        let iter = self.storage.read_files(files)?;
+        Ok(PlanResult::ByteStream(iter))
+    }
+
+    fn execute_write_bytes(
+        &self,
+        url: &url::Url,
+        data: Bytes,
+        overwrite: bool,
+    ) -> DeltaResult<PlanResult> {
+        self.storage.put(url, data, overwrite)?;
+        Ok(PlanResult::Unit)
+    }
+
+    fn execute_head_file(&self, url: &url::Url) -> DeltaResult<PlanResult> {
+        let meta = self.storage.head(url)?;
+        let batch = file_metas_to_engine_data(std::slice::from_ref(&meta))?;
+        let iter: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>>> =
+            Box::new(std::iter::once(Ok(batch)));
+        Ok(PlanResult::Data(iter))
+    }
+}
+
+/// Convert a slice of [`FileMeta`] into a single [`ArrowEngineData`] batch matching
+/// [`FILE_META_SCHEMA`].
+fn file_metas_to_engine_data(metas: &[FileMeta]) -> DeltaResult<Box<dyn EngineData>> {
+    let paths: StringArray = metas.iter().map(|m| Some(m.location.as_str())).collect();
+    let last_modified: Int64Array = metas.iter().map(|m| Some(m.last_modified)).collect();
+    let sizes: Int64Array = metas
+        .iter()
+        .map(|m| {
+            i64::try_from(m.size)
+                .ok()
+                .map(Some)
+                .unwrap_or(Some(i64::MAX))
+        })
+        .collect();
+
+    let _ = &*FILE_META_SCHEMA;
+
+    let arrow_schema = Arc::new(ArrowSchema::new(vec![
+        ArrowField::new("path", ArrowDataType::Utf8, false),
+        ArrowField::new("last_modified", ArrowDataType::Int64, false),
+        ArrowField::new("size", ArrowDataType::Int64, false),
+    ]));
+
+    let columns: Vec<ArrayRef> = vec![Arc::new(paths), Arc::new(last_modified), Arc::new(sizes)];
+
+    let batch = RecordBatch::try_new(arrow_schema, columns)
+        .map_err(|e| Error::generic(format!("Failed to create RecordBatch: {e}")))?;
+
+    Ok(Box::new(ArrowEngineData::new(batch)))
+}