Join operator lost data due to decimal join key.

[beliefer]

Backend

VL (Velox)

Bug description

There are two tables: A and B.

// The main schema of A
CREATE TABLE A (
  invoiceid DECIMAL(20,0),
  time TIMESTAMP,
  update_time TIMESTAMP,
  dt STRING,
  hour STRING)
USING orc
PARTITIONED BY (dt, hour)
LOCATION 'hdfs://test/A'
TBLPROPERTIES (
  'transient_lastDdlTime'='1718612895')

// The main schema of B.
CREATE TABLE B (
  `invoiceid` DECIMAL(20,0),
  `status` STRING,
  `update_time` TIMESTAMP,
  `dt` STRING,
  `hour` STRING)
USING orc
PARTITIONED BY (dt, hour)
LOCATION 'hdfs://test/B'
TBLPROPERTIES (
  'transient_lastDdlTime'='1718612913')

// The orc file schema in hdfs://test/A

Type: struct<invoiceid:decimal(38,18), time:timestamp, update_time:timestamp>

// The orc file schema in hdfs://test/B

Type: struct<invoiceid:decimal(38,18), status:string, update_time:timestamp>

// The query info.
SELECT count(*)
FROM (
   SELECT c.invoiceid AS invoice_id,
       c.orderid AS order_id,
       c.time AS order_time,
       c.update_time AS update_time
FROM
    (SELECT *
     FROM
         (SELECT *,
                 ROW_NUMBER() OVER (PARTITION BY invoiceid
                                    ORDER BY update_time DESC) AS rn
          FROM A
          WHERE dt >= '20260421') b
     WHERE rn = 1) c
JOIN
    (SELECT *
     FROM B
     WHERE dt >= '20260421'
         AND status = 7) d ON c.invoiceid = d.invoiceid
);

The root cause

Scan operator of A is Spark Scan hive due to the time and update_time are TIMESTAMP.
Scan operator of B is Gluten NativeScan hive.

Left side: Scan hive (vanilla) → DECIMAL (20,2) row → RowToVeloxColumnar → DECIMAL(20,2) columnar                                                                         
Right: NativeScan hive (Velox) → may be widen when reading ORC internally DECIMAL(38,2) columnar                                                                                   
                                                                                                                                                                      
→ Inconsistent join key types on both sides → Key matching failed in Velox hash join → 0 line output
                                                                                                                                                                      
The problem with RowToVeloxColumnarEXE itself:
When converting RowToVeloxColumnar, the Velox Vector is constructed strictly according to the DECIMAL (20,2) declared in Spark schema, without actively expanding the precision. When NativeScan reads Hive ORC, Velox will internally decode it
Type regularization, for example, DECIMAL (20,2) may be mapped to an internal precision of 128 bit representation DECIMAL (38,2) or Velox ShortDecimal/LongDecimal.
                                                                                                                                                                      
This results in a situation where the plan level type is consistent (both are DECIMAL (20,2)), but the type tags of the actual columnar vectors are inconsistent, causing the key comparison logic of Velox HashJoin to go to the type mismatch branch → join
Output 0 rows.

Gluten version

Gluten-1.5

Spark version

Spark-3.5.x

Spark configurations

No response

System information

No response

Relevant logs

[beliefer]

I will try to fix this issue.

[jinchengchenghh]

For DECIMAL (20,2),both RowToColumnar and scan ORC should output int128_t RowVector, if one side is not, we should investigate why it is not int128_t

[beliefer]

@jinchengchenghh The table schema(DECIMAL(20,0)) is not consistent with orc file schema(decimal(38,18)).

[beliefer]

Take for example.

set spark.gluten.sql.columnar.forceShuffledHashJoin=false;
set spark.gluten.sql.columnar.shuffledHashJoin=false;
SELECT /*+ MERGE(htmp1) */ htmp1.c2 AS 1c2, htmp1.c3 AS 1c3, htmp2.c2 AS 2c2 FROM htmp1 JOIN htmp2 ON htmp1.c1 = htmp2.c1；

Execution plan structure (confirmed)

  SortMergeJoinExecTransformer  →  Output 0 rows, 0.0 B                                       
  :- SortExecTransformer (left)  →  100 rows, 2.3 KiB                                                                                                                   
  :  +- InputIteratorTransformer                                                                                                                                      
  :     +- RowToVeloxColumnar          ← htmp1 Follow the vanilla path                                                                                                        
  :        +- Scan hive htmp1          ← 100 rows                                                                                                                       
  +- SortExecTransformer (right) →  100 rows, 4.2 KiB                                                                                                                   
     +- InputIteratorTransformer                                                                                                                                      
        +- NativeScan hive htmp2       ← htmp2 takes the Velox native path, 100 rows

100 rows on each side are sent to SortExec, 100 rows on each side are sent to SMJ, and SMJ outputs 0 rows. The data is fine, the problem lies in comparing the join keys.

---

Root cause: Different internal representations of DECIMAL generated by two read paths

It's not an issue with RowToVeloxColumnar itself, it's just a format bridge. The real root cause lies in the fact that two paths generate semantically equivalent but internally encoded DECIMAL values.

Left side (Scan hive → RowToVeloxColumnar):

