From 82242c7d557d6e939fddfe6a7f97df5f1e263efa Mon Sep 17 00:00:00 2001 From: Hazmi Date: Wed, 26 Aug 2026 12:55:17 +0000 Subject: [PATCH 1/2] [VL][Iceberg] Fix target file size bytes to session config --- .../enhanced/VeloxIcebergSuite.scala | 115 +++++++++--------- cpp/velox/utils/ConfigExtractor.cc | 2 +- 2 files changed, 57 insertions(+), 60 deletions(-) diff --git a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala index f3d8e290d4..4f426ec51d 100644 --- a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala +++ b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala @@ -475,75 +475,72 @@ class VeloxIcebergSuite extends IcebergSuite { ) } } - ignore("disabled test") { - test("iceberg native write respects target file size bytes") { - withTable("iceberg_small_target_tbl") { + test("iceberg native write respects target file size bytes") { + withTable("iceberg_small_target_tbl") { + spark.sql( + """ + |CREATE TABLE iceberg_small_target_tbl ( + | id INT, + | payload STRING + |) USING iceberg + |TBLPROPERTIES ( + | 'write.format.default' = 'parquet', + | 'write.parquet.compression-codec' = 'uncompressed', + | 'write.parquet.page-size-bytes' = '1024B', + | 'write.target-file-size-bytes' = '8192' + |) + |""".stripMargin) + + checkAnswer( spark.sql( """ - |CREATE TABLE iceberg_small_target_tbl ( - | id INT, - | payload STRING - |) USING iceberg - |TBLPROPERTIES ( - | 'write.format.default' = 'parquet', - | 'write.parquet.compression-codec' = 'uncompressed', - | 'write.parquet.row-group-size-bytes' = '4096', - | 'write.parquet.page-size-bytes' = '1024B', - | 'write.target-file-size-bytes' = '8192' - |) - |""".stripMargin) - - checkAnswer( - spark.sql( - """ - |SHOW TBLPROPERTIES iceberg_small_target_tbl - |('write.target-file-size-bytes') - |""".stripMargin), - Seq(Row("write.target-file-size-bytes", "8192")) - ) + |SHOW TBLPROPERTIES iceberg_small_target_tbl + |('write.target-file-size-bytes') + |""".stripMargin), + Seq(Row("write.target-file-size-bytes", "8192")) + ) - val df = spark.sql( - """ - |INSERT INTO iceberg_small_target_tbl - |SELECT /*+ COALESCE(1) */ - | CAST(id AS INT), - | concat( - | CAST(id AS STRING), - | '-', - | sha2(CAST(id AS STRING), 256), - | '-', - | sha2(CAST(id + 1000 AS STRING), 256) - | ) - |FROM range(1000) - |""".stripMargin) + val df = spark.sql( + """ + |INSERT INTO iceberg_small_target_tbl + |SELECT /*+ COALESCE(1) */ + | CAST(id AS INT), + | concat( + | CAST(id AS STRING), + | '-', + | sha2(CAST(id AS STRING), 256), + | '-', + | sha2(CAST(id + 1000 AS STRING), 256) + | ) + |FROM range(1000) + |""".stripMargin) - val commandPlan = - df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan + val commandPlan = + df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan - assert(commandPlan.isInstanceOf[VeloxIcebergAppendDataExec]) + assert(commandPlan.isInstanceOf[VeloxIcebergAppendDataExec]) - checkAnswer( - spark.sql("SELECT COUNT(*) FROM iceberg_small_target_tbl"), - Seq(Row(1000L))) + checkAnswer( + spark.sql("SELECT COUNT(*) FROM iceberg_small_target_tbl"), + Seq(Row(1000L))) - val files = spark.sql( - """ - |SELECT file_size_in_bytes - |FROM default.iceberg_small_target_tbl.files - |""".stripMargin).collect().map(_.getLong(0)) + val files = spark.sql( + """ + |SELECT file_size_in_bytes + |FROM default.iceberg_small_target_tbl.files + |""".stripMargin).collect().map(_.getLong(0)) - assert(files.nonEmpty) + assert(files.nonEmpty) - assert( - files.length > 1, - s"Expected write.target-file-size-bytes=8192 to create multiple files, " + - s"but got files=${files.mkString("[", ", ", "]")}") + assert( + files.length > 1, + s"Expected write.target-file-size-bytes=8192 to create multiple files, " + + s"but got files=${files.mkString("[", ", ", "]")}") - assert( - files.max < 64L * 1024L, - s"Expected small target file size to keep max file size reasonably small, " + - s"but got files=${files.mkString("[", ", ", "]")}") - } + assert( + files.max < 64L * 1024L, + s"Expected small target file size to keep max file size reasonably small, " + + s"but got files=${files.mkString("[", ", ", "]")}") } } diff --git a/cpp/velox/utils/ConfigExtractor.cc b/cpp/velox/utils/ConfigExtractor.cc index d38745416b..fa75cd3db6 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -265,7 +265,7 @@ std::shared_ptr createHiveConnectorSessionC configs[facebook::velox::connector::hive::HiveConfig::kReadTimestampUnitSession] = std::string("6"); configs[facebook::velox::connector::hive::HiveConfig::kMaxPartitionsPerWritersSession] = conf->get(kMaxPartitions, "10000"); - configs[facebook::velox::connector::hive::HiveConfig::kParquetMaxTargetFileSize] = + configs[facebook::velox::connector::hive::HiveConfig::kParquetMaxTargetFileSizeSession] = conf->get(kParquetMaxTargetFileSize, "0B"); // 0 means no limit on target file size configs[facebook::velox::connector::hive::HiveConfig::kIgnoreMissingFilesSession] = conf->get(kIgnoreMissingFiles, false) ? "true" : "false"; From 8b5e4bacbd4715c8b590c6cd4a5c6c8fbef8880a Mon Sep 17 00:00:00 2001 From: Hazmi Date: Thu, 27 Aug 2026 13:08:43 +0000 Subject: [PATCH 2/2] [VL][Iceberg] Normalize target file size capacity --- .../scala/org/apache/gluten/execution/IcebergWriteExec.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala index 270745063a..227b7e4f7b 100644 --- a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala +++ b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala @@ -57,7 +57,7 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec { } protected def getTargetFileSizeBytes: String = { - IcebergWriteUtil.getWriteConf(write).targetDataFileSize().toString + normalizeCapacityString(IcebergWriteUtil.getWriteConf(write).targetDataFileSize().toString) } protected def getDictSizeBytes: String = {