github-actions[bot] commented on code in PR #67687:
URL: https://github.com/apache/doris/pull/67687#discussion_r3976726377
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java:
##########
@@ -758,30 +758,45 @@ static boolean isValidRelatedTable(Table table) {
/**
* The cross-query PARTITIONS-scan de-duplication seam (PERF-02): when
{@code cache} is non-null the raw
* partition list is served from / populated into the per-catalog {@link
IcebergPartitionCache} keyed by
- * {@code (id, snapshotId)} — a snapshot is immutable, so the derived
partitions are a pure function of that
- * key and safe to reuse across queries (restoring the legacy
IcebergExternalMetaCache partition-info cache).
+ * {@code (id, snapshotId, schemaId, specId)}. The schema/spec generation
is required because metadata-only
+ * evolution can change the unified partition projection without creating
a snapshot.
* A {@code null} cache (offline unit tests / the no-cache catalog) reads
live every call. The cached list is
- * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; the loader's exception (e.g. the
- * dropped-partition-source-column {@link ValidationException}) propagates
verbatim so callers keep their own
- * degradation, and a failed scan is not cached.
+ * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; loader exceptions propagate
+ * verbatim so callers keep their own degradation, and a failed scan is
not cached.
*/
private static List<IcebergRawPartition> loadRawPartitions(TableIdentifier
id, Table table, long snapshotId,
IcebergPartitionCache cache) {
if (cache == null) {
return loadRawPartitionsUncached(table, snapshotId);
}
- return cache.getOrLoad(new IcebergPartitionCache.Key(id, snapshotId),
+ return cache.getOrLoad(new IcebergPartitionCache.Key(
+ id, snapshotId, table.schema().schemaId(),
table.spec().specId()),
Review Comment:
[P1] Please carry the spec generation into the derived partition-view caches
too. `getMvccPartitionView` and `listPartitions` first query caches keyed only
by `(db, table, snapshotId, schemaId)`; an `updateSpec` commit changes neither
snapshot nor schema, so a warm derived cache can return the old view for its
independent 24-hour TTL without ever reaching this new spec-aware raw key. This
can leave MTMV partition state and generic partition-count/pruning metadata
stale after spec evolution. Please include the spec/projection generation in
those keys (or invalidate them) and cover a warm derived cache across a
spec-only commit.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergPartitionUtils.java:
##########
@@ -758,30 +758,45 @@ static boolean isValidRelatedTable(Table table) {
/**
* The cross-query PARTITIONS-scan de-duplication seam (PERF-02): when
{@code cache} is non-null the raw
* partition list is served from / populated into the per-catalog {@link
IcebergPartitionCache} keyed by
- * {@code (id, snapshotId)} — a snapshot is immutable, so the derived
partitions are a pure function of that
- * key and safe to reuse across queries (restoring the legacy
IcebergExternalMetaCache partition-info cache).
+ * {@code (id, snapshotId, schemaId, specId)}. The schema/spec generation
is required because metadata-only
+ * evolution can change the unified partition projection without creating
a snapshot.
* A {@code null} cache (offline unit tests / the no-cache catalog) reads
live every call. The cached list is
- * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; the loader's exception (e.g. the
- * dropped-partition-source-column {@link ValidationException}) propagates
verbatim so callers keep their own
- * degradation, and a failed scan is not cached.
+ * unmodifiable so a shared entry cannot be mutated by a concurrent
reader; loader exceptions propagate
+ * verbatim so callers keep their own degradation, and a failed scan is
not cached.
*/
private static List<IcebergRawPartition> loadRawPartitions(TableIdentifier
id, Table table, long snapshotId,
IcebergPartitionCache cache) {
if (cache == null) {
return loadRawPartitionsUncached(table, snapshotId);
}
- return cache.getOrLoad(new IcebergPartitionCache.Key(id, snapshotId),
+ return cache.getOrLoad(new IcebergPartitionCache.Key(
+ id, snapshotId, table.schema().schemaId(),
table.spec().specId()),
() ->
Collections.unmodifiableList(loadRawPartitionsUncached(table, snapshotId)));
}
private static List<IcebergRawPartition> loadRawPartitionsUncached(Table
table, long snapshotId) {
+ StructType unifiedPartitionType = Partitioning.partitionType(table);
+ Map<Integer, Integer> partitionFieldOrdinals = new HashMap<>();
+ for (int i = 0; i < unifiedPartitionType.fields().size(); i++) {
+
partitionFieldOrdinals.put(unifiedPartitionType.fields().get(i).fieldId(), i);
+ }
+ boolean hasUnrepresentableField = table.specs().values().stream()
Review Comment:
[P2] Please apply this safety check only to specs represented by live
entries in the selected snapshot. Iceberg's `$partitions` table builds rows
from `liveEntries()` in that snapshot; a retained historical spec with no live
files contributes no row. After old-spec files are rewritten/deleted away and
its source column is dropped, this table-wide scan still returns an empty
display even though every current partition row is representable. Preserve the
empty fallback for a live unrepresentable spec, but do not let an unused
retained spec hide all current partitions; add the no-live-old-files case.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -1201,7 +1202,12 @@ private static boolean
isPositionDeletesPartitionColumnRequested(List<ConnectorC
*/
private TableScan buildScan(Table table, IcebergTableHandle handle,
Optional<ConnectorExpression> filter,
ConnectorSession session) {
- TableScan scan = table.newScan();
+ Schema selectedSchema = !handle.isSystemTable() &&
handle.hasSnapshotPin()
+ ? pinnedSchema(table, handle) : table.schema();
+ // Keep the SDK's native Table.newScan implementation unless an actual
historical schema needs the
+ // metadata-only snapshot fix; catalog-specific Table wrappers may
provide their own scan behavior.
+ TableScan scan = !handle.isSystemTable() &&
!selectedSchema.sameSchema(table.schema())
Review Comment:
[P1] This historical-schema branch discards the catalog's native scan
implementation. With Iceberg 1.11 REST `scan-planning-mode=server`,
`RESTTable.newScan()` returns `RESTTableScan`, whose `planFiles()` must call
the server planning endpoint; replacing it with `DataTableScan` silently
reverts time-travel reads to local manifest planning. Catalogs that require
server planning or server-only metadata access will fail specifically for
historical schemas. Please preserve/refine the native scan object—including
wrappers such as the Kerberos path that rebuild `BaseTable`—or reject this
configuration explicitly, and cover a historical REST server-planned scan.
##########
fe/pom.xml:
##########
@@ -360,7 +360,7 @@ under the License.
<!-- ATTN: avro version must be consistent with Iceberg version -->
<!-- Please modify iceberg.version and avro.version together,
you can find avro version info in iceberg mvn repository -->
- <iceberg.version>1.10.1</iceberg.version>
+ <iceberg.version>1.11.0</iceberg.version>
Review Comment:
[P1] The 1.11 upgrade also introduces scan-scoped data-file credentials, but
the connector never consumes them. `RESTTableScan.planFiles()` builds
`scan.fileIO()` from `PlanTableScanResponse.credentials()` only after planning;
Doris extracts from `table.io()` before `planFiles()`, and
`getScanNodeProperties` uses a separate unplanned scan. A server-planned scan
can therefore return valid tasks whose private files BE cannot open. Please
propagate the planned scan's FileIO credentials to the BE scan properties, or
disable server planning until that contract is supported, and test a plan whose
only data credentials arrive in the plan response.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -493,7 +493,8 @@ public long streamingSplitEstimate(ConnectorSession
session, ConnectorTableHandl
long threshold = sessionLong(session, NUM_FILES_IN_BATCH_MODE,
DEFAULT_NUM_FILES_IN_BATCH_MODE);
long fileCount = 0;
try (CloseableIterable<ManifestFile> matching = getMatchingManifest(
- snapshot.dataManifests(table.io()), table.specs(),
scan.filter())) {
+ snapshot.dataManifests(table.io()),
Review Comment:
[P1] Please bypass the local manifest prepaths when Iceberg selects REST
server-side planning. `PluginDrivenScanNode` calls this estimator for ordinary
scans and batch mode defaults on, but this line opens manifests through
`table.io()` before `RESTTableScan.planFiles()` can submit the server plan. A
server-mandated/current-schema scan can therefore fail before reaching its
endpoint. With Doris's manifest cache enabled, the synchronous/streaming/count
paths similarly construct tasks locally and never invoke the native plan.
Detect server-planned scans and defer all enumeration to `scan.planFiles()` (or
reject/force client mode), with a test that denies local manifest access.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
