Feature/issue 67 uda support changelog (#137)

* UDA support 'retract' on changelog stream. close #67

Author: sunset3000

src/AggregateFunctions/AggregateFunctionFactory.cpp

@@ -70,8 +70,14 @@ static DataTypes convertLowCardinalityTypesToNested(const DataTypes & types)
     return res_types;
 }
 
+/// proton: starts
 AggregateFunctionPtr AggregateFunctionFactory::get(
-    const String & name, const DataTypes & argument_types, const Array & parameters, AggregateFunctionProperties & out_properties) const
+    const String & name,
+    const DataTypes & argument_types,
+    const Array & parameters,
+    AggregateFunctionProperties & out_properties,
+    bool is_changelog_input) const
+/// proton: ends
 {
     auto types_without_low_cardinality = convertLowCardinalityTypesToNested(argument_types);
 
@@ -92,7 +98,7 @@ AggregateFunctionPtr AggregateFunctionFactory::get(
             [](const auto & type) { return type->onlyNull(); });
 
         AggregateFunctionPtr nested_function = getImpl(
-            name, nested_types, nested_parameters, out_properties, has_null_arguments);
+            name, nested_types, nested_parameters, out_properties, has_null_arguments, is_changelog_input);
 
         // Pure window functions are not real aggregate functions. Applying
         // combinators doesn't make sense for them, they must handle the
@@ -103,20 +109,22 @@ AggregateFunctionPtr AggregateFunctionFactory::get(
             return combinator->transformAggregateFunction(nested_function, out_properties, types_without_low_cardinality, parameters);
     }
 
-    auto with_original_arguments = getImpl(name, types_without_low_cardinality, parameters, out_properties, false);
+    auto with_original_arguments = getImpl(name, types_without_low_cardinality, parameters, out_properties, false, is_changelog_input);
 
     if (!with_original_arguments)
         throw Exception("Logical error: AggregateFunctionFactory returned nullptr", ErrorCodes::LOGICAL_ERROR);
     return with_original_arguments;
 }
 
-
+/// proton: starts
 AggregateFunctionPtr AggregateFunctionFactory::getImpl(
     const String & name_param,
     const DataTypes & argument_types,
     const Array & parameters,
     AggregateFunctionProperties & out_properties,
-    bool has_null_arguments) const
+    bool has_null_arguments,
+    bool is_changelog_input) const
+/// proton: ends
 {
     String name = getAliasToOrName(name_param);
     bool is_case_insensitive = false;
@@ -194,7 +202,7 @@ AggregateFunctionPtr AggregateFunctionFactory::getImpl(
     }
 
     /// proton: starts. Check user defined aggr function
-    auto aggr = UserDefinedFunctionFactory::getAggregateFunction(name, argument_types, parameters, out_properties);
+    auto aggr = UserDefinedFunctionFactory::getAggregateFunction(name, argument_types, parameters, out_properties, is_changelog_input);
     if (aggr)
         return aggr;
     /// proton: ends
@@ -211,12 +219,19 @@ AggregateFunctionPtr AggregateFunctionFactory::getImpl(
         throw Exception(ErrorCodes::UNKNOWN_AGGREGATE_FUNCTION, "Unknown aggregate function {}{}", name, extra_info);
 }
 
-
+/// proton: starts
 AggregateFunctionPtr AggregateFunctionFactory::tryGet(
-    const String & name, const DataTypes & argument_types, const Array & parameters, AggregateFunctionProperties & out_properties) const
+    const String & name,
+    const DataTypes & argument_types,
+    const Array & parameters,
+    AggregateFunctionProperties & out_properties,
+    bool is_changelog_input) const
+/// proton: ends
 {
     return isAggregateFunctionName(name)
-        ? get(name, argument_types, parameters, out_properties)
+        /// proton: starts
+        ? get(name, argument_types, parameters, out_properties, is_changelog_input)
+        /// proton: ends
         : nullptr;
 }
 

src/AggregateFunctions/AggregateFunctionFactory.h

@@ -63,18 +63,22 @@ class AggregateFunctionFactory final : private boost::noncopyable, public IFacto
         CaseSensitiveness case_sensitiveness = CaseSensitive);
 
     /// Throws an exception if not found.
