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/10/14 16:25:04 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6860-6d33dd80518afc56fbc889b3232e69794ef9a1bf created (now 2123b747a6)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6860-6d33dd80518afc56fbc889b3232e69794ef9a1bf
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 2123b747a6 perf: generate the in-memory cache reader's class once per executor (#6860)
This branch includes the following new commits:
new 2123b747a6 perf: generate the in-memory cache reader's class once per executor (#6860)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: perf: generate the in-memory cache reader's class once per executor (#6860)
Posted by gi...@apache.org.
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-6860-6d33dd80518afc56fbc889b3232e69794ef9a1bf
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 2123b747a63eb74d613f710ef095fc9d38a2dc07
Author: Andy Grove <ag...@apache.org>
AuthorDate: Wed Oct 14 16:24:40 2026 +0000
perf: generate the in-memory cache reader's class once per executor (#6860)
* perf: generate the columnar-to-row projection once per executor
CometColumnarToRowExec.doExecute built its UnsafeProjection with
UnsafeProjection.create in every partition, which regenerates the
projection's Java source each time. The plan runs that path whenever it
is outside whole-stage codegen, which a schema with more nested fields
than spark.sql.codegen.maxFields always is. For 100 columns nested three
levels deep, generating the source took about 19 ms per partition, half
the time of a 201-partition JVM shuffle of 1,000 rows.
Cache the compiled class per executor by column layout and create a new
instance for each partition.
* ci: run CometUnsafeProjectionSuite on Linux and macOS
* perf: generate the cache reader's class once per executor
Spark reads Comet's in-memory cache through CachedBatchRowIterator, once
per partition, whenever it does not read the cache as batches. That is
always the case for a schema with more fields than
spark.sql.codegen.maxFields. Every call generated the reader's source
again; Spark's compiled class cache only saved the compile. For a
1,500-column table that was about 18 ms per partition, and for 100
columns nested three levels deep about 78 ms.
Keep the class per executor by column layout, as CometUnsafeProjection
does for CometColumnarToRowExec, with the cache both now use moved into
GeneratedClassCache. Each reader gets its own batches through its own
copy of the references. The fallback for a method over the huge-method
limit builds its projection through CometUnsafeProjection.
* fix: bound the generated class caches by spark.sql.codegen.cache.maxEntries
GeneratedClassCache kept the 100 most recently used classes, the default
of spark.sql.codegen.cache.maxEntries, without reading that conf, so
lowering it to bound metaspace left Comet's two caches at 100 each. Read
it when the cache is created, as Spark's CodeGenerator does for its own
cache, and drop the constructor parameter and constant nothing passed.
---
.../spark/sql/comet/CometUnsafeProjection.scala | 59 +++++----------
.../spark/sql/comet/GeneratedClassCache.scala | 87 ++++++++++++++++++++++
.../execution/arrow/CachedBatchRowIterator.scala | 85 ++++++++++++++++-----
.../sql/comet/CometUnsafeProjectionSuite.scala | 19 ++++-
.../arrow/CachedBatchRowIteratorSuite.scala | 36 +++++++++
5 files changed, 224 insertions(+), 62 deletions(-)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometUnsafeProjection.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometUnsafeProjection.scala
index d787460a8b..98847dfa2d 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometUnsafeProjection.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometUnsafeProjection.scala
@@ -19,13 +19,8 @@
package org.apache.spark.sql.comet
-import java.util.{LinkedHashMap => JLinkedHashMap, Map => JMap}
-import java.util.concurrent.atomic.AtomicLong
-
import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, CodeGeneratorWithInterpretedFallback, InterpretedUnsafeProjection, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.codegen.{CodeAndComment, CodeFormatter, CodegenContext, CodeGenerator, GeneratedClass, GenerateUnsafeProjection}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.DataType
/**
* Creates the projection that copies each row of a batch into an UnsafeRow, as
@@ -47,25 +42,16 @@ import org.apache.spark.sql.types.DataType
private[comet] object CometUnsafeProjection
extends CodeGeneratorWithInterpretedFallback[Seq[BoundReference], UnsafeProjection] {
- /**
- * What the generated source depends on: the type and nullability of each column, and the method
- * size at which `CodegenContext.splitExpressions` splits the field writes.
- */
- private case class Layout(columns: Seq[(DataType, Boolean)], methodSplitThreshold: Int)
-
- /** The same bound as Spark's compiled class cache, `spark.sql.codegen.cache.maxEntries`. */
- private[comet] val MaxCachedClasses = 100
-
- /** Least recently used first. Guarded by `classes.synchronized`. */
- private val classes = new JLinkedHashMap[Layout, GeneratedClass](16, 0.75f, true) {
- override def removeEldestEntry(eldest: JMap.Entry[Layout, GeneratedClass]): Boolean =
- size() > MaxCachedClasses
+ /** A compiled projection class and the objects its instances reference. */
+ private case class Generated(generatedClass: GeneratedClass, references: Array[Any]) {
+ def newProjection(): UnsafeProjection =
+ generatedClass.generate(references).asInstanceOf[UnsafeProjection]
}
- private val classesGenerated = new AtomicLong(0)
+ private val classes = new GeneratedClassCache[ColumnLayout, Generated]()
/** How many projection classes this executor has generated, for tests. */
- private[comet] def generatedClassCount: Long = classesGenerated.get()
+ private[comet] def generatedClassCount: Long = classes.generatedCount
/** A projection of rows with the columns of `output` to UnsafeRows. */
def create(output: Seq[Attribute]): UnsafeProjection =
@@ -74,34 +60,24 @@ private[comet] object CometUnsafeProjection
})
override protected def createCodeGeneratedObject(
- columns: Seq[BoundReference]): UnsafeProjection = {
- val layout =
- Layout(columns.map(c => (c.dataType, c.nullable)), SQLConf.get.methodSplitThreshold)
- val cached = classes.synchronized(classes.get(layout))
- if (cached != null) {
- cached.generate(Array.empty[Any]).asInstanceOf[UnsafeProjection]
- } else {
- // Generated outside the lock: tasks that miss on the same layout at once each generate
- // it, as every task does with UnsafeProjection.create.
- val (generated, references) = generate(columns)
- // A stored class is instantiated without references, so store only one that needs none.
- // GenerateUnsafeProjection references no objects for bound columns.
- if (references.isEmpty) {
- val _ = classes.synchronized(classes.putIfAbsent(layout, generated))
+ columns: Seq[BoundReference]): UnsafeProjection =
+ classes
+ .getOrGenerate(ColumnLayout.of(columns)) {
+ val generated = generate(columns)
+ // Instances of a shared class share its references, so share only a class that has
+ // none. GenerateUnsafeProjection references no objects for bound columns.
+ (generated, generated.references.isEmpty)
}
- generated.generate(references).asInstanceOf[UnsafeProjection]
- }
- }
+ .newProjection()
override protected def createInterpretedObject(columns: Seq[BoundReference]): UnsafeProjection =
InterpretedUnsafeProjection.createProjection(columns)
/**
* Generates and compiles the class that `GenerateUnsafeProjection.create` does, from the same
- * template, which is repeated here because that method returns only an instance. Returns the
- * class with the objects its instances reference.
+ * template, which is repeated here because that method returns only an instance.
*/
- private def generate(columns: Seq[BoundReference]): (GeneratedClass, Array[Any]) = {
+ private def generate(columns: Seq[BoundReference]): Generated = {
val ctx = new CodegenContext
val eval = GenerateUnsafeProjection.createCode(ctx, columns)
val body =
@@ -140,7 +116,6 @@ private[comet] object CometUnsafeProjection
val code = CodeFormatter.stripOverlappingComments(
new CodeAndComment(body, ctx.getPlaceHolderToComments()))
val (generatedClass, _) = CodeGenerator.compile(code)
- classesGenerated.incrementAndGet()
- (generatedClass, ctx.references.toArray)
+ Generated(generatedClass, ctx.references.toArray)
}
}
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/GeneratedClassCache.scala b/spark/src/main/scala/org/apache/spark/sql/comet/GeneratedClassCache.scala
new file mode 100644
index 0000000000..a9d3b160e2
--- /dev/null
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/GeneratedClassCache.scala
@@ -0,0 +1,87 @@
+/*
+ * 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.comet
+
+import java.util.{LinkedHashMap => JLinkedHashMap, Map => JMap}
+import java.util.concurrent.atomic.AtomicLong
+
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.DataType
+
+/**
+ * What code generated over a list of columns depends on: the type and nullability of each column,
+ * and the method size at which `CodegenContext.splitExpressions` splits the code for them.
+ */
+private[comet] case class ColumnLayout(
+ columns: Seq[(DataType, Boolean)],
+ methodSplitThreshold: Int)
+
+private[comet] object ColumnLayout {
+
+ /** The layout of `columns` under the current `spark.sql.codegen.methodSplitThreshold`. */
+ def of(columns: Seq[Expression]): ColumnLayout =
+ ColumnLayout(columns.map(c => (c.dataType, c.nullable)), SQLConf.get.methodSplitThreshold)
+}
+
+/**
+ * Generated classes kept for the life of an executor, under a key for what their source depends
+ * on.
+ *
+ * Spark caches a compiled class under its source, so an operator that generates code in every
+ * partition still generates the source every time, which for a wide or deeply nested schema can
+ * take longer than the partition's rows. Keeping the class under a cheaper key generates it once.
+ * Like Spark's cache, it holds the `spark.sql.codegen.cache.maxEntries` most recently used
+ * values.
+ */
+private[comet] class GeneratedClassCache[K, V <: AnyRef] {
+
+ /** A static conf, so read once, as Spark reads it for its own cache. Zero keeps nothing. */
+ private val maxEntries = SQLConf.get.codegenCacheMaxEntries
+
+ /** Least recently used first. Guarded by `entries.synchronized`. */
+ private val entries = new JLinkedHashMap[K, V](16, 0.75f, true) {
+ override def removeEldestEntry(eldest: JMap.Entry[K, V]): Boolean = size() > maxEntries
+ }
+
+ private val generated = new AtomicLong(0)
+
+ /** How many values this cache has generated, for tests. */
+ def generatedCount: Long = generated.get()
+
+ /**
+ * The value for `key`, generated if absent. `generate` returns the value and whether other
+ * callers may share it. It runs outside the lock, so callers that miss on the same key at once
+ * each generate the value, as they would without the cache.
+ */
+ def getOrGenerate(key: K)(generate: => (V, Boolean)): V = {
+ val cached = entries.synchronized(entries.get(key))
+ if (cached != null) {
+ cached
+ } else {
+ val (value, shareable) = generate
+ generated.incrementAndGet()
+ if (shareable) {
+ val _ = entries.synchronized(entries.putIfAbsent(key, value))
+ }
+ value
+ }
+ }
+}
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIterator.scala b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIterator.scala
index 82f73d2590..59181c28ed 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIterator.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIterator.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, BoundReference, CodeGeneratorWithInterpretedFallback, InterpretedUnsafeProjection, LeafExpression, UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.codegen._
import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.comet.{ColumnLayout, CometUnsafeProjection, GeneratedClassCache}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
@@ -37,6 +38,12 @@ import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
* splits the field writes of a wide projection into methods of bounded size, as it does for any
* Spark projection. If a generated method still exceeds the huge-method limit, the reader backs
* off the way WholeStageCodegenExec does, here to Spark's projection of each batch row.
+ *
+ * Spark reads a cache through this path, once per partition, when it does not read the cache as
+ * batches, which is always the case for a schema with more fields than
+ * `spark.sql.codegen.maxFields`, nested fields included. For such a schema generating the source
+ * can take longer than reading a partition's rows, so the reader's class is generated once per
+ * executor for each column layout.
*/
private[arrow] class CachedBatchRowIterator(attributes: Seq[Attribute])
extends CodeGeneratorWithInterpretedFallback[Iterator[ColumnarBatch], Iterator[InternalRow]] {
@@ -47,6 +54,35 @@ private[arrow] class CachedBatchRowIterator(attributes: Seq[Attribute])
override protected def createCodeGeneratedObject(
batches: Iterator[ColumnarBatch]): Iterator[InternalRow] = {
+ val reader = CachedBatchRowIterator.readers.getOrGenerate(ColumnLayout.of(attributes)) {
+ val generated = generate()
+ // Instances of a shared class share its references other than the batches, so share only
+ // a class that has no others.
+ (generated, generated.references.length == 1)
+ }
+ // Honor spark.sql.codegen.hugeMethodLimit as whole-stage codegen does, but never go above
+ // HotSpot's own limit: the config defaults to the largest method the JVM accepts, while this
+ // runs once per row and HotSpot never JIT-compiles a method longer than
+ // DEFAULT_JVM_HUGE_METHOD_LIMIT bytes.
+ val limit =
+ math.min(SQLConf.get.hugeMethodLimit, CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT)
+ if (reader.maxMethodCodeSize > limit) {
+ logInfo(
+ s"Generated cache reader for ${attributes.length} columns has a " +
+ s"${reader.maxMethodCodeSize}-byte method, above the $limit-byte limit; " +
+ "projecting cached rows with UnsafeProjection instead")
+ new ProjectedRows(batches, CometUnsafeProjection.create(attributes))
+ } else {
+ reader.newIterator(batches)
+ }
+ }
+
+ override protected def createInterpretedObject(
+ batches: Iterator[ColumnarBatch]): Iterator[InternalRow] =
+ new ProjectedRows(batches, InterpretedUnsafeProjection.createProjection(fields))
+
+ /** Generates and compiles the reader's class, which takes its batches as a reference. */
+ private def generate(): CachedBatchRowIterator.GeneratedReader = {
val ctx = new CodegenContext
val vectorClass = classOf[ColumnVector].getName
val batchClass = classOf[ColumnarBatch].getName
@@ -62,7 +98,9 @@ private[arrow] class CachedBatchRowIterator(attributes: Seq[Attribute])
// With ctx.currentVars unset, GenerateUnsafeProjection splits the field writes into methods
// that take the input row as their argument. The reads above ignore it.
val projection = GenerateUnsafeProjection.createCode(ctx, reads)
- val batchesRef = ctx.addReferenceObj("batches", batches, "scala.collection.Iterator")
+ // Each instance gets its own batches, set in a copy of the references.
+ val batchesIndex = ctx.references.length
+ val batchesRef = ctx.addReferenceObj("batches", null, "scala.collection.Iterator")
val code = s"""
public Object generate(Object[] references) {
return new SpecificCachedBatchRowIterator(references);
@@ -105,26 +143,37 @@ private[arrow] class CachedBatchRowIterator(attributes: Seq[Attribute])
"""
val (compiled, stats) =
CodeGenerator.compile(new CodeAndComment(code, ctx.getPlaceHolderToComments()))
- // Honor spark.sql.codegen.hugeMethodLimit as whole-stage codegen does, but never go above
- // HotSpot's own limit: the config defaults to the largest method the JVM accepts, while this
- // runs once per row and HotSpot never JIT-compiles a method longer than
- // DEFAULT_JVM_HUGE_METHOD_LIMIT bytes.
- val limit =
- math.min(SQLConf.get.hugeMethodLimit, CodeGenerator.DEFAULT_JVM_HUGE_METHOD_LIMIT)
- if (stats.maxMethodCodeSize > limit) {
- logInfo(
- s"Generated cache reader for ${attributes.length} columns has a " +
- s"${stats.maxMethodCodeSize}-byte method, above the $limit-byte limit; " +
- "projecting cached rows with UnsafeProjection instead")
- new ProjectedRows(batches, UnsafeProjection.create(fields))
- } else {
- compiled.generate(ctx.references.toArray).asInstanceOf[Iterator[InternalRow]]
+ CachedBatchRowIterator.GeneratedReader(
+ compiled,
+ stats.maxMethodCodeSize,
+ ctx.references.toArray,
+ batchesIndex)
+ }
+}
+
+private[arrow] object CachedBatchRowIterator {
+
+ /**
+ * A compiled reader class, the size of its largest method, and the objects its instances
+ * reference, with a null slot at `batchesIndex` for each instance's batches.
+ */
+ private case class GeneratedReader(
+ generatedClass: GeneratedClass,
+ maxMethodCodeSize: Int,
+ references: Array[Any],
+ batchesIndex: Int) {
+
+ def newIterator(batches: Iterator[ColumnarBatch]): Iterator[InternalRow] = {
+ val instanceReferences = references.clone()
+ instanceReferences(batchesIndex) = batches
+ generatedClass.generate(instanceReferences).asInstanceOf[Iterator[InternalRow]]
}
}
- override protected def createInterpretedObject(
- batches: Iterator[ColumnarBatch]): Iterator[InternalRow] =
- new ProjectedRows(batches, InterpretedUnsafeProjection.createProjection(fields))
+ private val readers = new GeneratedClassCache[ColumnLayout, GeneratedReader]()
+
+ /** How many reader classes this executor has generated, for tests. */
+ private[arrow] def generatedClassCount: Long = readers.generatedCount
}
/**
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/CometUnsafeProjectionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/CometUnsafeProjectionSuite.scala
index 07c6371f3e..c592d5b0a3 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/CometUnsafeProjectionSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/CometUnsafeProjectionSuite.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.{CatalystTypeConverters, InternalRow}
import org.apache.spark.sql.catalyst.expressions.{AttributeReference, InterpretedUnsafeProjection, UnsafeProjection}
import org.apache.spark.sql.execution.WholeStageCodegenExec
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
-import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.types._
class CometUnsafeProjectionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
@@ -135,7 +135,7 @@ class CometUnsafeProjectionSuite extends CometTestBase with AdaptiveSparkPlanHel
test("keeps the most recently used classes") {
withCodegenOnly() {
- val layouts = Seq.fill(CometUnsafeProjection.MaxCachedClasses + 1) {
+ val layouts = Seq.fill(SQLConf.get.codegenCacheMaxEntries + 1) {
val name = s"f_${UUID.randomUUID().toString.replace("-", "")}"
Seq(AttributeReference("c", StructType(Seq(StructField(name, IntegerType))))())
}
@@ -148,6 +148,21 @@ class CometUnsafeProjectionSuite extends CometTestBase with AdaptiveSparkPlanHel
}
}
+ test("keeps as many classes as spark.sql.codegen.cache.maxEntries") {
+ Seq(0, 2).foreach { maxEntries =>
+ // A static conf, so set where the cache reads it rather than in the session.
+ val conf = new SQLConf
+ conf.setConfString(StaticSQLConf.CODEGEN_CACHE_MAX_ENTRIES.key, maxEntries.toString)
+ val cache = SQLConf.withExistingConf(conf)(new GeneratedClassCache[Int, String])
+ def get(key: Int): String = cache.getOrGenerate(key)((key.toString, true))
+ Seq(1, 2, 3).foreach(get)
+ val before = cache.generatedCount
+ // Most recently used first, so every key still cached is read before a miss evicts one.
+ Seq(3, 2, 1).foreach(get)
+ assert(cache.generatedCount == before + 3 - maxEntries)
+ }
+ }
+
test("columnar to row outside whole-stage codegen reuses the generated class") {
withTempPath { dir =>
val path = dir.getCanonicalPath
diff --git a/spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIteratorSuite.scala b/spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIteratorSuite.scala
index 6d2a5385e8..108694a7b5 100644
--- a/spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIteratorSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/comet/execution/arrow/CachedBatchRowIteratorSuite.scala
@@ -133,6 +133,42 @@ class CachedBatchRowIteratorSuite extends AnyFunSuite {
}
}
+ if (mode == "CODEGEN_ONLY") {
+ test(s"$mode: readers of one layout share a class and read their own batches") {
+ withMode() {
+ val (attributes, first) = wideInput(41)
+ val (_, second) = wideInput(41)
+ try {
+ val firstRows = new CachedBatchRowIterator(attributes).createObject(first.iterator)
+ val generated = CachedBatchRowIterator.generatedClassCount
+ val secondRows = new CachedBatchRowIterator(attributes).createObject(second.iterator)
+ assert(CachedBatchRowIterator.generatedClassCount == generated)
+ assert(firstRows.getClass eq secondRows.getClass)
+ // Draining the second reader leaves the first one's batches unread.
+ checkWideRows(attributes, secondRows)
+ checkWideRows(attributes, firstRows)
+ } finally (first ++ second).foreach(_.close())
+ }
+ }
+
+ test(s"$mode: the huge-method limit applies to a reader class generated earlier") {
+ val (attributes, batches) = wideInput(43)
+ try {
+ withMode(SQLConf.WHOLESTAGE_HUGE_METHOD_LIMIT.key -> "1") {
+ val rows = new CachedBatchRowIterator(attributes).createObject(Iterator.empty)
+ assert(rows.isInstanceOf[ProjectedRows])
+ }
+ val generated = CachedBatchRowIterator.generatedClassCount
+ withMode() {
+ val rows = new CachedBatchRowIterator(attributes).createObject(batches.iterator)
+ assert(!rows.isInstanceOf[ProjectedRows])
+ checkWideRows(attributes, rows)
+ }
+ assert(CachedBatchRowIterator.generatedClassCount == generated)
+ } finally batches.foreach(_.close())
+ }
+ }
+
test(
s"$mode: a generated method above the huge-method limit falls back to UnsafeProjection") {
// Below any generated method, as if the reader had grown past HotSpot's limit.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org