From dd25d23dfac8ecf5452190b983fe6ccc8b9187eb Mon Sep 17 00:00:00 2001
From: NEUpanning <emmning@163.com>
Date: Wed, 13 Nov 2024 12:24:00 -0800
Subject: [PATCH] Disable aggregate pushdown for decimal type (#11298)

Summary:
Currently, `int64_t` enables push-down for decimal type. This PR disables aggregate pushdown for decimal type regardless of c++ type.

Fixes https://github.com/facebookincubator/velox/issues/11290

Pull Request resolved: https://github.com/facebookincubator/velox/pull/11298

Reviewed By: Yuhta

Differential Revision: D65834210

Pulled By: kevinwilfong

fbshipit-source-id: 422f7eda8f4184c6fa83055e7cf430ff5053d387
---
 velox/exec/tests/TableScanTest.cpp            | 27 +++++++++++++++++++
 .../lib/aggregates/MinMaxAggregateBase.cpp    | 24 +++++++++++------
 .../lib/aggregates/SimpleNumericAggregate.h   |  3 ++-
 3 files changed, 45 insertions(+), 9 deletions(-)

diff --git a/velox/exec/tests/TableScanTest.cpp b/velox/exec/tests/TableScanTest.cpp
index 24c202dae7e3..a7d66c56b26a 100644
--- a/velox/exec/tests/TableScanTest.cpp
+++ b/velox/exec/tests/TableScanTest.cpp
@@ -3492,6 +3492,33 @@ TEST_F(TableScanTest, aggregationPushdown) {
   EXPECT_EQ(0, loadedToValueHook(task));
 }
 
+TEST_F(TableScanTest, decimalDisableAggregationPushdown) {
+  vector_size_t size = 1'000;
+  auto rowVector = makeRowVector({
+      makeFlatVector<int64_t>(size, [](auto row) { return 1; }),
+      makeFlatVector<int64_t>(
+          size, [](auto row) { return row; }, nullptr, DECIMAL(18, 2)),
+  });
+
+  auto filePath = TempFilePath::create();
+  writeToFile(filePath->getPath(), {rowVector});
+
+  createDuckDbTable({rowVector});
+
+  auto rowType = asRowType(rowVector->type());
+  auto op = PlanBuilder()
+                .tableScan(rowType)
+                .singleAggregation({"c0"}, {"min(c1)", "max(c1)", "sum(c1)"})
+                .planNode();
+
+  auto task = assertQuery(
+      op,
+      {filePath},
+      "SELECT c0, min(c1), max(c1), sum(c1) FROM tmp GROUP BY 1");
+  auto stats = task->taskStats().pipelineStats[0].operatorStats[1].runtimeStats;
+  EXPECT_EQ(stats.end(), stats.find("loadedToValueHook"));
+}
+
 TEST_F(TableScanTest, bitwiseAggregationPushdown) {
   auto vectors = makeVectors(10, 1'000);
   auto filePath = TempFilePath::create();
diff --git a/velox/functions/lib/aggregates/MinMaxAggregateBase.cpp b/velox/functions/lib/aggregates/MinMaxAggregateBase.cpp
index 1499cbd74fef..3ac57d9a32ea 100644
--- a/velox/functions/lib/aggregates/MinMaxAggregateBase.cpp
+++ b/velox/functions/lib/aggregates/MinMaxAggregateBase.cpp
@@ -111,10 +111,14 @@ class MaxAggregate : public MinMaxAggregate<T> {
       const std::vector<VectorPtr>& args,
       bool mayPushdown) override {
     if constexpr (BaseAggregate::template kMayPushdown<T>) {
-      if (mayPushdown && args[0]->isLazy()) {
-        BaseAggregate::template pushdown<
-            velox::aggregate::MinMaxHook<T, false>>(groups, rows, args[0]);
-        return;
+      if (!args[0]->type()->isDecimal()) {
+        if (mayPushdown && args[0]->isLazy()) {
+          BaseAggregate::template pushdown<
+              velox::aggregate::MinMaxHook<T, false>>(groups, rows, args[0]);
+          return;
+        }
+      } else {
+        mayPushdown = false;
       }
     } else {
       mayPushdown = false;
@@ -208,10 +212,14 @@ class MinAggregate : public MinMaxAggregate<T> {
       const std::vector<VectorPtr>& args,
       bool mayPushdown) override {
     if constexpr (BaseAggregate::template kMayPushdown<T>) {
-      if (mayPushdown && args[0]->isLazy()) {
-        BaseAggregate::template pushdown<velox::aggregate::MinMaxHook<T, true>>(
-            groups, rows, args[0]);
-        return;
+      if (!args[0]->type()->isDecimal()) {
+        if (mayPushdown && args[0]->isLazy()) {
+          BaseAggregate::template pushdown<
+              velox::aggregate::MinMaxHook<T, true>>(groups, rows, args[0]);
+          return;
+        }
+      } else {
+        mayPushdown = false;
       }
     } else {
       mayPushdown = false;
diff --git a/velox/functions/lib/aggregates/SimpleNumericAggregate.h b/velox/functions/lib/aggregates/SimpleNumericAggregate.h
index b5a35ae0757e..3e9cca3db3b8 100644
--- a/velox/functions/lib/aggregates/SimpleNumericAggregate.h
+++ b/velox/functions/lib/aggregates/SimpleNumericAggregate.h
@@ -100,7 +100,8 @@ class SimpleNumericAggregate : public exec::Aggregate {
     DecodedVector decoded(*arg, rows, !mayPushdown);
     auto encoding = decoded.base()->encoding();
     if constexpr (kMayPushdown<TData>) {
-      if (encoding == VectorEncoding::Simple::LAZY) {
+      if (encoding == VectorEncoding::Simple::LAZY &&
+          !arg->type()->isDecimal()) {
         velox::aggregate::SimpleCallableHook<TData, UpdateSingleValue> hook(
             exec::Aggregate::offset_,
             exec::Aggregate::nullByte_,