judahrand opened a new issue, #7541: URL: https://github.com/apache/datafusion-comet/issues/7541
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](https://github.com/apache/datafusion/pull/26743). Could we backport it or update to a dependency containing it? ## Reproducer With Spark 4.1.3, Java 21, and the Comet JAR installed in `$SPARK_HOME/jars`, save this as `inline_repro.py`: ```python 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() ``` ```bash "$SPARK_HOME/bin/spark-submit" --master 'local[1]' --driver-memory 1g \ inline_repro.py ``` The 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: ```diff 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
