Skip to content
Open
17 changes: 17 additions & 0 deletions docs/source/user-guide/latest/iceberg.md
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,16 @@ results, but are not pushed into iceberg-rust, which binds accessors only for pr
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).

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, as in `NOT (id <=> 1)`, unless it is ANDed with an
`IS NOT NULL` check on that column, written in the query or inferred by Spark as for `b = false`.
`!=` reaches iceberg-rust as a separate comparison and is not affected.
- Comparisons other than `=` on a column whose type was promoted. This applies to every `BIGINT`
and `DOUBLE` column when expiring snapshots removed older schemas.

**Partitioning:**

- Standard partitioning with partition pruning
Expand Down Expand Up @@ -804,6 +814,13 @@ Which entries apply depends on which native paths are enabled:
a file as NULL, and the native scan returns the values stored in the file. A read of only some
of a struct's fields fails instead
([#6794](https://github.com/apache/datafusion-comet/issues/6794)).
- **Data files that store a narrower type than their column.** Files written before a type
promotion, such as `INT` to `BIGINT`, keep the older type, and files imported by `add_files`,
`migrate` or `snapshot` can store any narrower type. iceberg-rust compares such a file with a
filter literal incorrectly, so comparisons other than `=` pushed to the native scan can drop
matching rows. Comet avoids this for promotions that the schema history records, but not for
imported files or promotions made before Iceberg 0.12
([#7138](https://github.com/apache/datafusion-comet/issues/7138)).
- **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as
NULL, and two columns that swapped names return each other's values
([#6822](https://github.com/apache/datafusion-comet/issues/6822)).
Expand Down
56 changes: 40 additions & 16 deletions native/core/src/execution/planner.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2670,21 +2670,7 @@ fn parse_file_scan_tasks_from_common(
.residual_pool
.get(idx as usize)
.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.
// 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}");
None
}
}
})
.and_then(|pred| bind_residual(pred, Arc::clone(&schema_ref)))
} else {
None
};
Expand Down Expand Up @@ -3055,6 +3041,24 @@ fn iceberg_predicate_to_predicate(
}
}

/// Binds a residual predicate to the file's schema. The residual prunes reads and filters rows; the
/// post-scan filter enforces any omitted predicates. 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. 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).
fn bind_residual(
pred: iceberg::expr::Predicate,
schema: Arc<iceberg::spec::Schema>,
) -> Option<iceberg::expr::BoundPredicate> {
match pred.rewrite_not().bind(schema, true) {
Ok(bound) => Some(bound),
Err(e) => {
log::warn!("Skipping Iceberg predicate pushdown; bind failed: {e}");
None
}
}
}

/// 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
Expand Down Expand Up @@ -3178,7 +3182,7 @@ mod tests {

use crate::execution::operators::ExecutionError;
use crate::execution::planner::{
convert_spark_types_to_arrow_schema, literal_to_array_ref,
bind_residual, convert_spark_types_to_arrow_schema, literal_to_array_ref,
parse_file_scan_tasks_from_common,
};
use crate::execution::shuffle::CometPartitioning;
Expand All @@ -3199,6 +3203,26 @@ mod tests {
};
use datafusion_comet_spark_expr::EvalMode;

#[test]
fn iceberg_residual_bind_failure_skips_pushdown() {
use iceberg::expr::Reference;
use iceberg::spec::{Datum, NestedField, PrimitiveType, Schema as IcebergSchema, Type};

let schema = Arc::new(
IcebergSchema::builder()
.with_fields(vec![Arc::new(NestedField::optional(
1,
"id",
Type::Primitive(PrimitiveType::Int),
))])
.build()
.unwrap(),
);
let id = Reference::new("id");
assert!(bind_residual(!id.clone().equal_to(Datum::int(1)), Arc::clone(&schema)).is_some());
assert!(bind_residual(id.equal_to(Datum::bool(true)), schema).is_none());
}

#[test]
fn scan_default_rejects_struct_expressions() {
let planner = PhysicalPlanner::new(Arc::new(SessionContext::new()), 0);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 = fieldTypeOf(field).toString
val schemasById = getMethod(table.getClass, "schemas")
.invoke(table)
.asInstanceOf[java.util.Map[_, _]]
.asScala
val history = schemasById.values.toSeq
val ids = schemasById.keys.map(_.asInstanceOf[Integer].intValue).toSet
val historyComplete = ids.nonEmpty && ids.min == 0 && ids.max == ids.size - 1
getMethod(schema.getClass, "columns")
.invoke(schema)
.asInstanceOf[java.util.List[_]]
.asScala
.filter { column =>
val id = fieldIdOf(column).asInstanceOf[AnyRef]
val current = typeOf(column)
val promotable =
current == "long" || current == "double" || current.startsWith("decimal(")
(promotable && !historyComplete) ||
history.exists(older => Option(findField.invoke(older, id)).exists(typeOf(_) != current))
}
.map(fieldName)
.toSet
}

/**
* Validates partition column types for compatibility with iceberg-rust.
*
Expand Down
Loading