Merge branch 'main' into fd-remove

Author: Fokko

AGENTS.md

@@ -315,6 +315,10 @@ Keep this list updated when new protocol features are added to kernel.
   block closes, and emit via `warn!()` only then. (`on_event`'s visitor does no
   warning-eligible work, so it may run under the lock directly.) See
   `kernel/src/metrics/reporter.rs` for the canonical pattern.
+- **Keep tests with process-global state safe under concurrency:** `cargo test` runs tests as
+  parallel threads in one test binary, while nextest normally runs each test in a separate
+  process. Tests for global tracing subscribers and callbacks must not share capture buffers with
+  thread-local dispatch tests or assume no other thread can emit an event.
 
 ## Code Style
 

Cargo.lock

@@ -390,13 +390,11 @@ fn struct_columns_from_patch(
 /// types come from `output_type`, which must be a struct holding only primitive fields (matching
 /// the kernel evaluator, which supports only primitive targets).
 ///
-/// Each field extracts its value with `cast(get_field(map, name), T)`. For a numeric or temporal
-/// type the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null before
-/// the cast, so an empty string becomes null (kernel's `empty_string_partition_cast`) while an
-/// unparseable value fails the cast (kernel's hard parse error). String and Binary keep the raw
-/// value (empty is a valid empty string / empty bytes). A missing key or null value is already null
-/// via [`get_field`]. The whole struct is nulled where the input map row is null, via `<map> IS NOT
-/// NULL`.
+/// Each field extracts its value with `cast(get_field(map, name), T)`. For every type except String
+/// and Binary, the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null
+/// before the cast. String and Binary keep the raw value because empty strings and bytes are valid.
+/// A missing key or null value is already null via [`get_field`]. The whole struct is nulled where
+/// the input map row is null, via `<map> IS NOT NULL`.
 ///
 /// KNOWN DIVERGENCES from the kernel parser, confined to malformed or non-spec-compliant values
 /// (spec-compliant writers never emit them):
@@ -408,14 +406,20 @@ fn struct_columns_from_patch(
 ///   value's scale to match the target's exactly (and hard-errors otherwise).
 ///
 /// # Errors
-/// Returns an error when `output_type` is absent, not a struct, or has a non-primitive field, or
-/// from lowering the map expression.
+///
+/// Returns an error when options are configured, `output_type` is absent, not a struct, or has a
+/// non-primitive field, or from lowering the map expression.
 fn map_to_struct_to_df_expr(
     map_to_struct: &MapToStructExpression,
     input_schema: &StructType,
     output_type: Option<&KernelDataType>,
 ) -> DeltaResult<DFExpr> {
     let target = require_struct_output(output_type, "MapToStruct")?;
+    if !map_to_struct.options.is_default() {
+        return Err(Error::unsupported(
+            "DataFusion execution of MapToStruct with configured options",
+        ));
+    }
     let map = to_df_expr(&map_to_struct.map_expr, input_schema, None)?;
 
     let mut args = Vec::with_capacity(target.num_fields() * 2);
@@ -547,7 +551,7 @@ mod tests {
     use datafusion::physical_expr::execution_props::ExecutionProps;
     use delta_kernel::expressions::{
         col, lit, null_lit, ColumnName as KernelColumnName, Expression as KernelExpr,
-        ExpressionStructPatch, ExpressionStructPatchBuilder,
+        ExpressionStructPatch, ExpressionStructPatchBuilder, MapToStructOptions,
     };
     use delta_kernel::schema::{schema, schema_ref, ArrayType, DataType, MapType, StructType};
     use rstest::rstest;
@@ -1001,7 +1005,7 @@ mod tests {
     /// Lowers a `MapToStruct` over `pv` targeting `output_schema` and renders it as a `Display`
     /// string.
     fn lower_map_to_struct(output_schema: StructType) -> String {
-        let kernel = KernelExpr::map_to_struct(col!("pv"));
+        let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
         let target: DataType = output_schema.into();
         to_df_expr(&kernel, &pv_map_schema(), Some(&target))
             .unwrap()
@@ -1063,13 +1067,28 @@ mod tests {
         #[case] output_type: Option<DataType>,
         #[case] expected_message: &str,
     ) {
-        let kernel = KernelExpr::map_to_struct(col!("pv"));
+        let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
         let err = to_df_expr(&kernel, &pv_map_schema(), output_type.as_ref())
             .unwrap_err()
             .to_string();
         assert!(err.contains(expected_message), "{err}");
     }
 
+    #[test]
+    fn configured_map_to_struct_is_unsupported() {
+        let target = DataType::from(schema! { nullable "ts": TIMESTAMP });
+        let kernel = KernelExpr::map_to_struct(
+            col!("pv"),
+            MapToStructOptions::default().with_timestamp_timezone("America/Los_Angeles"),
+        );
+
+        let error = to_df_expr(&kernel, &pv_map_schema(), Some(&target))
+            .unwrap_err()
+            .to_string();
+
+        assert!(error.contains("MapToStruct with configured options"));
+    }
+
     // === ParseJson Shared Helpers ===
 
     /// Input schema for JSON tests: `{ j: string }`.

datafusion-executor/Cargo.lock

@@ -1,5 +1,5 @@
 use proc_macro2::{Ident, Span, TokenStream};
-use quote::{quote, quote_spanned, ToTokens};
+use quote::{quote, quote_spanned};
 use syn::parse::Parser;
 use syn::punctuated::Punctuated;
 use syn::spanned::Spanned;
@@ -551,23 +551,30 @@ pub fn internal_api(
     item: proc_macro::TokenStream,
 ) -> proc_macro::TokenStream {
     let input = parse_macro_input!(item as Item);
+    internal_api_impl(input).into()
+}
 
+fn internal_api_impl(input: Item) -> TokenStream {
     // Create a version with public visibility for the unstable feature
-    let public_version = make_public(input.clone());
+    let public_version = match make_public(input.clone()) {
+        Ok(public_version) => public_version,
+        Err(err) => {
+            let error = err.to_compile_error();
+            return quote! { #input #error };
+        }
+    };
 
     // The original item stays as-is for the non-unstable case
-    let output = quote! {
+    quote! {
         #[cfg(feature = "internal-api")]
         #public_version
 
         #[cfg(not(feature = "internal-api"))]
         #input
-    };
-
-    output.into()
+    }
 }
 
-fn make_public(mut item: Item) -> Item {
+fn make_public(mut item: Item) -> Result<Item, Error> {
     /// Transforms the passed visibility to be `pub`. We pass the original span that the visibility
     /// came from, and attach it to the newly created pub token. This means that the compiler treats
     /// it as user-written code and normal lints apply. We want this because it allows us to catch
@@ -591,7 +598,7 @@ fn make_public(mut item: Item) -> Item {
         }};
     }
 
-    let result = match &mut item {
+    match &mut item {
         Item::Fn(f) => set_vis!(f),
         Item::Struct(s) => set_vis!(s),
         Item::Enum(e) => set_vis!(e),
@@ -606,24 +613,30 @@ fn make_public(mut item: Item) -> Item {
             item.span(),
             format!("unsupported item type for #[internal_api]: {item:?}"),
         )),
-    };
-
-    if let Err(err) = result {
-        let error = err.to_compile_error();
-        let mut tokens = item.to_token_stream();
-        tokens.extend(error);
-        return syn::parse_quote!(#tokens);
-    }
-
-    item
+    }?;
+    Ok(item)
 }
 
 #[cfg(test)]
 mod tests {
     use rstest::rstest;
+    use syn::parse_quote;
 
     use super::*;
 
+    #[test]
+    fn internal_api_rejects_public_items_without_panicking() {
+        let input = parse_quote!(
+            pub fn already_public() {}
+        );
+
+        let output = internal_api_impl(input).to_string();
+
+        assert!(output.contains("pub fn already_public"));
+        assert!(output.contains("compile_error"));
+        assert!(output.contains("item is already public"));
+    }
+
     /// Expand `gen_schema_fields` for `input` and return the generated tokens as a string. Macro
     /// errors are embedded as `compile_error!` tokens in that string; `Err` only signals that the
     /// input itself failed to parse as a `DeriveInput`.

datafusion-executor/src/expression.rs

@@ -161,6 +161,7 @@ engine-owned memory.
 | `visit_schema` | Walk a `SharedSchema` by invoking per-field callbacks on an `EngineSchemaVisitor` |
 | `visit_protocol` | Invoke a `visit_versions` callback, then a `visit_feature` callback per reader/writer feature |
 | `visit_metadata` | Invoke a single callback with `(id, name, description, format_provider, has_created_time, created_time_ms)` |
+| `visit_metadata_format_options` | Iterate arbitrary format option key/value pairs from a `SharedMetadata` handle |
 | `visit_metadata_configuration` | Iterate the `configuration` key/value map (takes a snapshot handle, not a metadata handle) |
 | `visit_string_map` / `get_from_string_map` | Iterate or look up entries in an opaque `CStringMap` (used by both metadata and scan-metadata surfaces) |
 
@@ -383,7 +384,7 @@ pattern is the same in every case:
 
 Callbacks run synchronously on the same thread that called `visit_*`. Strings
 passed to callbacks (`KernelStringSlice`) are borrowed for the duration of the
-call; copy them if you need to retain them beyond the callback.
+call. Copy them if you need to retain them beyond the callback.
 
 ## Error handling
 

derive-macros/src/lib.rs

@@ -9,6 +9,12 @@ use crate::{
     OptionalValue, SharedExternEngine, SharedSnapshot, TryFromStringSlice,
 };
 
+/// The row-tracking high-water mark before any fresh row IDs have been assigned.
+pub const ROW_TRACKING_INITIAL_HIGH_WATER_MARK: i64 = -1;
+
+const _: [(); 1] = [(); (ROW_TRACKING_INITIAL_HIGH_WATER_MARK
+    == delta_kernel::ROW_TRACKING_INITIAL_HIGH_WATER_MARK) as usize];
+
 /// Get the domain metadata as an optional string allocated by `AllocatedStringFn` for a specific
 /// domain in this snapshot
 ///
@@ -40,6 +46,29 @@ fn get_domain_metadata_impl(
         .and_then(|config| allocate_fn(kernel_string_slice!(config))))
 }
 
+/// Get the row-tracking high-water mark for this snapshot.
+///
+/// Returns [`OptionalValue::None`] when the snapshot has no active `delta.rowTracking` domain
+/// metadata. [`OptionalValue::Some`] with a value of `ROW_TRACKING_INITIAL_HIGH_WATER_MARK` means
+/// row tracking is active but no row IDs have been assigned yet. Returns an error if the domain
+/// metadata cannot be read or its JSON configuration is malformed.
+///
+/// # Safety
+///
+/// Caller is responsible for passing valid snapshot and engine handles.
+#[no_mangle]
+pub unsafe extern "C" fn snapshot_row_tracking_high_water_mark(
+    snapshot: Handle<SharedSnapshot>,
+    engine: Handle<SharedExternEngine>,
+) -> ExternResult<OptionalValue<i64>> {
+    let engine_ref = unsafe { engine.as_ref() };
+    let snapshot = unsafe { snapshot.as_ref() };
+    snapshot
+        .get_row_tracking_high_water_mark(engine_ref.engine().as_ref())
+        .map(OptionalValue::from)
+        .into_extern_result(&engine_ref)
+}
+
 /// Signature of the callback invoked once per clustering column by
 /// [`visit_clustering_columns`], in the order the columns appear in the `delta.clustering`
 /// domain. Each invocation describes one column: