diff --git a/.github/workflows/pyarrow_udf_test.yml b/.github/workflows/pyarrow_udf_test.yml
index 6af67aa7e36..600847abef8 100644
--- a/.github/workflows/pyarrow_udf_test.yml
+++ b/.github/workflows/pyarrow_udf_test.yml
@@ -28,23 +28,29 @@ on:
     paths: &feature-paths
       - "pom.xml"
       - "common/pom.xml"
-      - "common/src/main/scala/org/apache/comet/CometConf.scala"
+      - "native/shuffle/src/spark_unsafe/row.rs"
       - "spark/pom.xml"
+      - "spark/src/main/java/org/apache/comet/vector/**"
+      - "spark/src/main/java/org/apache/spark/sql/comet/execution/shuffle/SpillWriter.java"
+      - "spark/src/main/scala/org/apache/comet/CometConf.scala"
       - "spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala"
+      - "spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala"
+      - "spark/src/main/scala/org/apache/comet/vector/StreamReader.scala"
       - "spark/src/main/scala/org/apache/spark/sql/comet/CometMapInBatchExec.scala"
       - "spark/src/main/scala/org/apache/spark/sql/comet/shims/MapInBatchInfo.scala"
       - "spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
       - "spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
       - "spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
-      - "spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
-      - "spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
       - "spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala"
+      - "spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
       - "spark/src/main/spark-4.1/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala"
+      - "spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala"
       - "spark/src/main/spark-4.2/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala"
       - "spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/Spark4xMapInBatchSupport.scala"
       - "spark/src/main/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerBase.scala"
       - "spark/src/test/resources/pyspark/conftest.py"
       - "spark/src/test/resources/pyspark/test_pyarrow_udf.py"
+      - "spark/src/test/resources/pyspark/test_pyarrow_udf_dictionary_shuffle.py"
       - "spark/src/test/spark-3.5/org/apache/spark/sql/comet/CometMapInBatchSuite.scala"
       - "spark/src/test/spark-4.x/org/apache/spark/sql/comet/CometMapInBatchSuite.scala"
       - "spark/src/test/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerSuite.scala"
@@ -130,3 +136,5 @@ jobs:
         run: |
           /tmp/venv/bin/python -m pytest -v \
             spark/src/test/resources/pyspark/test_pyarrow_udf.py
+          /tmp/venv/bin/python -m pytest -v \
+            spark/src/test/resources/pyspark/test_pyarrow_udf_dictionary_shuffle.py
diff --git a/docs/source/user-guide/latest/pyarrow-udfs.md b/docs/source/user-guide/latest/pyarrow-udfs.md
index 14975c42c31..e013dad6b3f 100644
--- a/docs/source/user-guide/latest/pyarrow-udfs.md
+++ b/docs/source/user-guide/latest/pyarrow-udfs.md
@@ -183,8 +183,9 @@ on the unoptimized path.
 - The optimization currently applies only to `mapInArrow` and `mapInPandas`. Scalar pandas UDFs
   (`@pandas_udf`) and grouped operations (`applyInPandas`) are not yet supported.
 - The optimization requires Arrow data on the input side. If a shuffle sits between the upstream
-  Comet operator and the Python UDF, you need Comet's native shuffle for the optimization to
-  apply. Set `spark.shuffle.manager` to
+  Comet operator and the Python UDF, use Comet's columnar shuffle for the optimization to apply.
+  Both the `jvm` and `native` shuffle modes can feed `CometMapInBatch`. Set
+  `spark.shuffle.manager` to
   `org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager` and enable
   `spark.comet.shuffle.enabled=true` at session startup. With a vanilla Spark `Exchange`
   in the plan the data leaves the shuffle as rows and the optimization cannot fire.
@@ -209,8 +210,12 @@ on the unoptimized path.
   requested by the configuration. `EliminateRedundantTransitions` therefore skips the rewrite
   and vanilla Spark handles the operation. Comet can read `large_string` and `large_binary`
   columns returned by a Python worker; that output support does not widen the input vectors.
-- Comet writes input Arrow IPC record batches directly from its existing vector buffers. The
-  only additional Arrow buffer is the validity bitmap for the non-null struct that wraps the
-  input columns. Writing the IPC bytes to the Python worker's pipe still requires one copy;
-  that copy is inherent to Spark's process-based Python transport. This path does not transfer
-  buffers between Arrow allocators or change their ownership.
+- Comet writes input Arrow IPC record batches directly from existing plain vector buffers. The
+  only additional Arrow buffer for plain inputs is the validity bitmap for the non-null struct
+  that wraps the input columns. Before decoding dictionary-encoded shuffle columns, Comet uses
+  Spark's Arrow record threshold and the decoded dictionary size against Spark's byte threshold to
+  split the compact batch. Each temporary logical slice is released after its synchronous write.
+  Plain-only inputs continue to preserve their upstream Comet batch boundaries. Writing the IPC
+  bytes to the Python worker's pipe still requires one copy; that copy is inherent to Spark's
+  process-based Python transport. Borrowed buffers are not transferred between Arrow allocators or
+  given new ownership.
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index a7d7b80db1f..5a84f8299ea 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -596,9 +596,10 @@ object CometConf extends ShimCometConf {
     .withAlternative("spark.comet.shuffle.preferDictionary.ratio")
     .category(CATEGORY_SHUFFLE)
     .doc(
-      "The ratio of total values to distinct values in a string column to decide whether to " +
+      "The ratio of total values to distinct values in a string or binary column to decide " +
+        "whether to " +
         "prefer dictionary encoding when shuffling the column. If the ratio is higher than " +
-        "this config, dictionary encoding will be used on shuffling string column. This config " +
+        "this config, dictionary encoding will be used when shuffling the column. This config " +
         "is effective if it is higher than 1.0. Note that this " +
         "config is only used when `spark.comet.shuffle.mode` is `jvm`.")
     .doubleConf
diff --git a/spark/src/main/scala/org/apache/comet/Native.scala b/spark/src/main/scala/org/apache/comet/Native.scala
index e6090016186..d72bf7f9128 100644
--- a/spark/src/main/scala/org/apache/comet/Native.scala
+++ b/spark/src/main/scala/org/apache/comet/Native.scala
@@ -136,9 +136,9 @@ class Native extends NativeBase {
    * @param file
    *   the file path to write to.
    * @param preferDictionaryRatio
-   *   the ratio of total values to distinct values in a string column that makes the writer to
-   *   prefer dictionary encoding. If it is larger than the specified ratio, dictionary encoding
-   *   will be used when writing columns of string type.
+   *   the ratio of total values to distinct values in a string or binary column that makes the
+   *   writer prefer dictionary encoding. If it is larger than the specified ratio, dictionary
+   *   encoding will be used when writing columns of either type.
    * @param batchSize
    *   the batch size on the native side to buffer outputs during the row to columnar conversion
    *   before writing them out to disk.
diff --git a/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
index 9808f38d3da..b0cd245586f 100644
--- a/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
+++ b/spark/src/main/spark-4.0/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
@@ -43,5 +43,7 @@ trait ShimCometMapInBatch extends Spark4xMapInBatchSupport {
       schema,
       runnerInputs.pythonRunnerConf,
       pythonMetrics,
-      runnerInputs.jobArtifactUUID).compute(batchIter, partitionId, context)
+      runnerInputs.jobArtifactUUID,
+      runnerInputs.arrowMaxRecordsPerBatch,
+      runnerInputs.arrowMaxBytesPerBatch).compute(batchIter, partitionId, context)
 }
