andygrove commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r3973655636
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -283,6 +284,156 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
private val requireNativeSupportedCompressionLevel: TriggerRule = ctx =>
IcebergWriteProtoTranslation.compressionLevelRejection(ctx.properties)
+ // These are Apache Parquet Java BlockSplitBloomFilter implementation
bounds, not Iceberg
+ // TableProperties constants, so they cannot be obtained through
IcebergReflection:
+ // scalastyle:off line.size.limit
+ //
https://github.com/apache/parquet-java/blob/78a8d3230eb4769db93de5f2f2e18363c04cae81/parquet-column/src/main/java/org/apache/parquet/column/values/bloomfilter/BlockSplitBloomFilter.java#L40-L50
+ // scalastyle:on line.size.limit
+ private val MinBloomFilterBytes = 32
+ private val MaxBloomFilterBytes = 128 * 1024 * 1024
+ private val BloomFilterHashProbes = 8
+ private val MaxNonOverflowingBloomFilterNdv = Long.MaxValue /
BloomFilterHashProbes
+
+ /**
+ * Keep only Bloom shape properties interpreted by the Iceberg runtime on
the classpath. Older
+ * Iceberg releases leave these table properties untouched but do not pass
them to parquet-mr.
+ * Ignoring them here preserves that version's JVM-writer behavior while
allowing the remaining
+ * supported Bloom configuration to execute natively.
+ */
+ private def interpretedBloomFilterProperties(
+ properties: Map[String, String]): Map[String, String] = {
+ val unsupportedPrefixes = Seq(
+ PropertyKeys.ParquetBloomFilterColumnFppPrefix ->
+ "PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX",
+ PropertyKeys.ParquetBloomFilterColumnNdvPrefix ->
+ "PARQUET_BLOOM_FILTER_COLUMN_NDV_PREFIX").collect {
+ case (prefix, constant) if
IcebergReflection.tablePropertyConstantOpt(constant).isEmpty =>
+ prefix
+ }
+ properties.filterNot { case (key, _) =>
unsupportedPrefixes.exists(key.startsWith) }
+ }
+
+ /**
+ * parquet-rs 58.x represents Bloom filters as a power-of-two number of
bytes. parquet-mr
+ * accepts arbitrary caps and, when one binds, serializes that exact length.
Keep those writes
+ * on the classic path instead of silently changing the number of usable
Bloom blocks.
+ */
+ private val requireNativeSupportedBloomFilterProperties: TriggerRule = ctx
=> {
+ val properties = interpretedBloomFilterProperties(ctx.properties)
+ val maxRejection =
+ properties.get(PropertyKeys.ParquetBloomFilterMaxBytes).flatMap { raw =>
+ scala.util.Try(java.lang.Integer.parseInt(raw)).toOption match {
+ case None => Some(s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$raw
is not a Java int")
+ case Some(value)
+ if value < MinBloomFilterBytes || value > MaxBloomFilterBytes ||
+ (value & (value - 1)) != 0 =>
+ Some(
+ s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$value must be a
power of two " +
+ s"in [$MinBloomFilterBytes, $MaxBloomFilterBytes] for native
writes")
+ case Some(_) => None
+ }
+ }
+
+ maxRejection.orElse {
+ val maxBytes = properties
+ .get(PropertyKeys.ParquetBloomFilterMaxBytes)
+ .flatMap(raw =>
scala.util.Try(java.lang.Integer.parseInt(raw)).toOption)
+ .getOrElse(IcebergWriteProtoTranslation.Defaults.BloomFilterMaxBytes)
+ // Iceberg visits every enabled-prefix entry and applies enabled, FPP,
then NDV. Validate
+ // the associated shape properties even for enabled=false; a valid NDV
also re-enables the
+ // filter in parquet-mr.
+ val configured = properties.iterator.collect {
+ case (key, _) if
key.startsWith(PropertyKeys.BloomFilterColumnEnabledPrefix) =>
+ key.substring(PropertyKeys.BloomFilterColumnEnabledPrefix.length)
+ }.toSeq
+ configured.iterator
+ .flatMap { column =>
+ val fppKey = PropertyKeys.ParquetBloomFilterColumnFppPrefix + column
+ val ndvKey = PropertyKeys.ParquetBloomFilterColumnNdvPrefix + column
+ val parsedFpp = properties.get(fppKey) match {
+ case Some(raw) =>
scala.util.Try(java.lang.Double.parseDouble(raw)).toOption
+ case None =>
Some(IcebergWriteProtoTranslation.Defaults.BloomFilterFpp)
+ }
+ val parsedNdv = properties
+ .get(ndvKey)
+ .flatMap(raw =>
scala.util.Try(java.lang.Long.parseLong(raw)).toOption)
+ val fppError = properties.get(fppKey).flatMap { raw =>
+ parsedFpp match {
+ case Some(value)
+ if value > 0.0d && value < 1.0d &&
java.lang.Double.isFinite(value) &&
+ bloomFilterSizesRepresentable(maxBytes, value) =>
+ None
+ case Some(value)
+ if value > 0.0d && value < 1.0d &&
java.lang.Double.isFinite(value) =>
+ Some(s"$fppKey=$raw cannot represent the configured native
Bloom sizes")
+ case _ => Some(s"$fppKey=$raw must be a finite double strictly
between 0 and 1")
+ }
+ }
+ val ndvError = properties.get(ndvKey).flatMap { raw =>
+ parsedNdv match {
+ case Some(value) if value > 0L && value <=
MaxNonOverflowingBloomFilterNdv => None
+ case Some(value) if value > MaxNonOverflowingBloomFilterNdv =>
+ Some(s"$ndvKey=$raw exceeds $MaxNonOverflowingBloomFilterNdv;
" +
+ "parquet-mr Bloom sizing may overflow")
+ case _ => Some(s"$ndvKey=$raw must be a positive Java long")
+ }
+ }
+ val ignoredMinimumCapError = parsedNdv.collect {
+ case ndv
+ if maxBytes == MinBloomFilterBytes && ndv > 0L &&
+ ndv <= MaxNonOverflowingBloomFilterNdv &&
+ parsedFpp.exists(
+ parquetMrRequestedBloomFilterBytes(ndv, _) >
MinBloomFilterBytes) =>
+
s"${PropertyKeys.ParquetBloomFilterMaxBytes}=$MinBloomFilterBytes is ignored by
" +
+ s"parquet-mr for $ndvKey=$ndv"
+ }
+ Seq(fppError, ndvError, ignoredMinimumCapError).flatten
+ }
+ .toSeq
+ .headOption
+ }
+ }
+
+ // Planning-time counterpart of the native inverse-NDV check. A target B is
safely encoded by
+ // aiming at 3B/4, in the interior of parquet-rs's (B/2, B] round-up
interval. Requiring every
+ // power-of-two through the configured cap is conservative and keeps
pathological-but-valid
+ // floating-point FPPs on the JVM path rather than discovering them after
task launch.
+ private def bloomFilterSizesRepresentable(maxBytes: Int, fpp: Double):
Boolean = {
Review Comment:
The encoding here is sound, so this is about drift rather than a bug.
`bloomFilterSizesRepresentable` and `parquetMrRequestedBloomFilterBytes`
reimplement the same arithmetic as `parquet_rs_bloom_filter_bytes` and
`parquet_mr_bloom_filter_bytes` on the native side, and nothing ties the two
together. The Rust `#[test]`s cover the Rust half, and these two are private
with no unit test at all, only indirect coverage through the detection suite.
If someone edits one side and the gate ends up saying "representable" where
`synthetic_ndv_for_bloom_filter_bytes` returns `Err`, the write does not fall
back to the classic writer, it fails inside the task.
Could we add one shared table of `(ndv, fpp, max-bytes)` to expected bytes
and assert it from both sides? `IcebergWriteProtoTranslationSuite` is a plain
`AnyFunSuite`, so it can host the Scala half cheaply if these become
`private[operator]`. While you are there, the successful branch of the binary
search is never exercised. `impossible_synthetic_ndv_is_rejected` does enter
the loop, but only on the degenerate path where every candidate returns 32
bytes, so the `Ok(low)` return has no coverage.
##########
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:
This row states the FPP and NDV rules unconditionally, but
`interpretedBloomFilterProperties` makes them depend on the Iceberg runtime. On
1.5.2 both prefixes are dropped before validation, on 1.8.1 and 1.10.0 only NDV
is, and only 1.11 interprets both. Someone on Iceberg 1.10 who sets
`bloom-filter-ndv.column.x` gets neither the requested sizing nor a fallback,
and the table as written does not predict that. Could the row name the versions
that interpret each property, the way the `write.parquet.shred-variants` row
already calls out Spark 4.x and Iceberg 1.11?
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -600,6 +604,453 @@ class CometIcebergWriteActionSuite
}
}
+ test("native acceleration: writes configured Iceberg Parquet bloom filters")
{
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(
+ warehouseDir,
+ "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("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(warehouseDir, "bloom_false_ndv_native", partitionSpec = "",
properties)
+ createTable(warehouseDir, "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(warehouseDir, 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(
+ warehouseDir,
+ "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(
+ warehouseDir,
+ "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 { _ =>
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.native_nested_bloom (
+ id INT,
+ tags ARRAY<STRING>,
+ attrs MAP<STRING, INT>
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.parquet.bloom-filter-enabled.column.tags.element'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.key'='true',
+ 'write.parquet.bloom-filter-enabled.column.attrs.value'='true'
+ )
+ """)
+
+ assertNativeWriteEngages("native_nested_bloom", Seq(1, 2)) {
+ spark.sql("""
+ INSERT INTO cat.db.native_nested_bloom VALUES
+ (1, array('red', 'green'), map('small', 10, 'large', 20)),
+ (2, array('blue'), map('medium', 30))
+ """)
+ }
+
+ assertParquetBloomFilters(
+ "native_nested_bloom",
+ enabledColumns = Set("tags.list.element", "attrs.key_value.key",
"attrs.key_value.value"))
+ }
+ }
+
+ 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,
+ s"expected the sanitized Bloom-filter path to fall back,
plans:\n${snapshot.plans.mkString("\n--\n")}")
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ spark.sql(s"INSERT INTO $catalog.$ns.bloom_quoted_name_jvm VALUES (1),
(2)")
+ }
+ val fallbackHasBloom = parquetBloomFilterPresent("bloom_quoted_name",
"order_x20id")
+ val jvmHasBloom = parquetBloomFilterPresent("bloom_quoted_name_jvm",
"order_x20id")
+ assert(
+ fallbackHasBloom == jvmHasBloom,
+ "expected the fallback write to match the Parquet Java footer")
+ if (icebergSupportsBloomFpp) {
+ assert(jvmHasBloom, "expected this Iceberg version to write the
quoted-column filter")
+ }
+ }
+ }
+
+ test("native bloom filter is byte-identical to parquet-mr when max-bytes
binds") {
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // This NDV/FPP pair requests a 128 MiB allocation before max-bytes is
applied. The 4 KiB
+ // maximum must therefore bind on both writers rather than merely
coinciding with the
+ // naturally selected size.
+ val configuredNdv = 100000000L
+ val bindingMaxBytes = 4 * 1024
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='0.01', " +
+ s"'write.parquet.bloom-filter-ndv.column.id'='$configuredNdv', " +
+ s"'write.parquet.bloom-filter-max-bytes'='$bindingMaxBytes'")
+ createTable(warehouseDir, "bloom_identity_native", partitionSpec = "",
properties)
+ createTable(warehouseDir, "bloom_identity_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) FROM range(0,
10000, 1, 1)")
+
+ assertNativeWriteEngages("bloom_identity_native", 0 until 10000) {
+ insert("bloom_identity_native")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")
{
+ insert("bloom_identity_jvm")
+ }
+
+ val native = parquetBloomFilterBytes("bloom_identity_native", "id")
+ val jvm = parquetBloomFilterBytes("bloom_identity_jvm", "id")
+ assert(
+ native.nonEmpty && native.forall(_.length == bindingMaxBytes),
+ s"native Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(
+ jvm.nonEmpty && jvm.forall(_.length == bindingMaxBytes),
+ s"JVM Bloom filter must be capped at $bindingMaxBytes bytes")
+ assert(native.size == jvm.size, "native and JVM writes must produce the
same file count")
+ assert(
+ native.zip(jvm).forall { case (left, right) =>
java.util.Arrays.equals(left, right) },
+ "expected byte-identical capped SBBF bitsets for identical values and
allocation")
+ }
+ }
+
+ test("native bloom sizing matches JVM for shape properties supported by the
Iceberg runtime") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ val enabled = "'write.parquet.bloom-filter-enabled.column.id'='true'"
+ val cases: Seq[(String, String, Boolean, Boolean)] = Seq(
+ (
+ "fpp_only",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.02'",
+ icebergSupportsBloomFpp,
+ false),
+ (
+ "ndv_only",
+ s"$enabled, 'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomNdv,
+ true),
+ (
+ "both",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.005', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true),
+ // The requested NDV/FPP needs far more than 64 bytes. Like
parquet-mr, max wins and the
+ // target FPP becomes impossible to guarantee, but membership must
remain correct.
+ (
+ "binding_cap",
+ s"$enabled, 'write.parquet.bloom-filter-fpp.column.id'='0.0001', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='1000000', " +
+ "'write.parquet.bloom-filter-max-bytes'='64'",
+ icebergSupportsBloomFpp && icebergSupportsBloomNdv,
+ true))
+
+ cases.filter(_._3).foreach { case (suffix, properties, _,
expectByteIdentity) =>
+ val nativeTable = s"bloom_shape_${suffix}_native"
+ val jvmTable = s"bloom_shape_${suffix}_jvm"
+ Seq(nativeTable, jvmTable).foreach { table =>
+ createTable(warehouseDir, table, partitionSpec = "", properties =
Some(properties))
+ }
+ // Keep the explicit-NDV cases at their estimated cardinality so
parquet-rs does not
+ // fold their allocation before byte-parity is checked.
+ val insertedCardinality = if (suffix == "ndv_only" || suffix ==
"both") 1000 else 256
+ val ids = 0 until insertedCardinality
+ def insert(table: String): Unit =
+ spark.sql(s"INSERT INTO cat.db.$table SELECT CAST(id AS INT),
'region', " +
+ s"CAST(id AS DOUBLE) FROM range(${ids.start}, ${ids.end}, 1, 1)")
+
+ assertNativeWriteEngages(nativeTable, ids) {
+ insert(nativeTable)
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key ->
"false") {
+ insert(jvmTable)
+ }
+ val native = parquetBloomFilterBytes(nativeTable, "id")
+ val jvm = parquetBloomFilterBytes(jvmTable, "id")
+ if (expectByteIdentity) {
+ assert(native.map(_.length) == jvm.map(_.length))
+ assert(native.zip(jvm).forall { case (left, right) =>
+ java.util.Arrays.equals(left, right)
+ })
+ } else {
+ // Without explicit NDV, parquet-rs can fold to the observed
cardinality while Parquet
+ // Java retains its initial maximum allocation. Folding is safe for
readers, so require
+ // only that native uses no more space and still contains every
inserted value.
+ assert(native.zip(jvm).forall { case (left, right) => left.length <=
right.length })
+ assertParquetBloomContainsInts(nativeTable, "id", ids)
+ }
+ if (suffix == "binding_cap") {
+ assert(native.forall(_.length == 64), s"expected binding 64-byte cap
for $nativeTable")
+ assertParquetBloomContainsInts(nativeTable, "id", ids)
+ }
+ }
+ }
+ }
+
+ test("explicit underestimated NDV retains parquet-mr allocation precedence")
{
+ assumeNativeAcceleration()
+ assumeIcebergBloomShapeProperties()
+ withIcebergCatalog { warehouseDir =>
+ // max-bytes is only a cap in parquet-mr; it does not enlarge a filter
whose explicit NDV
+ // was underestimated. This comparison prevents Comet from silently
replacing the user's
+ // NDV with an artificial NDV derived from the much larger maximum.
+ val properties = Some(
+ "'write.parquet.bloom-filter-enabled.column.id'='true', " +
+ "'write.parquet.bloom-filter-fpp.column.id'='0.01', " +
+ "'write.parquet.bloom-filter-ndv.column.id'='10', " +
+ "'write.parquet.bloom-filter-max-bytes'='67108864'")
+ createTable(warehouseDir, "bloom_low_ndv_native", partitionSpec = "",
properties)
+ createTable(warehouseDir, "bloom_low_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) FROM range(0,
4096, 1, 1)")
+
+ assertNativeWriteEngages("bloom_low_ndv_native", 0 until 4096) {
+ insert("bloom_low_ndv_native")
+ }
+ insert("bloom_low_ndv_jvm")
Review Comment:
`insert("bloom_low_ndv_jvm")` is the only JVM-side insert in the new tests
that is not wrapped in
`withSQLConf(CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> "false")`. It
works today because the conf defaults to false and the suite's `sparkConf` does
not set it, so the comparison really is native against parquet-mr. If that
default ever flips, this test quietly starts comparing Comet with Comet and
passes for the wrong reason. Could it use the same explicit wrapper as its
siblings?
--
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]
