From 49868013015a49962ee6fa9c25e4132824fc89ec Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Tue, 25 Aug 2026 00:04:55 +0800 Subject: [PATCH 1/4] feat: native `spark_unbase64` kernel --- .../expression-audits/string_funcs.md | 8 + native/spark-expr/Cargo.toml | 4 + native/spark-expr/benches/unbase64.rs | 164 ++++++++ native/spark-expr/src/comet_scalar_funcs.rs | 4 + native/spark-expr/src/string_funcs/mod.rs | 2 + .../spark-expr/src/string_funcs/unbase64.rs | 380 ++++++++++++++++++ .../org/apache/comet/serde/strings.scala | 29 +- .../expressions/string/to_binary.sql | 9 +- .../sql-tests/expressions/string/unbase64.sql | 66 ++- .../benchmark/CometUnBase64Benchmark.scala | 108 +++++ 10 files changed, 766 insertions(+), 8 deletions(-) create mode 100644 native/spark-expr/benches/unbase64.rs create mode 100644 native/spark-expr/src/string_funcs/unbase64.rs create mode 100644 spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala diff --git a/docs/source/contributor-guide/expression-audits/string_funcs.md b/docs/source/contributor-guide/expression-audits/string_funcs.md index 58b300ebc12..d3e84f70169 100644 --- a/docs/source/contributor-guide/expression-audits/string_funcs.md +++ b/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` 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. diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml index 6faa9fec4ec..0349c3300d5 100644 --- a/native/spark-expr/Cargo.toml +++ b/native/spark-expr/Cargo.toml @@ -100,6 +100,10 @@ harness = false name = "base64" harness = false +[[bench]] +name = "unbase64" +harness = false + [[bench]] name = "date_trunc" harness = false diff --git a/native/spark-expr/benches/unbase64.rs b/native/spark-expr/benches/unbase64.rs new file mode 100644 index 00000000000..43bb54a4b35 --- /dev/null +++ b/native/spark-expr/benches/unbase64.rs @@ -0,0 +1,164 @@ +// 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::>() + .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); diff --git a/native/spark-expr/src/comet_scalar_funcs.rs b/native/spark-expr/src/comet_scalar_funcs.rs index dcb6b1906ce..c68b998e617 100644 --- a/native/spark-expr/src/comet_scalar_funcs.rs +++ b/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) diff --git a/native/spark-expr/src/string_funcs/mod.rs b/native/spark-expr/src/string_funcs/mod.rs index 5dd422a04e3..dc51cfea1b3 100644 --- a/native/spark-expr/src/string_funcs/mod.rs +++ b/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; diff --git a/native/spark-expr/src/string_funcs/unbase64.rs b/native/spark-expr/src/string_funcs/unbase64.rs new file mode 100644 index 00000000000..b3d8065f084 --- /dev/null +++ b/native/spark-expr/src/string_funcs/unbase64.rs @@ -0,0 +1,380 @@ +// 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 std::sync::Arc; + +use arrow::array::{ + Array, AsArray, BinaryArray, BinaryBuilder, GenericStringArray, OffsetSizeTrait, +}; +use arrow::datatypes::DataType; +use datafusion::common::{exec_err, DataFusionError, ScalarValue}; +use datafusion::physical_plan::ColumnarValue; + +/// Spark `unbase64(str)`: decodes a base64 string to binary using JDK's MIME decoder rules. +/// +/// Matches `java.util.Base64.getMimeDecoder().decode(str.getBytes(ISO_8859_1))`: every byte +/// outside the base64 alphabet is skipped (so CRLF-wrapped output from Spark's own `base64` +/// round-trips cleanly), and the four terminal-shape error conditions are reproduced with +/// matching messages. +pub fn spark_unbase64(args: &[ColumnarValue]) -> Result { + if args.len() != 1 { + return exec_err!("unbase64 expects exactly one argument, got {}", args.len()); + } + match &args[0] { + ColumnarValue::Array(array) => match array.data_type() { + DataType::Utf8 => Ok(ColumnarValue::Array(Arc::new(decode_array( + array.as_string::(), + )?))), + DataType::LargeUtf8 => Ok(ColumnarValue::Array(Arc::new(decode_array( + array.as_string::(), + )?))), + // Comet's planner coerces string inputs back to Utf8 for well-supported UDF + // signatures, so Utf8View is not expected to reach us today. Fail loudly if it + // ever does, rather than silently materialise. + other => exec_err!("unbase64 expects a string argument, got {other}"), + }, + ColumnarValue::Scalar(ScalarValue::Utf8(value)) + | ColumnarValue::Scalar(ScalarValue::LargeUtf8(value)) => { + let decoded = value + .as_ref() + .map(|s| { + let mut out = Vec::new(); + decode(s.as_bytes(), &mut out).map(|_| out) + }) + .transpose()?; + Ok(ColumnarValue::Scalar(ScalarValue::Binary(decoded))) + } + ColumnarValue::Scalar(other) => { + exec_err!("unbase64 expects a string argument, got {other}") + } + } +} + +/// Sentinel in `BASE64_LUT` for bytes that are not in the base64 alphabet. +const INVALID: u8 = 0xFF; +/// Sentinel in `BASE64_LUT` for the padding character `=`. +const PADDING: u8 = 0xFE; + +/// Lookup table mapping each possible input byte to its 6-bit base64 value, `PADDING` for `=`, +/// or `INVALID` for every other byte. Built at compile time so the hot loop is a single indexed +/// load per input byte. +const BASE64_LUT: [u8; 256] = build_base64_lut(); + +const fn build_base64_lut() -> [u8; 256] { + let mut lut = [INVALID; 256]; + let mut i = 0; + while i < 26 { + lut[b'A' as usize + i] = i as u8; + lut[b'a' as usize + i] = (i + 26) as u8; + i += 1; + } + let mut i = 0; + while i < 10 { + lut[b'0' as usize + i] = (i + 52) as u8; + i += 1; + } + lut[b'+' as usize] = 62; + lut[b'/' as usize] = 63; + lut[b'=' as usize] = PADDING; + lut +} + +const ERR_WRONG_ENDING: &str = "Input byte array has wrong 4-byte ending unit"; +const ERR_NOT_ENOUGH_BITS: &str = "Last unit does not have enough valid bits"; + +/// Returns `exec_err` if `input_bytes` cannot fit inside a `BinaryArray`'s i32 offset range. +/// +/// `GenericByteBuilder::next_offset` in arrow-rs is `.expect("byte array offset overflow")`, so +/// once the running offset would cross `i32::MAX` the append panics. That panic bubbles across +/// the JNI boundary without Comet context; the upfront guard turns it into an `exec_err` with +/// the actual input size, which is what surfaces in the Spark task log. +/// +/// The bound is conservative: the decoder produces at most `input_bytes * 3 / 4` output bytes +/// (skipped non-alphabet bytes shrink it further), so an input at this bound decodes to well +/// under `i32::MAX`. Rejecting at input length keeps the check O(1) and independent of the +/// decoder implementation. +fn check_binary_capacity(input_bytes: usize) -> Result<(), DataFusionError> { + if input_bytes > i32::MAX as usize { + return exec_err!( + "unbase64 input of {input_bytes} bytes exceeds BinaryArray capacity ({} bytes)", + i32::MAX + ); + } + Ok(()) +} + +// Both Utf8 and LargeUtf8 inputs decode to i32-offset `BinaryArray`. Comet's shuffle path +// does not otherwise plumb `LargeBinaryArray`, so widening the return type would ripple +// through every downstream consumer of this column; the `check_binary_capacity` guard on the +// LargeUtf8 path turns the only reachable overflow into a clean `exec_err` instead. +fn decode_array( + array: &GenericStringArray, +) -> Result { + // Byte span of the slice (last offset − first offset), not the underlying buffer length. + // `value_data().len()` reports the whole buffer and ignores slicing, which would + // false-positive the capacity guard on a small slice into a large parent. + let offsets = array.value_offsets(); + let input_bytes = (offsets[array.len()] - offsets[0]).as_usize(); + if O::IS_LARGE { + check_binary_capacity(input_bytes)?; + } + // Upper bound: N alphabet chars decode to `ceil(N/4)*3` bytes, and alphabet chars <= total + // input bytes. Over-reserves for inputs padded heavily with skipped bytes (long CRLF-wrapped + // values), which is acceptable. + let capacity = input_bytes.div_ceil(4) * 3; + let mut builder = BinaryBuilder::with_capacity(array.len(), capacity); + let mut scratch = Vec::new(); + for i in 0..array.len() { + if array.is_null(i) { + builder.append_null(); + continue; + } + scratch.clear(); + decode(array.value(i).as_bytes(), &mut scratch)?; + builder.append_value(&scratch); + } + Ok(builder.finish()) +} + +/// Decodes `src` in MIME mode, appending decoded bytes to `out`. +/// +/// Mirrors `java.util.Base64.Decoder.decode0` with `isMIME = true`: every byte outside the base64 +/// alphabet is skipped, and the four terminal-shape error conditions are reproduced verbatim so +/// the error messages match Spark's codegen-dispatched path. +fn decode(src: &[u8], out: &mut Vec) -> Result<(), DataFusionError> { + out.reserve(src.len().div_ceil(4) * 3); + + let mut bits: u32 = 0; + // Position, in bits, where the next 6-bit group's high bit lands inside a 24-bit atom. + // Steps 18 -> 12 -> 6 -> 0 for the four chars of a group; wraps back to 18 after emit. + let mut shift: i32 = 18; + let mut sp = 0; + while sp < src.len() { + let raw = src[sp]; + sp += 1; + let v = BASE64_LUT[raw as usize]; + if v == INVALID { + continue; + } + if v == PADDING { + if shift == 18 { + // '=' with no data before it in this group. + return Err(DataFusionError::Execution(ERR_WRONG_ENDING.into())); + } + if shift == 6 { + // `xx=` shape: the immediate next byte must also be `=`. JDK reads it adjacent, + // without skipping non-alphabet bytes, so we do the same for message parity. + if sp >= src.len() || src[sp] != b'=' { + return Err(DataFusionError::Execution(ERR_WRONG_ENDING.into())); + } + sp += 1; + } + break; + } + bits |= (v as u32) << shift; + shift -= 6; + if shift < 0 { + out.push((bits >> 16) as u8); + out.push((bits >> 8) as u8); + out.push(bits as u8); + shift = 18; + bits = 0; + } + } + + // Finalize the partial group. shift reflects how many chars we consumed in the tail. + match shift { + 18 => {} + 12 => return Err(DataFusionError::Execution(ERR_NOT_ENOUGH_BITS.into())), + 6 => out.push((bits >> 16) as u8), + 0 => { + out.push((bits >> 16) as u8); + out.push((bits >> 8) as u8); + } + _ => unreachable!("shift is always one of 18, 12, 6, 0"), + } + + // After padding, only real alphabet characters are errors. JDK's trailing loop treats any + // byte with `base64[b] < 0` as skippable, which covers both non-alphabet bytes (INVALID) + // and additional padding bytes (PADDING) — so e.g. `YQ===` decodes cleanly to `a`. + while sp < src.len() { + let b = src[sp]; + sp += 1; + if BASE64_LUT[b as usize] < PADDING { + return Err(DataFusionError::Execution(format!( + "Input byte array has incorrect ending byte at {sp}" + ))); + } + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow::array::{LargeStringArray, StringArray}; + + fn dec(s: &str) -> Vec { + let mut out = Vec::new(); + decode(s.as_bytes(), &mut out).unwrap(); + out + } + + fn dec_err(s: &str) -> String { + decode(s.as_bytes(), &mut Vec::new()) + .unwrap_err() + .to_string() + } + + #[test] + fn empty_input() { + assert_eq!(dec(""), Vec::::new()); + } + + #[test] + fn standard_padded() { + assert_eq!(dec("YQ=="), b"a"); + assert_eq!(dec("YWI="), b"ab"); + assert_eq!(dec("YWJj"), b"abc"); + } + + #[test] + fn unpadded_accepted() { + // JDK MIME accepts truncated tails at clean boundaries. + assert_eq!(dec("YQ"), b"a"); + assert_eq!(dec("YWI"), b"ab"); + } + + #[test] + fn skips_non_alphabet_bytes() { + assert_eq!(dec("YW Jj"), b"abc"); + assert_eq!(dec("YWJj?"), b"abc"); + assert_eq!(dec("Y\r\nWJj"), b"abc"); + } + + #[test] + fn crlf_wrapped_round_trips() { + // 60-byte input encodes to 80 chars → one CRLF wrap at char 76 (matches spark_base64). + let raw: Vec = (0..60).map(|i| i as u8).collect(); + let encoded_flat = + "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8gISIjJCUmJygpKissLS4vMDEyMzQ1Njc4OTo7"; + let encoded_wrapped = format!("{}\r\n{}", &encoded_flat[..76], &encoded_flat[76..]); + assert_eq!(dec(encoded_flat), raw); + assert_eq!(dec(&encoded_wrapped), raw); + } + + #[test] + fn extra_trailing_padding_after_valid_pair() { + // Regression: JDK's MIME trailing loop uses `base64[b] < 0`, and '=' maps to -2 (< 0), + // so additional `=` bytes after the required pair are silently consumed. An earlier + // implementation compared against the INVALID sentinel only and rejected these. + assert_eq!(dec("YQ==="), b"a"); + assert_eq!(dec("YQ==\r\n="), b"a"); + } + + #[test] + fn err_dangling_single_char() { + assert!(dec_err("YWJjY").contains(ERR_NOT_ENOUGH_BITS)); + } + + #[test] + fn err_padding_without_data() { + assert!(dec_err("====").contains(ERR_WRONG_ENDING)); + } + + #[test] + fn err_missing_second_pad() { + // "YW=" is shape xx=, the second `=` is missing. + assert!(dec_err("YW=").contains(ERR_WRONG_ENDING)); + } + + #[test] + fn err_alphabet_after_padding() { + let err = dec_err("YQ==Z"); + assert!(err.contains("incorrect ending byte"), "{err}"); + } + + #[test] + fn decode_array_utf8_with_nulls() { + let input = StringArray::from(vec![Some("YWJj"), None, Some("YQ=="), Some("")]); + let out = decode_array(&input).unwrap(); + assert_eq!(out.len(), 4); + assert_eq!(out.value(0), b"abc"); + assert!(out.is_null(1)); + assert_eq!(out.value(2), b"a"); + assert_eq!(out.value(3), b""); + } + + #[test] + fn check_binary_capacity_at_boundary() { + assert!(check_binary_capacity(0).is_ok()); + assert!(check_binary_capacity(i32::MAX as usize).is_ok()); + assert!(check_binary_capacity(i32::MAX as usize + 1).is_err()); + } + + #[test] + fn spark_unbase64_scalar_utf8() { + let args = vec![ColumnarValue::Scalar(ScalarValue::Utf8(Some( + "YWJj".into(), + )))]; + match spark_unbase64(&args).unwrap() { + ColumnarValue::Scalar(ScalarValue::Binary(Some(v))) => assert_eq!(v, b"abc"), + other => panic!("unexpected result: {other:?}"), + } + } + + #[test] + fn spark_unbase64_scalar_null() { + let args = vec![ColumnarValue::Scalar(ScalarValue::Utf8(None))]; + match spark_unbase64(&args).unwrap() { + ColumnarValue::Scalar(ScalarValue::Binary(None)) => {} + other => panic!("unexpected result: {other:?}"), + } + } + + #[test] + fn spark_unbase64_wrong_arity() { + assert!(spark_unbase64(&[]).is_err()); + let two = vec![ + ColumnarValue::Scalar(ScalarValue::Utf8(Some("YWJj".into()))), + ColumnarValue::Scalar(ScalarValue::Utf8(Some("YWJj".into()))), + ]; + assert!(spark_unbase64(&two).is_err()); + } + + #[test] + fn spark_unbase64_wrong_scalar_type() { + let args = vec![ColumnarValue::Scalar(ScalarValue::Int32(Some(42)))]; + assert!(spark_unbase64(&args).is_err()); + } + + #[test] + fn spark_unbase64_large_utf8_array() { + let input = LargeStringArray::from(vec![Some("YWJj"), None, Some("YQ==")]); + match spark_unbase64(&[ColumnarValue::Array(Arc::new(input))]).unwrap() { + ColumnarValue::Array(out) => { + let bin = out.as_binary::(); + assert_eq!(bin.len(), 3); + assert_eq!(bin.value(0), b"abc"); + assert!(bin.is_null(1)); + assert_eq!(bin.value(2), b"a"); + } + other => panic!("unexpected result: {other:?}"), + } + } +} diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index ebf45089882..4513a54de1a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -713,7 +713,34 @@ object CometBase64 extends CometExpressionSerde[Base64] { } } -object CometUnBase64 extends CometCodegenDispatch[UnBase64] +// Base64.getMimeDecoder() semantics: skips non-alphabet bytes and matches Spark's codegen +// path. The native path handles the default UnBase64 (failOnError = false, reachable from SQL +// `unbase64(...)`). When failOnError = true (from `to_binary('base64')` / `try_to_binary`), +// Spark uses a stricter RFC 4648 validator, so those cases stay on the JVM codegen dispatcher +// via CodegenDispatchFallback. Error messages match Spark byte-for-byte (pinned in the Rust +// unit tests), but the wrapping exception class does not; kept as Compatible() because Spark +// surfaces these as bare IllegalArgumentException without a SQL error class. +object CometUnBase64 extends CometExpressionSerde[UnBase64] with CodegenDispatchFallback { + + private val failOnErrorReason = + "unbase64 with failOnError = true uses stricter RFC 4648 validation that is not yet" + + " implemented natively" + + override def getUnsupportedReasons(): Seq[String] = Seq(failOnErrorReason) + + override def getSupportLevel(expr: UnBase64): SupportLevel = { + if (expr.failOnError) Unsupported(Some(failOnErrorReason)) else Compatible() + } + + override def convert(expr: UnBase64, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = { + val childExpr = exprToProtoInternal(expr.child, inputs, binding) + scalarFunctionExprToProtoWithReturnType( + "unbase64", + BinaryType, + failOnError = false, + childExpr) + } +} object CometToCharacter extends CometCodegenDispatch[ToCharacter] diff --git a/spark/src/test/resources/sql-tests/expressions/string/to_binary.sql b/spark/src/test/resources/sql-tests/expressions/string/to_binary.sql index 67837919b32..504f75c42d8 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/to_binary.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/to_binary.sql @@ -16,7 +16,10 @@ -- under the License. -- to_binary with 'hex' format lowers to Unhex, which Comet accelerates. --- to_binary with 'utf-8' and 'base64' formats fall back to Spark. +-- to_binary with 'utf-8' falls back to Spark. +-- to_binary with 'base64' lowers to UnBase64(failOnError = true), which CometUnBase64 marks +-- Unsupported (strict RFC 4648 validation is not yet ported) and routes through the JVM codegen +-- dispatcher via CodegenDispatchFallback, so results are byte-exact to Spark. statement CREATE TABLE test_to_binary(s string) USING parquet @@ -36,6 +39,6 @@ SELECT to_binary('41', 'hex'), to_binary('0A1B', 'hex'), to_binary('', 'hex'), t query spark_answer_only SELECT to_binary(s, 'utf-8') FROM test_to_binary --- base64 format falls back to Spark -query spark_answer_only +-- base64 format runs in-pipeline via CodegenDispatchFallback and matches Spark +query SELECT to_binary(s, 'base64') FROM test_to_binary diff --git a/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql b/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql index 8640827b867..e143374f087 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql @@ -15,17 +15,75 @@ -- specific language governing permissions and limitations -- under the License. --- Routes unbase64 through the codegen dispatcher so behavior matches Spark exactly. +-- unbase64 runs on a native Rust kernel that ports the JDK's +-- java.util.Base64.getMimeDecoder() rules: bytes outside the base64 alphabet are skipped, and +-- the terminal-shape errors are reproduced with matching messages. This fixture pins the +-- happy-path shapes; the terminal-shape error messages are pinned in the Rust unit tests, since +-- Spark surfaces them as raw IllegalArgumentException without a SQL error class. statement CREATE TABLE test_unbase64(s string) USING parquet +-- 'YWJj' -> 'abc' (no padding), 'YQ==' -> 'a' (two-pad), 'YWI=' -> 'ab' (one-pad), '' -> '', +-- NULL -> NULL, 'YW Jj' -> 'abc' (embedded space, skipped), 'Y\r\nWJj' -> 'abc' (embedded CRLF). statement -INSERT INTO test_unbase64 VALUES ('YWJj'), ('aGVsbG8='), (NULL) +INSERT INTO test_unbase64 VALUES + ('YWJj'), + ('YQ=='), + ('YWI='), + (''), + (NULL), + ('YW Jj'), + ('Y\r\nWJj') query SELECT s, hex(unbase64(s)) FROM test_unbase64 --- literal arguments +-- Literal arguments across the same shapes. query -SELECT hex(unbase64('YWJj')), hex(unbase64('aGVsbG8=')) +SELECT hex(unbase64('YWJj')), + hex(unbase64('YQ==')), + hex(unbase64('YWI=')), + hex(unbase64('')), + hex(unbase64('YW Jj')), + hex(unbase64('Y\r\nWJj')) + +-- Unpadded input is accepted by the MIME decoder at clean 4-char boundaries. +-- 'YQ' exits with 2 alphabet chars consumed (shift=6, emits 1 byte). +-- 'YWI' and 'YWJ' both exit with 3 alphabet chars (shift=0, emits 2 bytes). 'YWJ' has non-zero +-- low bits in the third char that MIME mode silently discards — the JDK decoder does not +-- strict-check them, and this row pins that behavior against the standard decoder. +query +SELECT hex(unbase64('YQ')), + hex(unbase64('YWI')), + hex(unbase64('YWJ')) + +-- Multi-byte UTF-8 bytes are outside the base64 alphabet, so every byte of the codepoint is +-- skipped. 'YQ==é' decodes to 'a' just like 'YQ=='; the trailing 2-byte é (0xC3 0xA9) is treated +-- the same as any non-alphabet byte after padding. '€' (0xE2 0x82 0xAC) covers the 3-byte +-- codepoint case in the same skip loop. Pin non-ASCII skipping alongside the space and CRLF +-- cases already covered above. +query +SELECT hex(unbase64('YQ==é')), + hex(unbase64('YW€Jj')), + hex(unbase64('YéWJj')) + +-- Round-trip against Comet's own base64 output. With the default +-- `spark.sql.chunkBase64String.enabled = true`, values longer than 57 raw bytes encode with CRLF +-- separators every 76 characters; unbase64 must skip those separators so the round-trip returns +-- the original bytes. +statement +CREATE TABLE test_unbase64_roundtrip(s string) USING parquet + +statement +INSERT INTO test_unbase64_roundtrip VALUES + ('abc'), + ('hello'), + (''), + (NULL), + ('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'), + ('bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'), + ('cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc') + +query +SELECT s, cast(unbase64(base64(cast(s AS binary))) AS string) FROM test_unbase64_roundtrip diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala new file mode 100644 index 00000000000..d5562ba6d8d --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala @@ -0,0 +1,108 @@ +/* + * 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. + */ + +package org.apache.spark.sql.benchmark + +import org.apache.comet.CometConf + +/** + * Configuration for a single unbase64 input shape under benchmark. Each shape is materialized as + * its own column in the source table, so the same benchmark harness that measures the other + * string expressions can address it by name. + * + * @param name + * short label for the shape + * @param column + * name of the pre-encoded source column to feed into unbase64 + */ +case class UnBase64Shape(name: String, column: String) + +/** + * Benchmarks `unbase64` end-to-end against Spark's JVM codegen path. The native kernel lives in + * `native/spark-expr/src/string_funcs/unbase64.rs` and has its own criterion bench for the decode + * loop; this file measures the full plan (scan + project + shuffle-free) so the reported number + * captures the same overhead a real query pays. + * + * Shapes match the criterion bench so the two views are directly comparable: short single-line + * values, long single-line values, long CRLF-wrapped values (the default output of Spark's + * `base64` when `spark.sql.chunkBase64String.enabled = true`), and tiny values (one base64 char + * per row, the worst case for per-row overhead). + * + * To run: + * {{{ + * SPARK_GENERATE_BENCHMARK_FILES=1 \ + * make benchmark-org.apache.spark.sql.benchmark.CometUnBase64Benchmark + * }}} + * + * Results land in `spark/benchmarks/CometUnBase64Benchmark-**results.txt`. + */ +object CometUnBase64Benchmark extends CometBenchmarkBase { + + private val shapes = List( + UnBase64Shape("short", "b64_short"), + UnBase64Shape("long_single_line", "b64_long"), + UnBase64Shape("long_crlf_wrapped", "b64_long_wrapped"), + UnBase64Shape("tiny", "b64_tiny")) + + override def runCometBenchmark(mainArgs: Array[String]): Unit = { + runBenchmarkWithTable("unbase64", 8192) { v => + withTempPath { dir => + withTempTable("parquetV1Table") { + // Pre-encode each shape at write time so unbase64 is the only decoder in the plan. + // `base64` on a 16-byte payload fits in one line; `base64` on a 200-byte payload with + // Spark's default chunking wraps at 76 chars with CRLF, which is the round-trip shape + // real workloads see. The tiny column encodes a single byte per row, isolating per-row + // overhead (offset walk, capacity guard, null-check) from decode throughput. + // `translate(x, concat(chr(13), chr(10)), '')` deletes both CR and LF from the base64 + // output, giving a long single-line variant of the same payload the wrapped column + // encodes. Avoid `'\r\n'` in a SQL string literal because Spark's parser does not + // interpret backslash escapes there. + prepareTable( + dir, + spark.sql(s""" + SELECT + base64(cast(repeat('z', 16) AS binary)) AS b64_short, + base64(cast(repeat('q', 200) AS binary)) AS b64_long_wrapped, + translate( + base64(cast(repeat('q', 200) AS binary)), + concat(chr(13), chr(10)), + '') AS b64_long, + base64(cast('a' AS binary)) AS b64_tiny + FROM $tbl + """)) + + shapes.foreach { s => + val query = s"select unbase64(${s.column}) from parquetV1Table" + runBenchmark(s.name) { + runUnBase64Modes(s.name, v, query) + } + } + } + } + } + } + + /** Runs Spark vs Comet-native for a single unbase64 shape. */ + private def runUnBase64Modes(name: String, cardinality: Long, query: String): Unit = { + // CometUnBase64 is `Compatible`, so no allowIncompatible flag is required for the native + // arm. There is no meaningful JVM-fallback intermediate case here (unlike RLike), so this is + // a straight Spark-vs-Comet comparison. + runExpressionBenchmark(name, cardinality, query) + } +} From ddcb967b7ea9dac3cdc0bde4a39708e027379d36 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Tue, 25 Aug 2026 00:23:20 +0800 Subject: [PATCH 2/4] fix check style --- native/spark-expr/benches/unbase64.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/native/spark-expr/benches/unbase64.rs b/native/spark-expr/benches/unbase64.rs index 43bb54a4b35..35e01250228 100644 --- a/native/spark-expr/benches/unbase64.rs +++ b/native/spark-expr/benches/unbase64.rs @@ -94,7 +94,8 @@ fn criterion_benchmark(c: &mut Criterion) { 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)); + 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)); From adc6bd97017cb4dd1ee539f3817fe7f1b43cc632 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Tue, 25 Aug 2026 00:23:58 +0800 Subject: [PATCH 3/4] remove unused CometConf import --- .../org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala index d5562ba6d8d..20ebfb9df8f 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometUnBase64Benchmark.scala @@ -19,8 +19,6 @@ package org.apache.spark.sql.benchmark -import org.apache.comet.CometConf - /** * Configuration for a single unbase64 input shape under benchmark. Each shape is materialized as * its own column in the source table, so the same benchmark harness that measures the other From 8a88da0ab13c22776dd3d22c19d562d144ce7ca6 Mon Sep 17 00:00:00 2001 From: 0lai0 Date: Tue, 25 Aug 2026 22:02:49 +0800 Subject: [PATCH 4/4] add regression coverage --- .../org/apache/comet/serde/strings.scala | 29 +++++++++---- .../sql-tests/expressions/string/unbase64.sql | 18 ++++++++ ...nbase64_concat_short_circuit_spark_3_5.sql | 42 +++++++++++++++++++ 3 files changed, 82 insertions(+), 7 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/expressions/string/unbase64_concat_short_circuit_spark_3_5.sql diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala index 4513a54de1a..665c0d3b54e 100644 --- a/spark/src/main/scala/org/apache/comet/serde/strings.scala +++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala @@ -715,21 +715,36 @@ object CometBase64 extends CometExpressionSerde[Base64] { // Base64.getMimeDecoder() semantics: skips non-alphabet bytes and matches Spark's codegen // path. The native path handles the default UnBase64 (failOnError = false, reachable from SQL -// `unbase64(...)`). When failOnError = true (from `to_binary('base64')` / `try_to_binary`), -// Spark uses a stricter RFC 4648 validator, so those cases stay on the JVM codegen dispatcher -// via CodegenDispatchFallback. Error messages match Spark byte-for-byte (pinned in the Rust -// unit tests), but the wrapping exception class does not; kept as Compatible() because Spark -// surfaces these as bare IllegalArgumentException without a SQL error class. +// `unbase64(...)`) only when the child is a column reference or literal, since native +// ScalarFunctionExpr evaluates its arguments eagerly and would bypass Spark's short-circuit +// semantics for compound children (see apache/datafusion-comet#5451). More complex children +// stay on the JVM codegen dispatcher via CodegenDispatchFallback. When failOnError = true +// (from `to_binary('base64')` / `try_to_binary`), Spark uses a stricter RFC 4648 validator, so +// those cases also stay on the dispatcher. Error messages match Spark byte-for-byte (pinned in +// the Rust unit tests), but the wrapping exception class does not; kept as Compatible() because +// Spark surfaces these as bare IllegalArgumentException without a SQL error class. object CometUnBase64 extends CometExpressionSerde[UnBase64] with CodegenDispatchFallback { private val failOnErrorReason = "unbase64 with failOnError = true uses stricter RFC 4648 validation that is not yet" + " implemented natively" - override def getUnsupportedReasons(): Seq[String] = Seq(failOnErrorReason) + private val nonTrivialChildReason = + "unbase64 with a non-trivial child expression uses the JVM codegen dispatcher to preserve" + + " Spark's short-circuit evaluation (native path is limited to column and literal children)" + + override def getUnsupportedReasons(): Seq[String] = + Seq(failOnErrorReason, nonTrivialChildReason) override def getSupportLevel(expr: UnBase64): SupportLevel = { - if (expr.failOnError) Unsupported(Some(failOnErrorReason)) else Compatible() + if (expr.failOnError) { + Unsupported(Some(failOnErrorReason)) + } else { + expr.child match { + case _: Attribute | _: Literal => Compatible() + case _ => Unsupported(Some(nonTrivialChildReason)) + } + } } override def convert(expr: UnBase64, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = { diff --git a/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql b/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql index e143374f087..64f13dd37a9 100644 --- a/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql +++ b/spark/src/test/resources/sql-tests/expressions/string/unbase64.sql @@ -87,3 +87,21 @@ INSERT INTO test_unbase64_roundtrip VALUES query SELECT s, cast(unbase64(base64(cast(s AS binary))) AS string) FROM test_unbase64_roundtrip + +-- Regression for apache/datafusion-comet#5451: when unbase64 has a compound child, Spark's +-- generated code preserves the child's short-circuit semantics and never evaluates branches +-- that would otherwise raise. The native ScalarFunctionExpr path evaluates its argument +-- eagerly, so compound children stay on the JVM codegen dispatcher via CodegenDispatchFallback. +-- CASE WHEN is used here because its short-circuit is guaranteed across all supported Spark +-- versions; the row (NULL, 'A') exercises the skipped branch that would otherwise raise on the +-- unpadded 1-char input. +statement +CREATE TABLE test_unbase64_short_circuit(n string, bad string) USING parquet + +statement +INSERT INTO test_unbase64_short_circuit VALUES (NULL, 'A'), ('X', 'YWJj') + +query +SELECT hex(unbase64(case when n is null then null + else cast(unbase64(bad) as string) end)) +FROM test_unbase64_short_circuit diff --git a/spark/src/test/resources/sql-tests/expressions/string/unbase64_concat_short_circuit_spark_3_5.sql b/spark/src/test/resources/sql-tests/expressions/string/unbase64_concat_short_circuit_spark_3_5.sql new file mode 100644 index 00000000000..e6d3727cd3b --- /dev/null +++ b/spark/src/test/resources/sql-tests/expressions/string/unbase64_concat_short_circuit_spark_3_5.sql @@ -0,0 +1,42 @@ +-- 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. + +-- Spark 3.5-only regression for apache/datafusion-comet#5451 (review comment). Reproduces the +-- exact scenario from the PR review: a non-foldable Parquet column that is NULL as the first +-- argument to concat, followed by a sibling expression (`unbase64('A')`) that would raise +-- `Last unit does not have enough valid bits`. Spark 3.5's generated concat short-circuits on +-- the first NULL argument and never evaluates the sibling, so the whole projection returns +-- NULL. Before the fix on this branch, recursively serializing the compound child pushed +-- `unbase64(bad)` into a native ScalarFunctionExpr that evaluates arguments eagerly, turning +-- the previously-successful query into a task failure. The fix routes non-trivial children +-- through the JVM codegen dispatcher (CodegenDispatchFallback), restoring the whole-tree +-- dispatch behavior. This fixture is pinned to Spark 3.5 because Spark 4.x's generated concat +-- does not short-circuit the same way and its own reference execution raises on this input, so +-- the same query cannot be verified as returning NULL against a Spark 4.x reference. + +-- MinSparkVersion: 3.5 +-- MaxSparkVersion: 3.5 + +statement +CREATE TABLE test_unbase64_concat_short_circuit(n string, bad string) USING parquet + +statement +INSERT INTO test_unbase64_concat_short_circuit VALUES (NULL, 'A'), ('YWJj', 'YWJj') + +query +SELECT hex(unbase64(concat(n, cast(unbase64(bad) AS string)))) +FROM test_unbase64_concat_short_circuit