[fix](iceberg) Allow optional position delete fields without nulls

be/src/format/table/iceberg_delete_file_reader_helper.cpp

@@ -30,8 +30,10 @@
 #include "core/block/block.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column_dictionary.h"
+#include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
 #include "core/column/column_vector.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 "exec/common/endian.h"
@@ -90,9 +92,26 @@ Status visit_position_delete_block(const Block& block, size_t read_rows,
         return Status::InternalError("Position delete block is missing required columns");
     }
 
-    const auto* pos_column =
-            assert_cast<const ColumnInt64*>(block.get_by_position(pos_it->second).column.get());
-    const auto* path_column = block.get_by_position(path_it->second).column.get();
+    ColumnPtr path_column_ptr = block.get_by_position(path_it->second).column;
+    ColumnPtr pos_column_ptr = block.get_by_position(pos_it->second).column;
+    if (const auto* nullable_col = check_and_get_column<ColumnNullable>(*path_column_ptr);
+        nullable_col != nullptr) {
+        if (nullable_col->has_null(0, read_rows)) {
+            return Status::Corruption(
+                    "Iceberg position delete column file_path contains null values");
+        }
+        path_column_ptr = remove_nullable(path_column_ptr);
+    }
+    if (const auto* nullable_col = check_and_get_column<ColumnNullable>(*pos_column_ptr);
+        nullable_col != nullptr) {
+        if (nullable_col->has_null(0, read_rows)) {
+            return Status::Corruption("Iceberg position delete column pos contains null values");
+        }
+        pos_column_ptr = remove_nullable(pos_column_ptr);
+    }
+
+    const auto* pos_column = assert_cast<const ColumnInt64*>(pos_column_ptr.get());
+    const auto* path_column = path_column_ptr.get();
 
     if (const auto* string_column = check_and_get_column<ColumnString>(path_column);
         string_column != nullptr) {
@@ -252,16 +271,15 @@ Status read_iceberg_position_delete_file(const TIcebergDeleteFileDesc& delete_fi
         while (!eof) {
             Block block;
             if (dictionary_coded) {
-                block.insert(ColumnWithTypeAndName(ColumnDictI32::create(),
-                                                   std::make_shared<DataTypeString>(),
-                                                   ICEBERG_FILE_PATH));
+                block.insert(ColumnWithTypeAndName(
+                        ColumnNullable::create(ColumnDictI32::create(), ColumnUInt8::create()),
+                        make_nullable(std::make_shared<DataTypeString>()), ICEBERG_FILE_PATH));
             } else {
-                block.insert(ColumnWithTypeAndName(ColumnString::create(),
-                                                   std::make_shared<DataTypeString>(),
-                                                   ICEBERG_FILE_PATH));
+                block.insert(ColumnWithTypeAndName(
+                        make_nullable(std::make_shared<DataTypeString>()), ICEBERG_FILE_PATH));
             }
-            block.insert(ColumnWithTypeAndName(ColumnInt64::create(),
-                                               std::make_shared<DataTypeInt64>(), ICEBERG_ROW_POS));
+            block.insert(ColumnWithTypeAndName(make_nullable(std::make_shared<DataTypeInt64>()),
+                                               ICEBERG_ROW_POS));
             size_t read_rows = 0;
             RETURN_IF_ERROR(reader.get_next_block(&block, &read_rows, &eof));
             RETURN_IF_ERROR(visit_position_delete_block(block, read_rows, visitor));

be/src/format/table/iceberg_reader.cpp

@@ -37,9 +37,11 @@
 #include "core/block/block.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
+#include "core/column/column_nullable.h"
 #include "core/column/column_string.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_nullable.h"
 #include "core/data_type/define_primitive_type.h"
 #include "core/data_type/primitive_type.h"
 #include "core/string_ref.h"
@@ -386,23 +388,26 @@ Status IcebergParquetReader::_read_position_delete_file(const TFileRangeDesc* de
             break;
         }
     }
-    DataTypePtr data_type_file_path {new DataTypeString};
-    DataTypePtr data_type_pos {new DataTypeInt64};
+    DataTypePtr data_type_file_path = make_nullable(std::make_shared<DataTypeString>());
+    DataTypePtr data_type_pos = make_nullable(std::make_shared<DataTypeInt64>());
     bool eof = false;
     while (!eof) {
-        Block block = {dictionary_coded
-                               ? ColumnWithTypeAndName {ColumnDictI32::create(),
-                                                        data_type_file_path, ICEBERG_FILE_PATH}
-                               : ColumnWithTypeAndName {data_type_file_path, ICEBERG_FILE_PATH},
-
-                       {data_type_pos, ICEBERG_ROW_POS}};
+        Block block = {
+                dictionary_coded
+                        ? ColumnWithTypeAndName {ColumnNullable::create(ColumnDictI32::create(),
+                                                                        ColumnUInt8::create()),
+                                                 data_type_file_path, ICEBERG_FILE_PATH}
+                        : ColumnWithTypeAndName {data_type_file_path, ICEBERG_FILE_PATH},
+
+                {data_type_pos, ICEBERG_ROW_POS}};
         size_t read_rows = 0;
         RETURN_IF_ERROR(parquet_delete_reader.get_next_block(&block, &read_rows, &eof));
 
         if (read_rows <= 0) {
             break;
         }
-        _gen_position_delete_file_range(block, position_delete, read_rows, dictionary_coded);
+        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows,
+                                                        dictionary_coded));
     }
     return Status::OK();
 };
