This is an automated email from the ASF dual-hosted git repository.
jackylee-ch 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 79f0040e9f [VL] Support storeDecimalAsInteger for parquet native write
compatible with writeLegacyFormat (#12213)
79f0040e9f is described below
commit 79f0040e9fd6fad7e4f3b3d28132fc498df03d5e
Author: lifulong <[email protected]>
AuthorDate: Tue Jun 23 18:31:40 2026 +0800
[VL] Support storeDecimalAsInteger for parquet native write compatible with
writeLegacyFormat (#12213)
---
.../velox/VeloxParquetWriterInjects.scala | 3 ++
.../sql/execution/VeloxParquetWriteSuite.scala | 36 ++++++++++++++++++++++
cpp/core/config/GlutenConfig.h | 8 +++++
cpp/velox/utils/VeloxWriterUtils.cc | 14 +++++++++
.../org/apache/gluten/config/GlutenConfig.scala | 1 +
5 files changed, 62 insertions(+)
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 14f1c6d63b..1244aab279 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
@@ -31,6 +31,9 @@ class VeloxParquetWriterInjects extends
VeloxFormatWriterInjects {
// i.e., compression, block size, block rows.
val sparkOptions = new mutable.HashMap[String, String]()
sparkOptions.put(SQLConf.PARQUET_COMPRESSION.key, compressionCodec)
+ sparkOptions.put(
+ SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key,
+ SQLConf.get.writeLegacyParquetFormat.toString)
val blockSize = options.getOrElse(
GlutenConfig.PARQUET_BLOCK_SIZE,
GlutenConfig.get.columnarParquetWriteBlockSize.toString)
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 9de24103b0..fcd5d45e6c 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
@@ -21,11 +21,13 @@ import
org.apache.gluten.execution.VeloxWholeStageTransformerSuite
import org.apache.spark.SparkConf
import org.apache.spark.sql.Row
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.Utils
import org.apache.hadoop.fs.Path
import org.apache.parquet.hadoop.ParquetFileReader
import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.apache.parquet.schema.PrimitiveType
import java.io.File
@@ -74,6 +76,40 @@ class VeloxParquetWriteSuite extends
VeloxWholeStageTransformerSuite with WriteU
}
}
+ test("test write parquet decimal with writeLegacyFormat") {
+ // writeLegacyFormat details see `VeloxWriterUtils.cc`
+ Seq(
+ false -> PrimitiveType.PrimitiveTypeName.INT32,
+ true -> PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
+ .foreach {
+ case (legacyFormat, expectedType) =>
+ withSQLConf(SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key ->
legacyFormat.toString) {
+ withTempPath {
+ f =>
+ val path = f.getCanonicalPath
+ val expected = spark.sql("SELECT CAST(123.456 AS DECIMAL(8,
3)) AS value")
+ checkNativeWrite(
+ s"INSERT OVERWRITE DIRECTORY '$path' USING PARQUET " +
+ "SELECT CAST(123.456 AS DECIMAL(8, 3)) AS value")
+ val parquetFiles = f.list((_, name) =>
name.contains("parquet"))
+ assert(parquetFiles.nonEmpty)
+ parquetFiles.foreach {
+ file =>
+ val filePath = new Path(path, file)
+ val in = HadoopInputFile.fromPath(filePath,
spark.sessionState.newHadoopConf())
+ Utils.tryWithResource(ParquetFileReader.open(in)) {
+ reader =>
+ val physicalType =
reader.getFooter.getFileMetaData.getSchema
+
.getFields.get(0).asPrimitiveType.getPrimitiveTypeName
+ assert(physicalType == expectedType)
+ }
+ }
+ checkAnswer(spark.read.parquet(path), expected)
+ }
+ }
+ }
+ }
+
test("test write parquet with compression codec") {
// compression codec details see `VeloxParquetDatasource.cc`
Seq("snappy", "gzip", "zstd", "lz4", "none", "uncompressed")
diff --git a/cpp/core/config/GlutenConfig.h b/cpp/core/config/GlutenConfig.h
index 16140a3ac3..3dd99e4cf3 100644
--- a/cpp/core/config/GlutenConfig.h
+++ b/cpp/core/config/GlutenConfig.h
@@ -74,6 +74,14 @@ const std::string kParquetWriterVersion =
"parquet.writer.version";
const std::string kParquetCompressionCodec =
"spark.sql.parquet.compression.codec";
+/// Spark `spark.sql.parquet.writeLegacyFormat` (passed from the JVM as
"true"/"false").
+/// Used in VeloxWriterUtils to set
`WriterOptions::enableStoreDecimalAsInteger` (inverted).
+/// Velox decimal storage when enableStoreDecimalAsInteger is:
+/// - true (Spark legacy off / unset): INT32/INT64 for short DECIMAL
precisions; higher precisions
+/// use FIXED_LEN_BYTE_ARRAY.
+/// - false (Spark legacy on): FIXED_LEN_BYTE_ARRAY for all DECIMAL precisions.
+const std::string kParquetStoreDecimalAsInteger =
"spark.sql.parquet.writeLegacyFormat";
+
const std::string kColumnarToRowMemoryThreshold =
"spark.gluten.sql.columnarToRowMemoryThreshold";
const std::string kUGIUserName = "spark.gluten.ugi.username";
diff --git a/cpp/velox/utils/VeloxWriterUtils.cc
b/cpp/velox/utils/VeloxWriterUtils.cc
index bd3cca6854..5616cd9481 100644
--- a/cpp/velox/utils/VeloxWriterUtils.cc
+++ b/cpp/velox/utils/VeloxWriterUtils.cc
@@ -49,6 +49,20 @@ std::unique_ptr<WriterOptions> makeParquetWriteOption(const
std::unordered_map<s
}
auto writeOption = std::make_unique<WriterOptions>();
writeOption->parquetWriteTimestampUnit = TimestampPrecision::kMicroseconds
/*micro*/;
+ bool writeLegacyParquetFormat = false;
+ if (auto it = sparkConfs.find(kParquetStoreDecimalAsInteger); it !=
sparkConfs.end()) {
+ writeLegacyParquetFormat = boost::iequals(it->second, "true");
+ }
+ // Maps spark.sql.parquet.writeLegacyFormat to Velox
enableStoreDecimalAsInteger (inverted).
+ // Controls how DECIMAL values are stored by the Writer:
+ // - true (default when legacy format is off): store as integer using
INT32/INT64 for short
+ // DECIMAL precisions; higher precisions use FIXED_LEN_BYTE_ARRAY.
+ // - false (when spark.sql.parquet.writeLegacyFormat is true): store as
FIXED_LEN_BYTE_ARRAY
+ // regardless of precision (Spark legacy Parquet decimal layout).
+ // TODO: Only DECIMAL is handled here. Spark's writeLegacyFormat also
changes ARRAY (bag/array
+ // vs list/element) and MAP (map vs key_value) physical layouts, which are
not yet supported
+ // in Gluten native Parquet write.
+ writeOption->enableStoreDecimalAsInteger = !writeLegacyParquetFormat;
auto compressionCodec = CompressionKind::CompressionKind_SNAPPY;
if (auto it = sparkConfs.find(kParquetCompressionCodec); it !=
sparkConfs.end()) {
auto compressionCodecStr = it->second;
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 14610868eb..053d715125 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
@@ -652,6 +652,7 @@ object GlutenConfig extends ConfigRegistry {
DEBUG_ENABLED.key,
// datasource config
SPARK_SQL_PARQUET_COMPRESSION_CODEC,
+ SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key,
// datasource config end
GlutenCoreConfig.COLUMNAR_OVERHEAD_SIZE_IN_BYTES.key,
GlutenCoreConfig.COLUMNAR_OFFHEAP_SIZE_IN_BYTES.key,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
