mbutrovich commented on code in PR #6847: URL: https://github.com/apache/datafusion-comet/pull/6847#discussion_r4329274950
########## docs/source/user-guide/latest/iceberg.md: ########## @@ -113,8 +113,30 @@ The native Iceberg reader supports the following features: NULL checks on struct, array, and map columns still use native scans and return correct results, but are not pushed into iceberg-rust, which binds accessors only for primitive fields. -These residuals provide no native row-group pruning, nor do conjunctions containing them; -safe partial pruning is tracked in [#5883](https://github.com/apache/datafusion-comet/issues/5883). +These null checks provide no native row-group pruning. In a positive conjunction, supported +primitive predicates still prune row groups and pages, while the post-scan filter evaluates the +complete condition. `OR` requires a safely pushable predicate from both branches, and expressions +inside `NOT` must convert completely before they can be pushed. When only part of a condition can +be pushed, predicates on `BIGINT` and `DOUBLE` columns in it are pushed only for `=` and null +checks, because a data file can store a narrower type than its column without any record of it. + +Tables with a `schema.name-mapping.default` property push a conjunction only when every predicate +in it can be pushed. The `add_files`, `migrate` and `snapshot` procedures set this property when +they import data files, and an imported file can store a narrower type than its column. + +Spark treats `-0.0` as equal to `0.0` and sorts every NaN above all numbers, but iceberg-rust +uses IEEE 754 total ordering. On `FLOAT` and `DOUBLE` columns, range comparisons and comparisons +with `0.0`, `-0.0` or NaN are therefore evaluated after the native scan. + +Some other predicates are also evaluated only after the native scan: + +- Negated equality on a nullable column. Iceberg passes `<=>` as `=`, so for `NOT (a <=> 1)` the + native reader would drop NULL rows that Spark keeps, and `a <> 1` reaches Comet in the same + form. +- Comparisons other than `=` with a literal on a column whose type was promoted, for example + from `INT` to `BIGINT`, because older data files keep the original type. If expiring + snapshots removed older schemas, this applies to every `BIGINT` and `DOUBLE` column. A + promotion made before Iceberg 0.12, which kept no schema history, cannot be detected. Review Comment: Could this section be shorter? It grows from two sentences to 24 lines, and most of the new text explains why each rule exists, which the code comments and the Known Issues entry already cover. It also says the kept predicates "prune row groups and pages", but iceberg-rust also uses them to filter rows during decode. Here's a version that keeps every rule a user can act on: ```suggestion These null checks are not pushed, but the other predicates in a positive conjunction still are. iceberg-rust uses a pushed predicate to skip row groups and pages and to filter rows during decode, and the filter above the scan evaluates the complete condition. `OR` is pushed only when both branches can be, and `NOT` only when its whole child can be. Some predicates are left to the filter above the scan because iceberg-rust would return different rows than Spark: - Range comparisons on `FLOAT` and `DOUBLE` columns, and comparisons with `0.0`, `-0.0` or NaN. - `NOT` over equality on a nullable column. - Comparisons other than `=` on a column whose type was promoted. This applies to every `BIGINT` and `DOUBLE` column when only part of a condition is pushed, or when expiring snapshots removed older schemas. Tables with a `schema.name-mapping.default` property push a condition only when all of it can be pushed. ``` ########## spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala: ########## @@ -679,17 +679,66 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit * - An exception means reflection over Iceberg's expression API did something unexpected, so * no such decision was reached. It propagates instead of folding into `None`, which would * make the two indistinguishable; see [[serializeResidual]]. + * + * @param promotedColumns + * columns whose older data files may hold a narrower type; see + * [[IcebergReflection.promotedColumns]] + * @param partialPushdown + * whether a residual that does not convert whole may push the conjuncts that do, leaving the + * rest to the post-scan filter */ def icebergExprToProto( icebergExpr: Any, output: Seq[Attribute], - pageIndexUnsupportedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] = { + pageIndexUnsupportedColumns: Set[String], + promotedColumns: Set[String] = Set.empty, + partialPushdown: Boolean = true): Option[OperatorOuterClass.IcebergPredicate] = { Review Comment: Should `promotedColumns` and `partialPushdown` be required here and in `serializeResidual` (lines 838-839)? The one production call, at line 1263, passes both. The defaults are the least safe pair, no promoted columns and partial pushdown allowed, so a new caller that leaves them out skips the promotion guard without a compile error. As I read it, only `CometIcebergNativeScanSuite` relies on them. ########## spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala: ########## @@ -679,17 +679,66 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit * - An exception means reflection over Iceberg's expression API did something unexpected, so * no such decision was reached. It propagates instead of folding into `None`, which would * make the two indistinguishable; see [[serializeResidual]]. + * + * @param promotedColumns + * columns whose older data files may hold a narrower type; see + * [[IcebergReflection.promotedColumns]] + * @param partialPushdown + * whether a residual that does not convert whole may push the conjuncts that do, leaving the + * rest to the post-scan filter */ def icebergExprToProto( icebergExpr: Any, output: Seq[Attribute], - pageIndexUnsupportedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] = { + pageIndexUnsupportedColumns: Set[String], + promotedColumns: Set[String] = Set.empty, + partialPushdown: Boolean = true): Option[OperatorOuterClass.IcebergPredicate] = { + val attributes = output.map(attr => attr.name -> attr).toMap + def convert(promoted: Set[String], partial: Boolean) = + icebergExprToProto( + icebergExpr, + ResidualScope(attributes, pageIndexUnsupportedColumns, promoted, partial), + underNot = false) + // Pushing part of a residual must not rely on the schema history, which misses some + // promotions, such as those made before Iceberg 0.12. So when the residual does not convert + // whole, every BIGINT and DOUBLE column counts as promoted. Whole residuals rely on the + // history, since limiting them too would stop range pruning on all such columns until + // apache/iceberg-rust#3122 is fixed. + convert(promotedColumns, partial = false).orElse { + if (partialPushdown) { + val widened = output.collect { + case attr if attr.dataType == LongType || attr.dataType == DoubleType => attr.name + } + convert(promotedColumns ++ widened, partial = true) + } else { + None + } + } + } + + /** What [[icebergExprToProto]] consults at every node of a residual. */ + private case class ResidualScope( + attributes: Map[String, Attribute], + pageIndexUnsupportedColumns: Set[String], + promotedColumns: Set[String], + partialPushdown: Boolean) Review Comment: What does `ResidualScope` represent? The doc comment says it's what the walk "consults at every node", which fits any argument. As I read it, it holds the settings for one of the two passes at lines 707-716, and it mixes values that are the same in both passes (`attributes`, `pageIndexUnsupportedColumns`) with values that differ (`promotedColumns`, `partialPushdown`). Meanwhile `underNot`, the one value that changes during the walk, stays a separate argument. The field `partialPushdown` also means something different from the parameter of the same name at line 695. The parameter says whether the table allows partial pushdown at all. The field says whether this pass may drop a conjunct, and it is `false` on the first pass even when the table allows it. Could the type and that field get names that say what they hold (for example `dropUnconvertedConjuncts` for the field), with a short doc comment on each field? -- 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]
