feat: native spark_unbase64 kernel

docs/source/contributor-guide/expression-audits/string_funcs.md

@@ -245,6 +245,14 @@
 - Spark 3.5.8 (audited 2026-05-27): registry alias of `Upper`. Same support as `upper`.
 - Spark 4.0.1 (audited 2026-05-27): unchanged alias of `Upper`.
 
+## unbase64
+
+- Spark 3.4.3 (audited 2026-08-24): identical to 3.5.8.
+- Spark 3.5.8 (audited 2026-08-24): baseline. `doGenCode` emits `java.util.Base64.getMimeDecoder().decode(child.toString())`. The MIME decoder skips every byte outside the base64 alphabet, so CRLF-wrapped output from Spark's own `base64` round-trips cleanly; four terminal-shape errors surface as `IllegalArgumentException`. `failOnError = true` is set only when the node is constructed from `to_binary('base64')` / `try_to_binary`, which use a stricter RFC 4648 validator.
+- Spark 4.0.1 (audited 2026-08-24): `NullIntolerant` becomes `override def nullIntolerant: Boolean = true`; `inputTypes` widens to `StringTypeWithCollation(supportsTrimCollation = true)`. Behaviour is byte-level and collation-independent, so no divergence for `UTF8_BINARY` and no shim is needed.
+- Spark 4.1.1 (audited 2026-08-24): adds `contextIndependentFoldable`; no behavioural change on the decode path.
+- Comet native implementation (`spark_unbase64`, `native/spark-expr/src/string_funcs/unbase64.rs`) ports the JDK MIME decoder rules: 256-entry decode LUT, a reused per-batch scratch `Vec<u8>` copied into a preallocated `BinaryBuilder`, all four error messages reproduced verbatim. `CometUnBase64` handles `failOnError = false` natively; `failOnError = true` (reachable from `to_binary('base64')` / `try_to_binary`) requires strict RFC 4648 validation and is not yet implemented natively, so those cases stay on the JVM codegen dispatcher via `CodegenDispatchFallback`.
+
 ## upper
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.

native/spark-expr/Cargo.toml

@@ -99,6 +99,10 @@ harness = false
 name = "base64"
 harness = false
 
+[[bench]]
+name = "unbase64"
+harness = false
+
 [[bench]]
 name = "date_trunc"
 harness = false

native/spark-expr/benches/unbase64.rs