@@ -650,7 +655,7 @@ Status IcebergOrcReader::_read_position_delete_file(const TFileRangeDesc* delete
         size_t read_rows = 0;
         RETURN_IF_ERROR(orc_delete_reader.get_next_block(&block, &read_rows, &eof));
 
-        _gen_position_delete_file_range(block, position_delete, read_rows, false);
+        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows, false));
     }
     return Status::OK();
 }

be/src/format/table/iceberg_reader_mixin.h

@@ -247,8 +247,9 @@ class IcebergReaderMixin : public BaseReader, public TableSchemaChangeHelper {
     PositionDeleteRange _get_range(const ColumnString& file_path_column);
     static void _sort_delete_rows(const std::vector<std::vector<int64_t>*>& delete_rows_array,
                                   int64_t num_delete_rows, std::vector<int64_t>& result);
-    void _gen_position_delete_file_range(Block& block, DeleteFile* position_delete,
-                                         size_t read_rows, bool file_path_column_dictionary_coded);
+    Status _gen_position_delete_file_range(Block& block, DeleteFile* position_delete,
+                                           size_t read_rows,
+                                           bool file_path_column_dictionary_coded);
     void _generate_equality_delete_block(Block* block,
                                          const std::vector<std::string>& equality_delete_col_names,
                                          const std::vector<DataTypePtr>& equality_delete_col_types);
@@ -776,14 +777,29 @@ void IcebergReaderMixin<BaseReader>::_sort_delete_rows(
 }
 
 template <typename BaseReader>
-void IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
+Status IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
         Block& block, DeleteFile* position_delete, size_t read_rows,
         bool file_path_column_dictionary_coded) {
     SCOPED_TIMER(_iceberg_profile.parse_delete_file_time);
     auto name_to_pos_map = block.get_name_to_pos_map();
     ColumnPtr path_column = block.get_by_position(name_to_pos_map[ICEBERG_FILE_PATH]).column;
     DCHECK_EQ(path_column->size(), read_rows);
     ColumnPtr pos_column = block.get_by_position(name_to_pos_map[ICEBERG_ROW_POS]).column;
+    if (const auto* nullable_col = check_and_get_column<ColumnNullable>(*path_column);
+        nullable_col != nullptr) {
+        if (nullable_col->has_null(0, read_rows)) {
+            return Status::Corruption(
+                    "Iceberg position delete column file_path contains null values");
+        }
+        path_column = remove_nullable(path_column);
+    }
+    if (const auto* nullable_col = check_and_get_column<ColumnNullable>(*pos_column);
+        nullable_col != nullptr) {
+        if (nullable_col->has_null(0, read_rows)) {
+            return Status::Corruption("Iceberg position delete column pos contains null values");
+        }
+        pos_column = remove_nullable(pos_column);
+    }
     using ColumnType = typename PrimitiveTypeTraits<TYPE_BIGINT>::ColumnType;
     const int64_t* src_data = assert_cast<const ColumnType&>(*pos_column).get_data().data();
     PositionDeleteRange range;
@@ -810,6 +826,7 @@ void IcebergReaderMixin<BaseReader>::_gen_position_delete_file_range(
         int64_t* dest_position = &(*delete_rows)[origin_size];
         memcpy(dest_position, cpy_start, cpy_count * sizeof(int64_t));
     }
+    return Status::OK();
 }
 
 template <typename BaseReader>

be/src/format_v2/table/iceberg_reader.cpp

