comphead commented on code in PR #7156:
URL: https://github.com/apache/datafusion-comet/pull/7156#discussion_r4379458711
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -4306,6 +4306,229 @@ class CometIcebergNativeSuite
}
}
+ /**
+ * Compares `SELECT id FROM table WHERE predicate` with Spark for each
predicate, alone and
+ * after `l IS NOT NULL AND`. That array null check is a conjunct the serde
never pushes, as in
+ * the queries that found these bugs, so each predicate is also checked in a
residual left
+ * unpushed.
+ */
+ private def checkResidualPredicates(table: String, predicates: Seq[String]):
Unit =
+ for (predicate <- predicates; prefix <- Seq("", "l IS NOT NULL AND ")) {
Review Comment:
With the prefix, the residual never converts. The AND branch needs both
sides, and the null check on the array column `l` is declined by
`isComplexType`, so each prefixed query reruns the unprefixed one with nothing
pushed. The docstring says as much. That doubles the query count of the three
new tests without reaching the code this PR changes. The prefix starts to
matter with partial conjuncts in #6847. Could the prefix and the `l` column
move there, so this PR and its backports only run the cases they fix?
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -720,6 +738,24 @@ 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))
+ // Files written before the promotion still hold the older type,
which iceberg-rust
+ // compares with a literal incorrectly (see
IcebergReflection.promotedColumns),
+ // except for equality outside NOT: pruning keeps every row
group and page for it,
+ // and a literal the older type cannot hold matches no row there
anyway.
+ case op if promotedColumns.contains(columnName) && (op != EQ ||
underNot) =>
+ None
+ // Iceberg converts Spark's null-safe `id <=> 1` to `id = 1`,
which Spark evaluates
+ // to false for a NULL id and the native reader to NULL. Both
reject the row, but
+ // for `NOT (id <=> 1)` Spark keeps it while the native reader
still drops it. A
+ // NOT_NULL conjunct of an enclosing AND makes the column
non-nullable here.
+ case EQ if underNot && attribute.nullable => None
+ // Spark sorts every NaN above all numbers. The native row
filter sorts negative NaNs
+ // below them, and iceberg-rust prunes row groups on min/max
statistics that can
+ // leave NaN out, or hold it as both when parquet-mr 1.9 or
earlier starts a row
+ // group with NaN (apache/iceberg-rust#3494).
Review Comment:
apache/iceberg-rust#3494 was closed on 2026-10-23 by
apache/iceberg-rust#3502 ("treat NaN row group bounds as might-match for
inequalities"), which is 54 commits past the pinned `1f3bc34`. Could the
comment name #3502, so the next pin bump knows which part of this exclusion it
can revisit? The fix is for row-group bounds, so the page index side would
still need checking then.
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -1216,6 +1216,48 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * Top-level column names of `schema` whose type differs in some schema of
the table's history
+ * (`table.schemas()`), such as an int column promoted to long. Data files
written before the
+ * promotion keep the older type, and iceberg-rust compares literals against
it incorrectly: the
+ * page index skips every page for range and IN predicates, and the row
filter casts the literal
+ * to the older type, so an out-of-range value becomes NULL
(apache/iceberg-rust#3122).
+ *
+ * Expiring snapshots with `cleanExpiredMetadata` removes schemas that no
retained snapshot
+ * uses, even though files written under them can remain. Iceberg numbers
schemas consecutively
+ * from 0, so a gap means the history is incomplete, and every column of a
type that a promotion
+ * can produce (long, double, decimal) is returned. The history cannot
reveal a file whose type
+ * differs without a recorded schema change, such as one imported by
`add_files`, `migrate` or
+ * `snapshot`, or one promoted under metadata from before Iceberg 0.12,
which kept only the
+ * current schema. Called at serialization time, so it throws on reflection
failure.
+ */
+ def promotedColumns(table: Any, schema: Any): Set[String] = {
+ import scala.jdk.CollectionConverters._
+ val findField = getMethod(schema.getClass, "findField", classOf[Int])
+ def typeOf(field: Any): String = getMethod(field.getClass,
"type").invoke(field).toString
Review Comment:
`fieldTypeOf` at line 1416 already does this lookup, so this can be
`fieldTypeOf(field).toString`, the way this method already reuses `fieldIdOf`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
