github-actions[bot] commented on code in PR #69286: URL: https://github.com/apache/doris/pull/69286#discussion_r4313715856
########## 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: [P2] Expect two minima in the merge fixture This fixture selects `count_at_min` over partial inputs `{5,1,5}` and `{5,1,9}`. Each partial has minimum 1 with count 1, so merging equal extrema correctly returns 2. `EXPECT_EQ(5, ...)` therefore fails deterministically with the implementation in this PR. Expect 2, or change the fixture to contain five occurrences of the global minimum. ########## 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: [P2] Wait for the submitted refresh task before checking fallback This predicate accepts any latest terminal row without checking its TaskId. During `MTMVTask.onSuccess`, the current task is removed from the running list before `after()` publishes it into MV history, so `tasks()` can temporarily return the preceding completed refresh. At the last-min deletion, this helper can then return the previous INCREMENTAL task and fail the expected COMPLETE assertion; after the smaller insert, it can return the previous fallback task and fail `assertNoFallback`. `Suite.pollMTMVTaskTerminal` already handles this exact gap. Reuse that wait logic and inspect the same completed TaskId, or capture the previous TaskId and wait for a distinct terminal task. ########## 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: [P1] Subtract matching deletes when initializing the extreme count When `oldExtreme` is NULL, persisting `deltaInsertCount` ignores canceled delete contributions. A reachable plan is `Aggregate(MIN(L.v)) -> InnerJoin(L.k = R.k)`: initially L has `(k=0,v=NULL)` and R has keys 0,1,2,3. Before refresh, insert L rows `(1,1),(2,1),(3,2)` and delete R key 1. The join rewrite emits `+{1,1,2}` from `leftDelta JOIN rightPre` and `-{1}` from `leftPost JOIN rightDelta`. The resulting MIN is correctly 1, but this branch stores MIN_COUNT=2 instead of 1. On the next refresh, deleting L key 2 leaves only non-NULL value 2; `survivingOldCount=2-1` stays positive, so the guard silently retains MIN=1 instead of falling back. This later wrong result is introduced by the new count state; the previous guard would reject that boundary deletion. Initialize the count from net contributions at the selected extreme and preserve fallback when its survival cannot be proved. MAX has the symmetric failure. -- 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]
