parthchandra commented on code in PR #4752:
URL: https://github.com/apache/datafusion-comet/pull/4752#discussion_r3687466695
##########
native/core/src/execution/planner.rs:
##########
@@ -3835,6 +3835,41 @@ fn parse_file_scan_tasks_from_common(
})
.collect::<Result<Vec<_>, _>>()?;
+ // Compute unified partition type from the partition_type_pool.
+ // Each entry is a StructType JSON with the resolved field types for one
partition spec.
+ // Merge all specs into a single unified type (dedup by field_id).
+ //
+ // NOTE: The type information here comes from Iceberg Java's
PartitionSpec.partitionType()
+ // (via Scala reflection). This is the same type computation as
iceberg-rust's
+ // Transform::result_type() -- both implement the Iceberg spec's transform
result rules.
+ // When iceberg-rust's own scan planning is used (not Comet's proto path),
it computes
+ // this via compute_unified_partition_type(specs, schema) instead.
+ let unified_partition_type = {
+ let mut seen_field_ids = std::collections::HashSet::new();
+ let mut struct_fields: Vec<iceberg::spec::NestedFieldRef> = Vec::new();
+
+ for type_json in &proto_common.partition_type_pool {
+ match serde_json::from_str::<iceberg::spec::StructType>(type_json)
{
+ Ok(struct_type) => {
+ for field in struct_type.fields() {
+ if seen_field_ids.insert(field.id) {
+ struct_fields.push(Arc::clone(field));
+ }
+ }
+ }
+ Err(e) => {
+ return Err(ExecutionError::GeneralError(format!(
+ "Failed to deserialize partition type JSON from pool:
{e}"
+ )));
+ }
+ }
+ }
Review Comment:
You're right. This was wrong.
Fixed and this should match
`compute_unified_partition_type`/`Partitioning.buildPartitionProjectionType`.
To get `spec_id` on the native side, I made `partition_type_pool` index-aligned
with `partition_spec_pool` (documented in the .proto) and read
`partition_spec_cache[i].spec_id()`.
One caveat: because Comet ships resolved types (not the original
`Transform`), the void-vs-non-void type preference isn't replicated — but that
only affects a dropped-source-column field, which Comet already filters upstream
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -603,6 +626,25 @@ case class CometScanRule(session: SparkSession)
false
}
+ // The `_partition` metadata column projects the table's unified
partition type -- the
+ // merge of all historical specs. iceberg-rust computes that merge at
scan time and errors
+ // (DataInvalid) when specs bind one field id to incompatible
source/transform pairs (only
+ // possible in V1 tables, which do not keep partition field ids unique
across specs). That
+ // error would fail the native scan with no chance to fall back, so
when `_partition` is
+ // projected we run Iceberg Java's equivalent check here and fall back
if it cannot merge.
+ val unifiedPartitionTypeSupported =
+ if (scanExec.output.exists(a => a.isMetadataCol && a.name ==
"_partition")) {
+ IcebergReflection.validateUnifiedPartitionType(metadata.table)
match {
+ case Some(reason) =>
+ fallbackReasons += "Iceberg table has partition specs whose
unified partition " +
+ s"type cannot be computed for the _partition metadata
column: $reason"
+ false
+ case None => true
+ }
+ } else {
+ true
+ }
Review Comment:
Added a test. This couldn't be tested end to end via SQL because Spark's
analyzer catches it and throws a validation error. The test hand-builds the
conflicting V1 metadata.json (round-tripped through `TableMetadataParser` to
catch malformed edits) and asserts directly on
`IcebergReflection.validateUnifiedPartitionType`
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -361,9 +361,30 @@ case class CometScanRule(session: SparkSession)
return withFallbackReasons(scanExec, fallbackReasons.toSet)
}
+ // Check for unsupported metadata columns in Iceberg scans
+ val unsupportedMetadataCols =
scanExec.output.filter(_.isMetadataCol).filterNot { attr =>
+ CometIcebergNativeScan.MetadataFieldIds.keySet.contains(attr.name)
+ }
+ if (unsupportedMetadataCols.nonEmpty) {
+ fallbackReasons += "Unsupported Iceberg metadata columns: " +
+ unsupportedMetadataCols.map(_.name).mkString(", ")
+ return withFallbackReasons(scanExec, fallbackReasons.toSet)
+ }
Review Comment:
Test added.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -487,25 +501,48 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
}
}.toSeq
- // Only serialize partition data if we have non-unknown fields
- if (partitionValues.nonEmpty) {
- val partitionDataProto = OperatorOuterClass.PartitionData
- .newBuilder()
- .addAllValues(partitionValues.asJava)
- .build()
-
- // Deduplicate by protobuf bytes (use Base64 string as key)
- val partitionDataBytes = partitionDataProto.toByteArray
- val partitionDataKey =
Base64.getEncoder.encodeToString(partitionDataBytes)
-
- val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate(
- partitionDataKey, {
- val idx = partitionDataToPoolIndex.size
- commonBuilder.addPartitionDataPool(partitionDataProto)
+ // Native requires a task to carry both a partition spec and
partition data, or neither:
+ // iceberg-rust errors when the unified partition type has fields
but a task is missing
+ // its spec/data. A file written while a partition field was dropped
(partition
+ // evolution) has no partition values, so partitionValues is empty
here.
+ //
+ // For that value-less case we must NOT send the file's real spec:
it may retain a void
+ // field whose id collides with a unified field, which would make
iceberg-rust index the
+ // empty partition data out of range. Instead send an empty-fields
spec that keeps the
+ // real spec id (so _spec_id stays correct) plus empty data, so
native fills every
+ // unified _partition field with null -- matching Spark and the
pre-tightening behaviour.
+ if (partitionValues.isEmpty) {
+ val specId =
spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int]
+ val emptySpecJson = s"""{"spec-id":$specId,"fields":[]}"""
+ val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate(
+ emptySpecJson, {
+ val idx = partitionSpecToPoolIndex.size
+ commonBuilder.addPartitionSpecPool(emptySpecJson)
idx
})
- taskBuilder.setPartitionDataIdx(partitionDataIdx)
+ // Override the real spec registered above.
+ taskBuilder.setPartitionSpecIdx(emptySpecIdx)
}
Review Comment:
Fixed the JSON consistency: the empty-spec literal now uses the json4s DSL
like the rest of the file
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -451,15 +472,17 @@ case class CometScanRule(session: SparkSession)
}
// Now perform all validation using the pre-extracted metadata
- // Check if table uses a FileIO implementation compatible with
iceberg-rust
-
+ // Check if table uses a FileIO implementation compatible with
iceberg-rust.
+ // Comet's native reader uses object_store (Rust) for I/O, bypassing
Iceberg Java's
+ // FileIO entirely. Only allow known-compatible implementations whose
underlying
+ // storage object_store can reach via standard URL schemes.
val fileIOCompatible = IcebergReflection.getFileIO(metadata.table)
match {
- case Some(fileIO)
- if fileIO.getClass.getName ==
"org.apache.iceberg.inmemory.InMemoryFileIO" =>
- fallbackReasons += "InMemoryFileIO is not supported by Comet's
native reader"
- false
- case Some(_) =>
+ case Some(fileIO) if
CometScanRule.isCompatibleFileIO(fileIO.getClass.getName) =>
true
+ case Some(fileIO) =>
+ fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not
supported by " +
+ "Comet's native reader (object_store bypasses Iceberg Java
FileIO)"
+ false
case None =>
fallbackReasons += "Could not check FileIO compatibility"
false
Review Comment:
Right again. Fixed exactly as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
