NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4372890153
##########
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:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `test:
use physical Bloom names on Iceberg 1.5`. This PR keeps these older runtimes on
Java.
##########
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:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `fix:
normalize legacy Iceberg Bloom paths`. This PR keeps these older runtimes on
Java.
##########
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:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `fix:
fall back for colliding legacy Bloom aliases`. This PR keeps these older
runtimes on Java.
##########
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:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `fix:
preserve legacy Bloom lookup validation`. This PR keeps these older runtimes on
Java.
##########
docs/source/user-guide/latest/iceberg-writes.md:
##########
@@ -154,7 +154,9 @@ A write is eligible only when ALL of the following hold:
| `write.parquet.page-version`
| unset or `v1`
|
| `write.parquet.shred-variants`
| unset or
`false` (Spark 4.x / Iceberg 1.11 resolve this into every parquet write)
|
| `write.parquet.variant-inference-buffer-size`
| any value
(only meaningful when shredding, which is gated)
|
-| `write.parquet.bloom-filter-enabled.column.<col>`
| unset or
`false`
|
+| `write.parquet.bloom-filter-enabled.column.<col>`
| `true` or
`false`; an explicit NDV enables the column even when this value is `false`,
matching Iceberg's property application order
|
+| `write.parquet.bloom-filter-fpp.column.<col>` /
`write.parquet.bloom-filter-ndv.column.<col>`
| For every column named by an `enabled` property, FPP must be a
finite double strictly between 0 and 1 and NDV must be a positive Java long no
greater than `Long.MAX_VALUE / 8`; the Iceberg FPP default is `0.01`
|
Review Comment:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `docs:
describe native Bloom support on legacy Iceberg`. This PR keeps these older
runtimes on Java.
--
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]
