morningman commented on code in PR #69536:
URL: https://github.com/apache/doris/pull/69536#discussion_r4368992831
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -2327,8 +2343,73 @@ private static List<Schema> metadataSchemaHistory(Table
table) {
return new ArrayList<>(table.schemas().values());
}
- private static void addHistoricalEqualityFields(
- List<NestedField> fields, Set<Integer> missingFieldIds, Schema
historicalSchema) {
+ /**
+ * A writer commits with the schema that is current at commit, so a key
dropped before its delete commits loses
+ * the schema-only changes made before the drop from every snapshot. Field
IDs are never reused and a field
+ * only relaxes compatibly, by promotion or from required to optional, so
each field of the selected definition
+ * takes the widest promotion of its type and the optionality it has in
any retained schema; neither depends on
+ * the order of the schemas. A field required in every retained schema
stays required, and a promoted field
+ * keeps its defaults in the promoted type ({@link #promotedDefault}).
+ */
+ private static List<NestedField> withRetainedRelaxations(List<NestedField>
fields, List<Schema> schemas) {
+ List<NestedField> relaxed = new ArrayList<>(fields.size());
+ for (NestedField field : fields) {
+ Type type = field.type();
+ boolean optional = field.isOptional();
+ for (Schema schema : schemas) {
+ NestedField retained = schema.findField(field.fieldId());
+ if (retained == null) {
+ continue;
+ }
+ optional |= retained.isOptional();
+ if (type.isPrimitiveType() && retained.type().isPrimitiveType()
+ && isWideningPromotion(type.asPrimitiveType(),
retained.type().asPrimitiveType())) {
+ type = retained.type();
+ }
+ }
+ if (type.isStructType()) {
+ type =
Types.StructType.of(withRetainedRelaxations(type.asStructType().fields(),
schemas));
+ }
+ relaxed.add(type.equals(field.type()) && optional ==
field.isOptional()
+ ? field :
Types.NestedField.from(field).ofType(type).isOptional(optional)
+
.withInitialDefault(promotedDefault(field.initialDefaultLiteral(),
field.type(), type))
+
.withWriteDefault(promotedDefault(field.writeDefaultLiteral(), field.type(),
type))
+ .build());
+ }
+ return relaxed;
+ }
+
+ /**
+ * {@link TypeUtil#isPromotionAllowed} plus the v3 promotion of DATE to an
unzoned TIMESTAMP or TIMESTAMP_NS,
+ * which the SDK's schema update does not implement yet but the format
allows.
+ */
+ private static boolean isWideningPromotion(Type.PrimitiveType from,
Type.PrimitiveType to) {
+ if (TypeUtil.isPromotionAllowed(from, to)) {
+ return true;
+ }
+ if (from.typeId() != Type.TypeID.DATE) {
Review Comment:
Confirmed, and fixed in `9affde5af66`. The v3 promotion table lets `unknown`
become any type, but Iceberg 1.12.0's `TypeUtil.isPromotionAllowed` and
`SchemaUpdate` do not implement it, and `isWideningPromotion` only added DATE
on top. In your history the only snapshot that records k is S1, written while k
was `unknown`, so the carrier kept `unknown`. The scan dictionary encodes
`unknown` as STRING, and FileScannerV2 then casts both keys to STRING: the DATE
key becomes `2024-01-02` and the TIMESTAMP_NS tombstone `2024-01-02
00:00:00.000000000`, so the deleted row stays visible. Master's reverse lookup
took the retained TIMESTAMP_NS definition, so this PR introduced the
regression. V1 compares in the delete file's own key type and was not affected.
`isWideningPromotion` now treats `unknown` as promotable to any primitive
type, so the recovered definition takes the widest retained promotion,
TIMESTAMP_NS here. An `unknown` field only has null defaults, so no default
conversion is involved.
`equalityCarrierFollowsUnknownKeyPromotedWithoutASnapshot` replays your history
(both promotions published as schema-only changes, a pending append and a
pending delete committed after the drop) and checks the carrier and the scan
dictionary; without the new rule the carrier stays `unknown`.
If metadata cleanup has removed every schema that gave k a concrete type,
the carrier has nothing to promote to and stays `unknown`, as on master. That
is the same limitation as the cleanup case discussed earlier: the comparison
type then has to come from the files, which I'm tracking as the FileScannerV2
follow-up.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -2327,8 +2341,55 @@ private static List<Schema> metadataSchemaHistory(Table
table) {
return new ArrayList<>(table.schemas().values());
}
- private static void addHistoricalEqualityFields(
- List<NestedField> fields, Set<Integer> missingFieldIds, Schema
historicalSchema) {
+ /**
+ * A writer commits with the schema that is current at commit, so a key
dropped before its delete commits loses
+ * the schema-only changes made before the drop from every snapshot. Field
IDs are never reused and a field
+ * only relaxes compatibly, by promotion or from required to optional, so
each field of the selected definition
+ * takes the widest promotion of its type and the optionality it has in
any retained schema; neither depends on
+ * the order of the schemas. A field required in every retained schema
stays required.
+ */
+ private static List<NestedField> withRetainedRelaxations(List<NestedField>
fields, List<Schema> schemas) {
+ List<NestedField> relaxed = new ArrayList<>(fields.size());
+ for (NestedField field : fields) {
+ Type type = field.type();
+ boolean optional = field.isOptional();
+ for (Schema schema : schemas) {
+ NestedField retained = schema.findField(field.fieldId());
+ if (retained == null) {
+ continue;
+ }
+ optional |= retained.isOptional();
+ if (type.isPrimitiveType() && retained.type().isPrimitiveType()
+ && isWideningPromotion(type.asPrimitiveType(),
retained.type().asPrimitiveType())) {
+ type = retained.type();
+ }
+ }
+ if (type.isStructType()) {
+ type =
Types.StructType.of(withRetainedRelaxations(type.asStructType().fields(),
schemas));
+ }
+ relaxed.add(type.equals(field.type()) && optional ==
field.isOptional()
+ ? field :
Types.NestedField.from(field).ofType(type).isOptional(optional).build());
Review Comment:
You're right about the copied write default; fixed in `9affde5af66`. The
rebuilt carrier field took the write default of the lineage definition.
Converting a stale one outside the TIMESTAMP_NS range (1000-01-01 in your
history) overflowed in `DateTimeUtil.microsToNanos` before any reader ran, so
planning failed with `ArithmeticException: long overflow`. The carrier only
feeds the scan dictionary, which carries the initial default and never a write
default, so a rebuilt field now drops its write default instead of converting
it. The initial default is still converted: it cannot change once the field is
added, so the retained promoted definition already holds the same value.
`equalityCarrierDropsStaleWriteDefaultOfPromotedDateKey` replays your history
(write default cleared with `updateColumnDefault`, TIMESTAMP_NS published
without a snapshot, the delete pending across the drop), and the two earlier
default tests now expect no write default on the carrier.
--
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]
