Refine window when preceding and following boundary type are unbounded

dbms/src/AggregateFunctions/AggregateFunctionAvg.h

@@ -131,6 +131,25 @@ class AggregateFunctionAvg final
         }
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        if constexpr (IsDecimal<TResult>)
+        {
+            ScaleType left_scale = result_scale - scale;
+            TResult result = this->data(place).sum.value * getScaleMultiplier<TResult>(left_scale)
+                / static_cast<typename TResult::NativeType>(this->data(place).count);
+            auto & container = static_cast<ColumnDecimal<TResult> &>(to).getData();
+            container.resize_fill(container.size() + num, result);
+        }
+        else
+        {
+            auto & container = static_cast<ColumnFloat64 &>(to).getData();
+            container.resize_fill(
+                container.size() + num,
+                static_cast<Float64>(this->data(place).sum) / this->data(place).count);
+        }
+    }
+
     void create(AggregateDataPtr __restrict place) const override
     {
         using Data

dbms/src/AggregateFunctions/AggregateFunctionCount.h

@@ -120,6 +120,12 @@ class AggregateFunctionCount final
         static_cast<ColumnUInt64 &>(to).getData().push_back(data(place).count);
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        auto & container = static_cast<ColumnUInt64 &>(to).getData();
+        container.resize_fill(container.size() + num, data(place).count);
+    }
+
     /// May be used for optimization.
     static void addDelta(AggregateDataPtr __restrict place, UInt64 x) { data(place).count += x; }
 
@@ -209,6 +215,12 @@ class AggregateFunctionCountNotNullUnary final
         static_cast<ColumnUInt64 &>(to).getData().push_back(data(place).count);
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        auto & container = static_cast<ColumnUInt64 &>(to).getData();
+        container.resize_fill(container.size() + num, data(place).count);
+    }
+
     const char * getHeaderFilePath() const override { return __FILE__; }
 };
 
@@ -282,6 +294,12 @@ class AggregateFunctionCountNotNullVariadic final
         static_cast<ColumnUInt64 &>(to).getData().push_back(data(place).count);
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        auto & container = static_cast<ColumnUInt64 &>(to).getData();
+        container.resize_fill(container.size() + num, data(place).count);
+    }
+
     const char * getHeaderFilePath() const override { return __FILE__; }
 
 private:

dbms/src/AggregateFunctions/AggregateFunctionMinMaxAny.h

@@ -61,6 +61,20 @@ struct SingleValueDataFixed : public CommonImpl
             static_cast<ColumnType &>(to).insertDefault();
     }
 
+    void insertBatchResultInto(IColumn & to, size_t num) const
+    {
+        if (has())
+        {
+            auto & container = static_cast<ColumnType &>(to).getData();
+            container.resize_fill(num + container.size(), value);
+        }
+        else
+        {
+            static_cast<ColumnType &>(to).insertManyDefaults(num);
+        }
+    }
+
+
     void write(WriteBuffer & buf, const IDataType & /*data_type*/) const
     {
         writeBinary(has(), buf);
@@ -239,6 +253,17 @@ struct SingleValueDataString : public CommonImpl
             static_cast<ColumnString &>(to).insertDefault();
     }
 
+    void insertBatchResultInto(IColumn & to, size_t num) const
+    {
+        if (has())
+        {
+            for (size_t i = 0; i < num; ++i)
+                static_cast<ColumnString &>(to).insertDataWithTerminatingZero(getData(), size);
+        }
+        else
+            static_cast<ColumnString &>(to).insertManyDefaults(num);
+    }
+
     void setCollators(const TiDB::TiDBCollators & collators_)
     {
         collator = !collators_.empty() ? collators_[0] : nullptr;
@@ -451,6 +476,16 @@ struct SingleValueDataGeneric : public CommonImpl
             to.insertDefault();
     }
 
+    void insertBatchResultInto(IColumn & to, size_t num) const
+    {
+        if (has())
+        {
+            to.insertMany(value, num);
+        }
+        else
+            to.insertManyDefaults(num);
+    }
+
     void write(WriteBuffer & buf, const IDataType & data_type) const
     {
         if (!value.isNull())
@@ -791,6 +826,11 @@ class AggregateFunctionsSingleValue final
         this->data(place).insertResultInto(to);
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        this->data(place).insertBatchResultInto(to, num);
+    }
+
     const char * getHeaderFilePath() const override { return __FILE__; }
 };
 

dbms/src/AggregateFunctions/AggregateFunctionMinMaxWindow.h

@@ -41,23 +41,31 @@ struct SingleValueDataFixedForWindow
 
     template <bool is_min>
     void insertMinOrMaxResultInto(IColumn & to) const
