Skip to content

[VL] feat(iceberg): Added dictionary size bytes write config #12434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhouyuan merged 1 commit into from
Jul 4, 2026
Merged

[VL] feat(iceberg): Added dictionary size bytes write config #12434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...s-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
package org.apache.gluten.execution

import org.apache.gluten.IcebergNestedFieldVisitor
import org.apache.gluten.config.VeloxConfig.{MAX_TARGET_FILE_SIZE_SESSION, PARQUET_PAGE_SIZE_BYTES}
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}

import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -48,7 +48,8 @@ abstract class AbstractIcebergWriteExec extends IcebergWriteExec {

Seq(
PARQUET_PAGE_SIZE_BYTES.key -> getParquetPageSizeBytes,
MAX_TARGET_FILE_SIZE_SESSION.key -> getTargetFileSizeBytes
MAX_TARGET_FILE_SIZE_SESSION.key -> getTargetFileSizeBytes,
PARQUET_DICT_SIZE_BYTES.key -> getDictSizeBytes
).foreach {
case (key, value) =>
if (SQLConf.get.getConfString(key, null) == null) {
Expand Down
110 changes: 110 additions & 0 deletions ...velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ 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.column.Encoding
import org.apache.iceberg.shaded.org.apache.parquet.column.page.{DataPage, DataPageV1, DataPageV2}
import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader
import org.apache.iceberg.shaded.org.apache.parquet.hadoop.util.HadoopInputFile

Expand Down Expand Up @@ -543,6 +545,114 @@ class VeloxIcebergSuite extends IcebergSuite {
}
}
}

test("iceberg parquet writer respects dictionary page size bytes") {
val table = "iceberg_dict_page_size_tbl"

def parquetFiles(table: String): Seq[String] = {
spark.sql(s"""
|SELECT file_path
|FROM default.$table.files
|""".stripMargin).collect().map(_.getString(0)).toSeq
}

def pageEncoding(page: DataPage): Encoding = {
page.accept(new DataPage.Visitor[Encoding] {
override def visit(dataPageV1: DataPageV1): Encoding = dataPageV1.getValueEncoding
override def visit(dataPageV2: DataPageV2): Encoding = dataPageV2.getDataEncoding
})
}

def dataPageEncodings(table: String, columnName: String): Seq[Encoding] = {
val conf = spark.sparkContext.hadoopConfiguration

parquetFiles(table).flatMap {
file =>
val inputFile = HadoopInputFile.fromPath(new Path(file), conf)
val reader = ParquetFileReader.open(inputFile, ParquetReadOptions.builder().build())

try {
val column = reader
.getFooter
.getFileMetaData
.getSchema
.getColumns
.asScala
.find(_.getPath.toSeq == Seq(columnName))
.getOrElse {
fail(s"Column $columnName was not found in Parquet file $file")
}

val encodings = scala.collection.mutable.ArrayBuffer.empty[Encoding]

var rowGroup = reader.readNextRowGroup()
while (rowGroup != null) {
val pageReader = rowGroup.getPageReader(column)
pageReader.readDictionaryPage()

var page = pageReader.readPage()
while (page != null) {
encodings += pageEncoding(page)
page = pageReader.readPage()
}

rowGroup = reader.readNextRowGroup()
}

encodings
} finally {
reader.close()
}
}
}

withSQLConf(
"spark.sql.shuffle.partitions" -> "1"
) {
withTable(table) {
spark.sql(s"""
|CREATE TABLE $table (
| value SMALLINT
|) USING iceberg
|TBLPROPERTIES (
| 'write.format.default' = 'parquet',
| 'write.parquet.compression-codec' = 'uncompressed',
| 'write.parquet.dict-size-bytes' = '1B'
|)
|""".stripMargin)

val df = spark.sql(s"""
|INSERT INTO $table
|SELECT CAST(id + 1 AS SMALLINT)
|FROM range(0, 10000, 1, 1)
|""".stripMargin)

assert(
df.queryExecution.executedPlan
.asInstanceOf[CommandResultExec]
.commandPhysicalPlan
.isInstanceOf[VeloxIcebergAppendDataExec])

checkAnswer(
spark.sql(s"SELECT count(*) FROM $table"),
Seq(Row(10000L)))

val encodings = dataPageEncodings(table, "value")

assert(encodings.nonEmpty, "Expected at least one Parquet data page")
assert(
encodings.head == Encoding.RLE_DICTIONARY,
s"Expected the first data page to use dictionary encoding, " +
s"but got encodings=${encodings.mkString("[", ", ", "]")}"
)
assert(
encodings.contains(Encoding.PLAIN),
s"Expected write.parquet.dict-size-bytes=1B to make later data pages fall back " +
s"to PLAIN, but got encodings=${encodings.mkString("[", ", ", "]")}"
)
}
}
}
test("iceberg parquet writer default row group size test") {
val table = "iceberg_default_row_group_size"
val defaultRowGroupBytes = 128L * 1024 * 1024
Expand Down
6 changes: 6 additions & 0 deletions backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -862,6 +862,12 @@ object VeloxConfig extends ConfigRegistry {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("1MB")

val PARQUET_DICT_SIZE_BYTES =
buildConf("spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes")
.doc("The maximum size in bytes for a Parquet dictionary page")
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2MB")

val ENABLE_TIMESTAMP_NTZ_VALIDATION =
buildConf("spark.gluten.sql.columnar.backend.velox.enableTimestampNtzValidation")
.doc(
Expand Down
2 changes: 2 additions & 0 deletions cpp/velox/config/VeloxConfig.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,8 @@ const uint32_t kGlogSeverityLevelDefault = 1;

// Iceberg write configs
const std::string kWriteParquetPageSizeBytes = "spark.gluten.sql.columnar.backend.velox.parquet.pageSizeBytes";
const std::string kWriteParquetDictSizeBytes =
"spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes";

// Query trace
/// Enable query tracing flag.
Expand Down
2 changes: 2 additions & 0 deletions cpp/velox/utils/ConfigExtractor.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,6 +252,8 @@ std::shared_ptr<facebook::velox::config::ConfigBase> createHiveConnectorSessionC
conf->get<bool>(kOrcUseColumnNames, true) ? "true" : "false";
configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterPageSizeSession)] =
conf->get<std::string>(kWriteParquetPageSizeBytes, "1MB");
configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterDictionaryPageSizeLimitSession)] =
conf->get<std::string>(kWriteParquetDictSizeBytes, "2MB");
configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kNullStructIfAllFieldsMissingSession)] =
"true";

