Merge branch 'main' into manndp/write_domain_metadata

Author: MannDP

ffi/src/expressions/engine_visitor.rs

@@ -10,14 +10,16 @@ use delta_kernel::expressions::{
     ColumnName, Expression, ExpressionRef, JunctionPredicate, JunctionPredicateOp, MapData,
     OpaqueExpression, OpaqueExpressionOpRef, OpaquePredicate, OpaquePredicateOpRef, Predicate,
     Scalar, StructData, Transform, UnaryExpression, UnaryExpressionOp, UnaryPredicate,
-    UnaryPredicateOp,
+    UnaryPredicateOp, VariadicExpression, VariadicExpressionOp,
 };
 
 use std::ffi::c_void;
 
 type VisitLiteralFn<T> = extern "C" fn(data: *mut c_void, sibling_list_id: usize, value: T);
 type VisitUnaryFn = extern "C" fn(data: *mut c_void, sibling_list_id: usize, child_list_id: usize);
 type VisitBinaryFn = extern "C" fn(data: *mut c_void, sibling_list_id: usize, child_list_id: usize);
+type VisitVariadicFn =
+    extern "C" fn(data: *mut c_void, sibling_list_id: usize, child_list_id: usize);
 type VisitJunctionFn =
     extern "C" fn(data: *mut c_void, sibling_list_id: usize, child_list_id: usize);
 
@@ -163,6 +165,9 @@ pub struct EngineExpressionVisitor {
     /// Visits the `Divide` binary operator belonging to the list identified by `sibling_list_id`.
     /// The operands will be in a _two_ item list identified by `child_list_id`
     pub visit_divide: VisitBinaryFn,
+    /// Visits the `Coalesce` variadic operator belonging to the list identified by `sibling_list_id`.
+    /// The operands will be in a list identified by `child_list_id`
+    pub visit_coalesce: VisitVariadicFn,
     /// Visits the `column` belonging to the list identified by `sibling_list_id`.
     pub visit_column:
         extern "C" fn(data: *mut c_void, sibling_list_id: usize, name: KernelStringSlice),
@@ -610,6 +615,16 @@ fn visit_expression_impl(
             };
             visit_fn(visitor.data, sibling_list_id, child_list_id);
         }
+        Expression::Variadic(VariadicExpression { op, exprs }) => {
+            let child_list_id = call!(visitor, make_field_list, exprs.len());
+            for expr in exprs {
+                visit_expression_impl(visitor, expr, child_list_id);
+            }
+            let visit_fn = match op {
+                VariadicExpressionOp::Coalesce => visitor.visit_coalesce,
+            };
+            visit_fn(visitor.data, sibling_list_id, child_list_id);
+        }
         Expression::Opaque(OpaqueExpression { op, exprs }) => {
             visit_expression_opaque(visitor, op, exprs, sibling_list_id)
         }

ffi/src/scan.rs

@@ -6,7 +6,7 @@ use std::sync::{Arc, Mutex};
 
 use delta_kernel::scan::state::DvInfo;
 use delta_kernel::scan::{Scan, ScanMetadata};
-use delta_kernel::snapshot::Snapshot;
+use delta_kernel::snapshot::SnapshotRef;
 use delta_kernel::{DeltaResult, Error, Expression, ExpressionRef};
 use delta_kernel_ffi_macros::handle_descriptor;
 use tracing::debug;
