924060929 commented on code in PR #69298:
URL: https://github.com/apache/doris/pull/69298#discussion_r4321335041
##########
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:
Fixed in ff44a870a718eebf4b954177892218501b134dad. Source-schema recovery is
needed only when a partition result type is UNKNOWN. Bucket and time transforms
already have fixed result types, so expired source schemas no longer cause an
unnecessary recovery failure or second stream. Real retained-manifest coverage
verifies one stream after source-schema cleanup. Query-local adaptation also
preserves the physical value for UNKNOWN result fields, matching SDK
data/delete projections without changing cached values.
##########
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(
+ "Missing schema for partition source " +
field.sourceId()));
+ }
+ sources.put(field.sourceId(), Types.NestedField.optional(
+ field.sourceId(), "source_" + field.sourceId(),
source.type()));
+ }
+ PartitionSpec readable = spec.toUnbound().bind(new Schema(
+ new ArrayList<>(sources.values())), true);
Review Comment:
Fixed in ff44a870a718eebf4b954177892218501b134dad. Recovery reserves every
partition field name before generating synthetic source names and adds a suffix
on collisions. Added a real retained-manifest regression with a partition named
source_1 and an expired source column.
--
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]