+    {
+        insertBatchMinOrMaxResultInto<is_min>(to, 1);
+    }
+
+    template <bool is_min>
+    void insertBatchMinOrMaxResultInto(IColumn & to, size_t num) const
     {
         if (!saved_values.empty())
         {
             if constexpr (is_min)
             {
                 const auto & iter = saved_values.begin();
-                static_cast<ColumnType &>(to).getData().push_back(*iter);
+                auto & container = static_cast<ColumnType &>(to).getData();
+                container.resize_fill(num + container.size(), *iter);
             }
             else
             {
                 const auto & iter = saved_values.rbegin();
-                static_cast<ColumnType &>(to).getData().push_back(*iter);
+                auto & container = static_cast<ColumnType &>(to).getData();
+                container.resize_fill(num + container.size(), *iter);
             }
         }
         else
         {
-            static_cast<ColumnType &>(to).insertDefault();
+            static_cast<ColumnType &>(to).insertManyDefaults(num);
         }
     }
 
@@ -66,6 +74,10 @@ struct SingleValueDataFixedForWindow
 
     void insertMinResultInto(IColumn & to) const { insertMinOrMaxResultInto<true>(to); }
 
+    void insertBatchMaxResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<false>(to, num); }
+
+    void insertBatchMinResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<true>(to, num); }
+
     void reset() { saved_values.clear(); }
 
     void decrease(const IColumn & column, size_t row_num)
@@ -122,23 +134,35 @@ struct SingleValueDataStringForWindow
 
     template <bool is_min>
     void insertMinOrMaxResultInto(IColumn & to) const
+    {
+        insertBatchMinOrMaxResultInto<is_min>(to, 1);
+    }
+
+    template <bool is_min>
+    void insertBatchMinOrMaxResultInto(IColumn & to, size_t num) const
     {
         if (!saved_values.empty())
         {
             if constexpr (is_min)
             {
                 const auto & iter = saved_values.begin();
-                static_cast<ColumnString &>(to).insertDataWithTerminatingZero(iter->value.data, iter->value.size);
+                const char * data = iter->value.data;
+                size_t size = iter->value.size;
+                for (size_t i = 0; i < num; ++i)
+                    static_cast<ColumnString &>(to).insertDataWithTerminatingZero(data, size);
             }
             else
             {
                 const auto & iter = saved_values.rbegin();
-                static_cast<ColumnString &>(to).insertDataWithTerminatingZero(iter->value.data, iter->value.size);
+                const char * data = iter->value.data;
+                size_t size = iter->value.size;
+                for (size_t i = 0; i < num; ++i)
+                    static_cast<ColumnString &>(to).insertDataWithTerminatingZero(data, size);
             }
         }
         else
         {
-            static_cast<ColumnString &>(to).insertDefault();
+            static_cast<ColumnString &>(to).insertManyDefaults(num);
         }
     }
 
@@ -147,6 +171,10 @@ struct SingleValueDataStringForWindow
 
     void insertMinResultInto(IColumn & to) const { insertMinOrMaxResultInto<true>(to); }
 
+    void insertBatchMaxResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<false>(to, num); }
+
+    void insertBatchMinResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<true>(to, num); }
+
     void reset() { saved_values.clear(); }
 
     void decrease(const IColumn & column, size_t row_num)
@@ -179,23 +207,29 @@ struct SingleValueDataGenericForWindow
 
     template <bool is_min>
     void insertMinOrMaxResultInto(IColumn & to) const
+    {
+        insertBatchMinOrMaxResultInto<is_min>(to, 1);
+    }
+
+    template <bool is_min>
+    void insertBatchMinOrMaxResultInto(IColumn & to, size_t num) const
     {
         if (!saved_values.empty())
         {
             if constexpr (is_min)
             {
                 const auto & iter = saved_values.begin();
-                to.insert(*iter);
+                to.insertMany(*iter, num);
             }
             else
             {
                 const auto & iter = saved_values.rbegin();
-                to.insert(*iter);
+                to.insertMany(*iter, num);
             }
         }
         else
         {
-            to.insertDefault();
+            to.insertManyDefaults(num);
         }
     }
 
@@ -204,6 +238,10 @@ struct SingleValueDataGenericForWindow
 
     void insertMinResultInto(IColumn & to) const { insertMinOrMaxResultInto<true>(to); }
 
+    void insertBatchMaxResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<false>(to, num); }
+
+    void insertBatchMinResultInto(IColumn & to, size_t num) const { insertBatchMinOrMaxResultInto<true>(to, num); }
+
     void reset() { saved_values.clear(); }
 
     void decrease(const IColumn & column, size_t row_num)
