This is an automated email from the ASF dual-hosted git repository.
zhouyuan 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 aaaa026b8f [VL] test(iceberg): Test default bytes in row group (#12409)
aaaa026b8f is described below
commit aaaa026b8fb3b4362b9566278c9975b0295c8778
Author: inf <[email protected]>
AuthorDate: Thu Jul 2 20:01:44 2026 +0000
[VL] test(iceberg): Test default bytes in row group (#12409)
Adding test for default row group size in iceberg
---
.../execution/enhanced/VeloxIcebergSuite.scala | 112 +++++++++++++++++++++
1 file changed, 112 insertions(+)
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 0ec53f4bab..0cc6cddda2 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
@@ -16,6 +16,7 @@
*/
package org.apache.gluten.execution.enhanced
+import org.apache.gluten.config.VeloxConfig.MAX_TARGET_FILE_SIZE_SESSION
import org.apache.gluten.execution._
import org.apache.gluten.tags.EnhancedFeaturesTest
@@ -26,6 +27,13 @@ import
org.apache.spark.sql.execution.datasources.v2.AppendDataExec
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.gluten.TestUtils
+import org.apache.hadoop.fs.Path
+import org.apache.iceberg.shaded.org.apache.parquet.ParquetReadOptions
+import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.iceberg.shaded.org.apache.parquet.hadoop.util.HadoopInputFile
+
+import scala.jdk.CollectionConverters._
+
@EnhancedFeaturesTest
class VeloxIcebergSuite extends IcebergSuite {
@@ -535,4 +543,108 @@ class VeloxIcebergSuite extends IcebergSuite {
}
}
}
+ test("iceberg parquet writer default row group size test") {
+ val table = "iceberg_default_row_group_size"
+ val defaultRowGroupBytes = 128L * 1024 * 1024
+
+ def parquetFiles(table: String): Seq[String] = {
+ spark.sql(s"""
+ SELECT file_path
+ FROM default.$table.files
+ """).collect().map(_.getString(0)).toSeq
+ }
+
+ case class RowGroupInfo(
+ file: String,
+ ordinal: Int,
+ rowCount: Long,
+ totalByteSize: Long,
+ compressedSize: Long)
+
+ def collectRowGroups(table: String): Seq[RowGroupInfo] = {
+ val conf = spark.sparkContext.hadoopConfiguration
+
+ parquetFiles(table).flatMap {
+ file =>
+ val path = new Path(file)
+ val inputFile = HadoopInputFile.fromPath(path, conf)
+ val options = ParquetReadOptions.builder().build()
+
+ val stream = inputFile.newStream()
+ val footer =
+ try {
+ ParquetFileReader.readFooter(inputFile, options, stream)
+ } finally {
+ stream.close()
+ }
+
+ footer.getBlocks.asScala.zipWithIndex.map {
+ case (block, index) =>
+ val compressedSize =
+ block.getColumns.asScala.map(_.getTotalSize).sum
+
+ RowGroupInfo(
+ file = file,
+ ordinal = index,
+ rowCount = block.getRowCount,
+ totalByteSize = block.getTotalByteSize,
+ compressedSize = compressedSize)
+ }
+ }
+ }
+
+ withSQLConf(
+ MAX_TARGET_FILE_SIZE_SESSION.key -> "0",
+ "spark.sql.shuffle.partitions" -> "1"
+ ) {
+ withTable(table) {
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id BIGINT,
+ payload STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.compression-codec' = 'uncompressed'
+ )
+ """)
+
+ val df = spark.sql(s"""
+ INSERT INTO $table
+ SELECT
+ id,
+ array_join(
+ transform(
+ sequence(0, 63),
+ x -> md5(concat(CAST(id AS STRING), ':', CAST(x AS STRING)))
+ ),
+ ''
+ ) AS payload
+ FROM range(0, 90000, 1, 1)
+ """)
+
+ assert(
+ df.queryExecution.executedPlan
+ .asInstanceOf[CommandResultExec]
+ .commandPhysicalPlan
+ .isInstanceOf[VeloxIcebergAppendDataExec])
+
+ checkAnswer(
+ spark.sql(s"SELECT count(*) FROM $table"),
+ Seq(Row(90000L)))
+ val rowGroups = collectRowGroups(table)
+
+ assert(
+ rowGroups.size == 2,
+ "Expected 2 row groups")
+
+ assert(
+ rowGroups.head.totalByteSize < defaultRowGroupBytes,
+ "Expected row group to contain less than default value")
+
+ assert(
+ rowGroups(1).totalByteSize < defaultRowGroupBytes,
+ "Expected row group to contain less than default value")
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
