[iceberg] Incorrect results when reading decimal fields

[andygrove]

Describe the bug

During fuzz testing, I am seeing errors reading decimals (or maybe when sorting decimals).

SQL

SELECT c7, cast(c7 as SMALLINT), try_cast(c7 as SMALLINT) FROM local.db.test0 ORDER BY c7;

Note that c7 is defined as Decimal128(10, 2).

Spark Plan

AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(2) Sort [c7#6669 ASC NULLS FIRST], true, 0
   +- AQEShuffleRead coalesced
      +- ShuffleQueryStage 0
         +- Exchange rangepartitioning(c7#6669 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, [plan_id=8959]
            +- *(1) Project [c7#6669, cast(c7#6669 as smallint) AS c7#6675, try_cast(c7#6669 as smallint) AS c7#6676]
               +- *(1) ColumnarToRow
                  +- BatchScan local.db.test0[c7#6669] local.db.test0 (branch=null) [filters=, groupedBy=] RuntimeFilters: []
+- == Initial Plan ==
   Sort [c7#6669 ASC NULLS FIRST], true, 0
   +- Exchange rangepartitioning(c7#6669 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, [plan_id=8946]
      +- Project [c7#6669, cast(c7#6669 as smallint) AS c7#6675, try_cast(c7#6669 as smallint) AS c7#6676]
         +- BatchScan local.db.test0[c7#6669] local.db.test0 (branch=null) [filters=, groupedBy=] RuntimeFilters: []

Comet Plan

AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   *(1) CometColumnarToRow
   +- CometSort [c7#6697, c7#6703, c7#6704], [c7#6697 ASC NULLS FIRST]
      +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 0
            +- CometColumnarExchange rangepartitioning(c7#6697 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=9021]
               +- CometProject [c7#6697, c7#6703, c7#6704], [c7#6697, cast(c7#6697 as smallint) AS c7#6703, try_cast(c7#6697 as smallint) AS c7#6704]
                  +- CometBatchScan local.db.test0[c7#6697] local.db.test0 (branch=null) [filters=, groupedBy=] RuntimeFilters: []
+- == Initial Plan ==
   CometSort [c7#6697, c7#6703, c7#6704], [c7#6697 ASC NULLS FIRST]
   +- CometColumnarExchange rangepartitioning(c7#6697 ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometColumnarShuffle, [plan_id=9008]
      +- CometProject [c7#6697, c7#6703, c7#6704], [c7#6697, cast(c7#6697 as smallint) AS c7#6703, try_cast(c7#6697 as smallint) AS c7#6704]
         +- CometBatchScan local.db.test0[c7#6697] local.db.test0 (branch=null) [filters=, groupedBy=] RuntimeFilters: []

First difference at row 1:
Spark: 0.01,0,0
Comet: 0.00,0,0

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response

[andygrove]

Another test failed and confirms that the issue is with reading decimals.

The value for c7 here is 0.98 in Spark but 0.65 in Comet.

First difference at row 2:
Spark: `false,-51,-20818,-2147483648,-32768,0.75996923,0.847645392651163,0.98,0.292978451873507573,2024-06-13,2024-05-06 06:52:54.765,䨕罅齺௚쉨햇⚤䇃,[49,50,51],false,109,-16556,-2050503136,8525030663349370862,0.27201045,NULL,0.05,0.212145654451256882,2024-06-16,2024-05-02 00:12:05.49,123,[55,49,57,51,54,55,56,53,48,50,48,48,50,53,48,55,48,50,55]`
Comet: `false,-51,-20818,-2147483648,-32768,0.75996923,0.847645392651163,0.65,0.292978451873507573,2024-06-13,2024-05-06 06:52:54.765,䨕罅齺௚쉨햇⚤䇃,[49,50,51],false,109,-16556,-2050503136,8525030663349370862,0.27201045,NULL,0.48,0.212145654451256882,2024-06-16,2024-05-02 00:12:05.49,123,[55,49,57,51,54,55,56,53,48,50,48,48,50,53,48,55,48,50,55]`

[andygrove]

Fixed in #1985