This is an automated email from the ASF dual-hosted git repository. HappenLee pushed a commit to branch fix/doris-28490-agg-state-cast in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9daea77946c976635c93ff30cedfcf3c645fd52e Author: happenlee <[email protected]> AuthorDate: Wed Sep 9 01:59:47 2026 +0800 [fix](agg-state) Reject casts from ordinary values to aggregate states ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Casting raw strings or variants to AGG_STATE can admit invalid serialized bytes and produce incorrect results or out-of-bounds reads when those states are merged. Reject non-state inputs in FE cast checks and in BE before lowering aggregate states to their serialized types. Preserve typed NULLs, identical states and state-combinator argument coercion. Also reject parsing ordinary values into containers of aggregate states. Update raw state reload tests to expect rejection and verify typed state copies remain supported. ### Release note Explicit and implicit casts from ordinary values to AGG_STATE are rejected. This also rejects raw state reloads that rely on VARCHAR-to-AGG_STATE casts, including Parquet Stream Load. Generate states with aggregate combinators or copy existing states of the matching type. ### Check List (For Author) - Test: 41 FE unit tests passed; FE Checkstyle, BE header hygiene, clang-format 16, and syntax checks of both changed C++ files passed. Full BE unit-test build blocked by a missing AWS SDK header; cluster regression not run. clang-tidy attempted but blocked by toolchain resource and existing NOLINT errors. - Behavior changed: Yes, reject casts from ordinary values to aggregate states. - Does this need documentation: Yes, document the raw state import restriction. --- be/src/exprs/function/cast/function_cast.cpp | 6 ++ .../exprs/function/cast/cast_to_agg_state_test.cpp | 88 ++++++++++++++++ .../nereids/rules/expression/check/CheckCast.java | 11 +- .../rules/expression/check/AggStateCastTest.java | 113 +++++++++++++++++++++ .../agg_state/test_agg_state_cast.groovy | 50 +++++++++ .../agg_state/test_outfile_agg_state.groovy | 23 ++++- .../agg_state_array/test_outfile_agg_array.groovy | 21 +++- .../test_outfile_agg_state_bitmap.groovy | 25 +++-- 8 files changed, 320 insertions(+), 17 deletions(-) diff --git a/be/src/exprs/function/cast/function_cast.cpp b/be/src/exprs/function/cast/function_cast.cpp index 5561a3a76de..d093b23df31 100644 --- a/be/src/exprs/function/cast/function_cast.cpp +++ b/be/src/exprs/function/cast/function_cast.cpp @@ -241,6 +241,12 @@ WrapperType prepare_remove_nullable(FunctionContext* context, const DataTypePtr& // NOLINTNEXTLINE(readability-function-size) WrapperType prepare_impl(FunctionContext* context, const DataTypePtr& origin_from_type, const DataTypePtr& origin_to_type) { + // Check before lowering AggState to its serialized type, which can match ordinary input. + if (origin_to_type->get_primitive_type() == TYPE_AGG_STATE && + origin_from_type->get_primitive_type() != TYPE_AGG_STATE) { + return CastWrapper::create_unsupport_wrapper( + "Cast to AggState only supports AggState input"); + } auto to_type = get_serialized_type(origin_to_type); auto from_type = get_serialized_type(origin_from_type); if (from_type->equals(*to_type)) { diff --git a/be/test/exprs/function/cast/cast_to_agg_state_test.cpp b/be/test/exprs/function/cast/cast_to_agg_state_test.cpp new file mode 100644 index 00000000000..89b0189a236 --- /dev/null +++ b/be/test/exprs/function/cast/cast_to_agg_state_test.cpp @@ -0,0 +1,88 @@ +// 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 "agent/be_exec_version_manager.h" +#include "core/column/column_nullable.h" +#include "core/data_type/data_type_agg_state.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_string.h" +#include "core/data_type/data_type_variant.h" +#include "exprs/function/cast/cast_base.h" + +namespace doris { +namespace { + +DataTypePtr create_state_type() { + return std::make_shared<DataTypeAggState>(DataTypes {std::make_shared<DataTypeString>()}, true, + "group_concat", + BeExecVersionManager::get_newest_version()); +} + +} // namespace + +TEST(CastToAggStateTest, RejectOrdinaryInput) { + auto state_type = create_state_type(); + DataTypes input_types {std::make_shared<DataTypeString>(), std::make_shared<DataTypeVariant>(), + std::make_shared<DataTypeInt32>()}; + for (const auto& input_type : input_types) { + for (bool nullable : {false, true}) { + DataTypePtr from_type = nullable ? make_nullable(input_type) : input_type; + DataTypePtr to_type = nullable ? make_nullable(state_type) : state_type; + auto input = from_type->create_column(); + input->insert_default(); + Block block {{std::move(input), from_type, "input"}, {nullptr, to_type, "result"}}; + auto wrapper = CastWrapper::prepare_unpack_dictionaries(nullptr, from_type, to_type); + auto status = wrapper(nullptr, block, {0}, 1, 1, nullptr); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("Cast to AggState only supports AggState input"), + std::string::npos); + } + } +} + +TEST(CastToAggStateTest, PreserveIdenticalState) { + auto state_type = create_state_type(); + auto function = assert_cast<const DataTypeAggState*>(state_type.get())->get_nested_function(); + Arena arena; + auto* place = reinterpret_cast<AggregateDataPtr>( + arena.aligned_alloc(function->size_of_data(), function->align_of_data())); + function->create(place); + auto input = state_type->create_column(); + function->serialize_without_key_to_column(place, *input); + function->destroy(place); + + Block block {{input->get_ptr(), state_type, "input"}, {nullptr, state_type, "result"}}; + auto wrapper = CastWrapper::prepare_unpack_dictionaries(nullptr, state_type, state_type); + ASSERT_TRUE(wrapper(nullptr, block, {0}, 1, 1, nullptr).ok()); + EXPECT_EQ(block.get_by_position(1).column, block.get_by_position(0).column); +} + +TEST(CastToAggStateTest, PreserveNullLiteral) { + auto null_type = std::make_shared<DataTypeUInt8>(); + null_type->set_null_literal(true); + auto from_type = make_nullable(null_type); + auto to_type = make_nullable(create_state_type()); + Block block {{nullptr, to_type, "result"}}; + auto wrapper = CastWrapper::prepare_unpack_dictionaries(nullptr, from_type, to_type); + ASSERT_TRUE(wrapper(nullptr, block, {}, 0, 1, nullptr).ok()); + EXPECT_TRUE(block.get_by_position(0).column->is_null_at(0)); +} + +} // namespace doris diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/check/CheckCast.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/check/CheckCast.java index a363226b3d4..f8d4f94dd09 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/check/CheckCast.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/check/CheckCast.java @@ -322,7 +322,6 @@ public class CheckCast implements ExpressionPatternRuleFactory { allowedTypes.add(MapType.class); allowedTypes.add(StructType.class); allowedTypes.add(VariantType.class); - allowedTypes.add(AggStateType.class); allowedTypes.add(QuantileStateType.class); } @@ -358,6 +357,16 @@ public class CheckCast implements ExpressionPatternRuleFactory { */ public static boolean check(DataType originalType, DataType targetType, boolean isStrictMode, boolean looseAggState) { + // Serialized values do not carry the aggregate function's state invariants. + // Matching containers are checked recursively below; parsing a value into a container + // of states must not bypass the same restriction (for example, Variant -> Array<AggState>). + if (checkTypeContainsType(targetType, AggStateType.class) + && !originalType.isAggStateType() && !originalType.isNullType() + && !(originalType.isArrayType() && targetType.isArrayType()) + && !(originalType.isMapType() && targetType.isMapType()) + && !(originalType.isStructType() && targetType.isStructType())) { + return false; + } if (originalType.isVariantType() && (targetType instanceof PrimitiveType || targetType.isArrayType())) { // variant could cast to primitive types and array return true; diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/expression/check/AggStateCastTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/expression/check/AggStateCastTest.java new file mode 100644 index 00000000000..115ea0456ec --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/expression/check/AggStateCastTest.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.rules.expression.check; + +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.types.AggStateType; +import org.apache.doris.nereids.types.ArrayType; +import org.apache.doris.nereids.types.BigIntType; +import org.apache.doris.nereids.types.BitmapType; +import org.apache.doris.nereids.types.BooleanType; +import org.apache.doris.nereids.types.CharType; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.DateV2Type; +import org.apache.doris.nereids.types.DoubleType; +import org.apache.doris.nereids.types.HllType; +import org.apache.doris.nereids.types.IntegerType; +import org.apache.doris.nereids.types.JsonType; +import org.apache.doris.nereids.types.MapType; +import org.apache.doris.nereids.types.NullType; +import org.apache.doris.nereids.types.QuantileStateType; +import org.apache.doris.nereids.types.StringType; +import org.apache.doris.nereids.types.StructType; +import org.apache.doris.nereids.types.VarBinaryType; +import org.apache.doris.nereids.types.VarcharType; +import org.apache.doris.nereids.types.VariantType; +import org.apache.doris.nereids.util.MemoTestUtils; +import org.apache.doris.nereids.util.PlanChecker; +import org.apache.doris.nereids.util.TypeCoercionUtils; + +import com.google.common.collect.ImmutableList; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class AggStateCastTest { + private final AggStateType stateType = new AggStateType("sum_map", + ImmutableList.of(MapType.of(StringType.INSTANCE, IntegerType.INSTANCE)), + ImmutableList.of(true), false); + + @Test + void testRejectNonStateInputs() { + for (DataType source : ImmutableList.of(CharType.SYSTEM_DEFAULT, VarcharType.SYSTEM_DEFAULT, + StringType.INSTANCE, VariantType.INSTANCE, VarBinaryType.INSTANCE, + IntegerType.INSTANCE, BigIntType.INSTANCE, DoubleType.INSTANCE, BooleanType.INSTANCE, + DateV2Type.INSTANCE, JsonType.INSTANCE, ArrayType.of(IntegerType.INSTANCE), + MapType.of(StringType.INSTANCE, IntegerType.INSTANCE), StructType.SYSTEM_DEFAULT, + BitmapType.INSTANCE, HllType.INSTANCE, QuantileStateType.INSTANCE)) { + for (boolean strict : new boolean[] {false, true}) { + Assertions.assertFalse(CheckCast.check(source, stateType, strict), source.toSql()); + Assertions.assertFalse(CheckCast.checkWithLooseAggState(source, stateType, strict), source.toSql()); + } + Assertions.assertThrows(AnalysisException.class, + () -> TypeCoercionUtils.checkCanCastTo(source, stateType), source.toSql()); + } + } + + @Test + void testPreserveNullAndIdenticalState() { + for (boolean strict : new boolean[] {false, true}) { + Assertions.assertTrue(CheckCast.check(NullType.INSTANCE, stateType, strict)); + Assertions.assertTrue(CheckCast.check(stateType, stateType, strict)); + } + } + + @Test + void testRejectNonStateToNestedState() { + for (boolean strict : new boolean[] {false, true}) { + for (DataType source : ImmutableList.of(StringType.INSTANCE, VariantType.INSTANCE, JsonType.INSTANCE)) { + Assertions.assertFalse(CheckCast.check(source, ArrayType.of(stateType), strict)); + Assertions.assertFalse(CheckCast.check(source, + MapType.of(StringType.INSTANCE, stateType), strict)); + } + Assertions.assertFalse(CheckCast.check(ArrayType.of(StringType.INSTANCE), + ArrayType.of(stateType), strict)); + Assertions.assertTrue(CheckCast.check(ArrayType.of(stateType), ArrayType.of(stateType), strict)); + } + } + + @Test + void testRejectRawStateSql() { + for (String input : ImmutableList.of( + "unhex('00020101016101010000000000000001010161010300000000000000')", + "cast('invalid state' as variant)", "cast('invalid state' as varbinary)", "1")) { + Assertions.assertThrows(AnalysisException.class, + () -> PlanChecker.from(MemoTestUtils.createConnectContext()).analyze( + "select cast(" + input + " as agg_state<sum_map(map<string,int>)>)")); + } + } + + @Test + void testPreserveStateConstructionAndCoercion() { + PlanChecker.from(MemoTestUtils.createConnectContext()).analyze( + "select avg_merge(cast(avg_state(cast(1 as int)) as agg_state<avg(bigint)>))"); + PlanChecker.from(MemoTestUtils.createConnectContext()).analyze( + "select cast(null as agg_state<avg(int)>)"); + PlanChecker.from(MemoTestUtils.createConnectContext()).analyze( + "select sum_map_merge(sum_map_state(map('a', 1)))"); + } +} diff --git a/regression-test/suites/datatype_p0/agg_state/test_agg_state_cast.groovy b/regression-test/suites/datatype_p0/agg_state/test_agg_state_cast.groovy new file mode 100644 index 00000000000..7d924e5f8d4 --- /dev/null +++ b/regression-test/suites/datatype_p0/agg_state/test_agg_state_cast.groovy @@ -0,0 +1,50 @@ +// 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("test_agg_state_cast") { + sql "set enable_agg_state = true" + sql "drop table if exists test_agg_state_cast" + sql """ + create table test_agg_state_cast ( + id int, + s agg_state<sum_map(map<string,int>)> generic + ) aggregate key(id) + distributed by hash(id) buckets 1 + properties("replication_num" = "1") + """ + + for (def strict : [false, true]) { + sql "set enable_strict_cast = ${strict}" + // Both well-formed and malformed bytes must be rejected without inspecting their payload. + for (def input : [ + "unhex('000101010161010100000000000000')", + "unhex('00020101016101010000000000000001010161010300000000000000')", + "cast('invalid state' as variant)", + "cast('invalid state' as varbinary)", + "1" + ]) { + test { + sql "select cast(${input} as agg_state<sum_map(map<string,int>)>)" + exception "cast" + } + test { + sql "insert into test_agg_state_cast values (1, ${input})" + exception "cast" + } + } + } +} diff --git a/regression-test/suites/query_p0/outfile/agg_state/test_outfile_agg_state.groovy b/regression-test/suites/query_p0/outfile/agg_state/test_outfile_agg_state.groovy index 6b132b177c0..5264f9b6730 100644 --- a/regression-test/suites/query_p0/outfile/agg_state/test_outfile_agg_state.groovy +++ b/regression-test/suites/query_p0/outfile/agg_state/test_outfile_agg_state.groovy @@ -62,11 +62,24 @@ suite("test_outfile_agg_state") { properties("replication_num" = "1"); """ - def filePath=testHelper.localDir+"/*" - cmd """ - curl --location-trusted -u ${context.config.jdbcUser}:${context.config.jdbcPassword} -H "format:PARQUET" -H "Expect:100-continue" -T ${filePath} -XPUT http://${context.config.feHttpAddress}/api/regression_test_query_p0_outfile_agg_state/a_table2/_stream_load - """ - Thread.sleep(10000) + // Raw serialized states cannot be cast from file columns to AGG_STATE. + def files = new File(testHelper.localDir).listFiles().findAll { it.isFile() } + assertTrue(!files.isEmpty()) + files.each { exportedFile -> + streamLoad { + table "a_table2" + set "format", "parquet" + file exportedFile.absolutePath + check { result, exception, startTime, endTime -> + assertTrue(exception == null) + def response = parseJson(result) + assertEquals("Fail", response.Status) + assertTrue(response.Message.contains("cast"), response.Message) + } + } + } + // Copying typed states between tables remains supported. + sql "insert into a_table2 select * from a_table" qt_test "select k1,max_by_merge(k2),group_concat_merge(k3) from a_table2 group by k1 order by k1;" testHelper.close() diff --git a/regression-test/suites/query_p0/outfile/agg_state_array/test_outfile_agg_array.groovy b/regression-test/suites/query_p0/outfile/agg_state_array/test_outfile_agg_array.groovy index 21ee6aacaeb..d02ee6ef5cb 100644 --- a/regression-test/suites/query_p0/outfile/agg_state_array/test_outfile_agg_array.groovy +++ b/regression-test/suites/query_p0/outfile/agg_state_array/test_outfile_agg_array.groovy @@ -62,11 +62,22 @@ suite("test_outfile_agg_state_array") { properties("replication_num" = "1"); """ - def filePath=testHelper.localDir+"/tmp_*" - cmd """ - curl --location-trusted -u ${context.config.jdbcUser}:${context.config.jdbcPassword} -H "format:PARQUET" -H "Expect:100-continue" -T ${filePath} http://${context.config.feHttpAddress}/api/regression_test_query_p0_outfile_agg_state_array/a_table2/_stream_load - """ - Thread.sleep(10000) + // Raw serialized states cannot be cast from file columns to AGG_STATE. + def files = new File(testHelper.localDir).listFiles().findAll { it.isFile() } + assertTrue(!files.isEmpty()) + files.each { exportedFile -> + streamLoad { + table "a_table2" + set "format", "parquet" + file exportedFile.absolutePath + check { result, exception, startTime, endTime -> + assertTrue(exception == null) + def response = parseJson(result) + assertEquals("Fail", response.Status) + assertTrue(response.Message.contains("cast"), response.Message) + } + } + } qt_test "select k1,max_by_merge(k2),group_concat_merge(k3) from a_table2 group by k1 order by k1;" testHelper.close() } diff --git a/regression-test/suites/query_p0/outfile/agg_state_bitmap/test_outfile_agg_state_bitmap.groovy b/regression-test/suites/query_p0/outfile/agg_state_bitmap/test_outfile_agg_state_bitmap.groovy index caacf90f624..72a8535f5c5 100644 --- a/regression-test/suites/query_p0/outfile/agg_state_bitmap/test_outfile_agg_state_bitmap.groovy +++ b/regression-test/suites/query_p0/outfile/agg_state_bitmap/test_outfile_agg_state_bitmap.groovy @@ -61,11 +61,24 @@ suite("test_outfile_agg_state_bitmap") { properties("replication_num" = "1"); """ - def filePath=testHelper.localDir+"/tmp_*" - cmd """ - curl --location-trusted -u ${context.config.jdbcUser}:${context.config.jdbcPassword} -H "format:PARQUET" -H "Expect:100-continue" -T ${filePath} http://${context.config.feHttpAddress}/api/regression_test_query_p0_outfile_agg_state_bitmap/a_table2/_stream_load - """ - Thread.sleep(10000) - qt_test "select k1,bitmap_to_string(bitmap_union_merge(k2)) from a_table group by k1 order by k1;" + // Raw serialized states cannot be cast from file columns to AGG_STATE. + def files = new File(testHelper.localDir).listFiles().findAll { it.isFile() } + assertTrue(!files.isEmpty()) + files.each { exportedFile -> + streamLoad { + table "a_table2" + set "format", "parquet" + file exportedFile.absolutePath + check { result, exception, startTime, endTime -> + assertTrue(exception == null) + def response = parseJson(result) + assertEquals("Fail", response.Status) + assertTrue(response.Message.contains("cast"), response.Message) + } + } + } + // Copying typed states between tables remains supported. + sql "insert into a_table2 select * from a_table" + qt_test "select k1,bitmap_to_string(bitmap_union_merge(k2)) from a_table2 group by k1 order by k1;" testHelper.close() } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
