Skip to content

[fix](iceberg) Report historical schema scans to the table's metrics reporter #68965

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morningman merged 1 commit into from
Oct 13, 2026
Merged

[fix](iceberg) Report historical schema scans to the table's metrics reporter #68965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion ...ector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,13 @@ private SchemaAwareDataTableScan(Table table, Schema schema, TableScanContext co
}

public static TableScan newScan(Table table) {
return new SchemaAwareDataTableScan(table, table.schema(), TableScanContext.empty());
// Inherit the context of the table's own scan: it carries the table's metrics reporter (for example a REST
// catalog's scan reports), so a historical scan reports like a current one. A catalog-specific Table may
// return its own scan type; such a table keeps the plain context it had before.
TableScan tableScan = table.newScan();
TableScanContext context = tableScan instanceof DataTableScan
? ((DataTableScan) tableScan).context() : TableScanContext.empty();
return new SchemaAwareDataTableScan(table, table.schema(), context);
}

/** Returns the selected snapshot's specs bound to the scan schema. */
Expand Down
40 changes: 40 additions & 0 deletions ...iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
Expand Down Expand Up @@ -76,6 +77,7 @@
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.io.StorageCredential;
import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.metrics.ScanReport;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
Expand Down Expand Up @@ -2265,6 +2267,44 @@ public void historicalPlanningIgnoresUnusedConflictingPartitionSpecs() throws IO
assertHistoricalPredicatePlansAfterSchemaEvolution(false, true);
}

@Test
public void historicalSchemaScanReportsToTheTableMetricsReporter() throws IOException {
Schema schema = new Schema(
Types.NestedField.optional(1, "x", Types.IntegerType.get()),
Types.NestedField.optional(2, "y", Types.IntegerType.get()));
Table table = createTable("historical_metrics_reporter", schema, PartitionSpec.unpartitioned(),
Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
table.newFastAppend().appendFile(dataFile(table.spec(), "s3://b/db/historical.parquet",
1024, null, null)).commit();
long snapshotId = table.currentSnapshot().snapshotId();
int schemaId = table.currentSnapshot().schemaId();
table.updateSchema().renameColumn("x", "renamed_x").commit();
List<ScanReport> reports = new ArrayList<>();
((BaseTable) table).combineMetricsReporter(report -> {
if (report instanceof ScanReport) {
reports.add((ScanReport) report);
}
});
IcebergScanPlanProvider provider = providerOver(table);

// Control: a scan of the current schema goes through table.newScan() and reports to the table's reporter.
provider.planScan(emptySession(), ConnectorScanRequest.builder(
new IcebergTableHandle("db1", "t1"), Collections.emptyList()).build());
Assertions.assertEquals(1, reports.size());

// WHY: a scan of a historical schema is built by SchemaAwareDataTableScan instead of table.newScan(), and it
// must keep the table's scan context, or a catalog's metrics reporter (for example a REST catalog's scan
// reports) misses the scans of an old snapshot or tag. branch-4.x DorisDataTableScan.wrap() kept it (#67479).
// MUTATION: building the historical scan from an empty context -> no second report -> red.
IcebergTableHandle historical = new IcebergTableHandle("db1", "t1").withSnapshot(snapshotId, null, schemaId);
List<ConnectorScanRange> ranges = provider.planScan(emptySession(), ConnectorScanRequest.builder(
historical, Collections.emptyList()).filter(Optional.of(eqInt("x", 1))).build());
Assertions.assertEquals(1, ranges.size());
Assertions.assertEquals(2, reports.size());
Assertions.assertEquals(snapshotId, reports.get(1).snapshotId());
Assertions.assertEquals(schemaId, reports.get(1).schemaId());
}

private void assertHistoricalPredicatePlansAfterSchemaEvolution(boolean dropColumn, boolean evolveSpec)
throws IOException {
for (boolean partitioned : new boolean[] {false, true}) {
Expand Down
Loading