Feature/null eq pushdown

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzerHelper.cpp

@@ -124,6 +124,39 @@ String DAGExpressionAnalyzerHelper::buildIfNullFunction(
     return analyzer->applyFunction(func_name, argument_names, actions, getCollatorFromExpr(expr));
 }
 
+String DAGExpressionAnalyzerHelper::buildNullEqFunction(
+    DAGExpressionAnalyzer * analyzer,
+    const tipb::Expr & expr,
+    const ExpressionActionsPtr & actions)
+{
+    if (expr.children_size() != 2)
+    {
+        throw TiFlashException("Invalid arguments of nullEq function", Errors::Coprocessor::BadRequest);
+    }
+
+    String col1 = analyzer->getActions(expr.children(0), actions, false);
+    String col2 = analyzer->getActions(expr.children(1), actions, false);
+
+    const Block & sample_block = actions->getSampleBlock();
+    bool col1_nullable = sample_block.getByName(col1).type->isNullable();
+    bool col2_nullable = sample_block.getByName(col2).type->isNullable();
+
+    if (!col1_nullable && !col2_nullable)
+    {
+        return analyzer->applyFunction("equals", {col1, col2}, actions, getCollatorFromExpr(expr));
+    }
+
+    String equals = analyzer->applyFunction("equals", {col1, col2}, actions, getCollatorFromExpr(expr));
+    String name = analyzer->getActions(constructInt64LiteralTiExpr(0), actions);    
+
+    String is_null_col1 = analyzer->applyFunction("isNull", {col1}, actions, getCollatorFromExpr(expr));
+    String is_null_col2 = analyzer->applyFunction("isNull", {col2}, actions, getCollatorFromExpr(expr));
+    String and_is_null = analyzer->applyFunction("and", {is_null_col1, is_null_col2}, actions, nullptr);
+    String not_null_equals = analyzer->applyFunction("coalesce", {equals, name}, actions, nullptr);
+
+    return analyzer->applyFunction("or", {and_is_null, not_null_equals}, actions, nullptr);
+}
+
 String DAGExpressionAnalyzerHelper::buildInFunction(
     DAGExpressionAnalyzer * analyzer,
     const tipb::Expr & expr,
@@ -547,6 +580,7 @@ DAGExpressionAnalyzerHelper::FunctionBuilderMap DAGExpressionAnalyzerHelper::fun
      {"tidbIn", DAGExpressionAnalyzerHelper::buildInFunction},
      {"tidbNotIn", DAGExpressionAnalyzerHelper::buildInFunction},
      {"ifNull", DAGExpressionAnalyzerHelper::buildIfNullFunction},
+     {"nullEq", DAGExpressionAnalyzerHelper::buildNullEqFunction},
      {"multiIf", DAGExpressionAnalyzerHelper::buildMultiIfFunction},
      {"tidb_cast", DAGExpressionAnalyzerHelper::buildCastFunction},
      {"cast_int_as_json", DAGExpressionAnalyzerHelper::buildSingleParamJsonRelatedFunctions},

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzerHelper.h

@@ -49,6 +49,11 @@ class DAGExpressionAnalyzerHelper
         const tipb::Expr & expr,
         const ExpressionActionsPtr & actions);
 