@@ -103,7 +103,7 @@ pub unsafe extern "C" fn scan(
 }
 
 fn scan_impl(
-    snapshot: Arc<Snapshot>,
+    snapshot: SnapshotRef,
     predicate: Option<&mut EnginePredicate>,
 ) -> DeltaResult<Handle<SharedScan>> {
     let mut scan_builder = snapshot.scan_builder();

kernel/examples/common/src/lib.rs

@@ -8,7 +8,7 @@ use delta_kernel::{
     engine::default::{executor::tokio::TokioBackgroundExecutor, DefaultEngine},
     scan::Scan,
     schema::Schema,
-    DeltaResult, Snapshot,
+    DeltaResult, SnapshotRef,
 };
 
 use url::Url;
@@ -62,7 +62,7 @@ pub fn get_engine(
 
 /// Construct a scan at the latest snapshot. This is over the specified table and using the passed
 /// engine. Parameters of the scan are controlled by the specified `ScanArgs`
-pub fn get_scan(snapshot: Arc<Snapshot>, args: &ScanArgs) -> DeltaResult<Option<Scan>> {
+pub fn get_scan(snapshot: SnapshotRef, args: &ScanArgs) -> DeltaResult<Option<Scan>> {
     if args.schema_only {
         println!("{:#?}", snapshot.schema());
         return Ok(None);

kernel/examples/write-table/src/main.rs

@@ -19,7 +19,7 @@ use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
 use delta_kernel::engine::default::DefaultEngine;
 use delta_kernel::schema::{DataType, SchemaRef, StructField, StructType};
 use delta_kernel::transaction::CommitResult;
-use delta_kernel::{DeltaResult, Engine, Error, Snapshot};
+use delta_kernel::{DeltaResult, Engine, Error, Snapshot, SnapshotRef};
 
 /// An example program that writes to a Delta table and creates it if necessary.
 #[derive(Parser)]
@@ -123,7 +123,7 @@ async fn create_or_get_base_snapshot(
     url: &Url,
     engine: &dyn Engine,
     schema_str: &str,
-) -> DeltaResult<Arc<Snapshot>> {
+) -> DeltaResult<SnapshotRef> {
     // Check if table already exists
     match Snapshot::builder_for(url.clone()).build(engine) {
         Ok(snapshot) => {

kernel/src/checkpoint/mod.rs

@@ -35,6 +35,7 @@
 //! # use delta_kernel::checkpoint::CheckpointWriter;
 //! # use delta_kernel::Engine;
 //! # use delta_kernel::Snapshot;
+//! # use delta_kernel::SnapshotRef;
 //! # use delta_kernel::DeltaResult;
 //! # use delta_kernel::Error;
 //! # use delta_kernel::FileMeta;
@@ -79,6 +80,7 @@
 //!
 //! [`CheckpointMetadata`]: crate::actions::CheckpointMetadata
 //! [`LastCheckpointHint`]: crate::last_checkpoint_hint::LastCheckpointHint
+//! [`Snapshot::checkpoint`]: crate::Snapshot::checkpoint
 // Future extensions:
 // - TODO(#837): Multi-file V2 checkpoints are not supported yet. The API is designed to be extensible for future
 //   multi-file support, but the current implementation only supports single-file checkpoints.
@@ -98,7 +100,7 @@ use crate::last_checkpoint_hint::LastCheckpointHint;
 use crate::log_replay::LogReplayProcessor;
 use crate::path::ParsedLogPath;
 use crate::schema::{DataType, SchemaRef, StructField, StructType, ToSchema as _};
-use crate::snapshot::Snapshot;
+use crate::snapshot::SnapshotRef;
 use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Engine, EngineData, Error, EvaluationHandlerExtension, FileMeta};
 
@@ -194,7 +196,7 @@ impl Iterator for CheckpointDataIterator {
 /// See the [module-level documentation](self) for the complete checkpoint workflow
 pub struct CheckpointWriter {
     /// Reference to the snapshot (i.e. version) of the table being checkpointed
-    pub(crate) snapshot: Arc<Snapshot>,
+    pub(crate) snapshot: SnapshotRef,
 
     /// The version of the snapshot being checkpointed.
     /// Note: Although the version is stored as a u64 in the snapshot, it is stored as an i64
@@ -210,7 +212,7 @@ impl RetentionCalculator for CheckpointWriter {
 
 impl CheckpointWriter {
     /// Creates a new [`CheckpointWriter`] for the given snapshot.
-    pub(crate) fn try_new(snapshot: Arc<Snapshot>) -> DeltaResult<Self> {
+    pub(crate) fn try_new(snapshot: SnapshotRef) -> DeltaResult<Self> {
         let version = i64::try_from(snapshot.version()).map_err(|e| {
             Error::CheckpointWrite(format!(
                 "Failed to convert checkpoint version from u64 {} to i64: {}",