Repository navigation
ANY_VALUE memory regression in Comet 1.2.0 #7541
Description
Comet 1.2.0 fails ANY_VALUE(array<string>) with an oversized memory reservation.
Spark lowers this to first, which uses DataFusion 55.2.0's
TrivialFirstValueAccumulator::merge_batch. That method retains the input's
backing buffers and counts them repeatedly across groups.
This is fixed by apache/datafusion#26743.
Could we update to a dependency containing it when released?
Reproducer
With Spark 4.1.3, Java 21, and the Comet JAR installed in $SPARK_HOME/jars,
save this as inline_repro.py:
from pyspark.sql import SparkSession
spark = (
SparkSession.builder.config("spark.ui.enabled", "false")
.config("spark.sql.adaptive.enabled", "false")
.config("spark.sql.shuffle.partitions", "2")
.config("spark.plugins", "org.apache.spark.CometPlugin")
.config(
"spark.shuffle.manager",
"org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
)
.config("spark.comet.enabled", "true")
.config("spark.comet.exec.enabled", "true")
.config("spark.comet.shuffle.enabled", "true")
.config("spark.comet.nativeLoadRequired", "true")
.config("spark.comet.exec.aggregate.enabled", "true")
.config("spark.memory.offHeap.enabled", "true")
.config("spark.memory.offHeap.size", "128m")
.getOrCreate()
)
try:
path = "/tmp/comet-any-value-input"
spark.range(256, numPartitions=4).selectExpr(
"CAST(id AS STRING) AS user_id",
"transform(sequence(0, 199), x -> concat(lpad(CAST(id AS STRING), 8, '0'), ':', lpad(CAST(x AS STRING), 4, '0'))) AS values",
).write.mode("overwrite").parquet(path)
spark.read.parquet(path).createOrReplaceTempView("input")
result = spark.sql(
"SELECT user_id, ANY_VALUE(values) AS values FROM input GROUP BY user_id"
)
result.explain()
rows = result.collect()
expected = {str(i): [f"{i:08d}:{j:04d}" for j in range(200)] for i in range(256)}
assert len(rows) == 256
assert {row.user_id: row["values"] for row in rows} == expected
finally:
spark.stop()"$SPARK_HOME/bin/spark-submit" --master 'local[1]' --driver-memory 1g \
inline_repro.pyThe script sets a 128 MiB native pool and two shuffle partitions. It fails in
OrderedFinalAggregateStream, requesting 80,775,440 bytes against a
67,108,864-byte share. The strings contain just 650 KiB of data.
The same script succeeds with Comet 1.1.0, or with
spark.comet.exec.aggregate.enabled set to false on 1.2.0. It checks every
output array.
Fix verification
Focused native tests use DataFusion 55.2.0 and Arrow 59.3.0. For the same
selected array, update_batch reports 3,832 bytes while merge_batch reports
1,311,296 bytes. Adding the upstream fix makes both report 3,832 bytes:
self.first = ScalarValue::try_from_array(first, 0)?;
+self.first.compact();
self.is_set = true;Both native regression tests fail before this change and pass afterward,
with unchanged output values. A rebuilt Comet JAR has not yet been tested.
Activity
[-]Backport the FIRST_VALUE merge compaction fix for array-valued ANY_VALUE[/-][+]ANY_VALUE memory regression in Comet 1.2.0[/+]kazuyukitanimura commentedon Nov 9, 2026
This will be closed by #7249