Copilot commented on code in PR #13073:
URL: https://github.com/apache/gluten/pull/13073#discussion_r4122184169
##########
backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala:
##########
@@ -54,18 +60,28 @@ abstract class AbstractIcebergWriteExec extends
IcebergWriteExec {
PARQUET_PAGE_SIZE_BYTES.key -> getParquetPageSizeBytes,
COLUMNAR_PARQUET_WRITE_BLOCK_SIZE.key -> getParquetRowGroupSizeBytes,
parquetPageRowLimitSession -> getParquetPageRowLimit,
+ parquetPageVersionSession -> getParquetPageVersion,
MAX_TARGET_FILE_SIZE_SESSION.key -> getTargetFileSizeBytes,
PARQUET_DICT_SIZE_BYTES.key -> getDictSizeBytes
).foreach {
case (key, value) =>
val overrideValue = SQLConf.get.getConfString(key, null)
if (overrideValue == null) {
icebergProperties.put(key, value)
- } else if (key != parquetPageRowLimitSession) {
+ } else if (key != parquetPageRowLimitSession && key !=
parquetPageVersionSession) {
icebergProperties.put(key, normalizeCapacityString(overrideValue))
}
Review Comment:
When this SQLConf override is present, this branch deliberately skips
inserting any value for `parquetPageVersionSession` into `icebergProperties`.
Since `IcebergDataWriteFactory` forwards that map as the native writer
configuration, the native writer never sees the session override and continues
using the table/default page version (the added V1/V2 override test will fail).
Preserve the raw override value for this non-capacity setting while continuing
to avoid byte normalization.
This issue also appears on line 79 of the same file.
##########
cpp/velox/utils/VeloxWriterUtils.cc:
##########
@@ -38,6 +38,17 @@ const int32_t kGzipWindowBits4k = 12;
const int32_t kZSTDDefaultCompressionLevel = 3;
} // namespace
+std::shared_ptr<dwio::common::FormatSpecificOptions>
GlutenParquetWriterFactory::createFormatOptions(
+ const config::ConfigBase& connectorConfig,
+ const config::ConfigBase& session) const {
+ auto options = ParquetWriterFactory::createFormatOptions(connectorConfig,
session);
+ if (auto level = session.get<int32_t>("writer_compression_level")) {
Review Comment:
The Scala path emits
`spark.gluten.sql.columnar.backend.velox.parquet_writer_compression_level`;
`createHiveConnectorSessionConfig` strips the dynamic prefix and passes
`parquet_writer_compression_level` to this factory. Reading
`writer_compression_level` therefore misses the production setting, so
compression levels are ignored end to end. Use the same prefixed Velox session
key here.
##########
gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala:
##########
@@ -51,6 +51,19 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec {
} else codec.toLowerCase(Locale.ROOT)
}
+ protected def getParquetCompressionLevel: Option[String] = {
+
Option(IcebergWriteUtil.getWriteProperty(write).get(PARQUET_COMPRESSION_LEVEL))
+
.orElse(Option(IcebergWriteUtil.getTable(write).properties().get(PARQUET_COMPRESSION_LEVEL)))
Review Comment:
`getWriteProperty` is only the `SparkWrite.writeProperties` map, while
`spark.sql.iceberg.compression-level` is a session-level `SparkWriteConf`
setting. This method therefore never reads the documented session override, so
the new `withSQLConf("spark.sql.iceberg.compression-level" -> "1")` cases will
still use the table level. Resolve the compression level through Iceberg's
`SparkWriteConf`/session configuration before falling back to the table
property.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