+    /// proton: starts. Add 'is_changelog_input' param to allow aggregate function being aware whether the input stream is a changelog
     AggregateFunctionPtr
     get(const String & name,
         const DataTypes & argument_types,
         const Array & parameters,
-        AggregateFunctionProperties & out_properties) const;
+        AggregateFunctionProperties & out_properties,
+        bool is_changelog_input = false) const;
 
     /// Returns nullptr if not found.
     AggregateFunctionPtr tryGet(
         const String & name,
         const DataTypes & argument_types,
         const Array & parameters,
-        AggregateFunctionProperties & out_properties) const;
+        AggregateFunctionProperties & out_properties,
+        bool is_changelog_input = false) const;
+    /// proton: ends
 
     /// Get properties if the aggregate function exists.
     std::optional<AggregateFunctionProperties> tryGetProperties(const String & name) const;
@@ -86,12 +90,15 @@ class AggregateFunctionFactory final : private boost::noncopyable, public IFacto
     /// proton: ends
 
 private:
+    /// proton: starts
     AggregateFunctionPtr getImpl(
         const String & name,
         const DataTypes & argument_types,
         const Array & parameters,
         AggregateFunctionProperties & out_properties,
-        bool has_null_arguments) const;
+        bool has_null_arguments,
+        bool is_changelog_input = false) const;
+    /// proton: ends
 
     std::optional<AggregateFunctionProperties> tryGetPropertiesImpl(const String & name) const;
 

src/AggregateFunctions/AggregateFunctionJavaScriptAdapter.cpp

@@ -6,6 +6,7 @@
 #include <V8/ConvertDataTypes.h>
 #include <V8/Utils.h>
 #include <Common/logger_useful.h>