@@ -0,0 +1,165 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow::array::builder::StringBuilder;
+use arrow::array::ArrayRef;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion::common::ScalarValue;
+use datafusion::physical_plan::ColumnarValue;
+use datafusion_comet_spark_expr::spark_unbase64;
+use std::hint::black_box;
+use std::sync::Arc;
+
+const LINE_LEN: usize = 76;
+
+fn unwrapped(bytes: &[u8]) -> String {
+    BASE64_STANDARD.encode(bytes)
+}
+
+/// Reproduces Spark's default `base64` output shape: MIME-wrapped at 76 chars with CRLF.
+fn crlf_wrapped(bytes: &[u8]) -> String {
+    let encoded = unwrapped(bytes);
+    encoded
+        .as_bytes()
+        .chunks(LINE_LEN)
+        .map(|line| std::str::from_utf8(line).unwrap())
+        .collect::<Vec<_>>()
+        .join("\r\n")
+}
+
+/// Density of null placement, expressed as "1 null every `stride` rows". Use `usize::MAX` for
+/// no-null batches. `stride == 1` produces the all-null batch.
+enum NullDensity {
+    None,
+    /// One null every `stride` rows. `stride == 10` is sparse, `stride == 2` is dense.
+    Every(usize),
+    All,
+}
+
+fn create_string_array(size: usize, value: &str, nulls: NullDensity) -> ArrayRef {
+    let mut builder = StringBuilder::new();
+    for i in 0..size {
+        let is_null = match nulls {
+            NullDensity::None => false,
+            NullDensity::All => true,
+            NullDensity::Every(stride) => i % stride == 0,
+        };
+        if is_null {
+            builder.append_null();
+        } else {
+            builder.append_value(value);
+        }
+    }
+    Arc::new(builder.finish())
+}
+
+/// Builds a batch of mostly-valid inputs sprinkled with malformed values that trip
+/// `Input byte array has wrong 4-byte ending unit`. The kernel returns early on the first bad
+/// row, so this bench measures the error-return path (partial decode + Result short-circuit)
+/// rather than the happy path.
+fn create_error_shaped_array(size: usize, valid: &str) -> ArrayRef {
+    let mut builder = StringBuilder::new();
+    // First row is the malformed one so the kernel short-circuits immediately.
+    builder.append_value("YW=");
+    for _ in 1..size {
+        builder.append_value(valid);
+    }
+    Arc::new(builder.finish())
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let size = 8192;
+    let short_bytes = vec![b'z'; 16];
+    let long_bytes = vec![b'q'; 200];
+
+    // Sparse-null (one every 10 rows) — the shape the previous bench measured, kept as the
+    // default null density for the short / long / tiny cases.
+    let short = create_string_array(size, &unwrapped(&short_bytes), NullDensity::Every(10));
+    let long_clean = create_string_array(size, &unwrapped(&long_bytes), NullDensity::Every(10));
+    // Long CRLF-wrapped values: matches `unbase64(base64(x))` when Spark's default
+    // `spark.sql.chunkBase64String.enabled = true` is in effect (also Comet's default).
+    let long_wrapped =
+        create_string_array(size, &crlf_wrapped(&long_bytes), NullDensity::Every(10));
+    // A batch dominated by tiny values, one per row (worst case for per-row overhead).
+    let tiny = create_string_array(size, &unwrapped(b"a"), NullDensity::Every(10));
+
+    // No-nulls / dense-nulls shapes on the long-single-line payload isolate the null-append
+    // branch from the decode branch. Dense-null uses stride 2 (~50% nulls) rather than an
+    // all-null shape so the decoder still runs on half the rows.
+    let long_no_nulls = create_string_array(size, &unwrapped(&long_bytes), NullDensity::None);
+    let long_dense_nulls =
+        create_string_array(size, &unwrapped(&long_bytes), NullDensity::Every(2));
+    let long_all_nulls = create_string_array(size, &unwrapped(&long_bytes), NullDensity::All);
+
+    // Error-path shape: the first row throws, the rest are valid but never decoded. Measures the
+    // early-return path independent of decode throughput.
+    let error_first = create_error_shaped_array(size, &unwrapped(&long_bytes));
+
+    c.bench_function("spark_unbase64: short", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&short))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: long, single line", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&long_clean))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: long, CRLF-wrapped", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&long_wrapped))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: tiny values", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&tiny))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: long, no nulls", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&long_no_nulls))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: long, dense nulls (50%)", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&long_dense_nulls))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: long, all nulls", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&long_all_nulls))];
+        b.iter(|| black_box(spark_unbase64(black_box(&args)).unwrap()))
+    });
+
+    c.bench_function("spark_unbase64: error on first row", |b| {
+        let args = vec![ColumnarValue::Array(Arc::clone(&error_first))];
+        b.iter(|| {
+            let result = spark_unbase64(black_box(&args));
+            debug_assert!(result.is_err());
+            black_box(result.err());
+        })
+    });
+
+    c.bench_function("spark_unbase64: scalar literal", |b| {
+        let arg = ColumnarValue::Scalar(ScalarValue::Utf8(Some(unwrapped(&long_bytes))));
+        b.iter(|| black_box(spark_unbase64(black_box(std::slice::from_ref(&arg))).unwrap()))
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);

native/spark-expr/src/comet_scalar_funcs.rs

@@ -217,6 +217,10 @@ pub fn create_comet_physical_fun_with_eval_mode(
             let func = Arc::new(crate::string_funcs::spark_base64);
             make_comet_scalar_udf!("base64", func, without data_type)
         }
+        "unbase64" => {
+            let func = Arc::new(crate::string_funcs::spark_unbase64);
+            make_comet_scalar_udf!("unbase64", func, without data_type)
+        }
         "split" => {
             let func = Arc::new(crate::string_funcs::spark_split);
             make_comet_scalar_udf!("split", func, without data_type)

native/spark-expr/src/string_funcs/mod.rs

@@ -23,6 +23,7 @@ mod regexp_extract;
 mod regexp_extract_all;
 mod regexp_extract_common;
 mod split;
+mod unbase64;
 
 pub use base64::spark_base64;
 pub use contains::SparkContains;
@@ -31,3 +32,4 @@ pub use levenshtein::spark_levenshtein;
 pub use regexp_extract::spark_regexp_extract;
 pub use regexp_extract_all::spark_regexp_extract_all;
 pub use split::{spark_split, spark_split_sql};
+pub use unbase64::spark_unbase64;