yujun777 commented on code in PR #69286:
URL: https://github.com/apache/doris/pull/69286#discussion_r4314470270
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggExtremalProcessor.java:
##########
@@ -109,48 +135,94 @@ public void appendApplyExpressions(IvmAggTarget target,
IvmAggApplyContext apply
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 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));
+
+ // 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 guardCond = new Or(ImmutableList.of(
- new EqualTo(newCount, new BigIntLiteral(0L)),
- new IsNull(deltaDel),
+ new EqualTo(newTotalCount, zero),
new IsNull(oldExtreme),
- deletedExtremeKeepsOldValueValid(deltaDel, oldExtreme,
deltaInsert)
- ));
+ new IsNull(deltaDelete),
+ new And(
+ new And(deletedPresent,
+ deletedExtremeIsAtLeastOld(deltaDelete,
oldExtreme)),
+ new Or(ImmutableList.of(
+ insertedBetter,
+ new GreaterThan(survivingOldCount, zero))))));
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 newCountRaw = new CaseWhen(
+ ImmutableList.of(
+ new WhenClause(new EqualTo(newTotalCount, zero), zero),
+ new WhenClause(new IsNull(oldExtreme),
deltaInsertCount),
Review Comment:
Fixed in 87a8d552ab5. When the old extreme is NULL, the count now comes from
the net insert/delete contributions at the insert-side extreme: subtract
deltaDeleteCount when the delete-side extreme equals it, use deltaInsertCount
when the delete extreme is farther from the aggregate extreme, and keep the
complete-refresh fallback when a delete is closer or cancels every inserted
row. I also added the two-refresh canceled-join regression
test_ivm_minmax_count_join, which reproduced the stale MIN=1 behavior before
the fix.
##########
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(5, result.get_element(0));
Review Comment:
Fixed in 87a8d552ab5. The merge fixture now expects 2: each partial has
minimum 1 with count 1, so merging equal extrema returns 2.
##########
regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_count.groovy:
##########
@@ -0,0 +1,149 @@
+// 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}")
+ }
+
+ def latestTask = {
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT 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].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
Review Comment:
Fixed in 87a8d552ab5. Both MIN/MAX count suites now remember the previous
TaskId and only accept a terminal row whose TaskId differs, so the
running-list-to-history gap can no longer return the preceding refresh. The new
join suite uses the same wait.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
