This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new b01655526d chore: remove dead Parquet parallel-IO configs (#4981)
b01655526d is described below
commit b01655526d677431c369717fac6bf4a2d2010334
Author: Andy Grove <[email protected]>
AuthorDate: Mon Jul 20 12:47:51 2026 -0600
chore: remove dead Parquet parallel-IO configs (#4981)
These configs were consumed by the JVM-side Parquet reader that has since
been removed. They are referenced nowhere in current native or JVM code
and mislead operators tuning benchmarks.
Removed configs:
- spark.comet.parquet.read.parallel.io.enabled
- spark.comet.parquet.read.parallel.io.thread-pool.size
- spark.comet.parquet.read.io.mergeRanges
- spark.comet.parquet.read.io.mergeRanges.delta
- spark.comet.parquet.read.io.adjust.readRange.skew
Also removes the stale ParquetReadSuite "test merge scan range" test,
which only asserted that the (dead) config propagated to the Hadoop conf.
Refs #4842.
---
.../main/scala/org/apache/comet/CometConf.scala | 46 ------------
.../apache/comet/parquet/ParquetReadSuite.scala | 86 ----------------------
2 files changed, 132 deletions(-)
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala
b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 119493ecfb..e49cf0db1d 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -156,52 +156,6 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
- val COMET_PARQUET_PARALLEL_IO_ENABLED: ConfigEntry[Boolean] =
- conf("spark.comet.parquet.read.parallel.io.enabled")
- .category(CATEGORY_PARQUET)
- .doc(
- "Whether to enable Comet's parallel reader for Parquet files. The
parallel reader reads " +
- "ranges of consecutive data in a file in parallel. It is faster for
large files and " +
- "row groups but uses more resources.")
- .booleanConf
- .createWithDefault(true)
-
- val COMET_PARQUET_PARALLEL_IO_THREADS: ConfigEntry[Int] =
- conf("spark.comet.parquet.read.parallel.io.thread-pool.size")
- .category(CATEGORY_PARQUET)
- .doc("The maximum number of parallel threads the parallel reader will
use in a single " +
- "executor. For executors configured with a smaller number of cores,
use a smaller number.")
- .intConf
- .createWithDefault(16)
-
- val COMET_IO_MERGE_RANGES: ConfigEntry[Boolean] =
- conf("spark.comet.parquet.read.io.mergeRanges")
- .category(CATEGORY_PARQUET)
- .doc(
- "When enabled the parallel reader will try to merge ranges of data
that are separated " +
- "by less than `comet.parquet.read.io.mergeRanges.delta` bytes.
Longer continuous reads " +
- "are faster on cloud storage.")
- .booleanConf
- .createWithDefault(true)
-
- val COMET_IO_MERGE_RANGES_DELTA: ConfigEntry[Int] =
- conf("spark.comet.parquet.read.io.mergeRanges.delta")
- .category(CATEGORY_PARQUET)
- .doc("The delta in bytes between consecutive read ranges below which the
parallel reader " +
- "will try to merge the ranges. The default is 8MB.")
- .intConf
- .createWithDefault(1 << 23) // 8 MB
-
- val COMET_IO_ADJUST_READRANGE_SKEW: ConfigEntry[Boolean] =
- conf("spark.comet.parquet.read.io.adjust.readRange.skew")
- .category(CATEGORY_PARQUET)
- .doc("In the parallel reader, if the read ranges submitted are skewed in
sizes, this " +
- "option will cause the reader to break up larger read ranges into
smaller ranges to " +
- "reduce the skew. This will result in a slightly larger number of
connections opened to " +
- "the file system but may give improved performance.")
- .booleanConf
- .createWithDefault(false)
-
val COMET_CONVERT_FROM_PARQUET_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.convert.parquet.enabled")
.category(CATEGORY_EXEC)
diff --git
a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
index 27ab885250..c294837d10 100644
--- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
@@ -1311,92 +1311,6 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
- test("test merge scan range") {
- def makeRawParquetFile(path: Path, n: Int): Seq[Option[Int]] = {
- val dictionaryPageSize = 1024
- val pageRowCount = 500
- val schemaStr =
- """
- |message root {
- | optional int32 _1(INT_16);
- | optional int32 _2;
- | optional int64 _3;
- |}
- """.stripMargin
-
- val schema = MessageTypeParser.parseMessageType(schemaStr)
- val writer = createParquetWriter(
- schema,
- path,
- dictionaryEnabled = true,
- dictionaryPageSize = dictionaryPageSize,
- pageRowCountLimit = pageRowCount)
-
- val rand = new scala.util.Random(42)
- val expected = (0 until n).map { i =>
- // use a single value for the first page, to make sure dictionary
encoding kicks in
- val value = if (i < pageRowCount) i % 8 else i
- if (rand.nextBoolean()) None
- else Some(value)
- }
-
- expected.foreach { opt =>
- val record = new SimpleGroup(schema)
- opt match {
- case Some(i) =>
- record.add(0, i.toShort)
- record.add(1, i)
- record.add(2, i.toLong)
- case _ =>
- }
- writer.write(record)
- }
-
- writer.close()
- expected
- }
-
- Seq(16, 128).foreach { batchSize =>
- Seq(1024, 1024 * 1024).foreach { mergeRangeDelta =>
- {
- withSQLConf(
- CometConf.COMET_BATCH_SIZE.key -> batchSize.toString,
- CometConf.COMET_IO_MERGE_RANGES.key -> "true",
- CometConf.COMET_IO_MERGE_RANGES_DELTA.key ->
mergeRangeDelta.toString) {
- withTempDir { dir =>
- val path = new Path(dir.toURI.toString, "part-r-0.parquet")
- val expected = makeRawParquetFile(path, 10000)
- val schema = StructType(
- Seq(StructField("_1", ShortType, true), StructField("_3",
LongType, true)))
- readParquetFile(path.toString, Some(schema)) { df =>
- {
- // CometScanExec calls sessionState.newHadoopConfWithOptions
which copies
- // the sqlConf and some additional options to the hadoopConf
and then
- // uses the result to create the inputRDD
(https://github.com/apache/datafusion-comet/blob/3783faaa01078a35bee93b299368f8c72869198d/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala#L181).
- // We don't have access to the created hadoop Conf, but can
confirm that the
- // result does contain the correct configuration
- assert(
- df.sparkSession.sessionState
- .newHadoopConfWithOptions(Map.empty)
- .get(CometConf.COMET_IO_MERGE_RANGES_DELTA.key)
- .equals(mergeRangeDelta.toString))
- checkAnswer(
- df,
- expected.map {
- case None =>
- Row(null, null)
- case Some(i) =>
- Row(i.toShort, i.toLong)
- })
- }
- }
- }
- }
- }
- }
- }
- }
-
def testScanner(cometEnabled: String, scanner: String, v1: Option[String] =
None): Unit = {
withSQLConf(
CometConf.COMET_ENABLED.key -> cometEnabled,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
