[VL] Support parquet.enable.page.index write option

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

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 {

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";

cpp/velox/utils/VeloxWriterUtils.cc

@@ -129,6 +129,14 @@ std::shared_ptr<facebook::velox::dwio::common::WriterOptions> 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;
 }

docs/velox-parquet-write-configuration.md

@@ -54,7 +54,7 @@ df.write.option("parquet.block.rows").save()
 	</tr>
 	<tr>
 		<td><code>page_index</code></td>
-		<td></td><td></td><td>false</td><td></td>
+		<td></td><td></td><td>true</td><td>parquet.enable.page.index</td>
 	</tr>
 	<tr>
 		<td><code>decimal_as_integer</code></td>

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."