This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1d03523339f [feature](ivm) Track MIN/MAX extreme counts for
incremental refresh (#69286)
1d03523339f is described below
commit 1d03523339fbd6ec27b70e58b6d45e06bc196c1e
Author: yujun <[email protected]>
AuthorDate: Tue Oct 20 16:13:13 2026 +0800
[feature](ivm) Track MIN/MAX extreme counts for incremental refresh (#69286)
Incremental refresh of a MIN/MAX materialized view currently falls back
to a
complete refresh whenever a deleted row could be the current extreme.
Deleting
one of several rows that share the minimum value is a common case, yet
the MV
does not remember how many rows hold that extreme, so the refresh cannot
tell
that the extreme survives.
This PR adds `COUNT_AT_MIN` / `COUNT_AT_MAX` and stores that count as a
hidden
IVM state column. The delta plan now emits the count at the insert-side
and
delete-side extreme. Apply keeps the old extreme while its surviving
count is
positive; only deleting the last row at the extreme still falls back to
a
complete refresh.
Trace issue: https://github.com/apache/doris/issues/65418
### What changed?
- Add public `COUNT_AT_MIN(expr)` / `COUNT_AT_MAX(expr)` aggregate
functions.
They return the number of input rows equal to the group's
minimum/maximum,
ignore NULLs, return 0 for empty or all-NULL input, and reject DISTINCT.
They are equivalent to Oracle's
`COUNT(*) KEEP (DENSE_RANK FIRST/LAST ORDER BY expr)`.
- Add `MIN_COUNT` / `MAX_COUNT` IVM hidden state keys and hidden columns
named
in the existing `__DORIS_IVM_AGG_{ordinal}_{STATE}_COL__` format.
- Emit insert/delete extreme counts in the delta aggregate.
- Use the surviving extreme count in the apply guard and update the
hidden
count state.
- Add BE/FE unit coverage and a regression suite.
---
.../aggregate_function_count_at_min_max.cpp | 178 +++++++++++++++++++
.../aggregate_function_count_at_min_max.h | 167 ++++++++++++++++++
.../aggregate_function_simple_factory.cpp | 2 +
be/src/exprs/aggregate/single_value_data.h | 10 ++
be/src/exprs/function/function_dict_get_many.cpp | 1 +
.../exprs/aggregate/agg_count_at_min_max_test.cpp | 163 ++++++++++++++++++
.../doris/catalog/BuiltinAggregateFunctions.java | 4 +
.../mtmv/ivm/agg/IvmAggExpressionBuilder.java | 2 +
.../mtmv/ivm/agg/IvmAggExtremalProcessor.java | 191 +++++++++++++++++----
.../doris/mtmv/ivm/agg/IvmAggFunctionKind.java | 2 +
.../mtmv/ivm/agg/IvmAggFunctionProcessor.java | 6 +
.../doris/mtmv/ivm/agg/IvmAggMaxProcessor.java | 31 ++--
.../doris/mtmv/ivm/agg/IvmAggMinProcessor.java | 31 ++--
.../apache/doris/mtmv/ivm/agg/IvmAggStateKey.java | 4 +-
.../expressions/functions/agg/CountAtMax.java | 113 ++++++++++++
.../expressions/functions/agg/CountAtMin.java | 113 ++++++++++++
.../visitor/AggregateFunctionVisitor.java | 10 ++
.../doris/mtmv/ivm/IvmAggDeltaHandlerTest.java | 10 +-
.../mtmv/ivm/IvmNormalizeUniqKeysMTMVTest.java | 14 +-
.../doris/mtmv/ivm/agg/IvmAggMaxProcessorTest.java | 30 +++-
.../doris/mtmv/ivm/agg/IvmAggMinProcessorTest.java | 30 +++-
.../mtmv/ivm/agg/IvmAggProcessorTestBase.java | 23 +++
.../mtmv_p0/ivm/test_ivm_agg_expr_over_agg_1.out | 2 +
.../mtmv_p0/ivm/test_ivm_agg_expr_over_agg_2.out | 2 +
.../data/mtmv_p0/ivm/test_ivm_minmax_count.out | 25 +++
.../mtmv_p0/ivm/test_ivm_minmax_count_join.out | 13 ++
.../data/query_p0/aggregate/count_at_min_max.out | 36 ++++
.../mtmv_p0/ivm/test_ivm_minmax_count.groovy | 162 +++++++++++++++++
.../mtmv_p0/ivm/test_ivm_minmax_count_join.groovy | 148 ++++++++++++++++
.../query_p0/aggregate/count_at_min_max.groovy | 126 ++++++++++++++
30 files changed, 1564 insertions(+), 85 deletions(-)
diff --git a/be/src/exprs/aggregate/aggregate_function_count_at_min_max.cpp
b/be/src/exprs/aggregate/aggregate_function_count_at_min_max.cpp
new file mode 100644
index 00000000000..df67ac115a3
--- /dev/null
+++ b/be/src/exprs/aggregate/aggregate_function_count_at_min_max.cpp
@@ -0,0 +1,178 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "exprs/aggregate/aggregate_function_count_at_min_max.h"
+
+#include "common/exception.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/define_primitive_type.h"
+#include "core/data_type/primitive_type.h"
+#include "exprs/aggregate/aggregate_function_simple_factory.h"
+#include "exprs/aggregate/factory_helpers.h"
+#include "exprs/aggregate/helpers.h"
+
+namespace doris {
+
+namespace {
+
+template <template <typename> class ExtremeData>
+AggregateFunctionPtr create_aggregate_function_count_at_extreme(const String&
name,
+ const
DataTypes& argument_types,
+ const
DataTypePtr& result_type,
+ const bool
result_is_nullable,
+ const
AggregateFunctionAttr& attr) {
+ assert_arity_range(name, argument_types, 1, 1);
+ switch (argument_types[0]->get_primitive_type()) {
+ case PrimitiveType::TYPE_STRING:
+ case PrimitiveType::TYPE_CHAR:
+ case PrimitiveType::TYPE_VARCHAR:
+ case PrimitiveType::TYPE_JSONB:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataString>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DATE:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_DATE>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DATETIME:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_DATETIME>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DATEV2:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_DATEV2>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DATETIMEV2:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DATETIMEV2>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_TIMESTAMP_NS:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_TIMESTAMP_NS>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_TIMESTAMPTZ:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_TIMESTAMPTZ>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_TIMEV2:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_TIMEV2>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_IPV4:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_IPV4>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_IPV6:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_IPV6>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_UUID:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_UUID>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_BOOLEAN:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_BOOLEAN>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_TINYINT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_TINYINT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_SMALLINT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_SMALLINT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_INT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_INT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_BIGINT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_BIGINT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_LARGEINT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_LARGEINT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_FLOAT:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_FLOAT>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DOUBLE:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataFixed<TYPE_DOUBLE>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DECIMAL32:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DECIMAL32>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DECIMAL64:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DECIMAL64>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DECIMALV2:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DECIMALV2>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DECIMAL128I:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DECIMAL128I>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_DECIMAL256:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+ AggregateFunctionsCountAtExtreme<
+ ExtremeData<SingleValueDataFixed<TYPE_DECIMAL256>>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_ARRAY:
+ case PrimitiveType::TYPE_MAP:
+ case PrimitiveType::TYPE_STRUCT:
+ case PrimitiveType::TYPE_AGG_STATE:
+ case PrimitiveType::TYPE_BITMAP:
+ case PrimitiveType::TYPE_HLL:
+ case PrimitiveType::TYPE_QUANTILE_STATE:
+ return
creator_without_type::create_unary_arguments_return_not_nullable<
+
AggregateFunctionsCountAtExtreme<ExtremeData<SingleValueDataColumn>>>(
+ argument_types, result_is_nullable, attr);
+ case PrimitiveType::TYPE_VARBINARY:
+ throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+ "Aggregate function {} does not support "
+ "VARBINARY arguments",
+ name);
+ default:
+ return nullptr;
+ }
+}
+
+} // namespace
+
+void
register_aggregate_function_count_at_min_max(AggregateFunctionSimpleFactory&
factory) {
+ factory.register_function_both(
+ "count_at_min",
+
create_aggregate_function_count_at_extreme<AggregateFunctionCountAtMinData>);
+ factory.register_function_both(
+ "count_at_max",
+
create_aggregate_function_count_at_extreme<AggregateFunctionCountAtMaxData>);
+}
+
+} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_count_at_min_max.h
b/be/src/exprs/aggregate/aggregate_function_count_at_min_max.h
new file mode 100644
index 00000000000..91d327947f9
--- /dev/null
+++ b/be/src/exprs/aggregate/aggregate_function_count_at_min_max.h
@@ -0,0 +1,167 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_number.h"
+#include "core/types.h"
+#include "exprs/aggregate/aggregate_function.h"
+#include "exprs/aggregate/single_value_data.h"
+
+namespace doris {
+
+/**
+ * State shared by COUNT_AT_MIN and COUNT_AT_MAX: the current extreme value
and the
+ * number of input rows that are equal to it. The wrapped Data type owns the
value and
+ * decides whether a new value is better (smaller for MIN, greater for MAX).
+ */
+template <typename Data>
+struct AggregateFunctionCountAtMinData : Data {
+ using Self = AggregateFunctionCountAtMinData;
+ int64_t count = 0;
+
+ AggregateFunctionCountAtMinData() { reset(); }
+
+ void change_if_better(const IColumn& column, size_t row_num, Arena& arena)
{
+ if (!this->has()) {
+ this->set(column, row_num, arena);
+ count = 1;
+ } else if (this->is_equal_to(column, row_num)) {
+ ++count;
+ } else if (this->set_if_smaller(column, row_num, arena)) {
+ count = 1;
+ }
+ }
+
+ void change_if_better(const Self& to, Arena& arena) {
+ if (!to.has()) {
+ return;
+ }
+ if (!this->has()) {
+ this->set(to, arena);
+ count = to.count;
+ } else if (this->is_equal_to(to)) {
+ count += to.count;
+ } else if (this->set_if_smaller(to, arena)) {
+ count = to.count;
+ }
+ }
+
+ void reset() {
+ Data::reset();
+ count = 0;
+ }
+
+ static const char* name() { return "count_at_min"; }
+};
+
+template <typename Data>
+struct AggregateFunctionCountAtMaxData : Data {
+ using Self = AggregateFunctionCountAtMaxData;
+ int64_t count = 0;
+
+ AggregateFunctionCountAtMaxData() { reset(); }
+
+ void change_if_better(const IColumn& column, size_t row_num, Arena& arena)
{
+ if (!this->has()) {
+ this->set(column, row_num, arena);
+ count = 1;
+ } else if (this->is_equal_to(column, row_num)) {
+ ++count;
+ } else if (this->set_if_greater(column, row_num, arena)) {
+ count = 1;
+ }
+ }
+
+ void change_if_better(const Self& to, Arena& arena) {
+ if (!to.has()) {
+ return;
+ }
+ if (!this->has()) {
+ this->set(to, arena);
+ count = to.count;
+ } else if (this->is_equal_to(to)) {
+ count += to.count;
+ } else if (this->set_if_greater(to, arena)) {
+ count = to.count;
+ }
+ }
+
+ void reset() {
+ Data::reset();
+ count = 0;
+ }
+
+ static const char* name() { return "count_at_max"; }
+};
+
+/**
+ * Aggregate function returning the number of rows equal to the minimum (or
maximum)
+ * value in the group. NULL inputs are ignored. An empty group, or a group
whose inputs
+ * are all NULL, returns 0.
+ */
+template <typename Data>
+class AggregateFunctionsCountAtExtreme final
+ : public IAggregateFunctionDataHelper<Data,
AggregateFunctionsCountAtExtreme<Data>> {
+private:
+ using Base = IAggregateFunctionDataHelper<Data,
AggregateFunctionsCountAtExtreme<Data>>;
+
+public:
+ explicit AggregateFunctionsCountAtExtreme(const DataTypes& arguments) :
Base(arguments) {}
+
+ String get_name() const override { return Data::name(); }
+
+ DataTypePtr get_return_type() const override { return
std::make_shared<DataTypeInt64>(); }
+
+ void create(AggregateDataPtr __restrict place) const override { new
(place) Data; }
+
+ void add(AggregateDataPtr __restrict place, const IColumn** columns,
ssize_t row_num,
+ Arena& arena) const override {
+ this->data(place).change_if_better(*columns[0], row_num, arena);
+ }
+
+ void reset(AggregateDataPtr place) const override {
this->data(place).reset(); }
+
+ void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
+ Arena& arena) const override {
+ this->data(place).change_if_better(this->data(rhs), arena);
+ }
+
+ void serialize(ConstAggregateDataPtr __restrict place, BufferWritable&
buf) const override {
+ this->data(place).write(buf, this->argument_types[0],
IAggregateFunction::version);
+ buf.write_binary(this->data(place).count);
+ }
+
+ void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
+ Arena& arena) const override {
+ this->data(place).read(buf, this->argument_types[0],
IAggregateFunction::version, arena);
+ buf.read_binary(this->data(place).count);
+ }
+
+ void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn&
to) const override {
+ assert_cast<ColumnInt64&,
TypeCheckOnRelease::DISABLE>(to).insert_value(
+ this->data(place).count);
+ }
+};
+
+} // namespace doris
diff --git a/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
b/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
index e0e48cef8c4..49f3da1f981 100644
--- a/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
+++ b/be/src/exprs/aggregate/aggregate_function_simple_factory.cpp
@@ -32,6 +32,7 @@ void
register_aggregate_function_combinator_foreachv2(AggregateFunctionSimpleFac
void register_aggregate_function_sum(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_minmax(AggregateFunctionSimpleFactory&
factory);
+void
register_aggregate_function_count_at_min_max(AggregateFunctionSimpleFactory&
factory);
void register_aggregate_function_max_min_by(AggregateFunctionSimpleFactory&
factory);
void register_aggregate_function_avg(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_count(AggregateFunctionSimpleFactory&
factory);
@@ -89,6 +90,7 @@ AggregateFunctionSimpleFactory&
AggregateFunctionSimpleFactory::instance() {
std::call_once(oc, [&]() {
register_aggregate_function_sum(instance);
register_aggregate_function_minmax(instance);
+ register_aggregate_function_count_at_min_max(instance);
register_aggregate_function_max_min_by(instance);
register_aggregate_function_avg(instance);
register_aggregate_function_count(instance);
diff --git a/be/src/exprs/aggregate/single_value_data.h
b/be/src/exprs/aggregate/single_value_data.h
index ff947577d5d..798725414f9 100644
--- a/be/src/exprs/aggregate/single_value_data.h
+++ b/be/src/exprs/aggregate/single_value_data.h
@@ -137,6 +137,10 @@ public:
return Compare::equal(_value_at(column, row_num), value);
}
+ bool is_equal_to(const Self& to) const {
+ return has() && to.has() && Compare::equal(value, to.value);
+ }
+
bool set_if_greater(const Self& to, Arena& arena) {
if (to.has() && (!has() || Compare::greater(to.value, value))) {
set(to, arena);
@@ -301,6 +305,8 @@ public:
}
return _value_at(column, row_num) == _value();
}
+
+ bool is_equal_to(const Self& to) const { return has() && to.has() &&
_value() == to._value(); }
};
static_assert(sizeof(SingleValueDataString) ==
SingleValueDataString::AUTOMATIC_STORAGE_SIZE);
@@ -412,6 +418,10 @@ public:
bool is_equal_to(const IColumn& column, size_t row_num) const {
return has() && column.compare_at(row_num, 0, *_value, 1) == 0;
}
+
+ bool is_equal_to(const Self& to) const {
+ return has() && to.has() && _value->compare_at(0, 0, *to._value, 1) ==
0;
+ }
};
} // namespace doris
diff --git a/be/src/exprs/function/function_dict_get_many.cpp
b/be/src/exprs/function/function_dict_get_many.cpp
index e077041b98c..a2cb5a4d1f7 100644
--- a/be/src/exprs/function/function_dict_get_many.cpp
+++ b/be/src/exprs/function/function_dict_get_many.cpp
@@ -32,6 +32,7 @@
#include "exprs/function/dictionary_factory.h"
#include "exprs/function/function.h"
#include "exprs/function/simple_function_factory.h"
+#include "runtime/exec_env.h"
namespace doris {
diff --git a/be/test/exprs/aggregate/agg_count_at_min_max_test.cpp
b/be/test/exprs/aggregate/agg_count_at_min_max_test.cpp
new file mode 100644
index 00000000000..db0bc9eabf3
--- /dev/null
+++ b/be/test/exprs/aggregate/agg_count_at_min_max_test.cpp
@@ -0,0 +1,163 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <cstring>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "core/column/column.h"
+#include "core/column/column_string.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/types.h"
+#include "exprs/aggregate/aggregate_function.h"
+#include "exprs/aggregate/aggregate_function_simple_factory.h"
+#include "exprs/aggregate/single_value_data.h"
+
+namespace doris {
+
+void
register_aggregate_function_count_at_min_max(AggregateFunctionSimpleFactory&
factory);
+
+namespace {
+
+AggregateFunctionPtr create_function(AggregateFunctionSimpleFactory& factory,
+ const std::string& name, const DataTypes&
argument_types) {
+ auto function = factory.get(name, argument_types, nullptr, false, -1);
+ EXPECT_NE(function, nullptr);
+ return function;
+}
+
+void aggregate_int32(IAggregateFunction* function, Arena& arena, const
IColumn& column,
+ int64_t expected) {
+ std::unique_ptr<char[]> memory(new char[function->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ function->create(place);
+ const IColumn* columns[1] = {&column};
+ for (size_t row = 0; row < column.size(); ++row) {
+ function->add(place, columns, row, arena);
+ }
+ ColumnInt64 result;
+ function->insert_result_into(place, result);
+ EXPECT_EQ(expected, result.get_element(0));
+ function->destroy(place);
+}
+
+} // namespace
+
+TEST(CountAtMinMaxTest, Int32) {
+ AggregateFunctionSimpleFactory factory;
+ register_aggregate_function_count_at_min_max(factory);
+ DataTypes data_types = {std::make_shared<DataTypeInt32>()};
+
+ auto column = ColumnInt32::create();
+ for (int value : {1, 2, 1, 3, 1, 2}) {
+ column->insert_value(value);
+ }
+
+ Arena arena;
+ auto count_at_min = create_function(factory, "count_at_min", data_types);
+ aggregate_int32(count_at_min.get(), arena, *column, 3);
+
+ auto count_at_max = create_function(factory, "count_at_max", data_types);
+ aggregate_int32(count_at_max.get(), arena, *column, 1);
+}
+
+TEST(CountAtMinMaxTest, EmptyInput) {
+ AggregateFunctionSimpleFactory factory;
+ register_aggregate_function_count_at_min_max(factory);
+ DataTypes data_types = {std::make_shared<DataTypeInt32>()};
+ Arena arena;
+
+ auto column = ColumnInt32::create();
+ for (const auto& name : {"count_at_min", "count_at_max"}) {
+ auto function = create_function(factory, name, data_types);
+ aggregate_int32(function.get(), arena, *column, 0);
+ }
+}
+
+TEST(CountAtMinMaxTest, Merge) {
+ AggregateFunctionSimpleFactory factory;
+ register_aggregate_function_count_at_min_max(factory);
+ DataTypes data_types = {std::make_shared<DataTypeInt32>()};
+ Arena arena;
+
+ auto first = ColumnInt32::create();
+ auto second = ColumnInt32::create();
+ for (int value : {5, 1, 5}) {
+ first->insert_value(value);
+ }
+ for (int value : {5, 1, 9}) {
+ second->insert_value(value);
+ }
+
+ auto function = create_function(factory, "count_at_min", data_types);
+ std::unique_ptr<char[]> memory(new char[function->size_of_data() * 2]);
+ AggregateDataPtr lhs = memory.get();
+ AggregateDataPtr rhs = memory.get() + function->size_of_data();
+ function->create(lhs);
+ function->create(rhs);
+
+ const IColumn* first_columns[1] = {first.get()};
+ const IColumn* second_columns[1] = {second.get()};
+ for (size_t row = 0; row < first->size(); ++row) {
+ function->add(lhs, first_columns, row, arena);
+ }
+ for (size_t row = 0; row < second->size(); ++row) {
+ function->add(rhs, second_columns, row, arena);
+ }
+ function->merge(lhs, rhs, arena);
+
+ ColumnInt64 result;
+ function->insert_result_into(lhs, result);
+ EXPECT_EQ(2, result.get_element(0));
+ function->destroy(lhs);
+ function->destroy(rhs);
+}
+
+TEST(CountAtMinMaxTest, String) {
+ AggregateFunctionSimpleFactory factory;
+ register_aggregate_function_count_at_min_max(factory);
+ DataTypes data_types = {std::make_shared<DataTypeString>()};
+ Arena arena;
+
+ auto column = ColumnString::create();
+ for (const auto& value : {"b", "a", "b", "a", "c", "a"}) {
+ column->insert_data(value, strlen(value));
+ }
+
+ for (const auto& [name, expected] : {std::pair {"count_at_min", 3},
{"count_at_max", 1}}) {
+ auto function = create_function(factory, name, data_types);
+ std::unique_ptr<char[]> memory(new char[function->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ function->create(place);
+ const IColumn* columns[1] = {column.get()};
+ for (size_t row = 0; row < column->size(); ++row) {
+ function->add(place, columns, row, arena);
+ }
+ ColumnInt64 result;
+ function->insert_result_into(place, result);
+ EXPECT_EQ(expected, result.get_element(0));
+ function->destroy(place);
+ }
+}
+
+} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
index e20a45baf03..4aefb09b1e1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
@@ -36,6 +36,8 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.CollectSet;
import org.apache.doris.nereids.trees.expressions.functions.agg.Corr;
import org.apache.doris.nereids.trees.expressions.functions.agg.CorrWelford;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMax;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMin;
import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
import org.apache.doris.nereids.trees.expressions.functions.agg.CountMap;
import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
@@ -149,6 +151,8 @@ public class BuiltinAggregateFunctions implements
FunctionHelper {
agg(CorrWelford.class, "corr_welford"),
agg(ExponentialMovingAverage.class,
"exponential_moving_average"),
agg(Count.class, "count"),
+ agg(CountAtMin.class, "count_at_min"),
+ agg(CountAtMax.class, "count_at_max"),
agg(CountByEnum.class, "count_by_enum"),
agg(CountMap.class, "count_map"),
agg(Covar.class, "covar", "covar_pop"),
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
index 24990b7a72b..7e7e8f19c36 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExpressionBuilder.java
@@ -143,6 +143,8 @@ public class IvmAggExpressionBuilder {
return target.stateColumnName(IvmAggStateKey.SUM);
case MIN:
case MAX:
+ case MIN_COUNT:
+ case MAX_COUNT:
case BITMAP_UNION:
return IvmUtil.ivmAggHiddenColumnName(target.getOrdinal(),
slotKind.name());
default:
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExtremalProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExtremalProcessor.java
index 3954792fd1e..446466252a3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExtremalProcessor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExtremalProcessor.java
@@ -19,14 +19,19 @@ package org.apache.doris.mtmv.ivm.agg;
import org.apache.doris.mtmv.ivm.IvmException;
import org.apache.doris.mtmv.ivm.IvmFailureReason;
+import org.apache.doris.nereids.trees.expressions.Add;
import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.And;
import org.apache.doris.nereids.trees.expressions.CaseWhen;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.IsNull;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Not;
import org.apache.doris.nereids.trees.expressions.Or;
import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.Subtract;
import org.apache.doris.nereids.trees.expressions.WhenClause;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
@@ -35,26 +40,50 @@ import
org.apache.doris.nereids.trees.expressions.functions.scalar.If;
import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.types.BigIntType;
import org.apache.doris.nereids.util.TypeCoercionUtils;
import com.google.common.collect.ImmutableList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
/**
- * Shared logic for MIN/MAX.
+ * Shared delta/apply logic for MIN/MAX.
*
- * <p>The delta aggregate produces an insert-only extreme, a delete-only
extreme, and a non-NULL count delta. Apply can
- * merge inserts directly, but if a deleted row may have been the old visible
extreme, the assert guard forces fallback
- * to complete refresh because the new extreme cannot be derived from current
MV state plus delta alone.
+ * <p>Besides the current extreme value, the MV stores the number of rows
equal to that
+ * extreme ({@code MIN_COUNT}/{@code MAX_COUNT}). The delta aggregate produces
the extreme
+ * on the insert and delete sides together with the count of rows equal to
each side's
+ * extreme. Apply uses those counts to decide whether the old extreme survived:
+ *
+ * <ul>
+ * <li>An inserted value better than the old extreme becomes the new
extreme.</li>
+ * <li>A group whose old extreme is NULL starts from the insert-side
extreme, and its
+ * count is first reduced by deletes that cancel the same value (a join
can emit
+ * such a delete for a row that the same refresh inserted
elsewhere).</li>
+ * <li>Otherwise the old extreme survives while
+ * {@code oldCount + equal inserts - equal deletes > 0}.</li>
+ * <li>If the old extreme is exhausted and no inserted value is better, the
new extreme
+ * cannot be derived from the current state and delta; the assert guard
keeps the
+ * existing fallback to a complete refresh.</li>
+ * </ul>
+ *
+ * <p>The count state is what makes deleting a duplicate extreme incremental:
deleting one
+ * of several rows equal to the minimum no longer trips the fallback.
*/
abstract class IvmAggExtremalProcessor extends IvmAggFunctionProcessor {
+ private static final String INSERT_COUNT_SLOT = "EXTREME_INS_COUNT";
+ private static final String DELETE_COUNT_SLOT = "EXTREME_DEL_COUNT";
+
private final IvmAggFunctionKind functionKind;
+ private final IvmAggStateKey countStateKey;
private final String deleteSlotName;
- IvmAggExtremalProcessor(IvmAggFunctionKind functionKind, String
deleteSlotName) {
+ IvmAggExtremalProcessor(IvmAggFunctionKind functionKind, IvmAggStateKey
countStateKey,
+ String deleteSlotName) {
this.functionKind = functionKind;
+ this.countStateKey = countStateKey;
this.deleteSlotName = deleteSlotName;
}
@@ -65,7 +94,7 @@ abstract class IvmAggExtremalProcessor extends
IvmAggFunctionProcessor {
@Override
protected List<IvmAggStateKey> hiddenStateKeys(AggregateFunction function)
{
- return ImmutableList.of(IvmAggStateKey.COUNT);
+ return ImmutableList.of(IvmAggStateKey.COUNT, countStateKey);
}
@Override
@@ -78,6 +107,10 @@ abstract class IvmAggExtremalProcessor extends
IvmAggFunctionProcessor {
deleteDeltaColumnName(target, ctx)));
outputs.add(new Alias(new Sum(ctx.nonNullDeltaCountValue(arg,
dmlFactorSlot)),
ctx.deltaColumnName(target, IvmAggFunctionKind.COUNT)));
+ outputs.add(new
Alias(buildExtremeCountAggregate(ctx.insertOnlyValue(arg, dmlFactorSlot)),
+ ctx.transientDeltaColumnName(target, INSERT_COUNT_SLOT)));
+ outputs.add(new
Alias(buildExtremeCountAggregate(ctx.deleteOnlyValue(arg, dmlFactorSlot)),
+ ctx.transientDeltaColumnName(target, DELETE_COUNT_SLOT)));
}
@Override
@@ -87,16 +120,12 @@ abstract class IvmAggExtremalProcessor extends
IvmAggFunctionProcessor {
@Override
void mapApplyDeltaSlots(IvmAggTarget target, Map<String, Slot>
outputByName,
- Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots, Slot
deltaGroupCountSlot, IvmAggExpressionBuilder ctx) {
+ Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots, Slot
deltaGroupCountSlot,
+ IvmAggExpressionBuilder ctx) {
super.mapApplyDeltaSlots(target, outputByName, applyDeltaSlots,
deltaGroupCountSlot, ctx);
- String columnName = deleteDeltaColumnName(target, ctx);
- Slot slot = outputByName.get(columnName);
- if (slot == null) {
- throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
- "IVM agg delta rewrite failed to resolve delta output
slot: "
- + columnName + " for target " + target);
- }
- applyDeltaSlots.put(deleteSlotRef(target), slot);
+ resolveDeltaSlot(target, deleteSlotName, outputByName,
applyDeltaSlots, ctx);
+ resolveDeltaSlot(target, INSERT_COUNT_SLOT, outputByName,
applyDeltaSlots, ctx);
+ resolveDeltaSlot(target, DELETE_COUNT_SLOT, outputByName,
applyDeltaSlots, ctx);
}
@Override
@@ -109,48 +138,126 @@ abstract class IvmAggExtremalProcessor extends
IvmAggFunctionProcessor {
IvmAggExpressionBuilder ctx = applyContext.expressions();
Slot oldExtreme =
applyContext.rawMvSlot(target.getValueStateColumnName());
Expression deltaInsert = applyContext.deltaSlotValue(target,
functionKind);
- Expression deltaDel = applyContext.deltaSlotValue(target,
deleteSlotRef(target));
- Expression newCount = applyContext.buildNewHiddenCount(target);
+ Expression deltaDelete = applyContext.deltaSlotValue(target,
deleteSlotRef(target));
+ Expression deltaInsertCount = applyContext.deltaSlotValue(target,
insertCountSlotRef(target));
+ Expression deltaDeleteCount = applyContext.deltaSlotValue(target,
deleteCountSlotRef(target));
+ Slot oldCountSlot =
Objects.requireNonNull(target.getHiddenStateSlot(countStateKey),
+ "missing extreme count state for " + target);
+ Expression oldCount =
ctx.zeroIfNull(applyContext.rawMvSlot(oldCountSlot.getName()));
+ Expression newTotalCount = applyContext.buildNewHiddenCount(target);
+ Expression zero = new BigIntLiteral(0L);
+
+ // Keep every predicate non-NULL: an empty insert/delete side must
read as false, not
+ // as unknown, so the guard below is a plain boolean.
+ Expression insertedPresent = new Not(new IsNull(deltaInsert));
+ Expression deletedPresent = new Not(new IsNull(deltaDelete));
+ Expression oldExtremeNull = new IsNull(oldExtreme);
+ Expression oldExtremePresent = new Not(oldExtremeNull);
+ Expression insertedBetter = new And(insertedPresent,
insertedIsBetter(deltaInsert, oldExtreme));
+ Expression insertedEqual = new And(insertedPresent, new
EqualTo(deltaInsert, oldExtreme));
+ Expression deletedEqual = new And(deletedPresent, new
EqualTo(deltaDelete, oldExtreme));
+ Expression survivingOldCount = new Subtract(
+ new Add(oldCount, new If(insertedEqual, deltaInsertCount,
zero)),
+ new If(deletedEqual, deltaDeleteCount, zero));
+
+ // When the old extreme is NULL the new extreme comes from the insert
side. Its count
+ // is the insert-side count minus the deletes that cancel the same
value. That
+ // subtraction is only possible when the delete-side extreme is
exactly that value,
+ // or is farther from the extreme (so no delete touches it). A delete
closer to the
+ // extreme, or deletes that cancel every inserted row, leaves the
count unprovable
+ // and keeps the complete-refresh fallback.
+ Expression deleteEqualsInsert = new And(ImmutableList.of(
+ deletedPresent, insertedPresent, new EqualTo(deltaDelete,
deltaInsert)));
+ Expression insertExtremeNetCount = new Subtract(deltaInsertCount,
deltaDeleteCount);
+ Expression insertExtremeProvable = new And(insertedPresent, new
Or(ImmutableList.of(
+ new IsNull(deltaDelete),
+ new And(deletedPresent,
+ deletedExtremeIsBeyondInsert(deltaDelete,
deltaInsert)),
+ new And(deleteEqualsInsert, new
GreaterThan(insertExtremeNetCount, zero)))));
+ Expression oldNullGuard = new And(oldExtremeNull, new
Or(ImmutableList.of(
+ new EqualTo(newTotalCount, zero),
+ insertExtremeProvable)));
+
+ // The old extreme survives when no delete is below it and its
remaining count is
+ // positive. An inserted value that beats the old extreme also makes
the merge
+ // derivable without a complete refresh.
+ Expression oldPresentNoDeletes = new And(oldExtremePresent, new
IsNull(deltaDelete));
+ Expression oldPresentBoundary = new And(ImmutableList.of(
+ oldExtremePresent,
+ deletedPresent,
+ deletedExtremeIsAtLeastOld(deltaDelete, oldExtreme),
+ new Or(ImmutableList.of(
+ insertedBetter,
+ new GreaterThan(survivingOldCount, zero)))));
Expression guardCond = new Or(ImmutableList.of(
- new EqualTo(newCount, new BigIntLiteral(0L)),
- new IsNull(deltaDel),
- new IsNull(oldExtreme),
- deletedExtremeKeepsOldValueValid(deltaDel, oldExtreme,
deltaInsert)
- ));
+ new EqualTo(newTotalCount, zero),
+ oldNullGuard,
+ oldPresentNoDeletes,
+ oldPresentBoundary));
Expression guard = new AssertTrue(guardCond, new
StringLiteral(fallbackMessage()));
+
Expression newExtremeRaw = new CaseWhen(
ImmutableList.of(
- new WhenClause(new EqualTo(newCount, new
BigIntLiteral(0L)),
+ new WhenClause(new EqualTo(newTotalCount, zero),
new NullLiteral(oldExtreme.getDataType())),
new WhenClause(new IsNull(oldExtreme), deltaInsert),
- new WhenClause(new IsNull(deltaInsert), oldExtreme)
- ),
- mergeOldAndInsertedExtreme(oldExtreme, deltaInsert)
- );
+ new WhenClause(insertedBetter, deltaInsert)),
+ oldExtreme);
Expression newExtremeGuarded = new If(guard, newExtremeRaw,
new NullLiteral(newExtremeRaw.getDataType()));
- applyContext.putFinalExpression(target,
target.getHiddenStateSlot(IvmAggStateKey.COUNT).getName(), newCount);
+ Expression insertExtremeCount = new If(
+ deleteEqualsInsert, insertExtremeNetCount, deltaInsertCount);
+ Expression newCountRaw = new CaseWhen(
+ ImmutableList.of(
+ new WhenClause(new EqualTo(newTotalCount, zero), zero),
+ new WhenClause(oldExtremeNull, insertExtremeCount),
+ new WhenClause(insertedBetter, deltaInsertCount)),
+ survivingOldCount);
+ Expression newCountGuarded = new If(guard, newCountRaw,
+ new NullLiteral(BigIntType.INSTANCE));
+
+ applyContext.putFinalExpression(target, oldCountSlot.getName(),
newCountGuarded);
+ applyContext.putFinalExpression(target,
target.getHiddenStateSlot(IvmAggStateKey.COUNT).getName(),
+ newTotalCount);
applyContext.putFinalExpression(target,
target.getVisibleSlot().getName(),
- new If(ctx.isPositive(newCount),
-
TypeCoercionUtils.castIfNotMatchType(newExtremeGuarded,
target.getVisibleSlot().getDataType()),
+ new If(ctx.isPositive(newTotalCount),
+ TypeCoercionUtils.castIfNotMatchType(newExtremeGuarded,
+ target.getVisibleSlot().getDataType()),
new
NullLiteral(target.getVisibleSlot().getDataType())));
}
+ private void resolveDeltaSlot(IvmAggTarget target, String slotName,
Map<String, Slot> outputByName,
+ Map<IvmAggDeltaSlotRef, Slot> applyDeltaSlots,
IvmAggExpressionBuilder ctx) {
+ String columnName = ctx.transientDeltaColumnName(target, slotName);
+ Slot slot = outputByName.get(columnName);
+ if (slot == null) {
+ throw new IvmException(IvmFailureReason.PLAN_REWRITE_FAILED,
+ "IVM agg delta rewrite failed to resolve delta output
slot: "
+ + columnName + " for target " + target);
+ }
+ applyDeltaSlots.put(new IvmAggDeltaSlotRef(target.getOrdinal(),
slotName), slot);
+ }
+
/** Builds MIN or MAX over a filtered insert/delete expression. */
protected abstract Expression buildExtremeAggregate(Expression input);
+ /** Builds COUNT_AT_MIN or COUNT_AT_MAX over a filtered insert/delete
expression. */
+ protected abstract Expression buildExtremeCountAggregate(Expression input);
+
+ /** True when an inserted value becomes the new extreme. */
+ protected abstract Expression insertedIsBetter(Expression deltaInsert,
Slot oldExtreme);
+
+ /** True when the deleted extreme does not fall below/above the old
extreme. */
+ protected abstract Expression deletedExtremeIsAtLeastOld(Expression
deltaDelete, Slot oldExtreme);
+
/**
- * Returns true when the old visible extreme is still deterministically
valid after deletes,
- * either because no delete hit the boundary point, or because the insert
side provides a
- * value that equals or improves the old extreme position.
+ * True when the delete-side extreme is farther from the aggregate extreme
than the
+ * insert-side extreme, so no deleted row can cancel the insert-side
extreme.
*/
- protected abstract Expression deletedExtremeKeepsOldValueValid(Expression
deltaDeletedExtreme,
- Slot oldExtreme, Expression deltaInsertExtreme);
-
- /** Merges the old visible extreme with the insert-only delta extreme. */
- protected abstract Expression mergeOldAndInsertedExtreme(Slot oldExtreme,
Expression deltaInsertExtreme);
+ protected abstract Expression deletedExtremeIsBeyondInsert(
+ Expression deltaDelete, Expression deltaInsert);
/** Assertion message used when the delta is not sufficient and complete
refresh is required. */
protected abstract String fallbackMessage();
@@ -159,6 +266,14 @@ abstract class IvmAggExtremalProcessor extends
IvmAggFunctionProcessor {
return new IvmAggDeltaSlotRef(target.getOrdinal(), deleteSlotName);
}
+ private IvmAggDeltaSlotRef insertCountSlotRef(IvmAggTarget target) {
+ return new IvmAggDeltaSlotRef(target.getOrdinal(), INSERT_COUNT_SLOT);
+ }
+
+ private IvmAggDeltaSlotRef deleteCountSlotRef(IvmAggTarget target) {
+ return new IvmAggDeltaSlotRef(target.getOrdinal(), DELETE_COUNT_SLOT);
+ }
+
private String deleteDeltaColumnName(IvmAggTarget target,
IvmAggExpressionBuilder ctx) {
return ctx.transientDeltaColumnName(target, deleteSlotName);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
index b7f216b2a32..801d6c9abd2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionKind.java
@@ -29,6 +29,8 @@ public enum IvmAggFunctionKind {
AVG,
MIN,
MAX,
+ MIN_COUNT,
+ MAX_COUNT,
BITMAP_UNION,
BITMAP_UNION_COUNT,
ARRAY_AGG,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
index 2466f90b8a9..18ab268cdc0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java
@@ -27,6 +27,8 @@ import org.apache.doris.nereids.trees.expressions.Slot;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnion;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMax;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMin;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import com.google.common.collect.ImmutableList;
@@ -155,6 +157,10 @@ public abstract class IvmAggFunctionProcessor {
return new Sum(function.child(0));
case BITMAP_UNION:
return new BitmapUnion(function.child(0));
+ case MIN_COUNT:
+ return new CountAtMin(function.child(0));
+ case MAX_COUNT:
+ return new CountAtMax(function.child(0));
default:
throw new IllegalArgumentException("Unsupported IVM aggregate
state: " + stateKey);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessor.java
index 9d1edf7ea16..dad7d15430a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessor.java
@@ -19,18 +19,18 @@ package org.apache.doris.mtmv.ivm.agg;
import org.apache.doris.mtmv.ivm.IvmFailureClassifier;
import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.GreaterThanEqual;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.LessThan;
-import org.apache.doris.nereids.trees.expressions.Or;
+import org.apache.doris.nereids.trees.expressions.LessThanEqual;
import org.apache.doris.nereids.trees.expressions.Slot;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMax;
import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.Greatest;
-/** MAX-specific extreme comparisons and merge expression. */
+/** MAX-specific extreme comparisons and delta state. */
class IvmAggMaxProcessor extends IvmAggExtremalProcessor {
IvmAggMaxProcessor() {
- super(IvmAggFunctionKind.MAX, "DELMAX");
+ super(IvmAggFunctionKind.MAX, IvmAggStateKey.MAX_COUNT, "DELMAX");
}
@Override
@@ -44,16 +44,23 @@ class IvmAggMaxProcessor extends IvmAggExtremalProcessor {
}
@Override
- protected Expression deletedExtremeKeepsOldValueValid(Expression
deltaDeletedExtreme,
- Slot oldExtreme, Expression deltaInsertExtreme) {
- return new Or(
- new LessThan(deltaDeletedExtreme, oldExtreme),
- new GreaterThanEqual(deltaInsertExtreme, oldExtreme));
+ protected Expression buildExtremeCountAggregate(Expression input) {
+ return new CountAtMax(input);
}
@Override
- protected Expression mergeOldAndInsertedExtreme(Slot oldExtreme,
Expression deltaInsertExtreme) {
- return new Greatest(oldExtreme, deltaInsertExtreme);
+ protected Expression insertedIsBetter(Expression deltaInsert, Slot
oldExtreme) {
+ return new GreaterThan(deltaInsert, oldExtreme);
+ }
+
+ @Override
+ protected Expression deletedExtremeIsAtLeastOld(Expression deltaDelete,
Slot oldExtreme) {
+ return new LessThanEqual(deltaDelete, oldExtreme);
+ }
+
+ @Override
+ protected Expression deletedExtremeIsBeyondInsert(Expression deltaDelete,
Expression deltaInsert) {
+ return new LessThan(deltaDelete, deltaInsert);
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessor.java
index d1b6434907e..30cd23e01ac 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessor.java
@@ -20,17 +20,17 @@ package org.apache.doris.mtmv.ivm.agg;
import org.apache.doris.mtmv.ivm.IvmFailureClassifier;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.GreaterThan;
-import org.apache.doris.nereids.trees.expressions.LessThanEqual;
-import org.apache.doris.nereids.trees.expressions.Or;
+import org.apache.doris.nereids.trees.expressions.GreaterThanEqual;
+import org.apache.doris.nereids.trees.expressions.LessThan;
import org.apache.doris.nereids.trees.expressions.Slot;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMin;
import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.Least;
-/** MIN-specific extreme comparisons and merge expression. */
+/** MIN-specific extreme comparisons and delta state. */
class IvmAggMinProcessor extends IvmAggExtremalProcessor {
IvmAggMinProcessor() {
- super(IvmAggFunctionKind.MIN, "DELMIN");
+ super(IvmAggFunctionKind.MIN, IvmAggStateKey.MIN_COUNT, "DELMIN");
}
@Override
@@ -44,16 +44,23 @@ class IvmAggMinProcessor extends IvmAggExtremalProcessor {
}
@Override
- protected Expression deletedExtremeKeepsOldValueValid(Expression
deltaDeletedExtreme,
- Slot oldExtreme, Expression deltaInsertExtreme) {
- return new Or(
- new GreaterThan(deltaDeletedExtreme, oldExtreme),
- new LessThanEqual(deltaInsertExtreme, oldExtreme));
+ protected Expression buildExtremeCountAggregate(Expression input) {
+ return new CountAtMin(input);
}
@Override
- protected Expression mergeOldAndInsertedExtreme(Slot oldExtreme,
Expression deltaInsertExtreme) {
- return new Least(oldExtreme, deltaInsertExtreme);
+ protected Expression insertedIsBetter(Expression deltaInsert, Slot
oldExtreme) {
+ return new LessThan(deltaInsert, oldExtreme);
+ }
+
+ @Override
+ protected Expression deletedExtremeIsAtLeastOld(Expression deltaDelete,
Slot oldExtreme) {
+ return new GreaterThanEqual(deltaDelete, oldExtreme);
+ }
+
+ @Override
+ protected Expression deletedExtremeIsBeyondInsert(Expression deltaDelete,
Expression deltaInsert) {
+ return new GreaterThan(deltaDelete, deltaInsert);
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggStateKey.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggStateKey.java
index de83a22e1b3..76dfb3b303a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggStateKey.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggStateKey.java
@@ -21,5 +21,7 @@ package org.apache.doris.mtmv.ivm.agg;
public enum IvmAggStateKey {
COUNT,
SUM,
- BITMAP_UNION
+ BITMAP_UNION,
+ MIN_COUNT,
+ MAX_COUNT
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMax.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMax.java
new file mode 100644
index 00000000000..9d504a78ba7
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMax.java
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.agg;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DecimalV2Type;
+import org.apache.doris.nereids.types.DecimalV3Type;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * count_at_max(expr): the number of input rows whose value equals MAX(expr).
+ *
+ * <p>NULL inputs are ignored. An empty input, or an input whose values are
all NULL, returns 0.
+ * This is a Doris extension; it is equivalent to
+ * {@code COUNT(*) KEEP (DENSE_RANK LAST ORDER BY expr)} in Oracle.
+ */
+public class CountAtMax extends NotNullableAggregateFunction
+ implements UnaryExpression, CustomSignature,
NullIgnoringAggregateFunction {
+
+ public CountAtMax(Expression arg) {
+ this(false, false, arg);
+ }
+
+ public CountAtMax(boolean distinct, Expression arg) {
+ this(distinct, false, arg);
+ }
+
+ private CountAtMax(boolean distinct, boolean alwaysNullable, Expression
arg) {
+ super("count_at_max", distinct, false, arg);
+ }
+
+ /** constructor for withChildren and reuse signature */
+ private CountAtMax(AggregateFunctionParams functionParams) {
+ super(functionParams);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ if (distinct) {
+ throw new AnalysisException("COUNT_AT_MAX does not support
DISTINCT");
+ }
+ if (getArgumentType(0).isOnlyMetricType() &&
!getArgumentType(0).isArrayType()) {
+ throw new AnalysisException(Type.OnlyMetricTypeErrorMsg);
+ }
+ }
+
+ @Override
+ public void checkLegalityAfterRewrite() {
+ checkLegalityBeforeTypeCoercion();
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ DataType dataType = getArgument(0).getDataType();
+ if (dataType instanceof DecimalV2Type) {
+ dataType = DecimalV3Type.forType(dataType);
+ }
+ return FunctionSignature.ret(BigIntType.INSTANCE).args(dataType);
+ }
+
+ @Override
+ protected List<DataType> intermediateTypes() {
+ DataType dataType = getArgument(0).getDataType();
+ if (dataType instanceof DecimalV2Type) {
+ dataType = DecimalV3Type.forType(dataType);
+ }
+ return ImmutableList.of(dataType, BigIntType.INSTANCE);
+ }
+
+ @Override
+ public Expression resultForEmptyInput() {
+ return new BigIntLiteral(0);
+ }
+
+ @Override
+ public CountAtMax withDistinctAndChildren(boolean distinct,
List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new CountAtMax(getFunctionParams(distinct, children));
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitCountAtMax(this, context);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMin.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMin.java
new file mode 100644
index 00000000000..ad84f82afb3
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CountAtMin.java
@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.agg;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DecimalV2Type;
+import org.apache.doris.nereids.types.DecimalV3Type;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * count_at_min(expr): the number of input rows whose value equals MIN(expr).
+ *
+ * <p>NULL inputs are ignored. An empty input, or an input whose values are
all NULL, returns 0.
+ * This is a Doris extension; it is equivalent to
+ * {@code COUNT(*) KEEP (DENSE_RANK FIRST ORDER BY expr)} in Oracle.
+ */
+public class CountAtMin extends NotNullableAggregateFunction
+ implements UnaryExpression, CustomSignature,
NullIgnoringAggregateFunction {
+
+ public CountAtMin(Expression arg) {
+ this(false, false, arg);
+ }
+
+ public CountAtMin(boolean distinct, Expression arg) {
+ this(distinct, false, arg);
+ }
+
+ private CountAtMin(boolean distinct, boolean alwaysNullable, Expression
arg) {
+ super("count_at_min", distinct, false, arg);
+ }
+
+ /** constructor for withChildren and reuse signature */
+ private CountAtMin(AggregateFunctionParams functionParams) {
+ super(functionParams);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ if (distinct) {
+ throw new AnalysisException("COUNT_AT_MIN does not support
DISTINCT");
+ }
+ if (getArgumentType(0).isOnlyMetricType() &&
!getArgumentType(0).isArrayType()) {
+ throw new AnalysisException(Type.OnlyMetricTypeErrorMsg);
+ }
+ }
+
+ @Override
+ public void checkLegalityAfterRewrite() {
+ checkLegalityBeforeTypeCoercion();
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ DataType dataType = getArgument(0).getDataType();
+ if (dataType instanceof DecimalV2Type) {
+ dataType = DecimalV3Type.forType(dataType);
+ }
+ return FunctionSignature.ret(BigIntType.INSTANCE).args(dataType);
+ }
+
+ @Override
+ protected List<DataType> intermediateTypes() {
+ DataType dataType = getArgument(0).getDataType();
+ if (dataType instanceof DecimalV2Type) {
+ dataType = DecimalV3Type.forType(dataType);
+ }
+ return ImmutableList.of(dataType, BigIntType.INSTANCE);
+ }
+
+ @Override
+ public Expression resultForEmptyInput() {
+ return new BigIntLiteral(0);
+ }
+
+ @Override
+ public CountAtMin withDistinctAndChildren(boolean distinct,
List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new CountAtMin(getFunctionParams(distinct, children));
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitCountAtMin(this, context);
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
index 5a77fdf9d7a..40d1c4311cd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
@@ -36,6 +36,8 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.CollectSet;
import org.apache.doris.nereids.trees.expressions.functions.agg.Corr;
import org.apache.doris.nereids.trees.expressions.functions.agg.CorrWelford;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMax;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMin;
import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
import org.apache.doris.nereids.trees.expressions.functions.agg.CovarSamp;
@@ -188,6 +190,14 @@ public interface AggregateFunctionVisitor<R, C> {
return visitAggregateFunction(count, context);
}
+ default R visitCountAtMin(CountAtMin countAtMin, C context) {
+ return visitAggregateFunction(countAtMin, context);
+ }
+
+ default R visitCountAtMax(CountAtMax countAtMax, C context) {
+ return visitAggregateFunction(countAtMax, context);
+ }
+
default R visitCountByEnum(CountByEnum count, C context) {
return visitAggregateFunction(count, context);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
index 579af9f450a..4a433111459 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
@@ -439,7 +439,7 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
// SUM, AVG's hidden COUNT dedups with SUM's hidden COUNT, and MIN/MAX
reuse visible cnt_name as
// their hidden COUNT, so shared state removes two duplicate COUNT
delta outputs.
List<String> outputNames =
deltaAgg.getOutput().stream().map(Slot::getName).collect(Collectors.toList());
- Assertions.assertEquals(10, outputNames.size());
+ Assertions.assertEquals(14, outputNames.size());
Assertions.assertTrue(outputNames.contains(Column.IVM_DELTA_GROUP_COUNT_COL));
// ordinal 0: COUNT(name) -> hidden COUNT (kept)
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"COUNT")));
@@ -450,14 +450,18 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(2,
"SUM")));
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(2,
"COUNT")));
Assertions.assertTrue(outputNames.contains("sum_id"));
- // ordinal 3: MIN(name) -> MIN + DELMIN kept; COUNT absorbed by
visible cnt_name
+ // ordinal 3: MIN(name) -> MIN + DELMIN + insert/delete extreme
counts; COUNT absorbed by visible cnt_name
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(3,
"MIN")));
Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(3,
"DELMIN")));
+ Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(3,
"EXTREME_INS_COUNT")));
+ Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(3,
"EXTREME_DEL_COUNT")));
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(3,
"COUNT")));
Assertions.assertTrue(outputNames.contains("cnt_name"));
- // ordinal 4: MAX(name) -> MAX + DELMAX kept; COUNT absorbed by
visible cnt_name
+ // ordinal 4: MAX(name) -> MAX + DELMAX + insert/delete extreme
counts; COUNT absorbed by visible cnt_name
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(4,
"MAX")));
Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(4,
"DELMAX")));
+ Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(4,
"EXTREME_INS_COUNT")));
+ Assertions.assertTrue(outputNames.contains(transientDeltaColumnName(4,
"EXTREME_DEL_COUNT")));
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(4,
"COUNT")));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmNormalizeUniqKeysMTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmNormalizeUniqKeysMTMVTest.java
index 80b9f5fa505..b1c66199ace 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmNormalizeUniqKeysMTMVTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmNormalizeUniqKeysMTMVTest.java
@@ -798,9 +798,13 @@ class IvmNormalizeUniqKeysMTMVTest {
Pair.of(new Count(nameSlot), concat(groupCount,
IvmUtil.ivmAggHiddenColumnName(0, "COUNT"))),
Pair.of(new Min(nameSlot), concat(groupCount,
- IvmUtil.ivmAggHiddenColumnName(0, "COUNT"),
IvmUtil.ivmAggHiddenColumnName(0, "MIN"))),
+ IvmUtil.ivmAggHiddenColumnName(0, "COUNT"),
+ IvmUtil.ivmAggHiddenColumnName(0, "MIN_COUNT"),
+ IvmUtil.ivmAggHiddenColumnName(0, "MIN"))),
Pair.of(new Max(nameSlot), concat(groupCount,
- IvmUtil.ivmAggHiddenColumnName(0, "COUNT"),
IvmUtil.ivmAggHiddenColumnName(0, "MAX"))),
+ IvmUtil.ivmAggHiddenColumnName(0, "COUNT"),
+ IvmUtil.ivmAggHiddenColumnName(0, "MAX_COUNT"),
+ IvmUtil.ivmAggHiddenColumnName(0, "MAX"))),
Pair.of(new CollectList(nameSlot), concat(groupCount,
IvmUtil.ivmAggHiddenColumnName(0, "COLLECT_LIST"))),
Pair.of(new BitmapUnion(new BitmapFromString(nameSlot)),
concat(groupCount,
@@ -1068,7 +1072,7 @@ class IvmNormalizeUniqKeysMTMVTest {
JobContext jobContext = newJobContextForRoot(agg, true);
Plan result = new IvmNormalizeUniqKeysMTMV().normalize(agg,
jobContext);
- // Normalization should succeed; no hidden MIN column — only hidden
COUNT
+ // Normalization should succeed; no hidden MIN column, but COUNT and
MIN_COUNT state columns.
Assertions.assertInstanceOf(LogicalProject.class, result);
IvmAggMeta aggMeta =
jobContext.getCascadesContext().getIvmRewriteResult().get().getAggMeta();
Assertions.assertNotNull(aggMeta);
@@ -1076,12 +1080,14 @@ class IvmNormalizeUniqKeysMTMVTest {
IvmAggTarget target = aggMeta.getAggTargets().get(0);
Assertions.assertEquals(IvmAggFunctionKind.MIN,
target.getFunctionKind());
Assertions.assertNotNull(target.getHiddenStateSlot(IvmAggStateKey.COUNT));
+
Assertions.assertNotNull(target.getHiddenStateSlot(IvmAggStateKey.MIN_COUNT));
LogicalProject<?> topProject = (LogicalProject<?>) result;
Set<String> outputNames = topProject.getOutput().stream()
.map(Slot::getName).collect(Collectors.toSet());
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"MIN")));
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"COUNT")));
+
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"MIN_COUNT")));
}
@Test
@@ -1104,12 +1110,14 @@ class IvmNormalizeUniqKeysMTMVTest {
IvmAggTarget target = aggMeta.getAggTargets().get(0);
Assertions.assertEquals(IvmAggFunctionKind.MAX,
target.getFunctionKind());
Assertions.assertNotNull(target.getHiddenStateSlot(IvmAggStateKey.COUNT));
+
Assertions.assertNotNull(target.getHiddenStateSlot(IvmAggStateKey.MAX_COUNT));
LogicalProject<?> topProject = (LogicalProject<?>) result;
Set<String> outputNames = topProject.getOutput().stream()
.map(Slot::getName).collect(Collectors.toSet());
Assertions.assertFalse(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"MAX")));
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"COUNT")));
+
Assertions.assertTrue(outputNames.contains(IvmUtil.ivmAggHiddenColumnName(0,
"MAX_COUNT")));
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessorTest.java
index 928a82e1383..c79c064c26c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMaxProcessorTest.java
@@ -17,13 +17,15 @@
package org.apache.doris.mtmv.ivm.agg;
+import org.apache.doris.mtmv.ivm.IvmUtil;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.LessThan;
+import org.apache.doris.nereids.trees.expressions.LessThanEqual;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMax;
import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.Greatest;
import org.apache.doris.nereids.types.BigIntType;
import org.apache.doris.nereids.types.IntegerType;
@@ -41,22 +43,34 @@ class IvmAggMaxProcessorTest extends
IvmAggProcessorTestBase {
IvmAggMaxProcessor processor = new IvmAggMaxProcessor();
Assertions.assertTrue(processor.supportsOriginalFunction(new
Max(value)));
Assertions.assertEquals(IvmAggFunctionKind.MAX,
processor.handledFunctionKind());
- Assertions.assertEquals(ImmutableList.of(IvmAggStateKey.COUNT),
processor.hiddenStateKeys(new Max(value)));
+ Assertions.assertEquals(ImmutableList.of(IvmAggStateKey.COUNT,
IvmAggStateKey.MAX_COUNT),
+ processor.hiddenStateKeys(new Max(value)));
IvmAggTarget target = target(3, IvmAggFunctionKind.MAX, "max_v",
IntegerType.INSTANCE,
- ImmutableMap.of(IvmAggStateKey.COUNT, slot("__ivm_count",
BigIntType.INSTANCE)), valueArg());
+ ImmutableMap.of(IvmAggStateKey.COUNT, slot("__ivm_count",
BigIntType.INSTANCE),
+ IvmAggStateKey.MAX_COUNT,
slot(IvmUtil.ivmAggHiddenColumnName(3, "MAX_COUNT"),
+ BigIntType.INSTANCE)),
+ valueArg());
List<NamedExpression> outputs = deltaOutputs(processor, target);
- Assertions.assertEquals(3, outputs.size());
+ Assertions.assertEquals(5, outputs.size());
Assertions.assertTrue(outputs.get(0).child(0) instanceof Max);
Assertions.assertTrue(outputs.get(1).getName().contains("DELMAX"));
Assertions.assertTrue(outputs.get(2).child(0) instanceof Sum);
+ Assertions.assertTrue(outputs.get(3).child(0) instanceof CountAtMax);
+ Assertions.assertTrue(outputs.get(4).child(0) instanceof CountAtMax);
Map<String, Expression> finalByName = apply(processor, target,
- ImmutableList.of(slot("max_v", IntegerType.INSTANCE),
slot("__ivm_count", BigIntType.INSTANCE)),
+ ImmutableList.of(slot("max_v", IntegerType.INSTANCE),
+ slot("__ivm_count", BigIntType.INSTANCE),
+ slot(IvmUtil.ivmAggHiddenColumnName(3, "MAX_COUNT"),
BigIntType.INSTANCE)),
mappedDeltaSlots(processor, target, outputs),
slot("delta_group_count", BigIntType.INSTANCE));
Expression visible = finalByName.get("max_v");
Assertions.assertTrue(visible.anyMatch(node -> node instanceof
AssertTrue));
- Assertions.assertTrue(visible.anyMatch(node -> node instanceof
LessThan));
- Assertions.assertTrue(visible.anyMatch(node -> node instanceof
Greatest));
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
LessThanEqual));
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
CaseWhen));
+
Assertions.assertTrue(finalByName.containsKey(IvmUtil.ivmAggHiddenColumnName(3,
"MAX_COUNT")));
+ Expression hiddenCount =
finalByName.get(IvmUtil.ivmAggHiddenColumnName(3, "MAX_COUNT"));
+ Assertions.assertTrue(containsSlotSubtract(hiddenCount,
"EXTREME_INS_COUNT", "EXTREME_DEL_COUNT"),
+ "MAX count initialization must subtract matching deletes");
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessorTest.java
index 67924bfa566..768e933842d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggMinProcessorTest.java
@@ -17,13 +17,15 @@
package org.apache.doris.mtmv.ivm.agg;
+import org.apache.doris.mtmv.ivm.IvmUtil;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.GreaterThanEqual;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CountAtMin;
import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
-import org.apache.doris.nereids.trees.expressions.functions.scalar.Least;
import org.apache.doris.nereids.types.BigIntType;
import org.apache.doris.nereids.types.IntegerType;
@@ -41,22 +43,34 @@ class IvmAggMinProcessorTest extends
IvmAggProcessorTestBase {
IvmAggMinProcessor processor = new IvmAggMinProcessor();
Assertions.assertTrue(processor.supportsOriginalFunction(new
Min(value)));
Assertions.assertEquals(IvmAggFunctionKind.MIN,
processor.handledFunctionKind());
- Assertions.assertEquals(ImmutableList.of(IvmAggStateKey.COUNT),
processor.hiddenStateKeys(new Min(value)));
+ Assertions.assertEquals(ImmutableList.of(IvmAggStateKey.COUNT,
IvmAggStateKey.MIN_COUNT),
+ processor.hiddenStateKeys(new Min(value)));
IvmAggTarget target = target(2, IvmAggFunctionKind.MIN, "min_v",
IntegerType.INSTANCE,
- ImmutableMap.of(IvmAggStateKey.COUNT, slot("__ivm_count",
BigIntType.INSTANCE)), valueArg());
+ ImmutableMap.of(IvmAggStateKey.COUNT, slot("__ivm_count",
BigIntType.INSTANCE),
+ IvmAggStateKey.MIN_COUNT,
slot(IvmUtil.ivmAggHiddenColumnName(2, "MIN_COUNT"),
+ BigIntType.INSTANCE)),
+ valueArg());
List<NamedExpression> outputs = deltaOutputs(processor, target);
- Assertions.assertEquals(3, outputs.size());
+ Assertions.assertEquals(5, outputs.size());
Assertions.assertTrue(outputs.get(0).child(0) instanceof Min);
Assertions.assertTrue(outputs.get(1).getName().contains("DELMIN"));
Assertions.assertTrue(outputs.get(2).child(0) instanceof Sum);
+ Assertions.assertTrue(outputs.get(3).child(0) instanceof CountAtMin);
+ Assertions.assertTrue(outputs.get(4).child(0) instanceof CountAtMin);
Map<String, Expression> finalByName = apply(processor, target,
- ImmutableList.of(slot("min_v", IntegerType.INSTANCE),
slot("__ivm_count", BigIntType.INSTANCE)),
+ ImmutableList.of(slot("min_v", IntegerType.INSTANCE),
+ slot("__ivm_count", BigIntType.INSTANCE),
+ slot(IvmUtil.ivmAggHiddenColumnName(2, "MIN_COUNT"),
BigIntType.INSTANCE)),
mappedDeltaSlots(processor, target, outputs),
slot("delta_group_count", BigIntType.INSTANCE));
Expression visible = finalByName.get("min_v");
Assertions.assertTrue(visible.anyMatch(node -> node instanceof
AssertTrue));
- Assertions.assertTrue(visible.anyMatch(node -> node instanceof
GreaterThan));
- Assertions.assertTrue(visible.anyMatch(node -> node instanceof Least));
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
GreaterThanEqual));
+ Assertions.assertTrue(visible.anyMatch(node -> node instanceof
CaseWhen));
+
Assertions.assertTrue(finalByName.containsKey(IvmUtil.ivmAggHiddenColumnName(2,
"MIN_COUNT")));
+ Expression hiddenCount =
finalByName.get(IvmUtil.ivmAggHiddenColumnName(2, "MIN_COUNT"));
+ Assertions.assertTrue(containsSlotSubtract(hiddenCount,
"EXTREME_INS_COUNT", "EXTREME_DEL_COUNT"),
+ "MIN count initialization must subtract matching deletes");
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggProcessorTestBase.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggProcessorTestBase.java
index 788202d8c04..b86da454861 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggProcessorTestBase.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/agg/IvmAggProcessorTestBase.java
@@ -21,6 +21,7 @@ import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.Subtract;
import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnion;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import org.apache.doris.nereids.trees.expressions.functions.scalar.BitmapCount;
@@ -102,6 +103,28 @@ abstract class IvmAggProcessorTestBase {
return ((Slot) expression).getName();
}
+ /**
+ * True when the expression subtracts the two delta count slots, which is
what the
+ * old-NULL branch must do to cancel deletes that match the insert-side
extreme.
+ */
+ protected boolean containsSlotSubtract(Expression expression, String
leftName, String rightName) {
+ if (expression instanceof Subtract) {
+ Expression left = expression.child(0);
+ Expression right = expression.child(1);
+ if (left instanceof Slot && right instanceof Slot
+ && ((Slot) left).getName().contains(leftName)
+ && ((Slot) right).getName().contains(rightName)) {
+ return true;
+ }
+ }
+ for (Expression child : expression.children()) {
+ if (containsSlotSubtract(child, leftName, rightName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
protected void assertTwoBitmapDeltaOutputs(List<NamedExpression> outputs) {
Assertions.assertEquals(2, outputs.size());
Assertions.assertTrue(outputs.get(0).child(0) instanceof BitmapUnion);
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_1.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_1.out
index ace0a73ecdc..e0761b78a2a 100644
--- a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_1.out
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_1.out
@@ -26,6 +26,7 @@ k int Yes false \N NONE
m2 bigint Yes false \N NONE
__DORIS_IVM_AGG_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_COUNT_COL__ bigint No false \N NONE
+__DORIS_IVM_AGG_0_MIN_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_MIN_COL__ int Yes false \N NONE
__DORIS_DELETE_SIGN__ tinyint No false 0 NONE
__DORIS_VERSION_COL__ bigint No false 0 NONE
@@ -37,6 +38,7 @@ k int Yes false \N NONE
m1 bigint Yes false \N NONE
__DORIS_IVM_AGG_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_COUNT_COL__ bigint No false \N NONE
+__DORIS_IVM_AGG_0_MAX_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_MAX_COL__ int Yes false \N NONE
__DORIS_DELETE_SIGN__ tinyint No false 0 NONE
__DORIS_VERSION_COL__ bigint No false 0 NONE
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_2.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_2.out
index 0c772bad147..02316300cce 100644
--- a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_2.out
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_expr_over_agg_2.out
@@ -165,6 +165,7 @@ __DORIS_IVM_ROW_ID_COL__ tinyint No true \N
m varchar(65533) Yes false \N NONE
__DORIS_IVM_AGG_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_COUNT_COL__ bigint No false \N NONE
+__DORIS_IVM_AGG_0_MIN_COUNT_COL__ bigint No false \N NONE
__DORIS_DELETE_SIGN__ tinyint No false 0 NONE
__DORIS_VERSION_COL__ bigint No false 0 NONE
__DORIS_SEQUENCE_COL__ bigint Yes false \N NONE
@@ -251,6 +252,7 @@ k int Yes false \N NONE
m2 bigint Yes false \N NONE
__DORIS_IVM_AGG_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_COUNT_COL__ bigint No false \N NONE
+__DORIS_IVM_AGG_0_MIN_COUNT_COL__ bigint No false \N NONE
__DORIS_IVM_AGG_0_MIN_COL__ int Yes false \N NONE
__DORIS_DELETE_SIGN__ tinyint No false 0 NONE
__DORIS_VERSION_COL__ bigint No false 0 NONE
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count.out
new file mode 100644
index 00000000000..65ee6c479c4
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count.out
@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !initial --
+a 10 30 3
+b 5 5 3
+
+-- !after_duplicate_min_delete --
+a 10 30 2
+b 5 5 3
+
+-- !after_duplicate_max_delete --
+a 10 30 2
+b 5 5 2
+
+-- !after_last_min_delete --
+a 30 30 1
+b 5 5 2
+
+-- !after_smaller_insert --
+a 7 30 3
+b 5 5 2
+
+-- !source --
+a 7 30 3
+b 5 5 2
+
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count_join.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count_join.out
new file mode 100644
index 00000000000..51b58359e29
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_minmax_count_join.out
@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !initial --
+\N
+
+-- !after_canceled_delete --
+1
+
+-- !after_boundary --
+2
+
+-- !source --
+2
+
diff --git a/regression-test/data/query_p0/aggregate/count_at_min_max.out
b/regression-test/data/query_p0/aggregate/count_at_min_max.out
new file mode 100644
index 00000000000..86f8c34874b
--- /dev/null
+++ b/regression-test/data/query_p0/aggregate/count_at_min_max.out
@@ -0,0 +1,36 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !grouped_integer --
+a 2 1
+b 2 2
+c 0 0
+
+-- !grouped_string --
+a 1 1
+b 1 1
+c 0 0
+
+-- !grouped_decimal --
+a 2 1
+b 2 2
+c 0 0
+
+-- !grouped_datetime --
+a 1 1
+b 1 1
+c 0 0
+
+-- !scalar --
+2 1 1 1
+
+-- !empty_input --
+0 0
+
+-- !all_null_group --
+0 0
+
+-- !foreach_all_null_element --
+[1, 0] [1, 0]
+
+-- !foreach_all_null_element_old_adapter --
+[1, 0] [1, 0]
+
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count.groovy
new file mode 100644
index 00000000000..793cba7ca32
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count.groovy
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_ivm_minmax_count") {
+ // The base-table queries below assert the same rows as the MV, so
transparent rewrite must be off.
+ sql """set enable_materialized_view_rewrite = false;"""
+ sql """set enable_dml_materialized_view_rewrite = false;"""
+ sql """drop materialized view if exists ivm_mm_count_mv"""
+ sql """drop table if exists ivm_mm_count_t"""
+
+ sql """
+ CREATE TABLE ivm_mm_count_t (
+ id BIGINT NOT NULL,
+ category VARCHAR(16) NOT NULL,
+ score BIGINT
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ // category a: min=10 twice, max=30 once
+ // category b: min=5 three times, max=5
+ sql """INSERT INTO ivm_mm_count_t VALUES
+ (1, 'a', 10), (2, 'a', 10), (3, 'a', 30),
+ (4, 'b', 5), (5, 'b', 5), (6, 'b', 5)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ivm_mm_count_mv
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ KEY(category)
+ DISTRIBUTED BY HASH(category) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT category, MIN(score) AS min_score, MAX(score) AS max_score,
COUNT(*) AS row_count
+ FROM ivm_mm_count_t
+ GROUP BY category
+ """
+
+ def assertNoFallback = { task ->
+ def reason = task.IvmFallbackReason
+ assertTrue(reason == null || reason.toString() == "\\N",
+ "unexpected IVM fallback: ${reason}")
+ }
+
+ // A task that just finished is briefly absent from tasks(): the job
removes it from its
+ // running list before the MV history publishes it, so the newest row can
temporarily be
+ // the previous refresh. Require a different TaskId before trusting a
terminal row.
+ String lastTaskId = null
+ def latestTask = {
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status, RefreshMode, IvmFallbackReason, ErrorMsg
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = 'ivm_mm_count_mv'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != lastTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ lastTaskId = taskResult[0].TaskId.toString()
+ return taskResult[0]
+ }
+
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_count_mv INCREMENTAL"""
+ waitingMTMVTaskFinishedByMvName("ivm_mm_count_mv")
+ lastTaskId = sql_return_maparray("""
+ SELECT TaskId
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = 'ivm_mm_count_mv'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)[0].TaskId.toString()
+ order_qt_initial """
+ SELECT category, min_score, max_score, row_count
+ FROM ivm_mm_count_mv
+ """
+
+ // Delete one of the two rows that hold the minimum of category a. The
minimum survives,
+ // so the hidden min-count state keeps the refresh incremental.
+ sql """DELETE FROM ivm_mm_count_t WHERE id = 1"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_count_mv INCREMENTAL"""
+ def duplicateMinTask = latestTask()
+ assertEquals("SUCCESS", duplicateMinTask.Status.toString())
+ assertNoFallback(duplicateMinTask)
+ order_qt_after_duplicate_min_delete """
+ SELECT category, min_score, max_score, row_count
+ FROM ivm_mm_count_mv
+ """
+
+ // Delete one of the three rows that hold the maximum of category b. The
maximum survives.
+ sql """DELETE FROM ivm_mm_count_t WHERE id = 4"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_count_mv INCREMENTAL"""
+ def duplicateMaxTask = latestTask()
+ assertEquals("SUCCESS", duplicateMaxTask.Status.toString())
+ assertNoFallback(duplicateMaxTask)
+ order_qt_after_duplicate_max_delete """
+ SELECT category, min_score, max_score, row_count
+ FROM ivm_mm_count_mv
+ """
+
+ // Delete the last row holding category a's minimum. Its count reaches
zero and no smaller
+ // value was inserted, so the refresh still falls back to a complete
refresh.
+ sql """DELETE FROM ivm_mm_count_t WHERE id = 2"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_count_mv INCREMENTAL FALLBACK"""
+ def lastMinTask = latestTask()
+ assertEquals("SUCCESS", lastMinTask.Status.toString())
+ assertEquals("COMPLETE", lastMinTask.RefreshMode.toString())
+ assertEquals("MIN_MAX_BOUNDARY_HIT",
lastMinTask.IvmFallbackReason.toString())
+ order_qt_after_last_min_delete """
+ SELECT category, min_score, max_score, row_count
+ FROM ivm_mm_count_mv
+ """
+
+ // A smaller inserted value becomes the new minimum without a complete
refresh.
+ sql """INSERT INTO ivm_mm_count_t VALUES (7, 'a', 7), (8, 'a', 7)"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_count_mv INCREMENTAL"""
+ def smallerInsertTask = latestTask()
+ assertEquals("SUCCESS", smallerInsertTask.Status.toString())
+ assertNoFallback(smallerInsertTask)
+ order_qt_after_smaller_insert """
+ SELECT category, min_score, max_score, row_count
+ FROM ivm_mm_count_mv
+ """
+
+ // The source query must agree with the MV after every step.
+ order_qt_source """
+ SELECT category, MIN(score), MAX(score), COUNT(*)
+ FROM ivm_mm_count_t
+ GROUP BY category
+ """
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count_join.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count_join.groovy
new file mode 100644
index 00000000000..d2709f2a0b8
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count_join.groovy
@@ -0,0 +1,148 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_ivm_minmax_count_join") {
+ // The base-table queries below assert the same rows as the MV, so
transparent rewrite must be off.
+ sql """set enable_materialized_view_rewrite = false;"""
+ sql """set enable_dml_materialized_view_rewrite = false;"""
+ sql """drop materialized view if exists ivm_mm_join_mv"""
+ sql """drop table if exists ivm_mm_join_l"""
+ sql """drop table if exists ivm_mm_join_r"""
+
+ sql """
+ CREATE TABLE ivm_mm_join_l (
+ k BIGINT NOT NULL,
+ v BIGINT
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE TABLE ivm_mm_join_r (
+ k BIGINT NOT NULL
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ // The initial join produces only NULL values, so the MV starts with MIN =
NULL and the
+ // hidden MIN_COUNT is zero.
+ sql """INSERT INTO ivm_mm_join_l VALUES (0, NULL)"""
+ sql """INSERT INTO ivm_mm_join_r VALUES (0), (1), (2), (3)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ivm_mm_join_mv
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT MIN(l.v) AS min_v
+ FROM ivm_mm_join_l l
+ INNER JOIN ivm_mm_join_r r ON l.k = r.k
+ """
+
+ def assertNoFallback = { task ->
+ def reason = task.IvmFallbackReason
+ assertTrue(reason == null || reason.toString() == "\\N",
+ "unexpected IVM fallback: ${reason}")
+ }
+
+ // A task that just finished is briefly absent from tasks(): the job
removes it from its
+ // running list before the MV history publishes it, so the newest row can
temporarily be
+ // the previous refresh. Require a different TaskId before trusting a
terminal row.
+ String lastTaskId = null
+ def latestTask = {
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status, RefreshMode, IvmFallbackReason, ErrorMsg
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = 'ivm_mm_join_mv'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != lastTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ lastTaskId = taskResult[0].TaskId.toString()
+ return taskResult[0]
+ }
+
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_join_mv INCREMENTAL"""
+ waitingMTMVTaskFinishedByMvName("ivm_mm_join_mv")
+ lastTaskId = sql_return_maparray("""
+ SELECT TaskId
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = 'ivm_mm_join_mv'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)[0].TaskId.toString()
+ order_qt_initial """
+ SELECT min_v FROM ivm_mm_join_mv
+ """
+
+ // Insert L rows for R keys 1, 2 and 3, then delete R key 1. The join
rewrite emits
+ // +{1,1,2} from leftDelta JOIN rightPre and -{1} from leftPost JOIN
rightDelta, so the
+ // new minimum 1 has net count 1 (the delete cancels one of the two
inserted rows).
+ sql """INSERT INTO ivm_mm_join_l VALUES (1, 1), (2, 1), (3, 2)"""
+ sql """DELETE FROM ivm_mm_join_r WHERE k = 1"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_join_mv INCREMENTAL"""
+ def canceledDeleteTask = latestTask()
+ assertEquals("SUCCESS", canceledDeleteTask.Status.toString())
+ assertNoFallback(canceledDeleteTask)
+ order_qt_after_canceled_delete """
+ SELECT min_v FROM ivm_mm_join_mv
+ """
+
+ // The stored count for value 1 is 1. Deleting the remaining row that
produces value 1
+ // exhausts it, so the refresh must fall back instead of retaining the
stale MIN = 1.
+ sql """DELETE FROM ivm_mm_join_l WHERE k = 2"""
+ Thread.sleep(1000)
+ sql """REFRESH MATERIALIZED VIEW ivm_mm_join_mv INCREMENTAL FALLBACK"""
+ def boundaryTask = latestTask()
+ assertEquals("SUCCESS", boundaryTask.Status.toString())
+ assertEquals("COMPLETE", boundaryTask.RefreshMode.toString())
+ assertEquals("MIN_MAX_BOUNDARY_HIT",
boundaryTask.IvmFallbackReason.toString())
+ order_qt_after_boundary """
+ SELECT min_v FROM ivm_mm_join_mv
+ """
+
+ order_qt_source """
+ SELECT MIN(l.v)
+ FROM ivm_mm_join_l l
+ INNER JOIN ivm_mm_join_r r ON l.k = r.k
+ """
+}
diff --git a/regression-test/suites/query_p0/aggregate/count_at_min_max.groovy
b/regression-test/suites/query_p0/aggregate/count_at_min_max.groovy
new file mode 100644
index 00000000000..4addaa28ac5
--- /dev/null
+++ b/regression-test/suites/query_p0/aggregate/count_at_min_max.groovy
@@ -0,0 +1,126 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("count_at_min_max") {
+ sql """drop table if exists count_at_min_max_t"""
+
+ sql """
+ CREATE TABLE count_at_min_max_t (
+ id INT NOT NULL,
+ grp VARCHAR(8) NOT NULL,
+ i BIGINT,
+ s STRING,
+ d DECIMAL(10, 2),
+ dt DATETIME
+ )
+ ENGINE = OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 2
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+
+ // grp a: i min=10 twice, max=30 once; s min='a', max='c'
+ // grp b: i min=max=5 twice; the NULL row must be ignored
+ // grp c: only NULL values, so both counts are zero
+ sql """INSERT INTO count_at_min_max_t VALUES
+ (1, 'a', 10, 'b', 10.50, '2024-01-01 10:00:00'),
+ (2, 'a', 10, 'a', 10.50, '2024-01-01 09:00:00'),
+ (3, 'a', 30, 'c', 30.50, '2024-01-01 11:00:00'),
+ (4, 'b', 5, 'x', 5.50, '2024-01-02 10:00:00'),
+ (5, 'b', 5, 'y', 5.50, '2024-01-02 09:00:00'),
+ (6, 'b', NULL, NULL, NULL, NULL),
+ (7, 'c', NULL, NULL, NULL, NULL)"""
+
+ order_qt_grouped_integer """
+ SELECT grp, count_at_min(i) AS min_cnt, count_at_max(i) AS max_cnt
+ FROM count_at_min_max_t
+ GROUP BY grp
+ """
+
+ order_qt_grouped_string """
+ SELECT grp, count_at_min(s) AS min_cnt, count_at_max(s) AS max_cnt
+ FROM count_at_min_max_t
+ GROUP BY grp
+ """
+
+ order_qt_grouped_decimal """
+ SELECT grp, count_at_min(d) AS min_cnt, count_at_max(d) AS max_cnt
+ FROM count_at_min_max_t
+ GROUP BY grp
+ """
+
+ order_qt_grouped_datetime """
+ SELECT grp, count_at_min(dt) AS min_cnt, count_at_max(dt) AS max_cnt
+ FROM count_at_min_max_t
+ GROUP BY grp
+ """
+
+ order_qt_scalar """
+ SELECT count_at_min(i) AS min_i_cnt, count_at_max(i) AS max_i_cnt,
+ count_at_min(s) AS min_s_cnt, count_at_max(s) AS max_s_cnt
+ FROM count_at_min_max_t
+ """
+
+ // Empty input returns zero, not NULL.
+ order_qt_empty_input """
+ SELECT count_at_min(i) AS min_cnt, count_at_max(i) AS max_cnt
+ FROM count_at_min_max_t
+ WHERE id < 0
+ """
+
+ // A group whose only rows are NULL has no non-NULL value.
+ order_qt_all_null_group """
+ SELECT count_at_min(i) AS min_cnt, count_at_max(i) AS max_cnt
+ FROM count_at_min_max_t
+ WHERE grp = 'c'
+ """
+
+ // The foreach combinator applies the aggregate per array position. An
all-NULL
+ // position must return 0, not NULL, for both the default (V2) and old
nullable
+ // adapters.
+ order_qt_foreach_all_null_element """
+ SELECT count_at_min_foreach(a) AS min_cnt, count_at_max_foreach(a) AS
max_cnt
+ FROM (
+ SELECT CAST([1, NULL] AS ARRAY<INT>) AS a
+ UNION ALL
+ SELECT CAST([2, NULL] AS ARRAY<INT>)
+ ) t
+ """
+
+ sql "set enable_aggregate_function_null_v2 = false"
+ order_qt_foreach_all_null_element_old_adapter """
+ SELECT count_at_min_foreach(a) AS min_cnt, count_at_max_foreach(a) AS
max_cnt
+ FROM (
+ SELECT CAST([1, NULL] AS ARRAY<INT>) AS a
+ UNION ALL
+ SELECT CAST([2, NULL] AS ARRAY<INT>)
+ ) t
+ """
+ sql "set enable_aggregate_function_null_v2 = true"
+
+ test {
+ sql "SELECT count_at_min(DISTINCT i) FROM count_at_min_max_t"
+ exception "COUNT_AT_MIN does not support DISTINCT"
+ }
+
+ test {
+ sql "SELECT count_at_max(DISTINCT i) FROM count_at_min_max_t"
+ exception "COUNT_AT_MAX does not support DISTINCT"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
