sunchao commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4367734576
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -795,6 +872,20 @@ object IcebergReflection extends Logging {
.toMap
}
+ /**
+ * Returns the field id resolved by Iceberg's name lookup, including
runtime-supported aliases.
+ */
+ def findFieldIdByName(schema: Any, name: String): Option[Int] = {
+ try {
+ val findFieldMethod = getMethod(schema.getClass, "findField",
classOf[String])
+ Option(findFieldMethod.invoke(schema, name)).map { field =>
+ getMethod(field.getClass, "fieldId").invoke(field).asInstanceOf[Int]
+ }
+ } catch {
+ case _: Exception => None
Review Comment:
[P2] On Iceberg 1.8.1, setting `write.parquet.bloom-filter-enabled.column.`
to `true` makes Java call `Schema.findField("")` and reject the write. This
catch treats the validation exception as a missing field, allowing an opted-in
native write to proceed with no filter. Preserve lookup failures as fallback
reasons and validate the empty suffix for every enabled-prefix entry, including
entries set to `false`.
Evidence: The tagged Java writer raised `IllegalArgumentException: Invalid
column name: (empty)`. The exact-head resolver returned `Some(Map())` and
serialized zero filters. With `enabled=false` and an interpreted FPP of `0.01`
on the empty suffix, Java also rejected the configuration while Comet selected
zero columns. A nonempty missing-name control was skipped by both.
##########
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala:
##########
@@ -112,6 +162,29 @@ object IcebergWriteProtoTranslation {
parseJavaInt(props, Keys.ParquetDictSizeBytes,
Defaults.DictSizeBytes.toInt).toLong
val pageRowLimit = parseJavaInt(props, Keys.ParquetPageRowLimit,
Defaults.PageRowLimit)
val compression = resolveCompression(props)
+ // Iceberg properties use logical schema paths, while Parquet writer
properties require the
+ // physical leaf path. Missing fields are skipped, matching Iceberg Java's
writer behavior.
+ val bloomFilterColumns = enabledBloomFilterColumnNames(props)
Review Comment:
[P2] On Iceberg 1.8.1, `tags.a` and `tags.element.a` can name the same leaf
of `ARRAY<STRUCT<a: INT>>`. Enable both and set FPP `0.0001` only on `tags.a`.
Java retains the explicit FPP. Comet emits two entries for
`tags.list.element.a`, and the second resets it to `0.01`. Resolve
enabled-prefix entries by physical leaf before discarding disabled names or
applying defaults. Falling back on collisions also preserves Java’s behavior.
Evidence: The exact-head translator emitted FPPs `[0.0001, 0.01]`. The
tagged 1.8.1 Java setter retained `0.0001`. With integers `0..99`, the
two-entry native file had a 128-byte filter and 9,088 positives among one
million absent probes. The single-setting control had 512 bytes and 13
positives. A second fixture with `tags.a=true` and `tags.element.a=false` left
the Java filter disabled while Comet enabled it.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -1120,9 +1317,62 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
// `option("write-parquet-compression-codec", "gzip")`) survive into the
native writer.
val resolvedWriteProperties =
IcebergReflection.getWritePropertiesFromSparkWrite(sparkWrite).getOrElse(Map.empty)
- val effectiveProperties = properties ++ resolvedWriteProperties
- val parquetSettings =
- IcebergWriteProtoTranslation.buildParquetSettings(effectiveProperties,
createdBy)
+ val effectiveProperties =
+ interpretedBloomFilterProperties(properties ++ resolvedWriteProperties)
+ val parquetPathByIcebergColumnName =
+ if
(IcebergWriteProtoTranslation.hasEnabledBloomFilters(effectiveProperties)) {
+ val resolution =
IcebergReflection.getParquetPathResolution(writeSchema).getOrElse {
+ withFallbackReason(op, "Could not resolve physical Parquet paths for
Bloom filters")
+ return None
+ }
+ val configuredColumns =
+
IcebergWriteProtoTranslation.enabledBloomFilterColumnNames(effectiveProperties)
+ val icebergVersion = IcebergReflection.icebergVersion()
+ val versionParts = icebergVersion.split("[.-]", 3)
+ val version =
+ if (versionParts.length >= 2) {
+ scala.util.Try((versionParts(0).toInt,
versionParts(1).toInt)).toOption
+ } else {
+ None
+ }
+ val resolvedPaths = version match {
+ case Some((1, minor)) if minor < 8 =>
+ // Iceberg 1.5 passes the configured name straight to parquet-mr.
+ configuredColumns.filter(resolution.physicalPaths).map(path =>
path -> path).toMap
Review Comment:
[P2] Iceberg 1.5.2 passes the configured spelling directly to parquet-mr.
For an `id INT` column, `write.parquet.bloom-filter-enabled.column.id.=true`
therefore enables the `id` filter because `ColumnPath.fromDotString` drops the
trailing empty component. The exact-string lookup here drops the setting, so
the native file loses the requested filter. Normalize raw paths with
parquet-mr’s rules before matching them, or fall back for these spellings.
Evidence: A parquet-mr 1.13.1 writer produced a 1,048,576-byte filter on
`id` for both `id.` and `id`. The exact-head 1.5.2 resolution branch returned
an empty map for `id.` and `Map(id -> id)` for the control.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -1120,9 +1317,62 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
// `option("write-parquet-compression-codec", "gzip")`) survive into the
native writer.
val resolvedWriteProperties =
IcebergReflection.getWritePropertiesFromSparkWrite(sparkWrite).getOrElse(Map.empty)
- val effectiveProperties = properties ++ resolvedWriteProperties
- val parquetSettings =
- IcebergWriteProtoTranslation.buildParquetSettings(effectiveProperties,
createdBy)
+ val effectiveProperties =
+ interpretedBloomFilterProperties(properties ++ resolvedWriteProperties)
+ val parquetPathByIcebergColumnName =
+ if
(IcebergWriteProtoTranslation.hasEnabledBloomFilters(effectiveProperties)) {
+ val resolution =
IcebergReflection.getParquetPathResolution(writeSchema).getOrElse {
+ withFallbackReason(op, "Could not resolve physical Parquet paths for
Bloom filters")
+ return None
+ }
+ val configuredColumns =
+
IcebergWriteProtoTranslation.enabledBloomFilterColumnNames(effectiveProperties)
+ val icebergVersion = IcebergReflection.icebergVersion()
+ val versionParts = icebergVersion.split("[.-]", 3)
+ val version =
+ if (versionParts.length >= 2) {
+ scala.util.Try((versionParts(0).toInt,
versionParts(1).toInt)).toOption
+ } else {
+ None
+ }
+ val resolvedPaths = version match {
+ case Some((1, minor)) if minor < 8 =>
+ // Iceberg 1.5 passes the configured name straight to parquet-mr.
+ configuredColumns.filter(resolution.physicalPaths).map(path =>
path -> path).toMap
+ case Some((1, minor)) if minor < 10 =>
+ // Iceberg 1.8 resolves canonical and short collection aliases
through field ids.
+ configuredColumns.flatMap { configuredName =>
+ IcebergReflection
+ .findFieldIdByName(writeSchema, configuredName)
+ .flatMap(resolution.pathByFieldId.get)
+ .map(configuredName -> _)
+ }.toMap
+ case _ =>
Review Comment:
[P2] A repackaged Iceberg 1.8.1 runtime can report `unknown` when its build
metadata omits `git.build.version`. This catch-all then selects canonical-only
lookup, silently dropping a valid `tags.a` filter on `ARRAY<STRUCT<a: INT>>`.
Fall back when the runtime’s naming convention cannot be identified, or
determine it from a writer capability.
Evidence: With the version property omitted, the tagged `IcebergBuild`
returned `unknown`. The exact-head resolver and serializer emitted zero entries
for `tags.a=true`, while the tagged 1.8.1 writer enabled `tags.list.element.a`.
Supplying `1.8.1` to the same resolver restored the mapping.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -1120,9 +1317,62 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
// `option("write-parquet-compression-codec", "gzip")`) survive into the
native writer.
val resolvedWriteProperties =
IcebergReflection.getWritePropertiesFromSparkWrite(sparkWrite).getOrElse(Map.empty)
- val effectiveProperties = properties ++ resolvedWriteProperties
- val parquetSettings =
- IcebergWriteProtoTranslation.buildParquetSettings(effectiveProperties,
createdBy)
+ val effectiveProperties =
+ interpretedBloomFilterProperties(properties ++ resolvedWriteProperties)
+ val parquetPathByIcebergColumnName =
+ if
(IcebergWriteProtoTranslation.hasEnabledBloomFilters(effectiveProperties)) {
+ val resolution =
IcebergReflection.getParquetPathResolution(writeSchema).getOrElse {
+ withFallbackReason(op, "Could not resolve physical Parquet paths for
Bloom filters")
+ return None
+ }
+ val configuredColumns =
+
IcebergWriteProtoTranslation.enabledBloomFilterColumnNames(effectiveProperties)
+ val icebergVersion = IcebergReflection.icebergVersion()
+ val versionParts = icebergVersion.split("[.-]", 3)
+ val version =
+ if (versionParts.length >= 2) {
+ scala.util.Try((versionParts(0).toInt,
versionParts(1).toInt)).toOption
+ } else {
+ None
+ }
+ val resolvedPaths = version match {
+ case Some((1, minor)) if minor < 8 =>
+ // Iceberg 1.5 passes the configured name straight to parquet-mr.
+ configuredColumns.filter(resolution.physicalPaths).map(path =>
path -> path).toMap
+ case Some((1, minor)) if minor < 10 =>
+ // Iceberg 1.8 resolves canonical and short collection aliases
through field ids.
+ configuredColumns.flatMap { configuredName =>
+ IcebergReflection
+ .findFieldIdByName(writeSchema, configuredName)
+ .flatMap(resolution.pathByFieldId.get)
+ .map(configuredName -> _)
+ }.toMap
+ case _ =>
+ // Iceberg 1.10+ keys writer properties by canonical
findColumnName output.
+ resolution.pathByIcebergColumnName
Review Comment:
[P2] On Iceberg 1.10 and 1.11, this branch returns the whole schema map. An
otherwise eligible table with `id INT` and a quoted `order id INT` falls back
to Java when only `id` has a Bloom filter. The rename check identifies the
unrelated `order id` as a configured Bloom column and disables the requested
native write. Intersect the canonical map with `configuredColumns` before
checking for renamed paths.
Evidence: The exact-head resolver returned fallback for `order id` on both
modern runtime branches with only `id=true`. The requested leaf mapped directly
to physical `id`. A native file control had a filter on `id` containing all
1,000 inserted values and no filter on the unconfigured column.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -1241,6 +1241,626 @@ class CometIcebergWriteActionSuite
}
}
+ test("native acceleration: writes configured Iceberg Parquet bloom filters")
{
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(
+ "native_bloom",
+ partitionSpec = "",
+ properties =
Some("'write.parquet.bloom-filter-enabled.column.id'='true'"))
+
+ val expectedIds = 0 until 256
+ assertNativeWriteEngages("native_bloom", expectedIds) {
+ spark.sql(
+ "INSERT INTO cat.db.native_bloom " +
+ "SELECT CAST(id AS INT), CONCAT('region-', CAST(id % 4 AS
STRING)), " +
+ "CAST(id AS DOUBLE) FROM range(256)")
+ }
+
+ assertParquetBloomFilters("native_bloom", enabledColumns = Set("id"))
+
+ }
+ }
+
+ test("Bloom-filter folding config selects the native or JVM Iceberg writer")
{
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val properties =
Some("'write.parquet.bloom-filter-enabled.column.id'='true'")
+ createTable("bloom_folding_enabled", partitionSpec = "", properties)
+ createTable("bloom_folding_disabled", partitionSpec = "", properties)
+
+ assertNativeWriteEngages("bloom_folding_enabled", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_folding_enabled VALUES (1,
'region', 1.0)")
+ }
+ assertNativeWriteDoesNotEngage(
+ "bloom_folding_disabled",
+ Seq(1),
+ bloomFilterFoldingEnabled = false) {
+ spark.sql("INSERT INTO cat.db.bloom_folding_disabled VALUES (1,
'region', 1.0)")
+ }
+ assertParquetBloomFilters("bloom_folding_enabled", enabledColumns =
Set("id"))
+ assertParquetBloomFilters("bloom_folding_disabled", enabledColumns =
Set("id"))
+ }
+ }
+
+ test("enabled=false plus NDV matches JVM behavior for the Iceberg runtime") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val configuredFpp = 0.01
+ val configuredNdv = 1000
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='false', " +
+ s"'write.parquet.bloom-filter-fpp.column.id'='$configuredFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv'")
+ createTable("bloom_false_ndv_native", partitionSpec = "", properties)
+ createTable("bloom_false_ndv_jvm", partitionSpec = "", properties)
+
+ def insert(table: String): Unit = spark.sql(
+ s"INSERT INTO cat.db.$table " +
+ "SELECT CAST(id AS INT), 'region', CAST(id AS DOUBLE) " +
+ s"FROM range(0, $configuredNdv, 1, 1)")
+
+ assertNativeWriteEngages("bloom_false_ndv_native", 0 until
configuredNdv) {
+ insert("bloom_false_ndv_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("bloom_false_ndv_jvm")
+ }
+
+ val nativeHasBloom = parquetBloomFilterPresent("bloom_false_ndv_native",
"id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_false_ndv_jvm", "id")
+ assert(nativeHasBloom == jvmHasBloom)
+ if (icebergSupportsBloomNdv) {
+ assert(nativeHasBloom, "an interpreted NDV must re-enable the Bloom
filter")
+ val native = parquetBloomFilterBytes("bloom_false_ndv_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_false_ndv_jvm", "id")
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ assert(!nativeHasBloom, "an uninterpreted NDV must not override
enabled=false")
+ }
+ }
+ }
+
+ test("enabled=false preserves JVM validation errors for malformed FPP and
NDV") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ Seq(
+ ("fpp", "'write.parquet.bloom-filter-fpp.column.id'='garbage'",
icebergSupportsBloomFpp),
+ ("ndv", "'write.parquet.bloom-filter-ndv.column.id'='garbage'",
icebergSupportsBloomNdv))
+ .foreach { case (suffix, malformedProperty, interpreted) =>
+ val properties =
+ Some(s"'write.parquet.bloom-filter-enabled.column.id'='false',
$malformedProperty")
+ val nativeTable = s"bloom_false_bad_${suffix}_native"
+ val jvmTable = s"bloom_false_bad_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(table, partitionSpec = "", properties = properties)
+ }
+
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table VALUES (1, 'region', 1.0)")
+
+ if (interpreted) {
+ val withComet = intercept[Throwable] {
+ withNativeEnabled(insert(nativeTable))
+ }
+ val withJvm = intercept[Throwable] {
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key ->
"false") {
+ insert(jvmTable)
+ }
+ }
+ val cometCause = exceptionChain(withComet).last
+ val jvmCause = exceptionChain(withJvm).last
+ assert(cometCause.getClass == jvmCause.getClass)
+ assert(cometCause.getMessage == jvmCause.getMessage)
+ } else {
+ assertNativeWriteEngages(nativeTable, Seq(1))(insert(nativeTable))
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key ->
"false") {
+ insert(jvmTable)
+ }
+ assert(
+ !parquetBloomFilterPresent(nativeTable, "id") &&
+ !parquetBloomFilterPresent(jvmTable, "id"),
+ s"uninterpreted $suffix must not enable a Bloom filter")
+ }
+ }
+ }
+ }
+
+ test("max-bytes=32 falls back only when parquet-mr ignores the cap") {
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ val minimumBytes = 32
+ val naturallyMinimumNdv = 1
+ val bindingNdv = 1000000
+ val bindingFpp = 0.0001
+ val parquetMrBindingBytes = 4 * 1024 * 1024
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true', "
+ createTable(
+ "bloom_minimum_native",
+ partitionSpec = "",
+ properties = Some(
+ enabled +
s"'write.parquet.bloom-filter-ndv.column.id'='$naturallyMinimumNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+ createTable(
+ "bloom_minimum_fallback",
+ partitionSpec = "",
+ properties = Some(
+ enabled +
s"'write.parquet.bloom-filter-fpp.column.id'='$bindingFpp', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$bindingNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$minimumBytes'"))
+
+ assertNativeWriteEngages("bloom_minimum_native", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_native VALUES (1,
'region', 1.0)")
+ }
+ assertNativeWriteDoesNotEngage("bloom_minimum_fallback", Seq(1)) {
+ spark.sql("INSERT INTO cat.db.bloom_minimum_fallback VALUES (1,
'region', 1.0)")
+ }
+
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_native", "id").forall(_.length
== minimumBytes))
+ assert(
+ parquetBloomFilterBytes("bloom_minimum_fallback", "id")
+ .forall(_.length == parquetMrBindingBytes))
+ }
+ }
+
+ test("native bloom filters resolve list and map leaves to physical Parquet
paths") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("native_nested_bloom", "native_nested_bloom_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY<STRUCT<a: INT>>,
+ attrs MAP<STRING, STRUCT<b: INT>>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.element.a'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.key'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.value.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10), named_struct('a', 20)),
+ map('small', named_struct('b', 30), 'large', named_struct('b',
40))),
+ (2, array(named_struct('a', 50)), map('medium', named_struct('b',
60)))
+ """)
+
+ assertNativeWriteEngages("native_nested_bloom", Seq(1, 2)) {
+ insert("native_nested_bloom")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("native_nested_bloom_jvm")
+ }
+
+ val physicalColumns =
+ Set("tags.list.element.a", "attrs.key_value.key",
"attrs.key_value.value.b")
+ // Iceberg 1.5 passes the configured logical names directly to Parquet
Java, where they do
+ // not match these physical list/map paths. Iceberg 1.8+ resolves
canonical names first.
+ val expectedColumns =
+ if (icebergVersionAtLeast(1, 8)) physicalColumns else Set.empty[String]
+ assertParquetBloomFilters("native_nested_bloom", enabledColumns =
expectedColumns)
+ assertParquetBloomFilters("native_nested_bloom_jvm", enabledColumns =
expectedColumns)
+ }
+ }
+
+ test("short nested bloom aliases match the Iceberg runtime footer") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("nested_alias_native", "nested_alias_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY<STRUCT<a: INT>>,
+ attrs MAP<STRING, STRUCT<b: INT>>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.a'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10)), map('first', named_struct('b',
20))),
+ (2, array(named_struct('a', 30)), map('second', named_struct('b',
40)))
+ """)
+
+ assertNativeWriteEngages("nested_alias_native", Seq(1, 2)) {
+ insert("nested_alias_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("nested_alias_jvm")
+ }
+
+ val icebergUsesFieldIdResolution =
+ icebergVersionAtLeast(1, 8) && !icebergVersionAtLeast(1, 10)
+ Seq("tags.list.element.a", "attrs.key_value.value.b").foreach {
physicalColumn =>
+ val nativeHasBloom = parquetBloomFilterPresent("nested_alias_native",
physicalColumn)
+ val jvmHasBloom = parquetBloomFilterPresent("nested_alias_jvm",
physicalColumn)
+ assert(nativeHasBloom == jvmHasBloom, s"native and JVM footer differ
for $physicalColumn")
+ assert(
+ jvmHasBloom == icebergUsesFieldIdResolution,
+ s"unexpected short-alias behavior for $physicalColumn on this
Iceberg runtime")
+ }
+ }
+ }
+
+ test("Iceberg 1.5 physical nested bloom paths execute natively and match the
JVM footer") {
+ assumeNativeAcceleration()
+ assume(
+ !icebergVersionAtLeast(1, 8),
+ "physical-path Bloom properties are specific to Iceberg 1.5")
+ withIcebergCatalog { _ =>
+ Seq("physical_path_native", "physical_path_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ id INT,
+ tags ARRAY<STRUCT<a: INT>>,
+ attrs MAP<STRING, STRUCT<b: INT>>
+ ) USING iceberg
+ TBLPROPERTIES (
+
'write.parquet.bloom-filter-enabled.column.tags.list.element.a'='true',
+
'write.parquet.bloom-filter-enabled.column.attrs.key_value.value.b'='true'
+ )
+ """)
+ }
+
+ def insert(table: String): Unit = spark.sql(s"""
+ INSERT INTO $catalog.$ns.$table VALUES
+ (1, array(named_struct('a', 10)), map('first', named_struct('b',
20))),
+ (2, array(named_struct('a', 30)), map('second', named_struct('b',
40)))
+ """)
+
+ assertNativeWriteEngages("physical_path_native", Seq(1, 2)) {
+ insert("physical_path_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("physical_path_jvm")
+ }
+
+ Seq("tags.list.element.a", "attrs.key_value.value.b").foreach {
physicalColumn =>
+ assert(
+ parquetBloomFilterPresent("physical_path_native", physicalColumn),
+ s"expected native Bloom filter for $physicalColumn")
+ assert(
+ parquetBloomFilterPresent("physical_path_jvm", physicalColumn),
+ s"expected JVM Bloom filter for $physicalColumn")
+ }
+ }
+ }
+
+ test("quoted bloom-filter columns renamed by Iceberg fall back to the JVM
footer") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ Seq("bloom_quoted_name", "bloom_quoted_name_jvm").foreach { table =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.$table (
+ `order id` INT
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.order id'='true'
+ )
+ """)
+ }
+
+ val snapshot = withNativeEnabled {
+ captureWrite("bloom_quoted_name") {
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name VALUES (1),
(2)")
+ }
+ }
+ assertExactlyOneCommit(snapshot)
+ val nativeExecs = snapshot.plans.flatMap { plan =>
+ collectWithSubqueries(plan) { case exec: CometIcebergWriteExec => exec
}
+ }
+ assert(
+ nativeExecs.isEmpty,
Review Comment:
[P2] This assertion expects JVM fallback on Spark 3.4/Iceberg 1.5.2, but the
fixture configures logical `order id`. That runtime uses raw Parquet names, so
this spelling matches no physical leaf and the planner permits native
execution. Configure `order_x20id` on 1.5.2 to exercise the intended fallback
and retain the logical spelling on runtimes that resolve it.
Evidence: The 1.5.2 resolution branch returned `Some(Map())` for `order id`
and selected fallback for `order_x20id`. Parquet 1.13.1 footers showed zero
filter bytes for the logical spelling and 1,048,576 for the physical spelling.
This verifies the assertion mismatch at the component level. The full Spark 3.4
suite stopped before test execution.
##########
native/core/src/execution/operators/iceberg_write.rs:
##########
@@ -2310,8 +2313,147 @@ fn build_writer_properties(settings:
&IcebergParquetWriteSettings) -> DFResult<W
.set_dictionary_page_size_limit(settings.dict_size_bytes as usize)
.set_data_page_row_count_limit(settings.page_row_limit as usize)
.set_statistics_enabled(EnabledStatistics::Page)
- .set_statistics_truncate_length(None)
- .build())
+ .set_statistics_truncate_length(None);
+ for bloom_filter in &settings.bloom_filters {
+ let path = parquet_column_path(&bloom_filter.parquet_path);
+ let fpp = bloom_filter.fpp;
+ let ndv = bloom_filter.ndv;
+ let max_bytes = settings.bloom_filter_max_bytes as usize;
+ validate_bloom_filter_inputs(fpp, max_bytes)?;
+ let target_bytes = parquet_mr_bloom_filter_bytes(ndv, fpp, max_bytes);
+ let synthetic_ndv = synthetic_ndv_for_bloom_filter_bytes(target_bytes,
fpp)?;
+ builder = builder
+ .set_column_bloom_filter_enabled(path.clone(), true)
Review Comment:
[P1] A Bloom-enabled `DOUBLE` leaf can make Spark skip a row group
containing matching NaNs. Write 4,096 copies of
`Double.longBitsToDouble(0x7ff8000000000001L)` and read the file with `f =
CAST('NaN' AS DOUBLE)`. Spark considers those values equal, but the native
filter hashes the stored payload and rejects the canonical NaN used by the
predicate. Keep affected floating-point Bloom writes on the Java path until the
native filter can honor Spark’s equality semantics. A predicate read of the
written file would cover this regression.
Evidence: The exact-head property builder and pinned iceberg-rust writer
produced one dictionary-encoded row group. Spark 3.5.9 returned 0 rows with
Bloom pruning and 4,096 with `parquet.filter.bloom.enabled=false` on the
identical file. Both the native no-filter control and the Iceberg 1.11.0 Java
file returned 4,096 with pruning enabled. Java omitted the filter for its
dictionary-only chunk.
--
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]
