Fix bug, support nullable function, and optimal (apache#67)

be/src/vec/aggregate_functions/aggregate_function_null.h

@@ -125,11 +125,13 @@ class AggregateFunctionNullBase : public IAggregateFunctionHelper<Derived> {
     }
 
     void deserialize(AggregateDataPtr place, std::istream& buf, Arena* arena) const override {
-        bool flag = 1;
+        bool flag = true;
         if (result_is_nullable) readBinary(flag, buf);
         if (flag) {
             setFlag(place);
             nested_function->deserialize(nestedPlace(place), buf, arena);
+        } else {
+            initFlag(place);
         }
     }
 

be/src/vec/exec/vaggregation_node.cpp

@@ -302,7 +302,7 @@ Status AggregationNode::get_next(RuntimeState* state, Block* block, bool* eos) {
 Status AggregationNode::close(RuntimeState* state) {
     RETURN_IF_ERROR(ExecNode::close(state));
     VExpr::close(_probe_expr_ctxs, state);
-    _executor.close();
+    if (_executor.close) _executor.close();
     return Status::OK();
 }
 

be/src/vec/exec/volap_scanner.cpp

@@ -50,7 +50,7 @@ Status VOlapScanner::get_block(RuntimeState* state, vectorized::Block* block, bo
         block->clear();
         std::vector<vectorized::MutableColumnPtr> columns;
         for (auto slot : get_query_slots()) {
-            columns.emplace_back(slot->get_empty_mutable_column());
+            columns.emplace_back(slot->get_empty_mutable_column())->reserve(state->batch_size());
         }
         while (true) {
             // block is full, break

be/src/vec/exprs/vectorized_agg_fn.cpp

@@ -76,7 +76,8 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const RowDescriptor& desc, M
         // Now, For correctness. We have to treat each AggFn argument as nullable. which will cause execute slowly
         // TODO: RECHCK THE BEHAVIOR
         auto data_type = _input_exprs_ctxs[i]->root()->data_type();
-        argument_types.emplace_back(data_type->isNullable() ? data_type : std::make_shared<DataTypeNullable>(data_type));
+//        argument_types.emplace_back(data_type->isNullable() ? data_type : std::make_shared<DataTypeNullable>(data_type));
+        argument_types.emplace_back(data_type);
         child_expr_name.emplace_back(_input_exprs_ctxs[i]->root()->expr_name());
     }
     _function = AggregateFunctionSimpleFactory::instance().get(_fn.name.function_name,
@@ -165,7 +166,7 @@ std::vector<ColumnPtr> AggFnEvaluator::_get_argment_columns(Block* block) const
         int column_id = -1;
         _input_exprs_ctxs[i]->execute(block, &column_id);
         auto ptr = block->getByPosition(column_id).column->convertToFullColumnIfConst();
-        columns[i] = makeNullable(ptr);
+        columns[i] = ptr;
     }
     return columns;
 }

fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java

@@ -1717,6 +1717,16 @@ public Expr getResultValue() throws AnalysisException {
      * overwrite this method to plan correct
      */
     public boolean isNullable() {
+        if (fn != null) {
+            if (fn.getNullableMode().equals(Function.NullableMode.DEPEND_ON_ARGUMENT)) {
+                for (Expr expr : children) {
+                    if (expr.isNullable()) return true;
+                }
+                return false;
+            } else {
+                return fn.getNullableMode().equals(Function.NullableMode.ALWAYS_NULLABLE);
+            }
+        }
         return true;
     }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/Function.java

@@ -83,6 +83,16 @@ public enum CompareMode {
         IS_MATCHABLE
     }
 
+    public enum NullableMode {
+        // Whether output column is nullable is depend on the input column is nullable
+        DEPEND_ON_ARGUMENT,
+        // like 'str_to_date', 'cast', 'date_format' etc, the output column is nullable
+        // depend on input content
+        ALWAYS_NULLABLE,
+        // like 'count', the output column is always not nullable
+        ALWAYS_NOT_NULLABLE
+    }
+
     public static final long UNIQUE_FUNCTION_ID = 0;
     // Function id, every function has a unique id. Now all built-in functions' id is 0
     private long id = 0;
@@ -105,6 +115,8 @@ public enum CompareMode {
     private HdfsURI location;
     private TFunctionBinaryType binaryType;
 
+    private NullableMode nullableMode = NullableMode.DEPEND_ON_ARGUMENT;
+
     private boolean vectorized;
 
     // library's checksum to make sure all backends use one library to serve user's request
@@ -734,4 +746,8 @@ public List<Comparable> getInfo(boolean isVerbose) {
     boolean isVectorized() {
         return vectorized;
     }
+
+    public NullableMode getNullableMode() {
+        return nullableMode;
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java

@@ -1060,8 +1060,9 @@ public void addBuiltinBothScalaAndVectorized(Function fn) {
             vecFns = Lists.newArrayList();
             vectorizedFunctions.put(fn.functionName(), vecFns);
         }
-        vecFns.add(new Function(fn.getId(), fn.getFunctionName(), fn.getArgs(), fn.getReturnType(), fn.hasVarArgs(),
-                fn.getBinaryType(), fn.isUserVisible(), true));
+        ScalarFunction scalarFunction = (ScalarFunction)fn;
+        vecFns.add(ScalarFunction.createVecBuiltin(scalarFunction.functionName(), scalarFunction.getSymbolName(),
+                Lists.newArrayList(scalarFunction.getArgs()), scalarFunction.hasVarArgs(), scalarFunction.getReturnType(), scalarFunction.isUserVisible()));
     }
 
 

fe/fe-core/src/main/java/org/apache/doris/catalog/ScalarFunction.java

@@ -227,6 +227,7 @@ public static ScalarFunction createBuiltinVecOperator(
         return createVecBuiltin(name, symbol, argTypes, false, retType, false);
     }
 
+    //TODO: This method should not be here, move to other place in the future
     public static ScalarFunction createVecBuiltin(
             String name, String symbol, ArrayList<Type> argTypes,
             boolean hasVarArgs, Type retType, boolean userVisible) {