yujun777 commented on code in PR #69456:
URL: https://github.com/apache/doris/pull/69456#discussion_r4348577909
##########
regression-test/data/query_p0/aggregate/topk_min_max_count.out:
##########
@@ -0,0 +1,143 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !min_k0 --
+a {"values":[], "bound":1}
+b {"values":[], "bound":5}
+c {"values":[], "bound":null}
+d {"values":[], "bound":10}
+e {"values":[], "bound":0}
+
+-- !max_k0 --
+a {"values":[], "bound":3}
+b {"values":[], "bound":5}
+c {"values":[], "bound":null}
+d {"values":[], "bound":17}
+e {"values":[], "bound":0}
+
+-- !min_k1 --
+a {"values":[{"value":1, "count":2}], "bound":1}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":10, "count":1}], "bound":10}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !max_k1 --
+a {"values":[{"value":3, "count":2}], "bound":3}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":17, "count":1}], "bound":17}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !min_k2 --
+a {"values":[{"value":1, "count":2}, {"value":2, "count":1}], "bound":2}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":10, "count":1}, {"value":11, "count":1}],
"bound":11}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !max_k2 --
+a {"values":[{"value":3, "count":2}, {"value":2, "count":1}], "bound":2}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":17, "count":1}, {"value":16, "count":1}],
"bound":16}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !min_k8 --
+a {"values":[{"value":1, "count":2}, {"value":2, "count":1}, {"value":3,
"count":2}], "bound":null}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":10, "count":1}, {"value":11, "count":1},
{"value":12, "count":1}, {"value":13, "count":1}, {"value":14, "count":1},
{"value":15, "count":1}, {"value":16, "count":1}, {"value":17, "count":1}],
"bound":null}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !max_k8 --
+a {"values":[{"value":3, "count":2}, {"value":2, "count":1}, {"value":1,
"count":2}], "bound":null}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[], "bound":null}
+d {"values":[{"value":17, "count":1}, {"value":16, "count":1},
{"value":15, "count":1}, {"value":14, "count":1}, {"value":13, "count":1},
{"value":12, "count":1}, {"value":11, "count":1}, {"value":10, "count":1}],
"bound":null}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !min_empty_input --
+{"values":[], "bound":null}
+
+-- !min_all_null_group --
+c {"values":[], "bound":null}
+
+-- !min_not_null_type --
+a {"values":[{"value":1, "count":2}, {"value":2, "count":1}], "bound":2}
+b {"values":[{"value":5, "count":1}], "bound":null}
+c {"values":[{"value":0, "count":2}], "bound":null}
+d {"values":[{"value":10, "count":1}, {"value":11, "count":1}],
"bound":11}
+e {"values":[{"value":0, "count":5}], "bound":null}
+
+-- !min_not_null_type_empty --
+{"values":[], "bound":null}
+
+-- !min_string --
+a {"values":[{"value":"a", "count":1}, {"value":"b", "count":1}],
"bound":"b"} {"values":[{"value":"z", "count":1}, {"value":"c", "count":3}],
"bound":"c"}
+b {"values":[{"value":"x", "count":1}], "bound":null}
{"values":[{"value":"x", "count":1}], "bound":null}
+c {"values":[], "bound":null} {"values":[], "bound":null}
+d {"values":[{"value":"j", "count":1}, {"value":"k", "count":1}],
"bound":"k"} {"values":[{"value":"q", "count":1}, {"value":"p", "count":1}],
"bound":"p"}
+e {"values":[{"value":"s", "count":5}], "bound":null}
{"values":[{"value":"s", "count":5}], "bound":null}
+
+-- !min_decimal --
+a {"values":[{"value":1.10, "count":2}, {"value":2.20, "count":1}],
"bound":2.20} {"values":[{"value":3.30, "count":2}, {"value":2.20, "count":1}],
"bound":2.20}
+b {"values":[{"value":5.50, "count":1}], "bound":null}
{"values":[{"value":5.50, "count":1}], "bound":null}
+c {"values":[], "bound":null} {"values":[], "bound":null}
+d {"values":[{"value":10.10, "count":1}, {"value":11.10, "count":1}],
"bound":11.10} {"values":[{"value":17.10, "count":1}, {"value":16.10,
"count":1}], "bound":16.10}
+e {"values":[{"value":0.00, "count":5}], "bound":null}
{"values":[{"value":0.00, "count":5}], "bound":null}
+
+-- !min_datetime --
+a {"values":[{"value":"2024-01-01 09:00:00", "count":1},
{"value":"2024-01-01 10:00:00", "count":1}], "bound":"2024-01-01 10:00:00"}
{"values":[{"value":"2024-01-03 10:00:00", "count":2}, {"value":"2024-01-02
10:00:00", "count":1}], "bound":"2024-01-02 10:00:00"}
+b {"values":[{"value":"2024-02-01 10:00:00", "count":1}], "bound":null}
{"values":[{"value":"2024-02-01 10:00:00", "count":1}], "bound":null}
+c {"values":[], "bound":null} {"values":[], "bound":null}
+d {"values":[{"value":"2024-03-01 10:00:00", "count":1},
{"value":"2024-03-02 10:00:00", "count":1}], "bound":"2024-03-02 10:00:00"}
{"values":[{"value":"2024-03-08 10:00:00", "count":1}, {"value":"2024-03-07
10:00:00", "count":1}], "bound":"2024-03-07 10:00:00"}
+e {"values":[], "bound":null} {"values":[], "bound":null}
+
+-- !double_special_values --
+e {"values":[{"value":-0, "count":2}, {"value":1, "count":1},
{"value":NaN, "count":2}], "bound":null} {"values":[{"value":NaN,
"count":2}, {"value":1, "count":1}, {"value":-0, "count":2}], "bound":null}
Review Comment:
Fixed in 56a622307d2. The case no longer depends on scan order: it is an
explicit `topk_merge_min` with `-0.0` on the left and `0.0` inserted on the
right, so the retained representative is deterministic, and the golden pins
that single merge result.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/TopKMergeFunction.java:
##########
@@ -0,0 +1,139 @@
+// 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.scalar;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
+import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.TopKCountFunction;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.StructField;
+import org.apache.doris.nereids.types.StructType;
+
+import java.util.List;
+
+/**
+ * Shared implementation of {@code topk_merge_min} and {@code topk_merge_max},
the apply-side
+ * merge of a MIN/MAX top-K state with the insert/delete top-K lists of one
delta batch:
+ *
+ * <pre>
+ * topk_merge_min(S, b, I, b_i, D, b_d, n, dn, k)
+ * -> STRUCT<values ARRAY<STRUCT<value T, count BIGINT>>,
bound T>
+ * </pre>
+ *
+ * <p>The result is the state itself and has exactly the type the aggregate
returns, so apply can
+ * write it back to the hidden top-K state columns without rebuilding a
struct. The caller derives
+ * {@code n' = n + dn} and {@code exhausted = (size(values) == 0 && n' > 0)}
itself.
+ *
+ * <p>A NULL list is an empty list; a NULL {@code b}/{@code b_i}/{@code b_d}
means that side is
+ * complete/untruncated. The result struct itself is never NULL, only its
{@code bound} field can
+ * be: the merge only trusts values in {@code v <= min(b, b_i, b_d)}, so
{@code bound} is the
+ * effective upper bound (for MIN; the lower bound for MAX) of the resulting
state.
+ */
+public abstract class TopKMergeFunction extends ScalarFunction
+ implements CustomSignature, AlwaysNotNullable {
+
+ protected TopKMergeFunction(String name, List<Expression> children) {
+ super(name, children);
+ }
+
+ /** constructor for withChildren and reuse signature */
+ protected TopKMergeFunction(ScalarFunctionParams functionParams) {
+ super(functionParams);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ if (arity() != 9) {
+ throw new AnalysisException(getName() + " requires 9 arguments,
but got " + arity());
+ }
+ DataType valueType = inferValueType();
+ if (valueType == null) {
+ throw new AnalysisException(getName() + " can not infer the value
type from arguments: "
+ + toSql());
+ }
+ if (!TopKCountFunction.isSupportedValueType(valueType)) {
+ throw new AnalysisException(getName() + " does not support type "
+ valueType.toSql());
+ }
+ for (int index : new int[] {0, 2, 4}) {
+ DataType type = getArgumentType(index);
+ if (type.isNullType()) {
+ continue;
+ }
+ if (!(type instanceof ArrayType)
+ || !(((ArrayType) type).getItemType() instanceof
StructType)) {
+ throw new AnalysisException(getName() + " requires argument "
+ (index + 1)
+ + " to be ARRAY<STRUCT<value, count>>, but got " +
type.toSql());
+ }
+ }
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ DataType valueType = inferValueType();
+ if (valueType == null) {
+ throw new AnalysisException(getName() + " can not infer the value
type from arguments: "
+ + toSql());
+ }
+ ArrayType listType =
ArrayType.of(TopKCountFunction.pairType(valueType));
+ return FunctionSignature.ret(resultType(valueType)).args(
Review Comment:
Fixed in e4acad346dd. `TopKMergeFunction` now overrides `computePrecision`
and returns the signature unchanged, so a NULL bound can no longer contribute
DECIMAL(1,0) and narrow the nested lists.
`TopKFunctionSignatureTest.testMergeKeepsDecimalPrecisionWithNullBounds` covers
S=[(1.23,1)] with NULL bounds, and `order_qt_merge_decimal_null_bounds` covers
it end-to-end.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/TopKMergeFunction.java:
##########
@@ -0,0 +1,139 @@
+// 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.scalar;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable;
+import org.apache.doris.nereids.trees.expressions.functions.CustomSignature;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.TopKCountFunction;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.StructField;
+import org.apache.doris.nereids.types.StructType;
+
+import java.util.List;
+
+/**
+ * Shared implementation of {@code topk_merge_min} and {@code topk_merge_max},
the apply-side
+ * merge of a MIN/MAX top-K state with the insert/delete top-K lists of one
delta batch:
+ *
+ * <pre>
+ * topk_merge_min(S, b, I, b_i, D, b_d, n, dn, k)
+ * -> STRUCT<values ARRAY<STRUCT<value T, count BIGINT>>,
bound T>
+ * </pre>
+ *
+ * <p>The result is the state itself and has exactly the type the aggregate
returns, so apply can
+ * write it back to the hidden top-K state columns without rebuilding a
struct. The caller derives
+ * {@code n' = n + dn} and {@code exhausted = (size(values) == 0 && n' > 0)}
itself.
+ *
+ * <p>A NULL list is an empty list; a NULL {@code b}/{@code b_i}/{@code b_d}
means that side is
+ * complete/untruncated. The result struct itself is never NULL, only its
{@code bound} field can
+ * be: the merge only trusts values in {@code v <= min(b, b_i, b_d)}, so
{@code bound} is the
+ * effective upper bound (for MIN; the lower bound for MAX) of the resulting
state.
+ */
+public abstract class TopKMergeFunction extends ScalarFunction
+ implements CustomSignature, AlwaysNotNullable {
+
+ protected TopKMergeFunction(String name, List<Expression> children) {
+ super(name, children);
+ }
+
+ /** constructor for withChildren and reuse signature */
+ protected TopKMergeFunction(ScalarFunctionParams functionParams) {
+ super(functionParams);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ if (arity() != 9) {
+ throw new AnalysisException(getName() + " requires 9 arguments,
but got " + arity());
+ }
+ DataType valueType = inferValueType();
+ if (valueType == null) {
+ throw new AnalysisException(getName() + " can not infer the value
type from arguments: "
+ + toSql());
+ }
+ if (!TopKCountFunction.isSupportedValueType(valueType)) {
+ throw new AnalysisException(getName() + " does not support type "
+ valueType.toSql());
+ }
+ for (int index : new int[] {0, 2, 4}) {
+ DataType type = getArgumentType(index);
+ if (type.isNullType()) {
+ continue;
+ }
+ if (!(type instanceof ArrayType)
+ || !(((ArrayType) type).getItemType() instanceof
StructType)) {
+ throw new AnalysisException(getName() + " requires argument "
+ (index + 1)
+ + " to be ARRAY<STRUCT<value, count>>, but got " +
type.toSql());
+ }
+ }
+ }
+
+ @Override
+ public FunctionSignature customSignature() {
+ DataType valueType = inferValueType();
+ if (valueType == null) {
+ throw new AnalysisException(getName() + " can not infer the value
type from arguments: "
+ + toSql());
+ }
+ ArrayType listType =
ArrayType.of(TopKCountFunction.pairType(valueType));
+ return FunctionSignature.ret(resultType(valueType)).args(
+ listType, valueType,
+ listType, valueType,
+ listType, valueType,
+ BigIntType.INSTANCE, BigIntType.INSTANCE, BigIntType.INSTANCE);
+ }
+
+ /**
+ * The value type is read from the first argument that carries it: the
per-value type of S/I/D
+ * (argument positions 1, 3, 5) or, when all lists are untyped NULL
literals, the bound type
+ * (positions 2, 4, 6).
+ */
+ private DataType inferValueType() {
+ for (int index : new int[] {0, 2, 4}) {
+ DataType type = getArgumentType(index);
+ if (!(type instanceof ArrayType)) {
+ continue;
+ }
+ DataType itemType = ((ArrayType) type).getItemType();
+ if (!(itemType instanceof StructType)) {
+ continue;
+ }
+ List<StructField> fields = ((StructType) itemType).getFields();
+ if (fields.size() != 2) {
+ continue;
+ }
+ return
TopKCountFunction.normalizeValueType(fields.get(0).getDataType());
Review Comment:
Fixed in e4acad346dd. `inferValueType` now requires the same value type in
every typed list and bound and raises an analysis error for mixed types, so no
list is coerced to a narrower type.
`TopKFunctionSignatureTest.testMergeRejectsMixedValueTypes` covers empty INT
lists with a DOUBLE list, and the regression asserts the same error.
##########
be/src/exprs/function/function_topk_merge.cpp:
##########
@@ -0,0 +1,465 @@
+// 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 <glog/logging.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "common/check.h"
+#include "common/compare.h"
+#include "common/exception.h"
+#include "common/status.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/block/column_numbers.h"
+#include "core/block/column_with_type_and_name.h"
+#include "core/column/column.h"
+#include "core/column/column_array.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
+#include "core/data_type/primitive_type.h"
+#include "core/types.h"
+#include "exprs/function/function.h"
+#include "exprs/function/simple_function_factory.h"
+
+namespace doris {
+class FunctionContext;
+} // namespace doris
+
+namespace doris {
+
+namespace {
+
+/// Upper bound of k. Keep in sync with the aggregate's MAX_TOPK_K
+/// (aggregate_function_topk_count.h) and FE TopKCountFunction.MAX_TOPK_K.
+constexpr int64_t MAX_TOPK_K = 65536;
+
+/// One row of an ARRAY<STRUCT<value, count>> argument. A NULL array is an
empty view.
+struct TopKArrayView {
+ const IColumn* values = nullptr;
+ const ColumnInt64* counts = nullptr;
+ /// NULL bitmap of the array element (the pair struct).
+ const NullMap* null_map = nullptr;
+ /// NULL bitmap of the `value` field; NULL values are not part of the
contract.
+ const NullMap* value_null_map = nullptr;
+ /// NULL bitmap of the `count` field; NULL counts are not part of the
contract.
+ const NullMap* count_null_map = nullptr;
+ size_t begin = 0;
+ size_t end = 0;
+};
+
+TopKArrayView make_array_view(const IColumn& column, size_t row) {
+ TopKArrayView view;
+ const IColumn* array_column = &column;
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(&column)) {
+ if (nullable->is_null_at(row)) {
+ return view;
+ }
+ array_column = &nullable->get_nested_column();
+ }
+ const auto& array = assert_cast<const ColumnArray&>(*array_column);
+ // offsets[i] is the cumulative size after row i, so row's range is
[offsets[row-1], offsets[row]).
+ // offset_at(0) reads offsets[-1], which PaddedPODArray keeps as 0.
+ view.begin = array.offset_at(static_cast<ssize_t>(row));
+ view.end = array.offset_at(static_cast<ssize_t>(row) + 1);
+
+ const IColumn* pair_column = &array.get_data();
+ if (const auto* element_nullable =
check_and_get_column<ColumnNullable>(pair_column)) {
+ view.null_map = &element_nullable->get_null_map_data();
+ pair_column = &element_nullable->get_nested_column();
+ }
+ const auto& pairs = assert_cast<const ColumnStruct&>(*pair_column);
+ view.values = &pairs.get_column(0);
+ // A cast may widen the `value` field to a nullable column; the entries
themselves are never
+ // NULL, so only the nested column is used.
+ if (const auto* value_nullable =
check_and_get_column<ColumnNullable>(view.values)) {
+ view.value_null_map = &value_nullable->get_null_map_data();
+ view.values = &value_nullable->get_nested_column();
+ }
+ // The same holds for `count`: a nested cast may have made the field
nullable.
+ const auto* count_column = &pairs.get_column(1);
+ if (const auto* count_nullable =
check_and_get_column<ColumnNullable>(count_column)) {
+ view.count_null_map = &count_nullable->get_null_map_data();
+ count_column = &count_nullable->get_nested_column();
+ }
+ view.counts = &assert_cast<const ColumnInt64&>(*count_column);
+ return view;
+}
+
+/// One scalar bound argument: NULL means the corresponding side was not
truncated. `row` is the
+/// index of the current input row inside `value`: bounds are ordinary per-row
columns, so a block
+/// can carry a different bound in every row.
+struct TopKBoundView {
+ const IColumn* value = nullptr;
+ size_t row = 0;
+ bool is_null = true;
+};
+
+TopKBoundView make_bound_view(const IColumn& column, size_t row) {
+ TopKBoundView view;
+ const IColumn* value_column = &column;
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(&column)) {
+ if (nullable->is_null_at(row)) {
+ return view;
+ }
+ value_column = &nullable->get_nested_column();
+ }
+ view.value = value_column;
+ view.row = row;
+ view.is_null = false;
+ return view;
+}
+
+int64_t read_int64(const IColumn& column, size_t row, const char*
argument_name) {
+ const IColumn* nested = &column;
+ if (const auto* nullable = check_and_get_column<ColumnNullable>(&column)) {
+ // The contract declares these arguments as non-NULL BIGINT.
+ DORIS_CHECK(!nullable->is_null_at(row)) << argument_name << " must not
be NULL";
+ nested = &nullable->get_nested_column();
+ }
+ return assert_cast<const ColumnInt64&,
TypeCheckOnRelease::DISABLE>(*nested).get_element(row);
+}
+
+int64_t checked_add(int64_t lhs, int64_t rhs, const char* function_name) {
+ int64_t result = 0;
+ if (UNLIKELY(__builtin_add_overflow(lhs, rhs, &result))) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "{}: value count
overflow", function_name);
+ }
+ return result;
+}
+
+int64_t checked_sub(int64_t lhs, int64_t rhs, const char* function_name) {
+ int64_t result = 0;
+ if (UNLIKELY(__builtin_sub_overflow(lhs, rhs, &result))) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT, "{}: value count
overflow", function_name);
+ }
+ return result;
+}
+
+/// FLOAT/DOUBLE use CompareFloat so NaN sorts greatest and is distinct from
every number; other
+/// types use the column comparison, i.e. exactly the MIN/MAX comparison.
+int compare_values(PrimitiveType value_type, const IColumn& lhs, size_t
lhs_row, const IColumn& rhs,
+ size_t rhs_row) {
+ if (value_type == TYPE_FLOAT) {
+ return CompareFloat(
+ assert_cast<const ColumnFloat32&,
TypeCheckOnRelease::DISABLE>(lhs).get_element(
+ lhs_row),
+ assert_cast<const ColumnFloat32&,
TypeCheckOnRelease::DISABLE>(rhs).get_element(
+ rhs_row));
+ }
+ if (value_type == TYPE_DOUBLE) {
+ return CompareFloat(
+ assert_cast<const ColumnFloat64&,
TypeCheckOnRelease::DISABLE>(lhs).get_element(
+ lhs_row),
+ assert_cast<const ColumnFloat64&,
TypeCheckOnRelease::DISABLE>(rhs).get_element(
+ rhs_row));
+ }
+ return lhs.compare_at(lhs_row, rhs_row, rhs, 1);
+}
+
+DataTypePtr find_value_type(const DataTypes& arguments) {
+ // Prefer the per-value type of S/I/D; it is also present when all bound
arguments are NULL.
+ for (size_t index : {0, 2, 4}) {
+ const auto& type = remove_nullable(arguments[index]);
+ if (type->get_primitive_type() != TYPE_ARRAY) {
+ continue;
+ }
+ const auto& nested =
+ remove_nullable(assert_cast<const
DataTypeArray&>(*type).get_nested_type());
+ if (nested->get_primitive_type() != TYPE_STRUCT) {
+ continue;
+ }
+ const auto& fields = assert_cast<const
DataTypeStruct&>(*nested).get_elements();
+ if (fields.size() != 2) {
+ continue;
+ }
+ // The result struct always declares a non-nullable `value` field,
while an argument cast
+ // may have produced a nullable one; drop it so both sides agree on
the return type.
+ return remove_nullable(fields[0]);
+ }
+ for (size_t index : {1, 3, 5}) {
+ const auto& type = remove_nullable(arguments[index]);
+ if (!type->is_null_literal()) {
+ return type;
+ }
+ }
+ return nullptr;
+}
+
+/**
+ * topk_merge_min / topk_merge_max: one apply step of the IVM MIN/MAX top-K
state.
+ *
+ * Arguments: S, b, I, b_i, D, b_d, n, dn, k. S is the pre-image state, I/D
are the insert/delete
+ * top-K lists of this delta batch and b/i/d bounds are their truncation
points (NULL = not
+ * truncated). Only values in [<= bE] are trusted, where bE = min of all
non-NULL bounds (max for
+ * MAX): outside it at least one side may have dropped values. The post count
of a value is
+ * S + I - D; non-positive counts are removed (net-zero cancellations), then
the first k values are
+ * kept.
+ *
+ * The result is the state itself, i.e. the same type
topk_min_count/topk_max_count return:
+ * STRUCT<values ARRAY<STRUCT<value T, count BIGINT>>, bound T NULL>. n' = n +
dn and
+ * exhausted = (size(values) == 0 && n' > 0) are derived by the caller, which
writes n' to the
+ * COUNT state column and uses exhausted to trigger the fallback.
+ */
+template <bool IsMin>
+class FunctionTopKMerge : public IFunction {
+public:
+ static constexpr auto name = IsMin ? "topk_merge_min" : "topk_merge_max";
+
+ static FunctionPtr create() { return
std::make_shared<FunctionTopKMerge>(); }
+
+ String get_name() const override { return name; }
+
+ size_t get_number_of_arguments() const override { return 9; }
+
+ /// A NULL list is an empty list and a NULL bound is a valid value: the
default implementation
+ /// would turn the whole row into NULL and break the contract.
+ bool use_default_implementation_for_nulls() const override { return false;
}
+
+ DataTypePtr get_return_type_impl(const DataTypes& arguments) const
override {
+ DataTypePtr value_type = find_value_type(arguments);
+ if (value_type == nullptr) {
+ throw Exception(ErrorCode::INVALID_ARGUMENT,
+ "{} can not infer the value type from arguments",
name);
+ }
+ // The result is the state type, identical to the aggregate's return
type, so that the
+ // apply expression can write it straight back to the hidden
MIN_TOPK/MAX_TOPK and BOUND
+ // columns.
+ auto pair_type = std::make_shared<DataTypeStruct>(
+ DataTypes {value_type, std::make_shared<DataTypeInt64>()},
+ Strings {"value", "count"});
+ auto values_type =
std::make_shared<DataTypeArray>(make_nullable(pair_type));
+ return std::make_shared<DataTypeStruct>(DataTypes {values_type,
make_nullable(value_type)},
+ Strings {"values", "bound"});
+ }
+
+ Status execute_impl(FunctionContext* /*context*/, Block& block, const
ColumnNumbers& arguments,
+ uint32_t result, size_t input_rows_count) const
override {
+ std::vector<ColumnPtr> columns(arguments.size());
+ DataTypes argument_types(arguments.size());
+ for (size_t i = 0; i != arguments.size(); ++i) {
+ columns[i] =
+
block.get_by_position(arguments[i]).column->convert_to_full_column_if_const();
+ argument_types[i] = block.get_by_position(arguments[i]).type;
+ }
+ const DataTypePtr value_type = find_value_type(argument_types);
+ // An untyped call is rejected while planning get_return_type(); the
plan always carries T.
+ DCHECK(value_type != nullptr);
+ const PrimitiveType primitive_type = value_type->get_primitive_type();
+
+ auto result_column =
block.get_by_position(result).type->create_column();
+ auto& result_struct = assert_cast<ColumnStruct&>(*result_column);
+ auto& out_values =
assert_cast<ColumnArray&>(result_struct.get_column(0));
+ auto& out_pair_nullable =
assert_cast<ColumnNullable&>(out_values.get_data());
+ auto& out_pair =
assert_cast<ColumnStruct&>(out_pair_nullable.get_nested_column());
+ auto& out_value_column = out_pair.get_column(0);
+ auto& out_count_column =
assert_cast<ColumnInt64&>(out_pair.get_column(1));
+ auto& out_bound_column =
assert_cast<ColumnNullable&>(result_struct.get_column(1));
+
+ for (size_t row = 0; row != input_rows_count; ++row) {
+ const int64_t k = read_int64(*columns[8], row, "k");
+ if (UNLIKELY(k < 1 || k > MAX_TOPK_K)) {
+ return Status::InvalidArgument("{} requires 1 <= k <= {}, but
got {}", name,
+ MAX_TOPK_K, k);
+ }
+ const int64_t n = read_int64(*columns[6], row, "n");
+ const int64_t dn = read_int64(*columns[7], row, "dn");
+ const int64_t new_n = checked_add(n, dn, name);
+
+ const TopKArrayView s_view = make_array_view(*columns[0], row);
+ const TopKArrayView i_view = make_array_view(*columns[2], row);
+ const TopKArrayView d_view = make_array_view(*columns[4], row);
+ const TopKBoundView b_view = make_bound_view(*columns[1], row);
+ const TopKBoundView bi_view = make_bound_view(*columns[3], row);
+ const TopKBoundView bd_view = make_bound_view(*columns[5], row);
+ const TopKBoundView effective_bound =
+ effective_bound_of(b_view, bi_view, bd_view,
primitive_type);
+
+ size_t positive = 0;
+ RETURN_IF_ERROR(append_trusted_values(s_view, i_view, d_view,
effective_bound, k,
+ primitive_type,
out_value_column,
+ out_count_column,
out_pair_nullable, &positive));
+ out_values.get_offsets().push_back(out_pair.size());
+
+ // bound: NULL while the state is complete, the k-th kept value
only when values were
+ // actually dropped (pos > k), or the effective bound otherwise.
+ // b IS NULL means the pre-image covers every distinct value; a
NULL b_i/b_d means
+ // that side of the delta was not truncated.
+ // pos == k is NOT truncation: every known value fit, so keep bE
instead of shrinking
+ // the trusted range to max(S').
+ const bool sides_complete = b_view.is_null && bi_view.is_null &&
bd_view.is_null;
+ if (new_n == 0 || (sides_complete && std::cmp_less_equal(positive,
k))) {
+ out_bound_column.insert_default();
+ } else if (std::cmp_greater(positive, k)) {
+
out_bound_column.get_nested_column().insert_from(out_value_column,
+
out_pair.size() - 1);
+ out_bound_column.get_null_map_data().push_back(0);
+ } else {
+ DCHECK(!effective_bound.is_null);
+
out_bound_column.get_nested_column().insert_from(*effective_bound.value,
+
effective_bound.row);
+ out_bound_column.get_null_map_data().push_back(0);
+ }
+ }
+
+ block.replace_by_position(result, std::move(result_column));
+ return Status::OK();
+ }
+
+private:
+ /// bE: the best (smallest for MIN, largest for MAX) of the non-NULL
bounds, or nullptr when
+ /// every side is complete/untruncated.
+ static TopKBoundView effective_bound_of(const TopKBoundView& b_view,
+ const TopKBoundView& bi_view,
+ const TopKBoundView& bd_view,
PrimitiveType type) {
+ TopKBoundView bound;
+ for (const TopKBoundView* view : {&b_view, &bi_view, &bd_view}) {
+ if (view->is_null) {
+ continue;
+ }
+ if (bound.is_null) {
+ bound = *view;
+ continue;
+ }
+ int cmp = compare_values(type, *view->value, view->row,
*bound.value, bound.row);
+ if (IsMin ? cmp < 0 : cmp > 0) {
+ bound = *view;
+ }
+ }
+ return bound;
+ }
+
+ /// Finds the best head (smallest for MIN, largest for MAX) among the
three lists; `head` is
+ /// set to nullptr when all of them are exhausted.
+ static void best_head(const TopKArrayView& s_view, size_t s_index, const
TopKArrayView& i_view,
+ size_t i_index, const TopKArrayView& d_view, size_t
d_index,
+ PrimitiveType type, const IColumn** head, size_t*
head_row) {
+ *head = nullptr;
+ const TopKArrayView* views[] = {&s_view, &i_view, &d_view};
+ const size_t indexes[] = {s_index, i_index, d_index};
+ for (size_t v = 0; v != 3; ++v) {
+ if (indexes[v] == views[v]->end) {
+ continue;
+ }
+ if (*head == nullptr) {
+ *head = views[v]->values;
+ *head_row = indexes[v];
+ continue;
+ }
+ int cmp = compare_values(type, *views[v]->values, indexes[v],
**head, *head_row);
+ if (IsMin ? cmp < 0 : cmp > 0) {
+ *head = views[v]->values;
+ *head_row = indexes[v];
+ }
+ }
+ }
+
+ /// Consumes one entry equal to the current head value, returning its
count.
+ static bool consume_equal(const TopKArrayView& view, size_t& index, const
IColumn& head,
+ size_t head_row, PrimitiveType type, int64_t*
count) {
+ if (index == view.end || compare_values(type, *view.values, index,
head, head_row) != 0) {
+ return false;
+ }
+ // The top-K producers never emit a NULL entry, a NULL value or a NULL
count.
+ DCHECK(view.null_map == nullptr || (*view.null_map)[index] == 0);
+ DCHECK(view.value_null_map == nullptr || (*view.value_null_map)[index]
== 0);
Review Comment:
Fixed in e4acad346dd. `check_entry()` runs before any comparison or
consumption and returns `INVALID_ARGUMENT` when an entry, its `value` or its
`count` is NULL. NULL top-level lists are still accepted as empty states
(`order_qt_merge_all_null`), and the NULL-value case is in the regression.
--
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]
