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 b014a42706ff29..4e04d5bfa1dd30 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 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 fields; + private final int requiredFieldCount; private final String timezone; private CloseableIterator reader; @@ -60,15 +56,22 @@ public IcebergSysTableJniScanner(int batchSize, Map 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 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 selectSchema(StructType schema, String[] requiredFields) { - List schemaFields = schema.fields(); - List 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 55da4f30ad7043..e510f05bfb6910 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 @@ 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 parti 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 aeb539d1f3fa3d..14047dd44ff26b 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 String getSysTableType() { 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 70ac1d26b108fe..b592883f860344 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.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 expressions, boolean caseSensitive) + throws UserException { + List projectedFields = new ArrayList<>(); + Set projectedFieldIds = new HashSet<>(); + List 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 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 planFileScanTask(TableScan scan) { if (!IcebergUtils.isManifestCacheEnabled(source.getCatalog())) { return splitFiles(scan); @@ -1199,7 +1262,7 @@ private List 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 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 f34ea0d633db3e..e70933b5b64580 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 List 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) { 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 2a331f57462152..d4273f6faa4cc5 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.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 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 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(); 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 865bba61e1f3ad..ac0c76f80907b9 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 @@ 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 @@ 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 00000000000000..a7d5acf75963f6 --- /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, + dt int + ) ENGINE=iceberg + PARTITION BY LIST (dt) () + PROPERTIES ( + "write-format" = "${writeFormat}" + ); + """ + sql """ + INSERT INTO ${tableName} + VALUES + (1, MAP(true, false), 20260702); + """ + + List> 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> 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> 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> 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> 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> 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> 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") +}