[GLUTEN-13010][VL] Support Arrow StringView for Velox batch loading

[WangGuangxin]

What changes are proposed in this pull request?

Previously, Velox exported VARCHAR and VARBINARY columns using Arrow’s traditional Utf8 and Binary layouts. These layouts store all values in one contiguous data buffer and use signed 32-bit offsets. Consequently, a column fails to export when its aggregate payload exceeds approximately 2 GiB, even if every individual value is small.

This PR adds an optional Arrow StringView export path for Velox-to-Java conversion. Utf8View and BinaryView use fixed-size descriptors that can reference multiple variadic data buffers, removing the requirement that an entire column fit in one contiguous buffer smaller than 2 GiB.

The change includes:

Exporting Velox strings as Utf8View/BinaryView through ColumnarBatches.load when supported.
Adding Java accessors for ViewVarCharVector and ViewVarBinaryVector.
Preserving View types during Arrow schema conversion.
Preserving variadic-buffer metadata during load → offload → load round trips.
Falling back to traditional Utf8/Binary when the Arrow runtime does not support StringView.
Keeping shuffle, serialization, and writer paths unchanged to limit compatibility impact.
This removes the approximately 2 GiB aggregate payload limit for a string or binary column in this conversion path. It does not support an individual value larger than approximately 2 GiB, and other paths that still use traditional Utf8/Binary retain the original limit.

Why did the previous path have a 2 GiB limit?

Traditional Arrow Utf8 and Binary arrays store all values in one contiguous data buffer and use signed 32-bit offsets:

validity buffer
offset buffer: [0, len0, len0 + len1, ...]
data buffer: all values stored contiguously
The last offset represents the column’s total payload size and cannot exceed Integer.MAX_VALUE (2,147,483,647). Therefore, conversion can fail when the aggregate payload of one column exceeds approximately 2 GiB, even if every individual value is small.

How does the new path bypass this limit?

Arrow Utf8View and BinaryView use fixed-size 16-byte descriptors. Short values can be stored inline, while longer values reference one of multiple variadic data buffers using a buffer index and offset.

This removes the requirement that all values in a column be stored in one contiguous buffer smaller than 2 GiB. The aggregate payload can therefore exceed 2 GiB by being distributed across multiple backing buffers.

What cases are still not addressed?

A single string or binary value larger than approximately 2 GiB is still unsupported because its length is represented using a 32-bit integer, and Java arrays and Spark UTF8String have similar limits.
Paths that still use traditional Utf8/Binary, such as shuffle, serialization, and writer paths, retain the original aggregate 2 GiB limit.
Environments without Arrow StringView support, such as the default Arrow 15/JDK 8 configuration, fall back to traditional Utf8/Binary.
Normal memory and allocator limits still apply. StringView avoids the contiguous-buffer offset limit; it does not provide unlimited capacity.

How was this patch tested?

Added tests covering:

Velox export using Utf8View and BinaryView.
Java access through ArrowWritableColumnVector and ArrowColumnVector.
Null, inline, and out-of-line View values.
View variadic-buffer metadata.
Velox → Arrow load and load → offload → load round trips.
Arrow 15 fallback compatibility.

Besides,

we can manually test using following scripts

spark-3.5.5-bin-hadoop3/bin/spark-shell \
  --master 'local[1]' \
  --driver-memory 10g \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.driver.extraClassPath=/home/guangxinw/code/gluten2/package/target/gluten-velox-bundle-spark3.5_2.12-linux_aarch64-1.8.0-SNAPSHOT.jar \
  --conf spark.executor.extraClassPath=/home/guangxinw/code/gluten2/package/target/gluten-velox-bundle-spark3.5_2.12-linux_aarch64-1.8.0-SNAPSHOT.jar \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=8g \
  --conf spark.gluten.sql.columnar.maxBatchSize=4096 \
  --conf spark.gluten.sql.columnar.backend.velox.preferredBatchBytes=4g \
  --conf spark.sql.adaptive.enabled=false \
  --conf spark.driver.extraJavaOptions=-Darrow.allocation.manager.type=Unsafe \
  --conf spark.executor.extraJavaOptions=-Darrow.allocation.manager.type=Unsafe \
  < test.scala