diff --git a/spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala b/spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
index b7d9eeabbfc..663de882006 100644
--- a/spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
+++ b/spark/src/main/spark-4.0/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
@@ -38,7 +38,9 @@ class CometArrowPythonRunner(
     override val schema: StructType,
     override val workerConf: Map[String, String],
     override val pythonMetrics: Map[String, SQLMetric],
-    jobArtifactUUID: Option[String])
+    jobArtifactUUID: Option[String],
+    override val arrowMaxRecordsPerBatch: Int,
+    override val arrowMaxBytesPerBatch: Long)
     extends BasePythonRunner[Iterator[ColumnarBatch], ColumnarBatch](
       funcs.map(_._1),
       evalType,
diff --git a/spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
index 56241d53fc4..32aec976857 100644
--- a/spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
+++ b/spark/src/main/spark-4.1/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
@@ -44,5 +44,7 @@ trait ShimCometMapInBatch extends Spark4xMapInBatchSupport {
       runnerInputs.pythonRunnerConf,
       pythonMetrics,
       runnerInputs.jobArtifactUUID,
-      None).compute(batchIter, partitionId, context)
+      None,
+      runnerInputs.arrowMaxRecordsPerBatch,
+      runnerInputs.arrowMaxBytesPerBatch).compute(batchIter, partitionId, context)
 }
diff --git a/spark/src/main/spark-4.1/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala b/spark/src/main/spark-4.1/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
index c6f3a4f8917..658e559adcf 100644
--- a/spark/src/main/spark-4.1/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
+++ b/spark/src/main/spark-4.1/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
@@ -40,7 +40,9 @@ class CometArrowPythonRunner(
     override val workerConf: Map[String, String],
     override val pythonMetrics: Map[String, SQLMetric],
     jobArtifactUUID: Option[String],
-    sessionUUID: Option[String])
+    sessionUUID: Option[String],
+    override val arrowMaxRecordsPerBatch: Int,
+    override val arrowMaxBytesPerBatch: Long)
     extends BasePythonRunner[Iterator[ColumnarBatch], ColumnarBatch](
       funcs.map(_._1),
       evalType,
diff --git a/spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala b/spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
index 56241d53fc4..32aec976857 100644
--- a/spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
+++ b/spark/src/main/spark-4.2/org/apache/spark/sql/comet/shims/ShimCometMapInBatch.scala
@@ -44,5 +44,7 @@ trait ShimCometMapInBatch extends Spark4xMapInBatchSupport {
       runnerInputs.pythonRunnerConf,
       pythonMetrics,
       runnerInputs.jobArtifactUUID,
-      None).compute(batchIter, partitionId, context)
+      None,
+      runnerInputs.arrowMaxRecordsPerBatch,
+      runnerInputs.arrowMaxBytesPerBatch).compute(batchIter, partitionId, context)
 }
