github-actions[bot] commented on code in PR #69298:
URL: https://github.com/apache/doris/pull/69298#discussion_r4374678360
##########
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:
[P2] Recover dropped sources without ordering schema IDs
For a v2 table with retained schema history S100(p INT) then S20(p LONG),
keep a live manifest written under S20, then remove identity(p) and drop p
while retaining both schemas. Schema IDs need only be unique
([specification](https://iceberg.apache.org/spec/#schema-evolution)); the [SDK
parser preserves the supplied
IDs](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/TableMetadataParser.java#L369-L420).
This sort selects S100 and labels the manifest's physical Long(7) partition as
INTEGER. The SDK [reads physical LONG as
Long](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/avro/InternalReader.java#L220-L221)
and [stores it in the expected partition
struct](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/avro/InternalReaders.java#L48-L78),
so forScan to the current UNKNOWN spec reaches IcebergPartitionValueAdapter's
Integer cast
and throws ClassCastException. With manifest caching enabled, a cold
streaming scan fails during lazy iteration, outside streamingFileScanTasks'
fallback. Recover a compatible source type from metadata history or the
manifest header instead of the greatest numeric ID. This differs from the prior
UNKNOWN finding: one entirely cold data manifest fails without deletes, mixed
cache warmth or schema expiration.
--
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]
