mbutrovich commented on code in PR #4752:
URL: https://github.com/apache/datafusion-comet/pull/4752#discussion_r3686517366
##########
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:
This PR flips the FileIO compatibility check from a deny-list (block only
`InMemoryFileIO`, allow everything else) to an allow-list of seven exact class
names plus an `EncryptingFileIO` prefix match:
```scala
private val CompatibleFileIOClasses: Set[String] = Set(
"org.apache.iceberg.hadoop.HadoopFileIO",
"org.apache.iceberg.aws.s3.S3FileIO",
"org.apache.iceberg.gcp.gcs.GCSFileIO",
"org.apache.iceberg.io.ResolvingFileIO",
"org.apache.iceberg.spark.SparkFileIO",
"org.apache.iceberg.azure.adlsv2.ADLSFileIO",
"org.apache.iceberg.CachingFileIO")
```
Failing closed on an unrecognized FileIO is the right default here: Comet's
native reader bypasses Iceberg Java's `FileIO` entirely and goes straight to
`object_store`, so a deny-list would silently accelerate any new FileIO a
future Iceberg release ships, with no way to know whether its I/O behavior
(credential vending, request signing, etc.) is actually replicated on the
native side. That's the same fail-closed pattern already used for
`formatVersionSupported` (falls back on `v > 3` rather than assuming forward
compatibility), so the allow-list approach itself fits the codebase.
The actual gap is the matching mechanism: `isCompatibleFileIO` matches on
exact class name (`CompatibleFileIOClasses.contains(className)`), not on the
class hierarchy. A FileIO that subclasses one of these seven with no change to
actual I/O behavior (a common pattern for adding metrics, retry logic, or
multi-tenant credential routing on top of `S3FileIO`/`HadoopFileIO`) will no
longer match, and will now silently fall back to Spark where it previously ran
natively.
Matching should walk the already-loaded `fileIO.getClass` ancestry rather
than loading each candidate name via `IcebergReflection.loadClass`.
`ClassLoaders.loadClass`
(`spark/src/main/java/org/apache/comet/util/ClassLoaders.java:32`) declares
`throws ClassNotFoundException` uncaught, and `isCompatibleFileIO` is only
reached from `fileIOCompatible` at plan time (`CometScanRule.scala:480`), where
the codebase's established invariant is that a reflection failure becomes a
fallback reason, never an uncaught exception (see every other check in
`transformV2Scan`, e.g. `defaultValuesSupported`, `schemaTypesSupported`,
`taskValidation`). A deployment that only bundles `iceberg-aws` (no
`iceberg-gcp`/`iceberg-azure` on the classpath) would hit
`ClassNotFoundException` trying to load `GCSFileIO`/`ADLSFileIO` just to test
`isInstance`, crashing planning instead of falling back — likely why the PR
used a string-name comparison against the live instance in the first place.
`IcebergReflection.scala` already has both pieces of idiom this needs, just
not combined: `findMethodInHierarchy` (`IcebergReflection.scala:112-126`) walks
a `Class[_]`'s superclass chain with a plain `while` loop, and
`isIcebergScanClass`/`ICEBERG_SCAN_CLASSES.contains(name)` matches a live
object's `getClass.getName` against a `Set[String]`, the same pattern this PR's
`isCompatibleFileIO` already uses at the leaf class only. Combining them (and
placing the result in `IcebergReflection`, alongside the other "is this class
one of these known names" helpers, rather than in `CometScanRule`'s companion
object) avoids ever loading a class that might not exist in this JVM:
```scala
// in IcebergReflection
def classNameInHierarchy(clazz: Class[_], names: Set[String]): Boolean = {
var current: Class[_] = clazz
while (current != null) {
if (names.contains(current.getName)) return true
current = current.getSuperclass
}
false
}
```
```scala
// in CometScanRule
def isCompatibleFileIO(fileIO: AnyRef): Boolean =
IcebergReflection.classNameInHierarchy(fileIO.getClass,
CompatibleFileIOClasses) ||
fileIO.getClass.getName.startsWith(EncryptingFileIOPrefix)
```
This also means the call site at `CometScanRule.scala:480` needs to change
from passing `fileIO.getClass.getName` to passing `fileIO` itself, since
matching subclasses requires the live `Class[_]`, not just its name.
##########
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:
This builds the unified `_partition` type by scanning `partition_type_pool`
in pool-insertion order and keeping the first-seen field for each field id.
Pool-insertion order is whatever order the Scala side happens to encounter
distinct per-spec partition types while iterating `DataSourceRDD` partitions
(`CometIcebergNativeScan.scala`, `serializePartitionData`), not a defined order.
Compare with iceberg-rust's own `compute_unified_partition_type`
(`crates/iceberg/src/partitioning.rs` in iceberg-rust, pinned rev `3d84c81`),
which this code's own comment says it's meant to match: it sorts specs by
`spec_id` descending before merging, and explicitly sorts the final field id
list ascending before building the struct. Comet's version does neither.
For a simple case (fields only ever added, never dropped and re-added),
ascending field ids happen to fall out of the pool naturally, since each spec's
own field list is already ascending and ids only increase across specs. But the
"re-add dropped partition field" scenario breaks that: a field dropped in one
spec (voided, so it's filtered out of that spec's serialized partition type
JSON, see `CometIcebergNativeScan.scala` around the `fieldsJson` construction)
and re-added in a later spec gets a new, higher field id. If the pool happens
to serialize that later spec's JSON before the id of the original,
lower-numbered field is (re-)encountered from an earlier spec's JSON,
`struct_fields` ends up with the higher id before the lower one, i.e. not
ascending.
`CometIcebergNativeSuite.scala` has a test for exactly this shape
("partition evolution - re-add dropped partition field"), but it only calls
`checkIcebergNativeScan` without asserting on struct field values directly, so
I can't tell from reading the diff alone whether the ordering actually diverges
in that test's specific data, or whether it happens to work out. Given
`build_partition_constant` in iceberg-rust (`record_batch_transformer.rs`)
builds the Arrow struct by iterating `unified_partition_type.fields()` in
whatever order it's given, an order mismatch against the schema Spark's
analyzer fixed for `_partition` would show up as wrong values in the wrong
named fields, not a crash.
Could you confirm whether the field order this computation produces is
actually guaranteed to be ascending by field id in all cases, or add the same
sort iceberg-rust's `compute_unified_partition_type` does?
##########
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:
This is new fallback logic for the case that motivated it: a V1 table where
two specs bind the same partition field id to incompatible source/transform
pairs. I didn't find a test that constructs such a table and confirms Comet
actually falls back (with the right reason) instead of trying the native path.
Since this is the one case this whole check exists to catch, it seems worth a
dedicated test, even if constructing the conflicting-spec table requires going
through Iceberg's Java API directly (as the other partition-evolution tests in
this PR already do via `updateSpec()`).
##########
spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql:
##########
@@ -0,0 +1,850 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Native Iceberg scan is unsupported on Spark 4.2: no Iceberg spark-runtime
is published for
+-- 4.2 yet, so the build reuses the 4.0 runtime, which is binary-incompatible
with 4.2 (e.g.
+-- connector.catalog.View became a class). Mirrors the assume(!isSpark42Plus)
guard in
+-- CometIcebergNativeSuite. See
https://github.com/apache/datafusion-comet/issues/4969.
+-- MaxSparkVersion: 4.1
+
+-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog
+-- Config: spark.sql.catalog.test_cat.type=hadoop
+-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test
+-- Config: spark.comet.enabled=true
+-- Config: spark.comet.exec.enabled=true
+-- Config: spark.comet.iceberg.native.enabled=true
+
+-- All `query` assertions below implicitly validate that Comet's partition type
+-- computation (from partition_type_pool JSON) matches Spark's (from Iceberg
Java's
+-- PartitionSpec.partitionType()). If the two diverged,
checkSparkAnswerAndOperator
+-- would detect differing struct values. This covers the "dual computation
path"
+-- cross-validation concern.
+
+-- =============================================================
+-- Setup: Create table with bucket partitioning (format v2 for MoR)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.meta_test
+
+statement
+CREATE TABLE test_cat.db.meta_test (id INT, name STRING, value DOUBLE) USING
iceberg PARTITIONED BY (bucket(4, id)) TBLPROPERTIES ('format-version' = '2')
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (1, 'alice', 10.0), (2, 'bob', 20.0),
(3, 'charlie', 30.0)
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (4, 'dave', 40.0), (5, 'eve', 50.0)
+
+-- =============================================================
+-- _file: basic projection (native)
+-- =============================================================
+
+query
+SELECT id, _file FROM test_cat.db.meta_test ORDER BY id
+
+-- Multiple inserts produce multiple files
+query
+SELECT COUNT(DISTINCT _file) > 1 FROM test_cat.db.meta_test
+
+-- _file in a WHERE clause exercises filter pushdown interaction
+query
+SELECT id, name FROM test_cat.db.meta_test WHERE _file LIKE '%.parquet' ORDER
BY id
+
+-- =============================================================
+-- _partition: basic struct projection (native)
+-- =============================================================
+
+query
+SELECT id, _partition FROM test_cat.db.meta_test ORDER BY id
+
+-- Access individual partition fields within the struct
+query
+SELECT id, _partition.id_bucket FROM test_cat.db.meta_test ORDER BY id
+
+-- =============================================================
+-- _spec_id: partition spec ID (scalar per file)
+-- =============================================================
+
+query
+SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id
+
+-- All rows should have spec_id = 0 (initial spec)
+query
+SELECT DISTINCT _spec_id FROM test_cat.db.meta_test
+
+-- =============================================================
+-- _pos: row position within each file (0-based)
+-- =============================================================
+
+-- Basic: positions should be 0-based within each file
+query
+SELECT id, _file, _pos FROM test_cat.db.meta_test ORDER BY _file, _pos
+
+-- Positions should be contiguous 0..N-1 per file
+query
+SELECT _file, MIN(_pos) as min_pos, MAX(_pos) as max_pos, COUNT(*) as cnt FROM
test_cat.db.meta_test GROUP BY _file
+
+-- =============================================================
+-- Partition evolution requires IcebergSparkSessionExtensions
+-- (ALTER TABLE ADD PARTITION FIELD / CALL system.add_partition_field).
+-- The SQL file test framework cannot register session extensions via
+-- Config directives. Partition evolution is covered in
+-- CometIcebergNativeSuite which has full session control.
+-- =============================================================
+
+-- =============================================================
+-- Combined: all metadata columns together (native -- all supported)
+-- =============================================================
+
+query
+SELECT id, _file, _pos, _spec_id, _partition FROM test_cat.db.meta_test ORDER
BY _file, _pos
+
+-- Row-level operation simulation: what CoW/MoR would project
+query
+SELECT _file, _pos, _spec_id, _partition, id, name FROM test_cat.db.meta_test
WHERE id > 3 ORDER BY _file, _pos
+
+-- =============================================================
+-- _file + _partition combined (native -- both supported)
+-- =============================================================
+
+query
+SELECT id, _file, _partition FROM test_cat.db.meta_test ORDER BY id
+
+statement
+DROP TABLE test_cat.db.meta_test
+
+-- =============================================================
+-- Identity partition transform
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.identity_part
+
+statement
+CREATE TABLE test_cat.db.identity_part (id INT, name STRING, dept STRING)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.identity_part VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+query
+SELECT id, _partition FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _partition.dept FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _file, _partition FROM test_cat.db.identity_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.identity_part
+
+-- =============================================================
+-- Multiple partition fields
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_part
+
+statement
+CREATE TABLE test_cat.db.multi_part (id INT, year INT, month INT, data STRING)
USING iceberg PARTITIONED BY (year, month)
+
+statement
+INSERT INTO test_cat.db.multi_part VALUES (1, 2023, 6, 'a'), (2, 2023, 7,
'b'), (3, 2024, 1, 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.multi_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.multi_part
+
+-- =============================================================
+-- Partition with filter
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.filter_part
+
+statement
+CREATE TABLE test_cat.db.filter_part (id INT, dept STRING, value DOUBLE) USING
iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.filter_part VALUES (1, 'eng', 10.5), (2, 'sales',
20.3), (3, 'eng', 30.7)
+
+query
+SELECT id, _partition FROM test_cat.db.filter_part WHERE dept = 'eng' ORDER BY
id
+
+statement
+DROP TABLE test_cat.db.filter_part
+
+-- =============================================================
+-- Edge case: Unpartitioned table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.unpart
+
+statement
+CREATE TABLE test_cat.db.unpart (id INT, val STRING) USING iceberg
+
+statement
+INSERT INTO test_cat.db.unpart VALUES (1, 'a'), (2, 'b')
+
+-- _partition on unpartitioned table is an empty struct (null), which Comet's
+-- shuffle doesn't support. Verify correctness only, not native operator
coverage.
+query spark_answer_only
+SELECT id, _partition FROM test_cat.db.unpart ORDER BY id
+
+-- Without ORDER BY (no shuffle), the native scan executes end-to-end
+query
+SELECT id, _partition FROM test_cat.db.unpart
+
+query
+SELECT id, _file FROM test_cat.db.unpart ORDER BY id
+
+-- _spec_id on unpartitioned: should be 0
+-- (spark_answer_only because _partition is an empty struct that Comet shuffle
rejects)
+query spark_answer_only
+SELECT id, _spec_id, _partition FROM test_cat.db.unpart ORDER BY id
+
+statement
+DROP TABLE test_cat.db.unpart
+
+-- =============================================================
+-- Edge case: Large batch (tests multi-batch _pos correctness)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.large_pos
+
+statement
+CREATE TABLE test_cat.db.large_pos (id INT) USING iceberg
+
+statement
+INSERT INTO test_cat.db.large_pos SELECT id FROM range(10000)
+
+query
+SELECT id, _pos FROM test_cat.db.large_pos ORDER BY id LIMIT 5
+
+statement
+DROP TABLE test_cat.db.large_pos
+
+-- =============================================================
+-- Edge case: Multiple row groups (forces _pos offset tracking)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_rg
+
+statement
+CREATE TABLE test_cat.db.multi_rg (id INT, data STRING) USING iceberg
TBLPROPERTIES ('write.parquet.row-group-size-bytes' = '1024')
+
+statement
+INSERT INTO test_cat.db.multi_rg SELECT id, REPEAT('x', 100) FROM range(1000)
+
+query
+SELECT _pos, id FROM test_cat.db.multi_rg WHERE _pos IN (0, 1, 998, 999) ORDER
BY _pos
+
+statement
+DROP TABLE test_cat.db.multi_rg
+
+-- =============================================================
+-- DML operations (DELETE, UPDATE, MERGE) internally project metadata columns
+-- (_file, _pos, _spec_id, _partition) to identify rows. All four are now
+-- supported natively. The subsequent SELECT queries verify that reads after
+-- DML return correct results through the native scan path (applying position
+-- deletes, seeing updated data).
+-- =============================================================
+
+-- Copy-on-write DELETE
+statement
+DROP TABLE IF EXISTS test_cat.db.cow_delete
+
+statement
+CREATE TABLE test_cat.db.cow_delete (id INT, name STRING, dept STRING) USING
iceberg PARTITIONED BY (dept) TBLPROPERTIES
('write.delete.mode'='copy-on-write')
+
+statement
+INSERT INTO test_cat.db.cow_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.cow_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.cow_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.cow_delete
+
+-- Merge-on-read DELETE (native read applies position deletes)
+statement
+DROP TABLE IF EXISTS test_cat.db.mor_delete
+
+statement
+CREATE TABLE test_cat.db.mor_delete (id INT, name STRING, dept STRING) USING
iceberg PARTITIONED BY (dept) TBLPROPERTIES
('write.delete.mode'='merge-on-read')
+
+statement
+INSERT INTO test_cat.db.mor_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.mor_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.mor_delete ORDER BY id
+
+-- After MoR delete, _pos still references original file positions
+query
+SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.mor_delete
+
+-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not
+-- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead.
Review Comment:
I searched `CometIcebergNativeSuite.scala` (including this PR's additions)
for any `UPDATE` statement and found none, with or without metadata columns.
There's no MoR UPDATE test anywhere in the suite, so this comment is inaccurate
and MoR UPDATE's interaction with `_file`/`_pos`/`_spec_id`/`_partition` has no
coverage at all right now. Could you either add that coverage (maybe gated on a
newer Spark version where `UPDATE` is supported, similar to other version-gated
tests in this suite) or fix the comment so it doesn't claim coverage that
doesn't exist?
##########
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:
The old test that verified an unsupported metadata column triggers fallback
(`"CometScanRule should report unsupported metadata columns"`) was replaced
with tests for the four now-supported columns, but I don't see a replacement
test that queries one of the still-unsupported Iceberg metadata columns (e.g.
`_deleted`, `_row_id`, `_change_type`) and checks it still falls back
correctly. Worth keeping a test for that path so a future change to
`MetadataFieldIds` or this filter doesn't silently regress it.
##########
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:
Two things here:
- Everywhere else in this file, spec/type JSON is produced through
`PartitionSpecParser.toJson()` (real spec) or the json4s DSL
(`compact(render(...))`, for the partition type). This is the one spot that
hand-interpolates a JSON string. It's safe today since `specId` is an `Int`,
but it's inconsistent with the rest of the file's approach, and any later
change that adds a string-typed field to this literal would need to remember to
escape it. Might be worth building it the same way (json4s DSL) as the
partition type JSON just above it.
- By the time this branch runs, the "real" spec for this task has already
been resolved, JSON-serialized via `PartitionSpecParser.toJson()`, and interned
into `partitionSpecToPoolIndex` earlier in the same method (the
`specIdx`/`taskBuilder.setPartitionSpecIdx(specIdx)` call above this block).
This code then unconditionally overwrites that with the empty-spec index. For
every value-less task, that's a reflection call plus a JSON serialization done
and then thrown away. It's probably not hot enough to matter (value-less tasks
only occur for partition-evolution edge cases), but if it's easy to restructure
so the empty-spec check happens before the real spec is computed, it'd avoid
the redundant work entirely.
--
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]
