fix fe iceberg checkstyle

Author: hubgeter

be/src/exec/scan/file_scanner.cpp

@@ -1205,6 +1205,16 @@ Status FileScanner::_init_parquet_reader(FileMetaCache* file_meta_cache_ptr,
             RETURN_IF_ERROR(_create_row_id_column_iterator());
             iceberg_reader->set_row_id_column_iterator(_row_id_column_iterator_pair);
         }
+        if (_row_lineage_columns != nullptr) {
+            auto row_lineage_columns = std::make_shared<RowGroupReader::RowLineageColumns>();
+            row_lineage_columns->row_id_column_idx = _row_lineage_columns->row_id_column_idx;
+            row_lineage_columns->last_updated_sequence_number_column_idx =
+                    _row_lineage_columns->last_updated_sequence_number_column_idx;
+            row_lineage_columns->first_row_id = range.table_format_params.iceberg_params.first_row_id;
+            row_lineage_columns->last_updated_sequence_number =
+                    range.table_format_params.iceberg_params.last_updated_sequence_number;
+            iceberg_reader->set_row_lineage_columns(row_lineage_columns);
+        }
         iceberg_reader->set_push_down_agg_type(_get_push_down_agg_type());
 
         init_status = iceberg_reader->init_reader(
@@ -1761,6 +1771,19 @@ Status FileScanner::_init_expr_ctxes() {
             _row_id_column_iterator_pair.second = _default_val_row_desc->get_column_id(slot_id);
             continue;
         }
+        if (it->second->col_name().starts_with("_row_id")) {
+            if (_row_lineage_columns == nullptr) {
+                _row_lineage_columns = std::make_shared<RowGroupReader::RowLineageColumns>();
+            }
+            _row_lineage_columns->row_id_column_idx = _default_val_row_desc->get_column_id(slot_id);
+        }
+        if (it->second->col_name().starts_with("_last_updated_sequence_number")) {
+            if (_row_lineage_columns == nullptr) {
+                _row_lineage_columns = std::make_shared<RowGroupReader::RowLineageColumns>();
+            }
+            _row_lineage_columns->last_updated_sequence_number_column_idx =
+                    _default_val_row_desc->get_column_id(slot_id);
+        }
 
         LOG(INFO) << "[DEBUG _init_expr_ctxes] slot: '" << it->second->col_name()
                   << "' is_file_slot=" << slot_info.is_file_slot << " is_partition="

be/src/exec/scan/file_scanner.h

@@ -231,6 +231,7 @@ class FileScanner : public Scanner {
 
     std::pair<std::shared_ptr<RowIdColumnIteratorV2>, int> _row_id_column_iterator_pair = {nullptr,
                                                                                            -1};
+    std::shared_ptr<RowGroupReader::RowLineageColumns> _row_lineage_columns = nullptr;
     int64_t _last_bytes_read_from_local = 0;
     int64_t _last_bytes_read_from_remote = 0;
 

be/src/exec/sink/viceberg_delete_rewrite_loader.cpp

@@ -0,0 +1,86 @@
+// 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 "exec/sink/viceberg_delete_rewrite_loader.h"
+
+#include "common/status.h"
+#include "format/table/iceberg_delete_file_reader_helper.h"
+#include "io/hdfs_builder.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+namespace {
+class RewriteBitmapVisitor final : public IcebergPositionDeleteVisitor {
+public:
+    RewriteBitmapVisitor(const std::string& referenced_data_file_path,
+                         roaring::Roaring64Map* rows_to_delete)
+            : _referenced_data_file_path(referenced_data_file_path),
+              _rows_to_delete(rows_to_delete) {}
+
+    Status visit(const std::string& file_path, int64_t pos) override {
+        if (_rows_to_delete == nullptr) {
+            return Status::InvalidArgument("rows_to_delete is null");
+        }
+        if (file_path == _referenced_data_file_path) {
+            _rows_to_delete->add(static_cast<uint64_t>(pos));
+        }
+        return Status::OK();
+    }
+
+private:
+    const std::string& _referenced_data_file_path;
+    roaring::Roaring64Map* _rows_to_delete;
+};
+} // namespace
+
+Status VIcebergDeleteRewriteLoader::load(RuntimeState* state, RuntimeProfile* profile,
+                                         const std::string& referenced_data_file_path,
+                                         const std::vector<TIcebergDeleteFileDesc>& delete_files,
+                                         const std::map<std::string, std::string>& hadoop_conf,
+                                         TFileType::type file_type,
+                                         const std::vector<TNetworkAddress>& broker_addresses,
+                                         roaring::Roaring64Map* rows_to_delete) {
+    if (rows_to_delete == nullptr) {
+        return Status::InvalidArgument("rows_to_delete is null");
+    }
+    if (state == nullptr || profile == nullptr || delete_files.empty()) {
+        return Status::OK();
+    }
+
+    TFileScanRangeParams params =
+            build_iceberg_delete_scan_range_params(hadoop_conf, file_type, broker_addresses);
+    IcebergDeleteFileIOContext delete_file_io_ctx(state);
+    IcebergDeleteFileReaderOptions options;
+    options.state = state;
+    options.profile = profile;
+    options.scan_params = &params;
+    options.io_ctx = &delete_file_io_ctx.io_ctx;
+    options.batch_size = 102400;
+
+    for (const auto& delete_file : delete_files) {
+        if (is_iceberg_deletion_vector(delete_file)) {
+            RETURN_IF_ERROR(read_iceberg_deletion_vector(delete_file, options, rows_to_delete));
+            continue;
+        }
+        RewriteBitmapVisitor visitor(referenced_data_file_path, rows_to_delete);
+        RETURN_IF_ERROR(read_iceberg_position_delete_file(delete_file, options, &visitor));
+    }
+    return Status::OK();
+}
+
+} // namespace doris

be/src/exec/sink/viceberg_delete_rewrite_loader.h

@@ -0,0 +1,45 @@
+// 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.
+
+#pragma once
+
+#include <gen_cpp/PlanNodes_types.h>
+
+#include <map>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "roaring/roaring64map.hh"
+
+namespace doris {
+
+class RuntimeState;
+class RuntimeProfile;
+
+class VIcebergDeleteRewriteLoader {
+public:
+    static Status load(RuntimeState* state, RuntimeProfile* profile,
+                       const std::string& referenced_data_file_path,
+                       const std::vector<TIcebergDeleteFileDesc>& delete_files,
+                       const std::map<std::string, std::string>& hadoop_conf,
+                       TFileType::type file_type,
+                       const std::vector<TNetworkAddress>& broker_addresses,
+                       roaring::Roaring64Map* rows_to_delete);
+};
+
+} // namespace doris