[POC] Expose PlanExecutor through FFI

ffi/Cargo.toml

@@ -22,6 +22,9 @@ tracing = "0.1"
 tracing-core = { version = "0.1", optional = true }
 tracing-subscriber = { version = "0.3", optional = true, features = [ "json" ] }
 url = "2"
+bytes = { version = "1.10", optional = true }
+prost = { version = "0.13", optional = true }
+serde_json = { version = "1.0", optional = true }
 delta_kernel = { path = "../kernel", default-features = false, features = [
   "internal-api",
 ] }
@@ -32,6 +35,7 @@ unity-catalog-delta-client-api = { path = "../unity-catalog-delta-client-api", o
 [build-dependencies]
 cbindgen = "0.29.2"
 libc = "0.2.175"
+prost-build = "0.13"
 
 [dev-dependencies]
 paste = "1.0"
@@ -57,7 +61,7 @@ arrow-57 = ["delta_kernel/arrow-57"]
 # This is an 'internal' feature flag which has all the shared bits from default-engine-native-tls and
 # default-engine-rustls. There is a check in kernel/lib.rs to ensure you have enabled one of
 # default-engine-native-tls or default-engine-rustls, so default-engine-base will not work by itself
-default-engine-base = ["delta_kernel/default-engine-base"]
+default-engine-base = ["delta_kernel/default-engine-base", "dep:bytes", "dep:prost", "dep:serde_json"]
 
 tracing = [ "tracing-core", "tracing-subscriber" ]
 internal-api = []

ffi/build.rs

@@ -17,9 +17,26 @@ fn get_target_dir(manifest_dir: &str) -> PathBuf {
     }
 }
 
+fn compile_protos(crate_dir: &str) {
+    // The proto module is only consumed under the `default-engine-base` feature, but we always
+    // generate the Rust bindings so that turning on/off features doesn't trigger extra rebuilds
+    // of the generated file.
+    let proto_path = Path::new(crate_dir)
+        .join("proto")
+        .join("declarative_plan_node.proto");
+    let proto_dir = Path::new(crate_dir).join("proto");
+    println!("cargo:rerun-if-changed={}", proto_path.display());
+    prost_build::Config::new()
+        .compile_protos(&[proto_path.as_path()], &[proto_dir.as_path()])
+        .expect("failed to compile declarative_plan_node.proto");
+}
+
 fn main() {
     let crate_dir = env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR should be set");
     let package_name = env::var("CARGO_PKG_NAME").expect("CARGO_PKG_NAME should be set");
+
+    compile_protos(&crate_dir);
+
     let target_dir = get_target_dir(crate_dir.as_str());
     let cbindgen_toml = Path::new(&crate_dir).join("cbindgen.toml");
     let mut config = Config::from_file(&cbindgen_toml)

ffi/proto/declarative_plan_node.proto

@@ -0,0 +1,79 @@
+syntax = "proto3";
+
+package delta_kernel_ffi.plan;
+
+// Wire format for delta_kernel::plan::DeclarativePlanNode.
+//
+// Mirrors the Rust enum variants. Used to ferry plan nodes from Rust to a
+// non-Rust executor (e.g. Java) over FFI as serialized bytes.
+
+// File format for a Scan operation. Mirrors delta_kernel::plan::ScanFileFormat.
+enum ScanFileFormat {
+  SCAN_FILE_FORMAT_UNSPECIFIED = 0;
+  SCAN_FILE_FORMAT_PARQUET = 1;
+  SCAN_FILE_FORMAT_JSON = 2;
+}
+
+// File metadata; mirrors delta_kernel::FileMeta. Path is a fully qualified URL.
+message FileMeta {
+  string location = 1;
+  int64 last_modified = 2;
+  uint64 size = 3;
+}
+
+// Optional half-open byte range [start, end). Mirrors `std::ops::Range<u64>`.
+message ByteRange {
+  uint64 start = 1;
+  uint64 end = 2;
+}
+
+// File slice; mirrors `delta_kernel::FileSlice = (Url, Option<Range<u64>>)`.
+message FileSlice {
+  string url = 1;
+  ByteRange range = 2; // unset = read whole file
+}
+
+// List files at the given URL.
+message FileListing {
+  string url = 1;
+}
+
+// Read raw bytes from one or more file slices.
+message ReadBytes {
+  repeated FileSlice files = 1;
+}
+
+// Write raw bytes to a file at the given URL.
+message WriteBytes {
+  string url = 1;
+  bytes data = 2;
+  bool overwrite = 3;
+}
+
+// HEAD request for a single file.
+message HeadFile {
+  string url = 1;
+}
+
+// Read and parse structured data files (Parquet or JSON).
+//
+// physical_schema_json is a JSON-serialized delta_kernel::schema::StructType (the standard
+// kernel schema serialization). predicate is intentionally opaque for now and may be empty;
+// a future revision will define a structured predicate message.
+message Scan {
+  ScanFileFormat format = 1;
+  repeated FileMeta files = 2;
+  string physical_schema_json = 3;
+  bytes predicate = 4; // TODO: structured predicate; opaque for now
+}
+
+// Top-level declarative plan node. Exactly one of the `op` fields is set.
+message DeclarativePlanNode {
+  oneof op {
+    FileListing file_listing = 1;
+    ReadBytes read_bytes = 2;
+    WriteBytes write_bytes = 3;
+    HeadFile head_file = 4;
+    Scan scan = 5;
+  }
+}