@@ -150,10 +150,20 @@ Status IcebergTableReader::PositionDeleteRowsCollector::collect(const Block& blo
     if (read_rows == 0) {
         return Status::OK();
     }
-    const auto& file_path_column = assert_cast<const ColumnString&>(
-            *remove_nullable((block.get_by_position(ICEBERG_FILE_PATH_BLOCK_POSITION).column)));
-    const auto& pos_column = assert_cast<const ColumnInt64&>(
-            *remove_nullable(block.get_by_position(ICEBERG_ROW_POS_BLOCK_POSITION).column));
+    const auto& file_path_column_ptr =
+            block.get_by_position(ICEBERG_FILE_PATH_BLOCK_POSITION).column;
+    const auto& pos_column_ptr = block.get_by_position(ICEBERG_ROW_POS_BLOCK_POSITION).column;
+    if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*file_path_column_ptr);
+        nullable_column != nullptr && nullable_column->has_null(0, read_rows)) {
+        return Status::Corruption("Iceberg position delete column file_path contains null values");
+    }
+    if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*pos_column_ptr);
+        nullable_column != nullptr && nullable_column->has_null(0, read_rows)) {
+        return Status::Corruption("Iceberg position delete column pos contains null values");
+    }
+    const auto& file_path_column =
+            assert_cast<const ColumnString&>(*remove_nullable(file_path_column_ptr));
+    const auto& pos_column = assert_cast<const ColumnInt64&>(*remove_nullable(pos_column_ptr));
     for (size_t row = 0; row < read_rows; ++row) {
         const auto file_path = file_path_column.get_data_at(row).to_string();
         if (file_path == _data_file_path) {

be/test/format/table/iceberg/iceberg_delete_file_reader_helper_test.cpp

@@ -17,13 +17,17 @@
 
 #include "format/table/iceberg_delete_file_reader_helper.h"
 
+#include <arrow/api.h>
+#include <arrow/io/api.h>
 #include <gen_cpp/PlanNodes_types.h>
 #include <gen_cpp/Types_types.h>
 #include <gtest/gtest.h>
+#include <parquet/arrow/writer.h>
 
 #include <cstring>
 #include <filesystem>
 #include <fstream>
+#include <optional>
 #include <string>
 #include <thread>
 #include <unordered_map>
@@ -99,6 +103,45 @@ int64_t write_iceberg_deletion_vector_file(const std::string& file_path,
     return static_cast<int64_t>(blob.size());
 }
 
+std::string temp_parquet_path(const std::string& filename) {
+    return (std::filesystem::temp_directory_path() / filename).string();
+}
+
+void write_position_delete_parquet(const std::string& path,
+                                   const std::vector<std::optional<std::string>>& file_paths,
+                                   const std::vector<std::optional<int64_t>>& positions) {
+    ASSERT_EQ(file_paths.size(), positions.size());
+
+    arrow::StringBuilder path_builder;
+    arrow::Int64Builder pos_builder;
+    for (size_t i = 0; i < file_paths.size(); ++i) {
+        if (file_paths[i].has_value()) {
+            ASSERT_TRUE(path_builder.Append(*file_paths[i]).ok());
+        } else {
+            ASSERT_TRUE(path_builder.AppendNull().ok());
+        }
+        if (positions[i].has_value()) {
+            ASSERT_TRUE(pos_builder.Append(*positions[i]).ok());
+        } else {
+            ASSERT_TRUE(pos_builder.AppendNull().ok());
+        }
+    }
+
+    std::shared_ptr<arrow::Array> path_array;
+    std::shared_ptr<arrow::Array> pos_array;
+    ASSERT_TRUE(path_builder.Finish(&path_array).ok());
+    ASSERT_TRUE(pos_builder.Finish(&pos_array).ok());
+
+    auto schema = arrow::schema({arrow::field("file_path", arrow::utf8(), true),
+                                 arrow::field("pos", arrow::int64(), true)});
+    auto table = arrow::Table::Make(schema, {path_array, pos_array});
+
+    std::shared_ptr<arrow::io::FileOutputStream> output;
+    ASSERT_TRUE(arrow::io::FileOutputStream::Open(path).Value(&output).ok());
+    PARQUET_THROW_NOT_OK(
+            parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), output, 1024));
+}
+
 IcebergDeleteFileReaderOptions make_delete_file_reader_options(
         RuntimeState* state, RuntimeProfile* profile, const TFileScanRangeParams* scan_params,
         io::IOContext* io_ctx) {
@@ -110,6 +153,21 @@ IcebergDeleteFileReaderOptions make_delete_file_reader_options(
     };
 }
 
