[fix](be) Reject incompatible Arrow arrays without crashing Stream Load (#69872)

### What problem does this PR solve?

Issue Number: close #69858

Related PR: None

Problem Summary:

The shared Arrow BOOLEAN and DECIMAL SerDes dereference unchecked
`dynamic_cast` results. Passing an `Int32Array` to the BOOLEAN reader or
a `StringArray` to the DECIMAL64 reader can therefore terminate the BE
instead of returning a load error. MAP and STRUCT have the same
unchecked input-cast pattern.

Reject incompatible arrays before accessing buffers, decimal metadata,
or values, independently of `enable_arrow_input_validation`. Reject
mismatched STRUCT field counts before indexing children. Add the column
name and source/destination types to V1/V2 Arrow load errors.

The current V2 Stream Load reader already binds supported input types
with safe Arrow casts. Preserve those conversions, including valid
integer-to-BOOLEAN and string-to-DECIMAL inputs, and test the shared
SerDes directly for incompatible arrays.

be/src/core/data_type_serde/data_type_decimal_serde.cpp

@@ -921,6 +921,11 @@ Status DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
     // Decimal<Int128I> for deicmalv3
     if constexpr (T == TYPE_DECIMALV2) {
         const auto* concrete_array = dynamic_cast<const arrow::DecimalArray*>(arrow_array);
+        // A string or a different decimal width cannot be read as Decimal128, including NULLs.
+        if (concrete_array == nullptr) {
+            return Status::InvalidArgument("Expected Arrow Decimal128Array for {}, got {}",
+                                           get_name(), arrow_array->type()->ToString());
+        }
         const auto* arrow_decimal_type =
                 static_cast<const arrow::DecimalType*>(arrow_array->type().get());
         const auto arrow_scale = arrow_decimal_type->scale();
@@ -948,6 +953,11 @@ Status DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
         }
     } else if constexpr (T == TYPE_DECIMAL32 || T == TYPE_DECIMAL64 || T == TYPE_DECIMAL128I) {
         const auto* concrete_array = dynamic_cast<const arrow::DecimalArray*>(arrow_array);
+        // A string or a different decimal width cannot be read as Decimal128, including NULLs.
+        if (concrete_array == nullptr) {
+            return Status::InvalidArgument("Expected Arrow Decimal128Array for {}, got {}",
+                                           get_name(), arrow_array->type()->ToString());
+        }
         for (auto value_i = start; value_i < end; ++value_i) {
             const auto* value = concrete_array->Value(value_i);
             auto decimal_value = unaligned_load<FieldType>(value);
@@ -955,6 +965,11 @@ Status DataTypeDecimalSerDe<T>::read_column_from_arrow(IColumn& column,
         }
     } else if constexpr (T == TYPE_DECIMAL256) {
         const auto* concrete_array = dynamic_cast<const arrow::Decimal256Array*>(arrow_array);
+        // Reject incompatible arrays before Value(), independently of buffer validation.
+        if (concrete_array == nullptr) {
+            return Status::InvalidArgument("Expected Arrow Decimal256Array, got {}",
+                                           arrow_array->type()->ToString());
+        }
         for (auto value_i = start; value_i < end; ++value_i) {
             auto decimal_value = unaligned_load<FieldType>(concrete_array->Value(value_i));
             column_data.emplace_back(decimal_value);

be/src/core/data_type_serde/data_type_map_serde.cpp

@@ -549,6 +549,11 @@ Status DataTypeMapSerDe::read_column_from_arrow(IColumn& column, const arrow::Ar
     auto& column_map = static_cast<ColumnMap&>(column);
     auto& offsets_data = column_map.get_offsets();
     const auto* concrete_map = dynamic_cast<const arrow::MapArray*>(arrow_array);
+    // A valid non-MAP Arrow column has no map offsets; reject it before accessing them.
+    if (concrete_map == nullptr) {
+        return Status::InvalidArgument("Expected Arrow MapArray, got {}",
+                                       arrow_array->type()->ToString());
+    }
     auto arrow_offsets_array = concrete_map->offsets();
     auto* arrow_offsets = dynamic_cast<arrow::Int32Array*>(arrow_offsets_array.get());
     if (config::enable_arrow_input_validation) {

be/src/core/data_type_serde/data_type_number_serde.cpp

@@ -956,6 +956,11 @@ Status DataTypeNumberSerDe<T>::read_column_from_arrow(IColumn& column,
     // now uint8 for bool
     if constexpr (T == TYPE_BOOLEAN) {
         const auto* concrete_array = dynamic_cast<const arrow::BooleanArray*>(arrow_array);
+        // An Int32Array sent to a BOOLEAN column must fail even when buffer validation is off.
+        if (concrete_array == nullptr) {
+            return Status::InvalidArgument("Expected Arrow BooleanArray, got {}",
+                                           arrow_array->type()->ToString());
+        }
         if (config::enable_arrow_input_validation) {
             check_arrow_boolean_buffer(*concrete_array);
         }

be/src/core/data_type_serde/data_type_struct_serde.cpp

@@ -551,7 +551,16 @@ Status DataTypeStructSerDe::read_column_from_arrow(IColumn& column, const arrow:
     }
     auto& struct_column = static_cast<ColumnStruct&>(column);
     const auto* concrete_struct = dynamic_cast<const arrow::StructArray*>(arrow_array);
-    DCHECK_EQ(struct_column.tuple_size(), concrete_struct->num_fields());
+    // Reject non-STRUCT input before reading its fields, even with buffer validation disabled.
+    if (concrete_struct == nullptr) {
+        return Status::InvalidArgument("Expected Arrow StructArray, got {}",
+                                       arrow_array->type()->ToString());
+    }
+    // A shorter input struct would otherwise index past its children; extra fields lose data.
+    if (struct_column.tuple_size() != concrete_struct->num_fields()) {
+        return Status::InvalidArgument("Expected Arrow StructArray with {} fields, got {}",
+                                       struct_column.tuple_size(), concrete_struct->num_fields());
+    }
     for (auto i = 0; i < struct_column.tuple_size(); ++i) {
         RETURN_IF_ERROR(elem_serdes_ptrs[i]->read_column_from_arrow(
                 struct_column.get_column(i), concrete_struct->field(i).get(), start, end, ctz));

be/src/format/arrow/arrow_stream_reader.cpp

@@ -114,10 +114,15 @@ Status ArrowStreamReader::_do_get_next_block(Block* block, size_t* read_rows, bo
                                                  column_name_in_block, column_name);
                 }
 
-                RETURN_IF_ERROR(
-                        columns_guard.get_datatype_by_position(c)
-                                ->get_serde()
-                                ->read_column_from_arrow(*columns[c], column, 0, num_rows, _ctzz));
+                const auto& data_type = columns_guard.get_datatype_by_position(c);
+                auto status = data_type->get_serde()->read_column_from_arrow(*columns[c], column, 0,
+                                                                             num_rows, _ctzz);
+                // Type mismatches are load errors; retain their code and identify the input column.
+                if (!status.ok()) {
+                    return status.prepend("Arrow column '" + column_name + "' (" +
+                                          column->type()->ToString() + " -> " +
+                                          data_type->get_name() + "): ");
+                }
             } catch (Exception& e) {
                 return Status::InternalError("Failed to convert from arrow to block: {}", e.what());
             }

be/src/format_v2/arrow/arrow_stream_reader.cpp

@@ -429,10 +429,16 @@ Status ArrowStreamReader::_prepare_arrays() {
         RETURN_IF_ERROR(normalize_arrow_array(_batch->column(_arrow_positions[id.value()]),
                                               _arrow_pool.get(), &array));
         arrow::compute::ExecContext context(_arrow_pool.get());
-        std::shared_ptr<arrow::DataType> binding;
-        RETURN_DORIS_STATUS_IF_RESULT_ERROR(
-                binding, input_binding_type(array->type(), _arrow_types[id.value()],
-                                            _file_slots[id.value()]->type()));
+        auto binding_result = input_binding_type(array->type(), _arrow_types[id.value()],
+                                                 _file_slots[id.value()]->type());
+        // A STRUCT field-count mismatch needs the same column context as a failed value cast.
+        if (!binding_result.ok()) {
+            return Status::InvalidArgument(
+                    "Cannot decode Arrow column '{}' ({} -> {}): {}",
+                    _file_slots[id.value()]->col_name(), array->type()->ToString(),
+                    _arrow_types[id.value()]->ToString(), binding_result.status().ToString());
+        }
+        auto binding = std::move(binding_result).ValueUnsafe();
         // Bind differing numeric widths before concrete SerDe array casts. Timestamp units
         // stay on the wire even inside containers, so Doris retains its own rounding semantics.
         if (!array->type()->Equals(binding)) {
@@ -443,8 +449,9 @@ Status ArrowStreamReader::_prepare_arrays() {
             options.allow_invalid_utf8 = true;
             auto converted = arrow::compute::Cast(array, binding, options, &context);
             if (!converted.ok()) {
-                return Status::InvalidArgument("Cannot decode Arrow column '{}': {}",
+                return Status::InvalidArgument("Cannot decode Arrow column '{}' ({} -> {}): {}",
                                                _file_slots[id.value()]->col_name(),
+                                               array->type()->ToString(), binding->ToString(),
                                                converted.status().ToString());
             }
             array = converted->make_array();

be/test/core/data_type_serde/data_type_serde_arrow_validation_test.cpp

@@ -35,6 +35,13 @@
 #include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
 #include "core/column/column_varbinary.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_map.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_struct.h"
 #include "core/data_type/primitive_type.h"
 #include "core/data_type_serde/data_type_array_serde.h"
 #include "core/data_type_serde/data_type_datev2_serde.h"
@@ -514,4 +521,142 @@ TEST(DataTypeSerDeArrowValidationTest, RejectsMapOffsetsBeyondKeysLength) {
             "map offsets beyond keys length should be rejected");
 }
 
+// These arrays are well formed: the failure is their pairing with the destination type,
+// not a malformed buffer. Empty/all-NULL input must also reject the incompatible schema.
+TEST(DataTypeSerDeArrowValidationTest, RejectsIncompatibleArrayTypes) {
+    const DataTypes types = {
+            std::make_shared<DataTypeUInt8>(),
+            std::make_shared<DataTypeDecimalV2>(27, 9),
+            std::make_shared<DataTypeDecimal32>(9, 2),
+            std::make_shared<DataTypeDecimal64>(18, 2),
+            std::make_shared<DataTypeDecimal128>(38, 2),
+            std::make_shared<DataTypeDecimal256>(76, 2),
+            std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeUInt8>())),
+            std::make_shared<DataTypeMap>(make_nullable(std::make_shared<DataTypeString>()),
+                                          make_nullable(std::make_shared<DataTypeUInt8>())),
+            std::make_shared<DataTypeStruct>(
+                    DataTypes {make_nullable(std::make_shared<DataTypeUInt8>())}, Strings {"b"})};
+    arrow::Int32Builder builder;
+    ASSERT_TRUE(builder.Append(1).ok());
+    std::shared_ptr<arrow::Array> non_null;
+    ASSERT_TRUE(builder.Finish(&non_null).ok());
+    for (const bool validation : {false, true}) {
+        ScopedArrowInputValidation scoped_validation(validation);
+        for (const auto& type : types) {
+            for (const bool nullable : {false, true}) {
+                const auto target = nullable ? make_nullable(type) : type;
+                for (const auto length : {0, 1}) {
+                    const auto nulls = arrow::MakeArrayOfNull(arrow::int32(), length);
+                    ASSERT_TRUE(nulls.ok());
+                    for (const auto& input : {non_null, *nulls}) {
+                        SCOPED_TRACE(target->get_name() + ": " + input->ToString());
+                        auto column = target->create_column();
+                        const auto status = target->get_serde()->read_column_from_arrow(
+                                *column, input.get(), 0, input->length(), cctz::utc_time_zone());
+                        EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+                    }
+                }
+            }
+        }
+    }
+}
+
+TEST(DataTypeSerDeArrowValidationTest, RejectsStringAndWrongWidthDecimalArrays) {
+    const DataTypes types = {
+            std::make_shared<DataTypeDecimalV2>(27, 9), std::make_shared<DataTypeDecimal32>(9, 2),
+            std::make_shared<DataTypeDecimal64>(18, 2), std::make_shared<DataTypeDecimal128>(38, 2),
+            std::make_shared<DataTypeDecimal256>(76, 2)};
+    arrow::StringBuilder builder;
+    ASSERT_TRUE(builder.Append("1.50").ok());
+    std::shared_ptr<arrow::Array> strings;
+    ASSERT_TRUE(builder.Finish(&strings).ok());
+    for (const bool validation : {false, true}) {
+        ScopedArrowInputValidation scoped_validation(validation);
+        for (const auto& type : types) {
+            const auto wrong_width = type->get_primitive_type() == TYPE_DECIMAL256
+                                             ? arrow::decimal128(10, 2)
+                                             : arrow::decimal256(76, 2);
+            auto decimal = arrow::MakeArrayOfNull(wrong_width, 1);
+            ASSERT_TRUE(decimal.ok());
+            for (const bool nullable : {false, true}) {
+                const auto target = nullable ? make_nullable(type) : type;
+                for (const auto& input : {strings, *decimal}) {
+                    auto column = target->create_column();
+                    const auto status = target->get_serde()->read_column_from_arrow(
+                            *column, input.get(), 0, input->length(), cctz::utc_time_zone());
+                    EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+                }
+            }
+        }
+    }
+}
+
+TEST(DataTypeSerDeArrowValidationTest, RejectsStructFieldCountMismatch) {
+    const auto boolean = make_nullable(std::make_shared<DataTypeUInt8>());
+    const auto type = std::make_shared<DataTypeStruct>(DataTypes {boolean}, Strings {"b"});
+    for (const bool validation : {false, true}) {
+        ScopedArrowInputValidation scoped_validation(validation);
+        for (const auto& arrow_type :
+             {arrow::struct_({}), arrow::struct_({arrow::field("b", arrow::boolean()),
+                                                  arrow::field("extra", arrow::boolean())})}) {
+            auto input = arrow::MakeArrayOfNull(arrow_type, 1);
+            ASSERT_TRUE(input.ok());
+            auto column = type->create_column();
+            const auto status = type->get_serde()->read_column_from_arrow(*column, input->get(), 0,
+                                                                          1, cctz::utc_time_zone());
+            EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+        }
+    }
+}
+
+TEST(DataTypeSerDeArrowValidationTest, RejectsIncompatibleNestedArrayTypes) {
+    const auto boolean = make_nullable(std::make_shared<DataTypeUInt8>());
+    const DataTypes targets = {
+            std::make_shared<DataTypeArray>(boolean),
+            std::make_shared<DataTypeMap>(make_nullable(std::make_shared<DataTypeString>()),
+                                          boolean),
+            std::make_shared<DataTypeStruct>(DataTypes {boolean}, Strings {"b"})};
+    const std::vector<std::shared_ptr<arrow::DataType>> inputs = {
+            arrow::list(arrow::int32()), arrow::map(arrow::utf8(), arrow::int32()),
+            arrow::struct_({arrow::field("b", arrow::int32())})};
+    for (const bool validation : {false, true}) {
+        ScopedArrowInputValidation scoped_validation(validation);
+        for (size_t i = 0; i < targets.size(); ++i) {
+            auto input = arrow::MakeArrayOfNull(inputs[i], 1);
+            ASSERT_TRUE(input.ok());
+            auto column = targets[i]->create_column();
+            const auto status = targets[i]->get_serde()->read_column_from_arrow(
+                    *column, input->get(), 0, 1, cctz::utc_time_zone());
+            EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+        }
+    }
+}
+
+TEST(DataTypeSerDeArrowValidationTest, AcceptsMatchingArrayTypesWithValidationOnAndOff) {
+    const DataTypes targets = {std::make_shared<DataTypeUInt8>(),
+                               std::make_shared<DataTypeDecimalV2>(27, 9),
+                               std::make_shared<DataTypeDecimal32>(9, 2),
+                               std::make_shared<DataTypeDecimal64>(18, 2),
+                               std::make_shared<DataTypeDecimal128>(38, 2),
+                               std::make_shared<DataTypeDecimal256>(76, 2)};
+    const std::vector<std::shared_ptr<arrow::DataType>> inputs = {
+            arrow::boolean(),         arrow::decimal128(27, 9), arrow::decimal128(9, 2),
+            arrow::decimal128(18, 2), arrow::decimal128(38, 2), arrow::decimal256(76, 2)};
+    for (const bool validation : {false, true}) {
+        ScopedArrowInputValidation scoped_validation(validation);
+        for (size_t i = 0; i < targets.size(); ++i) {
+            const auto target = make_nullable(targets[i]);
+            for (const auto length : {0, 2}) {
+                auto input = arrow::MakeArrayOfNull(inputs[i], length);
+                ASSERT_TRUE(input.ok());
+                auto column = target->create_column();
+                const auto status = target->get_serde()->read_column_from_arrow(
+                        *column, input->get(), 0, length, cctz::utc_time_zone());
+                EXPECT_TRUE(status.ok()) << status;
+                EXPECT_EQ(column->size(), length);
+            }
+        }
+    }
+}
+
 } // namespace doris

be/test/format_v2/table/load_reader_test.cpp

@@ -1238,6 +1238,63 @@ TEST_F(LoadArrowPipeTest, NumericIntStillRejectsAnOverflowingUInt32) {
     ASSERT_TRUE(reader.close().ok());
 }
 
+TEST_F(LoadArrowPipeTest, ConvertsTheReportedInt32BooleanInput) {
+    set_column_type(make_nullable(std::make_shared<DataTypeUInt8>()));
+    arrow::Int32Builder builder;
+    ASSERT_TRUE(builder.AppendValues({1, 0}).ok());
+    auto values = builder.Finish();
+    ASSERT_TRUE(values.ok());
+    append_batch(arrow::RecordBatch::Make(
+            arrow::schema({arrow::field("raw_value", arrow::int32())}), 2, {*values}));
+    ASSERT_TRUE(_pipe->finish().ok());
+    LoadTableReader reader(source.get(), destination.get());
+    ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+    Block output(destination->slots(), 0);
+    bool eos = false;
+    const auto status = reader.get_block(&output, &eos);
+    ASSERT_TRUE(status.ok()) << status;
+    ASSERT_EQ(output.rows(), 2);
+    const auto& column = assert_cast<const ColumnNullable&>(*output.get_by_position(0).column);
+    EXPECT_EQ(column.get_nested_column().get_int(0), 1);
+    EXPECT_EQ(column.get_nested_column().get_int(1), 0);
+    ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, InvalidBooleanInputIncludesTheColumnAndBothTypes) {
+    set_column_type(make_nullable(std::make_shared<DataTypeUInt8>()));
+    append_stream({"not-a-boolean"}, "raw_value");
+    ASSERT_TRUE(_pipe->finish().ok());
+    LoadTableReader reader(source.get(), destination.get());
+    ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+    Block output(destination->slots(), 0);
+    bool eos = false;
+    const auto status = reader.get_block(&output, &eos);
+    EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+    EXPECT_NE(status.to_string().find("Arrow column 'raw_value' (string -> bool)"),
+              std::string::npos)
+            << status;
+    ASSERT_TRUE(reader.close().ok());
+}
+
+TEST_F(LoadArrowPipeTest, StructFieldCountErrorIncludesTheColumnAndBothTypes) {
+    set_column_type(make_nullable(std::make_shared<DataTypeStruct>(
+            DataTypes {make_nullable(std::make_shared<DataTypeUInt8>())}, Strings {"b"})));
+    auto input = arrow::MakeArrayOfNull(arrow::struct_({}), 1);
+    ASSERT_TRUE(input.ok());
+    append_batch(arrow::RecordBatch::Make(
+            arrow::schema({arrow::field("raw_value", (*input)->type())}), 1, {*input}));
+    ASSERT_TRUE(_pipe->finish().ok());
+    LoadTableReader reader(source.get(), destination.get());
+    ASSERT_TRUE(prepare(reader, FileFormat::ARROW).ok());
+    Block output(destination->slots(), 0);
+    bool eos = false;
+    const auto status = reader.get_block(&output, &eos);
+    EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT) << status;
+    EXPECT_NE(status.to_string().find("Arrow column 'raw_value'"), std::string::npos) << status;
+    EXPECT_NE(status.to_string().find("struct<> -> struct<b: bool>"), std::string::npos) << status;
+    ASSERT_TRUE(reader.close().ok());
+}
+
 TEST_F(LoadArrowPipeTest, ReadsCompressedIpcBatches) {
     arrow::StringBuilder builder;
     ASSERT_TRUE(builder.Append("compressed").ok());

regression-test/data/load_p0/stream_load/arrow_type_validation/generate.py

@@ -0,0 +1,98 @@
+# 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.
+
+"""Regenerate checked-in IPC fixtures with Python 3.12 and pyarrow==21.0.0.
+
+The regression suite only reads the .arrow files and does not require Python.
+Every stream includes an empty batch, followed by one-row batches, to exercise EOF,
+nullable values and batch transitions. Failed loads can follow a complete valid stream.
+"""
+
+from decimal import Decimal
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.ipc as ipc
+
+ROOT = Path(__file__).resolve().parent
+
+
+def stream(data_type, values):
+    schema = pa.schema([pa.field("k", pa.int32(), False), pa.field("v", data_type)])
+    sink = pa.BufferOutputStream()
+    with ipc.new_stream(sink, schema) as writer:
+        writer.write_batch(
+            pa.record_batch(
+                [pa.array([], pa.int32()), pa.array([], data_type)], schema=schema
+            )
+        )
+        for key, value in enumerate(values, 1):
+            writer.write_batch(
+                pa.record_batch(
+                    [pa.array([key], pa.int32()), pa.array([value], data_type)], schema=schema
+                )
+            )
+    return sink.getvalue().to_pybytes()
+
+
+def write(name, data_type, values):
+    (ROOT / (name + ".arrow")).write_bytes(stream(data_type, values))
+
+
+TYPES = {
+    "bool": (pa.bool_(), [True, False]),
+    "decimal128": (pa.decimal128(38, 2), [Decimal("1.50"), Decimal("-2.25")]),
+    "decimal256": (pa.decimal256(76, 2), [Decimal("1.50"), Decimal("-2.25")]),
+    "array": (pa.list_(pa.bool_()), [[True, False], []]),
+    "map": (pa.map_(pa.string(), pa.bool_()), [[("x", True)], []]),
+    "struct": (pa.struct([("b", pa.bool_())]), [{"b": True}, {"b": False}]),
+}
+for name, (data_type, values) in TYPES.items():
+    write(name + "_good", data_type, values)
+    write(name + "_nullable", data_type, values + [None])
+    write(name + "_null", data_type, [None])
+
+# These are the original issue's wire types; V2 safely casts their valid values.
+write("int32", pa.int32(), [1, 0])
+write("int32_null", pa.int32(), [None])
+write("decimal_string", pa.string(), ["1.50", "-2.25"])
+write("bad_bool", pa.string(), ["not-a-boolean"])
+write("bad_decimal", pa.string(), ["not-a-decimal"])
+# Exceed the declared precision without overflowing the source Arrow decimal's bit width.
+write("overflow_decimal128", pa.decimal256(76, 2), [Decimal("1" + "0" * 40 + ".00")])
+write("overflow_decimal256", pa.string(), ["1" + "0" * 74 + ".00"])
+write("scale_loss_decimal", pa.decimal128(10, 3), [Decimal("1.234")])
+write("bad_array", pa.list_(pa.string()), [["not-a-boolean"]])
+write("bad_map", pa.map_(pa.string(), pa.string()), [[("x", "not-a-boolean")]])
+write("bad_struct", pa.struct([("b", pa.string())]), [{"b": "not-a-boolean"}])
+write("short_struct", pa.struct([]), [{}])
+write(
+    "long_struct",
+    pa.struct([("b", pa.bool_()), ("extra", pa.bool_())]),
+    [{"b": True, "extra": False}],
+)
+write("array_nested_null", pa.list_(pa.bool_()), [[True, None]])
+write("map_nested_null", pa.map_(pa.string(), pa.bool_()), [[("x", None)]])
+write("struct_nested_null", pa.struct([("b", pa.bool_())]), [{"b": None}])
+# The first stream produces rows before the second stream fails. Neither may commit.
+(ROOT / "bool_late_failure.arrow").write_bytes(
+    stream(pa.bool_(), [True]) + stream(pa.string(), ["not-a-boolean"])
+)
+(ROOT / "decimal_late_failure.arrow").write_bytes(
+    stream(pa.decimal128(38, 2), [Decimal("1.50")])
+    + stream(pa.string(), ["not-a-decimal"])
+)

regression-test/data/load_p0/stream_load/test_arrow_stream_load_type_validation.out

@@ -0,0 +1,391 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !bool_required_bad --
+0
+
+-- !bool_required_rollback --
+0
+
+-- !bool_required_not_null --
+0
+
+-- !bool_required_recovery --
+1	true
+2	false
+
+-- !bool_required_conversion_0 --
+1	true
+2	false
+
+-- !bool_nullable_bad --
+0
+
+-- !bool_nullable_rollback --
+0
+
+-- !bool_nullable_recovery --
+1	true
+2	false
+3	\N
+
+-- !bool_nullable_conversion_0 --
+1	true
+2	false
+
+-- !bool_nullable_typed_null --
+1	\N
+
+-- !decimal32_required_bad --
+0
+
+-- !decimal32_required_overflow --
+0
+
+-- !decimal32_required_scale_loss --
+0
+
+-- !decimal32_required_integer_precision --
+0
+
+-- !decimal32_required_rollback --
+0
+
+-- !decimal32_required_not_null --
+0
+
+-- !decimal32_required_recovery --
+1	1.50
+2	-2.25
+
+-- !decimal32_required_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal32_required_conversion_1 --
+1	1.50
+2	-2.25
+
+-- !decimal32_nullable_bad --
+0
+
+-- !decimal32_nullable_overflow --
+0
+
+-- !decimal32_nullable_scale_loss --
+0
+
+-- !decimal32_nullable_integer_precision --
+0
+
+-- !decimal32_nullable_null_integer_precision --
+0
+
+-- !decimal32_nullable_rollback --
+0
+
+-- !decimal32_nullable_recovery --
+1	1.50
+2	-2.25
+3	\N
+
+-- !decimal32_nullable_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal32_nullable_conversion_1 --
+1	1.50
+2	-2.25
+
+-- !decimal32_nullable_typed_null --
+1	\N
+
+-- !decimal64_required_bad --
+0
+
+-- !decimal64_required_overflow --
+0
+
+-- !decimal64_required_scale_loss --
+0
+
+-- !decimal64_required_rollback --
+0
+
+-- !decimal64_required_not_null --
+0
+
+-- !decimal64_required_recovery --
+1	1.50
+2	-2.25
+
+-- !decimal64_required_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal64_required_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal64_required_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal64_nullable_bad --
+0
+
+-- !decimal64_nullable_overflow --
+0
+
+-- !decimal64_nullable_scale_loss --
+0
+
+-- !decimal64_nullable_rollback --
+0
+
+-- !decimal64_nullable_recovery --
+1	1.50
+2	-2.25
+3	\N
+
+-- !decimal64_nullable_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal64_nullable_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal64_nullable_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal64_nullable_typed_null --
+1	\N
+
+-- !decimal128_required_bad --
+0
+
+-- !decimal128_required_overflow --
+0
+
+-- !decimal128_required_scale_loss --
+0
+
+-- !decimal128_required_rollback --
+0
+
+-- !decimal128_required_not_null --
+0
+
+-- !decimal128_required_recovery --
+1	1.50
+2	-2.25
+
+-- !decimal128_required_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal128_required_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal128_required_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal128_nullable_bad --
+0
+
+-- !decimal128_nullable_overflow --
+0
+
+-- !decimal128_nullable_scale_loss --
+0
+
+-- !decimal128_nullable_rollback --
+0
+
+-- !decimal128_nullable_recovery --
+1	1.50
+2	-2.25
+3	\N
+
+-- !decimal128_nullable_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal128_nullable_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal128_nullable_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal128_nullable_typed_null --
+1	\N
+
+-- !decimal256_required_bad --
+0
+
+-- !decimal256_required_overflow --
+0
+
+-- !decimal256_required_scale_loss --
+0
+
+-- !decimal256_required_rollback --
+0
+
+-- !decimal256_required_not_null --
+0
+
+-- !decimal256_required_recovery --
+1	1.50
+2	-2.25
+
+-- !decimal256_required_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal256_required_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal256_required_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal256_nullable_bad --
+0
+
+-- !decimal256_nullable_overflow --
+0
+
+-- !decimal256_nullable_scale_loss --
+0
+
+-- !decimal256_nullable_rollback --
+0
+
+-- !decimal256_nullable_recovery --
+1	1.50
+2	-2.25
+3	\N
+
+-- !decimal256_nullable_conversion_0 --
+1	1.50
+2	-2.25
+
+-- !decimal256_nullable_conversion_1 --
+1	1.00
+2	0.00
+
+-- !decimal256_nullable_conversion_2 --
+1	1.50
+2	-2.25
+
+-- !decimal256_nullable_typed_null --
+1	\N
+
+-- !array_required_bad --
+0
+
+-- !array_required_container --
+0
+
+-- !array_required_not_null --
+0
+
+-- !array_required_recovery --
+1	[1, 0]
+2	[]
+
+-- !array_required_conversion_0 --
+1	[1, null]
+
+-- !array_nullable_bad --
+0
+
+-- !array_nullable_container --
+0
+
+-- !array_nullable_recovery --
+1	[1, 0]
+2	[]
+3	\N
+
+-- !array_nullable_conversion_0 --
+1	[1, null]
+
+-- !map_required_bad --
+0
+
+-- !map_required_container --
+0
+
+-- !map_required_not_null --
+0
+
+-- !map_required_recovery --
+1	{"x":1}
+2	{}
+
+-- !map_required_conversion_0 --
+1	{"x":null}
+
+-- !map_nullable_bad --
+0
+
+-- !map_nullable_container --
+0
+
+-- !map_nullable_recovery --
+1	{"x":1}
+2	{}
+3	\N
+
+-- !map_nullable_conversion_0 --
+1	{"x":null}
+
+-- !struct_required_bad --
+0
+
+-- !struct_required_container --
+0
+
+-- !struct_required_short_struct --
+0
+
+-- !struct_required_long_struct --
+0
+
+-- !struct_required_not_null --
+0
+
+-- !struct_required_recovery --
+1	{"b":1}
+2	{"b":0}
+
+-- !struct_required_conversion_0 --
+1	{"b":null}
+
+-- !struct_nullable_bad --
+0
+
+-- !struct_nullable_container --
+0
+
+-- !struct_nullable_short_struct --
+0
+
+-- !struct_nullable_long_struct --
+0
+
+-- !struct_nullable_recovery --
+1	{"b":1}
+2	{"b":0}
+3	\N
+
+-- !struct_nullable_conversion_0 --
+1	{"b":null}
+

regression-test/suites/load_p0/stream_load/test_arrow_stream_load_type_validation.groovy

@@ -0,0 +1,148 @@
+// 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_arrow_stream_load_type_validation", "p0") {
+    sql "set enable_decimal256 = true"
+    def backendIps = [:]
+    def backendPorts = [:]
+    getBackendIpHttpPort(backendIps, backendPorts)
+    // Pin every failed load and its recovery load to the same BE, without HTTP retries.
+    def backendId = backendIps.keySet().first()
+    def backendIp = backendIps[backendId]
+    def backendPort = backendPorts[backendId] as int
+
+    def load = { tableName, fixture, expectedError, typeError = true ->
+        streamLoad {
+            table tableName
+            directToBe backendIp, backendPort
+            retryIfHttpError false
+            set 'format', 'arrow'
+            set 'columns', 'k,v'
+            set 'strict_mode', 'true'
+            set 'max_filter_ratio', '0'
+            file "arrow_type_validation/${fixture}.arrow"
+            time 30000
+            check { result, exception, startTime, endTime ->
+                if (exception != null) {
+                    throw exception
+                }
+                def response = parseJson(result)
+                if (expectedError != null) {
+                    assertEquals('Fail', response.Status)
+                    if (typeError) {
+                        assertTrue(response.Message.contains("Arrow column 'v'"), response.Message)
+                        assertTrue(response.Message.contains('->'), response.Message)
+                    }
+                    assertTrue(response.Message.toLowerCase().contains(expectedError.toLowerCase()), response.Message)
+                } else {
+                    assertEquals('Success', response.Status)
+                    assertEquals(0, response.NumberFilteredRows)
+                }
+            }
+        }
+    }
+
+    // Different decimal storage widths share the decimal128 wire representation.
+    def cases = [
+        [name: 'bool', type: 'BOOLEAN', good: 'bool', bad: 'bad_bool', reason: 'boolean',
+            conversions: ['int32'], late: 'bool_late_failure'],
+        [name: 'decimal32', type: 'DECIMAL(9,2)', good: 'decimal128', bad: 'bad_decimal',
+            reason: 'decimal', conversions: ['decimal_string', 'decimal256_good'],
+            late: 'decimal_late_failure'],
+        [name: 'decimal64', type: 'DECIMAL(18,2)', good: 'decimal128', bad: 'bad_decimal',
+            reason: 'decimal', conversions: ['decimal_string', 'int32', 'decimal256_good'],
+            late: 'decimal_late_failure'],
+        [name: 'decimal128', type: 'DECIMAL(38,2)', good: 'decimal128', bad: 'bad_decimal',
+            reason: 'decimal', conversions: ['decimal_string', 'int32', 'decimal256_good'],
+            late: 'decimal_late_failure'],
+        [name: 'decimal256', type: 'DECIMAL(76,2)', good: 'decimal256', bad: 'bad_decimal',
+            reason: 'decimal', conversions: ['decimal_string', 'int32', 'decimal128_good'],
+            late: 'decimal_late_failure'],
+        [name: 'array', type: 'ARRAY<BOOLEAN>', good: 'array', bad: 'bad_array',
+            reason: 'boolean', conversions: ['array_nested_null']],
+        [name: 'map', type: 'MAP<STRING,BOOLEAN>', good: 'map', bad: 'bad_map',
+            reason: 'boolean', conversions: ['map_nested_null']],
+        [name: 'struct', type: 'STRUCT<b:BOOLEAN>', good: 'struct', bad: 'bad_struct',
+            reason: 'boolean', conversions: ['struct_nested_null']],
+    ]
+    cases.each { c ->
+        [false, true].each { nullable ->
+            // This is a type/nullability matrix, so each case retains a separate table for debugging.
+            def suffix = "${c.name}_${nullable ? 'nullable' : 'required'}"
+            def tableName = "test_arrow_type_${suffix}"
+            sql "DROP TABLE IF EXISTS ${tableName}"
+            sql """CREATE TABLE ${tableName} (k INT NOT NULL, v ${c.type} ${nullable ? 'NULL' : 'NOT NULL'})
+                DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+                PROPERTIES ("replication_num" = "1")"""
+
+            load(tableName, c.bad, c.reason)
+            "qt_${suffix}_bad"("SELECT count(*) FROM ${tableName}")
+            if (c.name.startsWith('decimal')) {
+                load(tableName, c.name == 'decimal256' ? 'overflow_decimal256' : 'overflow_decimal128', 'precision')
+                "qt_${suffix}_overflow"("SELECT count(*) FROM ${tableName}")
+                load(tableName, 'scale_loss_decimal', 'data loss')
+                "qt_${suffix}_scale_loss"("SELECT count(*) FROM ${tableName}")
+            }
+            if (c.name == 'decimal32') {
+                // Arrow requires precision 12 to cast int32 to a scale-two decimal, even for 0/1.
+                load(tableName, 'int32', 'precision')
+                "qt_${suffix}_integer_precision"("SELECT count(*) FROM ${tableName}")
+                if (nullable) {
+                    load(tableName, 'int32_null', 'precision')
+                    "qt_${suffix}_null_integer_precision"("SELECT count(*) FROM ${tableName}")
+                }
+            }
+            if (c.name in ['array', 'map', 'struct']) {
+                load(tableName, 'int32', '')
+                "qt_${suffix}_container"("SELECT count(*) FROM ${tableName}")
+            }
+            if (c.name == 'struct') {
+                ['short_struct', 'long_struct'].each { fixture ->
+                    load(tableName, fixture, 'field count')
+                    "qt_${suffix}_${fixture}"("SELECT count(*) FROM ${tableName}")
+                }
+            }
+            if (c.late != null) {
+                load(tableName, c.late, c.reason)
+                "qt_${suffix}_rollback"("SELECT count(*) FROM ${tableName}")
+            }
+
+            if (!nullable) {
+                // Correctly typed NULL input still has to obey the destination NOT NULL constraint.
+                load(tableName, "${c.good}_null", '', false)
+                "qt_${suffix}_not_null"("SELECT count(*) FROM ${tableName}")
+            }
+            // An empty batch and multiple nonempty batches precede the final NULL in nullable controls.
+            load(tableName, "${c.good}_${nullable ? 'nullable' : 'good'}", null)
+            sql 'sync'
+            "order_qt_${suffix}_recovery"("SELECT * FROM ${tableName} ORDER BY k")
+
+            c.conversions.eachWithIndex { fixture, index ->
+                sql "TRUNCATE TABLE ${tableName}"
+                load(tableName, fixture, null)
+                sql 'sync'
+                "order_qt_${suffix}_conversion_${index}"("SELECT * FROM ${tableName} ORDER BY k")
+            }
+            if (nullable && !(c.name in ['array', 'map', 'struct'])) {
+                sql "TRUNCATE TABLE ${tableName}"
+                load(tableName, c.name == 'decimal32' ? 'decimal128_null' : 'int32_null', null)
+                sql 'sync'
+                "order_qt_${suffix}_typed_null"("SELECT * FROM ${tableName} ORDER BY k")
+            }
+        }
+    }
+}