Implement Agg_BitOr function push down.

dbms/src/AggregateFunctions/AggregateFunctionNull.h

@@ -434,6 +434,12 @@ class AggregateFunctionNullUnary final : public AggregateFunctionNullBase<result
                 const IColumn * nested_column = &column->getNestedColumn();
                 this->nested_function->add(this->nestedPlace(place), &nested_column, row_num, arena);
             }
+            else if (const String & nested_func_name = this->nested_function->getName(); nested_func_name == "groupBitOr" || nested_func_name == "groupBitAnd" || nested_func_name == "groupBitXor")
+            {
+                // If nested_function is bitwise aggregation function,
+                // the result is always not null.
+                this->setFlag(place);
+            }
         }
         else
         {
@@ -466,6 +472,13 @@ class AggregateFunctionNullUnary final : public AggregateFunctionNullBase<result
                 arena,
                 if_argument_pos);
 
+            if (const String & nested_func_name = this->nested_function->getName(); nested_func_name == "groupBitOr" || nested_func_name == "groupBitAnd" || nested_func_name == "groupBitXor")
+            {
+                // If nested_function is bitwise aggregation function,
+                // the result is always not null.
+                this->setFlag(place);
+            }
+
             if constexpr (result_is_nullable)
                 if (!mem_utils::memoryIsByte(null_map, batch_size, std::byte{1}))
                     this->setFlag(place);

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp

@@ -211,6 +211,36 @@ void DAGExpressionAnalyzer::fillArgumentDetail(
     arg_collators.push_back(removeNullable(arg_types.back())->isString() ? getCollatorFromExpr(arg) : nullptr);
 }
 
