This is an automated email from the ASF dual-hosted git repository.
morningman 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 dfac7bd3b44 [fix](iceberg) Report historical schema scans to the
table's metrics reporter (#68965)
dfac7bd3b44 is described below
commit dfac7bd3b44546166580bf13e660daeea8977e7c
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Oct 13 14:54:07 2026 +0800
[fix](iceberg) Report historical schema scans to the table's metrics
reporter (#68965)
### What problem does this PR solve?
Issue Number: None
Related PR: #67479, #67687
Problem Summary:
**In short.** On master, a scan of an Iceberg snapshot or tag taken
before a column rename or drop is not reported to the table's metrics
reporter, so a REST catalog with metrics reporting receives no
ScanReport for it. This PR makes such scans inherit the table's scan
context, as branch-4.x does since #67479. Query results do not change.
**Background.** Iceberg reports each scan's metrics (planning time,
manifests and files scanned) through the `MetricsReporter` in the scan's
context. `BaseTable.newScan()` puts the table's reporter into that
context, and catalogs such as REST forward the reports to their server.
To read a historical schema correctly, master builds the scan with
`SchemaAwareDataTableScan` (#67687, the master port of #67479), which
binds the partition specs to the snapshot's schema; branch-4.x does the
same through `DorisDataTableScan.wrap(table.newScan())`, which keeps the
context of the table's scan.
**The problem, and what it cost.**
| Scan | master | with this PR |
|---|---|---|
| Current schema | reported to the table's reporter | unchanged |
| Historical schema (a snapshot or tag from before a rename or drop) |
built from `TableScanContext.empty()`, whose reporter is
`LoggingMetricsReporter`: the catalog's reporter gets nothing, and the
FE log gets a "Received metrics report" INFO line | reported to the
table's reporter |
**How this PR fixes it.** `SchemaAwareDataTableScan.newScan()` takes the
context of `table.newScan()`, which carries the table's reporter, and
binds the partition specs as before. A catalog-specific table whose own
scan is not a `DataTableScan` keeps the plain context it had.
**Results.** The new unit test puts a recording reporter on an in-memory
table: a current-schema scan and a historical-schema scan each send one
ScanReport, the second with the snapshot's id and schema id. Without the
fix the historical scan sends none. All fe-connector-iceberg tests pass
(1525).
Co-authored-by: Gabriel <[email protected]>
---
.../apache/iceberg/SchemaAwareDataTableScan.java | 8 ++++-
.../iceberg/IcebergScanPlanProviderTest.java | 40 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 1 deletion(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
index 31ec1e3ea68..50d7849942a 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/iceberg/SchemaAwareDataTableScan.java
@@ -29,7 +29,13 @@ public final class SchemaAwareDataTableScan extends
DataTableScan {
}
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. */
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index f0d616d7431..3366a38ac88 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -47,6 +47,7 @@ import org.apache.doris.thrift.schema.external.TFieldPtr;
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;
@@ -76,6 +77,7 @@ import org.apache.iceberg.io.InputFile;
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;
@@ -2265,6 +2267,44 @@ public class IcebergScanPlanProviderTest {
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}) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
