fix: bind CALL $params per Eval and keep them in params_type

.github/workflows/neug-extension-test.yml

@@ -145,6 +145,7 @@ jobs:
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_sniffer.py -k "parquet"
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_copy_temp.py -k "parquet or httpfs"
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_gds_temp_graph.py
+        NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_call_params.py
 
     - name: Run GDS Tests
       run: |

extension/test_out_of_tree/test_out_of_tree_extension.cc

@@ -14,15 +14,127 @@
  * limitations under the License.
  */
 
+#include "neug/common/columns/value_columns.h"
 #include "neug/compiler/extension/extension_api.h"
+#include "neug/compiler/function/neug_call_function.h"
+#include "neug/execution/common/context.h"
+#include "neug/execution/common/params_map.h"
+#include "neug/utils/exception/exception.h"
+
+namespace {
+
+// String literal, or $param (ParamsMap key without '$').
+struct DeferredCallArg {
+  bool is_param = false;
+  std::string param_name;
+  std::string literal;
+
+  static DeferredCallArg FromLiteral(std::string value) {
+    DeferredCallArg arg;
+    arg.literal = std::move(value);
+    return arg;
+  }
+
+  static DeferredCallArg FromParam(std::string name) {
+    DeferredCallArg arg;
+    arg.is_param = true;
+    arg.param_name = std::move(name);
+    return arg;
+  }
+
+  std::string resolve(const neug::execution::ParamsMap& params) const {
+    if (!is_param) {
+      return literal;
+    }
+    auto it = params.find(param_name);
+    if (it == params.end() || it->second.IsNull()) {
+      THROW_INVALID_ARGUMENT_EXCEPTION(
+          "Missing parameter for TEST_ECHO_PARAM: " + param_name);
+    }
+    return it->second.GetValue<std::string>();
+  }
+};
+
+struct EchoParamFuncInput : public neug::function::CallFuncInputBase {
+  DeferredCallArg arg;
+  std::string value;
+
+  std::unique_ptr<neug::function::CallFuncInputBase> bindParams(
+      const neug::execution::ParamsMap& params) const override {
+    auto bound = std::make_unique<EchoParamFuncInput>(*this);
+    bound->value = arg.resolve(params);
+    return bound;
+  }
+};
+
+struct TestEchoParamFunction {
+  static constexpr const char* name = "TEST_ECHO_PARAM";
+
+  static neug::function::function_set getFunctionSet() {
+    auto func = std::make_unique<neug::function::NeugCallFunction>(
+        name,
+        neug::function::call_input_types{
+            neug::common::DataType(neug::common::DataTypeId::kVarchar)},
+        neug::function::call_output_columns{
+            {"value",
+             neug::common::DataType(neug::common::DataTypeId::kVarchar)}});
+
+    func->bindFunc =
+        [](const neug::Schema&, const neug::execution::ContextMeta&,
+           const ::physical::PhysicalPlan& plan,
+           int op_idx) -> std::unique_ptr<neug::function::CallFuncInputBase> {
+      const auto& procedure = plan.plan(op_idx).opr().procedure_call();
+      if (procedure.query().arguments_size() < 1) {
+        THROW_INVALID_ARGUMENT_EXCEPTION(
+            "TEST_ECHO_PARAM requires 1 VARCHAR argument");
+      }
+      auto input = std::make_unique<EchoParamFuncInput>();
+      const auto& arg = procedure.query().arguments(0);
+      if (arg.has_param()) {
+        input->arg = DeferredCallArg::FromParam(arg.param().name());
+      } else if (arg.has_const_() && arg.const_().has_str()) {
+        input->arg = DeferredCallArg::FromLiteral(arg.const_().str());
+      } else {
+        THROW_INVALID_ARGUMENT_EXCEPTION(
+            "TEST_ECHO_PARAM requires a string constant or $param");
+      }
+      return input;
+    };
+
+    func->execFunc = [](const neug::function::CallFuncInputBase& input_base,
+                        neug::IStorageInterface&) {
+      const auto& input = static_cast<const EchoParamFuncInput&>(input_base);
+
+      neug::ValueColumnBuilder<std::string> builder;
+      builder.reserve(1);
+      builder.push_back_opt(input.value);
+
+      neug::execution::Context ctx;
+      neug::DataChunk chunk;
+      chunk.set(0, builder.finish());
+      ctx.append_chunk(std::move(chunk));
+      ctx.tag_ids = {0};
+      return ctx;
+    };
+
+    neug::function::function_set set;
+    set.push_back(std::move(func));
+    return set;
+  }
+};
+
+}  // namespace
 
 extern "C" {
 
 void Init() {
+  neug::extension::ExtensionAPI::registerFunction<TestEchoParamFunction>(
+      neug::catalog::CatalogEntryType::TABLE_FUNCTION_ENTRY);
   neug::extension::ExtensionAPI::registerExtension(
       neug::extension::ExtensionInfo{
           "test_out_of_tree",
-          "Test extension for validating out-of-tree extension builds."});
+          "Test extension for out-of-tree builds and CALL $params "
+          "(TEST_ECHO_PARAM)."});
 }
 
 const char* Name() { return "TEST_OUT_OF_TREE"; }