+    static String buildNullEqFunction(
+        DAGExpressionAnalyzer * analyzer,
+        const tipb::Expr & expr,
+        const ExpressionActionsPtr & actions);
+
     static String buildLogicalFunction(
         DAGExpressionAnalyzer * analyzer,
         const tipb::Expr & expr,

dbms/src/Flash/Coprocessor/DAGUtils.cpp

@@ -226,13 +226,13 @@ const std::unordered_map<tipb::ScalarFuncSig, String> scalar_func_map({
     //{tipb::ScalarFuncSig::NEJson, "notEquals"},
     {tipb::ScalarFuncSig::NEVectorFloat32, "notEquals"},
 
-    //{tipb::ScalarFuncSig::NullEQInt, "cast"},
-    //{tipb::ScalarFuncSig::NullEQReal, "cast"},
-    //{tipb::ScalarFuncSig::NullEQString, "cast"},
-    //{tipb::ScalarFuncSig::NullEQDecimal, "cast"},
-    //{tipb::ScalarFuncSig::NullEQTime, "cast"},
-    //{tipb::ScalarFuncSig::NullEQDuration, "cast"},
-    //{tipb::ScalarFuncSig::NullEQJson, "cast"},
+    {tipb::ScalarFuncSig::NullEQInt, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQReal, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQString, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQDecimal, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQTime, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQDuration, "nullEq"},
+    {tipb::ScalarFuncSig::NullEQJson, "nullEq"},
 
     {tipb::ScalarFuncSig::PlusReal, "plus"},
     {tipb::ScalarFuncSig::PlusDecimal, "plus"},

dbms/src/Storages/DeltaMerge/Filter/NullEqual.h

@@ -0,0 +1,69 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// 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.
+
+#pragma once
+
+#include <Storages/DeltaMerge/Filter/RSOperator.h>
+#include <Storages/DeltaMerge/Index/RoughCheck.h>
+
+namespace DB::DM
+{
+
+class NullEqual : public ColCmpVal
+{
+public:
+    NullEqual(const Attr & attr_, const Field & value_)
+        : ColCmpVal(attr_, value_)
+    {}
+
+    String name() override { return "nullEqual"; }
+
+    RSResults roughCheck(size_t start_pack, size_t pack_count, const RSCheckParam & param) override
+    {
+        auto rs_index = getRSIndex(param, attr);
+        if (!rs_index)
+            return RSResults(pack_count, RSResult::Some);
+
+        if (value.isNull())
+        {
+            return rs_index->minmax->checkIsNull(start_pack, pack_count);
+        }
+        else
+        {
+            return rs_index->minmax->checkCmp<RoughCheck::CheckEqual>(start_pack, pack_count, value, rs_index->type);
+        }
+    }
+
+    ColumnRangePtr buildSets(const google::protobuf::RepeatedPtrField<tipb::ColumnarIndexInfo> & index_infos) override
+    {
+        if (value.isNull())
+            return UnsupportedColumnRange::create();
+
+        if (auto set = IntegerSet::createValueSet(attr.type, {value}); set)
+        {
+            auto iter = std::find_if(index_infos.begin(), index_infos.end(), [&](const auto & info) {
+                return info.index_type() == tipb::ColumnarIndexType::TypeInverted
+                    && info.inverted_query_info().column_id() == attr.col_id;
+            });
+            if (iter != index_infos.end())
+                return SingleColumnRange::create(
+                    iter->inverted_query_info().column_id(),
+                    iter->inverted_query_info().index_id(),
+                    set);
+        }
+        return UnsupportedColumnRange::create();
+    }
+};
+
+} // namespace DB::DM

dbms/src/Storages/DeltaMerge/Filter/RSOperator.cpp

@@ -25,6 +25,7 @@
 #include <Storages/DeltaMerge/Filter/Like.h>
 #include <Storages/DeltaMerge/Filter/Not.h>
 #include <Storages/DeltaMerge/Filter/NotEqual.h>
+#include <Storages/DeltaMerge/Filter/NullEqual.h>
 #include <Storages/DeltaMerge/Filter/Or.h>
 #include <Storages/DeltaMerge/Filter/RSOperator.h>
 #include <Storages/DeltaMerge/Filter/Unsupported.h>
@@ -45,6 +46,7 @@ RSOperatorPtr createLess(const Attr & attr, const Field & value)
 RSOperatorPtr createLessEqual(const Attr & attr, const Field & value)           { return std::make_shared<LessEqual>(attr, value); }
 RSOperatorPtr createLike(const Attr & attr, const Field & value)                { return std::make_shared<Like>(attr, value); }
 RSOperatorPtr createNot(const RSOperatorPtr & op)                               { return std::make_shared<Not>(op); }
+RSOperatorPtr createNullEqual(const Attr & attr, const Field & value)            { return std::make_shared<NullEqual>(attr, value); }
 RSOperatorPtr createNotEqual(const Attr & attr, const Field & value)            { return std::make_shared<NotEqual>(attr, value); }
 RSOperatorPtr createOr(const RSOperators & children)                            { return std::make_shared<Or>(children); }
 RSOperatorPtr createIsNull(const Attr & attr)                                   { return std::make_shared<IsNull>(attr);}

dbms/src/Storages/DeltaMerge/Filter/RSOperator.h

@@ -181,6 +181,7 @@ RSOperatorPtr createAnd(const RSOperators & children);
 // compare
 RSOperatorPtr createEqual(const Attr & attr, const Field & value);
 RSOperatorPtr createNotEqual(const Attr & attr, const Field & value);
+RSOperatorPtr createNullEqual(const Attr & attr, const Field & value);
 RSOperatorPtr createGreater(const Attr & attr, const Field & value);
 RSOperatorPtr createGreaterEqual(const Attr & attr, const Field & value);
 RSOperatorPtr createLess(const Attr & attr, const Field & value);

dbms/src/Storages/DeltaMerge/FilterParser/FilterParser.cpp

@@ -237,6 +237,8 @@ inline RSOperatorPtr parseTiCompareExpr( //
             return createLessEqual(attr, values[0]);
     case FilterParser::RSFilterType::In:
         return createIn(attr, values);
+    case FilterParser::RSFilterType::NullEqual:
+        return createNullEqual(attr, values[0]);
     default:
         return createUnsupported(fmt::format("Unknown compare type: {}", tipb::ExprType_Name(expr.tp())));
     }
@@ -308,6 +310,7 @@ RSOperatorPtr parseTiExpr(
         case FilterParser::RSFilterType::Less:
         case FilterParser::RSFilterType::LessEqual:
         case FilterParser::RSFilterType::In:
+        case FilterParser::RSFilterType::NullEqual:
             return parseTiCompareExpr(expr, filter_type, scan_column_infos, id_to_attr, timezone_info);
 
         case FilterParser::RSFilterType::IsNull:
@@ -583,13 +586,14 @@ std::unordered_map<tipb::ScalarFuncSig, FilterParser::RSFilterType> FilterParser
     {tipb::ScalarFuncSig::NEDuration, FilterParser::RSFilterType::NotEqual},
     {tipb::ScalarFuncSig::NEJson, FilterParser::RSFilterType::NotEqual},
 
-    //{tipb::ScalarFuncSig::NullEQInt, "cast"},
-    //{tipb::ScalarFuncSig::NullEQReal, "cast"},
-    //{tipb::ScalarFuncSig::NullEQString, "cast"},
-    //{tipb::ScalarFuncSig::NullEQDecimal, "cast"},
-    //{tipb::ScalarFuncSig::NullEQTime, "cast"},
-    //{tipb::ScalarFuncSig::NullEQDuration, "cast"},
-    //{tipb::ScalarFuncSig::NullEQJson, "cast"},
+    {tipb::ScalarFuncSig::NullEQInt, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQReal, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQString, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQDecimal, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQTime, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQDuration, FilterParser::RSFilterType::NullEqual},
+    {tipb::ScalarFuncSig::NullEQJson, FilterParser::RSFilterType::NullEqual},
+
 
     // {tipb::ScalarFuncSig::PlusReal, "plus"},
     // {tipb::ScalarFuncSig::PlusDecimal, "plus"},

dbms/src/Storages/DeltaMerge/FilterParser/FilterParser.h

@@ -83,6 +83,8 @@ class FilterParser
         // NotLike, TiDB will convert it to Not(Like)
 
         IsNull,
+
+        NullEqual,
     };
 
     static std::unordered_map<tipb::ScalarFuncSig, RSFilterType> scalar_func_rs_filter_map;