Comet produces bloated results in comparison with Spark

[sandugood]

Describe the bug

Was testing out Comet in a multi-stage ETL process (contains multiple joins and native Iceberg scans). Some of the steps look like a usual Comet execution:

CometSinkPlaceHolder
+- CometColumnarExchange
   +- Project
      +- SortMergeJoin
         :- CometSort
         :  +- CometSinkPlaceHolder
         :     +- CometColumnarExchange
         :        +- Project
         :           +- SortMergeJoin
         :              :- CometSort
         :              :  +- CometSinkPlaceHolder
         :              :     +- CometColumnarExchange
         :              :        +- Project
         :              :           +- SortMergeJoin
         :              :              :- CometSort
         :              :              :  +- CometSinkPlaceHolder
         :              :              :     +- CometColumnarExchange
         :              :              :        +- Project
         :              :              :           +- SortMergeJoin
         :              :              :              :- CometSort
         :              :              :              :  +- CometSinkPlaceHolder
         :              :              :              :     +- CometColumnarExchange
         :              :              :              :        +- Project
         :              :              :              :           +- SortMergeJoin
         :              :              :              :              :- Sort
         :              :              :              :              :  +-  HashAggregate [COMET: Unsupported data type: TimestampNTZType, Unsupported aggregate expression(s)]
         :              :              :              :              :     +- CometSinkPlaceHolder
         :              :              :              :              :        +- CometColumnarExchange
         :              :              :              :              :           +- HashAggregate
         :              :              :              :              :              +-  InMemoryTableScan [COMET: InMemoryTableScan is not supported]
         :              :              :              :              :                    +- InMemoryRelation
         :              :              :              :              :                          +- CometExchange
         :              :              :              :              :                             +- CometProject
         :              :              :              :              :                                +- CometSortMergeJoin
         :              :              :              :              :                                   :- CometSort
         :              :              :              :              :                                   :  +- CometExchange
         :              :              :              :              :                                   :     +- CometProject
         :              :              :              :              :                                   :        +- CometFilter
         :              :              :              :              :                                   :           +- CometIcebergNativeScan
         :              :              :              :              :                                   +- CometSort
         :              :              :              :              :                                      +- CometFilter
         :              :              :              :              :                                         +- CometHashAggregate
         :              :              :              :              :                                            +- CometExchange
         :              :              :              :              :                                               +- CometHashAggregate
         :              :              :              :              :                                                  +- CometUnion
         :              :              :              :              :                                                     :- CometProject
         :              :              :              :              :                                                     :  +- CometFilter
         :              :              :              :              :                                                     :     +- CometIcebergNativeScan
         :              :              :              :              :                                                     +- CometIcebergNativeScan

However when aggregating the final result (to test Comet vs Spark sanity) - getting bloated values for sum aggregation. We can even see that without the aggregation, just comparing single values for users (domain-specific) across both resulting tables.
Note:

• using "spark.comet.scan.icebergNative.enabled": "true". So native Iceberg scan is enabled. And using it from both iceberg-rust and iceberg-storage-opendal repo main branch. Why? There was a fix regarding reading .parquet files that didn't contain page index. Before that - whole query would faild. Now it works, but produces inconsistent results. Check fix(reader): graceful handling of missing column index iceberg-rust#2693
• bloated values are consistent and deterministic. So it isn't related to spark.comet.exec.strictFloatingPoint being set to true or false. Wrong results are being consistent.

When disabling native Iceberg scan - query runs as intended, results are being stored in an iceberg table and results are consistent across vanilla Spark and Comet. So the problem itself lies somewhere in the iceberg-rust integration in Comet.

Steps to reproduce

No response

Expected behavior

Same (or almost the same down to floating point precision) values for both Spark and Comet

Additional context

No response

[comphead]

Thanks @sandugood for reporting this. Please clarify on bloated results? Is it a correctness issue? output row count doesn't match or aggregated values inflated?

[sandugood]

Purely a correctness issue. This ETL produces a feature-store-like table. Row count matches perfectly, though values are sometimes identical, sometimes almost identical and sometimes different.
This is internal data, so can't share the values exactly, but view it as a per-user-feature-values.

Using aggregation after the table formation just to test the sanity.

[sandugood]

By the way, a follow-up to the presented physical plan where InMemoryTableScan [COMET: InMemoryTableScan is not supported] was stated.
Later on, we tested full pipeline again, disabling all of the .cache() invocations (we are using PySpark) and still got the same bloated results.
So, I would confirm that this is mostly Iceberg issue, however I'm down to all sorts of thoughts and suggestions.

[mbutrovich]

What are the data types for the incorrect sums? Are they floating point? I understand you can't share internal data, but anything we can do to reproduce would be helpful (schema, rough query, etc.). Thanks for reporting!

[sandugood]

Returning with additional info:

1. When tried to debug and rerun the pipeline, once got: WARN CometIcebergNativeScan: Failed to serialize delete file: null
2. Additionally, when performing a single .count() over data - Comet's row count is higher, than vanilla Spark.

For context. Using spark-4.0.3 version (official image) + additional .jar files inside the image (all of them pulled from maven-central proxy):

• compiled Comet (from main branch, with iceberg-rust crate pulled also from main branch)
• iceberg-spark-runtime-4.0_2.13-1.11.0.jar
• iceberg-aws-bundle-1.11.0.jar

In-comparison - vanilla Spark has all of the .jar files listed, besides the Comet one.

So, right now, I would say that these wrong values comes purely from reading with native Iceberg scan on. When I did enable Comet, but didn't enable native scan - it was the same as vanilla Spark.

[mbutrovich]

Returning with additional info:

• iceberg-spark-runtime-4.0_2.13-1.11.0.jar
• iceberg-aws-bundle-1.11.0.jar

Thanks for the added details! Are you able to use Iceberg 1.10 to reproduce the issue? Comet does not support Iceberg 1.11 yet.

[sandugood]

Yes, just now reproduced it with iceberg-spark-runtime-4.0_2.13-1.10.1.jar and iceberg-aws-bundle-1.10.1.jar inside the container

[sandugood]

Additional context - screenshots from Spark WEB-UI. The experiment was quite simple. Applying .count() on the same table with both Spark and Spark+Comet
Spark's regular BatchScan

Comet native Iceberg scan:

As we can see - after Comet scans the table we get almost 42% more rows. And there is a problem with stats gathering, although couldn't make it work. Seems like it can't scan all of the metadata.

[mbutrovich]

Trying to get all the variables here: you have delete files, COW or MOR, equality or row-level deletes, and table format is V2? As much information as you can send our way. I basically need to reproduce this in a standalone test.

[comphead]

@sandugood wondering if this issue also can be reproduced on plain parquet files?