[fix](iceberg) Project Iceberg system table scans

fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java

@@ -28,15 +28,11 @@
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.io.CloseableIterator;
-import org.apache.iceberg.types.Types.NestedField;
-import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.SerializationUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
 import java.util.stream.Collectors;
@@ -50,7 +46,7 @@ public class IcebergSysTableJniScanner extends JniScanner {
     private final ClassLoader classLoader;
     private final PreExecutionAuthenticator preExecutionAuthenticator;
     private final FileScanTask scanTask;
-    private final List<SelectedField> fields;
+    private final int requiredFieldCount;
     private final String timezone;
     private CloseableIterator<StructLike> reader;
 
@@ -60,15 +56,22 @@ public IcebergSysTableJniScanner(int batchSize, Map<String, String> params) {
         Preconditions.checkArgument(serializedSplitParams != null && !serializedSplitParams.isEmpty(),
                 "serialized_split should not be empty");
         this.scanTask = SerializationUtil.deserializeFromBase64(serializedSplitParams);
-        String[] requiredFields = params.get("required_fields").split(",");
-        this.fields = selectSchema(scanTask.schema().asStruct(), requiredFields);
+        String requiredFieldsParam = params.get("required_fields");
+        Preconditions.checkArgument(requiredFieldsParam != null && !requiredFieldsParam.isEmpty(),
+                "required_fields should not be empty");
+        String[] requiredFields = requiredFieldsParam.split(",");
+        this.requiredFieldCount = requiredFields.length;
         this.timezone = params.getOrDefault("time_zone", TimeZone.getDefault().getID());
         Map<String, String> hadoopOptionParams = params.entrySet().stream()
                 .filter(kv -> kv.getKey().startsWith(HADOOP_OPTION_PREFIX))
                 .collect(Collectors
                         .toMap(kv1 -> kv1.getKey().substring(HADOOP_OPTION_PREFIX.length()), kv1 -> kv1.getValue()));
         this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(hadoopOptionParams);
-        ColumnType[] requiredTypes = parseRequiredTypes(params.get("required_types").split("#"), requiredFields);
+        String requiredTypesParam = params.get("required_types");
+        Preconditions.checkArgument(requiredTypesParam != null && !requiredTypesParam.isEmpty(),
+                "required_types should not be empty");
+        String[] requiredTypeStrings = requiredTypesParam.split("#");
+        ColumnType[] requiredTypes = parseRequiredTypes(requiredTypeStrings, requiredFields);
         initTableInfo(requiredTypes, requiredFields, batchSize);
     }
 
@@ -106,9 +109,10 @@ protected int getNext() throws IOException {
                     break;
                 }
                 StructLike row = reader.next();
-                for (int i = 0; i < fields.size(); i++) {
-                    SelectedField field = fields.get(i);
-                    Object value = row.get(field.sourceIndex, field.field.type().typeId().javaClass());
+                for (int i = 0; i < requiredFieldCount; i++) {
+                    // FE keeps the fields requested by BE at the start of the Iceberg projection.
+                    // FileScanTask.schema() is not the row schema for every DataTask implementation.
+                    Object value = row.get(i, Object.class);
                     ColumnValue columnValue = new IcebergSysTableColumnValue(value, timezone);
                     appendData(i, columnValue);
                 }
@@ -129,35 +133,10 @@ public void close() throws IOException {
         }
     }
 
-    private static List<SelectedField> selectSchema(StructType schema, String[] requiredFields) {
-        List<NestedField> schemaFields = schema.fields();
-        List<SelectedField> selectedFields = new ArrayList<>();
-        for (String requiredField : requiredFields) {
-            NestedField field = schema.field(requiredField);
-            if (field == null) {
-                throw new IllegalArgumentException("RequiredField " + requiredField + " not found in schema");
-            }
-            int sourceIndex = schemaFields.indexOf(field);
-            if (sourceIndex < 0) {
-                throw new IllegalArgumentException(
-                        "RequiredField " + requiredField + " not found in source schema fields");
-            }
-            selectedFields.add(new SelectedField(sourceIndex, field));
-        }
-        return selectedFields;
-    }
-
-    private static final class SelectedField {
-        private final int sourceIndex;
-        private final NestedField field;
-
-        private SelectedField(int sourceIndex, NestedField field) {
-            this.sourceIndex = sourceIndex;
-            this.field = field;
-        }
-    }
-
     private static ColumnType[] parseRequiredTypes(String[] typeStrings, String[] requiredFields) {
+        Preconditions.checkArgument(typeStrings.length == requiredFields.length,
+                "required_types size %s does not match required_fields size %s",
+                typeStrings.length, requiredFields.length);
         ColumnType[] requiredTypes = new ColumnType[typeStrings.length];
         for (int i = 0; i < typeStrings.length; i++) {
             String type = typeStrings[i];

fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java

@@ -181,7 +181,7 @@ protected void initSchemaParams() throws UserException {
             slotInfo.setSlotId(slot.getId().asInt());
             TColumnCategory category = classifyColumn(slot, partitionKeys);
             slotInfo.setCategory(category);
-            slotInfo.setIsFileSlot(category == TColumnCategory.REGULAR || category == TColumnCategory.GENERATED);
+            slotInfo.setIsFileSlot(isFileSlot(category));
             params.addToRequiredSlots(slotInfo);
         }
         setDefaultValueExprs(getTargetTable(), destSlotDescByName, null, params, false);
@@ -210,7 +210,7 @@ private void updateRequiredSlots() throws UserException {
             slotInfo.setSlotId(slot.getId().asInt());
             TColumnCategory category = classifyColumn(slot, partitionKeys);
             slotInfo.setCategory(category);
-            slotInfo.setIsFileSlot(category == TColumnCategory.REGULAR || category == TColumnCategory.GENERATED);
+            slotInfo.setIsFileSlot(isFileSlot(category));
             params.addToRequiredSlots(slotInfo);
         }
         // Update required slots and column_idxs in scanRangeLocations.
@@ -228,6 +228,10 @@ protected TColumnCategory classifyColumn(SlotDescriptor slot, List<String> parti
         return TColumnCategory.REGULAR;
     }
 
+    protected boolean isFileSlot(TColumnCategory category) {
+        return category == TColumnCategory.REGULAR || category == TColumnCategory.GENERATED;
+    }
+
     public void setTableSample(TableSample tSample) {
         this.tableSample = tSample;
     }

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSysExternalTable.java

@@ -80,6 +80,10 @@ public String getSysTableType() {
         return sysTableType;
     }
 
+    public boolean isPositionDeletesTable() {
+        return MetadataTableType.POSITION_DELETES.name().equalsIgnoreCase(sysTableType);
+    }
+
     public Table getSysIcebergTable() {
         if (sysIcebergTable == null) {
             synchronized (this) {

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java

@@ -96,6 +96,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Binder;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
@@ -108,6 +109,7 @@
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.util.ScanTaskUtil;
 import org.apache.iceberg.util.SerializationUtil;
@@ -121,11 +123,13 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -674,11 +678,70 @@ public TableScan createTableScan() throws UserException {
             this.pushdownIcebergPredicates.add(predicate.toString());
         }
 
+        // Doris reads normal Iceberg table files in BE and applies column pruning through scan range params.
+        // System tables are different: Iceberg SDK DataTask materializes rows using the projected scan
+        // schema. Keep Doris file slots in the same order as the JNI reader's required fields.
+        if (isSystemTable) {
+            Schema projectedSchema = getSystemTableProjectedSchema(expressions, scan.isCaseSensitive());
+            Preconditions.checkState(!projectedSchema.columns().isEmpty(),
+                    "Iceberg system table scan must materialize at least one file slot");
+            scan = scan.project(projectedSchema);
+        }
+
         icebergTableScan = scan.planWith(source.getCatalog().getThreadPoolWithPreAuth());
 
         return icebergTableScan;
     }
 
+    @VisibleForTesting
+    Schema getSystemTableProjectedSchema(List<Expression> expressions, boolean caseSensitive)
+            throws UserException {
+        List<NestedField> projectedFields = new ArrayList<>();
+        Set<Integer> projectedFieldIds = new HashSet<>();
+        List<String> partitionKeys = getPathPartitionKeys();
+        for (SlotDescriptor slot : desc.getSlots()) {
+            Column column = slot.getColumn();
+            String columnName = column.getName();
+            if (!isFileSlot(classifyColumn(slot, partitionKeys))) {
+                continue;
+            }
+
+            NestedField field = caseSensitive
+                    ? icebergTable.schema().findField(columnName)
+                    : icebergTable.schema().caseInsensitiveFindField(columnName);
+            if (field == null) {
+                throw new UserException("Column " + columnName + " not found in Iceberg system table schema");
+            }
+            if (projectedFieldIds.add(field.fieldId())) {
+                projectedFields.add(field);
+            }
+        }
+
+        Set<Integer> filterFieldIds = Binder.boundReferences(
+                icebergTable.schema().asStruct(), expressions, caseSensitive);
+        for (Integer fieldId : filterFieldIds) {
+            NestedField field = getTopLevelSystemTableField(fieldId);
+            if (field == null) {
+                throw new UserException(
+                        "Column with field id " + fieldId + " not found in Iceberg system table schema");
+            }
+            if (!projectedFieldIds.contains(field.fieldId())) {
+                throw new UserException("Iceberg system table filter column " + field.name()
+                        + " is not materialized by the planner");
+            }
+        }
+        return new Schema(projectedFields);
+    }
+
+    private NestedField getTopLevelSystemTableField(int fieldId) {
+        for (NestedField field : icebergTable.schema().columns()) {
+            if (field.fieldId() == fieldId || TypeUtil.getProjectedIds(field.type()).contains(fieldId)) {
+                return field;
+            }
+        }
+        return null;
+    }
+
     private CloseableIterable<FileScanTask> planFileScanTask(TableScan scan) {
         if (!IcebergUtils.isManifestCacheEnabled(source.getCatalog())) {
             return splitFiles(scan);
@@ -1199,7 +1262,7 @@ private List<Split> doGetSystemTableSplits() throws UserException {
     private boolean isPositionDeletesSystemTable() {
         TableIf targetTable = source.getTargetTable();
         return targetTable instanceof IcebergSysExternalTable
-                && "position_deletes".equalsIgnoreCase(((IcebergSysExternalTable) targetTable).getSysTableType());
+                && ((IcebergSysExternalTable) targetTable).isPositionDeletesTable();
     }
 
     private List<Split> doGetPositionDeletesSystemTableSplits() throws UserException {

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java

@@ -233,8 +233,13 @@ public List<Slot> computeAsteriskOutput() {
     @Override
     public boolean supportPruneNestedColumn() {
         ExternalTable table = getTable();
-        if (table instanceof IcebergExternalTable || table instanceof IcebergSysExternalTable) {
+        if (table instanceof IcebergExternalTable) {
             return true;
+        } else if (table instanceof IcebergSysExternalTable) {
+            // Position deletes use the native reader, which supports nested column pruning. Other
+            // Iceberg system tables are materialized as StructLike rows by the SDK and consumed by
+            // ordinal in the JNI reader, so their nested struct layout must remain unchanged.
+            return ((IcebergSysExternalTable) table).isPositionDeletesTable();
         } else if (table instanceof HMSExternalTable) {
             HMSExternalTable hmsTable = (HMSExternalTable) table;
             if (hmsTable.getDlaType() == HMSExternalTable.DLAType.HUDI) {

fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java

@@ -17,8 +17,12 @@
 
 package org.apache.doris.datasource.iceberg.source;
 
+import org.apache.doris.analysis.SlotDescriptor;
+import org.apache.doris.analysis.SlotId;
 import org.apache.doris.analysis.TupleDescriptor;
 import org.apache.doris.analysis.TupleId;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Type;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.LocationPath;
 import org.apache.doris.datasource.TableFormatType;
@@ -38,6 +42,8 @@
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ScanTaskUtil;
 import org.junit.Assert;
@@ -88,6 +94,64 @@ public boolean isBatchMode() {
         protected boolean getEnableMappingVarbinary() {
             return enableMappingVarbinary;
         }
+
+        @Override
+        public List<String> getPathPartitionKeys() {
+            return Collections.emptyList();
+        }
+
+        void addSlot(int slotId, Column column) {
+            SlotDescriptor slot = new SlotDescriptor(new SlotId(slotId), desc.getId());
+            slot.setColumn(column);
+            desc.addSlot(slot);
+        }
+    }
+
+    @Test
+    public void testSystemTableProjectionMatchesFileSlotOrder() throws Exception {
+        Schema systemTableSchema = new Schema(
+                Types.NestedField.required(1, "file_path", Types.StringType.get()),
+                Types.NestedField.required(2, "record_count", Types.LongType.get()),
+                Types.NestedField.optional(3, "readable_metrics", Types.StructType.of(
+                        Types.NestedField.optional(4, "id", Types.StructType.of(
+                                Types.NestedField.optional(5, "lower_bound", Types.IntegerType.get()))))));
+        Table systemTable = Mockito.mock(Table.class);
+        Mockito.when(systemTable.schema()).thenReturn(systemTableSchema);
+
+        TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable());
+        setIcebergTable(node, systemTable);
+        node.addSlot(1, new Column("RECORD_COUNT", Type.BIGINT));
+        node.addSlot(2, new Column(Column.GLOBAL_ROWID_COL + "system_table", Type.BIGINT));
+        node.addSlot(3, new Column("FILE_PATH", Type.STRING));
+
+        List<Expression> filters = Collections.singletonList(Expressions.greaterThan("record_count", 0L));
+        Schema projectedSchema = node.getSystemTableProjectedSchema(filters, false);
+
+        Assert.assertEquals(2, projectedSchema.columns().size());
+        Assert.assertEquals("record_count", projectedSchema.columns().get(0).name());
+        Assert.assertEquals("file_path", projectedSchema.columns().get(1).name());
+        Assert.assertNull(projectedSchema.findField("readable_metrics"));
+    }
+
+    @Test
+    public void testSystemTableProjectionRejectsUnmaterializedFilterColumn() throws Exception {
+        Schema systemTableSchema = new Schema(
+                Types.NestedField.required(1, "file_path", Types.StringType.get()),
+                Types.NestedField.required(2, "record_count", Types.LongType.get()));
+        Table systemTable = Mockito.mock(Table.class);
+        Mockito.when(systemTable.schema()).thenReturn(systemTableSchema);
+
+        TestIcebergScanNode node = new TestIcebergScanNode(new SessionVariable());
+        setIcebergTable(node, systemTable);
+        node.addSlot(1, new Column("record_count", Type.BIGINT));
+
+        try {
+            node.getSystemTableProjectedSchema(
+                    Collections.singletonList(Expressions.equal("file_path", "data.parquet")), true);
+            Assert.fail("Filter columns must be materialized by the planner");
+        } catch (UserException e) {
+            Assert.assertTrue(e.getMessage().contains("filter column file_path is not materialized"));
+        }
     }
 
     @Test
@@ -142,6 +206,12 @@ public void testInitialDefaultMetadataUsesSystemTableSchemaWithoutTableScan() th
         Mockito.verify(node, Mockito.never()).createTableScan();
     }
 
+    private static void setIcebergTable(IcebergScanNode node, Table table) throws Exception {
+        Field icebergTableField = IcebergScanNode.class.getDeclaredField("icebergTable");
+        icebergTableField.setAccessible(true);
+        icebergTableField.set(node, table);
+    }
+
     @Test
     public void testDetermineTargetFileSplitSizeHonorsMaxFileSplitNum() throws Exception {
         SessionVariable sv = new SessionVariable();