1. Hive Metastore record c1 column type is DECIMAL(20, 0)
2. However, the type written in the footer of the ORC file is decimal (38, 18) (which is the schema inconsistency you observed)
3. Scan hive (vanilla Spark ORC reader) to read using Metastore's type DECIMAL (20,0), and the values read are 1, 2, 3 .. (integer)
4. RowToVeloxColumnar converts Spark's DECIMAL (20,0) to Velox's int128 with an accuracy of 20 scale 0， Internal value example: 1 → 0x00000001

Right side (NativeScan hive):

1. Velox's ORC reader directly reads the type in the footer of the ORC file, which is decimal (38, 18)
2. Under the encoding of scale=18, the internal value of int128 is 1 × 10 ^ 18=10000000000000000
3. Namely: 1 → 0 x0DE0B6B3A7640000

SMJ performs key comparison in the Velox layer:

Left key: int128=1 (scale=0)
Right key: int128=10 ^ 18 (scale=18)
Comparison result: Unequal → No matching rows

Therefore, no match can be found in the Cartesian space of 100 × 100, and 0 rows are output.

---

RowToVeloxColumnar's role
RowToVeloxColumnar is not the root cause, but the key diagnostic clue is:
-Its appearance indicates that the htmp1 side did not go through NativeScan, but rather went through it vanilla Scan hive
-This is precisely the manifestation of asymmetry: different ORC readers are used on both sides, and different readers interpret the type of the DECIMAL column of the same ORC file differently (one using Metastore type, the other using file footer type)
-If both sides are NativeScan (or both are Scan hive+RowToVeloxColumnar), the internal representation of DECIMAL will be consistent, and join will match correctly

It can be understood that RowToVeloxColumnar is the symptom, and the asymmetric scan path is the cause.

[jinchengchenghh]

Do you know how JVM Spark handle this case? I assume after table scan, we should get the decimal(20, 0) instead of decimal(38, 18)

the type written in the footer of the ORC file is decimal (38, 18)

And even if the table scan outputs decimal(38, 18), the value is still equal, looks like the comparison between different scale has some problems in sort merge join? The function decimal_equalto can compare the decimal value with different scale

Left key: int128=1 (scale=0)
Right key: int128=10 ^ 18 (scale=18)
Comparison result: Unequal → No matching rows

[jinchengchenghh]

Do you know the ORC table scan output data type? Each operator should output corresponding data type

case class FileSourceScanExecTransformer(
    @transient override val relation: HadoopFsRelation,
    @transient stream: Option[SparkDataStream],
    override val output: Seq[Attribute],
    override val requiredSchema: StructType,
    override val partitionFilters: Seq[Expression],
    override val optionalBucketSet: Option[BitSet],
    override val optionalNumCoalescedBuckets: Option[Int],
    override val dataFilters: Seq[Expression],
    override val tableIdentifier: Option[TableIdentifier],
    override val disableBucketedScan: Boolean = false,
    override val pushDownFilters: Option[Seq[Expression]] = None)

[beliefer]

Do you know how JVM Spark handle this case? I assume after table scan, we should get the decimal(20, 0) instead of decimal(38, 18)

the type written in the footer of the ORC file is decimal (38, 18)

You can think writing orc data before creating hive table.

[beliefer]

@jinchengchenghh You can reproduce this issue with below SQL.

CREATE TABLE htmp1_wide (c1 DECIMAL(38, 18), c2 INT, c3 TIMESTAMP) STORED AS ORC;
INSERT INTO htmp1_wide
SELECT cast(id AS decimal(38,18)),
       id % 3,
       cast(id % 9 AS TIMESTAMP)
FROM range(1, 101);
CREATE TABLE htmp1 (c1 DECIMAL(20, 0), c2 INT, c3 TIMESTAMP) STORED AS ORC LOCATION 'hdfs://.../htmp1_wide';

CREATE TABLE htmp2_wide (c1 DECIMAL(38, 18), c2 INT, c3 TIMESTAMP) STORED AS ORC;
INSERT INTO htmp2_wide
SELECT cast(id AS decimal(38,18)),
       id % 3,
       cast(id % 5 AS TIMESTAMP)
FROM range(1, 101);
CREATE TABLE htmp2 (c1 DECIMAL(20, 0), c2 INT, c3 TIMESTAMP) STORED AS ORC LOCATION 'hdfs://.../htmp2_wide';

set spark.gluten.sql.columnar.forceShuffledHashJoin=false;
set spark.gluten.sql.columnar.shuffledHashJoin=false;
SELECT /*+ MERGE(htmp1) */ htmp1.c2 AS 1c2, htmp1.c3 AS 1c3, htmp2.c2 AS 2c2 FROM htmp1 JOIN htmp2 ON htmp1.c1 = htmp2.c1;

[jinchengchenghh]

#11980 (comment) Do you know the decimal comparison logic in sort merge join? @JkSelf

[jinchengchenghh]

This is because you create a table with decimal(20, 0) but the source orc file schema is decimal(38, 18), Velox outputs the incorrect case in this corner case. Usually, the source table data type should align with the file data type, we support read data different with other data type in parquet reader, but we don't have similar logic in orc. You may need a feature like this one https://github.com/facebookincubator/velox/pull/15343/changes

[JkSelf]

#11980 (comment) Do you know the decimal comparison logic in sort merge join? @JkSelf

Same with basic type comparation and no extra logic for Decimal type. https://github.com/facebookincubator/velox/blob/f5a10e84573c910e5f0393c3febddeaa7b347a4e/velox/exec/MergeJoin.cpp#L238-L276