+void DAGExpressionAnalyzer::fillArgumentDetailForAggFuncBitwise(
+    const ExpressionActionsPtr & actions,
+    const tipb::Expr & arg,
+    Names & arg_names,
+    DataTypes & arg_types,
+    TiDB::TiDBCollators & arg_collators)
+{
+    DataTypePtr uint64_type = std::make_shared<DataTypeUInt64>();
+    const Block & sample_block = actions->getSampleBlock();
+    String name = getActions(arg, actions);
+    DataTypePtr orig_type = sample_block.getByName(name).type;
+    // Bitwise operations can only be applied to integer types.
+    // So we need to cast the argument to UInt64.
+    if (!removeNullable(orig_type)->equals(*uint64_type))
+    {
+        if (orig_type->isNullable())
+        {
+            name = appendCast(makeNullable(uint64_type), actions, name);
+        }
+        else
+        {
+            name = appendCast(uint64_type, actions, name);
+        }
+    }
+
+    arg_names.push_back(name);
+    arg_types.push_back(orig_type->isNullable() ? makeNullable(uint64_type) : uint64_type);
+    arg_collators.push_back(removeNullable(arg_types.back())->isString() ? getCollatorFromExpr(arg) : nullptr);
+}
+
 void DAGExpressionAnalyzer::buildGroupConcat(
     const tipb::Expr & expr,
     const ExpressionActionsPtr & actions,
@@ -330,14 +360,23 @@ void DAGExpressionAnalyzer::buildCommonAggFunc(
     NamesAndTypes & aggregated_columns,
     bool empty_input_as_null)
 {
+    tipb::ExprType tp = expr.tp();
+    bool is_bitwise = tp == tipb::ExprType::Agg_BitAnd || tp == tipb::ExprType::Agg_BitOr || tp == tipb::ExprType::Agg_BitXor;
     auto child_size = expr.children_size();
     Names arg_names;
     DataTypes arg_types;
     TiDB::TiDBCollators arg_collators;
 
     for (Int32 i = 0; i < child_size; ++i)
     {
-        fillArgumentDetail(actions, expr.children(i), arg_names, arg_types, arg_collators);
+        if (is_bitwise)
+        {
+            fillArgumentDetailForAggFuncBitwise(actions, expr.children(i), arg_names, arg_types, arg_collators);
+        }
+        else
+        {
+            fillArgumentDetail(actions, expr.children(i), arg_names, arg_types, arg_collators);
+        }
     }
     // For count(not null column), we can transform it to count() to avoid the cost of convertToFullColumn.
     if (expr.tp() == tipb::ExprType::Count && !expr.has_distinct() && child_size == 1 && !arg_types[0]->isNullable())

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h

@@ -249,6 +249,13 @@ class DAGExpressionAnalyzer : private boost::noncopyable
         DataTypes & arg_types,
         TiDB::TiDBCollators & arg_collators);
 
+    void fillArgumentDetailForAggFuncBitwise(
+        const ExpressionActionsPtr & actions,
+        const tipb::Expr & arg,
+        Names & arg_names,
+        DataTypes & arg_types,
+        TiDB::TiDBCollators & arg_collators);
+
     void makeExplicitSet(
         const tipb::Expr & expr,
         const Block & sample_block,

dbms/src/Flash/Coprocessor/DAGUtils.cpp

@@ -53,7 +53,7 @@ const std::unordered_map<tipb::ExprType, String> agg_func_map({
     {tipb::ExprType::GroupConcat, "groupArray"},
     //{tipb::ExprType::Avg, ""},
     //{tipb::ExprType::Agg_BitAnd, ""},
-    //{tipb::ExprType::Agg_BitOr, ""},
+    {tipb::ExprType::Agg_BitOr, "groupBitOr"},
     //{tipb::ExprType::Agg_BitXor, ""},
     //{tipb::ExprType::Std, ""},
     //{tipb::ExprType::Stddev, ""},

tests/fullstack-test/expr/agg_bit_or_pushdown.test

@@ -0,0 +1,76 @@
+# Copyright 2022 PingCAP, Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+mysql> drop table if exists test.all_null
+mysql> create table test.all_null (a int, b int)
+mysql> alter table test.all_null set tiflash replica 1
+mysql> insert into test.all_null values(1, NULL), (1, NULL), (2, NULL), (2, NULL), (3, NULL);
+
+mysql> drop table if exists test.not_all_null
+mysql> create table test.not_all_null (a int, b int)
+mysql> alter table test.not_all_null set tiflash replica 1
+mysql> insert into test.not_all_null values(1, 1), (1, 2), (2, 2), (2, 4), (3, NULL), (3, 4);
+
+mysql> drop table if exists test.char
+mysql> create table test.char (a int, b char(10))
+mysql> alter table test.char set tiflash replica 1
+mysql> insert into test.char values(1, '1a'), (1, '2a2'), (2, '2bbb'), (2, '4ccc11'), (3, 'adasda'), (3, 'ASD');
+
+mysql> set @@tidb_isolation_read_engines='tiflash'
+
+mysql> select bit_or(b) from test.all_null;
++-----------+
+| bit_or(b) |
++-----------+
+|         0 |
++-----------+
+mysql> select a, bit_or(b) from test.all_null group by a;
++------+-----------+
+| a    | bit_or(b) |
++------+-----------+
+|    1 |         0 |
+|    2 |         0 |
+|    3 |         0 |
++------+-----------+
+
+
+mysql> select bit_or(b) from test.not_all_null;
++-----------+
+| bit_or(b) |
++-----------+
+|         7 |
++-----------+
+mysql> select a, bit_or(b) from test.not_all_null group by a;
++------+-----------+
+| a    | bit_or(b) |
++------+-----------+
+|    1 |         3 |
+|    2 |         6 |
+|    3 |         4 |
++------+-----------+
+
+mysql> select bit_or(b) from test.char;
++-----------+
+| bit_or(b) |
++-----------+
+|         7 |
++-----------+
+mysql> select a, bit_or(b) from test.char group by a;
++------+-----------+
+| a    | bit_or(b) |
++------+-----------+
+|    1 |         3 |
+|    2 |         6 |
+|    3 |         0 |
++------+-----------+