Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions docs/source/user-guide/latest/iceberg.md
Original file line number Diff line number Diff line change
Expand Up @@ -122,8 +122,22 @@ 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 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.

**Partitioning:**

Expand Down Expand Up @@ -801,6 +815,14 @@ Which entries apply depends on which native paths are enabled:
- **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such
a file as NULL, and the native scan returns the values stored in the file
([#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. When only part
of a condition is pushed, it also avoids this for promotions made before Iceberg 0.12 and for
tables with a name mapping, which those procedures set. Other cases can still drop rows
([#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 @@ -2572,21 +2572,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 @@ -2953,6 +2939,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 @@ -3071,7 +3075,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 @@ -3092,6 +3096,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 = getMethod(field.getClass, "type").invoke(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
Loading