[VL][Iceberg] Propagate Parquet row group size to Velox

backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala

@@ -17,6 +17,7 @@
 package org.apache.gluten.execution
 
 import org.apache.gluten.IcebergNestedFieldVisitor
+import org.apache.gluten.config.GlutenConfig.COLUMNAR_PARQUET_WRITE_BLOCK_SIZE
 import org.apache.gluten.config.VeloxConfig.{MAX_TARGET_FILE_SIZE_SESSION, PARQUET_DICT_SIZE_BYTES, PARQUET_PAGE_SIZE_BYTES}
 import org.apache.gluten.connector.write.{ColumnarBatchDataWriterFactory, ColumnarStreamingDataWriterFactory, IcebergDataWriteFactory}
 
@@ -48,6 +49,7 @@ abstract class AbstractIcebergWriteExec extends IcebergWriteExec {
 
     Seq(
       PARQUET_PAGE_SIZE_BYTES.key -> getParquetPageSizeBytes,
+      COLUMNAR_PARQUET_WRITE_BLOCK_SIZE.key -> getParquetRowGroupSizeBytes,
       MAX_TARGET_FILE_SIZE_SESSION.key -> getTargetFileSizeBytes,
       PARQUET_DICT_SIZE_BYTES.key -> getDictSizeBytes
     ).foreach {

backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala

@@ -655,10 +655,9 @@ class VeloxIcebergSuite extends IcebergSuite {
     }
   }
 
-  // Ignored due to velox parquet row-group flush semantics change after velox#16998.
-  test("iceberg parquet writer default row group size test") {
-    val table = "iceberg_default_row_group_size"
-    val defaultRowGroupBytes = 128L * 1024 * 1024
+  test("iceberg parquet writer respects row group size") {
+    val table = "iceberg_row_group_size"
+    val rowGroupBytes = 2L * 1024 * 1024
 
     def parquetFiles(table: String): Seq[String] = {
       spark.sql(s"""
@@ -717,7 +716,8 @@ class VeloxIcebergSuite extends IcebergSuite {
           payload STRING
         ) USING iceberg
         TBLPROPERTIES (
-          'write.parquet.compression-codec' = 'uncompressed'
+          'write.parquet.compression-codec' = 'uncompressed',
+          'write.parquet.row-group-size-bytes' = '$rowGroupBytes'
         )
       """)
 
@@ -727,12 +727,12 @@ class VeloxIcebergSuite extends IcebergSuite {
           id,
           array_join(
             transform(
-              sequence(0, 63),
+              sequence(0, 31),
               x -> md5(concat(CAST(id AS STRING), ':', CAST(x AS STRING)))
             ),
             ''
           ) AS payload
-        FROM range(0, 90000, 1, 1)
+        FROM range(0, 10000, 1, 1)
       """)
 
         assert(
@@ -743,7 +743,7 @@ class VeloxIcebergSuite extends IcebergSuite {
 
         checkAnswer(
           spark.sql(s"SELECT count(*) FROM $table"),
-          Seq(Row(90000L)))
+          Seq(Row(10000L)))
         val rowGroups =
           collectRowGroups(table).sortBy(info => (info.file, info.ordinal))
 
@@ -752,27 +752,16 @@ class VeloxIcebergSuite extends IcebergSuite {
           s"Expected one Parquet file, found: ${rowGroups.map(_.file).distinct}")
 
         assert(
-          rowGroups.size == 2,
-          s"Expected 2 row groups, found ${rowGroups.size}: $rowGroups")
-
-        assert(
-          rowGroups.map(_.rowCount).sum == 90000L,
-          s"Expected 90000 rows across all row groups: $rowGroups")
-
-        val firstRowGroup = rowGroups.head
-        val finalRowGroup = rowGroups.last
+          rowGroups.size > 1,
+          s"Expected the Iceberg row-group size to create multiple row groups: $rowGroups")
 
         assert(
-          firstRowGroup.compressedSize >= defaultRowGroupBytes,
-          s"Expected the first row group to reach the default row-group size " +
-            s"$defaultRowGroupBytes, but found ${firstRowGroup.compressedSize}"
-        )
+          rowGroups.map(_.rowCount).sum == 10000L,
+          s"Expected 10000 rows across all row groups: $rowGroups")
 
         assert(
-          finalRowGroup.compressedSize < defaultRowGroupBytes,
-          s"Expected the final row group to be smaller than the default row-group " +
-            s"size $defaultRowGroupBytes, but found ${finalRowGroup.compressedSize}"
-        )
+          rowGroups.dropRight(1).forall(_.compressedSize >= rowGroupBytes),
+          s"Expected each complete row group to reach $rowGroupBytes bytes: $rowGroups")
       }
     }
   }

cpp/core/config/GlutenConfig.h

@@ -57,6 +57,9 @@ const bool kCheckUsageLeakDefault = true;
 
 const std::string kSparkBatchSize = "spark.gluten.sql.columnar.maxBatchSize";
 
+// Parquet historically uses "block" to mean a row group.
+const std::string kColumnarParquetWriteBlockSize = "spark.gluten.sql.columnar.parquet.write.blockSize";
+
 const std::string kParquetBlockSize = "parquet.block.size";
 
 const std::string kParquetBlockRows = "parquet.block.rows";

cpp/velox/utils/ConfigExtractor.cc

@@ -283,6 +283,8 @@ std::shared_ptr<facebook::velox::config::ConfigBase> createHiveConnectorSessionC
       conf->get<bool>(kOrcForcePositionalEvolution, false) ? "false" : "true";
   configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterPageSizeSession)] =
       conf->get<std::string>(kWriteParquetPageSizeBytes, "1MB");
+  configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterRowGroupSizeSession)] =
+      conf->get<std::string>(kColumnarParquetWriteBlockSize, "128MB");
   configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterDictionaryPageSizeLimitSession)] =
       conf->get<std::string>(kWriteParquetDictSizeBytes, "2MB");
   configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kNullStructIfAllFieldsMissingSession)] =

gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala

@@ -19,7 +19,7 @@ package org.apache.gluten.execution
 import org.apache.gluten.backendsapi.BackendsApiManager
 
 import org.apache.iceberg.{FileFormat, PartitionField, PartitionSpec, Schema, TableProperties}
-import org.apache.iceberg.TableProperties.{ORC_COMPRESSION, ORC_COMPRESSION_DEFAULT, PARQUET_COMPRESSION, PARQUET_COMPRESSION_DEFAULT, PARQUET_DICT_SIZE_BYTES, PARQUET_DICT_SIZE_BYTES_DEFAULT, PARQUET_PAGE_SIZE_BYTES, PARQUET_PAGE_SIZE_BYTES_DEFAULT}
+import org.apache.iceberg.TableProperties.{ORC_COMPRESSION, ORC_COMPRESSION_DEFAULT, PARQUET_COMPRESSION, PARQUET_COMPRESSION_DEFAULT, PARQUET_DICT_SIZE_BYTES, PARQUET_DICT_SIZE_BYTES_DEFAULT, PARQUET_PAGE_SIZE_BYTES, PARQUET_PAGE_SIZE_BYTES_DEFAULT, PARQUET_ROW_GROUP_SIZE_BYTES, PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT}
 import org.apache.iceberg.avro.AvroSchemaUtil
 import org.apache.iceberg.spark.source.IcebergWriteUtil
 import org.apache.iceberg.types.Type.TypeID
@@ -60,6 +60,13 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec {
     IcebergWriteUtil.getWriteConf(write).targetDataFileSize().toString
   }
 
+  protected def getParquetRowGroupSizeBytes: String = {
+    val tableProps = IcebergWriteUtil.getTable(write).properties()
+    tableProps.getOrDefault(
+      normalizeCapacityString(PARQUET_ROW_GROUP_SIZE_BYTES),
+      normalizeCapacityString(PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT.toString))
+  }
+
   protected def getDictSizeBytes: String = {
     val tableProps = IcebergWriteUtil.getTable(write).properties()
     tableProps.getOrDefault(

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

@@ -492,6 +492,7 @@ object GlutenConfig extends ConfigRegistry {
     BENCHMARK_SAVE_DIR.key,
     GlutenCoreConfig.COLUMNAR_TASK_OFFHEAP_SIZE_IN_BYTES.key,
     COLUMNAR_MAX_BATCH_SIZE.key,
+    COLUMNAR_PARQUET_WRITE_BLOCK_SIZE.key,
     SHUFFLE_WRITER_BUFFER_SIZE.key,
     COLUMNAR_CUDF_ENABLED.key,
     SQLConf.LEGACY_SIZE_OF_NULL.key,