diff --git a/spark/src/main/spark-4.2/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala b/spark/src/main/spark-4.2/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
index 789e4b97a96..2ec17fd4cdf 100644
--- a/spark/src/main/spark-4.2/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
+++ b/spark/src/main/spark-4.2/org/apache/spark/sql/execution/python/CometArrowPythonRunner.scala
@@ -39,7 +39,9 @@ class CometArrowPythonRunner(
     pythonRunnerConf: Map[String, String],
     override val pythonMetrics: Map[String, SQLMetric],
     jobArtifactUUID: Option[String],
-    sessionUUID: Option[String])
+    sessionUUID: Option[String],
+    override val arrowMaxRecordsPerBatch: Int,
+    override val arrowMaxBytesPerBatch: Long)
     extends BasePythonRunner[Iterator[ColumnarBatch], ColumnarBatch](
       funcs.map(_._1),
       evalType,
diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/Spark4xMapInBatchSupport.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/Spark4xMapInBatchSupport.scala
index 0586a755593..e096f87cd35 100644
--- a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/Spark4xMapInBatchSupport.scala
+++ b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/Spark4xMapInBatchSupport.scala
@@ -63,7 +63,9 @@ trait Spark4xMapInBatchSupport {
   protected case class RunnerInputs(
       chainedFunc: Seq[(ChainedPythonFunctions, Long)],
       pythonRunnerConf: Map[String, String],
-      jobArtifactUUID: Option[String])
+      jobArtifactUUID: Option[String],
+      arrowMaxRecordsPerBatch: Int,
+      arrowMaxBytesPerBatch: Long)
 
   /**
    * Resolves the `SQLConf`-derived inputs the `ArrowPythonRunner` needs. Must be called on the
@@ -74,5 +76,7 @@ trait Spark4xMapInBatchSupport {
     RunnerInputs(
       chainedFunc = Seq((ChainedPythonFunctions(Seq(pythonUDF.func)), pythonUDF.resultId.id)),
       pythonRunnerConf = ArrowPythonRunner.getPythonRunnerConfMap(conf),
-      jobArtifactUUID = JobArtifactSet.getCurrentJobArtifactState.map(_.uuid))
+      jobArtifactUUID = JobArtifactSet.getCurrentJobArtifactState.map(_.uuid),
+      arrowMaxRecordsPerBatch = conf.arrowMaxRecordsPerBatch,
+      arrowMaxBytesPerBatch = conf.arrowMaxBytesPerBatch)
 }
diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerBase.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerBase.scala
index 651a9f25bc7..b977ced2c64 100644
--- a/spark/src/main/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerBase.scala
+++ b/spark/src/main/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerBase.scala
@@ -27,8 +27,9 @@ import java.util.concurrent.atomic.AtomicBoolean
 import scala.jdk.CollectionConverters._
 
 import org.apache.arrow.memory.{ArrowBuf, BufferAllocator}
-import org.apache.arrow.vector.{FieldVector, VectorSchemaRoot, VectorUnloader}
+import org.apache.arrow.vector.{BaseFixedWidthVector, BaseLargeVariableWidthVector, BaseVariableWidthVector, FieldVector, NullVector, VectorSchemaRoot, VectorUnloader}
 import org.apache.arrow.vector.complex.StructVector
+import org.apache.arrow.vector.dictionary.DictionaryEncoder
 import org.apache.arrow.vector.ipc.{ArrowStreamReader, ArrowStreamWriter, WriteChannel}
 import org.apache.arrow.vector.ipc.message.{ArrowFieldNode, ArrowRecordBatch, MessageSerializer}
 import org.apache.arrow.vector.types.pojo.{ArrowType, Field, FieldType}
@@ -41,7 +42,7 @@ import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
 
 import org.apache.comet.CometArrowAllocator
-import org.apache.comet.vector.{CometDecodedVector, CometVector}
+import org.apache.comet.vector.{CometDecodedVector, CometDictionaryVector, CometVector}
 
 /**
  * Shared base for Comet's Arrow Python runners (Spark 4.0 / 4.1 / 4.2).
@@ -97,6 +98,10 @@ private[python] trait CometArrowPythonRunnerBase
    */
   protected def schema: StructType
 
+  /** Arrow input limits captured from the driver-side SQLConf. */
+  protected def arrowMaxRecordsPerBatch: Int
+  protected def arrowMaxBytesPerBatch: Long
+
   override val pythonExec: String =
     SQLConf.get.pysparkWorkerPythonExecutable.getOrElse(funcs.head.funcs.head.pythonExec)
 
@@ -178,44 +183,49 @@ private[python] trait CometArrowPythonRunnerBase
 
         val cometBatch = currentGroup.next()
         val startData = dataOut.size()
-        val sourceVectors = (0 until cometBatch.numCols()).map { i =>
-          cometBatch
-            .column(i)
-            .asInstanceOf[CometDecodedVector]
-            .getValueVector
-            .asInstanceOf[FieldVector]
+        val columns = (0 until cometBatch.numCols()).map { i =>
+          cometBatch.column(i).asInstanceOf[CometDecodedVector]
         }
-        val batchFields = sourceVectors.map(_.getField)
-
-        if (arrowWriter == null) {
-          // Build the schema-only struct root once from the first batch's child fields.
-          // mapInArrow/mapInPandas exchange the columns under a single non-nullable struct.
-          // Comet's FFI-imported vectors leave the Arrow Field name null, so restore the real
-          // column names from the input schema (the worker reads columns by name, and shaded
-          // Arrow rejects a null field name). Keep the field types and child structure as-is so
-          // the advertised schema matches the source buffers. Keeping the type as-is also means
-          // a TimestampType reaches the worker with Comet's UTC time zone
-          // rather than the session zone vanilla Spark would label it with; this is a documented
-          // limitation (see pyarrow-udfs.md), not a value difference, since the stored instant is
-          // identical.
-          val childNames = inputStructType.fieldNames
-          streamFields = batchFields.zipWithIndex.map { case (field, i) =>
-            renamed(field, childNames(i), forceNullable = true)
+        CometArrowPythonRunnerBase.foreachInputBatch(
+          columns,
+          cometBatch.numRows(),
+          arrowMaxRecordsPerBatch,
+          arrowMaxBytesPerBatch,
+          allocator) { (sourceVectors, numRows) =>
+          val batchFields = sourceVectors.map(_.getField)
+
+          if (arrowWriter == null) {
+            // Build the schema-only struct root once from the first batch's child fields.
+            // mapInArrow/mapInPandas exchange the columns under a single non-nullable struct.
+            // Comet's FFI-imported vectors leave the Arrow Field name null, so restore the real
+            // column names from the input schema (the worker reads columns by name, and shaded
+            // Arrow rejects a null field name). Keep the field types and child structure as-is
+            // so the advertised schema matches the source buffers. Keeping the type as-is also
+            // means a TimestampType reaches the worker with Comet's UTC time zone rather than
+            // the session zone vanilla Spark would label it with; this is a documented
+            // limitation (see pyarrow-udfs.md), not a value difference, since the stored instant
+            // is identical.
+            val childNames = inputStructType.fieldNames
+            streamFields = batchFields.zipWithIndex.map { case (field, i) =>
+              renamed(field, childNames(i), forceNullable = true)
+            }
+            startWriter(streamFields, dataOut)
           }
-          startWriter(streamFields, dataOut)
-        }
-
-        // Union branches may differ in names, nullability, or descriptive metadata. Only
-        // differences that change how the advertised schema interprets the buffers are invalid.
-        require(
-          CometArrowPythonRunnerBase.hasCompatibleSchema(streamFields, batchFields),
-          s"Arrow input schema changed between batches: expected $streamFields, got $batchFields")
 
-        CometArrowPythonRunnerBase.serializeBatch(
-          new WriteChannel(Channels.newChannel(dataOut)),
-          sourceVectors,
-          cometBatch.numRows(),
-          allocator)
+          // Union branches may differ in names, nullability, or descriptive metadata. Only
+          // differences that change how the advertised schema interprets the buffers are
+          // invalid.
+          require(
+            CometArrowPythonRunnerBase.hasCompatibleSchema(streamFields, batchFields),
+            s"Arrow input schema changed between batches: expected $streamFields, " +
+              s"got $batchFields")
+
+          CometArrowPythonRunnerBase.serializeBatch(
+            new WriteChannel(Channels.newChannel(dataOut)),
+            sourceVectors,
+            numRows,
+            allocator)
+        }
 
         pythonMetrics("pythonDataSent") += dataOut.size() - startData
         true
@@ -338,6 +348,170 @@ private[python] trait CometArrowPythonRunnerBase
 
 private[python] object CometArrowPythonRunnerBase {
 
+  // A regular Arrow variable-width data buffer uses signed 32-bit offsets. The Spark setting is
+  // already restricted to this range, but cap it here as a final guard for direct test callers.
+  private val MaxDecodedBatchBytes = Int.MaxValue.toLong
+
+  private def dictionaryVector(column: CometDictionaryVector): FieldVector = {
+    val indices = column.getValueVector
+    val encoding = indices.getField.getDictionary
+    column.getDictionaryProvider.lookup(encoding.getId).getVector
+  }
+
+  private def initialDecodedBytes(values: FieldVector): Long =
+    values match {
+      case _: BaseVariableWidthVector => BaseVariableWidthVector.OFFSET_WIDTH
+      case _: BaseLargeVariableWidthVector => BaseLargeVariableWidthVector.OFFSET_WIDTH
+      case _ => 0L
+    }
+
+  /** Conservative logical bytes added by one decoded dictionary value. */
+  private def decodedValueBytes(
+      column: CometDictionaryVector,
+      values: FieldVector,
+      row: Int,
+      batchRow: Int): Long = {
+    val dictionaryIndex = if (column.isNullAt(row)) -1 else column.indices.getInt(row)
+    val validityBytes = if ((batchRow & 7) == 0) 1L else 0L
+    values match {
+      case vector: BaseVariableWidthVector =>
+        val valueBytes = if (dictionaryIndex < 0) 0L else vector.getValueLength(dictionaryIndex)
+        valueBytes + BaseVariableWidthVector.OFFSET_WIDTH + validityBytes
+      case vector: BaseLargeVariableWidthVector =>
+        val valueBytes = if (dictionaryIndex < 0) 0L else vector.getValueLength(dictionaryIndex)
+        valueBytes + BaseLargeVariableWidthVector.OFFSET_WIDTH + validityBytes
+      case vector: BaseFixedWidthVector =>
+        vector.getBufferSizeFor(batchRow + 1).toLong -
+          vector.getBufferSizeFor(batchRow).toLong
+      case _: NullVector => 0L
+      case vector =>
+        // Comet's JVM shuffle currently dictionary-encodes only strings and binary values.
+        // If another Arrow type reaches this path, the complete dictionary is a safe upper
+        // bound for any one selected value and favors smaller batches over a large allocation.
+        math.max(1L, vector.getBufferSize.toLong)
+    }
+  }
+
+  private def saturatedAdd(left: Long, right: Long): Long =
+    if (right >= Long.MaxValue - left) Long.MaxValue else left + right
+
+  /**
+   * Split a compact dictionary batch before decoding it.
+   *
+   * The byte estimate covers the temporary logical dictionary vectors. Plain input vectors are
+   * already allocated and remain zero-copy when no dictionary column is present. Every returned
+   * range is applied to all columns so rows stay aligned. A single oversized row is allowed,
+   * matching Spark's Arrow batching contract.
+   */
+  private[python] def inputBatchRanges(
+      columns: Seq[CometDecodedVector],
+      numRows: Int,
+      maxRecordsPerBatch: Int,
+      maxBytesPerBatch: Long): Seq[(Int, Int)] = {
+    require(numRows >= 0, s"Input batch row count must be non-negative: $numRows")
+
+    val dictionaries = columns.collect { case column: CometDictionaryVector =>
+      column -> dictionaryVector(column)
+    }
+    if (numRows == 0 || dictionaries.isEmpty) {
+      return Seq(0 -> numRows)
+    }
+
+    val recordLimit =
+      if (maxRecordsPerBatch > 0) maxRecordsPerBatch else Int.MaxValue
+    val byteLimit =
+      if (maxBytesPerBatch > 0) math.min(maxBytesPerBatch, MaxDecodedBatchBytes)
+      else MaxDecodedBatchBytes
+    val initialBytes = dictionaries.foldLeft(0L) { case (bytes, (_, values)) =>
+      saturatedAdd(bytes, initialDecodedBytes(values))
+    }
+
+    val ranges = Seq.newBuilder[(Int, Int)]
+    var start = 0
+    var row = 0
+    var decodedBytes = initialBytes
+    while (row < numRows) {
+      var rowsInBatch = row - start
+      var rowBytes = dictionaries.foldLeft(0L) { case (bytes, (column, values)) =>
+        saturatedAdd(bytes, decodedValueBytes(column, values, row, rowsInBatch))
+      }
+      // Spark checks the configured byte limit before adding the next row, so the row that
+      // crosses that soft limit stays in the current batch. The separate hard check prevents a
+      // regular variable-width buffer from crossing Arrow's signed 32-bit allocation ceiling.
+      val exceedsArrowLimit =
+        decodedBytes >= MaxDecodedBatchBytes ||
+          rowBytes > MaxDecodedBatchBytes - decodedBytes
+      if (rowsInBatch > 0 &&
+        (rowsInBatch >= recordLimit || decodedBytes >= byteLimit || exceedsArrowLimit)) {
+        ranges += start -> rowsInBatch
+        start = row
+        decodedBytes = initialBytes
+        rowsInBatch = 0
+        rowBytes = dictionaries.foldLeft(0L) { case (bytes, (column, values)) =>
+          saturatedAdd(bytes, decodedValueBytes(column, values, row, rowsInBatch))
+        }
+      }
+      decodedBytes = saturatedAdd(decodedBytes, rowBytes)
+      row += 1
+    }
+    ranges += start -> (numRows - start)
+    ranges.result()
+  }
+
+  /** Materialize and visit each safely sized, row-aligned input range synchronously. */
+  private[python] def foreachInputBatch(
+      columns: Seq[CometDecodedVector],
+      numRows: Int,
+      maxRecordsPerBatch: Int,
+      maxBytesPerBatch: Long,
+      allocator: BufferAllocator)(body: (Seq[FieldVector], Int) => Unit): Unit = {
+    inputBatchRanges(columns, numRows, maxRecordsPerBatch, maxBytesPerBatch).foreach {
+      case (0, length) if length == numRows =>
+        withMaterializedInputVectors(columns, allocator)(body(_, length))
+      case (offset, length) =>
+        val slices = new ArrayList[CometDecodedVector]()
+        try {
+          columns.foreach { column =>
+            slices.add(column.slice(offset, length).asInstanceOf[CometDecodedVector])
+          }
+          withMaterializedInputVectors(slices.asScala.toSeq, allocator)(body(_, length))
+        } finally {
+          slices.asScala.reverseIterator.foreach(_.close())
+        }
+    }
+  }
+
+  /**
+   * Supply logical Arrow vectors to the serializer for the duration of the body.
+   *
+   * Plain Comet vectors already expose their logical values and remain borrowed.
+   * Dictionary-backed shuffle columns expose only their integer indices through getValueVector,
+   * so materialize those columns first. The temporary decoded vectors own their buffers and are
+   * closed after the synchronous write, including schema and serialization failures.
+   */
+  private[python] def withMaterializedInputVectors[T](
+      columns: Seq[CometDecodedVector],
+      allocator: BufferAllocator)(body: Seq[FieldVector] => T): T = {
+    val materialized = new ArrayList[FieldVector]()
+    try {
+      val vectors = columns.map {
+        case dictionaryVector: CometDictionaryVector =>
+          val indices = dictionaryVector.getValueVector
+          val encoding = indices.getField.getDictionary
+          val dictionary = dictionaryVector.getDictionaryProvider.lookup(encoding.getId)
+          val decoded = DictionaryEncoder
+            .decode(indices, dictionary, allocator)
+            .asInstanceOf[FieldVector]
+          materialized.add(decoded)
+          decoded
+        case column => column.getValueVector.asInstanceOf[FieldVector]
+      }
+      body(vectors)
+    } finally {
+      materialized.asScala.foreach(_.close())
+    }
+  }
+
   // Extensions can change interpretation even when their underlying storage types match.
   private val extensionMetadataKeys = Seq(
     ArrowType.ExtensionType.EXTENSION_METADATA_KEY_NAME,
diff --git a/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py b/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
index 7d21c9822b3..e05763910e0 100644
--- a/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
+++ b/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
@@ -32,7 +32,7 @@
               (per-row InternalRow.getXXX() loop inside ArrowWriter.write)
   * optimized: CometScan -> CometMapInBatchExec -> CometArrowPythonRunner
               (Arrow IPC serialization directly from Comet's source vectors;
-              no row materialization or intermediate vector-buffer copy)
+              no row materialization; dictionary inputs use temporary decoded vectors)
 
 Results are wall-clock seconds, so they include Python interpreter,
 Arrow IPC, and downstream count() costs. That's intentional: the
@@ -58,6 +58,8 @@
     BENCHMARK_ROWS=2000000                rows per run
     BENCHMARK_WARMUP=2                    warmup iterations per case
     BENCHMARK_ITERS=5                     measured iterations per case
+    BENCHMARK_WORKLOAD='dictionary JVM shuffle'
+                                          run only the named workload
 """
 
 import contextlib
@@ -88,6 +90,11 @@ def _build_spark() -> SparkSession:
             "spark.shuffle.manager",
             "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
         )
+        # Force the JVM shuffle and a low dictionary threshold so the dictionary workload
+        # exercises CometDictionaryVector rather than the native shuffle's plain vectors.
+        .config("spark.comet.shuffle.mode", "jvm")
+        .config("spark.comet.shuffle.jvm.preferDictionary.ratio", "1.01")
+        .config("spark.sql.shuffle.partitions", "2")
         .config("spark.memory.offHeap.enabled", "true")
         .config("spark.memory.offHeap.size", "4g")
         .config("spark.driver.memory", "4g")
@@ -127,18 +134,28 @@ def _mixed_with_strings(spark: SparkSession, n: int):
     )
 
 
+def _low_cardinality_strings(spark: SparkSession, n: int):
+    return spark.range(n).selectExpr(
+        "id as id_long",
+        """case
+             when id % 23 = 0 then cast(null as string)
+             when id % 17 = 0 then ''
+             else concat('value_', cast(id % 8 as string))
+           end as repeated_str""",
+    )
+
+
 def _wide_rows(spark: SparkSession, n: int):
     types = ["int", "long", "double"]
-    cols = [
-        f"cast(id + {i} as {types[i % len(types)]}) as col_{i}" for i in range(50)
-    ]
+    cols = [f"cast(id + {i} as {types[i % len(types)]}) as col_{i}" for i in range(50)]
     return spark.range(n).selectExpr(*cols)
 
 
 WORKLOADS = [
-    ("narrow primitives", _narrow_primitives),
-    ("mixed with strings", _mixed_with_strings),
-    ("wide rows (50 cols)", _wide_rows),
+    ("narrow primitives", _narrow_primitives, False),
+    ("mixed with strings", _mixed_with_strings, False),
+    ("wide rows (50 cols)", _wide_rows, False),
+    ("dictionary JVM shuffle", _low_cardinality_strings, True),
 ]
 
 
@@ -150,12 +167,20 @@ def _temp_parquet(spark: SparkSession, build_df, n: int):
         yield path
 
 
-def _time_run(spark: SparkSession, parquet_path: str, accelerate: bool, api: str) -> float:
+def _time_run(
+    spark: SparkSession,
+    parquet_path: str,
+    accelerate: bool,
+    api: str,
+    shuffle_input: bool,
+) -> float:
     spark.conf.set(
         "spark.comet.exec.pyarrowUDF.enabled",
         "true" if accelerate else "false",
     )
     df = spark.read.parquet(parquet_path)
+    if shuffle_input:
+        df = df.repartition(2, "id_long")
     schema = df.schema
     if api == "mapInArrow":
         df = df.mapInArrow(_passthrough_arrow, schema)
@@ -165,6 +190,8 @@ def _time_run(spark: SparkSession, parquet_path: str, accelerate: bool, api: str
     if ("CometMapInBatch" in plan) != accelerate:
         expected = "CometMapInBatch" if accelerate else "vanilla Python execution"
         raise RuntimeError(f"Expected {expected} for {api}, but found:\n{plan}")
+    if shuffle_input and "CometColumnarExchange" not in plan:
+        raise RuntimeError(f"Expected Comet JVM shuffle for {api}, but found:\n{plan}")
     t0 = time.perf_counter()
     df.count()
     return time.perf_counter() - t0
@@ -174,6 +201,15 @@ def main() -> None:
     rows = int(os.environ.get("BENCHMARK_ROWS", 1024 * 1024))
     warmup = int(os.environ.get("BENCHMARK_WARMUP", 2))
     iters = int(os.environ.get("BENCHMARK_ITERS", 5))
+    workload_name = os.environ.get("BENCHMARK_WORKLOAD")
+    workloads = WORKLOADS
+    if workload_name:
+        workloads = [workload for workload in WORKLOADS if workload[0] == workload_name]
+        if not workloads:
+            names = ", ".join(name for name, _, _ in WORKLOADS)
+            raise ValueError(
+                f"Unknown BENCHMARK_WORKLOAD {workload_name!r}; choose from: {names}"
+            )
 
     spark = _build_spark()
     spark.sparkContext.setLogLevel("WARN")
@@ -188,16 +224,16 @@ def main() -> None:
     print(header)
     print("  " + "-" * (len(header) - 2))
 
-    for name, build_df in WORKLOADS:
+    for name, build_df, shuffle_input in workloads:
         print(f"\n=== {name} ===")
         with _temp_parquet(spark, build_df, rows) as parquet_path:
             for api in ("mapInArrow", "mapInPandas"):
                 samples_by_mode = {}
                 for mode, accelerate in (("vanilla", False), ("optimized", True)):
                     for _ in range(warmup):
-                        _time_run(spark, parquet_path, accelerate, api)
+                        _time_run(spark, parquet_path, accelerate, api, shuffle_input)
                     samples = [
-                        _time_run(spark, parquet_path, accelerate, api)
+                        _time_run(spark, parquet_path, accelerate, api, shuffle_input)
                         for _ in range(iters)
                     ]
                     samples_by_mode[mode] = samples
diff --git a/spark/src/test/resources/pyspark/test_pyarrow_udf_dictionary_shuffle.py b/spark/src/test/resources/pyspark/test_pyarrow_udf_dictionary_shuffle.py
new file mode 100755
index 00000000000..791bcc108a6
--- /dev/null
+++ b/spark/src/test/resources/pyspark/test_pyarrow_udf_dictionary_shuffle.py
@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Regression coverage for dictionary-encoded Comet shuffle input to Python UDFs."""
+
+import os
+
+import pyarrow as pa
+import pytest
+from conftest import resolve_comet_jar
+from pyspark.sql import SparkSession, types as T
+
+
+@pytest.fixture(scope="session")
+def spark():
+    jar = resolve_comet_jar()
+    os.environ["PYSPARK_SUBMIT_ARGS"] = (
+        f"--jars {jar} --driver-class-path {jar} pyspark-shell"
+    )
+    session = (
+        SparkSession.builder.master("local[2]")
+        .appName("comet-pyarrow-udf-dictionary-shuffle-tests")
+        .config("spark.plugins", "org.apache.spark.CometPlugin")
+        .config("spark.comet.enabled", "true")
+        .config("spark.comet.exec.enabled", "true")
+        .config("spark.comet.exec.pyarrowUDF.enabled", "true")
+        .config(
+            "spark.shuffle.manager",
+            "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
+        )
+        .config("spark.comet.shuffle.mode", "jvm")
+        .config("spark.comet.shuffle.jvm.preferDictionary.ratio", "1.01")
+        .config("spark.sql.adaptive.enabled", "false")
+        .config("spark.sql.shuffle.partitions", "2")
+        .config("spark.memory.offHeap.enabled", "true")
+        .config("spark.memory.offHeap.size", "2g")
+        .getOrCreate()
+    )
+    try:
+        yield session
+    finally:
+        session.stop()
+
+
+def _comparable(row):
+    return (
+        row.id,
+        row.text,
+        None if row.data is None else bytes(row.data),
+    )
+
+
+@pytest.mark.parametrize("api", ["mapInArrow", "mapInPandas"])
+def test_dictionary_shuffle_input(spark, tmp_path, api: str):
+    spark.conf.set("spark.sql.execution.arrow.useLargeVarTypes", "false")
+    rows = []
+    for index in range(200):
+        text = None if index % 23 == 0 else ("" if index % 17 == 0 else "same-text")
+        data = (
+            None
+            if index % 29 == 0
+            else (bytearray() if index % 19 == 0 else bytearray(b"same-binary"))
+        )
+        rows.append((index, text, data))
+
+    path = str(tmp_path / "dictionary-shuffle.parquet")
+    spark.createDataFrame(rows, "id int, text string, data binary").write.parquet(path)
+    source = spark.read.parquet(path).repartition(2, "id")
+
+    if api == "mapInArrow":
+
+        def passthrough(iterator):
+            for batch in iterator:
+                text_type = batch.schema.field("text").type
+                data_type = batch.schema.field("data").type
+                assert pa.types.is_string(text_type)
+                assert pa.types.is_binary(data_type)
+                yield batch
+
+        result = source.mapInArrow(passthrough, source.schema)
+    else:
+
+        def passthrough(iterator):
+            yield from iterator
+
+        result = source.mapInPandas(passthrough, source.schema)
+
+    plan = result._jdf.queryExecution().executedPlan().toString()
+    assert "CometColumnarExchange" in plan, plan
+    assert "CometMapInBatch" in plan, plan
+    assert "ColumnarToRow" not in plan, plan
+
+    actual = sorted(_comparable(row) for row in result.collect())
+    expected = sorted(
+        (
+            index,
+            text,
+            None if data is None else bytes(data),
+        )
+        for index, text, data in rows
+    )
+    assert actual == expected
+
+
+@pytest.mark.parametrize("api", ["mapInArrow", "mapInPandas"])
+@pytest.mark.parametrize(
+    "max_records,max_bytes,expected_batch_sizes",
+    [
+        (2, 256 * 1024 * 1024, [2, 2, 2, 2, 2]),
+        (100, 4096, [1] * 10),
+    ],
+)
+def test_dictionary_shuffle_input_respects_arrow_batch_limits(
+    spark,
+    tmp_path,
+    api: str,
+    max_records: int,
+    max_bytes: int,
+    expected_batch_sizes: list[int],
+):
+    """Split compact shuffle dictionaries using their decoded logical size."""
+    previous_records = spark.conf.get("spark.sql.execution.arrow.maxRecordsPerBatch")
+    previous_bytes = spark.conf.get("spark.sql.execution.arrow.maxBytesPerBatch")
+    spark.conf.set("spark.sql.execution.arrow.useLargeVarTypes", "false")
+    spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", str(max_records))
+    spark.conf.set("spark.sql.execution.arrow.maxBytesPerBatch", str(max_bytes))
+    try:
+        text_values = ["a" * (32 * 1024), "b" * (32 * 1024)]
+        binary_values = [bytearray(b"c" * (32 * 1024)), bytearray(b"d" * (32 * 1024))]
+        rows = [
+            (index, text_values[index % 2], binary_values[index % 2])
+            for index in range(10)
+        ]
+
+        path = str(tmp_path / "dictionary-shuffle-batch-limits.parquet")
+        spark.createDataFrame(rows, "id int, text string, data binary").coalesce(
+            1
+        ).write.parquet(path)
+        source = spark.read.parquet(path).repartition(1, "id")
+        output_schema = T.StructType(
+            [
+                *source.schema.fields,
+                T.StructField("input_batch_id", T.IntegerType(), nullable=False),
+                T.StructField("input_batch_rows", T.IntegerType(), nullable=False),
+            ]
+        )
+
+        if api == "mapInArrow":
+
+            def annotate_batches(iterator):
+                for batch_id, batch in enumerate(iterator):
+                    yield pa.RecordBatch.from_arrays(
+                        [
+                            *batch.columns,
+                            pa.array([batch_id] * batch.num_rows, type=pa.int32()),
+                            pa.array(
+                                [batch.num_rows] * batch.num_rows, type=pa.int32()
+                            ),
+                        ],
+                        names=output_schema.fieldNames(),
+                    )
+
+            result = source.mapInArrow(annotate_batches, output_schema)
+        else:
+
+            def annotate_batches(iterator):
+                for batch_id, frame in enumerate(iterator):
+                    yield frame.assign(
+                        input_batch_id=batch_id,
+                        input_batch_rows=len(frame),
+                    )
+
+            result = source.mapInPandas(annotate_batches, output_schema)
+
+        plan = result._jdf.queryExecution().executedPlan().toString()
+        assert "CometColumnarExchange" in plan, plan
+        assert "CometMapInBatch" in plan, plan
+        assert "ColumnarToRow" not in plan, plan
+
+        output = result.collect()
+        observed_batches = {}
+        for row in output:
+            observed_batches.setdefault(row.input_batch_id, []).append(row)
+        assert sorted(observed_batches) == list(range(len(expected_batch_sizes)))
+        assert [
+            len(observed_batches[batch_id]) for batch_id in sorted(observed_batches)
+        ] == expected_batch_sizes
+        for batch_rows in observed_batches.values():
+            assert {row.input_batch_rows for row in batch_rows} == {len(batch_rows)}
+
+        actual = sorted(_comparable(row) for row in output)
+        expected = sorted((index, text, bytes(data)) for index, text, data in rows)
+        assert actual == expected
+    finally:
+        spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", previous_records)
+        spark.conf.set("spark.sql.execution.arrow.maxBytesPerBatch", previous_bytes)
diff --git a/spark/src/test/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerSuite.scala b/spark/src/test/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerSuite.scala
index 04c0a16cc6e..d0ff798e492 100644
--- a/spark/src/test/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerSuite.scala
+++ b/spark/src/test/spark-4.x/org/apache/spark/sql/execution/python/CometArrowPythonRunnerSuite.scala
@@ -22,8 +22,10 @@ package org.apache.spark.sql.execution.python
 import java.io.{ByteArrayInputStream, ByteArrayOutputStream, IOException}
 import java.nio.ByteBuffer
 import java.nio.channels.{Channels, WritableByteChannel}
+import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Paths}
 
+import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._
 
 import org.scalatest.funsuite.AnyFunSuite
@@ -31,12 +33,15 @@ import org.scalatest.matchers.should.Matchers
 
 import org.apache.arrow.c.{ArrowArray, ArrowSchema, Data}
 import org.apache.arrow.memory.{BufferAllocator, RootAllocator}
-import org.apache.arrow.vector.{FieldVector, IntVector, NullVector, VarCharVector, VectorSchemaRoot}
+import org.apache.arrow.vector.{FieldVector, IntVector, NullVector, VarBinaryVector, VarCharVector, VectorSchemaRoot}
 import org.apache.arrow.vector.complex.{ListVector, MapVector, StructVector}
+import org.apache.arrow.vector.dictionary.{Dictionary, DictionaryProvider}
 import org.apache.arrow.vector.ipc.{ArrowStreamReader, ArrowStreamWriter, WriteChannel}
 import org.apache.arrow.vector.types.TimeUnit
 import org.apache.arrow.vector.types.pojo.{ArrowType, DictionaryEncoding, Field, FieldType, Schema}
-import org.apache.spark.sql.execution.python.CometArrowPythonRunnerBase.{hasCompatibleSchema, serializeBatch}
+import org.apache.spark.sql.execution.python.CometArrowPythonRunnerBase.{foreachInputBatch, hasCompatibleSchema, serializeBatch, withMaterializedInputVectors}
+
+import org.apache.comet.vector.{CometDecodedVector, CometDictionary, CometDictionaryVector, CometPlainVector}
 
 class CometArrowPythonRunnerSuite extends AnyFunSuite with Matchers {
 
@@ -71,6 +76,59 @@ class CometArrowPythonRunnerSuite extends AnyFunSuite with Matchers {
     }
   }
 
+  private case class LargeDictionaryInput(
+      columns: Seq[CometDecodedVector],
+      values: VarCharVector,
+      indices: IntVector,
+      expected: Seq[String],
+      allocator: BufferAllocator)
+      extends AutoCloseable {
+    override def close(): Unit = {
+      columns.foreach(_.close())
+      allocator.close()
+    }
+  }
+
+  private def largeDictionaryInput(): LargeDictionaryInput = {
+    val allocator = new RootAllocator(Long.MaxValue)
+    val intType = new ArrowType.Int(32, true)
+    val encoding = new DictionaryEncoding(21L, false, intType)
+    val values = new VarCharVector("text", allocator)
+    val indices =
+      new IntVector("text", new FieldType(true, intType, encoding), allocator)
+    val dictionary = new Dictionary(values, encoding)
+    val provider = new DictionaryProvider {
+      override def lookup(id: Long): Dictionary = {
+        require(id == encoding.getId)
+        dictionary
+      }
+
+      override def getDictionaryIds: java.util.Set[java.lang.Long] =
+        Set(java.lang.Long.valueOf(encoding.getId)).asJava
+    }
+
+    val dictionaryValues = Seq("a" * (64 * 1024), "b" * (64 * 1024))
+    values.allocateNew()
+    dictionaryValues.zipWithIndex.foreach { case (value, index) =>
+      values.setSafe(index, value.getBytes(StandardCharsets.UTF_8))
+    }
+    values.setValueCount(dictionaryValues.size)
+    indices.allocateNew()
+    val expected = (0 until 10).map { index =>
+      val valueIndex = index % dictionaryValues.size
+      indices.setSafe(index, valueIndex)
+      dictionaryValues(valueIndex)
+    }
+    indices.setValueCount(expected.size)
+
+    val columns = Seq[CometDecodedVector](
+      new CometDictionaryVector(
+        new CometPlainVector(indices),
+        new CometDictionary(new CometPlainVector(values)),
+        provider))
+    LargeDictionaryInput(columns, values, indices, expected, allocator)
+  }
+
   test("input schema compatibility preserves physical types and nested layouts") {
     val intType = new ArrowType.Int(32, true)
     def nested(dataType: ArrowType): Seq[Field] = Seq(
@@ -336,6 +394,242 @@ class CometArrowPythonRunnerSuite extends AnyFunSuite with Matchers {
     }
   }
 
+  for (failSerialization <- Seq(false, true)) {
+    test(s"dictionary inputs materialize logical values (failure: $failSerialization)") {
+      val sourceAllocator = new RootAllocator(Long.MaxValue)
+      val writerAllocator = new RootAllocator(Long.MaxValue)
+      val intType = new ArrowType.Int(32, true)
+      val textEncoding = new DictionaryEncoding(11L, false, intType)
+      val binaryEncoding = new DictionaryEncoding(12L, false, intType)
+      val textValues = new VarCharVector("text", sourceAllocator)
+      val binaryValues = new VarBinaryVector("data", sourceAllocator)
+      val textIndices =
+        new IntVector("text", new FieldType(true, intType, textEncoding), sourceAllocator)
+      val binaryIndices =
+        new IntVector("data", new FieldType(true, intType, binaryEncoding), sourceAllocator)
+      val dictionaries = Map(
+        textEncoding.getId -> new Dictionary(textValues, textEncoding),
+        binaryEncoding.getId -> new Dictionary(binaryValues, binaryEncoding))
+      val provider = new DictionaryProvider {
+        override def lookup(id: Long): Dictionary = dictionaries(id)
+
+        override def getDictionaryIds: java.util.Set[java.lang.Long] =
+          dictionaries.keys.map(id => java.lang.Long.valueOf(id)).toSet.asJava
+      }
+      val columns = Seq[CometDecodedVector](
+        new CometDictionaryVector(
+          new CometPlainVector(textIndices),
+          new CometDictionary(new CometPlainVector(textValues)),
+          provider),
+        new CometDictionaryVector(
+          new CometPlainVector(binaryIndices),
+          new CometDictionary(new CometPlainVector(binaryValues)),
+          provider))
+      var failWrites = false
+      val output = new ByteArrayOutputStream() {
+        override def write(bytes: Array[Byte], offset: Int, length: Int): Unit = {
+          if (failWrites) {
+            throw new IOException("injected dictionary IPC write failure")
+          }
+          super.write(bytes, offset, length)
+        }
+      }
+      try {
+        textValues.allocateNew()
+        Seq("same", "", "λ中文").zipWithIndex.foreach { case (value, index) =>
+          textValues.setSafe(index, value.getBytes(StandardCharsets.UTF_8))
+        }
+        textValues.setValueCount(3)
+        binaryValues.allocateNew()
+        Seq(Array[Byte](1, 2), Array.emptyByteArray, Array[Byte](0, -1)).zipWithIndex.foreach {
+          case (value, index) => binaryValues.setSafe(index, value)
+        }
+        binaryValues.setValueCount(3)
+        textIndices.allocateNew()
+        Seq(0, 1, 0, 2).zipWithIndex.foreach { case (value, index) =>
+          textIndices.setSafe(index, value)
+        }
+        textIndices.setNull(2)
+        textIndices.setValueCount(4)
+        binaryIndices.allocateNew()
+        Seq(2, 0, 0, 1).zipWithIndex.foreach { case (value, index) =>
+          binaryIndices.setSafe(index, value)
+        }
+        binaryIndices.setNull(2)
+        binaryIndices.setValueCount(4)
+
+        val sourceVectors = Seq(textValues, binaryValues, textIndices, binaryIndices)
+        val sourceBuffers = sourceVectors.flatMap(_.getFieldBuffers.asScala)
+        val sourceRefs = sourceBuffers.map(_.refCnt())
+        val sourceBytes = sourceAllocator.getAllocatedMemory
+
+        def writeDictionaryBatch(): Unit =
+          withMaterializedInputVectors(columns, writerAllocator) { vectors =>
+            vectors.map(_.getField.getDictionary) shouldBe Seq(null, null)
+            vectors.head.getObject(0).toString shouldBe "same"
+            vectors.head.getObject(1).toString shouldBe ""
+            vectors.head.isNull(2) shouldBe true
+            vectors.head.getObject(3).toString shouldBe "λ中文"
+            vectors(1).getObject(0).asInstanceOf[Array[Byte]] shouldBe Array[Byte](0, -1)
+            vectors(1).isNull(2) shouldBe true
+
+            withWriter(vectors.map(_.getField), writerAllocator, Channels.newChannel(output)) {
+              channel =>
+                failWrites = failSerialization
+                try {
+                  serializeBatch(new WriteChannel(channel), vectors, 4, writerAllocator)
+                } finally {
+                  failWrites = false
+                }
+            }
+          }
+
+        if (failSerialization) {
+          val error = intercept[IOException](writeDictionaryBatch())
+          error.getMessage shouldBe "injected dictionary IPC write failure"
+        } else {
+          writeDictionaryBatch()
+          withReader(output.toByteArray) { reader =>
+            reader.loadNextBatch() shouldBe true
+            val struct = reader.getVectorSchemaRoot.getVector(0).asInstanceOf[StructVector]
+            val resultText = struct.getChild("text")
+            val resultData = struct.getChild("data")
+            resultText.getField.getType shouldBe ArrowType.Utf8.INSTANCE
+            resultData.getField.getType shouldBe ArrowType.Binary.INSTANCE
+            resultText.getObject(0).toString shouldBe "same"
+            resultText.getObject(1).toString shouldBe ""
+            resultText.isNull(2) shouldBe true
+            resultText.getObject(3).toString shouldBe "λ中文"
+            resultData.getObject(0).asInstanceOf[Array[Byte]] shouldBe Array[Byte](0, -1)
+            resultData.isNull(2) shouldBe true
+            reader.loadNextBatch() shouldBe false
+          }
+        }
+
+        writerAllocator.getAllocatedMemory shouldBe 0L
+        sourceAllocator.getAllocatedMemory shouldBe sourceBytes
+        sourceBuffers.map(_.refCnt()) shouldBe sourceRefs
+        textValues.getObject(0).toString shouldBe "same"
+        binaryValues.getObject(2).asInstanceOf[Array[Byte]] shouldBe Array[Byte](0, -1)
+      } finally {
+        columns.foreach(_.close())
+        writerAllocator.close()
+        sourceAllocator.close()
+      }
+    }
+  }
+
+  test("dictionary inputs are sliced before decoding to the Arrow batch limits") {
+    val input = largeDictionaryInput()
+    val sourceVectors = Seq(input.values, input.indices)
+    val sourceBuffers = sourceVectors.flatMap(_.getFieldBuffers.asScala)
+    val sourceRefs = sourceBuffers.map(_.refCnt())
+    val sourceBytes = input.allocator.getAllocatedMemory
+    val fullDecodedDataBytes =
+      input.expected.map(_.getBytes(StandardCharsets.UTF_8).length.toLong).sum
+    try {
+      val cases = Seq(
+        (2, Int.MaxValue.toLong, Seq.fill(5)(2)),
+        (100, 100L * 1024L, Seq.fill(5)(2)),
+        (100, 4096L, Seq.fill(10)(1)))
+      cases.foreach { case (maxRecords, maxBytes, expectedBatchSizes) =>
+        val writerAllocator = new RootAllocator(256 * 1024)
+        val output = new ByteArrayOutputStream()
+        try {
+          withWriter(Seq(input.values.getField), writerAllocator, Channels.newChannel(output)) {
+            channel =>
+              foreachInputBatch(
+                input.columns,
+                input.expected.size,
+                maxRecords,
+                maxBytes,
+                writerAllocator) { (vectors, numRows) =>
+                serializeBatch(new WriteChannel(channel), vectors, numRows, writerAllocator)
+              }
+          }
+
+          val actualBatchSizes = ArrayBuffer.empty[Int]
+          val actualValues = ArrayBuffer.empty[String]
+          withReader(output.toByteArray) { reader =>
+            while (reader.loadNextBatch()) {
+              val root = reader.getVectorSchemaRoot
+              actualBatchSizes += root.getRowCount
+              val struct = root.getVector(0).asInstanceOf[StructVector]
+              val text = struct.getChild("text").asInstanceOf[VarCharVector]
+              (0 until root.getRowCount).foreach { row =>
+                actualValues += text.getObject(row).toString
+              }
+            }
+          }
+
+          actualBatchSizes.toSeq shouldBe expectedBatchSizes
+          actualValues.toSeq shouldBe input.expected
+          writerAllocator.getAllocatedMemory shouldBe 0L
+          writerAllocator.getPeakMemoryAllocation should be < fullDecodedDataBytes
+          input.allocator.getAllocatedMemory shouldBe sourceBytes
+          sourceBuffers.map(_.refCnt()) shouldBe sourceRefs
+          input.indices.get(9) shouldBe 1
+          input.values.getObject(0).toString shouldBe input.expected.head
+        } finally {
+          writerAllocator.close()
+        }
+      }
+    } finally {
+      input.close()
+    }
+  }
+
+  test("dictionary input slices are released when a later write fails") {
+    val input = largeDictionaryInput()
+    val sourceVectors = Seq(input.values, input.indices)
+    val sourceBuffers = sourceVectors.flatMap(_.getFieldBuffers.asScala)
+    val sourceRefs = sourceBuffers.map(_.refCnt())
+    val sourceBytes = input.allocator.getAllocatedMemory
+    val writerAllocator = new RootAllocator(256 * 1024)
+    var failWrites = false
+    var slicesEntered = 0
+    val output = new ByteArrayOutputStream() {
+      override def write(bytes: Array[Byte], offset: Int, length: Int): Unit = {
+        if (failWrites) {
+          throw new IOException("injected sliced dictionary IPC write failure")
+        }
+        super.write(bytes, offset, length)
+      }
+    }
+    try {
+      val error = intercept[IOException] {
+        withWriter(Seq(input.values.getField), writerAllocator, Channels.newChannel(output)) {
+          channel =>
+            foreachInputBatch(
+              input.columns,
+              input.expected.size,
+              maxRecordsPerBatch = 2,
+              maxBytesPerBatch = Int.MaxValue.toLong,
+              allocator = writerAllocator) { (vectors, numRows) =>
+              slicesEntered += 1
+              failWrites = slicesEntered == 2
+              try {
+                serializeBatch(new WriteChannel(channel), vectors, numRows, writerAllocator)
+              } finally {
+                failWrites = false
+              }
+            }
+        }
+      }
+      error.getMessage shouldBe "injected sliced dictionary IPC write failure"
+      slicesEntered shouldBe 2
+      writerAllocator.getAllocatedMemory shouldBe 0L
+      input.allocator.getAllocatedMemory shouldBe sourceBytes
+      sourceBuffers.map(_.refCnt()) shouldBe sourceRefs
+      input.indices.get(8) shouldBe 0
+      input.values.getObject(1).toString shouldBe input.expected(1)
+    } finally {
+      failWrites = false
+      writerAllocator.close()
+      input.close()
+    }
+  }
+
   test("direct batches preserve nested list, struct, map, and null field layouts") {
     val sourceAllocator = new RootAllocator(Long.MaxValue)
     val writerAllocator = new RootAllocator(Long.MaxValue)