+#include <span>
 
 namespace DB
 {
@@ -86,10 +87,12 @@ JavaScriptBlueprint::JavaScriptBlueprint(const String & name, const String & sou
 
         {
             v8::Local<v8::Value> val;
-            if (!obj->Get(local_ctx, V8::to_v8(isolate_, "has_customized_emit")).ToLocal(&val) || !val->IsUndefined())
+            if (obj->Get(local_ctx, V8::to_v8(isolate_, "has_customized_emit")).ToLocal(&val) && !val->IsUndefined())
             {
-                LOG_INFO(&Poco::Logger::get("JavaScriptAggregateFunction"), "JavaScript UDA '{}' has defined its own emit strategy", name);
-                has_user_defined_emit_strategy = true;
+                has_user_defined_emit_strategy = V8::from_v8<bool>(isolate_, val);
+                if (has_user_defined_emit_strategy)
+                    LOG_INFO(
+                        &Poco::Logger::get("JavaScriptAggregateFunction"), "JavaScript UDA '{}' has defined its own emit strategy", name);
             }
         }
 
@@ -110,9 +113,19 @@ JavaScriptBlueprint::~JavaScriptBlueprint() noexcept
 }
 
 JavaScriptAggrFunctionState::JavaScriptAggrFunctionState(
-    const JavaScriptBlueprint & blueprint, const std::vector<UserDefinedFunctionConfiguration::Argument> & arguments)
+    const JavaScriptBlueprint & blueprint,
+    const std::vector<UserDefinedFunctionConfiguration::Argument> & arguments,
+    const bool is_changelog_input_)
+    : is_changelog_input(is_changelog_input_)
 {
     columns.reserve(arguments.size());
+
+    /// check _tp_delta column
+    if (unlikely(arguments.back().type->getTypeId() != TypeIndex::Int8))
+        throw Exception(
+            ErrorCodes::NOT_IMPLEMENTED,
+            "Tha last argument of JavaScript UDA is '_tp_delta' column, which should be 'int8'. Invalid type.");
+
     for (const auto & arg : arguments)
     {
         auto col = arg.type->createColumn();
@@ -206,11 +219,28 @@ JavaScriptAggrFunctionState::~JavaScriptAggrFunctionState()
 
 void JavaScriptAggrFunctionState::add(const IColumn ** src_columns, size_t row_num)
 {
-    for (size_t i = 0; auto & col : columns)
-    {
-        col->insertFrom(*src_columns[i], row_num);
-        i++;
-    }
+    assert(columns.size() >= 1);
+    size_t num_of_input_columns = columns.size() - 1;
+
+    for (size_t i = 0; i < num_of_input_columns; i++)
+        columns[i]->insertFrom(*src_columns[i], row_num);
+
+    /// _tp_delta column
+    if (is_changelog_input)
+        columns.back()->insert(1);
+}
+
+void JavaScriptAggrFunctionState::negate(const IColumn ** src_columns, size_t row_num)
+{
+    assert(columns.size() >= 1);
+    size_t num_of_input_columns = columns.size() - 1;
+
+    for (size_t i = 0; i < num_of_input_columns; i++)
+        columns[i]->insertFrom(*src_columns[i], row_num);
+
+    /// _tp_delta column
+    if (is_changelog_input)
+        columns.back()->insert(-1);
 }
 
 void JavaScriptAggrFunctionState::reinitCache()
@@ -237,10 +267,12 @@ AggregateFunctionJavaScriptAdapter::AggregateFunctionJavaScriptAdapter(
     JavaScriptUserDefinedFunctionConfigurationPtr config_,
     const DataTypes & types,
     const Array & params_,
+    bool is_changelog_input_,
     size_t max_v8_heap_size_in_bytes_)
     : IAggregateFunctionHelper<AggregateFunctionJavaScriptAdapter>(types, params_)
     , config(config_)
     , num_arguments(types.size())
+    , is_changelog_input(is_changelog_input_)
     , max_v8_heap_size_in_bytes(max_v8_heap_size_in_bytes_)
     , blueprint(config->name, config->source)
 {
@@ -260,7 +292,7 @@ DataTypePtr AggregateFunctionJavaScriptAdapter::getReturnType() const
 void AggregateFunctionJavaScriptAdapter::create(AggregateDataPtr __restrict place) const
 {
     V8::checkHeapLimit(blueprint.isolate.get(), max_v8_heap_size_in_bytes);
-    new (place) Data(blueprint, config->arguments);
+    new (place) Data(blueprint, config->arguments, is_changelog_input);
 }
 
 /// destroy instance of UDF
@@ -307,6 +339,11 @@ void AggregateFunctionJavaScriptAdapter::add(AggregateDataPtr __restrict place,
     this->data(place).add(columns, row_num);
 }
 
+void AggregateFunctionJavaScriptAdapter::negate(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const
+{
+    this->data(place).negate(columns, row_num);
+}
+
 void AggregateFunctionJavaScriptAdapter::merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const
 {
     auto & data = this->data(place);
@@ -376,11 +413,13 @@ size_t AggregateFunctionJavaScriptAdapter::flush(AggregateDataPtr __restrict pla
         v8::Local<v8::Function> local_func = v8::Local<v8::Function>::New(isolate_, data.process_func);
 
         /// Second, convert the input column into the corresponding object used by UDF
-        auto argv = V8::prepareArguments(isolate_, config->arguments, data.columns);
+        /// remove the _tp_delta column if the input stream is not changelog
+        auto column_size = is_changelog_input ? config->arguments.size() : config->arguments.size() - 1;
+        auto argv = V8::prepareArguments(isolate_, std::span(config->arguments.begin(), column_size), data.columns);
 
         /// Third, execute the UDF and get aggregate state (only support the final state now, intermediate state is not supported
         v8::Local<v8::Value> res;
-        if (!local_func->Call(ctx, local_obj, static_cast<int>(config->arguments.size()), argv.data()).ToLocal(&res))
+        if (!local_func->Call(ctx, local_obj, static_cast<int>(column_size), argv.data()).ToLocal(&res))
             V8::throwException(
                 isolate_,
                 try_catch,

src/AggregateFunctions/AggregateFunctionJavaScriptAdapter.h

@@ -42,6 +42,8 @@ struct JavaScriptAggrFunctionState
 
     /// the number of emits, 0 means no emit, >1 means it has some aggregate results to emit
     size_t emit_times = 0;
+    /// whether or not it is used in changelog input stream
+    bool is_changelog_input = false;
 
     /// JavaScript UDA code looks like below. For far, it can only contain member functions and the has_customized_emit bool data member or function
     /// {
@@ -55,12 +57,16 @@ struct JavaScriptAggrFunctionState
     ///    has_customized_emit : false /// Define if the the aggregation has user defined emit strategy
     /// }
     JavaScriptAggrFunctionState(
-        const JavaScriptBlueprint & blueprint, const std::vector<UserDefinedFunctionConfiguration::Argument> & arguments);
+        const JavaScriptBlueprint & blueprint,
+        const std::vector<UserDefinedFunctionConfiguration::Argument> & arguments,
+        const bool is_changelog_input_);
 
     ~JavaScriptAggrFunctionState();
 
     void add(const IColumn ** src_columns, size_t row_num);
 
+    void negate(const IColumn ** src_columns, size_t row_num);
+
     void reinitCache();
 };
 
@@ -82,6 +88,7 @@ class AggregateFunctionJavaScriptAdapter final : public IAggregateFunctionHelper
 
     const JavaScriptUserDefinedFunctionConfigurationPtr config;
     size_t num_arguments;
+    bool is_changelog_input = false;
     size_t max_v8_heap_size_in_bytes;
     JavaScriptBlueprint blueprint;
 
@@ -90,6 +97,8 @@ class AggregateFunctionJavaScriptAdapter final : public IAggregateFunctionHelper
         JavaScriptUserDefinedFunctionConfigurationPtr config_,
         const DataTypes & types,
         const Array & params_,
+        /// If the input stream is changelog, aggregate function will pass _tp_delta column to JavaScript function
+        bool is_changelog_input_,
         size_t max_v8_heap_size_in_bytes_);
 
     String getName() const override;
@@ -113,6 +122,9 @@ class AggregateFunctionJavaScriptAdapter final : public IAggregateFunctionHelper
     /// get instance of UDF from AggregateData and execute UDF
     void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override;
 
+    /// for changelog processing, delete existing row from current aggregation result
+    void negate(AggregateDataPtr __restrict place, const IColumn ** columns, size_t /*row_num*/, Arena * /*arena*/) const override;
+
     /// Merge with other Aggregate Data, maybe used before finalize result
     void merge(AggregateDataPtr __restrict /*place*/, ConstAggregateDataPtr /*rhs*/, Arena *) const override;
 

src/AggregateFunctions/tests/gtest_uda.cpp

@@ -19,7 +19,7 @@ using namespace DB;
 
 std::unique_ptr<v8::Platform> platform;
 
-String ARGS_CEP1 = R"###([{ "name": "value","type": "int64"}])###";
+String ARGS_CEP1 = R"###([{ "name": "value","type": "int64"}, {"name": "_tp_delta", "type": "int8"}])###";
 String RETURN_CEP1 = "uint32";
 String UDA_CEP1 = R"###(
 {
@@ -49,7 +49,7 @@ String UDA_CEP1 = R"###(
     has_customized_emit : true
 })###";
 
-String ARGS_UDA1 = R"###([{ "name": "value","type": "int64"}])###";
+String ARGS_UDA1 = R"###([{ "name": "value","type": "int64"}, {"name": "_tp_delta", "type": "int8"}])###";
 String RETURN_UDA1 = "float32";
 String UDA1 = R"###(
 {
@@ -202,7 +202,7 @@ TEST_F(UDATestCase, add)
     DataTypes types = getDataTypes(ARGS_CEP1);
     Array params;
     size_t max_heap_size = 100 * 1024 * 1024;
-    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, max_heap_size);
+    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, false, max_heap_size);
 
     ASSERT_TRUE(aggr_function.hasUserDefinedEmit());
 
@@ -242,7 +242,7 @@ TEST_F(UDATestCase, CheckPoint)
     DataTypes types = getDataTypes(ARGS_UDA1);
     Array params;
     size_t max_heap_size = 100 * 1024 * 1024;
-    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, max_heap_size);
+    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, false, max_heap_size);
 
     std::unique_ptr<AggregateFunctionJavaScriptAdapter::Data[], AggregateFunctionJavaScriptAdapter::DataDeleter> places{
         static_cast<AggregateFunctionJavaScriptAdapter::Data *>(malloc(aggr_function.sizeOfData())),
@@ -284,7 +284,7 @@ TEST_F(UDATestCase, Merge)
     DataTypes types = getDataTypes(ARGS_UDA1);
     Array params;
     size_t max_heap_size = 100 * 1024 * 1024;
-    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, max_heap_size);
+    auto aggr_function = AggregateFunctionJavaScriptAdapter(config, types, params, false, max_heap_size);
 
     std::unique_ptr<AggregateFunctionJavaScriptAdapter::Data[], AggregateFunctionJavaScriptAdapter::DataDeleter> places{
         static_cast<AggregateFunctionJavaScriptAdapter::Data *>(malloc(2 * aggr_function.sizeOfData())),

src/Functions/UserDefined/JavaScriptUserDefinedFunction.cpp

@@ -1,6 +1,7 @@
 #include <Functions/UserDefined/JavaScriptUserDefinedFunction.h>
 #include <V8/ConvertDataTypes.h>
 #include <V8/Utils.h>
+#include <span>
 
 namespace DB
 {
@@ -64,7 +65,7 @@ ColumnPtr JavaScriptUserDefinedFunction::userDefinedExecuteImpl(
         v8::Local<v8::Function> local_func = v8::Local<v8::Function>::New(isolate_, js_ctx->func);
 
         /// Second, convert the input column into the corresponding object used by UDF
-        auto argv = V8::prepareArguments(isolate_, config->arguments, columns);
+        auto argv = V8::prepareArguments(isolate_, std::span(config->arguments), columns);
 
         /// Third, execute the UDF and get result
         v8::Local<v8::Value> res;