[fix](iceberg) support reading iceberg partition table without partition value stored in file

be/src/common/config.cpp

@@ -1494,6 +1494,9 @@ DEFINE_mBool(skip_loading_stale_rowset_meta, "false");
 
 DEFINE_Bool(enable_file_logger, "true");
 
+// Enable partition column fallback when partition columns are missing from file
+DEFINE_Bool(enable_iceberg_partition_column_fallback, "true");
+
 // The minimum row group size when exporting Parquet files. default 128MB
 DEFINE_Int64(min_row_group_size, "134217728");
 

be/src/common/config.h

@@ -1555,6 +1555,9 @@ DECLARE_mBool(skip_loading_stale_rowset_meta);
 // Only works when starting BE with --console.
 DECLARE_Bool(enable_file_logger);
 
+// Enable partition column fallback when partition columns are missing from file
+DECLARE_Bool(enable_iceberg_partition_column_fallback);
+
 // The minimum row group size when exporting Parquet files.
 DECLARE_Int64(min_row_group_size);
 

be/src/vec/exec/scan/file_scanner.cpp

@@ -1382,6 +1382,17 @@ Status FileScanner::_set_fill_or_truncate_columns(bool need_to_get_parsed_schema
         _slot_lower_name_to_col_type.emplace(to_lower(col_name), col_type);
     }
 
+    if (!_fill_partition_from_path && config::enable_iceberg_partition_column_fallback) {
+        // check if the cols of _partition_col_descs are in _missing_cols
+        // if so, set _fill_partition_from_path to true and remove the col from _missing_cols
+        for (const auto& [col_name, col_type] : _partition_col_descs) {
+            if (_missing_cols.contains(col_name)) {
+                _fill_partition_from_path = true;
+                _missing_cols.erase(col_name);
+            }
+        }
+    }
+
     RETURN_IF_ERROR(_generate_missing_columns());
     if (_fill_partition_from_path) {
         RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs, _missing_col_descs));