Added iceberg write configs

Author: infvg

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala

@@ -751,4 +751,40 @@ object VeloxConfig extends ConfigRegistry {
       .doc("Maps table field names to file field names using names, not indices for Parquet files.")
       .booleanConf
       .createWithDefault(true)
+
+  val PARQUET_COMPRESSION_CODEC =
+    buildConf("spark.gluten.sql.columnar.backend.velox.parquet.compression-codec")
+      .doc("Compression codec to use for Parquet write operations.")
+      .stringConf
+      .createWithDefault("UNCOMPRESSED")
+
+  val PARQUET_COMPRESSION_LEVEL =
+    buildConf("spark.gluten.sql.columnar.backend.velox.parquet.compression-level")
+      .doc("Compression level for Parquet write operations.")
+      .intConf
+      .createOptional
+
+  val VELOX_TARGET_FILE_SIZE =
+    buildConf("spark.gluten.sql.columnar.backend.velox.target-file-size-byte")
+      .doc("Target file size in bytes for write operations.")
+      .bytesConf(ByteUnit.BYTE)
+      .createWithDefaultString("0")
+
+  val ICEBERG_WRITE_PARQUET_ROW_GROUP_SIZE_BYTES =
+    buildConf("spark.gluten.sql.columnar.backend.velox.iceberg.write.parquet.row-group-size-bytes")
+      .doc("Row group size in bytes for Iceberg Parquet write operations.")
+      .bytesConf(ByteUnit.BYTE)
+      .createWithDefaultString("128MB")
+
+  val ICEBERG_WRITE_PARQUET_PAGE_SIZE_BYTES =
+    buildConf("spark.gluten.sql.columnar.backend.velox.iceberg.write.parquet.page-size-bytes")
+      .doc("Page size in bytes for Iceberg Parquet write operations.")
+      .bytesConf(ByteUnit.BYTE)
+      .createWithDefaultString("1MB")
+
+  val ICEBERG_WRITE_PARQUET_PAGE_ROW_LIMIT =
+    buildConf("spark.gluten.sql.columnar.backend.velox.iceberg.write.parquet.page-row-limit")
+      .doc("Maximum number of rows per page for Iceberg Parquet write operations.")
+      .intConf
+      .createWithDefault(20000)
 }

cpp/velox/config/VeloxConfig.h

@@ -166,9 +166,24 @@ const std::string kMemoryPoolCapacityTransferAcrossTasks =
 const std::string kOrcUseColumnNames = "spark.gluten.sql.columnar.backend.velox.orcUseColumnNames";
 const std::string kParquetUseColumnNames = "spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames";
 
-// write fies
+// write files
 const std::string kMaxPartitions = "spark.gluten.sql.columnar.backend.velox.maxPartitionsPerWritersSession";
 
+// Iceberg write configs
+const std::string kWriteTargetFileSizeBytes =
+    "spark.gluten.sql.columnar.backend.velox.target-file-size-bytes";
+const std::string kWriteParquetCompressionCodec =
+    "spark.gluten.sql.columnar.backend.velox.parquet.compression-codec";
+const std::string kWriteParquetCompressionLevel =
+    "spark.gluten.sql.columnar.backend.velox.parquet.compression-level";
+
+const std::string kWriteParquetRowGroupSizeBytes =
+    "spark.gluten.sql.columnar.backend.velox.parquet.row-group-size-bytes";
+const std::string kWriteParquetPageSizeBytes =
+    "spark.gluten.sql.columnar.backend.velox.parquet.page-size-bytes";
+const std::string kWriteParquetPageRowLimit =
+    "spark.gluten.sql.columnar.backend.velox.parquet.page-row-limit";
+
 const std::string kGlogVerboseLevel = "spark.gluten.sql.columnar.backend.velox.glogVerboseLevel";
 const uint32_t kGlogVerboseLevelDefault = 0;
 const uint32_t kGlogVerboseLevelMaximum = 99;

cpp/velox/utils/ConfigExtractor.cc

@@ -240,6 +240,15 @@ std::shared_ptr<facebook::velox::config::ConfigBase> createHiveConnectorSessionC
   configs[facebook::velox::connector::hive::HiveConfig::kOrcUseColumnNamesSession] =
       conf->get<bool>(kOrcUseColumnNames, true) ? "true" : "false";
 
+  configs[facebook::velox::connector::hive::HiveConfig::kMaxTargetFileSize] =
+      conf->get<std::string>(kWriteTargetFileSizeBytes, "0");
+  configs[facebook::velox::parquet::WriterOptions::kParquetSessionWritePageSize] =
+      conf->get<std::string>(kWriteParquetCompressionCodec, "UNCOMPRESSED");
+  // configs[facebook::velox::parquet::WriterOptions::kParquetHiveConnectorCompressionLevel] =
+  //     conf->get<std::string>(kWriteParquetCompressionLevel, "0");
+  // configs[facebook::velox::parquet::DefaultFlushPolicy::kDefaultBytesInRowGroup] =
+  //   conf->get<std::string>(kWriteParquetRowGroupSizeBytes, "UNCOMPRESSED");
+
   overwriteVeloxConf(conf.get(), configs, kDynamicBackendConfPrefix);
   return std::make_shared<facebook::velox::config::ConfigBase>(std::move(configs));
 }