[fix](iceberg) Guard synthesized slots in _create_column_ids

be/src/format/table/iceberg_reader.cpp

@@ -364,15 +364,27 @@ ColumnIdResult IcebergParquetReader::_create_column_ids(
                 IcebergParquetNestedColumnUtils::extract_nested_column_ids);
     };
 
+    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
+    // table columns. When present, resolve each column by name through it so the column-id set
+    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
+    // otherwise fall back to matching by Iceberg field id.
+    const auto* struct_node =
+            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
+
     for (const auto* slot : tuple_descriptor->slots()) {
         const FieldSchema* field_schema = nullptr;
-        if (table_info_node != nullptr) {
-            if (table_info_node->children_column_exists(slot->col_name())) {
+        if (struct_node != nullptr) {
+            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
+            // never registered as children, so check membership before querying: calling
+            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
+            // throws std::out_of_range from .at() in release builds.
+            if (struct_node->get_children().contains(slot->col_name()) &&
+                struct_node->children_column_exists(slot->col_name())) {
                 // Use the physical child selected by the schema-mapping pass. This keeps partial-id
                 // files in BY_NAME mode from binding a projected column through an unrelated stale
                 // field id.
                 const auto& file_column_name =
-                        table_info_node->children_file_column_name(slot->col_name());
+                        struct_node->children_file_column_name(slot->col_name());
                 for (int i = 0; i < field_desc->size(); ++i) {
                     const auto* candidate = field_desc->get_column(i);
                     if (candidate != nullptr && candidate->name == file_column_name) {
@@ -691,15 +703,27 @@ ColumnIdResult IcebergOrcReader::_create_column_ids(
                 IcebergOrcNestedColumnUtils::extract_nested_column_ids);
     };
 
+    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
+    // table columns. When present, resolve each column by name through it so the column-id set
+    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
+    // otherwise fall back to matching by Iceberg field id.
+    const auto* struct_node =
+            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
+
     for (const auto* slot : tuple_descriptor->slots()) {
         const orc::Type* orc_field = nullptr;
-        if (table_info_node != nullptr) {
-            if (table_info_node->children_column_exists(slot->col_name())) {
+        if (struct_node != nullptr) {
+            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
+            // never registered as children, so check membership before querying: calling
+            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
+            // throws std::out_of_range from .at() in release builds.
+            if (struct_node->get_children().contains(slot->col_name()) &&
+                struct_node->children_column_exists(slot->col_name())) {
                 // Select the physical child resolved by the shared schema-mapping pass. Hidden
                 // equality keys and projected columns must obey the same BY_NAME decision for
                 // partial-id ORC files.
                 const auto& file_column_name =
-                        table_info_node->children_file_column_name(slot->col_name());
+                        struct_node->children_file_column_name(slot->col_name());
                 for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
                     if (orc_type->getFieldName(i) == file_column_name) {
                         orc_field = orc_type->getSubtype(i);

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

@@ -28,6 +28,7 @@
 #include <unordered_map>
 #include <vector>
 
+#include "common/consts.h"
 #include "common/object_pool.h"
 #include "core/block/block.h"
 #include "core/block/column_with_type_and_name.h"
@@ -1166,4 +1167,96 @@ TEST_F(IcebergReaderCreateColumnIdsTest, test_create_column_ids_6) {
     }
 }
 
-} // namespace doris
+// Regression coverage for _create_column_ids() driven by the schema-mapping StructNode:
+//   1. Synthesized/metadata slots (e.g. the TopN global row-id column) are never serialized into
+//      the Iceberg schema tree, so they are absent from the node. Before the
+//      get_children().contains() guard, StructNode::children_column_exists() was called on such an
+//      unregistered name and hit DCHECK(children.contains(name)) -> abort in debug/ASAN builds
+//      (and threw std::out_of_range from .at() in release). The projected row-id slot reproduces
+//      that; with the guard it is skipped instead.
+//   2. A projected column must resolve to its physical file column BY NAME through the node, not by
+//      Iceberg field id and not by its own (table) name, so partial-id / name-mapping files stay
+//      correct. Table column "a" maps to physical "legacy_a", while an unrelated "stale" column
+//      carries a field id that collides with the projected slot's default field id (0). The result
+//      must be "legacy_a"'s column id, never "stale"'s -- a regression to id-only or identity-name
+//      binding would produce a different set and fail here.
+TEST_F(IcebergReaderCreateColumnIdsTest, parquet_name_mapped_and_synthesized_slots) {
+    // Physical Parquet schema (BY_NAME / partial-id file):
+    //   legacy_a: id-less real data column            -> column id 1
+    //   stale:    unrelated column carrying stale id 0 -> column id 2
+    FieldDescriptor field_desc;
+    FieldSchema legacy_a;
+    legacy_a.name = "legacy_a";
+    legacy_a.data_type =
+            DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    legacy_a.field_id = -1;
+    field_desc._fields.emplace_back(legacy_a);
+    FieldSchema stale;
+    stale.name = "stale";
+    stale.data_type =
+            DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    stale.field_id = 0; // collides with the projected slot's default col_unique_id (0)
+    field_desc._fields.emplace_back(stale);
+
+    // Table column "a" resolves BY NAME to physical "legacy_a"; the synthesized global row-id
+    // column is intentionally NOT registered.
+    auto struct_node = std::make_shared<TableSchemaChangeHelper::StructNode>();
+    struct_node->add_children("a", "legacy_a", TableSchemaChangeHelper::ConstNode::get_instance());
+    std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node = struct_node;
+
+    SlotDescriptor a_slot;
+    a_slot._type = DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    a_slot._col_name = "a";
+
+    SlotDescriptor row_id_slot;
+    row_id_slot._type =
+            DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    row_id_slot._col_name = BeConsts::GLOBAL_ROWID_COL;
+
+    TupleDescriptor tuple_desc;
+    tuple_desc.add_slot(&a_slot);
+    tuple_desc.add_slot(&row_id_slot);
+
+    // No abort on the synthesized slot; "a" resolves BY NAME to "legacy_a" (column id 1), never to
+    // "stale" (column id 2) through the colliding field id.
+    const ColumnIdResult result =
+            IcebergParquetReader::_create_column_ids(&field_desc, &tuple_desc, table_info_node);
+    EXPECT_EQ(result.column_ids, (std::set<uint64_t> {1}));
+    EXPECT_TRUE(result.filter_column_ids.empty());
+}
+
+TEST_F(IcebergReaderCreateColumnIdsTest, orc_name_mapped_and_synthesized_slots) {
+    // Physical ORC schema (BY_NAME): legacy_a -> column id 1, stale -> column id 2. "stale" carries
+    // an iceberg.id attribute colliding with the projected slot's default field id (0).
+    std::unique_ptr<orc::Type> orc_type(
+            orc::Type::buildTypeFromString("struct<legacy_a:bigint,stale:bigint>"));
+    orc_type->getSubtype(1)->setAttribute(IcebergOrcReader::ICEBERG_ORC_ATTRIBUTE, "0");
+
+    // Table column "a" resolves BY NAME to physical "legacy_a"; the synthesized global row-id
+    // column is intentionally NOT registered.
+    auto struct_node = std::make_shared<TableSchemaChangeHelper::StructNode>();
+    struct_node->add_children("a", "legacy_a", TableSchemaChangeHelper::ConstNode::get_instance());
+    std::shared_ptr<TableSchemaChangeHelper::Node> table_info_node = struct_node;
+
+    SlotDescriptor a_slot;
+    a_slot._type = DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    a_slot._col_name = "a";
+
+    SlotDescriptor row_id_slot;
+    row_id_slot._type =
+            DataTypeFactory::instance().create_data_type(PrimitiveType::TYPE_BIGINT, true);
+    row_id_slot._col_name = BeConsts::GLOBAL_ROWID_COL;
+
+    TupleDescriptor tuple_desc;
+    tuple_desc.add_slot(&a_slot);
+    tuple_desc.add_slot(&row_id_slot);
+
+    // No abort on the synthesized slot; "a" resolves BY NAME to "legacy_a" (column id 1), never to
+    // "stale" (column id 2) through the colliding field id.
+    const ColumnIdResult result =
+            IcebergOrcReader::_create_column_ids(orc_type.get(), &tuple_desc, table_info_node);
+    EXPECT_EQ(result.column_ids, (std::set<uint64_t> {1}));
+    EXPECT_TRUE(result.filter_column_ids.empty());
+}
+
+} // namespace doris

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -1166,7 +1166,7 @@ public static double getHotValueThreshold() {
             "FileScanNode 扫描数据的最大并发，默认为 16", "The max threads to read data of FileScanNode, default 16"})
     public int maxFileScannersConcurrency = 16;
 
-    @VarAttrDef.VarAttr(name = ENABLE_FILE_SCANNER_V2, needForward = true, description = {
+    @VarAttrDef.VarAttr(name = ENABLE_FILE_SCANNER_V2, needForward = true, fuzzy = true, description = {
             "开启后 FileScanNode 会在支持的查询场景使用 FileScannerV2，默认开启",
             "When enabled, FileScanNode uses FileScannerV2 for supported query scans. Enabled by default."})
     public boolean enableFileScannerV2 = true;
@@ -3841,6 +3841,10 @@ public void initFuzzyModeVariables() {
         this.enableLocalExchange = random.nextBoolean();
         this.enableSharedExchangeSinkBuffer = random.nextBoolean();
         this.useSerialExchange = random.nextBoolean();
+        // Randomize the external file scanner engine (FileScannerV2 vs the legacy V1 path). Kept
+        // here rather than in setFuzzyForCatalog() so it also runs in the external regression
+        // pipeline, which enables fuzzy sessions with fuzzy_test_type=p1 (not "external").
+        this.enableFileScannerV2 = random.nextBoolean();
         this.disableStreamPreaggregations = random.nextBoolean();
         this.enableStreamingAggHashJoinForcePassthrough = random.nextBoolean();
         this.enableLocalExchangeBeforeAgg = random.nextBoolean();