924060929 commented on code in PR #69298:
URL: https://github.com/apache/doris/pull/69298#discussion_r4375127275
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java:
##########
@@ -274,46 +299,73 @@ 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 {
+ /** Recover dropped partition sources from retained schemas, or the
manifest when those schemas expired. */
+ private static Map<Integer, PartitionSpec> manifestReadSpecs(ManifestFile
manifest, Table table) {
+ PartitionSpec spec = table.specs().get(manifest.partitionSpecId());
+ // Fixed-result transforms (bucket/year/month/day/hour) remain
readable without their source
+ // schema. Iceberg resolves their result from UNKNOWN; do not require
expired source schemas.
+ boolean missingSource = spec.rawPartitionType().fields().stream()
+ .anyMatch(field -> field.type().typeId() ==
org.apache.iceberg.types.Type.TypeID.UNKNOWN);
+ if (!missingSource) {
+ return Collections.singletonMap(spec.specId(), spec);
+ }
+ Map<Integer, Types.NestedField> sources = new LinkedHashMap<>();
+ Set<String> reservedNames = new HashSet<>();
+ spec.fields().forEach(field -> reservedNames.add(field.name()));
+ 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())
Review Comment:
Fixed in 3c8c571aaf5. The cache now resolves partition source types from the
snapshot that added the manifest and its schema, without ordering schema IDs.
If that snapshot or schema has expired, it lets Iceberg read the physical type
from the manifest header. The new cold-manifest regression reproduced
ClassCastException on the previous code for both retained and expired writer
snapshots; both pass after the fix, along with all 31 IcebergManifestCacheTest
cases.
--
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]
