From f998f93d55fd20bd2e15dc4e72970d62d4db1244 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Wed, 12 Aug 2026 11:15:52 -0700 Subject: [PATCH] [VL] Support parquet.enable.page.index write option Add a parquet.enable.page.index write option that maps to the Velox ParquetWriterOptions::enableWritePageIndex field (facebookincubator/velox#18325): - GlutenConfig (Scala/C++): define the parquet.enable.page.index key. - VeloxParquetWriterInjects: forward the option to native write config. - VeloxWriterUtils::makeParquetWriteOption: set enableWritePageIndex, defaulting to on when the option is unset. When enabled, the writer emits the Parquet column index and offset index (page index); when disabled, they are omitted. The option defaults to true to match Spark/parquet-mr, which write the page index by default (SPARK-26345); Velox's writer leaves it opt-in. Set parquet.enable.page.index=false to turn it off. Documented in the parquet write configuration table and covered by a VeloxParquetWriteSuite test that asserts the ColumnIndex/OffsetIndex presence tracks the option (enabled, disabled, and unset/default) and that data round-trips. Co-Authored-By: Claude Opus 4.8 --- .../velox/VeloxParquetWriterInjects.scala | 1 + .../execution/VeloxParquetWriteSuite.scala | 51 +++++++++++++++++++ cpp/core/config/GlutenConfig.h | 2 + cpp/velox/utils/VeloxWriterUtils.cc | 8 +++ docs/velox-parquet-write-configuration.md | 2 +- .../apache/gluten/config/GlutenConfig.scala | 1 + 6 files changed, 64 insertions(+), 1 deletion(-) diff --git a/backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxParquetWriterInjects.scala b/backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxParquetWriterInjects.scala index 1244aab279d..45f9b0b7478 100644 --- a/backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxParquetWriterInjects.scala +++ b/backends-velox/src/main/scala/org/apache/spark/sql/execution/datasources/velox/VeloxParquetWriterInjects.scala @@ -55,6 +55,7 @@ class VeloxParquetWriterInjects extends VeloxFormatWriterInjects { GlutenConfig.PARQUET_ZSTD_COMPRESSION_LEVEL, GlutenConfig.PARQUET_DATAPAGE_SIZE, GlutenConfig.PARQUET_ENABLE_DICTIONARY, + GlutenConfig.PARQUET_ENABLE_PAGE_INDEX, GlutenConfig.PARQUET_WRITER_VERSION ).foreach(key => options.get(key).foreach(sparkOptions.put(key, _))) sparkOptions.asJava diff --git a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala index fcd5d45e6c4..9a6d423ed28 100644 --- a/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala +++ b/backends-velox/src/test/scala/org/apache/spark/sql/execution/VeloxParquetWriteSuite.scala @@ -306,6 +306,57 @@ class VeloxParquetWriteSuite extends VeloxWholeStageTransformerSuite with WriteU checkAnswer(parquetDf, spark.range(100).toDF("id")) } } + + test("test write parquet with page index enabled/disabled/default") { + Seq(Some(true), Some(false), None).foreach { + enablePageIndex => + withTempPath { + f => + val writer = spark + .range(0, 100000, 1, 1) + .selectExpr("id", "cast(id % 100 as int) as v") + .write + .format("parquet") + .option(GlutenConfig.PARQUET_DATAPAGE_SIZE, (4 * 1024).toString) + enablePageIndex.foreach( + v => writer.option(GlutenConfig.PARQUET_ENABLE_PAGE_INDEX, v.toString)) + writer.save(f.getCanonicalPath) + + val expectPageIndex = enablePageIndex.getOrElse(true) + val parquetFiles = f.list((_, name) => name.contains("parquet")) + assert(parquetFiles.nonEmpty) + val indexRefs = parquetFiles.flatMap { + file => + val path = new Path(f.getCanonicalPath, file) + val in = HadoopInputFile.fromPath(path, spark.sessionState.newHadoopConf()) + Utils.tryWithResource(ParquetFileReader.open(in)) { + reader => + reader.getFooter.getBlocks.asScala.flatMap { + block => + block.getColumns.asScala.map { + col => + ( + col.getColumnIndexReference != null, + col.getOffsetIndexReference != null) + } + } + } + } + val hasColumnIndex = indexRefs.exists(_._1) + val hasOffsetIndex = indexRefs.exists(_._2) + assert( + hasColumnIndex == expectPageIndex, + s"expected column index present=$expectPageIndex but found $hasColumnIndex") + assert( + hasOffsetIndex == expectPageIndex, + s"expected offset index present=$expectPageIndex but found $hasOffsetIndex") + + checkAnswer( + spark.read.parquet(f.getCanonicalPath), + spark.range(0, 100000, 1, 1).selectExpr("id", "cast(id % 100 as int) as v")) + } + } + } } class VeloxParquetWriteHadoopConfSuite extends VeloxWholeStageTransformerSuite with WriteUtils { diff --git a/cpp/core/config/GlutenConfig.h b/cpp/core/config/GlutenConfig.h index 9c556289917..dae818c57cd 100644 --- a/cpp/core/config/GlutenConfig.h +++ b/cpp/core/config/GlutenConfig.h @@ -70,6 +70,8 @@ const std::string kParquetDataPageSize = "parquet.page.size"; const std::string kParquetEnableDictionary = "parquet.enable.dictionary"; +const std::string kParquetEnablePageIndex = "parquet.enable.page.index"; + const std::string kParquetWriterVersion = "parquet.writer.version"; const std::string kParquetCompressionCodec = "spark.sql.parquet.compression.codec"; diff --git a/cpp/velox/utils/VeloxWriterUtils.cc b/cpp/velox/utils/VeloxWriterUtils.cc index 52551787c25..68987edd990 100644 --- a/cpp/velox/utils/VeloxWriterUtils.cc +++ b/cpp/velox/utils/VeloxWriterUtils.cc @@ -129,6 +129,14 @@ std::shared_ptr makeParquetWriteOp parquetOptions->enableDictionary = false; } } + // Write the Parquet page index (column index + offset index) by default to match + // Spark/parquet-mr (SPARK-26345); Velox's writer leaves it opt-in. Disable with + // parquet.enable.page.index=false. + bool enableWritePageIndex = true; + if (auto it = sparkConfs.find(kParquetEnablePageIndex); it != sparkConfs.end()) { + enableWritePageIndex = boost::iequals(it->second, "true"); + } + parquetOptions->enableWritePageIndex = enableWritePageIndex; writeOption->formatSpecificOptions = std::move(parquetOptions); return writeOption; } diff --git a/docs/velox-parquet-write-configuration.md b/docs/velox-parquet-write-configuration.md index 69a5d21c19a..a408df8c6bf 100644 --- a/docs/velox-parquet-write-configuration.md +++ b/docs/velox-parquet-write-configuration.md @@ -54,7 +54,7 @@ df.write.option("parquet.block.rows").save() page_index - false + trueparquet.enable.page.index decimal_as_integer diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 3bb5b0958f3..4971d89dd0b 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala @@ -416,6 +416,7 @@ object GlutenConfig extends ConfigRegistry { val PARQUET_ZSTD_COMPRESSION_LEVEL: String = "parquet.compression.codec.zstd.level" val PARQUET_DATAPAGE_SIZE: String = "parquet.page.size" val PARQUET_ENABLE_DICTIONARY: String = "parquet.enable.dictionary" + val PARQUET_ENABLE_PAGE_INDEX: String = "parquet.enable.page.index" val PARQUET_WRITER_VERSION: String = "parquet.writer.version" // Hadoop config val HADOOP_PREFIX = "spark.hadoop."