[Refactor] Extract common format scanner context types (#74866)

Signed-off-by: alvin-phoenix-ai <alvin.zhao@phoenixdata.ai>

Author: alvin-phoenix-ai

be/AGENTS.md

@@ -177,7 +177,7 @@ Core runtime building blocks without full Runtime/Exec/Storage coupling.
 ### FormatCore (`formatcore`)
 Format-oriented core primitives above ComputeEnv, ExprCore, RuntimeCore, FSCore, ChunkCore, ColumnCore, and Types.
 - Targets: `FormatCore`
-- Allowed internal include prefixes: `formats/column_evaluator.h`, `formats/deletion_bitmap.h`, `formats/disk_range.hpp`, `formats/file_writer.h`, `formats/io/`, `formats/utils.h`, `exprs/`, `runtime/`, `fs/`, `column/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
+- Allowed internal include prefixes: `formats/column_evaluator.h`, `formats/deletion_bitmap.h`, `formats/disk_range.hpp`, `formats/file_writer.h`, `formats/io/`, `formats/scan_context.h`, `formats/utils.h`, `exprs/`, `runtime/`, `fs/`, `column/`, `types/`, `common/`, `base/`, `gutil/`, `gen_cpp/`
 - Allowed target deps: `ComputeEnv`, `ExprCore`, `RuntimeCore`, `FSCore`, `ChunkCore`, `ColumnCore`, `Types`, `Common`, `Base`, `Gutil`, `StarRocksGen`
 - Core tests: `format_test`
 - Remediation: Keep FormatCore limited to reusable format primitives; move connector orchestration and higher execution policy upward.

be/module_boundary_manifest.json

@@ -319,6 +319,7 @@
         "be/src/formats/disk_range.hpp",
         "be/src/formats/file_writer.*",
         "be/src/formats/io/**",
+        "be/src/formats/scan_context.h",
         "be/src/formats/utils.h"
       ],
       "allowed_include_prefixes": [
@@ -327,6 +328,7 @@
         "formats/disk_range.hpp",
         "formats/file_writer.h",
         "formats/io/",
+        "formats/scan_context.h",
         "formats/utils.h",
         "exprs/",
         "runtime/",

be/src/connector/deletion_vector/deletion_vector.cpp

@@ -33,8 +33,8 @@ Status DeletionVector::fill_row_indexes(const SkipRowsContextPtr& skip_rows_ctx)
     } else {
         std::shared_ptr<SharedBufferedInputStream> shared_buffered_input_stream = nullptr;
         std::shared_ptr<CacheInputStream> cache_input_stream = nullptr;
-        HdfsScannerStats app_stats;
-        HdfsScannerStats fs_stats;
+        FormatScannerStats app_stats;
+        FormatScannerStats fs_stats;
 
         ASSIGN_OR_RETURN(auto path, get_absolute_path(_ctx.table_location));
         int64_t offset = _deletion_vector_descriptor->offset;
@@ -78,7 +78,7 @@ Status DeletionVector::fill_row_indexes(const SkipRowsContextPtr& skip_rows_ctx)
 }
 
 StatusOr<std::unique_ptr<RandomAccessFile>> DeletionVector::open_random_access_file(
-        const std::string& file_path, HdfsScannerStats& fs_stats, HdfsScannerStats& app_stats,
+        const std::string& file_path, FormatScannerStats& fs_stats, FormatScannerStats& app_stats,
         std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream,
         std::shared_ptr<CacheInputStream>& cache_input_stream) const {
     const OpenFileOptions options{.fs = _ctx.fs,
@@ -172,7 +172,7 @@ std::string DeletionVector::assemble_deletion_vector_path(const string& table_lo
 }
 
 void DeletionVector::update_dv_file_io_counter(
-        RuntimeProfile* parent_profile, const HdfsScannerStats& app_stats, const HdfsScannerStats& fs_stats,
+        RuntimeProfile* parent_profile, const FormatScannerStats& app_stats, const FormatScannerStats& fs_stats,
         const std::shared_ptr<CacheInputStream>& cache_input_stream,
         const std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream) {
     const std::string DV_TIMER = DeletionVector::DELETION_VECTOR;

be/src/connector/deletion_vector/deletion_vector.h

@@ -46,7 +46,7 @@ class DeletionVector {
 
 private:
     StatusOr<std::unique_ptr<RandomAccessFile>> open_random_access_file(
-            const std::string& file_path, HdfsScannerStats& fs_stats, HdfsScannerStats& app_stats,
+            const std::string& file_path, FormatScannerStats& fs_stats, FormatScannerStats& app_stats,
             std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream,
             std::shared_ptr<CacheInputStream>& cache_input_stream) const;
 
@@ -56,8 +56,8 @@ class DeletionVector {
     std::string assemble_deletion_vector_path(const std::string& table_location, std::string&& uuid,
                                               std::string& prefix) const;
 
-    void update_dv_file_io_counter(RuntimeProfile* parent_profile, const HdfsScannerStats& app_stats,
-                                   const HdfsScannerStats& fs_stats,
+    void update_dv_file_io_counter(RuntimeProfile* parent_profile, const FormatScannerStats& app_stats,
+                                   const FormatScannerStats& fs_stats,
                                    const std::shared_ptr<CacheInputStream>& cache_input_stream,
                                    const std::shared_ptr<SharedBufferedInputStream>& shared_buffered_input_stream);
 

be/src/exec/hdfs_scanner/hdfs_scanner.cpp

@@ -34,7 +34,8 @@ namespace starrocks {
 
 class CountedSeekableInputStream final : public io::SeekableInputStreamWrapper {
 public:
-    explicit CountedSeekableInputStream(const std::shared_ptr<io::SeekableInputStream>& stream, HdfsScannerStats* stats)
+    explicit CountedSeekableInputStream(const std::shared_ptr<io::SeekableInputStream>& stream,
+                                        FormatScannerStats* stats)
             : io::SeekableInputStreamWrapper(stream.get(), kDontTakeOwnership), _stream(stream), _stats(stats) {}
 
     ~CountedSeekableInputStream() override = default;
@@ -69,7 +70,7 @@ class CountedSeekableInputStream final : public io::SeekableInputStreamWrapper {
 
 private:
     std::shared_ptr<io::SeekableInputStream> _stream;
-    HdfsScannerStats* _stats;
+    FormatScannerStats* _stats;
 };
 
 Status HdfsScanner::init(RuntimeState* runtime_state, HdfsScannerContext* scanner_ctx) {
@@ -127,7 +128,7 @@ Status HdfsScanner::_build_scanner_context() {
             slot->col_name() == ICEBERG_LAST_UPDATED_SEQUENCE_NUMBER) {
             ctx.reserved_field_slots.emplace_back(slot);
         } else {
-            HdfsScannerContext::ColumnInfo column;
+            FormatColumnInfo column;
             column.slot_desc = slot;
             column.idx_in_chunk = _scanner_ctx->materialize_index_in_chunk[i];
             // A slot must be decoded eagerly if it is an output column OR if it
@@ -140,15 +141,15 @@ Status HdfsScanner::_build_scanner_context() {
 
     for (size_t i = 0; i < _scanner_ctx->partition_slots.size(); i++) {
         auto* slot = _scanner_ctx->partition_slots[i];
-        HdfsScannerContext::ColumnInfo column;
+        FormatColumnInfo column;
         column.slot_desc = slot;
         column.idx_in_chunk = _scanner_ctx->partition_index_in_chunk[i];
         ctx.partition_columns.emplace_back(column);
     }
 
     for (size_t i = 0; i < _scanner_ctx->extended_col_slots.size(); i++) {
         auto* slot = _scanner_ctx->extended_col_slots[i];
-        HdfsScannerContext::ColumnInfo column;
+        FormatColumnInfo column;
         column.slot_desc = slot;
         column.idx_in_chunk = _scanner_ctx->extended_col_index_in_chunk[i];
         ctx.extended_columns.emplace_back(column);

be/src/exec/hdfs_scanner/hdfs_scanner.h

@@ -32,8 +32,8 @@ struct OpenFileOptions {
     FileSystem* fs = nullptr;
     std::string file_path;
     int64_t file_size = -1;
-    HdfsScannerStats* fs_stats = nullptr;
-    HdfsScannerStats* app_stats = nullptr;
+    FormatScannerStats* fs_stats = nullptr;
+    FormatScannerStats* app_stats = nullptr;
 
     // for datacache
     DataCacheOptions datacache_options;
@@ -107,8 +107,8 @@ class HdfsScanner {
 protected:
     HdfsScannerContext* _scanner_ctx = nullptr;
     RuntimeState* _runtime_state = nullptr;
-    HdfsScannerStats _app_stats;
-    HdfsScannerStats _fs_stats;
+    FormatScannerStats _app_stats;
+    FormatScannerStats _fs_stats;
     std::unique_ptr<RandomAccessFile> _file;
     // by default it's no compression.
     CompressionTypePB _compression_type = CompressionTypePB::NO_COMPRESSION;

be/src/exec/hdfs_scanner/hdfs_scanner_context.cpp

@@ -96,7 +96,7 @@ void HdfsScannerContext::update_return_count_columns() {
     // special handling for ___count__ optimization.
     // this is different from `can_use_count_optimization` ,  which uses iceberg metadata to return count value
     // this optimizaton is to fill with `count` rows of default value from parquet/orc header
-    std::vector<ColumnInfo> updated_columns;
+    std::vector<FormatColumnInfo> updated_columns;
     for (auto& column : materialized_columns) {
         if (column.name() == kCountOptColumnName) {
             update_with_none_existed_slot(column.slot_desc);
@@ -111,7 +111,7 @@ void HdfsScannerContext::update_min_max_columns() {
     if (!options.use_min_max_opt) {
         return;
     }
-    std::vector<ColumnInfo> updated_columns;
+    std::vector<FormatColumnInfo> updated_columns;
     const std::map<int32_t, TExprMinMaxValue>& min_max_values = scan_range->min_max_values;
     for (auto& column : materialized_columns) {
         if (min_max_values.find(column.slot_id()) != min_max_values.end()) {
@@ -146,7 +146,7 @@ void HdfsScannerContext::update_min_max_columns() {
 }
 
 Status HdfsScannerContext::update_materialized_columns(const std::unordered_set<std::string>& names) {
-    std::vector<ColumnInfo> updated_columns;
+    std::vector<FormatColumnInfo> updated_columns;
     for (auto& column : materialized_columns) {
         auto col_name = column.formatted_name(options.case_sensitive);
         if (names.find(col_name) == names.end()) {
@@ -353,7 +353,7 @@ void HdfsScannerContext::append_or_update_extended_column_to_chunk(ChunkPtr* chu
 }
 
 void HdfsScannerContext::append_or_update_column_to_chunk(ChunkPtr* chunk, size_t row_count,
-                                                          const std::vector<ColumnInfo>& columns,
+                                                          const std::vector<FormatColumnInfo>& columns,
                                                           const Columns& values) {
     if (columns.size() == 0) return;