[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();