From b1967d3a2eeb752e4301c8ba1586841f8e7a2cb4 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 13:42:56 -0400 Subject: [PATCH 01/12] adjust delete file pool to reduce redundancy --- native/core/src/execution/planner.rs | 65 +++++++---- native/proto/src/proto/operator.proto | 9 +- .../operator/CometIcebergNativeScan.scala | 22 +++- .../comet/CometIcebergNativeSuite.scala | 108 ++++++++++++++++++ 4 files changed, 176 insertions(+), 28 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index a23a7b1f67..c0acba6f31 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3674,35 +3674,52 @@ fn parse_file_scan_tasks_from_common( }) .collect(); + // Flat pool of unique delete files. A delete file applies to many data files under Iceberg's + // default partition delete granularity, so DeleteFileList entries reference this pool by index + // rather than embedding copies. + let delete_file_pool: Vec = proto_common + .delete_file_pool + .iter() + .map(|del| { + let file_type = match del.content_type.as_str() { + "POSITION_DELETES" => iceberg::spec::DataContentType::PositionDeletes, + "EQUALITY_DELETES" => iceberg::spec::DataContentType::EqualityDeletes, + other => { + return Err(GeneralError(format!( + "Invalid delete content type '{}'", + other + ))) + } + }; + + Ok(iceberg::scan::FileScanTaskDeleteFile { + file_path: del.file_path.clone(), + file_type, + // Not serialized; filled in by IcebergScanExec::fill_delete_file_sizes. + file_size_in_bytes: 0, + partition_spec_id: del.partition_spec_id, + equality_ids: if del.equality_ids.is_empty() { + None + } else { + Some(del.equality_ids.clone()) + }, + }) + }) + .collect::, ExecutionError>>()?; + let delete_files_cache: Vec> = proto_common .delete_files_pool .iter() .map(|list| { - list.delete_files + list.delete_file_indices .iter() - .map(|del| { - let file_type = match del.content_type.as_str() { - "POSITION_DELETES" => iceberg::spec::DataContentType::PositionDeletes, - "EQUALITY_DELETES" => iceberg::spec::DataContentType::EqualityDeletes, - other => { - return Err(GeneralError(format!( - "Invalid delete content type '{}'", - other - ))) - } - }; - - Ok(iceberg::scan::FileScanTaskDeleteFile { - file_path: del.file_path.clone(), - file_type, - // Not serialized; filled in by IcebergScanExec::fill_delete_file_sizes. - file_size_in_bytes: 0, - partition_spec_id: del.partition_spec_id, - equality_ids: if del.equality_ids.is_empty() { - None - } else { - Some(del.equality_ids.clone()) - }, + .map(|&idx| { + delete_file_pool.get(idx as usize).cloned().ok_or_else(|| { + GeneralError(format!( + "Invalid delete_file_index: {} (pool size: {})", + idx, + delete_file_pool.len() + )) }) }) .collect::, ExecutionError>>() diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 00158644d1..03e6eaecb8 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -216,6 +216,12 @@ message IcebergScanCommon { // isolated provider instances. Empty string when the table has no catalog identity (e.g. // HadoopTables loaded by raw path). string catalog_name = 13; + + // Flat pool of unique delete files, referenced by index from DeleteFileList. Iceberg's default + // partition delete granularity has one delete file apply to many data files, so interning each + // delete file here (rather than embedding copies per delete_files_pool set) keeps the broadcast + // common message from growing with the number of references. + repeated IcebergDeleteFile delete_file_pool = 14; } message IcebergScan { @@ -233,7 +239,8 @@ message ProjectFieldIdList { // Helper message for deduplicating delete file lists message DeleteFileList { - repeated IcebergDeleteFile delete_files = 1; + // Indices into IcebergScanCommon.delete_file_pool + repeated uint32 delete_file_indices = 1; } // Iceberg FileScanTask containing data file, delete files, and residual filter diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 879762818f..30f6083082 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -727,8 +727,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val nameMappingToPoolIndex = mutable.HashMap[String, Int]() val projectFieldIdsToPoolIndex = mutable.HashMap[Seq[Int], Int]() val partitionDataToPoolIndex = mutable.HashMap[String, Int]() + // Individual delete files are interned into a flat pool keyed by path (a delete file's path is + // its identity); deleteFilesToPoolIndex then dedups the per-task sets as lists of indices into + // that pool. One delete file applies to many data files under Iceberg's default partition + // delete granularity, so interning avoids re-serializing it once per referencing FileScanTask. + val deleteFileToPoolIndex = mutable.HashMap[String, Int]() val deleteFilesToPoolIndex = - mutable.HashMap[Seq[OperatorOuterClass.IcebergDeleteFile], Int]() + mutable.HashMap[Seq[Int], Int]() val residualToPoolIndex = mutable.HashMap[Option[Expr], Int]() val perPartitionBuilders = mutable.ArrayBuffer[OperatorOuterClass.IcebergScan]() @@ -907,11 +912,21 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val deleteFilesList = extractDeleteFilesList(task, contentFileClass, fileScanTaskClass) if (deleteFilesList.nonEmpty) { + // Intern each delete file into the flat pool, then dedup this task's set as the + // resulting list of pool indices. + val deleteFileIndices = deleteFilesList.map { df => + deleteFileToPoolIndex.getOrElseUpdate( + df.getFilePath, { + val idx = deleteFileToPoolIndex.size + commonBuilder.addDeleteFilePool(df) + idx + }) + } val deleteFilesIdx = deleteFilesToPoolIndex.getOrElseUpdate( - deleteFilesList, { + deleteFileIndices, { val idx = deleteFilesToPoolIndex.size val listBuilder = OperatorOuterClass.DeleteFileList.newBuilder() - deleteFilesList.foreach(df => listBuilder.addDeleteFiles(df)) + deleteFileIndices.foreach(idx => listBuilder.addDeleteFileIndices(idx)) commonBuilder.addDeleteFilesPool(listBuilder.build()) idx }) @@ -996,6 +1011,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit nameMappingToPoolIndex.size, projectFieldIdsToPoolIndex.size, partitionDataToPoolIndex.size, + deleteFileToPoolIndex.size, deleteFilesToPoolIndex.size, residualToPoolIndex.size) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 7b9bd6b3d6..58e1e1452d 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -21,6 +21,7 @@ package org.apache.comet import java.io.File import java.net.URI +import java.nio.charset.StandardCharsets.UTF_8 import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -39,6 +40,7 @@ import org.apache.spark.sql.types.{StringType, TimestampType} import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} import org.apache.comet.iceberg.RESTCatalogHelper +import org.apache.comet.serde.OperatorOuterClass import org.apache.comet.testing.{FuzzDataGenerator, SchemaGenOptions} /** @@ -96,6 +98,24 @@ class CometIcebergNativeSuite s"Plan:\n$cometPlan") } + /** Counts non-overlapping occurrences of `needle` within `haystack`. */ + private def countByteOccurrences(haystack: Array[Byte], needle: Array[Byte]): Int = { + require(needle.nonEmpty, "needle must be non-empty") + var count = 0 + var i = 0 + while (i <= haystack.length - needle.length) { + var j = 0 + while (j < needle.length && haystack(i + j) == needle(j)) j += 1 + if (j == needle.length) { + count += 1 + i += needle.length + } else { + i += 1 + } + } + count + } + test("create and query simple Iceberg table with Hadoop catalog") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -575,6 +595,94 @@ class CometIcebergNativeSuite } } + // Under Iceberg's default partition delete granularity, one position-delete file applies to every + // data file in the partition with a compatible sequence number (DeleteFileIndex.forDataFile). + // Interleaving inserts and deletes staggers data-file sequence numbers, so each FileScanTask sees + // a different subset of the same delete files. A shared delete file must be serialized once in + // the broadcast IcebergScanCommon, not once per referencing set: duplicating it is quadratic in + // the number of delete commits and can overflow protobuf's 2 GiB message limit + // (getSerializedSize() returns int, so it wraps to a negative array length). + test("delete file pool does not duplicate shared delete files (serde size)") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.delete_pool_test (id INT, name STRING) + USING iceberg + TBLPROPERTIES ( + 'write.delete.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read', + 'write.delete.granularity' = 'partition' + ) + """) + + // Interleave inserts and deletes. Each INSERT lands a data file at an increasing sequence + // number; each DELETE lands one partition-granularity position-delete file that references + // (and applies to) every prior data file still holding a matching row. Data file written in + // round r therefore sees delete files from rounds r..N-1, a distinct set per data file. + val rounds = 5 + for (r <- 0 until rounds) { + val base = r * 100 + val values = (1 to 50).map(i => s"(${base + i}, 'n${base + i}')").mkString(", ") + spark.sql(s"INSERT INTO test_cat.db.delete_pool_test VALUES $values") + // Delete one still-present row from every batch inserted so far so this delete file + // references multiple data files (partition granularity) and applies to all of them. The + // per-round offset (r + 2) is unique, so no row is deleted twice (which would make the + // delete match only the newest batch and defeat the staggering). + val ids = (0 to r).map(b => b * 100 + (r + 2)).mkString(", ") + spark.sql(s"DELETE FROM test_cat.db.delete_pool_test WHERE id IN ($ids)") + } + + val (_, cometPlan) = + checkSparkAnswer("SELECT * FROM test_cat.db.delete_pool_test ORDER BY id") + val scans = collectIcebergNativeScans(cometPlan) + assert(scans.length == 1, s"expected one native scan, got ${scans.length}\n$cometPlan") + + val commonBytes = scans.head.commonData + val common = OperatorOuterClass.IcebergScanCommon.parseFrom(commonBytes) + + val distinctPaths = + common.getDeleteFilePoolList.asScala.map(_.getFilePath).toSeq + val totalReferences = + common.getDeleteFilesPoolList.asScala.map(_.getDeleteFileIndicesCount).sum + + // Guard against a vacuous pass: we must actually exercise the shared-delete-file case + // (more references than distinct files, i.e. at least one file shared across tasks). + assert( + distinctPaths.size >= 2 && totalReferences > distinctPaths.size, + s"test setup produced too few shared delete files: ${distinctPaths.size} files, " + + s"$totalReferences references") + + // Count raw-byte occurrences of each path so the check is agnostic to how the pool is + // structured (set-of-copies before the fix, flat index pool after). + val duplicated = distinctPaths + .map(p => p -> countByteOccurrences(commonBytes, p.getBytes(UTF_8))) + .filter(_._2 > 1) + + logInfo( + s"IcebergScanCommon = ${commonBytes.length} bytes, " + + s"${distinctPaths.size} distinct delete files, " + + s"$totalReferences delete-file references") + + assert( + duplicated.isEmpty, + s"delete files serialized more than once in IcebergScanCommon " + + s"(${commonBytes.length} bytes): " + + duplicated.map { case (p, c) => s"$p x$c" }.mkString(", ")) + + spark.sql("DROP TABLE test_cat.db.delete_pool_test") + } + } + } + test("bytes_scanned includes delete file I/O") { assume(icebergAvailable, "Iceberg not available in classpath") From 6262e1a70fc7977cf6f6882d44357e9d7b7b4c53 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 13:54:51 -0400 Subject: [PATCH 02/12] add instrumentation, fix format issue --- .../operator/CometIcebergNativeScan.scala | 69 ++++++++++++++++--- .../comet/CometIcebergNativeSuite.scala | 2 +- 2 files changed, 62 insertions(+), 9 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 30f6083082..513e1842bf 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -19,6 +19,8 @@ package org.apache.comet.serde.operator +import java.nio.charset.StandardCharsets.UTF_8 + import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -1027,16 +1029,67 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } - val partitionDataPoolBytes = commonBuilder.getPartitionDataPoolList.asScala - .map(_.getSerializedSize) - .sum + // Per-pool byte sizes to diagnose an oversized common message. Sizes sum as Long because a + // single pool at or past protobuf's 2 GiB message limit overflows the int getSerializedSize, + // and the logging runs before toByteArray so the breakdown survives even if that allocation + // fails. String pools carry JSON, whose serialized size is its UTF-8 length. + def sumSizes(sizes: Iterator[Int]): Long = sizes.map(_.toLong).sum + def sumStrBytes(strings: mutable.Buffer[String]): Long = + strings.iterator.map(_.getBytes(UTF_8).length.toLong).sum + val commonPoolBytes: Seq[(String, Int, Long)] = Seq( + ( + "schema", + commonBuilder.getSchemaPoolCount, + sumStrBytes(commonBuilder.getSchemaPoolList.asScala)), + ( + "partition_type", + commonBuilder.getPartitionTypePoolCount, + sumStrBytes(commonBuilder.getPartitionTypePoolList.asScala)), + ( + "partition_spec", + commonBuilder.getPartitionSpecPoolCount, + sumStrBytes(commonBuilder.getPartitionSpecPoolList.asScala)), + ( + "name_mapping", + commonBuilder.getNameMappingPoolCount, + sumStrBytes(commonBuilder.getNameMappingPoolList.asScala)), + ( + "project_field_ids", + commonBuilder.getProjectFieldIdsPoolCount, + sumSizes( + commonBuilder.getProjectFieldIdsPoolList.asScala.iterator.map(_.getSerializedSize))), + ( + "partition_data", + commonBuilder.getPartitionDataPoolCount, + sumSizes( + commonBuilder.getPartitionDataPoolList.asScala.iterator.map(_.getSerializedSize))), + ( + "delete_file", + commonBuilder.getDeleteFilePoolCount, + sumSizes(commonBuilder.getDeleteFilePoolList.asScala.iterator.map(_.getSerializedSize))), + ( + "delete_files_set", + commonBuilder.getDeleteFilesPoolCount, + sumSizes(commonBuilder.getDeleteFilesPoolList.asScala.iterator.map(_.getSerializedSize))), + ( + "residual", + commonBuilder.getResidualPoolCount, + sumSizes(commonBuilder.getResidualPoolList.asScala.iterator.map(_.getSerializedSize)))) + + val perPartitionSizes = perPartitionBuilders.map(_.getSerializedSize.toLong) + val perPartitionTotal = perPartitionSizes.sum + val perPartitionMax = if (perPartitionSizes.isEmpty) 0L else perPartitionSizes.max logInfo(s"IcebergScan: $totalTasks tasks, ${allPoolSizes.size} pools ($avgDedup% avg dedup)") - if (partitionDataToPoolIndex.nonEmpty) { - logInfo( - s" Partition data pool: ${partitionDataToPoolIndex.size} unique values, " + - s"$partitionDataPoolBytes bytes (protobuf)") - } + logInfo( + " Common pools (unique/bytes): " + + commonPoolBytes + .map { case (name, count, bytes) => s"$name=$count/$bytes" } + .mkString(", ") + + s"; total=${commonPoolBytes.map(_._3).sum} bytes") + logInfo( + s" Per-partition messages: ${perPartitionBuilders.size}, " + + s"total=$perPartitionTotal bytes, max=$perPartitionMax bytes") val commonBytes = commonBuilder.build().toByteArray val perPartitionBytes = perPartitionBuilders.map(_.toByteArray).toArray diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 58e1e1452d..c6746954f7 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -674,7 +674,7 @@ class CometIcebergNativeSuite assert( duplicated.isEmpty, - s"delete files serialized more than once in IcebergScanCommon " + + "delete files serialized more than once in IcebergScanCommon " + s"(${commonBytes.length} bytes): " + duplicated.map { case (p, c) => s"$p x$c" }.mkString(", ")) From 290b44b94dc8890fe2f5a25701b419ff10b92e48 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 14:50:45 -0400 Subject: [PATCH 03/12] update docs --- .../serde/operator/CometIcebergNativeScan.scala | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 513e1842bf..f09e35d4fa 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -706,6 +706,19 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit * Called after doPrepare() has resolved DPP subqueries. Builds pools and per-partition data in * one pass from the DPP-filtered partitions. * + * The result splits into a common block plus one block per Spark partition. The common block + * holds everything shared across partitions and is broadcast once per executor in the stage + * task binary; each per-partition block ships with its task. Protobuf serializes by value and + * shares nothing between repeated fields, so shared values (schemas, specs, partition data, + * residuals, delete files) are interned into pools in the common block and referenced by index. + * Pooling also keeps a message under protobuf's 2 GiB limit (getSerializedSize returns int and + * wraps past it), which matters for anything that scales with the number of tasks. + * + * Delete files use two levels: a flat pool of unique files, plus a per-task list of indices + * into it. One delete file applies to many data files under Iceberg's default partition delete + * granularity, so this stores each file once and references it many times, as Iceberg's own + * DeleteFileIndex does. + * * @param scanExec * The BatchScanExec whose inputRDD contains the DPP-filtered partitions * @param output From f88a972be6f02a5e0fa84279f455bafb11fdc675 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 18:36:16 -0400 Subject: [PATCH 04/12] =?UTF-8?q?Core=20refactor=20(Iceberg-native=20expre?= =?UTF-8?q?ssion=20serde):=20-=20proto:=20IcebergLiteral/IcebergDecimal/Ic?= =?UTF-8?q?ebergPredicate=20+=20IcebergPredicateOperator=20(mirrors=20iceb?= =?UTF-8?q?erg-rust);=20PartitionValue=20unified=20onto=20IcebergLiteral;?= =?UTF-8?q?=20residual=5Fpool=20=E2=86=92=20repeated=20IcebergPredicate.?= =?UTF-8?q?=20-=20Rust:=20iceberg=5Fpredicate=5Fto=5Fpredicate=20+=20icebe?= =?UTF-8?q?rg=5Fliteral=5Fto=5Fdatum,=20partition=5Fvalue=5Fto=5Fliteral?= =?UTF-8?q?=20rewrite,=20rewrite=5Fnot()=20before=20bind,=20old=20Spark-ex?= =?UTF-8?q?pr=20conversion=20cluster=20deleted.=20-=20Scala:=20icebergExpr?= =?UTF-8?q?ToProto=20+=20proto=20builders=20+=20predicateLiteralToProto,?= =?UTF-8?q?=20partitionValueToProto/icebergLiteralToProto,=20residual=20si?= =?UTF-8?q?te=20+=20pool=20type,=20old=20conversion=20cluster=20deleted,?= =?UTF-8?q?=20imports=20cleaned=20(ByteString/Base64=20imported).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests: - New predicate fuzz test in CometFuzzIcebergSuite (sample-from-data + Column API over primitive columns: comparisons, !=, explicit NOT, IN/NOT IN, IS [NOT] NULL, AND/OR) — the coverage that would have caught the NOT bug. - Residual dedup test tightened: asserts one pool entry shared by many task references (no longer vacuously passable), logInfo removed. - Delete-file dedup test: logInfo removed. --- native/core/src/execution/planner.rs | 378 ++++++------------ native/proto/src/proto/operator.proto | 90 ++++- .../operator/CometIcebergNativeScan.scala | 367 ++++++++++------- .../apache/comet/CometFuzzIcebergSuite.scala | 73 ++-- .../comet/CometIcebergNativeSuite.scala | 70 +++- 5 files changed, 532 insertions(+), 446 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index c0acba6f31..389a92b7ec 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3537,59 +3537,32 @@ fn align_shuffle_writer_input( fn partition_value_to_literal( proto_value: &spark_operator::PartitionValue, ) -> Result, ExecutionError> { - use spark_operator::partition_value::Value; + use spark_operator::iceberg_literal::Value; - if proto_value.is_null { + let Some(lit) = proto_value.literal.as_ref() else { + return Err(GeneralError( + "PartitionValue has no literal set".to_string(), + )); + }; + + if lit.is_null { return Ok(None); } - let literal = match &proto_value.value { + let literal = match &lit.value { + Some(Value::BoolVal(v)) => iceberg::spec::Literal::bool(*v), Some(Value::IntVal(v)) => iceberg::spec::Literal::int(*v), Some(Value::LongVal(v)) => iceberg::spec::Literal::long(*v), - Some(Value::DateVal(v)) => { - // Convert i64 to i32 for date (days since epoch) - let days = (*v) - .try_into() - .map_err(|_| GeneralError(format!("Date value out of range: {}", v)))?; - iceberg::spec::Literal::date(days) - } + Some(Value::FloatVal(v)) => iceberg::spec::Literal::float(*v), + Some(Value::DoubleVal(v)) => iceberg::spec::Literal::double(*v), + Some(Value::DateVal(v)) => iceberg::spec::Literal::date(*v), Some(Value::TimestampVal(v)) => iceberg::spec::Literal::timestamp(*v), Some(Value::TimestampTzVal(v)) => iceberg::spec::Literal::timestamptz(*v), Some(Value::StringVal(s)) => iceberg::spec::Literal::string(s.clone()), - Some(Value::DoubleVal(v)) => iceberg::spec::Literal::double(*v), - Some(Value::FloatVal(v)) => iceberg::spec::Literal::float(*v), - Some(Value::DecimalVal(bytes)) => { - // Deserialize unscaled BigInteger bytes to i128 - // BigInteger is serialized as signed big-endian bytes - if bytes.len() > 16 { - return Err(GeneralError(format!( - "Decimal bytes too large: {} bytes (max 16 for i128)", - bytes.len() - ))); - } - - // Convert big-endian bytes to i128 - let mut buf = [0u8; 16]; - let offset = 16 - bytes.len(); - buf[offset..].copy_from_slice(bytes); - - // Handle sign extension for negative numbers - let value = if !bytes.is_empty() && (bytes[0] & 0x80) != 0 { - // Negative number - sign extend - for byte in buf.iter_mut().take(offset) { - *byte = 0xFF; - } - i128::from_be_bytes(buf) - } else { - // Positive number - i128::from_be_bytes(buf) - }; - - iceberg::spec::Literal::decimal(value) - } - Some(Value::BoolVal(v)) => iceberg::spec::Literal::bool(*v), + Some(Value::DecimalVal(d)) => iceberg::spec::Literal::decimal(decimal_bytes_to_i128( + &d.unscaled, + )?), Some(Value::UuidVal(bytes)) => { - // Deserialize UUID from 16 bytes if bytes.len() != 16 { return Err(GeneralError(format!( "Invalid UUID bytes length: {} (expected 16)", @@ -3604,7 +3577,7 @@ fn partition_value_to_literal( Some(Value::BinaryVal(bytes)) => iceberg::spec::Literal::binary(bytes.to_vec()), None => { return Err(GeneralError( - "PartitionValue has no value set and is_null is false".to_string(), + "IcebergLiteral has no value set and is_null is false".to_string(), )); } }; @@ -3612,6 +3585,26 @@ fn partition_value_to_literal( Ok(Some(literal)) } +/// Decodes an unscaled decimal (two's-complement big-endian) into i128. +fn decimal_bytes_to_i128(bytes: &[u8]) -> Result { + if bytes.len() > 16 { + return Err(GeneralError(format!( + "Decimal bytes too large: {} bytes (max 16 for i128)", + bytes.len() + ))); + } + let mut buf = [0u8; 16]; + let offset = 16 - bytes.len(); + buf[offset..].copy_from_slice(bytes); + // Sign-extend negative values. + if !bytes.is_empty() && (bytes[0] & 0x80) != 0 { + for byte in buf.iter_mut().take(offset) { + *byte = 0xFF; + } + } + Ok(i128::from_be_bytes(buf)) +} + /// Converts a protobuf PartitionData to an iceberg Struct. /// /// Uses the existing Struct::from_iter() API from iceberg-rust to construct the struct @@ -3762,14 +3755,15 @@ fn parse_file_scan_tasks_from_common( proto_common .residual_pool .get(idx as usize) - .and_then(convert_spark_expr_to_predicate) + .and_then(iceberg_predicate_to_predicate) .and_then(|pred| { // The residual predicate only drives row-group pruning; the post-scan // filter still enforces correctness. iceberg-rust cannot bind a datum // whose type has no conversion to the column type, so on a bind failure - // we skip pushdown rather than fail the scan, mirroring the NOT IN - // handling above. - match pred.bind(Arc::clone(&schema_ref), true) { + // we skip pushdown rather than fail the scan. + // rewrite_not first: iceberg-rust's scan evaluators require negation-normal + // form and reject a bare NOT (e.g. from a `NOT (a = b)` residual). + match pred.rewrite_not().bind(Arc::clone(&schema_ref), true) { Ok(bound) => Some(bound), Err(e) => { log::warn!("Skipping Iceberg predicate pushdown; bind failed: {e}"); @@ -4112,232 +4106,102 @@ fn literal_to_array_ref( // always returns MIGHT_MATCH (never prunes row groups). These are handled by CometFilter post-scan. /// Converts a protobuf Spark expression to an Iceberg predicate for row-group filtering. -fn convert_spark_expr_to_predicate( - expr: &spark_expression::Expr, +/// Converts a serialized Iceberg residual predicate into an iceberg-rust `Predicate` for row-group +/// pruning. This is only a pruning hint -- the post-scan CometFilter enforces correctness -- so any +/// node or literal that cannot be represented degrades to `None` (no pushdown) rather than an +/// error. Mirrors the proto 1:1 onto iceberg-rust's `Reference` builders. +fn iceberg_predicate_to_predicate( + proto: &spark_operator::IcebergPredicate, ) -> Option { - use spark_expression::expr::ExprStruct; - - match &expr.expr_struct { - Some(ExprStruct::Eq(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::Eq, - ), - Some(ExprStruct::Neq(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::NotEq, - ), - Some(ExprStruct::Lt(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::LessThan, - ), - Some(ExprStruct::LtEq(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::LessThanOrEq, - ), - Some(ExprStruct::Gt(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::GreaterThan, - ), - Some(ExprStruct::GtEq(binary)) => convert_binary_to_predicate( - &binary.left, - &binary.right, - iceberg::expr::PredicateOperator::GreaterThanOrEq, - ), - Some(ExprStruct::IsNull(unary)) => { - if let Some(ref child) = unary.child { - extract_column_reference(child).map(|column| { - iceberg::expr::Predicate::Unary(iceberg::expr::UnaryExpression::new( - iceberg::expr::PredicateOperator::IsNull, - iceberg::expr::Reference::new(column), - )) - }) - } else { - None + use iceberg::expr::Reference; + use spark_operator::iceberg_predicate::Node; + use spark_operator::IcebergPredicateOperator as Op; + + match proto.node.as_ref()? { + Node::Unary(u) => { + let column = Reference::new(u.column.clone()); + match u.op() { + Op::IsNull => Some(column.is_null()), + Op::NotNull => Some(column.is_not_null()), + Op::IsNan => Some(column.is_nan()), + Op::NotNan => Some(column.is_not_nan()), + _ => None, } } - Some(ExprStruct::IsNotNull(unary)) => { - if let Some(ref child) = unary.child { - extract_column_reference(child).map(|column| { - iceberg::expr::Predicate::Unary(iceberg::expr::UnaryExpression::new( - iceberg::expr::PredicateOperator::NotNull, - iceberg::expr::Reference::new(column), - )) - }) - } else { - None + Node::Binary(b) => { + let column = Reference::new(b.column.clone()); + let datum = iceberg_literal_to_datum(b.value.as_ref()?)?; + match b.op() { + Op::Eq => Some(column.equal_to(datum)), + Op::NotEq => Some(column.not_equal_to(datum)), + Op::LessThan => Some(column.less_than(datum)), + Op::LessThanOrEq => Some(column.less_than_or_equal_to(datum)), + Op::GreaterThan => Some(column.greater_than(datum)), + Op::GreaterThanOrEq => Some(column.greater_than_or_equal_to(datum)), + Op::StartsWith => Some(column.starts_with(datum)), + Op::NotStartsWith => Some(column.not_starts_with(datum)), + _ => None, } } - Some(ExprStruct::And(binary)) => { - let left = binary - .left - .as_ref() - .and_then(|e| convert_spark_expr_to_predicate(e)); - let right = binary - .right - .as_ref() - .and_then(|e| convert_spark_expr_to_predicate(e)); - match (left, right) { - (Some(l), Some(r)) => Some(l.and(r)), - (Some(l), None) => Some(l), - (None, Some(r)) => Some(r), - _ => None, + Node::Set(s) => { + // Only IN prunes from stats; not_in is inherently unprunable, so it is never emitted. + if s.op() != Op::In { + return None; } + let column = Reference::new(s.column.clone()); + let datums = s + .values + .iter() + .map(iceberg_literal_to_datum) + .collect::>>()?; + Some(column.is_in(datums)) } - Some(ExprStruct::Or(binary)) => { - let left = binary - .left - .as_ref() - .and_then(|e| convert_spark_expr_to_predicate(e)); - let right = binary - .right - .as_ref() - .and_then(|e| convert_spark_expr_to_predicate(e)); + Node::And(l) => { + // Dropping a conjunct only weakens the pruning predicate, so a missing side is safe. + let left = l.left.as_deref().and_then(iceberg_predicate_to_predicate); + let right = l.right.as_deref().and_then(iceberg_predicate_to_predicate); match (left, right) { - (Some(l), Some(r)) => Some(l.or(r)), - _ => None, // OR requires both sides to be valid + (Some(l), Some(r)) => Some(l.and(r)), + (Some(p), None) | (None, Some(p)) => Some(p), + (None, None) => None, } } - Some(ExprStruct::Not(unary)) => unary - .child - .as_ref() - .and_then(|child| convert_spark_expr_to_predicate(child)) - .map(|p| !p), - Some(ExprStruct::In(in_expr)) => { - // NOT IN predicates don't work correctly with iceberg-rust's row-group filtering. - // The iceberg-rust RowGroupMetricsEvaluator::not_in() always returns MIGHT_MATCH - // (never prunes row groups), even in cases where pruning is possible (e.g., when - // min == max == value and value is in the NOT IN set). - // - // Workaround: Skip NOT IN in predicate pushdown and let CometFilter handle it - // post-scan. This sacrifices row-group pruning for NOT IN but ensures correctness. - if in_expr.negated { - return None; - } - - if let Some(ref value) = in_expr.in_value { - if let Some(column) = extract_column_reference(value) { - let datums: Vec = in_expr - .lists - .iter() - .filter_map(extract_literal_as_datum) - .collect(); - - if datums.len() == in_expr.lists.len() { - Some(iceberg::expr::Reference::new(column).is_in(datums)) - } else { - None - } - } else { - None - } - } else { - None - } + Node::Or(o) => { + // Dropping a disjunct would strengthen the predicate and wrongly prune; require both. + let left = o.left.as_deref().and_then(iceberg_predicate_to_predicate)?; + let right = o.right.as_deref().and_then(iceberg_predicate_to_predicate)?; + Some(left.or(right)) } - _ => None, // Unsupported expression - } -} - -fn convert_binary_to_predicate( - left: &Option>, - right: &Option>, - op: iceberg::expr::PredicateOperator, -) -> Option { - let left_ref = left.as_ref()?; - let right_ref = right.as_ref()?; - - if let (Some(column), Some(datum)) = ( - extract_column_reference(left_ref), - extract_literal_as_datum(right_ref), - ) { - return Some(iceberg::expr::Predicate::Binary( - iceberg::expr::BinaryExpression::new(op, iceberg::expr::Reference::new(column), datum), - )); - } - - if let (Some(datum), Some(column)) = ( - extract_literal_as_datum(left_ref), - extract_column_reference(right_ref), - ) { - let reversed_op = match op { - iceberg::expr::PredicateOperator::LessThan => { - iceberg::expr::PredicateOperator::GreaterThan - } - iceberg::expr::PredicateOperator::LessThanOrEq => { - iceberg::expr::PredicateOperator::GreaterThanOrEq - } - iceberg::expr::PredicateOperator::GreaterThan => { - iceberg::expr::PredicateOperator::LessThan - } - iceberg::expr::PredicateOperator::GreaterThanOrEq => { - iceberg::expr::PredicateOperator::LessThanOrEq - } - _ => op, // Eq and NotEq are symmetric - }; - return Some(iceberg::expr::Predicate::Binary( - iceberg::expr::BinaryExpression::new( - reversed_op, - iceberg::expr::Reference::new(column), - datum, - ), - )); + Node::Not(child) => iceberg_predicate_to_predicate(child).map(|p| !p), } - - None } -fn extract_column_reference(expr: &spark_expression::Expr) -> Option { - use spark_expression::expr::ExprStruct; +/// Converts a serialized `IcebergLiteral` into an iceberg-rust `Datum` for predicate pushdown. +/// Returns `None` for null and for byte-array-backed types (decimal/uuid/fixed/binary), which +/// iceberg-rust cannot use in the page index yet; the driver does not emit those for predicates, +/// so this only guards against unexpected input. Correctness is unaffected -- a dropped literal +/// just skips pushdown. +fn iceberg_literal_to_datum(lit: &spark_operator::IcebergLiteral) -> Option { + use spark_operator::iceberg_literal::Value; - match &expr.expr_struct { - Some(ExprStruct::Unbound(unbound_ref)) => Some(unbound_ref.name.clone()), - _ => None, + if lit.is_null { + return None; } -} - -fn extract_literal_as_datum(expr: &spark_expression::Expr) -> Option { - use spark_expression::expr::ExprStruct; - match &expr.expr_struct { - Some(ExprStruct::Literal(literal)) => { - if literal.is_null { - return None; - } - - match &literal.value { - Some(spark_expression::literal::Value::IntVal(v)) => { - Some(iceberg::spec::Datum::int(*v)) - } - Some(spark_expression::literal::Value::LongVal(v)) => { - Some(iceberg::spec::Datum::long(*v)) - } - Some(spark_expression::literal::Value::FloatVal(v)) => { - Some(iceberg::spec::Datum::double(*v as f64)) - } - Some(spark_expression::literal::Value::DoubleVal(v)) => { - Some(iceberg::spec::Datum::double(*v)) - } - Some(spark_expression::literal::Value::StringVal(v)) => { - Some(iceberg::spec::Datum::string(v.clone())) - } - Some(spark_expression::literal::Value::BoolVal(v)) => { - Some(iceberg::spec::Datum::bool(*v)) - } - Some(spark_expression::literal::Value::ByteVal(v)) => { - Some(iceberg::spec::Datum::int(*v)) - } - Some(spark_expression::literal::Value::ShortVal(v)) => { - Some(iceberg::spec::Datum::int(*v)) - } - _ => None, - } - } - _ => None, + match lit.value.as_ref()? { + Value::BoolVal(v) => Some(iceberg::spec::Datum::bool(*v)), + Value::IntVal(v) => Some(iceberg::spec::Datum::int(*v)), + Value::LongVal(v) => Some(iceberg::spec::Datum::long(*v)), + Value::FloatVal(v) => Some(iceberg::spec::Datum::float(*v)), + Value::DoubleVal(v) => Some(iceberg::spec::Datum::double(*v)), + Value::DateVal(v) => Some(iceberg::spec::Datum::date(*v)), + Value::TimestampVal(v) => Some(iceberg::spec::Datum::timestamp_micros(*v)), + Value::TimestampTzVal(v) => Some(iceberg::spec::Datum::timestamptz_micros(*v)), + Value::StringVal(v) => Some(iceberg::spec::Datum::string(v.clone())), + Value::DecimalVal(_) + | Value::UuidVal(_) + | Value::FixedVal(_) + | Value::BinaryVal(_) => None, } } diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 03e6eaecb8..826eed9ce6 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -161,25 +161,39 @@ message CsvOptions { bool truncated_rows = 8; } +// A typed Iceberg primitive value, shared by partition values and predicate literals. The oneof +// case is the type tag (iceberg-rust's Datum is self-typed and Predicate::bind does not coerce, so +// the type must be known at construction). Grows by adding a case as Iceberg adds primitives +// (GEOMETRY/VARIANT/nanosecond timestamps); unrecognized values decode to no-op, never an error. +message IcebergLiteral { + bool is_null = 1; + oneof value { + bool bool_val = 2; + int32 int_val = 3; + int64 long_val = 4; + float float_val = 5; + double double_val = 6; + int32 date_val = 7; // days since epoch + int64 timestamp_val = 8; // micros since epoch, no tz + int64 timestamp_tz_val = 9; // micros since epoch, utc + string string_val = 10; + bytes uuid_val = 11; + bytes fixed_val = 12; + bytes binary_val = 13; + IcebergDecimal decimal_val = 14; + } +} + +message IcebergDecimal { + bytes unscaled = 1; // unscaled value, two's-complement big-endian + int32 precision = 2; + int32 scale = 3; +} + // Partition value for Iceberg partition data message PartitionValue { int32 field_id = 1; - oneof value { - int32 int_val = 2; - int64 long_val = 3; - int64 date_val = 4; // days since epoch - int64 timestamp_val = 5; // microseconds since epoch - int64 timestamp_tz_val = 6; // microseconds with timezone - string string_val = 7; - double double_val = 8; - float float_val = 9; - bytes decimal_val = 10; // unscaled BigInteger bytes - bool bool_val = 11; - bytes uuid_val = 12; - bytes fixed_val = 13; - bytes binary_val = 14; - } - bool is_null = 15; + IcebergLiteral literal = 2; } // Collection of partition values for a single partition @@ -187,6 +201,48 @@ message PartitionData { repeated PartitionValue values = 1; } +// Iceberg residual predicate for row-group pruning, mirroring iceberg::expr::Predicate. Only a +// pruning hint: the post-scan CometFilter enforces correctness, so an unrepresentable predicate is +// simply omitted. Carries no expr_id/query_context (those are for native ANSI error reporting on +// executable expressions; a residual never errors), which also lets identical residuals dedup. +message IcebergPredicate { + oneof node { + IcebergUnaryPredicate unary = 1; // IS_NULL / NOT_NULL / IS_NAN / NOT_NAN + IcebergBinaryPredicate binary = 2; // column OP literal + IcebergSetPredicate set = 3; // IN only emitted; not_in is inherently unprunable from stats + IcebergLogicalPredicate and = 4; + IcebergLogicalPredicate or = 5; + IcebergPredicate not = 6; + } +} + +message IcebergUnaryPredicate { string column = 1; IcebergPredicateOperator op = 2; } +message IcebergBinaryPredicate { string column = 1; IcebergPredicateOperator op = 2; IcebergLiteral value = 3; } +message IcebergSetPredicate { string column = 1; IcebergPredicateOperator op = 2; repeated IcebergLiteral values = 3; } +message IcebergLogicalPredicate { IcebergPredicate left = 1; IcebergPredicate right = 2; } + +// Mirrors iceberg-rust PredicateOperator (crates/iceberg/src/expr/mod.rs) so the mapping on both +// sides is 1:1. The driver emits a subset by design: NOT_IN never prunes from column stats +// (inherent, matches Iceberg-Java), and IS_NAN/STARTS_WITH are not produced by the current residual +// path. Unspecified/unrecognized decodes to no-pushdown. +enum IcebergPredicateOperator { + Unspecified = 0; + IsNull = 1; + NotNull = 2; + IsNan = 3; + NotNan = 4; + LessThan = 5; + LessThanOrEq = 6; + GreaterThan = 7; + GreaterThanOrEq = 8; + Eq = 9; + NotEq = 10; + StartsWith = 11; + NotStartsWith = 12; + In = 13; + NotIn = 14; +} + // Common data shared by all partitions in split mode (sent once, captured in closure) message IcebergScanCommon { // Catalog-specific configuration for FileIO (credentials, S3/GCS config, etc.) @@ -206,7 +262,7 @@ message IcebergScanCommon { repeated ProjectFieldIdList project_field_ids_pool = 8; repeated PartitionData partition_data_pool = 9; repeated DeleteFileList delete_files_pool = 10; - repeated spark.spark_expression.Expr residual_pool = 11; + repeated IcebergPredicate residual_pool = 11; // Number of data files to read concurrently within a single task uint32 data_file_concurrency_limit = 12; diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index f09e35d4fa..8272cf2222 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -20,6 +20,7 @@ package org.apache.comet.serde.operator import java.nio.charset.StandardCharsets.UTF_8 +import java.util.Base64 import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -34,12 +35,13 @@ import org.apache.spark.sql.comet.shims.ShimDataSourceRDDPartition import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceRDD, DataSourceRDDPartition} import org.apache.spark.sql.types._ +import com.google.protobuf.ByteString + import org.apache.comet.{CometConf, ConfigEntry} import org.apache.comet.iceberg.{CometIcebergNativeScanMetadata, IcebergReflection} import org.apache.comet.serde.{CometOperatorSerde, OperatorOuterClass} -import org.apache.comet.serde.ExprOuterClass.Expr import org.apache.comet.serde.OperatorOuterClass.{Operator, SparkStructField} -import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType} +import org.apache.comet.serde.QueryPlanSerde.serializeDataType object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] with Logging { @@ -74,21 +76,33 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } /** - * Converts an Iceberg partition value to protobuf format. Protobuf is less verbose than JSON. - * The following types are also serialized as integer values instead of as strings - Timestamps, - * Dates, Decimals, FieldIDs + * Wraps an Iceberg partition value (a typed primitive) in a PartitionValue. The value encoding + * is shared with predicate literals via [[icebergLiteralToProto]]. */ private def partitionValueToProto( fieldId: Int, fieldTypeStr: String, - value: Any): OperatorOuterClass.PartitionValue = { - val builder = OperatorOuterClass.PartitionValue.newBuilder() - builder.setFieldId(fieldId) + value: Any): OperatorOuterClass.PartitionValue = + OperatorOuterClass.PartitionValue + .newBuilder() + .setFieldId(fieldId) + .setLiteral(icebergLiteralToProto(fieldTypeStr, value)) + .build() + + /** + * Converts an Iceberg primitive value to a typed IcebergLiteral, keyed on the Iceberg type + * name. Protobuf is less verbose than JSON; timestamps/dates/decimals are serialized as their + * integer or byte encodings rather than strings. Shared by partition values and predicate + * literals. + */ + private def icebergLiteralToProto( + fieldTypeStr: String, + value: Any): OperatorOuterClass.IcebergLiteral = { + val builder = OperatorOuterClass.IcebergLiteral.newBuilder() if (value == null) { builder.setIsNull(true) } else { - builder.setIsNull(false) fieldTypeStr match { case t if t.startsWith("timestamp") => val micros = value match { @@ -107,13 +121,17 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit builder.setDateVal(days) case d if d.startsWith("decimal(") => - // Serialize as unscaled BigInteger bytes val bigDecimal = value match { case bd: java.math.BigDecimal => bd case _ => new java.math.BigDecimal(value.toString) } - val unscaledBytes = bigDecimal.unscaledValue().toByteArray - builder.setDecimalVal(com.google.protobuf.ByteString.copyFrom(unscaledBytes)) + builder.setDecimalVal( + OperatorOuterClass.IcebergDecimal + .newBuilder() + .setUnscaled(ByteString.copyFrom(bigDecimal.unscaledValue().toByteArray)) + .setPrecision(bigDecimal.precision()) + .setScale(bigDecimal.scale()) + .build()) case "string" => builder.setStringVal(value.toString) @@ -173,7 +191,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit bb.putLong(uuid.getLeastSignificantBits) bb.array() } - builder.setUuidVal(com.google.protobuf.ByteString.copyFrom(uuidBytes)) + builder.setUuidVal(ByteString.copyFrom(uuidBytes)) case t if t.startsWith("fixed[") || t.startsWith("binary") => val bytes = value match { @@ -181,9 +199,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit case _ => value.toString.getBytes("UTF-8") } if (t.startsWith("fixed")) { - builder.setFixedVal(com.google.protobuf.ByteString.copyFrom(bytes)) + builder.setFixedVal(ByteString.copyFrom(bytes)) } else { - builder.setBinaryVal(com.google.protobuf.ByteString.copyFrom(bytes)) + builder.setBinaryVal(ByteString.copyFrom(bytes)) } // Fallback: infer type from Java type ? @@ -203,24 +221,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit builder.build() } - /** - * Helper to extract a literal from an Iceberg expression and build a binary predicate. - */ - private def buildBinaryPredicate( - exprClass: Class[_], - icebergExpr: Any, - attribute: Attribute, - builder: (Expression, Expression) => Expression): Option[Expression] = { - try { - val literalMethod = exprClass.getMethod("literal") - val literal = literalMethod.invoke(icebergExpr) - val value = convertIcebergLiteral(literal, attribute.dataType) - Some(builder(attribute, value)) - } catch { - case _: Exception => None - } - } - /** * Extracts delete files from an Iceberg FileScanTask as a list (for deduplication). * @@ -464,7 +464,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // Deduplicate by protobuf bytes (use Base64 string as key) val partitionDataBytes = partitionDataProto.toByteArray - val partitionDataKey = java.util.Base64.getEncoder.encodeToString(partitionDataBytes) + val partitionDataKey = Base64.getEncoder.encodeToString(partitionDataBytes) val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate( partitionDataKey, { @@ -534,111 +534,104 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } /** - * Converts Iceberg Expression objects to Spark Catalyst expressions. - * - * This is used to extract per-file residual expressions from Iceberg FileScanTasks. Residuals - * are created by Iceberg's ResidualEvaluator through partial evaluation of scan filters against - * each file's partition data. These residuals enable row-group level filtering in the Parquet - * reader. + * Converts an Iceberg residual Expression into an IcebergPredicate for native row-group + * pruning. * - * The conversion uses reflection because Iceberg expressions are not directly accessible from - * Spark's classpath during query planning. + * Residuals come from Iceberg's ResidualEvaluator (partial evaluation of the scan filter + * against each file's partition data). This is only a pruning hint: the CometFilter above the + * scan enforces correctness, so any node or literal we cannot represent yields None (no + * pushdown). Uses reflection because Iceberg's expression classes are not on Spark's classpath + * at planning time; residuals are unbound predicates carrying a NamedReference (column name) + * and a literal. */ - def convertIcebergExpression(icebergExpr: Any, output: Seq[Attribute]): Option[Expression] = { + def icebergExprToProto( + icebergExpr: Any, + output: Seq[Attribute]): Option[OperatorOuterClass.IcebergPredicate] = { try { val exprClass = icebergExpr.getClass val attributeMap = output.map(attr => attr.name -> attr).toMap - // Check for UnboundPredicate if (exprClass.getName.endsWith(Constants.ExpressionTypes.UNBOUND_PREDICATE)) { - val opMethod = exprClass.getMethod("op") - val termMethod = exprClass.getMethod("term") - val operation = opMethod.invoke(icebergExpr) - val term = termMethod.invoke(icebergExpr) - - // Get column name from term - val refMethod = term.getClass.getMethod("ref") - val ref = refMethod.invoke(term) - val nameMethod = ref.getClass.getMethod("name") - val columnName = nameMethod.invoke(ref).asInstanceOf[String] - - val attr = attributeMap.get(columnName) - - val opName = operation.toString - - attr.flatMap { attribute => - opName match { - case Constants.Operations.IS_NULL => - Some(IsNull(attribute)) - - case Constants.Operations.IS_NOT_NULL | Constants.Operations.NOT_NULL => - Some(IsNotNull(attribute)) - - case Constants.Operations.EQ => - buildBinaryPredicate(exprClass, icebergExpr, attribute, EqualTo) - - case Constants.Operations.NOT_EQ => - buildBinaryPredicate( + val operation = exprClass.getMethod("op").invoke(icebergExpr).toString + val term = exprClass.getMethod("term").invoke(icebergExpr) + val ref = term.getClass.getMethod("ref").invoke(term) + val columnName = ref.getClass.getMethod("name").invoke(ref).asInstanceOf[String] + + attributeMap.get(columnName).flatMap { attribute => + import Constants.Operations._ + import OperatorOuterClass.IcebergPredicateOperator + operation match { + case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull)) + case IS_NOT_NULL | NOT_NULL => + Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull)) + case EQ => + binaryPredicate( exprClass, icebergExpr, + columnName, attribute, - (a, v) => Not(EqualTo(a, v))) - - case Constants.Operations.LT => - buildBinaryPredicate(exprClass, icebergExpr, attribute, LessThan) - - case Constants.Operations.LT_EQ => - buildBinaryPredicate(exprClass, icebergExpr, attribute, LessThanOrEqual) - - case Constants.Operations.GT => - buildBinaryPredicate(exprClass, icebergExpr, attribute, GreaterThan) - - case Constants.Operations.GT_EQ => - buildBinaryPredicate(exprClass, icebergExpr, attribute, GreaterThanOrEqual) - - case Constants.Operations.IN => - val literalsMethod = exprClass.getMethod("literals") - val literals = literalsMethod.invoke(icebergExpr).asInstanceOf[java.util.List[_]] - val values = - literals.asScala.map(lit => convertIcebergLiteral(lit, attribute.dataType)) - Some(In(attribute, values.toSeq)) - - case Constants.Operations.NOT_IN => - val literalsMethod = exprClass.getMethod("literals") - val literals = literalsMethod.invoke(icebergExpr).asInstanceOf[java.util.List[_]] - val values = - literals.asScala.map(lit => convertIcebergLiteral(lit, attribute.dataType)) - Some(Not(In(attribute, values.toSeq))) - - case _ => - None + IcebergPredicateOperator.Eq) + case NOT_EQ => + binaryPredicate( + exprClass, + icebergExpr, + columnName, + attribute, + IcebergPredicateOperator.NotEq) + case LT => + binaryPredicate( + exprClass, + icebergExpr, + columnName, + attribute, + IcebergPredicateOperator.LessThan) + case LT_EQ => + binaryPredicate( + exprClass, + icebergExpr, + columnName, + attribute, + IcebergPredicateOperator.LessThanOrEq) + case GT => + binaryPredicate( + exprClass, + icebergExpr, + columnName, + attribute, + IcebergPredicateOperator.GreaterThan) + case GT_EQ => + binaryPredicate( + exprClass, + icebergExpr, + columnName, + attribute, + IcebergPredicateOperator.GreaterThanOrEq) + case IN => setPredicate(exprClass, icebergExpr, columnName, attribute) + // NOT_IN is inherently unprunable from column stats, so it is not pushed. + case _ => None } } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.AND)) { - val leftMethod = exprClass.getMethod("left") - val rightMethod = exprClass.getMethod("right") - val left = leftMethod.invoke(icebergExpr) - val right = rightMethod.invoke(icebergExpr) - - (convertIcebergExpression(left, output), convertIcebergExpression(right, output)) match { - case (Some(l), Some(r)) => Some(And(l, r)) - case _ => None + val left = icebergExprToProto(exprClass.getMethod("left").invoke(icebergExpr), output) + val right = icebergExprToProto(exprClass.getMethod("right").invoke(icebergExpr), output) + (left, right) match { + // Dropping a conjunct only weakens the pruning predicate, so keep any convertible side. + case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = true, l, r)) + case (Some(p), None) => Some(p) + case (None, Some(p)) => Some(p) + case (None, None) => None } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.OR)) { - val leftMethod = exprClass.getMethod("left") - val rightMethod = exprClass.getMethod("right") - val left = leftMethod.invoke(icebergExpr) - val right = rightMethod.invoke(icebergExpr) - - (convertIcebergExpression(left, output), convertIcebergExpression(right, output)) match { - case (Some(l), Some(r)) => Some(Or(l, r)) + val left = icebergExprToProto(exprClass.getMethod("left").invoke(icebergExpr), output) + val right = icebergExprToProto(exprClass.getMethod("right").invoke(icebergExpr), output) + // Dropping a disjunct would strengthen the predicate and wrongly prune, so require both. + (left, right) match { + case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = false, l, r)) case _ => None } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.NOT)) { - val childMethod = exprClass.getMethod("child") - val child = childMethod.invoke(icebergExpr) - - convertIcebergExpression(child, output).map(Not) + val child = exprClass.getMethod("child").invoke(icebergExpr) + icebergExprToProto(child, output).map(notPredicate) } else { None } @@ -648,23 +641,113 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } + private def unaryPredicate( + column: String, + op: OperatorOuterClass.IcebergPredicateOperator): OperatorOuterClass.IcebergPredicate = + OperatorOuterClass.IcebergPredicate + .newBuilder() + .setUnary(OperatorOuterClass.IcebergUnaryPredicate.newBuilder().setColumn(column).setOp(op)) + .build() + + private def binaryPredicate( + exprClass: Class[_], + icebergExpr: Any, + column: String, + attribute: Attribute, + op: OperatorOuterClass.IcebergPredicateOperator) + : Option[OperatorOuterClass.IcebergPredicate] = { + val literal = exprClass.getMethod("literal").invoke(icebergExpr) + predicateLiteralToProto(attribute.dataType, icebergLiteralValue(literal)).map { lit => + OperatorOuterClass.IcebergPredicate + .newBuilder() + .setBinary( + OperatorOuterClass.IcebergBinaryPredicate + .newBuilder() + .setColumn(column) + .setOp(op) + .setValue(lit)) + .build() + } + } + + private def setPredicate( + exprClass: Class[_], + icebergExpr: Any, + column: String, + attribute: Attribute): Option[OperatorOuterClass.IcebergPredicate] = { + val literals = + exprClass.getMethod("literals").invoke(icebergExpr).asInstanceOf[java.util.List[_]] + val protoLiterals = + literals.asScala.map(l => + predicateLiteralToProto(attribute.dataType, icebergLiteralValue(l))) + if (protoLiterals.isEmpty || protoLiterals.exists(_.isEmpty)) { + None + } else { + val setBuilder = OperatorOuterClass.IcebergSetPredicate + .newBuilder() + .setColumn(column) + .setOp(OperatorOuterClass.IcebergPredicateOperator.In) + // Sort literals so an IN list Iceberg happens to iterate in a different order per file still + // deduplicates in the residual pool. + protoLiterals.flatten + .sortBy(lit => Base64.getEncoder.encodeToString(lit.toByteArray)) + .foreach(setBuilder.addValues) + Some(OperatorOuterClass.IcebergPredicate.newBuilder().setSet(setBuilder).build()) + } + } + + private def logicalPredicate( + isAnd: Boolean, + left: OperatorOuterClass.IcebergPredicate, + right: OperatorOuterClass.IcebergPredicate): OperatorOuterClass.IcebergPredicate = { + val logical = OperatorOuterClass.IcebergLogicalPredicate + .newBuilder() + .setLeft(left) + .setRight(right) + val builder = OperatorOuterClass.IcebergPredicate.newBuilder() + if (isAnd) builder.setAnd(logical) else builder.setOr(logical) + builder.build() + } + + private def notPredicate( + child: OperatorOuterClass.IcebergPredicate): OperatorOuterClass.IcebergPredicate = + OperatorOuterClass.IcebergPredicate.newBuilder().setNot(child).build() + + /** Extracts the raw Java value from an Iceberg Literal via reflection. */ + private def icebergLiteralValue(icebergLiteral: Any): Any = { + val literalClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.LITERAL) + literalClass.getMethod("value").invoke(icebergLiteral) + } + /** - * Converts an Iceberg Literal to a Spark Literal + * Builds a predicate literal from its Iceberg value and the column's Spark type. Returns None + * for null (IS NULL is a separate op) and for types not pushed for pruning (byte-array-backed + * types, which iceberg-rust cannot use in the page index yet, and anything unmapped), so the + * predicate degrades to no pushdown rather than an incorrect one. */ - private def convertIcebergLiteral(icebergLiteral: Any, sparkType: DataType): Literal = { - // Load Literal interface to get value() method (use interface to avoid package-private issues) - val literalClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.LITERAL) - val valueMethod = literalClass.getMethod("value") - val value = valueMethod.invoke(icebergLiteral) - - // Convert Java types to Spark internal types - val sparkValue = (value, sparkType) match { - case (s: String, _: StringType) => - org.apache.spark.unsafe.types.UTF8String.fromString(s) - case (v, _) => v + private def predicateLiteralToProto( + sparkType: DataType, + value: Any): Option[OperatorOuterClass.IcebergLiteral] = { + if (value == null) { + return None } - - Literal(sparkValue, sparkType) + val builder = OperatorOuterClass.IcebergLiteral.newBuilder() + sparkType match { + case _: BooleanType => builder.setBoolVal(value.asInstanceOf[java.lang.Boolean]) + case _: IntegerType => builder.setIntVal(value.asInstanceOf[java.lang.Number].intValue()) + case _: LongType => builder.setLongVal(value.asInstanceOf[java.lang.Number].longValue()) + case _: FloatType => builder.setFloatVal(value.asInstanceOf[java.lang.Number].floatValue()) + case _: DoubleType => + builder.setDoubleVal(value.asInstanceOf[java.lang.Number].doubleValue()) + case _: DateType => builder.setDateVal(value.asInstanceOf[java.lang.Number].intValue()) + case _: StringType => builder.setStringVal(value.toString) + case _: TimestampType => + builder.setTimestampTzVal(value.asInstanceOf[java.lang.Number].longValue()) + case _: TimestampNTZType => + builder.setTimestampVal(value.asInstanceOf[java.lang.Number].longValue()) + case _ => return None + } + Some(builder.build()) } /** @@ -749,7 +832,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val deleteFileToPoolIndex = mutable.HashMap[String, Int]() val deleteFilesToPoolIndex = mutable.HashMap[Seq[Int], Int]() - val residualToPoolIndex = mutable.HashMap[Option[Expr], Int]() + val residualToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergPredicate, Int]() val perPartitionBuilders = mutable.ArrayBuffer[OperatorOuterClass.IcebergScan]() @@ -950,11 +1033,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val residualExprOpt = try { - val residualExpr = residualMethod.invoke(task) - val catalystExpr = convertIcebergExpression(residualExpr, output) - catalystExpr.flatMap { expr => - exprToProto(expr, output, binding = false) - } + icebergExprToProto(residualMethod.invoke(task), output) } catch { case e: Exception => logWarning( @@ -963,11 +1042,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit None } - residualExprOpt.foreach { residualExpr => + residualExprOpt.foreach { residual => val residualIdx = residualToPoolIndex.getOrElseUpdate( - Some(residualExpr), { + residual, { val idx = residualToPoolIndex.size - commonBuilder.addResidualPool(residualExpr) + commonBuilder.addResidualPool(residual) idx }) taskBuilder.setResidualIdx(residualIdx) diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala index c6ac428938..245b9d2b3f 100644 --- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala @@ -21,7 +21,9 @@ package org.apache.comet import scala.util.Random +import org.apache.spark.sql.Column import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec +import org.apache.spark.sql.functions.{col, lit} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType import org.apache.spark.sql.types._ @@ -118,32 +120,57 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase { } } - test("decode") { + test("filter pushdown - fuzz predicates over primitive columns") { val df = spark.table(icebergTableName) - // We want to make sure that the schema generator wasn't modified to accidentally omit - // BinaryType, since then this test would not run any queries and silently pass. - var testedBinary = false - for (field <- df.schema.fields if field.dataType == BinaryType) { - testedBinary = true - // Intentionally use odd capitalization of 'utf-8' to test normalization. - val sql = s"SELECT decode(${field.name}, 'utF-8') FROM $icebergTableName" - checkSparkAnswerAndOperator(sql) - } - assert(testedBinary) - } + val primitiveColumns = + df.schema.fields.filterNot(f => isComplexType(f.dataType)).map(_.name) + assert(primitiveColumns.nonEmpty, "expected at least one primitive column in the fuzz schema") + + for (name <- primitiveColumns) { + // Sample real values of the column's own type so predicates are always well-typed, avoiding + // per-type literal generation and SQL formatting. Skip NaN, whose comparison differs. + val values = df + .select(col(name)) + .where(col(name).isNotNull) + .distinct() + .limit(3) + .collect() + .map(_.get(0)) + .filterNot { + case d: Double => d.isNaN + case f: Float => f.isNaN + case _ => false + } + .toSeq + + val c = col(name) + val predicates = scala.collection.mutable.ArrayBuffer[Column](c.isNull, c.isNotNull) + values.headOption.foreach { v => + val l = lit(v) + // Comparisons, not-equal and explicit NOT (exercises the native rewrite_not path), and + // AND/OR combinations. + predicates ++= Seq( + c === l, + c =!= l, + c > l, + c >= l, + c < l, + c <= l, + !(c === l), + c.isNotNull && (c === l), + (c === l) || c.isNull) + } + if (values.nonEmpty) { + predicates += c.isin(values: _*) + predicates += !c.isin(values: _*) + } - test("regexp_replace") { - withSQLConf(CometConf.getExprAllowIncompatConfigKey("RegExpReplace") -> "true") { - val df = spark.table(icebergTableName) - // We want to make sure that the schema generator wasn't modified to accidentally omit - // StringType, since then this test would not run any queries and silently pass. - var testedString = false - for (field <- df.schema.fields if field.dataType == StringType) { - testedString = true - val sql = s"SELECT regexp_replace(${field.name}, 'a', 'b') FROM $icebergTableName" - checkSparkAnswerAndOperator(sql) + for (pred <- predicates) { + val (_, cometPlan) = checkSparkAnswer(df.where(pred)) + assert( + collectIcebergNativeScans(cometPlan).nonEmpty, + s"expected a native Iceberg scan for predicate on '$name': $pred") } - assert(testedString) } } diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index c6746954f7..e6f1fc8f2f 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -667,11 +667,6 @@ class CometIcebergNativeSuite .map(p => p -> countByteOccurrences(commonBytes, p.getBytes(UTF_8))) .filter(_._2 > 1) - logInfo( - s"IcebergScanCommon = ${commonBytes.length} bytes, " + - s"${distinctPaths.size} distinct delete files, " + - s"$totalReferences delete-file references") - assert( duplicated.isEmpty, "delete files serialized more than once in IcebergScanCommon " + @@ -3088,6 +3083,71 @@ class CometIcebergNativeSuite } } + // A large static filter on a non-partition column gives every file the identical residual, so + // the residual pool must collapse to a single entry. Residuals serialize as IcebergPredicate + // without expr_id/query_context, so structurally-equal residuals dedup; a regression that + // reintroduced per-node identity metadata would make the pool grow with the task count and could + // overflow the broadcast IcebergScanCommon. + test("residual pool dedups a shared non-partition filter to one entry") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.residual_dedup_test (id INT, region STRING, payload STRING) + USING iceberg + PARTITIONED BY (region) + TBLPROPERTIES ('format-version' = '2') + """) + + // One data file per region, so the scan yields many FileScanTasks. The filter below is on + // payload, a non-partition column, so each task's residual is the identical full IN. + val values = (0 until 40) + .flatMap(r => (0 until 25).map(i => s"($i, 'r$r', 'p$i')")) + .mkString(", ") + spark.sql(s"INSERT INTO test_cat.db.residual_dedup_test VALUES $values") + + val inList = (0 until 200).map(i => s"'p$i'").mkString(", ") + val (_, cometPlan) = checkSparkAnswer( + "SELECT * FROM test_cat.db.residual_dedup_test " + + s"WHERE payload IN ($inList) ORDER BY id, region") + val scans = collectIcebergNativeScans(cometPlan) + assert(scans.length == 1, s"expected one native scan, got ${scans.length}\n$cometPlan") + + val common = OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData) + val residuals = common.getResidualPoolList.asScala.toSeq + + // Count how many FileScanTasks reference a residual across all partitions. Every task here + // carries the same IN predicate, so a correct pool holds one entry shared by all of them. + val tasksWithResidual = scans.head.perPartitionData.map { bytes => + OperatorOuterClass.IcebergScan + .parseFrom(bytes) + .getFileScanTasksList + .asScala + .count(_.hasResidualIdx) + }.sum + + // Guard against a vacuous pass: the dedup is only meaningful if many tasks share the entry. + assert( + tasksWithResidual > 1, + s"expected multiple tasks to carry a residual, got $tasksWithResidual") + assert( + residuals.size == 1, + s"expected the shared non-partition residual to dedup to 1 pool entry across " + + s"$tasksWithResidual task references, got ${residuals.size}") + + spark.sql("DROP TABLE test_cat.db.residual_dedup_test") + } + } + } + // Regression test for https://github.com/apache/datafusion-comet/issues/3860 // Fixed in https://github.com/apache/iceberg-rust/pull/2307 test("filter with nested types in migrated table") { From a62e08fbdb35e55d00c13697d9d2e0f9dd453f13 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 18:40:26 -0400 Subject: [PATCH 05/12] remove int96 tests that don't do anything --- .../apache/comet/CometFuzzIcebergSuite.scala | 78 ------------------- 1 file changed, 78 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala index 245b9d2b3f..22776e9c83 100644 --- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala @@ -174,84 +174,6 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase { } } - test("Iceberg temporal types written as INT96") { - testIcebergTemporalTypes(ParquetOutputTimestampType.INT96) - } - - test("Iceberg temporal types written as TIMESTAMP_MICROS") { - testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MICROS) - } - - test("Iceberg temporal types written as TIMESTAMP_MILLIS") { - testIcebergTemporalTypes(ParquetOutputTimestampType.TIMESTAMP_MILLIS) - } - - private def testIcebergTemporalTypes( - outputTimestampType: ParquetOutputTimestampType.Value, - generateArray: Boolean = true, - generateStruct: Boolean = true): Unit = { - - val schema = FuzzDataGenerator.generateSchema( - SchemaGenOptions( - generateArray = generateArray, - generateStruct = generateStruct, - primitiveTypes = SchemaGenOptions.defaultPrimitiveTypes.filterNot { dataType => - // Disable decimals - iceberg-rust doesn't support FIXED_LEN_BYTE_ARRAY in page index yet - dataType.isInstanceOf[DecimalType] - })) - - val options = - DataGenOptions(generateNegativeZero = false) - - withTempPath { filename => - val random = new Random(42) - withSQLConf( - CometConf.COMET_ENABLED.key -> "false", - SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key -> outputTimestampType.toString, - SQLConf.SESSION_LOCAL_TIMEZONE.key -> defaultTimezone) { - ParquetGenerator.makeParquetFile(random, spark, filename.toString, schema, 100, options) - } - - Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz => - Seq(true, false).foreach { inferTimestampNtzEnabled => - Seq(true, false).foreach { int96TimestampConversion => - Seq(true, false).foreach { int96AsTimestamp => - withSQLConf( - CometConf.COMET_ENABLED.key -> "true", - SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz, - SQLConf.PARQUET_INT96_AS_TIMESTAMP.key -> int96AsTimestamp.toString, - SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION.key -> int96TimestampConversion.toString, - SQLConf.PARQUET_INFER_TIMESTAMP_NTZ_ENABLED.key -> inferTimestampNtzEnabled.toString) { - - val df = spark.table(icebergTableName) - - Seq(defaultTimezone, "UTC", "America/Denver").foreach { tz => - withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> tz) { - def hasTemporalType(t: DataType): Boolean = t match { - case DataTypes.DateType | DataTypes.TimestampType | - DataTypes.TimestampNTZType => - true - case t: StructType => t.exists(f => hasTemporalType(f.dataType)) - case t: ArrayType => hasTemporalType(t.elementType) - case _ => false - } - - val columns = - df.schema.fields.filter(f => hasTemporalType(f.dataType)).map(_.name) - - for (col <- columns) { - checkSparkAnswer(s"SELECT $col FROM $icebergTableName ORDER BY $col") - } - } - } - } - } - } - } - } - } - } - def collectCometShuffleExchanges(plan: org.apache.spark.sql.execution.SparkPlan) : Seq[org.apache.spark.sql.execution.SparkPlan] = { collect(plan) { From f2c2be60ef1ff899091233bf1ee38bbf862829a4 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 18:50:06 -0400 Subject: [PATCH 06/12] make format --- native/core/src/execution/planner.rs | 16 ++++++++-------- .../org/apache/comet/CometFuzzIcebergSuite.scala | 4 ---- .../apache/comet/CometIcebergNativeSuite.scala | 2 +- 3 files changed, 9 insertions(+), 13 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 389a92b7ec..451b8a8b8f 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3559,9 +3559,9 @@ fn partition_value_to_literal( Some(Value::TimestampVal(v)) => iceberg::spec::Literal::timestamp(*v), Some(Value::TimestampTzVal(v)) => iceberg::spec::Literal::timestamptz(*v), Some(Value::StringVal(s)) => iceberg::spec::Literal::string(s.clone()), - Some(Value::DecimalVal(d)) => iceberg::spec::Literal::decimal(decimal_bytes_to_i128( - &d.unscaled, - )?), + Some(Value::DecimalVal(d)) => { + iceberg::spec::Literal::decimal(decimal_bytes_to_i128(&d.unscaled)?) + } Some(Value::UuidVal(bytes)) => { if bytes.len() != 16 { return Err(GeneralError(format!( @@ -4169,7 +4169,10 @@ fn iceberg_predicate_to_predicate( Node::Or(o) => { // Dropping a disjunct would strengthen the predicate and wrongly prune; require both. let left = o.left.as_deref().and_then(iceberg_predicate_to_predicate)?; - let right = o.right.as_deref().and_then(iceberg_predicate_to_predicate)?; + let right = o + .right + .as_deref() + .and_then(iceberg_predicate_to_predicate)?; Some(left.or(right)) } Node::Not(child) => iceberg_predicate_to_predicate(child).map(|p| !p), @@ -4198,10 +4201,7 @@ fn iceberg_literal_to_datum(lit: &spark_operator::IcebergLiteral) -> Option Some(iceberg::spec::Datum::timestamp_micros(*v)), Value::TimestampTzVal(v) => Some(iceberg::spec::Datum::timestamptz_micros(*v)), Value::StringVal(v) => Some(iceberg::spec::Datum::string(v.clone())), - Value::DecimalVal(_) - | Value::UuidVal(_) - | Value::FixedVal(_) - | Value::BinaryVal(_) => None, + Value::DecimalVal(_) | Value::UuidVal(_) | Value::FixedVal(_) | Value::BinaryVal(_) => None, } } diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala index 22776e9c83..5e8cca0c3a 100644 --- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala @@ -24,12 +24,8 @@ import scala.util.Random import org.apache.spark.sql.Column import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec import org.apache.spark.sql.functions.{col, lit} -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType -import org.apache.spark.sql.types._ import org.apache.comet.DataTypeSupport.isComplexType -import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator, ParquetGenerator, SchemaGenOptions} class CometFuzzIcebergSuite extends CometFuzzIcebergBase { diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index e6f1fc8f2f..5cec19a81d 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -3140,7 +3140,7 @@ class CometIcebergNativeSuite s"expected multiple tasks to carry a residual, got $tasksWithResidual") assert( residuals.size == 1, - s"expected the shared non-partition residual to dedup to 1 pool entry across " + + "expected the shared non-partition residual to dedup to 1 pool entry across " + s"$tasksWithResidual task references, got ${residuals.size}") spark.sql("DROP TABLE test_cat.db.residual_dedup_test") From eedc6ae576fa5d23974cafd43bb3c32f8b66ed96 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Mon, 20 Jul 2026 19:26:54 -0400 Subject: [PATCH 07/12] clean up --- native/proto/src/proto/operator.proto | 4 + .../operator/CometIcebergNativeScan.scala | 88 +++++++------------ 2 files changed, 37 insertions(+), 55 deletions(-) diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 826eed9ce6..f242635f02 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -184,6 +184,10 @@ message IcebergLiteral { } } +// A decimal literal. `scale`/`precision` let the native side build a self-typed Datum for +// predicate pushdown (iceberg-rust's Datum is self-typed); partition-value decode reads only +// `unscaled`. Predicate decimal pushdown is deferred (iceberg-rust page-index gap), so today only +// `unscaled` is consumed. message IcebergDecimal { bytes unscaled = 1; // unscaled value, two's-complement big-endian int32 precision = 2; diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 8272cf2222..1acc0afa3a 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -19,8 +19,10 @@ package org.apache.comet.serde.operator +import java.math.BigDecimal +import java.nio.ByteBuffer import java.nio.charset.StandardCharsets.UTF_8 -import java.util.Base64 +import java.util.{Base64, UUID} import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -63,7 +65,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val GT = "GT" val GT_EQ = "GT_EQ" val IN = "IN" - val NOT_IN = "NOT_IN" } // Iceberg expression class name suffixes @@ -75,6 +76,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } + /** Iceberg binary comparison operation names mapped to their IcebergPredicate operator. */ + private val binaryOps: Map[String, OperatorOuterClass.IcebergPredicateOperator] = Map( + Constants.Operations.EQ -> OperatorOuterClass.IcebergPredicateOperator.Eq, + Constants.Operations.NOT_EQ -> OperatorOuterClass.IcebergPredicateOperator.NotEq, + Constants.Operations.LT -> OperatorOuterClass.IcebergPredicateOperator.LessThan, + Constants.Operations.LT_EQ -> OperatorOuterClass.IcebergPredicateOperator.LessThanOrEq, + Constants.Operations.GT -> OperatorOuterClass.IcebergPredicateOperator.GreaterThan, + Constants.Operations.GT_EQ -> OperatorOuterClass.IcebergPredicateOperator.GreaterThanOrEq) + /** * Wraps an Iceberg partition value (a typed primitive) in a PartitionValue. The value encoding * is shared with predicate literals via [[icebergLiteralToProto]]. @@ -122,8 +132,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit case d if d.startsWith("decimal(") => val bigDecimal = value match { - case bd: java.math.BigDecimal => bd - case _ => new java.math.BigDecimal(value.toString) + case bd: BigDecimal => bd + case _ => new BigDecimal(value.toString) } builder.setDecimalVal( OperatorOuterClass.IcebergDecimal @@ -178,15 +188,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit case "uuid" => // UUID as bytes (16 bytes) or string val uuidBytes = value match { - case uuid: java.util.UUID => - val bb = java.nio.ByteBuffer.wrap(new Array[Byte](16)) + case uuid: UUID => + val bb = ByteBuffer.wrap(new Array[Byte](16)) bb.putLong(uuid.getMostSignificantBits) bb.putLong(uuid.getLeastSignificantBits) bb.array() case _ => // Parse UUID string and convert to bytes - val uuid = java.util.UUID.fromString(value.toString) - val bb = java.nio.ByteBuffer.wrap(new Array[Byte](16)) + val uuid = UUID.fromString(value.toString) + val bb = ByteBuffer.wrap(new Array[Byte](16)) bb.putLong(uuid.getMostSignificantBits) bb.putLong(uuid.getLeastSignificantBits) bb.array() @@ -564,48 +574,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull)) case IS_NOT_NULL | NOT_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull)) - case EQ => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.Eq) - case NOT_EQ => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.NotEq) - case LT => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.LessThan) - case LT_EQ => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.LessThanOrEq) - case GT => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.GreaterThan) - case GT_EQ => - binaryPredicate( - exprClass, - icebergExpr, - columnName, - attribute, - IcebergPredicateOperator.GreaterThanOrEq) + case op if binaryOps.contains(op) => + binaryPredicate(exprClass, icebergExpr, columnName, attribute, binaryOps(op)) case IN => setPredicate(exprClass, icebergExpr, columnName, attribute) // NOT_IN is inherently unprunable from column stats, so it is not pushed. case _ => None @@ -636,7 +606,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit None } } catch { - case _: Exception => + // Reflection over Iceberg's expression classes can fail on an unexpected shape (e.g. an + // Iceberg version change). A residual is only a pruning hint, so skip pushdown rather than + // fail the scan, but log it: a persistent warning here signals a real API drift to fix. + case e: Exception => + logWarning( + "Skipping Iceberg residual pushdown; could not convert expression: " + + s"${e.getMessage}") None } } @@ -687,11 +663,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit .newBuilder() .setColumn(column) .setOp(OperatorOuterClass.IcebergPredicateOperator.In) - // Sort literals so an IN list Iceberg happens to iterate in a different order per file still - // deduplicates in the residual pool. + // Sort literals so an IN list that Iceberg happens to iterate in a different order per file + // still deduplicates in the residual pool. Encode each key once (sortBy re-invokes its key + // function per comparison). protoLiterals.flatten - .sortBy(lit => Base64.getEncoder.encodeToString(lit.toByteArray)) - .foreach(setBuilder.addValues) + .map(lit => (Base64.getEncoder.encodeToString(lit.toByteArray), lit)) + .sortBy(_._1) + .foreach { case (_, lit) => setBuilder.addValues(lit) } Some(OperatorOuterClass.IcebergPredicate.newBuilder().setSet(setBuilder).build()) } } From 3126842604250448da70746b2abdac2b4c0ac9f6 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 08:32:26 -0400 Subject: [PATCH 08/12] fix CI failures by adding stronger detection of unsupported types for column index and tests --- .../comet/iceberg/IcebergReflection.scala | 32 +++ .../operator/CometIcebergNativeScan.scala | 82 ++++-- .../comet/CometIcebergNativeSuite.scala | 239 ++++++++++++++++++ 3 files changed, 327 insertions(+), 26 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index 9f48ddfa9e..26c929f598 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -674,6 +674,38 @@ object IcebergReflection extends Logging { } } + /** + * Top-level column names whose Iceberg type is stored as Parquet FIXED_LEN_BYTE_ARRAY (decimal, + * uuid, fixed). iceberg-rust's page-index evaluator rejects that physical column-index type + * (page_index_evaluator.rs), so any residual predicate over such a column fails the native scan + * during pruning, including the IS NOT NULL that Iceberg adds for every filtered column. + * Callers must not push predicates on these columns. Extend this set as Iceberg adds + * FIXED_LEN_BYTE_ARRAY types (e.g. geometry). + */ + def pageIndexUnsupportedColumns(schema: Any): Set[String] = { + import scala.jdk.CollectionConverters._ + try { + val columns = schema.getClass + .getMethod("columns") + .invoke(schema) + .asInstanceOf[java.util.List[_]] + columns.asScala.flatMap { column => + val name = column.getClass.getMethod("name").invoke(column).asInstanceOf[String] + val typeStr = column.getClass.getMethod("type").invoke(column).toString + if (typeStr.startsWith("decimal(") || typeStr == "uuid" || typeStr.startsWith("fixed[")) { + Some(name) + } else { + None + } + }.toSet + } catch { + case e: Exception => + logWarning( + s"Failed to inspect schema for page-index-unsupported columns: ${e.getMessage}") + Set.empty[String] + } + } + /** * Validates partition column types for compatibility with iceberg-rust. * diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 1acc0afa3a..9bf680e9a5 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -550,13 +550,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit * Residuals come from Iceberg's ResidualEvaluator (partial evaluation of the scan filter * against each file's partition data). This is only a pruning hint: the CometFilter above the * scan enforces correctness, so any node or literal we cannot represent yields None (no - * pushdown). Uses reflection because Iceberg's expression classes are not on Spark's classpath - * at planning time; residuals are unbound predicates carrying a NamedReference (column name) - * and a literal. + * pushdown). Predicates over `pageIndexUnsupportedColumns` also yield None (iceberg-rust cannot + * use those columns in the page index). Uses reflection because Iceberg's expression classes + * are not on Spark's classpath at planning time; residuals are unbound predicates carrying a + * NamedReference (column name) and a literal. */ def icebergExprToProto( icebergExpr: Any, - output: Seq[Attribute]): Option[OperatorOuterClass.IcebergPredicate] = { + output: Seq[Attribute], + pageIndexUnsupportedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] = { try { val exprClass = icebergExpr.getClass val attributeMap = output.map(attr => attr.name -> attr).toMap @@ -570,30 +572,49 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit attributeMap.get(columnName).flatMap { attribute => import Constants.Operations._ import OperatorOuterClass.IcebergPredicateOperator - operation match { - case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull)) - case IS_NOT_NULL | NOT_NULL => - Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull)) - case op if binaryOps.contains(op) => - binaryPredicate(exprClass, icebergExpr, columnName, attribute, binaryOps(op)) - case IN => setPredicate(exprClass, icebergExpr, columnName, attribute) - // NOT_IN is inherently unprunable from column stats, so it is not pushed. - case _ => None + if (pageIndexUnsupportedColumns.contains(columnName)) { + // Any predicate on this column, including a unary IS [NOT] NULL, would reach the page + // index and fail, so drop the whole predicate; the post-scan CometFilter enforces it. + None + } else { + operation match { + case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull)) + case IS_NOT_NULL | NOT_NULL => + Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull)) + case op if binaryOps.contains(op) => + binaryPredicate(exprClass, icebergExpr, columnName, attribute, binaryOps(op)) + case IN => setPredicate(exprClass, icebergExpr, columnName, attribute) + // NOT_IN is inherently unprunable from column stats, so it is not pushed. + case _ => None + } } } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.AND)) { - val left = icebergExprToProto(exprClass.getMethod("left").invoke(icebergExpr), output) - val right = icebergExprToProto(exprClass.getMethod("right").invoke(icebergExpr), output) + val left = icebergExprToProto( + exprClass.getMethod("left").invoke(icebergExpr), + output, + pageIndexUnsupportedColumns) + val right = icebergExprToProto( + exprClass.getMethod("right").invoke(icebergExpr), + output, + pageIndexUnsupportedColumns) (left, right) match { - // Dropping a conjunct only weakens the pruning predicate, so keep any convertible side. + // Push the residual only if it converts whole. Dropping a conjunct is safe in positive + // position but strengthens the predicate under a NOT (De Morgan), which would wrongly + // prune, and tracking polarity across arbitrary nesting is error prone. So an + // unconvertible conjunct elides the whole residual; the post-scan CometFilter is exact. case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = true, l, r)) - case (Some(p), None) => Some(p) - case (None, Some(p)) => Some(p) - case (None, None) => None + case _ => None } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.OR)) { - val left = icebergExprToProto(exprClass.getMethod("left").invoke(icebergExpr), output) - val right = icebergExprToProto(exprClass.getMethod("right").invoke(icebergExpr), output) + val left = icebergExprToProto( + exprClass.getMethod("left").invoke(icebergExpr), + output, + pageIndexUnsupportedColumns) + val right = icebergExprToProto( + exprClass.getMethod("right").invoke(icebergExpr), + output, + pageIndexUnsupportedColumns) // Dropping a disjunct would strengthen the predicate and wrongly prune, so require both. (left, right) match { case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = false, l, r)) @@ -601,7 +622,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.NOT)) { val child = exprClass.getMethod("child").invoke(icebergExpr) - icebergExprToProto(child, output).map(notPredicate) + icebergExprToProto(child, output, pageIndexUnsupportedColumns).map(notPredicate) } else { None } @@ -699,9 +720,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit /** * Builds a predicate literal from its Iceberg value and the column's Spark type. Returns None - * for null (IS NULL is a separate op) and for types not pushed for pruning (byte-array-backed - * types, which iceberg-rust cannot use in the page index yet, and anything unmapped), so the - * predicate degrades to no pushdown rather than an incorrect one. + * for null (IS NULL is a separate op) and for Spark types this serde does not map to an + * IcebergLiteral (e.g. binary), so the predicate degrades to no pushdown rather than an + * incorrect one. Columns backed by Parquet FIXED_LEN_BYTE_ARRAY (decimal/uuid/fixed) are + * dropped earlier by the column gate in icebergExprToProto and never reach here. */ private def predicateLiteralToProto( sparkType: DataType, @@ -811,6 +833,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val deleteFilesToPoolIndex = mutable.HashMap[Seq[Int], Int]() val residualToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergPredicate, Int]() + // Columns whose Iceberg type iceberg-rust cannot use for page-index pruning; residual + // predicates over them are dropped (see icebergExprToProto). Computed once from the full table + // schema so a filter column projected out of the scan output is still recognized. + val pageIndexUnsupportedColumns = + IcebergReflection.pageIndexUnsupportedColumns(metadata.tableSchema) val perPartitionBuilders = mutable.ArrayBuffer[OperatorOuterClass.IcebergScan]() @@ -1011,7 +1038,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val residualExprOpt = try { - icebergExprToProto(residualMethod.invoke(task), output) + icebergExprToProto( + residualMethod.invoke(task), + output, + pageIndexUnsupportedColumns) } catch { case e: Exception => logWarning( diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 5cec19a81d..bc38877a1d 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -604,6 +604,11 @@ class CometIcebergNativeSuite // (getSerializedSize() returns int, so it wraps to a negative array length). test("delete file pool does not duplicate shared delete files (serde size)") { assume(icebergAvailable, "Iceberg not available in classpath") + // Reading scan.commonData forces Iceberg planning (ParallelIterable), which on older Iceberg + // leaves a prefetched manifest stream open that Spark's DebugFilesystem flags at teardown. + assume( + icebergVersionAtLeast(1, 8), + "ParallelIterable leaks manifest streams on older Iceberg") withTempIcebergDir { warehouseDir => withSQLConf( @@ -2645,6 +2650,235 @@ class CometIcebergNativeSuite } } + // Reproducer for the iceberg-rust FIXED_LEN_BYTE_ARRAY page-index gap (comet#4982), mirroring + // Iceberg's TestSparkReaderWithBloomFilter: a table with decimal columns (stored as + // FIXED_LEN_BYTE_ARRAY), bloom filters, tiny row groups (so Parquet writes a page index), and a + // wide AND filter that includes decimal equalities. Getting this to fail lets us instrument why a + // predicate reaches iceberg-rust's PageIndexEvaluator on a decimal column. + test("filter on a table with a decimal column does not fail the native scan") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.decimal_filter_test ( + id INT, + id_long BIGINT, + id_double DOUBLE, + id_float FLOAT, + id_string STRING, + id_boolean BOOLEAN, + id_date DATE, + id_int_decimal DECIMAL(8, 2), + id_long_decimal DECIMAL(14, 2), + id_fixed_decimal DECIMAL(31, 2) + ) USING iceberg + TBLPROPERTIES ( + 'format-version' = '2', + 'write.parquet.row-group-size-bytes' = '100', + 'write.parquet.bloom-filter-enabled.column.id' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_long' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_double' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_float' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_string' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_boolean' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_date' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_int_decimal' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_long_decimal' = 'true', + 'write.parquet.bloom-filter-enabled.column.id_fixed_decimal' = 'true' + ) + """) + + // 300 rows (ids 30..329), matching the Iceberg test's value formulas. + spark.sql(""" + INSERT INTO test_cat.db.decimal_filter_test + SELECT + CAST(id AS INT), + CAST(id + 1000 AS BIGINT), + CAST(id + 10000 AS DOUBLE), + CAST(id + 100000 AS FLOAT), + CONCAT('BINARY_', CAST(id AS STRING)), + true, + DATE '2021-09-05', + CAST(77.77 AS DECIMAL(8, 2)), + CAST(88.88 AS DECIMAL(14, 2)), + CAST(99.99 AS DECIMAL(31, 2)) + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.decimal_filter_test WHERE id = 30 AND id_long = 1030 " + + "AND id_double = 10030.0 AND id_float = 100030.0 AND id_string = 'BINARY_30' " + + "AND id_boolean = true AND id_date = '2021-09-05' AND id_int_decimal = 77.77 " + + "AND id_long_decimal = 88.88 AND id_fixed_decimal = 99.99") + + spark.sql("DROP TABLE test_cat.db.decimal_filter_test") + } + } + } + + // Iceberg stores uuid as Parquet FIXED_LEN_BYTE_ARRAY(16), the same physical layout as decimal + // and fixed. iceberg-rust's page-index evaluator does not support FIXED_LEN_BYTE_ARRAY, so a + // predicate pushed over such a column fails the native scan during page-index pruning. The scan + // must drop the residual and stay native, leaving the filter to the post-scan CometFilter. + // The trigger is IS NOT NULL (a unary predicate): it binds without a literal and reaches the + // page-index read. An equality would push a string literal that iceberg-rust cannot bind to a + // UUID field (Datum bind does not coerce), dropping the whole residual before any page probe. + test("filter on a table with a uuid column does not fail the native scan") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + import org.apache.iceberg.types.Types + import org.apache.iceberg.{PartitionSpec, Schema} + + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + + spark.sql("CREATE NAMESPACE IF NOT EXISTS test_cat.db") + + // uuid is not expressible via Spark SQL CREATE TABLE, so build the table with the Iceberg + // API. A tiny row-group size forces multiple row groups, so Parquet writes a column index + // for the uuid column that the pushed predicate then probes. + val schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "u", Types.UUIDType.get())) + val tableIdent = TableIdentifier.of("db", "uuid_filter_test") + val props = new java.util.HashMap[String, String]() + props.put("format-version", "2") + props.put("write.parquet.row-group-size-bytes", "100") + sparkCatalog.icebergCatalog + .createTable(tableIdent, schema, PartitionSpec.unpartitioned(), props) + + spark.sql(""" + INSERT INTO test_cat.db.uuid_filter_test + SELECT CAST(id AS INT), 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11' + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.uuid_filter_test WHERE id = 30 AND u IS NOT NULL") + + spark.sql("DROP TABLE test_cat.db.uuid_filter_test") + } + } + } + + // Iceberg stores fixed[N] as Parquet FIXED_LEN_BYTE_ARRAY(N). As with uuid and decimal, + // iceberg-rust's page-index evaluator does not support that physical type, so the IS NOT NULL + // pushed over the fixed column (added by Iceberg for any filtered column) would fail the native + // scan. The residual must be dropped so the scan stays native and CometFilter enforces the filter. + test("filter on a table with a fixed column does not fail the native scan") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + import org.apache.iceberg.types.Types + import org.apache.iceberg.{PartitionSpec, Schema} + + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + + spark.sql("CREATE NAMESPACE IF NOT EXISTS test_cat.db") + + val schema = new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.optional(2, "f", Types.FixedType.ofLength(16))) + val tableIdent = TableIdentifier.of("db", "fixed_filter_test") + val props = new java.util.HashMap[String, String]() + props.put("format-version", "2") + props.put("write.parquet.row-group-size-bytes", "100") + sparkCatalog.icebergCatalog + .createTable(tableIdent, schema, PartitionSpec.unpartitioned(), props) + + spark.sql(""" + INSERT INTO test_cat.db.fixed_filter_test + SELECT CAST(id AS INT), X'00112233445566778899aabbccddeeff' + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.fixed_filter_test WHERE id = 30 AND f IS NOT NULL") + + spark.sql("DROP TABLE test_cat.db.fixed_filter_test") + } + } + } + + // A residual can arrive as NOT over an AND that mixes a supported conjunct with an unsupported + // (FIXED_LEN_BYTE_ARRAY) one. Dropping only the unsupported conjunct is safe in positive + // position (it weakens the pruning predicate), but under a NOT it strengthens it: NOT(id < 200 + // AND d = 100.00) is really id >= 200 OR d != 100.00, yet dropping the decimal conjunct and + // negating leaves id >= 200, which prunes pages holding id < 200 rows that satisfy the filter + // via d != 100.00. The residual must not be pushed partially in that case. This asserts results + // match Spark; d is stored as decimal so its predicate is dropped by the page-index gate. + test("NOT over a partially supported AND does not drop rows") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.not_partial_residual_test ( + id INT, + d DECIMAL(10, 2) + ) USING iceberg + TBLPROPERTIES ( + 'format-version' = '2', + 'write.parquet.row-group-size-bytes' = '100' + ) + """) + + // id ascending so row groups hold contiguous id ranges; a stronger id >= 200 pushed + // predicate would prune the id < 200 groups entirely. + spark.sql(""" + INSERT INTO test_cat.db.not_partial_residual_test + SELECT CAST(id AS INT), CAST(id AS DECIMAL(10, 2)) + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.not_partial_residual_test " + + "WHERE NOT(id < 200 AND d = 100.00)") + + spark.sql("DROP TABLE test_cat.db.not_partial_residual_test") + } + } + } + // Regression test for https://github.com/apache/datafusion-comet/issues/3856 // Fixed in https://github.com/apache/iceberg-rust/pull/2301 test("migration - INT96 timestamp") { @@ -3090,6 +3324,11 @@ class CometIcebergNativeSuite // overflow the broadcast IcebergScanCommon. test("residual pool dedups a shared non-partition filter to one entry") { assume(icebergAvailable, "Iceberg not available in classpath") + // Reading scan.commonData forces Iceberg planning (ParallelIterable), which on older Iceberg + // leaves a prefetched manifest stream open that Spark's DebugFilesystem flags at teardown. + assume( + icebergVersionAtLeast(1, 8), + "ParallelIterable leaks manifest streams on older Iceberg") withTempIcebergDir { warehouseDir => withSQLConf( From e60b1923b7c9d29d4946f6711acb5ae8d6626c6a Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 08:32:46 -0400 Subject: [PATCH 09/12] fix style --- .../test/scala/org/apache/comet/CometIcebergNativeSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index bc38877a1d..e4fcee9815 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2784,7 +2784,7 @@ class CometIcebergNativeSuite // Iceberg stores fixed[N] as Parquet FIXED_LEN_BYTE_ARRAY(N). As with uuid and decimal, // iceberg-rust's page-index evaluator does not support that physical type, so the IS NOT NULL // pushed over the fixed column (added by Iceberg for any filtered column) would fail the native - // scan. The residual must be dropped so the scan stays native and CometFilter enforces the filter. + // scan. The residual must be dropped so the scan stays native and CometFilter enforces it. test("filter on a table with a fixed column does not fail the native scan") { assume(icebergAvailable, "Iceberg not available in classpath") From 67de5dc4f9926f29eeba731b48d3465fb87abad8 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 19:17:25 -0400 Subject: [PATCH 10/12] Address PR feedback. --- native/core/src/execution/planner.rs | 13 +----- .../comet/iceberg/IcebergReflection.scala | 18 +++++--- .../operator/CometIcebergNativeScan.scala | 14 ++++-- .../comet/CometIcebergNativeSuite.scala | 45 +++++++++++++++++++ 4 files changed, 67 insertions(+), 23 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 451b8a8b8f..744709db32 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4092,20 +4092,9 @@ fn literal_to_array_ref( } // ============================================================================ -// Spark Expression to Iceberg Predicate Conversion +// Iceberg Residual Predicate Conversion // ============================================================================ -// -// Predicates are converted through Spark expressions rather than directly from -// Iceberg Java to Iceberg Rust. This leverages Comet's existing expression -// serialization infrastructure, which handles hundreds of expression types. -// -// Conversion path: -// Iceberg Expression (Java) -> Spark Catalyst Expression -> Protobuf -> Iceberg Predicate (Rust) -// -// Note: NOT IN predicates are skipped because iceberg-rust's RowGroupMetricsEvaluator::not_in() -// always returns MIGHT_MATCH (never prunes row groups). These are handled by CometFilter post-scan. -/// Converts a protobuf Spark expression to an Iceberg predicate for row-group filtering. /// Converts a serialized Iceberg residual predicate into an iceberg-rust `Predicate` for row-group /// pruning. This is only a pruning hint -- the post-scan CometFilter enforces correctness -- so any /// node or literal that cannot be represented degrades to `None` (no pushdown) rather than an diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index 26c929f598..1a5cea7c5c 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -675,12 +675,15 @@ object IcebergReflection extends Logging { } /** - * Top-level column names whose Iceberg type is stored as Parquet FIXED_LEN_BYTE_ARRAY (decimal, - * uuid, fixed). iceberg-rust's page-index evaluator rejects that physical column-index type - * (page_index_evaluator.rs), so any residual predicate over such a column fails the native scan - * during pruning, including the IS NOT NULL that Iceberg adds for every filtered column. - * Callers must not push predicates on these columns. Extend this set as Iceberg adds - * FIXED_LEN_BYTE_ARRAY types (e.g. geometry). + * Top-level column names whose Iceberg type iceberg-rust's page-index evaluator cannot prune + * over, so callers must not push a residual predicate on them, not even the IS NOT NULL that + * Iceberg adds for every filtered column. Two physical layouts fail (page_index_evaluator.rs): + * - FIXED_LEN_BYTE_ARRAY (decimal, uuid, fixed): rejected outright as an unsupported index + * type, which fails the native scan. + * - BYTE_ARRAY backing a binary column: the evaluator decodes column-index min/max as UTF-8 + * (String::from_utf8(..).unwrap()) before the predicate closure runs, so non-UTF-8 bounds + * panic the native scan even for a bare IS [NOT] NULL. Extend this set as Iceberg adds + * types with either layout (e.g. geometry). */ def pageIndexUnsupportedColumns(schema: Any): Set[String] = { import scala.jdk.CollectionConverters._ @@ -692,7 +695,8 @@ object IcebergReflection extends Logging { columns.asScala.flatMap { column => val name = column.getClass.getMethod("name").invoke(column).asInstanceOf[String] val typeStr = column.getClass.getMethod("type").invoke(column).toString - if (typeStr.startsWith("decimal(") || typeStr == "uuid" || typeStr.startsWith("fixed[")) { + if (typeStr.startsWith("decimal(") || typeStr == "uuid" || typeStr.startsWith("fixed[") || + typeStr == "binary") { Some(name) } else { None diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index 9bf680e9a5..247a2f9245 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -569,6 +569,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val ref = term.getClass.getMethod("ref").invoke(term) val columnName = ref.getClass.getMethod("name").invoke(ref).asInstanceOf[String] + // Iceberg names a nested reference by its dotted path ("struct.field"), which never matches + // a top-level scan output attribute, so a residual on a nested field drops here. That miss + // is also why the top-level-only pageIndexUnsupportedColumns gate below stays sound. attributeMap.get(columnName).flatMap { attribute => import Constants.Operations._ import OperatorOuterClass.IcebergPredicateOperator @@ -721,9 +724,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit /** * Builds a predicate literal from its Iceberg value and the column's Spark type. Returns None * for null (IS NULL is a separate op) and for Spark types this serde does not map to an - * IcebergLiteral (e.g. binary), so the predicate degrades to no pushdown rather than an - * incorrect one. Columns backed by Parquet FIXED_LEN_BYTE_ARRAY (decimal/uuid/fixed) are - * dropped earlier by the column gate in icebergExprToProto and never reach here. + * IcebergLiteral, so the predicate degrades to no pushdown rather than an incorrect one. + * Columns backed by Parquet FIXED_LEN_BYTE_ARRAY (decimal/uuid/fixed) or BYTE_ARRAY (binary) + * are dropped earlier by the column gate in icebergExprToProto and never reach here. */ private def predicateLiteralToProto( sparkType: DataType, @@ -734,7 +737,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val builder = OperatorOuterClass.IcebergLiteral.newBuilder() sparkType match { case _: BooleanType => builder.setBoolVal(value.asInstanceOf[java.lang.Boolean]) - case _: IntegerType => builder.setIntVal(value.asInstanceOf[java.lang.Number].intValue()) + // Spark byte/short map to Iceberg int (32-bit); iceberg-rust has no narrower integer type, so + // widening to int_val matches how it stores and prunes these columns. + case _: ByteType | _: ShortType | _: IntegerType => + builder.setIntVal(value.asInstanceOf[java.lang.Number].intValue()) case _: LongType => builder.setLongVal(value.asInstanceOf[java.lang.Number].longValue()) case _: FloatType => builder.setFloatVal(value.asInstanceOf[java.lang.Number].floatValue()) case _: DoubleType => diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index e4fcee9815..ba61ebf76a 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2832,6 +2832,51 @@ class CometIcebergNativeSuite } } + // Iceberg stores binary as Parquet BYTE_ARRAY. Unlike the FIXED_LEN_BYTE_ARRAY types above, + // iceberg-rust's page-index evaluator accepts BYTE_ARRAY but decodes its column-index min/max as + // UTF-8 (String::from_utf8(..).unwrap()), so non-UTF-8 bounds panic the native scan. The decode + // happens in calc_row_selection before the null-count closure runs, so even a bare IS NOT NULL + // (added by Iceberg for any filtered column) triggers it. Binary must therefore be gated like the + // FIXED_LEN_BYTE_ARRAY types: the residual is dropped and CometFilter enforces the filter. + test("filter on a table with a binary column does not fail the native scan") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.binary_filter_test ( + id INT, + b BINARY + ) USING iceberg + TBLPROPERTIES ( + 'format-version' = '2', + 'write.parquet.row-group-size-bytes' = '100' + ) + """) + + // Prepend a 0xFF byte (never valid UTF-8) so the column-index min/max for `b` cannot decode + // as a UTF-8 string, which is what makes iceberg-rust's page-index evaluator panic. + spark.sql(""" + INSERT INTO test_cat.db.binary_filter_test + SELECT CAST(id AS INT), concat(X'FF', CAST(CAST(id AS STRING) AS BINARY)) + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.binary_filter_test WHERE id = 30 AND b IS NOT NULL") + + spark.sql("DROP TABLE test_cat.db.binary_filter_test") + } + } + } + // A residual can arrive as NOT over an AND that mixes a supported conjunct with an unsupported // (FIXED_LEN_BYTE_ARRAY) one. Dropping only the unsupported conjunct is safe in positive // position (it weakens the pruning predicate), but under a NOT it strengthens it: NOT(id < 200 From b41e7837a274ca833b7be211ed260ef03e218563 Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Tue, 21 Jul 2026 19:30:04 -0400 Subject: [PATCH 11/12] Address PR feedback. --- .../apache/comet/CometFuzzIcebergBase.scala | 2 +- .../apache/comet/CometFuzzIcebergSuite.scala | 67 +++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergBase.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergBase.scala index fbace69140..88d3ff9b08 100644 --- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergBase.scala +++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergBase.scala @@ -58,7 +58,7 @@ class CometFuzzIcebergBase extends CometTestBase with AdaptiveSparkPlanHelper { } } - private def isIcebergVersionLessThan(targetVersion: String): Boolean = { + protected def isIcebergVersionLessThan(targetVersion: String): Boolean = { try { val icebergVersion = org.apache.iceberg.IcebergBuild.version() // Parse version strings like "1.5.2" or "1.6.0-SNAPSHOT" diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala index 5e8cca0c3a..e9bda4c6b4 100644 --- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala @@ -24,8 +24,10 @@ import scala.util.Random import org.apache.spark.sql.Column import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec import org.apache.spark.sql.functions.{col, lit} +import org.apache.spark.sql.types._ import org.apache.comet.DataTypeSupport.isComplexType +import org.apache.comet.serde.OperatorOuterClass class CometFuzzIcebergSuite extends CometFuzzIcebergBase { @@ -170,6 +172,71 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase { } } + // Asserts that predicate pushdown actually happens for the types that support it. For each + // primitive column it runs an equality filter and inspects the serialized IcebergScanCommon: the + // residual pool must be non-empty exactly when the column's type is one Comet maps to an Iceberg + // literal (pushable), and empty for a gated type (binary). This works because the fuzz table is + // unpartitioned, so Iceberg's per-file residual is the full filter rather than a partition-pruned + // remnant. It exists because the correctness test above cannot see pushdown at all -- results + // match and the scan stays native whether or not a predicate pushes, since CometFilter enforces + // it either way -- so only a residual-pool check catches a regression like byte/short mapping to + // None or binary escaping the page-index gate. + test("filter pushdown - residual pool reflects whether the column type is pushable") { + // Reading commonData forces Iceberg planning (ParallelIterable), which on older Iceberg leaves + // a prefetched manifest stream open that Spark's DebugFilesystem flags at teardown. + assume( + !isIcebergVersionLessThan("1.8.0"), + "ParallelIterable leaks manifest streams on older Iceberg") + + // Types predicateLiteralToProto maps to an IcebergLiteral. Everything else present in the fuzz + // table (binary) is gated or unmapped and must not push. + def isPushable(dt: DataType): Boolean = dt match { + case _: BooleanType | _: ByteType | _: ShortType | _: IntegerType | _: LongType | + _: FloatType | _: DoubleType | _: DateType | _: StringType | _: TimestampType | + _: TimestampNTZType => + true + case _ => false + } + + val df = spark.table(icebergTableName) + val primitiveColumns = df.schema.fields.filterNot(f => isComplexType(f.dataType)) + assert(primitiveColumns.nonEmpty, "expected at least one primitive column in the fuzz schema") + + for (field <- primitiveColumns) { + val name = field.name + val sample = df + .select(col(name)) + .where(col(name).isNotNull) + .limit(1) + .collect() + .headOption + .map(_.get(0)) + // Skip NaN, whose equality semantics differ, so the sampled literal is well-behaved. + .filterNot { + case d: Double => d.isNaN + case f: Float => f.isNaN + case _ => false + } + + sample.foreach { v => + val (_, cometPlan) = checkSparkAnswer(df.where(col(name) === lit(v))) + val scans = collectIcebergNativeScans(cometPlan) + assert(scans.length == 1, s"expected one native scan for '$name'\n$cometPlan") + val common = OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData) + val pushed = common.getResidualPoolCount > 0 + if (isPushable(field.dataType)) { + assert( + pushed, + s"expected a pushed residual for pushable column '$name' (${field.dataType})") + } else { + assert( + !pushed, + s"expected no residual pushdown for gated column '$name' (${field.dataType})") + } + } + } + } + def collectCometShuffleExchanges(plan: org.apache.spark.sql.execution.SparkPlan) : Seq[org.apache.spark.sql.execution.SparkPlan] = { collect(plan) { From bd2418a95b6f4f58e4dcbcb22dc9a4a96f405a6a Mon Sep 17 00:00:00 2001 From: Matt Butrovich Date: Thu, 23 Jul 2026 10:56:07 -0400 Subject: [PATCH 12/12] address PR feedback --- native/core/src/execution/planner.rs | 52 ++++++++++++++----- .../comet/CometIcebergNativeSuite.scala | 44 ++++++++++++++++ 2 files changed, 82 insertions(+), 14 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 8bf604df98..a790de5ce6 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4104,7 +4104,7 @@ fn literal_to_array_ref( fn iceberg_predicate_to_predicate( proto: &spark_operator::IcebergPredicate, ) -> Option { - use iceberg::expr::Reference; + use iceberg::expr::{Predicate, Reference}; use spark_operator::iceberg_predicate::Node; use spark_operator::IcebergPredicateOperator as Op; @@ -4147,29 +4147,53 @@ fn iceberg_predicate_to_predicate( .collect::>>()?; Some(column.is_in(datums)) } + // And/Or are whole-or-nothing: both children must convert. Dropping a child is unsafe. + // A dropped disjunct strengthens an Or directly; a dropped conjunct is safe for a bare And + // but not under a NOT, where De Morgan turns And into Or (`NOT(A AND B)` = `NOT A OR NOT B`) + // and dropping strengthens it, wrongly pruning row groups. Since Not is only pushed to + // negation-normal form after decode (rewrite_not below), the Not(And(..)) shape does reach + // here, so the And arm cannot assume positive position. Degrading to no-pushdown is always + // safe; the post-scan CometFilter is exact. See `combine_logical` for the parity invariant. Node::And(l) => { - // Dropping a conjunct only weakens the pruning predicate, so a missing side is safe. let left = l.left.as_deref().and_then(iceberg_predicate_to_predicate); let right = l.right.as_deref().and_then(iceberg_predicate_to_predicate); - match (left, right) { - (Some(l), Some(r)) => Some(l.and(r)), - (Some(p), None) | (None, Some(p)) => Some(p), - (None, None) => None, - } + combine_logical(left, right, Predicate::and) } Node::Or(o) => { - // Dropping a disjunct would strengthen the predicate and wrongly prune; require both. - let left = o.left.as_deref().and_then(iceberg_predicate_to_predicate)?; - let right = o - .right - .as_deref() - .and_then(iceberg_predicate_to_predicate)?; - Some(left.or(right)) + let left = o.left.as_deref().and_then(iceberg_predicate_to_predicate); + let right = o.right.as_deref().and_then(iceberg_predicate_to_predicate); + combine_logical(left, right, Predicate::or) } Node::Not(child) => iceberg_predicate_to_predicate(child).map(|p| !p), } } +/// Combines the two children of a logical residual node (And/Or), returning `None` unless both +/// converted. A missing child is not expected: the Scala serde emits a logical node only when both +/// children convert, and Rust decodes every node it emits, so both sides always convert for a +/// well-formed residual. A `None` here therefore means the two serde paths have diverged (e.g. a +/// new `IcebergLiteral` variant on the serialize side without a matching decode arm), so it trips a +/// debug-only assertion and degrades to no-pushdown in release. See [`iceberg_predicate_to_predicate`] +/// for why a partial logical node must never be pushed. +fn combine_logical( + left: Option, + right: Option, + combine: fn(iceberg::expr::Predicate, iceberg::expr::Predicate) -> iceberg::expr::Predicate, +) -> Option { + match (left, right) { + (Some(l), Some(r)) => Some(combine(l, r)), + _ => { + debug_assert!( + false, + "Iceberg residual logical node with an unconvertible child: the Scala serde emits \ + both children and Rust decodes every node it emits, so the two serde paths have \ + diverged" + ); + None + } + } +} + /// Converts a serialized `IcebergLiteral` into an iceberg-rust `Datum` for predicate pushdown. /// Returns `None` for null and for byte-array-backed types (decimal/uuid/fixed/binary), which /// iceberg-rust cannot use in the page index yet; the driver does not emit those for predicates, diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index ba61ebf76a..bdad3163d4 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -2924,6 +2924,50 @@ class CometIcebergNativeSuite } } + // Companion to the partial-AND test above: here both conjuncts are pushable (both int), so the + // whole residual is pushed as NOT(a < 200 AND b > 100). rewrite_not on the native side turns that + // into a >= 200 OR b <= 100 (negation-normal form), exercising the both-convert-under-NOT path + // that the partial-AND elision test cannot reach. Asserts results match Spark. + test("NOT over a fully supported AND does not drop rows") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.not_full_residual_test ( + a INT, + b INT + ) USING iceberg + TBLPROPERTIES ( + 'format-version' = '2', + 'write.parquet.row-group-size-bytes' = '100' + ) + """) + + // a ascending so row groups hold contiguous a ranges; a stronger a >= 200 pushed + // predicate (from dropping the b conjunct under the NOT) would prune the a < 200 groups. + spark.sql(""" + INSERT INTO test_cat.db.not_full_residual_test + SELECT CAST(id AS INT), CAST(id AS INT) + FROM range(30, 330) + """) + + checkIcebergNativeScan( + "SELECT * FROM test_cat.db.not_full_residual_test " + + "WHERE NOT(a < 200 AND b > 100)") + + spark.sql("DROP TABLE test_cat.db.not_full_residual_test") + } + } + } + // Regression test for https://github.com/apache/datafusion-comet/issues/3856 // Fixed in https://github.com/apache/iceberg-rust/pull/2301 test("migration - INT96 timestamp") {