924060929 commented on code in PR #69298:
URL: https://github.com/apache/doris/pull/69298#discussion_r4321962000
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -632,11 +620,16 @@ public void close() {
* streaming source's {@code hasNext}. Cache disabled -> the SDK path,
byte-unchanged.
*/
private CloseableIterable<FileScanTask> streamingFileScanTasks(TableScan
scan, ConnectorSession session,
- Table table, Optional<ConnectorExpression> filter, long sliceSize)
{
+ Table table, Optional<ConnectorExpression> filter, long sliceSize,
+ ManifestReadControl readControl, IcebergTableHandle readHandle) {
if (isManifestCacheEnabled()) {
try {
return TableScanUtil.splitFiles(
- cacheBackedFileScanTasks(scan, session, table, filter,
null), sliceSize);
+ cacheBackedFileScanTasks(scan, session, table, filter,
null, readControl, readHandle),
+ sliceSize);
+ } catch (CancellationException e) {
Review Comment:
Fixed in 17859e3ff14d197db98233fdf329b7f27c847d98. QueryTaskGroups.get now
reports terminal CancellationException for the already-closed captured query,
rather than a retryable rejection. Iceberg checks that lifetime before touching
scan metadata and before every SDK fallback (streaming, materialized and both
COUNT variants). Added a real engine-session test closing the scope before
first scheduler acquisition, plus all four fallback paths under acquisition
cancellation and cache-failure/closure interleavings; no SDK planFiles restart
is allowed. Offline empty-table planning remains supported. Focused/package
tests, full FE build and Checkstyle passed.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java:
##########
@@ -274,46 +285,61 @@ private void sweepExpiredStats(long nowNanos) {
statsByQuery.entrySet().removeIf(e -> nowNanos -
e.getValue().lastTouchNanos >= statsTtlNanos);
}
- /**
- * {@code specsById} is the scan's spec map (bound to the scan's schema),
the same one its residuals and delete
- * index use: the reader converts partition values to these types, so they
must match what the scan evaluates.
- */
- private static IcebergManifestEntryKey keyOf(ManifestFile manifest,
Map<Integer, PartitionSpec> specsById) {
- PartitionSpec boundSpec = specsById.get(manifest.partitionSpecId());
- Preconditions.checkArgument(boundSpec != null, "No bound partition
spec %s for iceberg manifest %s",
- manifest.partitionSpecId(), manifest.path());
- return IcebergManifestEntryKey.of(manifest, boundSpec);
- }
-
private static ManifestCacheValue loadManifestCacheValue(ManifestFile
manifest, Table table,
- Map<Integer, PartitionSpec> specsById, ManifestContent content,
boolean estimateWeight) {
+ ManifestContent content, boolean estimateWeight) {
try {
if (content == ManifestContent.DELETES) {
- return
ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table, specsById),
estimateWeight);
+ return
ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table),
estimateWeight);
}
- return ManifestCacheValue.forDataFiles(loadDataFiles(manifest,
table, specsById), estimateWeight);
+ return ManifestCacheValue.forDataFiles(loadDataFiles(manifest,
table), estimateWeight);
} catch (IOException e) {
throw new RuntimeException("Failed to read iceberg manifest " +
manifest.path(), e);
}
}
- private static List<DataFile> loadDataFiles(ManifestFile manifest, Table
table,
- Map<Integer, PartitionSpec> specsById) throws IOException {
+ private static List<DataFile> loadDataFiles(ManifestFile manifest, Table
table) throws IOException {
List<DataFile> dataFiles = new ArrayList<>();
// .copy() is mandatory: the ManifestReader iterator reuses the same
object across iterations.
- try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest,
table.io(), specsById)) {
+ try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest,
table.io(),
+ manifestReadSpecs(manifest, table))) {
for (DataFile dataFile : reader) {
dataFiles.add(dataFile.copy());
}
}
return dataFiles;
}
- private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest,
Table table,
- Map<Integer, PartitionSpec> specsById) throws IOException {
+ /** Keep dropped partition sources readable for historical queries without
reopening the manifest header. */
+ private static Map<Integer, PartitionSpec> manifestReadSpecs(ManifestFile
manifest, Table table) {
+ PartitionSpec spec = table.specs().get(manifest.partitionSpecId());
+ boolean missingSource = spec.fields().stream()
+ .anyMatch(field -> spec.schema().findField(field.sourceId())
== null);
+ if (!missingSource) {
+ return Collections.singletonMap(spec.specId(), spec);
+ }
+ Map<Integer, Types.NestedField> sources = new LinkedHashMap<>();
+ for (PartitionField field : spec.fields()) {
+ Types.NestedField source =
spec.schema().findField(field.sourceId());
+ if (source == null) {
+ source = table.schemas().entrySet().stream()
+ .sorted(Map.Entry.<Integer,
Schema>comparingByKey().reversed())
+ .map(entry ->
entry.getValue().findField(field.sourceId()))
+ .filter(Objects::nonNull).findFirst()
+ .orElseThrow(() -> new IllegalArgumentException(
Review Comment:
Completed the remaining identity/truncate case in
17859e3ff14d197db98233fdf329b7f27c847d98. If expiration has removed every
source schema while a manifest is still live, the shared data/delete loader now
lets the SDK recover the physical spec from that manifest header. This
exceptional cleanup path uses the SDK header and entry streams; normal/promoted
types and fixed-result transforms still open one stream, and warm cache reads
open none. Added real cleanExpiredMetadata tests for identity/truncate data and
position-delete manifests, physical-value preservation and delete-index
matching. Shared cached values are not mutated. Package tests and FE build
passed.
--
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]
