[fix](parquet) Handle nested nullability in predicate scans

be/src/format_v2/parquet/parquet_scan.cpp

@@ -35,7 +35,10 @@
 #include "core/column/column_decimal.h"
 #include "core/column/column_nullable.h"
 #include "core/column/column_vector.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_map.h"
 #include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_struct.h"
 #include "exprs/expr_zonemap_filter.h"
 #include "exprs/vcompound_pred.h"
 #include "exprs/vectorized_fn_call.h"
@@ -103,6 +106,47 @@ bool should_sample_adaptive_predicate(size_t samples, size_t batch_sequence) {
     return samples < WARMUP_SAMPLES || batch_sequence % STEADY_STATE_INTERVAL == 0;
 }
 
+bool types_equal_ignoring_nested_nullability(const DataTypePtr& left, const DataTypePtr& right) {
+    const auto left_type = remove_nullable(left);
+    const auto right_type = remove_nullable(right);
+    if (left_type->get_primitive_type() != right_type->get_primitive_type()) {
+        return false;
+    }
+
+    switch (left_type->get_primitive_type()) {
+    case TYPE_ARRAY: {
+        const auto& left_array = assert_cast<const DataTypeArray&>(*left_type);
+        const auto& right_array = assert_cast<const DataTypeArray&>(*right_type);
+        return types_equal_ignoring_nested_nullability(left_array.get_nested_type(),
+                                                       right_array.get_nested_type());
+    }
+    case TYPE_MAP: {
+        const auto& left_map = assert_cast<const DataTypeMap&>(*left_type);
+        const auto& right_map = assert_cast<const DataTypeMap&>(*right_type);
+        return types_equal_ignoring_nested_nullability(left_map.get_key_type(),
+                                                       right_map.get_key_type()) &&
+               types_equal_ignoring_nested_nullability(left_map.get_value_type(),
+                                                       right_map.get_value_type());
+    }
+    case TYPE_STRUCT: {
+        const auto& left_struct = assert_cast<const DataTypeStruct&>(*left_type);
+        const auto& right_struct = assert_cast<const DataTypeStruct&>(*right_type);
+        if (left_struct.get_elements().size() != right_struct.get_elements().size()) {
+            return false;
+        }
+        for (size_t i = 0; i < left_struct.get_elements().size(); ++i) {
+            if (!types_equal_ignoring_nested_nullability(left_struct.get_element(i),
+                                                         right_struct.get_element(i))) {
+                return false;
+            }
+        }
+        return true;
+    }
+    default:
+        return left_type->equals(*right_type);
+    }
+}
+
 } // namespace detail
 
 #ifdef BE_TEST
@@ -2273,8 +2317,10 @@ Status ParquetScanScheduler::read_filter_columns(int64_t batch_rows,
         DORIS_CHECK(used_direct_reader_filter != nullptr);
         *used_dictionary_filter = false;
         *used_direct_reader_filter = false;
-        DCHECK(remove_nullable(column_reader->type())
-                       ->equals(*remove_nullable(file_block->get_by_position(block_position).type)))
+        // External table schemas may make required Parquet descendants nullable. Preserve the
+        // recursive type and shape checks while ignoring only nullability at every nesting level.
+        DCHECK(detail::types_equal_ignoring_nested_nullability(
+                column_reader->type(), file_block->get_by_position(block_position).type))
                 << column_reader->type()->get_name() << " "
                 << file_block->get_by_position(block_position).type->get_name() << " "
                 << column_reader->name() << " " << file_block->get_by_position(block_position).name;

be/src/format_v2/parquet/parquet_scan.h

@@ -105,6 +105,7 @@ Status select_native_row_groups_by_scan_range(const tparquet::FileMetaData& meta
                                               const ParquetScanRange& scan_range,
                                               std::vector<int64_t>* row_group_first_rows,
                                               std::vector<int>* selected_row_groups);
+bool types_equal_ignoring_nested_nullability(const DataTypePtr& left, const DataTypePtr& right);
 #ifdef BE_TEST
 void reset_physical_leaf_set_build_count();
 size_t physical_leaf_set_build_count();

be/test/format_v2/parquet/parquet_scan_test.cpp

@@ -48,6 +48,7 @@
 #include "core/data_type/data_type_factory.hpp"
 #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/field.h"
 #include "exprs/bloom_filter_func.h"
 #include "exprs/create_predicate_function.h"
@@ -83,6 +84,24 @@
 namespace doris {
 namespace {
 
+TEST(ParquetScanTypeCompatibilityTest, IgnoresOnlyNestedNullability) {
+    const auto int_type = std::make_shared<DataTypeInt32>();
+    const auto string_type = std::make_shared<DataTypeString>();
+    const auto reader_type = std::make_shared<DataTypeStruct>(
+            DataTypes {make_nullable(int_type), make_nullable(string_type)},
+            Strings {"field", "another_field"});
+    const auto block_type = make_nullable(std::make_shared<DataTypeStruct>(
+            DataTypes {int_type, string_type}, Strings {"field", "another_field"}));
+
+    EXPECT_TRUE(format::parquet::detail::types_equal_ignoring_nested_nullability(reader_type,
+                                                                                 block_type));
+
+    const auto incompatible_type = make_nullable(std::make_shared<DataTypeStruct>(
+            DataTypes {int_type, int_type}, Strings {"field", "another_field"}));
+    EXPECT_FALSE(format::parquet::detail::types_equal_ignoring_nested_nullability(
+            reader_type, incompatible_type));
+}
+
 format::LocalColumnIndex field_projection(int32_t column_id) {
     return format::LocalColumnIndex {.index = column_id};
 }