@@ -241,6 +279,8 @@ struct AggregateFunctionMinDataForWindow : Data
 
     void insertResultInto(IColumn & to) const { Data::insertMinResultInto(to); }
 
+    void insertBatchResultInto(IColumn & to, size_t num) const { Data::insertBatchMinResultInto(to, num); }
+
     static const char * name() { return "min_for_window"; }
 };
 
@@ -258,6 +298,8 @@ struct AggregateFunctionMaxDataForWindow : Data
 
     void insertResultInto(IColumn & to) const { Data::insertMaxResultInto(to); }
 
+    void insertBatchResultInto(IColumn & to, size_t num) const { Data::insertBatchMaxResultInto(to, num); }
+
     static const char * name() { return "max_for_window"; }
 };
 

dbms/src/AggregateFunctions/AggregateFunctionNothing.h

@@ -56,6 +56,8 @@ class AggregateFunctionNothing final : public IAggregateFunctionHelper<Aggregate
 
     void insertResultInto(ConstAggregateDataPtr, IColumn & to, Arena *) const override { to.insertDefault(); }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict, IColumn &, size_t, Arena *) const override {}
+
     const char * getHeaderFilePath() const override { return __FILE__; }
 };
 

dbms/src/AggregateFunctions/AggregateFunctionNull.h

@@ -629,6 +629,20 @@ class AggregateFunctionNullUnaryForWindow
         }
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        auto & to_concrete = static_cast<ColumnNullable &>(to);
+        if (getCounter(place) > 0)
+        {
+            nested_function->insertBatchResultInto(nestedPlace(place), to_concrete.getNestedColumn(), num, nullptr);
+            to_concrete.insertManyNulls(num, 0);
+        }
+        else
+        {
+            to_concrete.insertManyDefaults(num);
+        }
+    }
+
     bool hasTrivialDestructor() const override { return nested_function->hasTrivialDestructor(); }
 
     size_t sizeOfData() const override { return prefix_size + nested_function->sizeOfData(); }

dbms/src/AggregateFunctions/AggregateFunctionSum.h

@@ -449,6 +449,23 @@ class AggregateFunctionSum final
             static_cast<ColumnVector<TResult> &>(to).getData().push_back(this->data(place).get());
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena *) const override
+    {
+        if constexpr (IsDecimal<TResult>)
+        {
+            // TODO refine it
+            auto & container = static_cast<ColumnDecimal<TResult> &>(to).getData();
+            container.reserve(container.size() + num);
+            for (size_t i = 0; i < num; ++i)
+                static_cast<ColumnDecimal<TResult> &>(to).getData().push_back(this->data(place).get(), result_scale);
+        }
+        else
+        {
+            auto & container = static_cast<ColumnVector<TResult> &>(to).getData();
+            container.resize_fill(container.size() + num, this->data(place).get());
+        }
+    }
+
     const char * getHeaderFilePath() const override { return __FILE__; }
 };
 

dbms/src/AggregateFunctions/IAggregateFunction.h

@@ -119,6 +119,11 @@ class IAggregateFunction
     /// Inserts results into a column.
     virtual void insertResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, Arena * arena) const = 0;
 
+    /// Inserts batch results into a column
+    virtual void insertBatchResultInto(ConstAggregateDataPtr __restrict place, IColumn & to, size_t num, Arena * arena)
+        const
+        = 0;
+
     /** Returns true for aggregate functions of type -State.
       * They are executed as other aggregate functions, but not finalized (return an aggregation state that can be combined with another).
       */
@@ -384,6 +389,11 @@ class IAggregateFunctionHelper : public IAggregateFunction
         throw Exception("decrease function is not implemented yet");
     }
 
+    void insertBatchResultInto(ConstAggregateDataPtr __restrict, IColumn &, size_t, Arena *) const override
+    {
+        throw Exception("insertBatchResultInto function is not implemented yet");
+    }
+
     void reset(AggregateDataPtr __restrict) const override { throw Exception("reset function is not implemented yet"); }
 };
 

dbms/src/Columns/ColumnNullable.h

@@ -151,10 +151,15 @@ class ColumnNullable final : public COWPtrHelper<IColumn, ColumnNullable>
     void insertManyDefaults(size_t length) override
     {
         getNestedColumn().insertManyDefaults(length);
+        insertManyNulls(length, 1);
+    }
+
+    void insertManyNulls(size_t length, UInt8 val)
+    {
         auto & map = getNullMapData();
         size_t old_size = map.size();
         map.resize(old_size + length);
-        memset(map.data() + old_size, 1, length);
+        memset(map.data() + old_size, val, length);
     }
 
     void popBack(size_t n) override;