diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index cd971d4f92..6f3b578334 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,