test.scala

import org.apache.gluten.columnarbatch.ColumnarBatches
  import org.apache.gluten.memory.arrow.alloc.ArrowBufferAllocators
  import org.apache.gluten.vectorized.ArrowWritableColumnVector
  import org.apache.spark.sql.execution.SparkPlan
  try {
    // Velox's range source emits 1,000 rows per batch in this Spark setup. Keep exactly one
    // source batch and make its aggregate string payload exceed the int32 offset limit.
    val numRows = 1000L
    val bytesPerRow = 2200 * 1024
    val aggregatePayload = numRows * bytesPerRow
    println(
      f"Target payload: $aggregatePayload%,d bytes " +
        f"(${aggregatePayload / 1024.0 / 1024 / 1024}%.2f GiB)")
    val supportsStringView =
      try {
        classOf[ColumnarBatches]
          .getMethod("supportsArrowStringView")
          .invoke(null)
          .asInstanceOf[Boolean]
      } catch {
        case _: NoSuchMethodException => false
      }
    println(s"Arrow StringView available: $supportsStringView")
    require(
      aggregatePayload > Int.MaxValue,
      s"The test payload must exceed the Utf8 int32 offset limit: $aggregatePayload")
    // Including id prevents the large string expression from becoming one shared constant vector.
    val testDf = spark
      .range(0, numRows, 1, 1)
      .selectExpr(s"concat(cast(id as string), repeat('x', $bytesPerRow)) AS payload")
    def firstColumnarSubtree(plan: SparkPlan): Option[SparkPlan] = {
      if (plan.supportsColumnar) {
        Some(plan)
      } else {
        plan.children.iterator.map(firstColumnarSubtree).collectFirst { case Some(child) => child }
      }
    }
    val columnarPlan = firstColumnarSubtree(testDf.queryExecution.executedPlan).getOrElse {
      throw new IllegalStateException("The query has no native columnar subtree")
    }
    val results = columnarPlan.executeColumnar().mapPartitions {
      batches =>
        batches.map {
          nativeBatch =>
            val nativeRows = nativeBatch.numRows()
            val loaded =
              ColumnarBatches.load(ArrowBufferAllocators.contextInstance(), nativeBatch)
            try {
              val column = loaded.column(0).asInstanceOf[ArrowWritableColumnVector]
              val vectorClass = column.getValueVector.getClass.getName
              val firstLength = column.getUTF8String(0).numBytes()
              val lastLength = column.getUTF8String(nativeRows - 1).numBytes()
              require(
                vectorClass == "org.apache.arrow.vector.ViewVarCharVector",
                s"Expected ViewVarCharVector but found $vectorClass")
              s"rows=$nativeRows, vector=$vectorClass, firstLength=$firstLength, " +
                s"lastLength=$lastLength"
            } finally {
              loaded.close()
            }
        }
    }.collect()
    results.foreach(result => println(s"PASS: $result"))
    require(
      results.length == 1,
      s"Expected one batch, found ${results.length}: ${results.mkString(", ")}")
    println("PASS: Velox -> Arrow converted a VARCHAR column whose aggregate payload exceeds 2 GiB")
    System.exit(0)
  } catch {
    case error: Throwable =>
      error.printStackTrace()
      System.exit(1)
  }

without this patch, exception is

org.apache.gluten.exception.GlutenException:
Exception: VeloxRuntimeError
Error Source: RUNTIME
Error Code: INVALID_STATE
Reason: (2252802890 vs. 2147483647)
Function: exportStrings
Expression:
bufSize < std::numeric_limits<TOffsets>::max()
File:
velox/vector/arrow/Bridge.cpp:1027

Was this patch authored or co-authored using generative AI tooling?

Generated-by: GPT-5.6 Sol

Related issue: #13010

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86