[fix](lambda) Skip lambda evaluation for null containers

be/src/exprs/lambda_function/varray_map_function.cpp

@@ -113,9 +113,9 @@ class ArrayMapFunction : public LambdaFunction {
         // offset column
         MutableColumnPtr array_column_offset;
         size_t nested_array_column_rows = 0;
-        ColumnPtr first_array_offsets = nullptr;
         //2. get the result column from executed expr, and the needed is nested column of array
         std::vector<ColumnPtr> lambda_datas(arguments.size());
+        std::vector<ColumnPtr> lambda_offsets(arguments.size());
         DataTypes lambda_argument_types(arguments.size());
 
         for (int i = 0; i < arguments.size(); ++i) {
@@ -144,35 +144,90 @@ class ArrayMapFunction : public LambdaFunction {
 
             // here is the array column
             const auto& col_array = assert_cast<const ColumnArray&>(*column_array);
+            lambda_offsets[i] = col_array.get_offsets_ptr();
 
             if (i == 0) {
                 nested_array_column_rows = col_array.get_data_ptr()->size();
-                first_array_offsets = col_array.get_offsets_ptr();
                 const auto& off_data = col_array.get_offsets_column();
                 array_column_offset = off_data.clone_resized(col_array.get_offsets_column().size());
                 args_info.offsets_ptr = &col_array.get_offsets();
-            } else {
-                // select array_map((x,y)->x+y,c_array1,[0,1,2,3]) from array_test2;
-                // c_array1: [0,1,2,3,4,5,6,7,8,9]
-                const auto& array_offsets =
-                        assert_cast<const ColumnArray::ColumnOffsets&>(*first_array_offsets)
-                                .get_data();
-                if (nested_array_column_rows != col_array.get_data_ptr()->size() ||
-                    (!array_offsets.empty() &&
-                     memcmp(array_offsets.data(), col_array.get_offsets().data(),
-                            sizeof(array_offsets[0]) * array_offsets.size()) != 0)) {
-                    return Status::InvalidArgument(
-                            "in array map function, the input column size "
-                            "are "
-                            "not equal completely, nested column data rows 1st size is {}, {}th "
-                            "size is {}.",
-                            nested_array_column_rows, i + 1, col_array.get_data_ptr()->size());
-                }
             }
             lambda_datas[i] = col_array.get_data_ptr();
             const auto& col_type = assert_cast<const DataTypeArray&>(*type_array);
             lambda_argument_types[i] = col_type.get_nested_type();
         }
+
+        const auto& first_array_offsets =
+                assert_cast<const ColumnArray::ColumnOffsets&>(*lambda_offsets[0]).get_data();
+        const auto& outside_null_map_data = outside_null_map->get_data();
+        const bool has_outer_null =
+                std::ranges::any_of(outside_null_map_data, [](uint8_t is_null) { return is_null; });
+        bool has_hidden_nested_data = false;
+        if (!has_outer_null) {
+            // select array_map((x,y)->x+y,c_array1,[0,1,2,3]) from array_test2;
+            // c_array1: [0,1,2,3,4,5,6,7,8,9]
+            for (int i = 1; i < arguments.size(); ++i) {
+                const auto& offsets =
+                        assert_cast<const ColumnArray::ColumnOffsets&>(*lambda_offsets[i])
+                                .get_data();
+                if (nested_array_column_rows != lambda_datas[i]->size() ||
+                    (!first_array_offsets.empty() &&
+                     memcmp(first_array_offsets.data(), offsets.data(),
+                            sizeof(first_array_offsets[0]) * first_array_offsets.size()) != 0)) {
+                    return Status::InvalidArgument(
+                            "in array map function, the input column size are not equal "
+                            "completely, nested column data rows 1st size is {}, {}th size is {}.",
+                            nested_array_column_rows, i + 1, lambda_datas[i]->size());
+                }
+            }
+        } else {
+            std::vector<size_t> previous_offsets(arguments.size(), 0);
+            for (size_t row = 0; row < count; ++row) {
+                const size_t first_row_size = first_array_offsets[row] - previous_offsets[0];
+                has_hidden_nested_data |= outside_null_map_data[row] != 0 && first_row_size > 0;
+                for (int i = 1; i < arguments.size(); ++i) {
+                    const auto& offsets =
+                            assert_cast<const ColumnArray::ColumnOffsets&>(*lambda_offsets[i])
+                                    .get_data();
+                    const size_t row_size = offsets[row] - previous_offsets[i];
+                    has_hidden_nested_data |= outside_null_map_data[row] != 0 && row_size > 0;
+                    if (outside_null_map_data[row] == 0 && first_row_size != row_size) {
+                        return Status::InvalidArgument(
+                                "in array map function, the input column size are not equal "
+                                "completely at row {}, 1st size is {}, {}th size is {}.",
+                                row, first_row_size, i + 1, row_size);
+                    }
+                    previous_offsets[i] = offsets[row];
+                }
+                previous_offsets[0] = first_array_offsets[row];
+            }
+        }
+
+        // NULL rows are skipped. If they retain hidden payload, rebuild only result offsets;
+        // the bounded execution path reads each argument through its own original offsets.
+        if (has_hidden_nested_data) {
+            auto res_offsets = ColumnArray::ColumnOffsets::create();
+            auto& res_offsets_data = res_offsets->get_data();
+            res_offsets_data.reserve(count);
+            size_t previous_offset = 0;
+            size_t compacted_rows = 0;
+            for (size_t row = 0; row < count; ++row) {
+                const size_t current_offset = first_array_offsets[row];
+                if (outside_null_map_data[row] == 0) {
+                    const size_t row_size = current_offset - previous_offset;
+                    compacted_rows += row_size;
+                }
+                res_offsets_data.push_back(compacted_rows);
+                previous_offset = current_offset;
+            }
+
+            nested_array_column_rows = compacted_rows;
+            array_column_offset = std::move(res_offsets);
+            args_info.offsets_ptr =
+                    &assert_cast<const ColumnArray::ColumnOffsets&>(*array_column_offset)
+                             .get_data();
+        }
+
         std::set<int> required_input_column_ids;
         children[0]->collect_slot_column_ids(required_input_column_ids);
         context->lambda_execution_context().collect_visible_binding_column_positions(
@@ -238,7 +293,7 @@ class ArrayMapFunction : public LambdaFunction {
         // if column_array is NULL, we know the array_data_column will not write any data,
         // so the column is empty. eg : (x) -> concat('|',x + "1"). if still execute the lambda function, will cause the bolck rows are not equal
         // the x column is empty, but "|" is const literal, size of column is 1, so the block rows is 1, but the x column is empty, will be coredump.
-        if (std::ranges::any_of(lambda_datas, [](const auto& v) { return v->empty(); })) {
+        if (nested_array_column_rows == 0) {
             DataTypePtr nested_type;
             bool is_nullable = result_type->is_nullable();
             if (is_nullable) {
@@ -275,7 +330,8 @@ class ArrayMapFunction : public LambdaFunction {
         // Lambda arguments are already stored contiguously in the input arrays. When all nested
         // rows fit within the direct-execution limit, reuse those columns and only materialize
         // captured outer columns whose values depend on the outer row.
-        if (nested_array_column_rows > 0 && nested_array_column_rows <= lambda_fast_path_rows) {
+        if (!has_hidden_nested_data && nested_array_column_rows > 0 &&
+            nested_array_column_rows <= lambda_fast_path_rows) {
             Block lambda_block;
             PaddedPODArray<IColumn::ColumnIndex> captured_source_row_indices;
             MutableColumns captured_columns(lambda_argument_base);
@@ -369,10 +425,14 @@ class ArrayMapFunction : public LambdaFunction {
                 long max_step = lambda_batch_rows - columns[lambda_argument_base]->size();
                 long current_step = std::min(
                         max_step, (long)(args_info.cur_size - args_info.current_offset_in_array));
-                size_t pos = args_info.array_start + args_info.current_offset_in_array;
                 for (int i = 0; i < arguments.size() && current_step > 0; ++i) {
-                    columns[lambda_argument_base + i]->insert_range_from(*lambda_datas[i], pos,
-                                                                         current_step);
+                    const auto& source_offsets =
+                            assert_cast<const ColumnArray::ColumnOffsets&>(*lambda_offsets[i])
+                                    .get_data();
+                    const size_t source_pos = source_offsets[args_info.current_row_idx - 1] +
+                                              args_info.current_offset_in_array;
+                    columns[lambda_argument_base + i]->insert_range_from(*lambda_datas[i],
+                                                                         source_pos, current_step);
                 }
                 args_info.current_offset_in_array += current_step;
                 if (has_row_dependent_captures) {

be/test/exprs/lambda_function/array_map_function_test.cpp

@@ -583,6 +583,16 @@ static ColumnPtr make_int_array_column(const std::vector<std::vector<int32_t>>&
     return ColumnArray::create(std::move(nullable_int_column), std::move(offsets));
 }
 
+static ColumnPtr make_nullable_int_array_column(const std::vector<std::vector<int32_t>>& rows,
+                                                const std::vector<uint8_t>& outer_null_map) {
+    auto array_column = IColumn::mutate(make_int_array_column(rows));
+    auto null_map = ColumnUInt8::create();
+    for (uint8_t is_null : outer_null_map) {
+        null_map->insert_value(is_null);
+    }
+    return ColumnNullable::create(std::move(array_column), std::move(null_map));
+}
+
 static ColumnPtr make_nested_int_array_column() {
     // Two input rows:
     //   row 0: [[1, 2], [3]]
@@ -1015,6 +1025,102 @@ TEST(ArrayMapFunctionTest, MultiBatchPreservesCaptureMappingAcrossSelectedArrayR
     EXPECT_EQ(values.get_element(total_nested_rows - 1), 1349);
 }
 
+TEST(ArrayMapFunctionTest, HiddenPayloadAfterValidRowUsesOwnOffsetsAcrossBatchesAndSelector) {
+    constexpr int lambda_batch_size = 3;
+    auto int_type = std::make_shared<DataTypeInt32>();
+    auto array_int_type = std::make_shared<DataTypeArray>(int_type);
+    auto nullable_array_int_type = std::make_shared<DataTypeNullable>(array_int_type);
+    std::vector<size_t> observed_batch_sizes;
+
+    auto root =
+            VLambdaFunctionCallExpr::create_shared(make_lambda_call_node(nullable_array_int_type, 3));
+    auto lambda = VLambdaFunctionExpr::create_shared(make_lambda_expr_node(int_type, {"x", "y"}));
+    auto body = std::make_shared<MockAddExpr>(int_type, &observed_batch_sizes);
+    body->add_child(VColumnRef::create_shared(make_column_ref_node(0, "x", int_type)));
+    body->add_child(VColumnRef::create_shared(make_column_ref_node(1, "y", int_type)));
+    lambda->add_child(body);
+    root->add_child(lambda);
+    root->add_child(make_slot_ref(0, "left", nullable_array_int_type));
+    root->add_child(make_slot_ref(1, "right", array_int_type));
+
+    VExprContext context(root);
+    open_expr_with_batch_size(root, &context, lambda_batch_size);
+
+    Block block;
+    block.insert({make_nullable_int_array_column({{-100},
+                                                  {1, 2},
+                                                  {-200},
+                                                  {900, 901, 902},
+                                                  {-300},
+                                                  {3, 4, 5, 6, 7}},
+                                                 {0, 0, 0, 1, 0, 0}),
+                  nullable_array_int_type, "left"});
+    block.insert({make_int_array_column(
+                          {{-10}, {10, 20}, {-20}, {800, 801}, {-30}, {30, 40, 50, 60, 70}}),
+                  array_int_type, "right"});
+
+    Selector selector {1, 3, 5};
+    ColumnPtr result;
+    auto status = root->execute_column(&context, &block, &selector, selector.size(), result);
+    ASSERT_TRUE(status.ok()) << status.to_string();
+
+    ASSERT_EQ(observed_batch_sizes.size(), 3);
+    EXPECT_EQ(observed_batch_sizes[0], 3);
+    EXPECT_EQ(observed_batch_sizes[1], 3);
+    EXPECT_EQ(observed_batch_sizes[2], 1);
+
+    const auto& nullable_result = assert_cast<const ColumnNullable&>(*result);
+    ASSERT_EQ(nullable_result.size(), 3);
+    EXPECT_FALSE(nullable_result.is_null_at(0));
+    EXPECT_TRUE(nullable_result.is_null_at(1));
+    EXPECT_FALSE(nullable_result.is_null_at(2));
+
+    const auto& result_array =
+            assert_cast<const ColumnArray&>(nullable_result.get_nested_column());
+    EXPECT_EQ(result_array.get_offsets()[0], 2);
+    EXPECT_EQ(result_array.get_offsets()[1], 2);
+    EXPECT_EQ(result_array.get_offsets()[2], 7);
+
+    const auto& nullable_values =
+            assert_cast<const ColumnNullable&>(*result_array.get_data_ptr());
+    const auto& values =
+            assert_cast<const ColumnInt32&>(nullable_values.get_nested_column());
+    ASSERT_EQ(values.size(), 7);
+    EXPECT_EQ(values.get_element(0), 11);
+    EXPECT_EQ(values.get_element(1), 22);
+    EXPECT_EQ(values.get_element(2), 33);
+    EXPECT_EQ(values.get_element(3), 44);
+    EXPECT_EQ(values.get_element(4), 55);
+    EXPECT_EQ(values.get_element(5), 66);
+    EXPECT_EQ(values.get_element(6), 77);
+}
+
+TEST(ArrayMapFunctionTest, NonNullLengthMismatchStillReturnsErrorWithOuterNull) {
+    auto int_type = std::make_shared<DataTypeInt32>();
+    auto array_int_type = std::make_shared<DataTypeArray>(int_type);
+    auto nullable_array_int_type = std::make_shared<DataTypeNullable>(array_int_type);
+
+    auto root =
+            VLambdaFunctionCallExpr::create_shared(make_lambda_call_node(nullable_array_int_type, 3));
+    auto lambda =
+            VLambdaFunctionExpr::create_shared(make_lambda_expr_node(int_type, {"x", "y"}));
+    lambda->add_child(std::make_shared<MockBodyExpr>(int_type, "unused_body"));
+    root->add_child(lambda);
+    root->add_child(std::make_shared<MockColumnExpr>(
+            make_nullable_int_array_column({{100, 101}, {1}}, {1, 0}),
+            nullable_array_int_type, "left"));
+    root->add_child(std::make_shared<MockColumnExpr>(
+            make_int_array_column({{200}, {10, 20}}), array_int_type, "right"));
+
+    VExprContext context(root);
+    open_expr(root, &context);
+
+    Block block;
+    ColumnPtr result;
+    auto status = root->execute_column(&context, &block, nullptr, 2, result);
+    EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status.to_string();
+}
+
 TEST(ArrayMapFunctionTest, SparseCapturedColumnUsesColumnNothingForUnusedSlots) {
     auto int_type = std::make_shared<DataTypeInt32>();
     auto uint8_type = std::make_shared<DataTypeUInt8>();

regression-test/data/query_p0/sql_functions/array_functions/test_array_map_function.out

@@ -132,3 +132,7 @@
 -- !select_28 --
 []
 
+-- !array_map_null_container --
+1	\N
+2	[32]
+

regression-test/data/query_p0/sql_functions/map_functions/test_map_lambda.out

@@ -120,3 +120,7 @@ x:a	y:b
 
 -- !map_from_arrays_nested_empty_array --
 {1:[]}
+
+-- !transform_values_null_container --
+1	\N
+2	{1:12}

regression-test/suites/query_p0/sql_functions/array_functions/test_array_map_function.groovy

@@ -162,4 +162,31 @@ suite("test_array_map_function") {
 
         qt_select_27 """ select QC_result_list, array_map(                 x -> concat(                     '|',                     x + "1"                 ),                 QC_result_list             ) FROM db; """
         qt_select_28 """ select array_map((x,y)->x,[],[]); """
+
+        sql "DROP TABLE IF EXISTS array_map_null_container"
+        sql """
+            CREATE TABLE array_map_null_container (
+                id INT,
+                string_values ARRAY<STRING>,
+                int_values ARRAY<INT>
+            ) ENGINE=OLAP
+            DUPLICATE KEY(id)
+            DISTRIBUTED BY HASH(id) BUCKETS 1
+            PROPERTIES ("replication_num" = "1")
+        """
+        sql """
+            INSERT INTO array_map_null_container VALUES
+                (1, ['bad-number'], [100, 101]),
+                (2, ['10'], [20])
+        """
+        sql "SET enable_strict_cast = true"
+        sql "SET short_circuit_evaluation = false"
+        order_qt_array_map_null_container """
+            SELECT id,
+                   array_map((x, y) -> cast(x AS INT) + y + id,
+                             if(id = 1, cast(NULL AS ARRAY<STRING>), string_values),
+                             int_values)
+            FROM array_map_null_container
+            ORDER BY id
+        """
 }

regression-test/suites/query_p0/sql_functions/map_functions/test_map_lambda.groovy

@@ -397,4 +397,30 @@ suite("test_map_lambda", "p0") {
         """
         exception "must return a non-nullable struct with exactly two fields"
     }
+
+    sql "drop table if exists test_map_lambda_null_container"
+    sql """
+        create table test_map_lambda_null_container (
+            id int,
+            m map<int, string>
+        )
+        duplicate key(id)
+        distributed by hash(id) buckets 1
+        properties("replication_num" = "1")
+    """
+    sql """
+        insert into test_map_lambda_null_container values
+            (1, map(1, 'bad-number')),
+            (2, map(1, '10'))
+    """
+    sql "set enable_strict_cast = true"
+    sql "set short_circuit_evaluation = false"
+    order_qt_transform_values_null_container """
+        select id,
+               transform_values(
+                   (k, v) -> cast(v as int) + id,
+                   if(id = 1, cast(null as map<int, string>), m))
+        from test_map_lambda_null_container
+        order by id
+    """
 }