+IcebergDeleteFileReaderOptions delete_reader_options(RuntimeState* runtime_state,
+                                                     RuntimeProfile* profile,
+                                                     TFileScanRangeParams* scan_params,
+                                                     IcebergDeleteFileIOContext* io_context,
+                                                     FileMetaCache* meta_cache) {
+    IcebergDeleteFileReaderOptions options;
+    options.state = runtime_state;
+    options.profile = profile;
+    options.scan_params = scan_params;
+    options.io_ctx = &io_context->io_ctx;
+    options.meta_cache = meta_cache;
+    options.batch_size = 1024;
+    return options;
+}
+
 } // namespace
 
 TEST(IcebergDeleteFileReaderHelperTest, BuildDeleteFileRange) {
@@ -397,4 +455,67 @@ TEST(IcebergDeleteFileReaderHelperTest, ReadMixedEncodingParquetPositionDeleteFi
     EXPECT_EQ(it->second, expected_positions);
 }
 
+TEST(IcebergDeleteFileReaderHelperTest, ReadOptionalParquetPositionDeleteColumnsWithoutNulls) {
+    const auto delete_file_path =
+            temp_parquet_path("iceberg_optional_position_delete_without_nulls.parquet");
+    write_position_delete_parquet(delete_file_path,
+                                  {std::string(kTargetDataFilePath),
+                                   std::string(kTargetDataFilePath), "s3://other/file.parquet"},
+                                  {3, 9, 11});
+
+    RuntimeProfile profile("test_profile");
+    RuntimeState runtime_state((TQueryOptions()), TQueryGlobals());
+    FileMetaCache meta_cache(1024);
+    IcebergDeleteFileIOContext io_context(&runtime_state);
+
+    TFileScanRangeParams scan_params;
+    scan_params.file_type = TFileType::FILE_LOCAL;
+    scan_params.format_type = TFileFormatType::FORMAT_PARQUET;
+
+    TIcebergDeleteFileDesc delete_file;
+    delete_file.path = delete_file_path;
+    delete_file.file_format = TFileFormatType::FORMAT_PARQUET;
+    delete_file.__isset.file_format = true;
+
+    CollectPositionDeleteVisitor visitor;
+    auto options =
+            delete_reader_options(&runtime_state, &profile, &scan_params, &io_context, &meta_cache);
+    auto st = read_iceberg_position_delete_file(delete_file, options, &visitor);
+
+    std::filesystem::remove(delete_file_path);
+    ASSERT_TRUE(st.ok()) << st;
+    ASSERT_EQ(visitor.total_rows, 3);
+    EXPECT_EQ(visitor.delete_rows[kTargetDataFilePath], std::vector<int64_t>({3, 9}));
+}
+
+TEST(IcebergDeleteFileReaderHelperTest, RejectParquetPositionDeleteColumnsWithActualNulls) {
+    const auto delete_file_path = temp_parquet_path("iceberg_position_delete_with_nulls.parquet");
+    write_position_delete_parquet(delete_file_path,
+                                  {std::string(kTargetDataFilePath), std::nullopt}, {3, 9});
+
+    RuntimeProfile profile("test_profile");
+    RuntimeState runtime_state((TQueryOptions()), TQueryGlobals());
+    FileMetaCache meta_cache(1024);
+    IcebergDeleteFileIOContext io_context(&runtime_state);
+
+    TFileScanRangeParams scan_params;
+    scan_params.file_type = TFileType::FILE_LOCAL;
+    scan_params.format_type = TFileFormatType::FORMAT_PARQUET;
+
+    TIcebergDeleteFileDesc delete_file;
+    delete_file.path = delete_file_path;
+    delete_file.file_format = TFileFormatType::FORMAT_PARQUET;
+    delete_file.__isset.file_format = true;
+
+    CollectPositionDeleteVisitor visitor;
+    auto options =
+            delete_reader_options(&runtime_state, &profile, &scan_params, &io_context, &meta_cache);
+    auto st = read_iceberg_position_delete_file(delete_file, options, &visitor);
+
+    std::filesystem::remove(delete_file_path);
+    ASSERT_FALSE(st.ok());
+    EXPECT_NE(st.to_string().find("file_path contains null values"), std::string::npos)
+            << st.to_string();
+}
+
 } // namespace doris