Skip to content

chore: remove dead Parquet parallel-IO configs #4981

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mbutrovich merged 1 commit into from
Jul 20, 2026
Merged

chore: remove dead Parquet parallel-IO configs #4981

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 0 additions & 46 deletions spark/src/main/scala/org/apache/comet/CometConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
86 changes: 0 additions & 86 deletions spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading