This is an automated email from the ASF dual-hosted git repository.
924060929 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d8e75cf63cd [fix](iceberg) Project Iceberg system table scans (#65262)
d8e75cf63cd is described below
commit d8e75cf63cd8372a1d7e4b68d629859d9dfa75dc
Author: Socrates <[email protected]>
AuthorDate: Fri Jul 17 16:03:48 2026 +0800
[fix](iceberg) Project Iceberg system table scans (#65262)
Iceberg system table scans serialize Iceberg SDK `FileScanTask` objects
and the Java metadata scanner materializes rows from the task schema.
Doris did not pass the actually required system table columns into the
Iceberg SDK scan, so metadata tables such as `$files` and `$data_files`
were planned with the full schema. For files metadata tables, the full
schema can include virtual `readable_metrics` even when the SQL only
requests columns such as `file_size_in_bytes`.
This PR applies SDK top-level column projection only for Iceberg system
table scans. Normal Iceberg data table scans continue to rely on Doris
BE scan range params for column pruning.
### What changed?
- Add system-table-only `TableScan.select(...)` projection in
`IcebergScanNode` before planning the Iceberg SDK scan.
- Skip synthesized/global row id and Iceberg row lineage columns when
building the SDK projection list.
- Add regression coverage for ORC and Parquet Iceberg tables with
`map<boolean, boolean>` columns, verifying projected `$data_files` and
`$files` queries.
---
.../doris/iceberg/IcebergSysTableJniScanner.java | 57 +++----
.../apache/doris/datasource/FileQueryScanNode.java | 8 +-
.../iceberg/IcebergSysExternalTable.java | 4 +
.../datasource/iceberg/source/IcebergScanNode.java | 65 +++++++-
.../trees/plans/logical/LogicalFileScan.java | 7 +-
.../iceberg/source/IcebergScanNodeTest.java | 70 ++++++++
.../trees/plans/logical/LogicalFileScanTest.java | 22 +++
.../test_iceberg_system_table_projection.groovy | 177 +++++++++++++++++++++
8 files changed, 367 insertions(+), 43 deletions(-)
diff --git
a/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
index b014a42706f..4e04d5bfa1d 100644
---
a/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
+++
b/fe/be-java-extensions/iceberg-metadata-scanner/src/main/java/org/apache/doris/iceberg/IcebergSysTableJniScanner.java
@@ -28,15 +28,11 @@ import com.google.common.base.Preconditions;
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 class IcebergSysTableJniScanner extends JniScanner {
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 @@ public class IcebergSysTableJniScanner extends JniScanner {
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 class IcebergSysTableJniScanner extends JniScanner
{
}
}
- 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];
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
index 55da4f30ad7..e510f05bfb6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java
@@ -181,7 +181,7 @@ public abstract class FileQueryScanNode extends
FileScanNode {
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 @@ public abstract class FileQueryScanNode extends
FileScanNode {
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 @@ public abstract class FileQueryScanNode extends
FileScanNode {
return TColumnCategory.REGULAR;
}
+ protected boolean isFileSlot(TColumnCategory category) {
+ return category == TColumnCategory.REGULAR || category ==
TColumnCategory.GENERATED;
+ }
+
public void setTableSample(TableSample tSample) {
this.tableSample = tSample;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSysExternalTable.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSysExternalTable.java
index aeb539d1f3f..14047dd44ff 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSysExternalTable.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergSysExternalTable.java
@@ -80,6 +80,10 @@ public class IcebergSysExternalTable extends ExternalTable {
return sysTableType;
}
+ public boolean isPositionDeletesTable() {
+ return
MetadataTableType.POSITION_DELETES.name().equalsIgnoreCase(sysTableType);
+ }
+
public Table getSysIcebergTable() {
if (sysIcebergTable == null) {
synchronized (this) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
index 70ac1d26b10..b592883f860 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java
@@ -96,6 +96,7 @@ import org.apache.iceberg.SplittableScanTask;
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.MappedFields;
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.ArrayList;
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 class IcebergScanNode extends FileQueryScanNode {
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 @@ public class IcebergScanNode extends FileQueryScanNode {
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 {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java
index f34ea0d633d..e70933b5b64 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScan.java
@@ -233,8 +233,13 @@ public class LogicalFileScan extends
LogicalCatalogRelation implements SupportPr
@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) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
index 2a331f57462..d4273f6faa4 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/source/IcebergScanNodeTest.java
+++
b/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.Schema;
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 class IcebergScanNodeTest {
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 class IcebergScanNodeTest {
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();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java
index 865bba61e1f..ac0c76f8090 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/logical/LogicalFileScanTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.nereids.trees.plans.logical;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Type;
import org.apache.doris.datasource.iceberg.IcebergExternalTable;
+import org.apache.doris.datasource.iceberg.IcebergSysExternalTable;
import org.apache.doris.datasource.iceberg.IcebergUtils;
import org.apache.doris.nereids.trees.plans.RelationId;
import
org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
@@ -36,6 +37,27 @@ import java.util.stream.Collectors;
public class LogicalFileScanTest {
+ @Test
+ public void testNestedColumnPruningForIcebergSystemTables() {
+ IcebergSysExternalTable positionDeletes =
Mockito.mock(IcebergSysExternalTable.class);
+ Mockito.when(positionDeletes.initSelectedPartitions(Mockito.any()))
+ .thenReturn(SelectedPartitions.NOT_PRUNED);
+
Mockito.when(positionDeletes.isPositionDeletesTable()).thenReturn(true);
+ LogicalFileScan positionDeletesScan = new LogicalFileScan(new
RelationId(1), positionDeletes,
+ Collections.singletonList("db"), Collections.emptyList(),
+ Optional.empty(), Optional.empty(), Optional.empty(),
Optional.empty());
+ Assertions.assertTrue(positionDeletesScan.supportPruneNestedColumn());
+
+ IcebergSysExternalTable jniSystemTable =
Mockito.mock(IcebergSysExternalTable.class);
+ Mockito.when(jniSystemTable.initSelectedPartitions(Mockito.any()))
+ .thenReturn(SelectedPartitions.NOT_PRUNED);
+
Mockito.when(jniSystemTable.isPositionDeletesTable()).thenReturn(false);
+ LogicalFileScan jniSystemTableScan = new LogicalFileScan(new
RelationId(2), jniSystemTable,
+ Collections.singletonList("db"), Collections.emptyList(),
+ Optional.empty(), Optional.empty(), Optional.empty(),
Optional.empty());
+ Assertions.assertFalse(jniSystemTableScan.supportPruneNestedColumn());
+ }
+
@Test
public void
testComputeOutputIncludesInvisibleRowLineageColumnsForIcebergTable() {
Column rowIdColumn = new Column(IcebergUtils.ICEBERG_ROW_ID_COL,
Type.BIGINT, true);
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_system_table_projection.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_system_table_projection.groovy
new file mode 100644
index 00000000000..a7d5acf7596
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_system_table_projection.groovy
@@ -0,0 +1,177 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_system_table_projection", "p0,external,iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test.")
+ return
+ }
+
+ String catalogName = "test_iceberg_system_table_projection"
+ String dbName = "test_iceberg_system_table_projection_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type'='iceberg',
+ 'iceberg.catalog.type'='rest',
+ 'uri' = 'http://${externalEnvIp}:${restPort}',
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ );"""
+
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+
+ def verifySystemTableProjection = { String tableName, String writeFormat ->
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ id int,
+ t_map_boolean map<boolean, boolean>,
+ dt int
+ ) ENGINE=iceberg
+ PARTITION BY LIST (dt) ()
+ PROPERTIES (
+ "write-format" = "${writeFormat}"
+ );
+ """
+ sql """
+ INSERT INTO ${tableName}
+ VALUES
+ (1, MAP(true, false), 20260702);
+ """
+
+ List<List<Object>> dataFiles = sql """
+ SELECT file_size_in_bytes
+ FROM ${tableName}\$data_files
+ ORDER BY file_size_in_bytes;
+ """
+ assertEquals(1, dataFiles.size())
+ assertTrue(((Number) dataFiles[0][0]).longValue() > 0)
+
+ List<List<Object>> filesSize = sql """
+ SELECT file_size_in_bytes
+ FROM ${tableName}\$files
+ ORDER BY file_size_in_bytes;
+ """
+ assertEquals(1, filesSize.size())
+ assertTrue(((Number) filesSize[0][0]).longValue() > 0)
+
+ List<List<Object>> files = sql """
+ SELECT file_path, record_count
+ FROM ${tableName}\$files
+ ORDER BY file_path;
+ """
+ assertEquals(1, files.size())
+ assertTrue(files[0][0].toString().contains(tableName))
+ assertEquals(1L, ((Number) files[0][1]).longValue())
+
+ List<List<Object>> snapshots = sql """
+ SELECT snapshot_id, parent_id, operation
+ FROM ${tableName}\$snapshots
+ ORDER BY committed_at;
+ """
+ assertEquals(1, snapshots.size())
+ long snapshotId = ((Number) snapshots[0][0]).longValue()
+ assertTrue(snapshotId > 0)
+ assertEquals(null, snapshots[0][1])
+ assertEquals("append", snapshots[0][2])
+
+ List<List<Object>> history = sql """
+ SELECT snapshot_id, parent_id, is_current_ancestor
+ FROM ${tableName}\$history
+ ORDER BY made_current_at;
+ """
+ assertEquals(1, history.size())
+ assertEquals(snapshotId, ((Number) history[0][0]).longValue())
+ assertEquals(null, history[0][1])
+ assertEquals(true, history[0][2])
+ }
+
+ def verifyReadableMetricsProjection = { String tableName, String
writeFormat ->
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ id int,
+ name string,
+ dt int
+ ) ENGINE=iceberg
+ PARTITION BY LIST (dt) ()
+ PROPERTIES (
+ "write-format" = "${writeFormat}"
+ );
+ """
+ sql """
+ INSERT INTO ${tableName}
+ VALUES
+ (1, 'alice', 20260702),
+ (2, 'bob', 20260702);
+ """
+
+ List<List<Object>> files = sql """
+ SELECT readable_metrics
+ FROM ${tableName}\$files
+ ORDER BY file_path;
+ """
+ assertEquals(1, files.size())
+ String readableMetrics = files[0][0].toString()
+ assertTrue(readableMetrics.contains("\"id\""))
+ assertTrue(readableMetrics.contains("\"name\""))
+ assertTrue(readableMetrics.contains("\"lower_bound\""))
+ assertTrue(readableMetrics.contains("\"upper_bound\""))
+
+ // `name` is not the first field of readable_metrics. Keep nested
column pruning enabled
+ // to verify that the system table reader preserves the SDK struct
field ordinals.
+ List<List<Object>> nameUpperBound = sql """
+ SELECT /*+ SET_VAR(enable_prune_nested_column=true) */
readable_metrics.name.upper_bound
+ FROM ${tableName}\$files
+ ORDER BY file_path;
+ """
+ assertEquals(1, nameUpperBound.size())
+ assertEquals("bob", nameUpperBound[0][0])
+ }
+
+ verifySystemTableProjection("test_iceberg_system_table_projection_orc",
"orc")
+
verifySystemTableProjection("test_iceberg_system_table_projection_parquet",
"parquet")
+
+ test {
+ sql """SELECT COUNT(*) FROM
test_iceberg_system_table_projection_orc\$data_files"""
+ result([[1L]])
+ }
+ test {
+ sql """SELECT COUNT(*) FROM
test_iceberg_system_table_projection_orc\$files"""
+ result([[1L]])
+ }
+ test {
+ sql """SELECT COUNT(*) FROM
test_iceberg_system_table_projection_parquet\$data_files"""
+ result([[1L]])
+ }
+ test {
+ sql """SELECT COUNT(*) FROM
test_iceberg_system_table_projection_parquet\$files"""
+ result([[1L]])
+ }
+
+
verifyReadableMetricsProjection("test_iceberg_system_table_projection_readable_metrics",
"orc")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
