This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 0761e549a4 perf: dedupe Iceberg residuals and delete files in native
scan serde (#4982)
0761e549a4 is described below
commit 0761e549a43fee618af462a305645e70d663a380
Author: Matt Butrovich <[email protected]>
AuthorDate: Thu Jul 23 13:42:05 2026 -0400
perf: dedupe Iceberg residuals and delete files in native scan serde (#4982)
---
native/core/src/execution/planner.rs | 476 ++++++++-----------
native/proto/src/proto/operator.proto | 103 ++++-
.../apache/comet/iceberg/IcebergReflection.scala | 36 ++
.../serde/operator/CometIcebergNativeScan.scala | 505 ++++++++++++++-------
.../org/apache/comet/CometFuzzIcebergBase.scala | 2 +-
.../org/apache/comet/CometFuzzIcebergSuite.scala | 206 +++++----
.../org/apache/comet/CometIcebergNativeSuite.scala | 496 ++++++++++++++++++++
7 files changed, 1252 insertions(+), 572 deletions(-)
diff --git a/native/core/src/execution/planner.rs
b/native/core/src/execution/planner.rs
index 7c24bcabba..e43daed9a5 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3547,59 +3547,32 @@ fn align_shuffle_writer_input(
fn partition_value_to_literal(
proto_value: &spark_operator::PartitionValue,
) -> Result<Option<iceberg::spec::Literal>, 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::DecimalVal(d)) => {
+
iceberg::spec::Literal::decimal(decimal_bytes_to_i128(&d.unscaled)?)
}
- Some(Value::BoolVal(v)) => iceberg::spec::Literal::bool(*v),
Some(Value::UuidVal(bytes)) => {
- // Deserialize UUID from 16 bytes
if bytes.len() != 16 {
return Err(GeneralError(format!(
"Invalid UUID bytes length: {} (expected 16)",
@@ -3614,7 +3587,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(),
));
}
};
@@ -3622,6 +3595,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<i128, ExecutionError> {
+ 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
@@ -3684,36 +3677,53 @@ 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<iceberg::scan::FileScanTaskDeleteFile> =
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())
+ },
+ key_metadata: None,
+ })
+ })
+ .collect::<Result<Vec<_>, ExecutionError>>()?;
+
let delete_files_cache: Vec<Vec<iceberg::scan::FileScanTaskDeleteFile>> =
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())
- },
- key_metadata: None,
+ .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::<Result<Vec<_>, ExecutionError>>()
@@ -3756,14 +3766,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}");
@@ -4093,246 +4104,129 @@ 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.
-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<iceberg::expr::Predicate> {
- 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
- }
- }
- 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
- }
- }
- 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),
+ use iceberg::expr::{Predicate, 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::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));
- match (left, right) {
- (Some(l), Some(r)) => Some(l.or(r)),
- _ => None, // OR requires both sides to be valid
+ 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::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 {
+ 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;
}
-
- if let Some(ref value) = in_expr.in_value {
- if let Some(column) = extract_column_reference(value) {
- let datums: Vec<iceberg::spec::Datum> = 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
- }
+ let column = Reference::new(s.column.clone());
+ let datums = s
+ .values
+ .iter()
+ .map(iceberg_literal_to_datum)
+ .collect::<Option<Vec<_>>>()?;
+ 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) => {
+ let left =
l.left.as_deref().and_then(iceberg_predicate_to_predicate);
+ let right =
l.right.as_deref().and_then(iceberg_predicate_to_predicate);
+ combine_logical(left, right, Predicate::and)
}
- _ => None, // Unsupported expression
+ Node::Or(o) => {
+ 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),
}
}
-fn convert_binary_to_predicate(
- left: &Option<Box<spark_expression::Expr>>,
- right: &Option<Box<spark_expression::Expr>>,
- op: iceberg::expr::PredicateOperator,
+/// 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<iceberg::expr::Predicate>,
+ right: Option<iceberg::expr::Predicate>,
+ combine: fn(iceberg::expr::Predicate, iceberg::expr::Predicate) ->
iceberg::expr::Predicate,
) -> Option<iceberg::expr::Predicate> {
- 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,
- ),
- ));
+ 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
+ }
}
-
- None
}
-fn extract_column_reference(expr: &spark_expression::Expr) -> Option<String> {
- 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<iceberg::spec::Datum> {
+ 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<iceberg::spec::Datum> {
- 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 d00979f0a0..439c9613a4 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -161,25 +161,43 @@ 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;
+ }
+}
+
+// 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;
+ 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 +205,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 +266,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;
@@ -216,6 +276,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 +299,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/iceberg/IcebergReflection.scala
b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 9f48ddfa9e..1a5cea7c5c 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,42 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * 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._
+ 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[") ||
+ typeStr == "binary") {
+ 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 879762818f..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
@@ -19,6 +19,11 @@
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, UUID}
+
import scala.collection.mutable
import scala.jdk.CollectionConverters._
@@ -32,12 +37,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 {
@@ -59,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
@@ -71,22 +76,43 @@ 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)
+
/**
- * 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 {
@@ -105,13 +131,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)
+ case bd: BigDecimal => bd
+ case _ => new 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)
@@ -158,20 +188,20 @@ 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()
}
-
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 {
@@ -179,9 +209,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 ?
@@ -201,24 +231,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).
*
@@ -462,7 +474,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, {
@@ -532,137 +544,216 @@ 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). 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 convertIcebergExpression(icebergExpr: Any, output: Seq[Attribute]):
Option[Expression] = {
+ def icebergExprToProto(
+ icebergExpr: Any,
+ output: Seq[Attribute],
+ pageIndexUnsupportedColumns: Set[String]):
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(
- exprClass,
- icebergExpr,
- 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
+ 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]
+
+ // 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
+ 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 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))
+ val left = icebergExprToProto(
+ exprClass.getMethod("left").invoke(icebergExpr),
+ output,
+ pageIndexUnsupportedColumns)
+ val right = icebergExprToProto(
+ exprClass.getMethod("right").invoke(icebergExpr),
+ output,
+ pageIndexUnsupportedColumns)
+ (left, right) match {
+ // 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 _ => 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,
+ 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))
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,
pageIndexUnsupportedColumns).map(notPredicate)
} else {
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
}
}
+ 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 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
+ .map(lit => (Base64.getEncoder.encodeToString(lit.toByteArray), lit))
+ .sortBy(_._1)
+ .foreach { case (_, lit) => setBuilder.addValues(lit) }
+
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 Spark types this serde does
not map to an
+ * 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 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])
+ // 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 =>
+
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())
}
/**
@@ -704,6 +795,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
@@ -727,9 +831,19 @@ 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]()
- val residualToPoolIndex = mutable.HashMap[Option[Expr], Int]()
+ 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]()
@@ -907,11 +1021,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
})
@@ -920,11 +1044,10 @@ 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,
+ pageIndexUnsupportedColumns)
} catch {
case e: Exception =>
logWarning(
@@ -933,11 +1056,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)
@@ -996,6 +1119,7 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
nameMappingToPoolIndex.size,
projectFieldIdsToPoolIndex.size,
partitionDataToPoolIndex.size,
+ deleteFileToPoolIndex.size,
deleteFilesToPoolIndex.size,
residualToPoolIndex.size)
@@ -1011,16 +1135,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/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 c6ac428938..e9bda4c6b4 100644
--- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala
@@ -21,13 +21,13 @@ 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.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
+import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.types._
import org.apache.comet.DataTypeSupport.isComplexType
-import org.apache.comet.testing.{DataGenOptions, FuzzDataGenerator,
ParquetGenerator, SchemaGenOptions}
+import org.apache.comet.serde.OperatorOuterClass
class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
@@ -118,108 +118,120 @@ 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)
}
}
- 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)
- }
+ // 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
+ }
- 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)
- }
+ 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
+ }
- 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")
- }
- }
- }
- }
- }
- }
+ 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})")
}
}
}
diff --git
a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 7b9bd6b3d6..bdad3163d4 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")
+ // 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(
+ "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)
+
+ assert(
+ duplicated.isEmpty,
+ "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")
@@ -2542,6 +2650,324 @@ 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 it.
+ 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")
+ }
+ }
+ }
+
+ // 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
+ // 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")
+ }
+ }
+ }
+
+ // 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") {
@@ -2980,6 +3406,76 @@ 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")
+ // 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(
+ "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,
+ "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") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
