[VL] Support config velox parquet writer option enableStoreDecimalAsInteger for write parquet

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)

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

cpp/core/config/GlutenConfig.h

@@ -71,6 +71,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";

cpp/velox/utils/VeloxWriterUtils.cc

@@ -49,6 +49,17 @@ 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).
+  writeOption->enableStoreDecimalAsInteger = !writeLegacyParquetFormat;
   auto compressionCodec = CompressionKind::CompressionKind_SNAPPY;
   if (auto it = sparkConfs.find(kParquetCompressionCodec); it != sparkConfs.end()) {
     auto compressionCodecStr = it->second;

gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala

@@ -633,6 +633,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,