You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/18 01:30:14 UTC
(datafusion-comet) 01/01: fix: decode invalid UTF-8 at the JVM to native FFI import boundary (#5310)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5310-b8e22398be5ad3efa963f73d04b8796a95d3194c
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 6bef240b89ddeac252ae20cfb11bfef2e53564e4
Author: Manu Zhang <Ow...@gmail.com>
AuthorDate: Fri Sep 18 01:29:50 2026 +0000
fix: decode invalid UTF-8 at the JVM to native FFI import boundary (#5310)
* fix: decode invalid UTF-8 at the JVM to native FFI import boundary
Co-authored-by: Manu Zhang <ow...@gmail.com>
Co-authored-by: Codex <co...@openai.com>
* bench: add end-to-end benchmark for FFI string import validation
Measures the UTF-8 validation that native code now runs on every string
column imported from the JVM, at the two import sites that carry most
string data: `ScanExec` fed by a JVM operator (Spark's Parquet reader
converted to Arrow) and native columnar-to-row conversion. Each imported
case is paired with a control that does the same work without an import,
and every Comet case checks its executed plan before it is timed. Both
ASCII and multibyte wide strings are covered, since validation takes a
slower path on non-ASCII bytes.
Co-Authored-By: Claude Opus 5 (1M context) <no...@anthropic.com>
Claude-Session: https://claude.ai/code/session_015mypwig1QwDcPvDMadVB1d
* perf: validate imported UTF-8 with simdutf8 instead of std
`decode_string_arrays` validated every imported string column with
`std::str::from_utf8`. That has a fast ASCII path but falls back to a
byte-at-a-time loop on non-ASCII text: 1.65 GB/s, or 111 ns for a
184-byte row. Measured end to end on wide multibyte strings, importing
through a JVM Parquet scan was 2.1x slower and through native
columnar-to-row 3.3x slower, against 6-11% for ASCII.
simdutf8 validates the same buffers at 11.7 GB/s (15.7 ns/row), 7x
faster on multibyte data and 3x on ASCII. It is already in the
dependency graph through arrow-json, so this adds no new third-party
crate. The `basic` validator is enough here because an invalid buffer
falls through to the element-by-element decoder below, which finds the
bad bytes itself.
Co-Authored-By: Claude Opus 5 (1M context) <no...@anthropic.com>
Claude-Session: https://claude.ai/code/session_015mypwig1QwDcPvDMadVB1d
---------
Co-authored-by: Andy Grove <ag...@apache.org>
Co-authored-by: Codex <co...@openai.com>
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
---
.../user-guide/latest/compatibility/scans.md | 10 +-
native/Cargo.lock | 1 +
native/common/Cargo.toml | 1 +
native/common/src/lib.rs | 2 +
native/common/src/utf8.rs | 559 +++++++++++++++++++++
native/core/src/execution/jni_api.rs | 4 +-
.../execution/operators/aligned_stream_reader.rs | 5 +-
native/core/src/execution/operators/scan.rs | 37 +-
native/spark-expr/Cargo.toml | 6 +-
native/spark-expr/benches/utf8_decode.rs | 40 ++
native/spark-expr/src/jvm_udf/mod.rs | 6 +-
.../benchmark/CometStringFfiImportBenchmark.scala | 226 +++++++++
12 files changed, 889 insertions(+), 8 deletions(-)
diff --git a/docs/source/user-guide/latest/compatibility/scans.md b/docs/source/user-guide/latest/compatibility/scans.md
index d5ab79b4a9..36d245992e 100644
--- a/docs/source/user-guide/latest/compatibility/scans.md
+++ b/docs/source/user-guide/latest/compatibility/scans.md
@@ -67,9 +67,13 @@ The following limitations raise an error at scan time rather than falling back t
Arrow, whose string type is strictly UTF-8. Reading a Parquet file whose `STRING` column contains
non-UTF-8 bytes fails with `Parquet error: encountered non UTF-8 data`. Disable Comet for the
query, or cast the column to `BINARY` before persisting, if you need to preserve non-UTF-8 bytes.
- Separately, non-UTF-8 bytes that reach native execution from a JVM-side columnar source are not
- currently validated at the Arrow FFI import boundary. See [#4121](https://github.com/apache/datafusion-comet/issues/4121)
- and the tracking issue [#4764](https://github.com/apache/datafusion-comet/issues/4764).
+ By contrast, Comet decodes non-UTF-8 bytes at the JVM-to-native Arrow FFI boundaries using
+ JVM-compatible replacement semantics. This covers native query input from the JVM-exported Arrow
+ stream (including JVM scans, shuffle reads, and `mapInArrow`), columnar-to-row conversion, and JVM
+ UDF results. It does not change the native Parquet reader: rejecting invalid UTF-8 there remains
+ Gap A of [#4764](https://github.com/apache/datafusion-comet/issues/4764). See
+ [Strings with non-UTF-8 bytes](index.md#strings-with-non-utf-8-bytes),
+ [#4121](https://github.com/apache/datafusion-comet/issues/4121), and the tracking issue above.
- Reading `TimestampLTZ` as `TimestampNTZ` on Spark 3.x. Spark raises an error per
[SPARK-36182](https://issues.apache.org/jira/browse/SPARK-36182) because LTZ encodes UTC-adjusted
instants that cannot be safely reinterpreted as timezone-free values, and Comet matches this by
diff --git a/native/Cargo.lock b/native/Cargo.lock
index fabb56e2ec..d4c592e1f8 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -2031,6 +2031,7 @@ dependencies = [
"datafusion",
"serde",
"serde_json",
+ "simdutf8",
"thiserror 2.0.20",
]
diff --git a/native/common/Cargo.toml b/native/common/Cargo.toml
index 5704263eb6..0e5020b895 100644
--- a/native/common/Cargo.toml
+++ b/native/common/Cargo.toml
@@ -33,6 +33,7 @@ arrow = { workspace = true }
datafusion = { workspace = true }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
+simdutf8 = "0.1.5"
thiserror = { workspace = true }
[lib]
diff --git a/native/common/src/lib.rs b/native/common/src/lib.rs
index b7662844fe..12e9ac8a5e 100644
--- a/native/common/src/lib.rs
+++ b/native/common/src/lib.rs
@@ -20,10 +20,12 @@ mod query_context;
mod schema;
pub mod struct_nulls;
pub mod tracing;
+mod utf8;
mod utils;
pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult};
pub use query_context::{create_query_context_map, QueryContext, QueryContextMap};
pub use schema::{cast_and_stamp_schema, widen_nested_nullability};
pub use struct_nulls::{child_with_parent_nulls, children_with_parent_nulls};
+pub use utf8::decode_string_arrays;
pub use utils::{bytes_to_i128, decode_utf8_spark_lossy};
diff --git a/native/common/src/utf8.rs b/native/common/src/utf8.rs
new file mode 100644
index 0000000000..b565af46c3
--- /dev/null
+++ b/native/common/src/utf8.rs
@@ -0,0 +1,559 @@
+// 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 crate::decode_utf8_spark_lossy;
+use arrow::array::{
+ downcast_dictionary_array, Array, ArrayRef, FixedSizeListArray, GenericListArray,
+ GenericStringArray, GenericStringBuilder, MapArray, OffsetSizeTrait, StructArray,
+};
+use arrow::datatypes::DataType;
+use arrow::error::ArrowError;
+use std::sync::Arc;
+
+/// Ensure every `Utf8`/`LargeUtf8` array reachable from `array` holds valid UTF-8, decoding invalid
+/// bytes the way Spark renders `StringType`. Returns the same `Arc` (zero-copy) when nothing needed
+/// decoding. Used at the JVM->native FFI import boundary, where arrow's `from_ffi` builds string
+/// arrays via `new_unchecked` and does not validate UTF-8. Unsupported string-bearing Arrow types
+/// fail closed rather than preserving unchecked string data.
+pub fn decode_string_arrays(array: &ArrayRef) -> Result<ArrayRef, ArrowError> {
+ match array.data_type() {
+ DataType::Utf8 => decode_generic_string::<i32>(array),
+ DataType::LargeUtf8 => decode_generic_string::<i64>(array),
+ DataType::Dictionary(_, value_type) => {
+ if !data_type_contains_string(value_type) {
+ return Ok(Arc::clone(array));
+ }
+ // Capture the original Arc before `downcast_dictionary_array!` shadows `array`, so the
+ // unchanged branch returns it verbatim, preserving the zero-copy contract that the
+ // Struct/List/Map arms rely on via `Arc::ptr_eq`.
+ let original = Arc::clone(array);
+ downcast_dictionary_array!(
+ array => {
+ let values = array.values();
+ let decoded = decode_string_arrays(values)?;
+ if Arc::ptr_eq(&decoded, values) {
+ Ok(original)
+ } else {
+ Ok(Arc::new(array.with_values(decoded)))
+ }
+ }
+ t => unreachable!("dictionary key type validated by Arrow: {t}"),
+ )
+ }
+ DataType::Struct(fields) => {
+ let s = array
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .expect("data type checked by caller");
+ let mut changed = false;
+ let mut columns = Vec::with_capacity(s.num_columns());
+ for col in s.columns() {
+ let decoded = decode_string_arrays(col)?;
+ changed |= !Arc::ptr_eq(&decoded, col);
+ columns.push(decoded);
+ }
+ if !changed {
+ return Ok(Arc::clone(array));
+ }
+ Ok(Arc::new(StructArray::new(
+ fields.clone(),
+ columns,
+ s.nulls().cloned(),
+ )))
+ }
+ DataType::List(_) => decode_list::<i32>(array),
+ DataType::LargeList(_) => decode_list::<i64>(array),
+ DataType::FixedSizeList(field, size) => {
+ let list = array
+ .as_any()
+ .downcast_ref::<FixedSizeListArray>()
+ .expect("data type checked by caller");
+ let values = list.values();
+ let decoded = decode_string_arrays(values)?;
+ if Arc::ptr_eq(&decoded, values) {
+ return Ok(Arc::clone(array));
+ }
+ Ok(Arc::new(FixedSizeListArray::try_new(
+ Arc::clone(field),
+ *size,
+ decoded,
+ list.nulls().cloned(),
+ )?))
+ }
+ DataType::Map(field, ordered) => {
+ let map = array
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .expect("data type checked by caller");
+ let entries = map.entries();
+ let mut changed = false;
+ let mut columns = Vec::with_capacity(entries.num_columns());
+ for col in entries.columns() {
+ let decoded = decode_string_arrays(col)?;
+ changed |= !Arc::ptr_eq(&decoded, col);
+ columns.push(decoded);
+ }
+ if !changed {
+ return Ok(Arc::clone(array));
+ }
+ let decoded_entries =
+ StructArray::new(entries.fields().clone(), columns, entries.nulls().cloned());
+ Ok(Arc::new(MapArray::try_new(
+ Arc::clone(field),
+ map.offsets().clone(),
+ decoded_entries,
+ map.nulls().cloned(),
+ *ordered,
+ )?))
+ }
+ // Spark's current JVM producers emit StringType as Utf8/LargeUtf8 and the supported
+ // containers above. Fail closed if a different string-bearing representation reaches this
+ // boundary: passing it through would preserve the invalid UTF-8 this function exists to
+ // remove. Keep `data_type_contains_string` exhaustive so an Arrow upgrade adding another
+ // data type forces this safety audit to be revisited.
+ data_type => {
+ if data_type_contains_string(data_type) {
+ Err(unsupported_string_type(data_type))
+ } else {
+ Ok(Arc::clone(array))
+ }
+ }
+ }
+}
+
+fn unsupported_string_type(data_type: &DataType) -> ArrowError {
+ ArrowError::NotYetImplemented(format!(
+ "decoding FFI-imported string data in {data_type} is not supported"
+ ))
+}
+
+/// Whether `data_type` is, or can contain, a logical string representation.
+///
+/// This match is deliberately exhaustive: when Arrow adds a data type, the compiler should force
+/// us to decide whether it can carry strings before `decode_string_arrays` passes it through.
+fn data_type_contains_string(data_type: &DataType) -> bool {
+ match data_type {
+ DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => true,
+ DataType::List(field)
+ | DataType::ListView(field)
+ | DataType::FixedSizeList(field, _)
+ | DataType::LargeList(field)
+ | DataType::LargeListView(field) => data_type_contains_string(field.data_type()),
+ DataType::Struct(fields) => fields
+ .iter()
+ .any(|field| data_type_contains_string(field.data_type())),
+ DataType::Union(fields, _) => fields
+ .iter()
+ .any(|(_, field)| data_type_contains_string(field.data_type())),
+ DataType::Dictionary(_, value_type) => data_type_contains_string(value_type),
+ DataType::Map(field, _) => data_type_contains_string(field.data_type()),
+ DataType::RunEndEncoded(_, values) => data_type_contains_string(values.data_type()),
+ DataType::Null
+ | DataType::Boolean
+ | DataType::Int8
+ | DataType::Int16
+ | DataType::Int32
+ | DataType::Int64
+ | DataType::UInt8
+ | DataType::UInt16
+ | DataType::UInt32
+ | DataType::UInt64
+ | DataType::Float16
+ | DataType::Float32
+ | DataType::Float64
+ | DataType::Timestamp(_, _)
+ | DataType::Date32
+ | DataType::Date64
+ | DataType::Time32(_)
+ | DataType::Time64(_)
+ | DataType::Duration(_)
+ | DataType::Interval(_)
+ | DataType::Binary
+ | DataType::FixedSizeBinary(_)
+ | DataType::LargeBinary
+ | DataType::BinaryView
+ | DataType::Decimal32(_, _)
+ | DataType::Decimal64(_, _)
+ | DataType::Decimal128(_, _)
+ | DataType::Decimal256(_, _) => false,
+ }
+}
+
+fn decode_generic_string<O: OffsetSizeTrait>(array: &ArrayRef) -> Result<ArrayRef, ArrowError> {
+ let arr = array
+ .as_any()
+ .downcast_ref::<GenericStringArray<O>>()
+ .expect("data type checked by caller");
+ let len = arr.len();
+ if len == 0 {
+ return Ok(Arc::clone(array));
+ }
+ let values: &[u8] = arr.value_data();
+ let offsets = arr.value_offsets(); // &[O], length == len + 1
+ let start = offsets[0].as_usize();
+ let end = offsets[len].as_usize();
+
+ // Fast path: the used byte range parses as UTF-8 AND no element boundary splits a codepoint.
+ // Both are required: a whole-buffer-valid "é" (C3 A9) with per-element offsets [0,1,2] yields
+ // element slices `C3` and `A9`, each invalid, and `value()` would decode them unchecked (UB).
+ //
+ // Every imported string column pays this pass, so it uses simdutf8 rather than
+ // `std::str::from_utf8`: std is fast on ASCII but falls back to a byte-at-a-time loop on
+ // non-ASCII text, which made multibyte columns several times slower to import. The `basic`
+ // validator is enough because an invalid buffer takes the slow path below, which finds the
+ // bad bytes itself.
+ if simdutf8::basic::from_utf8(&values[start..end]).is_ok() {
+ let mut boundaries_ok = true;
+ for off in &offsets[1..len] {
+ let o = off.as_usize();
+ // A boundary landing on a UTF-8 continuation byte (0b10xx_xxxx) splits a codepoint.
+ if o < values.len() && (values[o] & 0xC0) == 0x80 {
+ boundaries_ok = false;
+ break;
+ }
+ }
+ if boundaries_ok {
+ return Ok(Arc::clone(array));
+ }
+ }
+
+ // Slow path: rebuild element-by-element via the Spark-lossy decoder. We slice the raw values
+ // buffer directly rather than calling `arr.value(i)`, which uses `from_utf8_unchecked`.
+ let mut builder = GenericStringBuilder::<O>::with_capacity(len, end - start);
+ for i in 0..len {
+ if arr.is_null(i) {
+ builder.append_null();
+ } else {
+ let s = offsets[i].as_usize();
+ let e = offsets[i + 1].as_usize();
+ builder.append_value(decode_utf8_spark_lossy(&values[s..e]));
+ }
+ }
+ Ok(Arc::new(builder.finish()))
+}
+
+fn decode_list<O: OffsetSizeTrait>(array: &ArrayRef) -> Result<ArrayRef, ArrowError> {
+ let list = array
+ .as_any()
+ .downcast_ref::<GenericListArray<O>>()
+ .expect("data type checked by caller");
+ let values = list.values();
+ let decoded = decode_string_arrays(values)?;
+ if Arc::ptr_eq(&decoded, values) {
+ return Ok(Arc::clone(array));
+ }
+ let field = match array.data_type() {
+ DataType::List(f) | DataType::LargeList(f) => Arc::clone(f),
+ _ => unreachable!("decode_list called on non-list"),
+ };
+ Ok(Arc::new(GenericListArray::<O>::try_new(
+ field,
+ list.offsets().clone(),
+ decoded,
+ list.nulls().cloned(),
+ )?))
+}
+
+#[cfg(test)]
+mod walker_tests {
+ use super::{data_type_contains_string, decode_string_arrays};
+ use arrow::array::{
+ make_array, Array, ArrayData, ArrayRef, LargeStringArray, StringArray, StringViewArray,
+ };
+ use arrow::buffer::Buffer;
+ use arrow::datatypes::DataType;
+ use arrow::error::ArrowError;
+ use std::sync::Arc;
+
+ /// Build a (possibly invalid) Utf8 array from raw offsets + value bytes, the way an FFI import
+ /// would deliver it (no validation). `build_unchecked` mirrors arrow's `from_ffi`.
+ fn utf8_unchecked(offsets: &[i32], values: &[u8], len: usize) -> ArrayRef {
+ let data = unsafe {
+ ArrayData::builder(DataType::Utf8)
+ .len(len)
+ .add_buffer(Buffer::from_slice_ref(offsets))
+ .add_buffer(Buffer::from(values.to_vec()))
+ .build_unchecked()
+ };
+ make_array(data)
+ }
+
+ #[test]
+ fn valid_utf8_is_zero_copy() {
+ let input: ArrayRef = Arc::new(StringArray::from(vec!["a", "é", "🦀"]));
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "valid input must be returned unchanged"
+ );
+ }
+
+ #[test]
+ fn unsupported_string_representation_fails_closed() {
+ let input: ArrayRef = Arc::new(StringViewArray::from(vec!["valid"]));
+ let err = decode_string_arrays(&input).unwrap_err();
+ assert!(matches!(
+ err,
+ ArrowError::NotYetImplemented(message) if message.contains("Utf8View")
+ ));
+ }
+
+ #[test]
+ fn invalid_bytes_decode_to_replacement() {
+ // element 0 = [0xFF, 0x41] -> "\u{FFFD}A", element 1 = [0x42] -> "B"
+ let input = utf8_unchecked(&[0, 2, 3], &[0xFF, 0x41, 0x42], 2);
+ let out = decode_string_arrays(&input).unwrap();
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(s.value(0), "\u{FFFD}A");
+ assert_eq!(s.value(1), "B");
+ }
+
+ #[test]
+ fn split_codepoint_boundary_is_rebuilt() {
+ // whole buffer "é" (C3 A9) is valid UTF-8, but offsets split it into two invalid slices.
+ let input = utf8_unchecked(&[0, 1, 2], &[0xC3, 0xA9], 2);
+ let out = decode_string_arrays(&input).unwrap();
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(s.value(0), "\u{FFFD}");
+ assert_eq!(s.value(1), "\u{FFFD}");
+ }
+
+ #[test]
+ fn nulls_are_preserved() {
+ let data = unsafe {
+ ArrayData::builder(DataType::Utf8)
+ .len(3)
+ .null_count(1)
+ .null_bit_buffer(Some(Buffer::from(vec![0b0000_0101_u8])))
+ .add_buffer(Buffer::from_slice_ref([0_i32, 1, 2, 3]))
+ .add_buffer(Buffer::from(vec![0xff, 0xfe, b'b']))
+ .build_unchecked()
+ };
+ let input = make_array(data);
+ let out = decode_string_arrays(&input).unwrap();
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+
+ assert_eq!(s.value(0), "\u{FFFD}");
+ assert!(s.is_null(1));
+ assert_eq!(s.value(2), "b");
+ }
+
+ #[test]
+ fn large_utf8_invalid_decodes() {
+ let data = unsafe {
+ ArrayData::builder(DataType::LargeUtf8)
+ .len(1)
+ .add_buffer(Buffer::from_slice_ref([0i64, 1]))
+ .add_buffer(Buffer::from(vec![0xFFu8]))
+ .build_unchecked()
+ };
+ let out = decode_string_arrays(&make_array(data)).unwrap();
+ let s = out.as_any().downcast_ref::<LargeStringArray>().unwrap();
+ assert_eq!(s.value(0), "\u{FFFD}");
+ }
+
+ use arrow::array::{
+ DictionaryArray, FixedSizeListArray, Int32Array, ListArray, MapArray, StructArray,
+ };
+ use arrow::datatypes::{Field, Fields, Int32Type};
+
+ #[test]
+ fn detects_unhandled_nested_string_representations() {
+ let string_field = Arc::new(Field::new("item", DataType::Utf8, true));
+ let list_view = DataType::ListView(Arc::clone(&string_field));
+ let large_list_view = DataType::LargeListView(Arc::clone(&string_field));
+ assert!(data_type_contains_string(&list_view));
+ assert!(data_type_contains_string(&large_list_view));
+
+ let run_ends = Arc::new(Field::new("run_ends", DataType::Int32, false));
+ let run_end_encoded = DataType::RunEndEncoded(run_ends, string_field);
+ assert!(data_type_contains_string(&run_end_encoded));
+
+ let int_field = Arc::new(Field::new("item", DataType::Int32, true));
+ assert!(!data_type_contains_string(&DataType::ListView(int_field)));
+ }
+
+ /// An invalid Utf8 leaf ["\u{FFFD}"] built from raw bytes.
+ fn invalid_leaf() -> ArrayRef {
+ utf8_unchecked(&[0, 1], &[0xFF], 1)
+ }
+
+ #[test]
+ fn dictionary_values_are_decoded() {
+ let values = invalid_leaf();
+ let keys = Int32Array::from(vec![0, 0]);
+ let dict: ArrayRef = Arc::new(DictionaryArray::<Int32Type>::new(keys, values));
+ let out = decode_string_arrays(&dict).unwrap();
+ let d = out
+ .as_any()
+ .downcast_ref::<DictionaryArray<Int32Type>>()
+ .unwrap();
+ let vals = d.values().as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(vals.value(0), "\u{FFFD}");
+ }
+
+ #[test]
+ fn struct_field_is_decoded() {
+ let field = Arc::new(Field::new("s", DataType::Utf8, true));
+ let input: ArrayRef = Arc::new(StructArray::new(
+ Fields::from(vec![field]),
+ vec![invalid_leaf()],
+ None,
+ ));
+ let out = decode_string_arrays(&input).unwrap();
+ let s = out.as_any().downcast_ref::<StructArray>().unwrap();
+ let col = s.column(0).as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(col.value(0), "\u{FFFD}");
+ }
+
+ #[test]
+ fn list_values_are_decoded() {
+ let field = Arc::new(Field::new("item", DataType::Utf8, true));
+ let offsets = arrow::buffer::OffsetBuffer::new(vec![0i32, 1].into());
+ let input: ArrayRef =
+ Arc::new(ListArray::try_new(field, offsets, invalid_leaf(), None).unwrap());
+ let out = decode_string_arrays(&input).unwrap();
+ let l = out.as_any().downcast_ref::<ListArray>().unwrap();
+ let vals = l.values().as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(vals.value(0), "\u{FFFD}");
+ }
+
+ #[test]
+ fn valid_struct_is_zero_copy() {
+ let field = Arc::new(Field::new("s", DataType::Utf8, true));
+ let leaf: ArrayRef = Arc::new(StringArray::from(vec!["ok"]));
+ let input: ArrayRef = Arc::new(StructArray::new(
+ Fields::from(vec![field]),
+ vec![leaf],
+ None,
+ ));
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "all-valid nested input must be unchanged"
+ );
+ }
+
+ #[test]
+ fn dictionary_valid_values_are_zero_copy() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
+ let keys = Int32Array::from(vec![0, 1, 0]);
+ let input: ArrayRef = Arc::new(DictionaryArray::<Int32Type>::new(keys, values));
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "dictionary with all-valid values must be returned as the original Arc"
+ );
+ }
+
+ #[test]
+ fn fixed_size_list_values_are_decoded() {
+ let field = Arc::new(Field::new("item", DataType::Utf8, true));
+ let input: ArrayRef =
+ Arc::new(FixedSizeListArray::try_new(field, 1, invalid_leaf(), None).unwrap());
+ let out = decode_string_arrays(&input).unwrap();
+ let l = out.as_any().downcast_ref::<FixedSizeListArray>().unwrap();
+ let vals = l.values().as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(vals.value(0), "\u{FFFD}");
+ }
+
+ #[test]
+ fn fixed_size_list_valid_is_zero_copy() {
+ let field = Arc::new(Field::new("item", DataType::Utf8, true));
+ let leaf: ArrayRef = Arc::new(StringArray::from(vec!["ok"]));
+ let input: ArrayRef = Arc::new(FixedSizeListArray::try_new(field, 1, leaf, None).unwrap());
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "fixed-size list with all-valid values must be unchanged"
+ );
+ }
+
+ /// Build a Map array whose single entry maps `key` -> `values` (a Utf8 leaf array of length 1).
+ fn build_map(key: &str, values: ArrayRef) -> ArrayRef {
+ let entries_fields = Fields::from(vec![
+ Arc::new(Field::new("keys", DataType::Utf8, false)),
+ Arc::new(Field::new("values", DataType::Utf8, true)),
+ ]);
+ let keys: ArrayRef = Arc::new(StringArray::from(vec![key]));
+ let entries = StructArray::new(entries_fields.clone(), vec![keys, values], None);
+ let map_field = Arc::new(Field::new(
+ "entries",
+ DataType::Struct(entries_fields),
+ false,
+ ));
+ let offsets = arrow::buffer::OffsetBuffer::new(vec![0i32, 1].into());
+ Arc::new(MapArray::try_new(map_field, offsets, entries, None, false).unwrap())
+ }
+
+ #[test]
+ fn map_values_are_decoded() {
+ let input = build_map("k", invalid_leaf());
+ let out = decode_string_arrays(&input).unwrap();
+ let m = out.as_any().downcast_ref::<MapArray>().unwrap();
+ let values = m
+ .entries()
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(values.value(0), "\u{FFFD}");
+ }
+
+ #[test]
+ fn map_valid_is_zero_copy() {
+ let values: ArrayRef = Arc::new(StringArray::from(vec!["ok"]));
+ let input = build_map("k", values);
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "map with all-valid values must be unchanged"
+ );
+ }
+
+ #[test]
+ fn trailing_empty_string_offset_is_handled() {
+ // element 0 = "ab", element 1 = "" (its offset lands exactly at values.len()).
+ let input = utf8_unchecked(&[0, 2, 2], b"ab", 2);
+ let out = decode_string_arrays(&input).unwrap();
+ assert!(
+ Arc::ptr_eq(&input, &out),
+ "valid input with a trailing empty element must be returned unchanged"
+ );
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(s.value(0), "ab");
+ assert_eq!(s.value(1), "");
+ }
+
+ #[test]
+ fn sliced_array_is_decoded() {
+ // Slicing shifts offsets[0] away from 0; the fast path must still handle this correctly.
+ let base: ArrayRef = Arc::new(StringArray::from(vec!["aa", "bb", "cc"]));
+ let sliced: ArrayRef = base.slice(1, 2);
+ let out = decode_string_arrays(&sliced).unwrap();
+ // Slicing already produces a new Arc distinct from `base`, so ptr_eq against `base` isn't
+ // meaningful here; what matters is that the fast path recognizes the slice as valid and
+ // returns it unchanged (same Arc as `sliced`) rather than rebuilding it.
+ assert!(
+ Arc::ptr_eq(&sliced, &out),
+ "valid sliced input must be returned unchanged"
+ );
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(s.value(0), "bb");
+ assert_eq!(s.value(1), "cc");
+ }
+}
diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index 6eb58f07ee..89262c5bc0 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -41,6 +41,7 @@ use datafusion::{
physical_plan::{display::DisplayableExecutionPlan, SendableRecordBatchStream},
prelude::{SessionConfig, SessionContext},
};
+use datafusion_comet_common::decode_string_arrays;
use datafusion_comet_proto::spark_expression::agg_expr::ExprStruct as AggExprStruct;
use datafusion_comet_proto::spark_operator::{AggregateMode, Operator, ShuffleScan};
use datafusion_comet_spark_expr::url_funcs::{CometParseUrl, CometTryParseUrl};
@@ -1629,7 +1630,8 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_columnarToRowConvert(
let array_data = from_ffi(ffi_array, &ffi_schema)
.map_err(|e| CometError::Internal(format!("Failed to import array: {}", e)))?;
- arrays.push(arrow::array::make_array(array_data));
+ let imported = arrow::array::make_array(array_data);
+ arrays.push(decode_string_arrays(&imported)?);
}
// Convert columnar to row
diff --git a/native/core/src/execution/operators/aligned_stream_reader.rs b/native/core/src/execution/operators/aligned_stream_reader.rs
index 598c2c6819..3f906be4c7 100644
--- a/native/core/src/execution/operators/aligned_stream_reader.rs
+++ b/native/core/src/execution/operators/aligned_stream_reader.rs
@@ -91,7 +91,10 @@ impl Iterator for AlignedArrowStreamReader {
/// JVM producer's 8-byte-aligned `Decimal128` buffer reaches us under-aligned and would panic in
/// `ScalarBuffer::<i128>::from` once `StructArray::from` materializes the typed column. See
/// apache/arrow-rs#10028; the arrow 59 fix (apache/arrow-rs#10030) folds this `align_buffers` call
-/// into `from_ffi_and_data_type` itself.
+/// into `from_ffi_and_data_type` itself. This reader is currently consumed only by `ScanExec`;
+/// after this function materializes the batch, `ScanExec::pull_next` passes every column through
+/// `import_column`, which validates and decodes string data before native execution can read it.
+/// Any future consumer must preserve that validation step.
fn batch_from_ffi(array: FFI_ArrowArray, schema: &SchemaRef) -> Result<RecordBatch, ArrowError> {
let dt = DataType::Struct(schema.fields().clone());
// SAFETY: the caller transfers ownership of a valid FFI array whose layout matches `schema`.
diff --git a/native/core/src/execution/operators/scan.rs b/native/core/src/execution/operators/scan.rs
index f05a703dab..b1bf199068 100644
--- a/native/core/src/execution/operators/scan.rs
+++ b/native/core/src/execution/operators/scan.rs
@@ -31,6 +31,7 @@ use datafusion::{
physical_expr::*,
physical_plan::{ExecutionPlan, *},
};
+use datafusion_comet_common::decode_string_arrays;
use futures::Stream;
use itertools::Itertools;
use std::{
@@ -157,7 +158,7 @@ impl ScanExec {
let columns = record_batch.columns();
let mut inputs: Vec<ArrayRef> = Vec::with_capacity(columns.len());
for col in columns {
- inputs.push(copy_or_unpack_array(col, &CopyMode::UnpackOrClone)?);
+ inputs.push(import_column(col)?);
}
Ok(InputBatch::new(inputs, Some(num_rows)))
}
@@ -165,6 +166,15 @@ impl ScanExec {
}
}
+/// Transform one FFI-imported column for native execution: first decode any invalid UTF-8 to the
+/// Spark-rendered form (arrow's `from_ffi` imports string buffers unchecked), then copy/unpack.
+/// Decoding runs before unpacking so a `Dictionary(_, Utf8)` decodes its compact values, not the
+/// expanded ones.
+fn import_column(col: &ArrayRef) -> Result<ArrayRef, CometError> {
+ let decoded = decode_string_arrays(col)?;
+ Ok(copy_or_unpack_array(&decoded, &CopyMode::UnpackOrClone)?)
+}
+
fn schema_from_data_types(data_types: &[DataType]) -> SchemaRef {
let fields = data_types
.iter()
@@ -380,3 +390,28 @@ impl InputBatch {
InputBatch::Batch(columns, num_rows)
}
}
+
+#[cfg(test)]
+mod import_tests {
+ use super::*;
+ use arrow::array::{make_array, Array, ArrayData, ArrayRef, StringArray};
+ use arrow::buffer::Buffer;
+ use arrow::datatypes::DataType;
+
+ #[test]
+ fn import_decodes_invalid_utf8_column() {
+ // An FFI-imported Utf8 column with invalid bytes [0xFF] -> must be decoded, not passed
+ // through, so downstream `value()` is sound.
+ let data = unsafe {
+ ArrayData::builder(DataType::Utf8)
+ .len(1)
+ .add_buffer(Buffer::from_slice_ref([0i32, 1]))
+ .add_buffer(Buffer::from(vec![0xFFu8]))
+ .build_unchecked()
+ };
+ let col: ArrayRef = make_array(data);
+ let out = import_column(&col).unwrap();
+ let s = out.as_any().downcast_ref::<StringArray>().unwrap();
+ assert_eq!(s.value(0), "\u{FFFD}");
+ }
+}
diff --git a/native/spark-expr/Cargo.toml b/native/spark-expr/Cargo.toml
index 5905bcea11..15d5c4b743 100644
--- a/native/spark-expr/Cargo.toml
+++ b/native/spark-expr/Cargo.toml
@@ -425,4 +425,8 @@ harness = false
[[bench]]
name = "json_array_length"
-harness = false
\ No newline at end of file
+harness = false
+
+[[bench]]
+name = "utf8_decode"
+harness = false
diff --git a/native/spark-expr/benches/utf8_decode.rs b/native/spark-expr/benches/utf8_decode.rs
new file mode 100644
index 0000000000..e4026abeff
--- /dev/null
+++ b/native/spark-expr/benches/utf8_decode.rs
@@ -0,0 +1,40 @@
+// 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::{ArrayRef, StringArray};
+use criterion::{criterion_group, criterion_main, Criterion};
+use datafusion_comet_common::decode_string_arrays;
+use std::hint::black_box;
+use std::sync::Arc;
+
+fn bench_valid_fast_path(c: &mut Criterion) {
+ // 8192-row batch of valid mixed ASCII/multibyte strings, exercising the fast path (from_utf8
+ // plus boundary check, zero-copy return) that every scanned string column pays.
+ let values: Vec<String> = (0..8192).map(|i| format!("row-{i}-café-日本語")).collect();
+ let refs: Vec<&str> = values.iter().map(|s| s.as_str()).collect();
+ let input: ArrayRef = Arc::new(StringArray::from(refs));
+
+ c.bench_function("decode_string_arrays/valid_8192", |b| {
+ b.iter(|| {
+ let out = black_box(decode_string_arrays(black_box(&input)).unwrap());
+ assert!(Arc::ptr_eq(&input, &out));
+ })
+ });
+}
+
+criterion_group!(benches, bench_valid_fast_path);
+criterion_main!(benches);
diff --git a/native/spark-expr/src/jvm_udf/mod.rs b/native/spark-expr/src/jvm_udf/mod.rs
index 8e148ef40a..75e75351fa 100644
--- a/native/spark-expr/src/jvm_udf/mod.rs
+++ b/native/spark-expr/src/jvm_udf/mod.rs
@@ -28,6 +28,7 @@ use datafusion::common::Result as DFResult;
use datafusion::logical_expr::ColumnarValue;
use datafusion::physical_expr::PhysicalExpr;
+use datafusion_comet_common::decode_string_arrays;
use datafusion_comet_jni_bridge::errors::{CometError, ExecutionError};
use datafusion_comet_jni_bridge::JVMClasses;
use jni::objects::{Global, JObject, JValue};
@@ -249,7 +250,10 @@ impl PhysicalExpr for JvmScalarUdfExpr {
// exactly once when the Box drops at end of scope.
let result_data = unsafe { from_ffi(*out_array, &out_schema) }
.map_err(|e| CometError::Arrow { source: e })?;
- Ok(ColumnarValue::Array(make_array(result_data)))
+ let imported = make_array(result_data);
+ let decoded =
+ decode_string_arrays(&imported).map_err(|e| CometError::Arrow { source: e })?;
+ Ok(ColumnarValue::Array(decoded))
}
fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometStringFfiImportBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometStringFfiImportBenchmark.scala
new file mode 100644
index 0000000000..284438b40b
--- /dev/null
+++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometStringFfiImportBenchmark.scala
@@ -0,0 +1,226 @@
+/*
+ * 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 scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.SparkConf
+import org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.comet.{CometColumnarToRowExec, CometNativeColumnarToRowExec, CometNativeScanExec, CometSparkToColumnarExec}
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.util.QueryExecutionListener
+
+import org.apache.comet.{CometConf, CometSparkSessionExtensions}
+
+/**
+ * Benchmark to measure the end-to-end cost of validating string data that native code imports
+ * from the JVM over the Arrow C Data Interface.
+ *
+ * arrow-rs builds FFI-imported string arrays without checking UTF-8, so Comet validates every
+ * imported `Utf8` column and decodes it if it is invalid. The data here is always valid, so the
+ * cases measure the validation pass that every query crossing an import site pays. Two import
+ * sites carry most string data:
+ * - `ScanExec`, which receives batches from JVM operators. Here it is fed by Spark's Parquet
+ * reader converted to Arrow.
+ * - native columnar-to-row conversion, which imports native batches back from the JVM.
+ *
+ * Each case that crosses an import site is paired with a control that does the same work without
+ * one: the native Parquet scan for `ScanExec`, and the JVM conversion for columnar-to-row. When
+ * comparing a build with and without validation, the controls should not move.
+ *
+ * To run this benchmark:
+ * {{{
+ * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometStringFfiImportBenchmark
+ * }}}
+ *
+ * Results will be written to "spark/benchmarks/CometStringFfiImportBenchmark-**results.txt".
+ */
+object CometStringFfiImportBenchmark extends CometBenchmarkBase {
+
+ override def getSparkSession: SparkSession = {
+ val conf = new SparkConf()
+ .setAppName("CometStringFfiImportBenchmark")
+ .set("spark.master", "local[1]")
+ .setIfMissing("spark.driver.memory", "3g")
+ .setIfMissing("spark.executor.memory", "3g")
+ .set(
+ "spark.shuffle.manager",
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+ .set("spark.memory.offHeap.enabled", "true")
+ .set("spark.memory.offHeap.size", "2g")
+
+ val sparkSession = SparkSession
+ .builder()
+ .config(conf)
+ .withExtensions(new CometSparkSessionExtensions)
+ .getOrCreate()
+
+ sparkSession.conf.set(SQLConf.ANSI_ENABLED.key, "false")
+ sparkSession.conf.set(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key, "true")
+ sparkSession.conf.set(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key, "true")
+ sparkSession.conf.set(CometConf.COMET_ENABLED.key, "false")
+ sparkSession.conf.set(CometConf.COMET_EXEC_ENABLED.key, "false")
+ // Every value is distinct, so dictionary pages would fall back anyway. Disabling them keeps
+ // both import sites on plain `Utf8` arrays rather than depending on where the fallback lands.
+ sparkSession.conf.set("parquet.enable.dictionary", "false")
+
+ sparkSession
+ }
+
+ private val cometConf =
+ Seq(CometConf.COMET_ENABLED.key -> "true", CometConf.COMET_EXEC_ENABLED.key -> "true")
+
+ /**
+ * Adds a case after checking that the query really takes the path the case name claims. The
+ * noop write is what the case times, and its plan can differ from the SELECT's, so the check
+ * inspects the plans the write executed.
+ */
+ private def addVerifiedCase(
+ benchmark: Benchmark,
+ name: String,
+ query: String,
+ conf: Seq[(String, String)],
+ required: Class[_ <: SparkPlan],
+ excluded: Class[_ <: SparkPlan]): Unit = {
+ withSQLConf(conf: _*) {
+ val plans = ArrayBuffer.empty[SparkPlan]
+ val listener = new QueryExecutionListener {
+ override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit = {
+ plans += qe.executedPlan
+ }
+
+ override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit =
+ ()
+ }
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ spark.listenerManager.register(listener)
+ try {
+ spark.sql(query).noop()
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+ val nodes = plans.flatMap(plan => collect(plan) { case node => node.getClass })
+ require(
+ nodes.contains(required) && !nodes.contains(excluded),
+ s"$name must execute ${required.getSimpleName} and not ${excluded.getSimpleName}.\n" +
+ plans.mkString("\n"))
+ benchmark.out.println(s"Verified $name")
+ }
+
+ benchmark.addCase(name) { _ =>
+ withSQLConf(conf: _*) {
+ spark.sql(query).noop()
+ }
+ }
+ }
+
+ /** A filter over a wide string column, with the column reaching native code from a scan. */
+ def scanImportBenchmark(values: Int, kind: String): Unit = {
+ val benchmark =
+ new Benchmark(s"Filter over wide $kind strings", values, output = output)
+ val query = "SELECT count(*) FROM parquetV1Table WHERE s LIKE '%999%'"
+
+ benchmark.addCase("Spark") { _ =>
+ spark.sql(query).noop()
+ }
+
+ addVerifiedCase(
+ benchmark,
+ "Comet, native Parquet scan (no import)",
+ query,
+ cometConf :+ (CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true"),
+ required = classOf[CometNativeScanExec],
+ excluded = classOf[CometSparkToColumnarExec])
+
+ addVerifiedCase(
+ benchmark,
+ "Comet, Spark Parquet scan converted to Arrow (imported)",
+ query,
+ cometConf ++ Seq(
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true"),
+ required = classOf[CometSparkToColumnarExec],
+ excluded = classOf[CometNativeScanExec])
+
+ benchmark.run()
+ }
+
+ /** Wide string columns returned to Spark as rows. */
+ def columnarToRowImportBenchmark(values: Int, kind: String): Unit = {
+ val benchmark =
+ new Benchmark(s"Columnar to row of wide $kind strings", values, output = output)
+ val query = "SELECT s FROM parquetV1Table"
+ val nativeScan = cometConf :+ (CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "true")
+
+ benchmark.addCase("Spark") { _ =>
+ spark.sql(query).noop()
+ }
+
+ addVerifiedCase(
+ benchmark,
+ "Comet, JVM columnar to row (no import)",
+ query,
+ nativeScan :+ (CometConf.COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED.key -> "false"),
+ required = classOf[CometColumnarToRowExec],
+ excluded = classOf[CometNativeColumnarToRowExec])
+
+ addVerifiedCase(
+ benchmark,
+ "Comet, native columnar to row (imported)",
+ query,
+ nativeScan :+ (CometConf.COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED.key -> "true"),
+ required = classOf[CometNativeColumnarToRowExec],
+ excluded = classOf[CometColumnarToRowExec])
+
+ benchmark.run()
+ }
+
+ override def runCometBenchmark(mainArgs: Array[String]): Unit = {
+ val values = 1024 * 1024 // 1M rows
+
+ // Roughly 200 bytes per value. The multibyte strings mix two- and three-byte UTF-8
+ // sequences, which take the slower per-character path through validation. They are built
+ // with `chr` and `unhex` so that this source file stays ASCII.
+ val strings = Seq(
+ "ASCII" -> "repeat(concat('value_', cast(id as string), '_'), 12)",
+ "multibyte" ->
+ ("repeat(concat('caf', chr(233), '_', decode(unhex('E697A5E69CACE8AA9E'), 'UTF-8'), " +
+ "'_', cast(id as string), '_'), 8)"))
+
+ strings.foreach { case (kind, expr) =>
+ withTempPath { dir =>
+ withTempTable("parquetV1Table") {
+ prepareTable(dir, spark.range(values).selectExpr(s"$expr AS s"))
+
+ runBenchmark(s"FFI String Import - Scan ($kind)") {
+ scanImportBenchmark(values, kind)
+ }
+
+ runBenchmark(s"FFI String Import - Columnar to Row ($kind)") {
+ columnarToRowImportBenchmark(values, kind)
+ }
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org