[VL]Vector encoding type check failed in DecimalSumAggregate

[CLTFOREVER]

Describe the bug
java.lang.RuntimeException: Exception: VeloxRuntimeError
Error Source: RUNTIME
Error Code: INVALID_STATE
Reason: (FLAT vs. ROW)
Retriable: False
Expression: decodedPartial_.base()->encoding() == VectorEncoding::Simple::ROW
Function: addSingleGroupIntermediateResults
File: gluten/ep/build-velox/build/velox_ep/./velox/functions/sparksql/aggregates/DecimalSumAggregate.h
Line: 317

To Reproduce
Steps to reproduce the behavior:
select sum(xxx) from table; xxx's type is decimal.

[CLTFOREVER]

@zhouyuan please have a look.

[liujiayi771]

@CLTFOREVER Could you post your Spark UI Plan graph, and post the results of desc formatted table; ? explain your sql can also help.

[CLTFOREVER]

The physical plan context is:
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
VeloxColumnarToRowExec
+- ^(2) HashAggregateTransformer(keys=[], functions=[sum(t1#0)], output=[sum(t1)#2])
+- ShuffleQueryStage 0
+- ColumnarExchange SinglePartition, ENSURE_REQUIREMENTS, [id=#679], [id=#679], [OUTPUT] List(sum:DecimalType(30,4)), [OUTPUT] List(sum:DecimalType(30,4))
+- ^(1) HashAggregateTransformer(keys=[], functions=[partial_sum(t1#0)], output=[sum#6])
+- ^(1) NativeFileScan orc mart_hdp.orc_decimal_test[t1#0] Batched: true, DataFilters: [], Format: ORC, Location: PartitionFilters: [], PushedFilters: [], ReadSchema: structt1:decimal(20,4)
^(1) Substrait plan:
{
"extensions": [{
"extensionFunction": {
"name": "sum:req_dec\u003c20,4\u003e"
}
}],
"relations": [{
"root": {
"input": {
"project": {
"common": {
"emit": {
"outputMapping": [1, 2]
}
},
"input": {
"aggregate": {
"common": {
"direct": {
}
},
"input": {
"read": {
"common": {
"direct": {
}
},
"baseSchema": {
"names": ["t1"],
"struct": {
"types": [{
"decimal": {
"scale": 4,
"precision": 20,
"nullability": "NULLABILITY_NULLABLE"
}
}]
},
"partitionColumns": {
"columnType": ["NORMAL_COL"]
}
}
}
},
"groupings": [{
}],
"measures": [{
"measure": {
"phase": "AGGREGATION_PHASE_INITIAL_TO_INTERMEDIATE",
"outputType": {
"struct": {
"types": [{
"decimal": {
"scale": 4,
"precision": 30,
"nullability": "NULLABILITY_NULLABLE"
}
}, {
"bool": {
"nullability": "NULLABILITY_REQUIRED"
}
}],
"nullability": "NULLABILITY_REQUIRED"
}
},
"arguments": [{
"value": {
"selection": {
"directReference": {
"structField": {
}
}
}
}
}]
}
}]
}
},
"expressions": [{
"selection": {
"directReference": {
"structField": {
"child": {
"structField": {
}
}
}
}
}
}, {
"selection": {
"directReference": {
"structField": {
"child": {
"structField": {
"field": 1
}
}
}
}
}
}]
}
},
"names": ["sum#6"]
}
}]
}

[CLTFOREVER]

@liujiayi771 The table is very simple ,just create with sql:create table orc_decimal_test(t1 decimal(20,4)) stored as orc;
I just insert into three pieces of data:1.000,2.0000,3.0000;

[liujiayi771]

@CLTFOREVER I cannot reproduce. Is your code up to date?

[liujiayi771]

@CLTFOREVER According to your substrait plan, I find that something is missing in aggregate section.

{
  "extensions": [{
    "extensionFunction": {
      "functionAnchor": 1,
      "name": "sum:req_dec\u003c20,4\u003e"
    }
  }, {
    "extensionFunction": {
      "name": "row_constructor_with_null:dec\u003c30,4\u003e_bool"
    }
  }],
  "relations": [{
    "root": {
      "input": {
        "project": {
          "common": {
            "emit": {
              "outputMapping": [1]
            }
          },
          "input": {
            "aggregate": {
              "common": {
                "direct": {
                }
              },
              "input": {
                "project": {
                  "common": {
                    "emit": {
                      "outputMapping": [2]
                    }
                  },
                  "input": {
                    "read": {
                      "common": {
                        "direct": {
                        }
                      },
                      "baseSchema": {
                        "names": ["sum#7", "isempty#8"],
                        "struct": {
                          "types": [{
                            "decimal": {
                              "scale": 4,
                              "precision": 30,
                              "nullability": "NULLABILITY_NULLABLE"
                            }
                          }, {
                            "bool": {
                              "nullability": "NULLABILITY_REQUIRED"
                            }
                          }]
                        }
                      },
                      "localFiles": {
                        "items": [{
                          "uriFile": "iterator:0"
                        }]
                      }
                    }
                  },
                  "expressions": [{
                    "scalarFunction": {
                      "outputType": {
                        "struct": {
                          "types": [{
                            "decimal": {
                              "scale": 4,
                              "precision": 30,
                              "nullability": "NULLABILITY_NULLABLE"
                            }
                          }, {
                            "bool": {
                              "nullability": "NULLABILITY_REQUIRED"
                            }
                          }],
                          "nullability": "NULLABILITY_REQUIRED"
                        }
                      },
                      "arguments": [{
                        "value": {
                          "selection": {
                            "directReference": {
                              "structField": {
                              }
                            }
                          }
                        }
                      }, {
                        "value": {
                          "selection": {
                            "directReference": {
                              "structField": {
                                "field": 1
                              }
                            }
                          }
                        }
                      }]
                    }
                  }]
                }
              },
              "groupings": [{
              }],
              "measures": [{
                "measure": {
                  "functionReference": 1,
                  "phase": "AGGREGATION_PHASE_INTERMEDIATE_TO_RESULT",
                  "outputType": {
                    "decimal": {
                      "scale": 4,
                      "precision": 30,
                      "nullability": "NULLABILITY_NULLABLE"
                    }
                  },
                  "arguments": [{
                    "value": {
                      "selection": {
                        "directReference": {
                          "structField": {
                          }
                        }
                      }
                    }
                  }]
                }
              }]
            }
          },
          "expressions": [{
            "selection": {
              "directReference": {
                "structField": {
                }
              }
            }
          }]
        }
      },
      "names": ["sum#2"]
    }
  }]
}

The input of the sum agg need to be a struct schema with sum and isEmpty.

[CLTFOREVER]

@liujiayi771 yes，the code is latest，but using our Spark version: 3.0.2，what‘s your spark version.

[CLTFOREVER]

@liujiayi771 I guess there were some differences when integrating the community Gluten with our Spark. I will use the community version for verification later. Thank you

[liujiayi771]

@liujiayi771 yes，the code is latest，but using our Spark version: 3.0.2，what‘s your spark version.

I use Spark 3.3.1, maybe your Spark have some optimizations that will change the plan.

[CLTFOREVER]

@liujiayi771 The problem has been resolved，thank you so much.
The reason for this issue is that the variable aggBufferAttributes in the sum function has a different number of parameters in Spark3.0 and Spark3.2. This issue causes the output of the HashAggregateTransformer operator to be different, resulting in a missing parameter when converting to a Substrate plan, as you mentioned earlier