This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 7374ac276e [VL] Support parquet.enable.page.index write option (#12760)
7374ac276e is described below
commit 7374ac276e990dd73ef3e440057247917538a1ef
Author: EJ Song <[email protected]>
AuthorDate: Tue Aug 18 05:53:12 2026 -0700
[VL] Support parquet.enable.page.index write option (#12760)
Add a parquet.enable.page.index write option that maps to the Velox
ParquetWriterOptions::enableWritePageIndex field
(facebookincubator/velox#18325):
---
.../velox/VeloxParquetWriterInjects.scala | 1 +
.../sql/execution/VeloxParquetWriteSuite.scala | 51 ++++++++++++++++++++++
cpp/core/config/GlutenConfig.h | 2 +
cpp/velox/utils/VeloxWriterUtils.cc | 8 ++++
docs/velox-parquet-write-configuration.md | 2 +-
.../org/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 1244aab279..45f9b0b747 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 fcd5d45e6c..9a6d423ed2 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 9c55628991..dae818c57c 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 52551787c2..68987edd99 100644
--- a/cpp/velox/utils/VeloxWriterUtils.cc
+++ b/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;
}
diff --git a/docs/velox-parquet-write-configuration.md
b/docs/velox-parquet-write-configuration.md
index 69a5d21c19..a408df8c6b 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()
</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>
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 3bb5b0958f..4971d89dd0 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."
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