include/neug/compiler/function/neug_call_function.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <memory>
 #include <vector>
 #include "neug/common/types.h"
 #include "neug/compiler/function/function.h"
@@ -35,8 +36,14 @@ class Context;
 namespace function {
 struct CallFuncInputBase {
   virtual ~CallFuncInputBase() = default;
-  // Resolve deferred $param args against query ParamsMap before execFunc.
-  virtual void bindParams(const execution::ParamsMap& /*params*/) {}
+  // Bind deferred $param args into a per-Eval input.
+  // Opr stores this object unbound after bindFunc; Eval calls bindParams and
+  // executes against the returned input. Default: no deferred params — return
+  // nullptr and Opr will exec with the unbound template (must stay immutable).
+  virtual std::unique_ptr<CallFuncInputBase> bindParams(
+      const execution::ParamsMap& /*params*/) const {
+    return nullptr;
+  }
 };
 
 using call_bind_func_t = std::function<std::unique_ptr<CallFuncInputBase>(

proto/stored_procedure.proto

@@ -29,6 +29,8 @@ message Argument {
   oneof value {
     common.Value const = 3;  // real value
     common.Variable var = 4;
+    // Dynamic query param ($name): resolved from ParamsMap at exec time.
+    common.DynamicParam param = 5;
   }
 }
 

src/compiler/gopt/g_query_converter.cpp

@@ -1294,11 +1294,23 @@ void GQueryConvertor::convertProcedureCall(
     const auto& param = params.at(pos);
     auto queryArgPB = std::make_unique<::procedure::Argument>();
     queryArgPB->set_param_ind(pos);
-    // Dynamic query params ($name): keep the name (no '$') and resolve from
-    // ParamsMap at exec time. Literals / variables keep const / var.
+    // $param -> typed DynamicParam for params_type / ParamsMap; literals/vars
+    // keep const / var.
     if (param->expressionType == common::ExpressionType::PARAMETER) {
-      queryArgPB->set_param_name(
-          param->constCast<binder::ParameterExpression>().getName());
+      const auto& paramExpr = param->constCast<binder::ParameterExpression>();
+      if (pos >= callFunc.parameterTypes.size()) {
+        THROW_EXCEPTION_WITH_FILE_LINE("CALL $param index out of signature: " +
+                                       paramExpr.getName());
+      }
+      auto dynPB = std::make_unique<::common::DynamicParam>();
+      dynPB->set_name(paramExpr.getName());
+      dynPB->set_index(static_cast<int32_t>(pos));
+      // CALL $params keep UNKNOWN in the binder; use the procedure signature
+      // type for ParamsMap deserialization.
+      dynPB->set_allocated_data_type(
+          typeConverter->convertLogicalType(callFunc.parameterTypes[pos].copy())
+              .release());
+      queryArgPB->set_allocated_param(dynPB.release());
     } else {
       auto paramPB = exprConvertor->convert(*param, {});
       if (paramPB->operators_size() == 0) {

src/execution/execute/ops/retrieve/gds_algo.cc

@@ -31,6 +31,7 @@ GDSAlgoOpr::GDSAlgoOpr(std::unique_ptr<function::CallFuncInputBase> algo_input,
 neug::result<neug::execution::Context> GDSAlgoOpr::Eval(
     IStorageInterface& graph_interface, const ParamsMap& params,
     neug::execution::Context&& ctx, neug::execution::OprTimer* timer) {
+  (void) ctx;
   (void) timer;
   if (algo_func_ == nullptr) {
     THROW_RUNTIME_ERROR("GDSAlgoOpr: GDSAlgoFunction pointer is null");
@@ -42,8 +43,9 @@ neug::result<neug::execution::Context> GDSAlgoOpr::Eval(
   if (algo_input_ == nullptr) {
     THROW_RUNTIME_ERROR("GDSAlgoOpr: algo input is null");
   }
-  algo_input_->bindParams(params);
-  return algo_func_->execFunc(*algo_input_, graph_interface);
+  auto bound_input = algo_input_->bindParams(params);
+  const auto& input = bound_input ? *bound_input : *algo_input_;
+  return algo_func_->execFunc(input, graph_interface);
 }
 
 neug::result<OpBuildResultT> GDSAlgoOprBuilder::Build(

src/execution/execute/ops/retrieve/procedure_call.cc

@@ -23,13 +23,14 @@ namespace execution {
 namespace ops {
 class ProcedureCallOpr : public IOperator {
  private:
-  std::unique_ptr<neug::function::CallFuncInputBase> callInput;
-  function::NeugCallFunction* callFunction;
+  // Unbound template from bindFunc (shared across concurrent Evals).
+  std::unique_ptr<neug::function::CallFuncInputBase> unboundInput_;
+  function::NeugCallFunction* callFunction_;
 
  public:
   ProcedureCallOpr(std::unique_ptr<neug::function::CallFuncInputBase> input,
                    function::NeugCallFunction* callFunction)
-      : callInput(std::move(input)), callFunction(callFunction) {}
+      : unboundInput_(std::move(input)), callFunction_(callFunction) {}
 
   ~ProcedureCallOpr() override = default;
 
@@ -39,14 +40,25 @@ class ProcedureCallOpr : public IOperator {
       IStorageInterface& graph, const ParamsMap& params,
       neug::execution::Context&& ctx,
       neug::execution::OprTimer* timer) override {
-    if (callFunction == nullptr) {
+    (void) ctx;
+    (void) timer;
+    if (callFunction_ == nullptr) {
       THROW_RUNTIME_ERROR("ProcedureCallOpr: callFunction is nullptr");
     }
-    callInput->bindParams(params);
+    if (unboundInput_ == nullptr) {
+      THROW_RUNTIME_ERROR("ProcedureCallOpr: unbound input is nullptr");
+    }
+    if (callFunction_->execFunc == nullptr) {
+      THROW_RUNTIME_ERROR("ProcedureCallOpr: execFunc is nullptr");
+    }
+    // bindParams returns a per-Eval bound input; nullptr means no deferred
+    // params and the unbound template is safe to exec as-is.
+    auto boundInput = unboundInput_->bindParams(params);
+    const auto& input = boundInput ? *boundInput : *unboundInput_;
     return neug::result<neug::execution::Context>(
-        callFunction->execFunc(*callInput, graph));
-  }  // namespace ops
-};   // namespace execution
+        callFunction_->execFunc(input, graph));
+  }
+};
 
 neug::result<OpBuildResultT> ProcedureCallOprBuilder::Build(
     const neug::Schema& schema, const ContextMeta& ctx_meta,
@@ -77,4 +89,4 @@ neug::result<OpBuildResultT> ProcedureCallOprBuilder::Build(
 
 }  // namespace ops
 }  // namespace execution
-}  // namespace neug
+}  // namespace neug

src/execution/execute/plan_parser.cc

@@ -537,6 +537,21 @@ static void parse_params_type_impl(const physical::PhysicalPlan& plan,
       expression_parse(unfold_opr.input_expr(), params_type);
       break;
     }
+    case physical::PhysicalOpr_Operator::OpKindCase::kProcedureCall: {
+      const auto& query = plan.plan(i).opr().procedure_call().query();
+      for (int j = 0; j < query.arguments_size(); ++j) {
+        const auto& arg = query.arguments(j);
+        if (!arg.has_param()) {
+          continue;
+        }
+        const auto& param = arg.param();
+        if (params_type.find(param.name()) != params_type.end()) {
+          continue;
+        }
+        params_type[param.name()] = parse_from_ir_data_type(param.data_type());
+      }
+      break;
+    }
     default: {
       break;
     }

tests/unittest/CMakeLists.txt

@@ -22,8 +22,6 @@ add_neug_test(test_mmap_container test_mmap_container.cc)
 
 add_neug_test(test_connection test_connection.cc)
 
-add_neug_test(test_call_params test_call_params.cc)
-
 add_neug_test(test_query_result test_query_result.cc)
 
 add_neug_test(test_column test_column.cc)