[bug](pipelineX) fix java-udaf failed with open pipelineX

zhangstar333 · Sep 27, 2023 · 141a1bc · 141a1bc
1 parent a3427cb
commit 141a1bc
Showing 1 changed file with 15 additions and 0 deletions.
diff --git a/be/src/vec/exprs/vectorized_agg_fn.cpp b/be/src/vec/exprs/vectorized_agg_fn.cpp
@@ -335,6 +335,21 @@ AggFnEvaluator::AggFnEvaluator(AggFnEvaluator& evaluator, RuntimeState* state)
           _function(evaluator._function),
           _expr_name(evaluator._expr_name),
           _agg_columns(evaluator._agg_columns) {
+    if (evaluator._fn.binary_type == TFunctionBinaryType::JAVA_UDF) {
+        DataTypes tmp_argument_types;
+        tmp_argument_types.reserve(evaluator._input_exprs_ctxs.size());
+        // prepare for argument
+        for (int i = 0; i < evaluator._input_exprs_ctxs.size(); ++i) {
+            auto data_type = evaluator._input_exprs_ctxs[i]->root()->data_type();
+            tmp_argument_types.emplace_back(data_type);
+        }
+        const DataTypes& argument_types =
+                _real_argument_types.empty() ? tmp_argument_types : _real_argument_types;
+        _function = AggregateJavaUdaf::create(evaluator._fn, argument_types, evaluator._data_type);
+        static_cast<AggregateJavaUdaf*>(_function.get())->check_udaf(evaluator._fn);
+    }
+    DCHECK(_function != nullptr);
+
     _input_exprs_ctxs.resize(evaluator._input_exprs_ctxs.size());
     for (size_t i = 0; i < _input_exprs_ctxs.size(); i++) {
         WARN_IF_ERROR(evaluator._input_exprs_ctxs[i]->clone(state, _input_exprs_ctxs[i]), "");