Expand Down
1 change: 1 addition & 0 deletions docs/velox-configuration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. |
| spark.gluten.sql.columnar.backend.velox.orc.scan.enabled | 🔄 Dynamic | true | Enable velox orc scan. If disabled, vanilla spark orc scan will be used. |
| spark.gluten.sql.columnar.backend.velox.orcUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for ORC files. |
| spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes | 🔄 Dynamic | 2MB | The maximum size in bytes for a Parquet dictionary page |
| spark.gluten.sql.columnar.backend.velox.parquet.pageSizeBytes | 🔄 Dynamic | 1MB | The page size in bytes is for compression. |
| spark.gluten.sql.columnar.backend.velox.parquetMaxTargetFileSize | 🔄 Dynamic | 0b | The target file size for each output file when writing data. 0 means no limit on target file size, and the actual file size will be determined by other factors such as max partition number and shuffle batch size. |
| spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames | 🔄 Dynamic | true | Maps table field names to file field names using names, not indices for Parquet files. |
Expand Down
9 changes: 8 additions & 1 deletion gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_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}
import org.apache.iceberg.avro.AvroSchemaUtil
import org.apache.iceberg.spark.source.IcebergWriteUtil
import org.apache.iceberg.types.Type.TypeID
Expand Down Expand Up @@ -60,6 +60,13 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec {
IcebergWriteUtil.getWriteConf(write).targetDataFileSize().toString
}

protected def getDictSizeBytes: String = {
val tableProps = IcebergWriteUtil.getTable(write).properties()
tableProps.getOrDefault(
normalizeCapacityString(PARQUET_DICT_SIZE_BYTES),
normalizeCapacityString(PARQUET_DICT_SIZE_BYTES_DEFAULT.toString))
}

protected def getPartitionSpec: PartitionSpec = {
IcebergWriteUtil.getPartitionSpec(write)
}
Expand Down
Loading