This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3d000b4e48e [fix](fe) Resolve dropped Iceberg equality-delete keys
after a schema ID is reused (#69536)
3d000b4e48e is described below
commit 3d000b4e48eff0386da476760c7546b806b7072f
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Oct 26 07:03:48 2026 +0000
[fix](fe) Resolve dropped Iceberg equality-delete keys after a schema ID is
reused (#69536)
### What problem does this PR solve?
Issue Number: None
Related PR: #66538, #69394
Problem Summary:
**In short.** This ports a fix from branch-4.1 #66538 (Iceberg V3
default values). When a dropped column is still referenced by a live
equality delete, the Iceberg connector now looks up the column's
definition along the scan snapshot's parent chain first, and only then
in the whole schema list. Master searched only the part of the schema
list before the scan schema. That breaks as soon as Iceberg reuses a
schema ID, and every query on such a table failed.
**Background.** An equality delete names its key columns by field ID. If
a key column has since been dropped, the scan schema no longer has it,
so the connector adds the column's historical definition to the schema
dictionary it sends to BE
(`IcebergScanPlanProvider.schemaForPotentialEqualityDeletes`). BE needs
that field ID, type and initial default to read the key from data files.
Master (#65851) took the definition from the metadata schema list,
searching backwards from the scan schema's position, on the assumption
that the list is in chronological order.
**The problem, and what it cost.** Schema IDs are not a sequence. When a
schema change produces a schema equal to an existing one, Iceberg reuses
that schema's ID and does not append to the list. Dropping a column that
was added after the table was created is the common case: the schema
returns to the initial one, so the current schema is the first entry of
the list, and the schema that still defines the key comes after it. The
backward search never reaches that schema, and the scan fails.
Reproduce: create a table with `id`; add column `k`; write data and an
equality delete keyed on `k`; drop `k`; query the table.
| | Master today | After this PR |
|---|---|---|
| Query after dropping an equality-delete key, when the schema returns
to an earlier one | fails: `Iceberg equality-delete fields are absent
from schema history: [2]` | succeeds |
| Same, after the snapshots that recorded the key have expired | fails
with the same error | succeeds (the key comes from the metadata schema
list) |
| Time travel to a snapshot whose lineage defines the key, while a later
schema renames it | earlier definition | earlier definition (unchanged)
|
**How this PR fixes it.** It uses the same order as
`IcebergScanNode.getSchemaFieldsForScan` on branch-4.1 and branch-4.2.
First, walk the scan snapshot's parent chain and take each missing key
from the schema that snapshot was written with. Then search the whole
metadata schema list, newest first, for keys that only schema-only
changes or expired ancestors can explain. Unlike branch-4.x, the walk
indexes each schema at most once and stops as soon as every key is
resolved, so a long lineage that shares one schema costs only map
lookups. When no key is missing, nothing changes and no snapshot is
read, as before.
When a dropped key has several historical definitions, the one recorded
on the scan's lineage now wins, as on branch-4.x. The field ID and the
initial default are the same either way. Only a name changed by a
schema-only commit after the last write can differ. The existing test
`equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey` (from #69394)
renamed and then dropped the key with no write in between. It now writes
a snapshot after each schema change, as the branch-4.1 test it was
ported from does, and still expects `k2`. A dropped key nevertheless
keeps the compatible changes its field ID has in any retained schema:
the widest promotion of its type (the SDK's promotions such as INT→LONG,
plus the v3 promotions that Iceberg 1.12.0's
`TypeUtil.isPromotionAllowed` does not know: `unknown` to any primitive
type, and DATE to an unzoned TIMESTAMP or TIMESTAMP_NS), and optional
when any retained schema has it optional. The widest promotion does not
depend on the order of the schemas. If the retained schemas promote the
key to two types that do not promote to each other, which takes setting
the current schema back to an earlier one (for example DATE→TIMESTAMP,
back to DATE, then DATE→TIMESTAMP_NS), planning fails with an explicit
error: neither type holds the other's values exactly, and the metadata
does not record which one a delete file uses. A DATE initial default
moves to midnight in the promoted type, since Iceberg 1.12.0's
`DateLiteral` converts only to DATE. The carrier drops the write default
of a rebuilt field: no reader uses it, and a stale one may not fit the
promoted type. A delete committed after a schema-only promotion or
required→optional change follows the relaxed definition even when no
snapshot on the lineage records it: FileScannerV2 compares keys in the
carrier's type, and both readers reject a data file that lacks a
required key. A key is never narrowed, and a key required in every
retained schema stays required.
**Results.** This PR ports four tests from #66538 into
`IcebergScanPlanProviderTest`:
- `equalityCarrierResolvesDroppedKeyWhenSchemaIdIsReused` uses a real
in-memory table and also checks the scan dictionary that
`getScanNodeProperties` builds.
- `equalityCarrierResolvesReusedSchemaIdAfterParentSnapshotsExpire` is
the same, with the snapshots that recorded the key expired.
- `equalityCarrierPrefersScanLineageOverLaterRenameForTimeTravel` and
`equalityCarrierKeepsDefaultWhenTimeTravelParentExpiredBeforeLaterRename`
use table doubles with schema histories that the SDK cannot build.
- `equalityCarrierKeepsPromotedTypeOfKeyDroppedBeforeItsDeleteCommits`
(new) promotes a key from INT to LONG without a snapshot, prepares an
equality delete, drops the key and then commits the delete; the carrier
must say LONG. It fails without the widening.
- `equalityCarrierKeepsLineageTimestampOfPromotedDateKey` and
`equalityCarrierWidensDateKeyPromotedWithoutASnapshot` (new) promote a
v3 DATE key to TIMESTAMP / TIMESTAMP_NS, published directly since
`SchemaUpdate` in 1.12.0 rejects these promotions; the carrier must keep
the lineage's TIMESTAMP and widen a schema-only DATE→TIMESTAMP_NS.
`equalityCarrierKeepsSchemaOnlyOptionalityOfDroppedKey` and
`equalityCarrierKeepsAlwaysRequiredDroppedKeyRequired` (new) drop a
required key while a delete on it is pending, with and without a
schema-only `makeColumnOptional`; the carrier must be optional and
required respectively.
- `equalityCarrierConvertsDefaultsOfDateKeyPromotedToTimestamp` and
`equalityCarrierConvertsDefaultsOfDateKeyPromotedToTimestampNs` (new)
give the v3 DATE key an initial and a write default, promote it to
TIMESTAMP / TIMESTAMP_NS without a snapshot and drop it while a delete
on it is pending; the carrier must hold the initial default as midnight
in the promoted type and no write default, and the scan dictionary must
carry `2024-01-02 00:00:00`. Without the conversion, planning fails with
`Cannot cast default value to timestamp`.
- `equalityCarrierFollowsUnknownKeyPromotedWithoutASnapshot` (new)
promotes a v3 key from `unknown` to DATE and then TIMESTAMP_NS without
snapshots, and commits a pending append and a pending delete after the
key is dropped, so the lineage only records `unknown`; the carrier and
the scan dictionary must say TIMESTAMP_NS. Without the `unknown`
promotion the carrier stays `unknown`, which the dictionary sends as
STRING. `equalityCarrierDropsStaleWriteDefaultOfPromotedDateKey` (new)
records a DATE write default outside the TIMESTAMP_NS range, clears it
and promotes the key to TIMESTAMP_NS without snapshots; planning must
not fail on the stale write default, which converting fails with `long
overflow`.
- `equalityCarrierRejectsDateKeyPromotedToTimestampThenTimestampNs` and
`equalityCarrierRejectsDateKeyPromotedToTimestampNsThenTimestamp` (new)
promote a v3 DATE key to one of TIMESTAMP / TIMESTAMP_NS without a
snapshot, set the DATE schema current again, promote it to the other and
drop it while a delete on it is pending; planning must fail in either
order and name both types. Without the check the carrier takes whichever
type the schema list holds first, and a TIMESTAMP carrier rounds a
nanosecond tombstone to the microsecond.
`equalityCarrierRejectsUnknownKeyPromotedToUnrelatedTypes` (new) does
the same with `unknown` promoted to INT and then STRING.
`equalityCarrierKeepsLineagePromotionBesideAnAbandonedOne` (new) commits
the delete while TIMESTAMP_NS is current, so the lineage records
TIMESTAMP_NS; the abandoned TIMESTAMP schema must not fail planning.
With master's search, the two reused-ID tests fail with the error above.
Without the lineage walk, the time-travel test gets `k2` instead of `k`.
All 1706 tests of fe-connector-iceberg pass.
Co-authored-by: daidai <[email protected]>
---
.../connector/iceberg/IcebergScanPlanProvider.java | 157 +++++-
.../iceberg/IcebergScanPlanProviderTest.java | 536 ++++++++++++++++++++-
2 files changed, 670 insertions(+), 23 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index d3245f604d1..e4709175800 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -79,6 +79,7 @@ import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.InclusiveMetricsEvaluator;
+import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.expressions.ManifestEvaluator;
import org.apache.iceberg.expressions.Projections;
import org.apache.iceberg.expressions.ResidualEvaluator;
@@ -92,6 +93,7 @@ import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.util.DateTimeUtil;
import org.apache.iceberg.util.ScanTaskUtil;
import org.apache.iceberg.util.TableScanUtil;
import org.apache.logging.log4j.LogManager;
@@ -2108,7 +2110,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// every branch so the default and current field type match BE's
read.
if (hasApplicableEqualityDeletes) {
List<NestedField> equalityFields =
schemaForPotentialEqualityDeletes(
- table, scanSchema, applicableEqualityDeleteFieldIds);
+ table, exactScan.snapshot(), scanSchema,
applicableEqualityDeleteFieldIds);
dict = IcebergSchemaUtils.encodeEqualitySchemaEvolutionProp(
table, equalityFields, appendRowLineage,
enableVarbinary, enableTimestampTz);
@@ -2280,29 +2282,27 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
* delete manifests does not enumerate data files or byte-split tasks, and
retained state is bounded by the
* number of equality keys rather than the table's entire schema history.
*
- * <p>The metadata schema list is searched in reverse chronology up to the
selected schema (schema IDs are
- * identifiers, not a sequence), so the latest definition of each stable
field ID wins. Current fields remain
- * first, allowing a dropped/re-added name to resolve the projected field
by name while a live historical
- * equality key resolves by its stable field ID.</p>
+ * <p>Schema IDs are identifiers, not a sequence: Iceberg reuses an ID
when evolution returns to an earlier
+ * schema, so a schema's position in the metadata list does not tell when
it was last active, and the list may
+ * also hold schemas committed after a time-travel or branch target. The
scan snapshot's parent chain is
+ * therefore searched first, so the definition active on that lineage
wins; the whole metadata list, newest
+ * first, is the fallback for schema-only changes and expired ancestors.
The selected definition keeps its name
+ * and initial default but takes the relaxations its field has in any
retained schema
+ * ({@link #withRetainedRelaxations}): a delete committed after a
schema-only promotion or required-to-optional
+ * change follows the relaxed definition even when no snapshot records it.
Current fields remain first,
+ * allowing a dropped/re-added name to resolve the projected field by name
while a live historical equality key
+ * resolves by its stable field ID.</p>
*/
@VisibleForTesting
static List<NestedField> schemaForPotentialEqualityDeletes(
Table table, TableScan scan, Schema scanSchema) {
return schemaForPotentialEqualityDeletes(
- table, scanSchema, applicableEqualityDeleteFieldIds(table,
scan));
+ table, scan.snapshot(), scanSchema,
applicableEqualityDeleteFieldIds(table, scan));
}
- private static List<NestedField> schemaForPotentialEqualityDeletes(
- Table table, Schema scanSchema, Set<Integer> equalityFieldIds) {
- List<Schema> metadataSchemas = metadataSchemaHistory(table);
- int selectedSchemaIndex = -1;
- for (int index = 0; index < metadataSchemas.size(); index++) {
- if (metadataSchemas.get(index).schemaId() ==
scanSchema.schemaId()) {
- selectedSchemaIndex = index;
- }
- }
- int lastRelevantIndex = selectedSchemaIndex >= 0
- ? selectedSchemaIndex : metadataSchemas.size() - 1;
+ @VisibleForTesting
+ static List<NestedField> schemaForPotentialEqualityDeletes(
+ Table table, Snapshot scanSnapshot, Schema scanSchema,
Set<Integer> equalityFieldIds) {
Set<Integer> missing = new HashSet<>(equalityFieldIds);
missing.removeAll(TypeUtil.indexById(scanSchema.asStruct()).keySet());
if (missing.isEmpty()) {
@@ -2310,8 +2310,24 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
}
List<NestedField> fields = new ArrayList<>(scanSchema.columns());
- for (int index = lastRelevantIndex; index >= 0 && !missing.isEmpty();
index--) {
- addHistoricalEqualityFields(fields, missing,
metadataSchemas.get(index));
+ List<Schema> metadataSchemas = metadataSchemaHistory(table);
+ // Snapshots on one lineage mostly share a schema, so each schema is
indexed at most once.
+ Set<Integer> lineageSchemaIds = new HashSet<>();
+ for (Snapshot snapshot = scanSnapshot; snapshot != null &&
!missing.isEmpty();
+ snapshot = snapshot.parentId() == null ? null :
table.snapshot(snapshot.parentId())) {
+ Integer schemaId = snapshot.schemaId();
+ if (schemaId == null || !lineageSchemaIds.add(schemaId)) {
+ continue;
+ }
+ Schema lineageSchema = table.schemas().get(schemaId);
+ if (lineageSchema == null) {
+ throw new IllegalStateException(
+ "Iceberg snapshot schema " + schemaId + " is absent
from table metadata");
+ }
+ addHistoricalEqualityFields(fields, missing, lineageSchema,
metadataSchemas);
+ }
+ for (int index = metadataSchemas.size() - 1; index >= 0 &&
!missing.isEmpty(); index--) {
+ addHistoricalEqualityFields(fields, missing,
metadataSchemas.get(index), metadataSchemas);
}
if (!missing.isEmpty()) {
throw new IllegalStateException(
@@ -2327,8 +2343,105 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
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 ({@link #widestPromotion}) 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; a rebuilt field keeps its initial default in the promoted
type ({@link #promotedDefault}) and drops
+ * its write default, which no reader uses and whose stale value may not
fit the promoted type.
+ */
+ 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();
+ List<Type.PrimitiveType> promotions = new ArrayList<>();
+ 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())) {
+ promotions.add(retained.type().asPrimitiveType());
+ }
+ }
+ if (type.isPrimitiveType()) {
+ type = widestPromotion(field, promotions);
+ }
+ 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((Literal<?>) null)
+ .build());
+ }
+ return relaxed;
+ }
+
+ /**
+ * The promotion of {@code field} that every other one in {@code
promotions} promotes to. Evolution moves a
+ * field only along promotions, so its retained promotions form a chain
unless the current schema was set back
+ * to an earlier one and the field was then promoted differently: DATE to
both TIMESTAMP and TIMESTAMP_NS, or
+ * UNKNOWN to two unrelated types. Neither type then holds the other's
values exactly, and the metadata does not
+ * record which one a delete file was written in, so planning fails
instead of comparing in either.
+ */
+ private static Type.PrimitiveType widestPromotion(NestedField field,
List<Type.PrimitiveType> promotions) {
+ Type.PrimitiveType widest = field.type().asPrimitiveType();
+ for (Type.PrimitiveType promotion : promotions) {
+ if (isWideningPromotion(widest, promotion)) {
+ widest = promotion;
+ }
+ }
+ for (Type.PrimitiveType promotion : promotions) {
+ if (!isWideningPromotion(promotion, widest)) {
+ throw new IllegalStateException("Iceberg equality-delete field
" + field.fieldId() + " ("
+ + field.name() + ") was promoted to both " + widest +
" and " + promotion
+ + " in schema history");
+ }
+ }
+ return widest;
+ }
+
+ /**
+ * {@link TypeUtil#isPromotionAllowed} plus the v3 promotions that the
SDK's schema update does not implement yet
+ * but the format allows: UNKNOWN to any primitive type, and DATE to an
unzoned TIMESTAMP or TIMESTAMP_NS.
+ */
+ private static boolean isWideningPromotion(Type.PrimitiveType from,
Type.PrimitiveType to) {
+ if (TypeUtil.isPromotionAllowed(from, to)) {
+ return true;
+ }
+ if (from.typeId() == Type.TypeID.UNKNOWN) {
+ return true;
+ }
+ if (from.typeId() != Type.TypeID.DATE) {
+ return false;
+ }
+ return to.typeId() == Type.TypeID.TIMESTAMP && !((Types.TimestampType)
to).shouldAdjustToUTC()
+ || to.typeId() == Type.TypeID.TIMESTAMP_NANO &&
!((Types.TimestampNanoType) to).shouldAdjustToUTC();
+ }
+
+ /**
+ * {@code value}, the initial default of a field of type {@code from}, in
the type {@code to} the field is rebuilt
+ * with. NestedField casts the initial default to its type, which converts
the literals of the SDK promotions,
+ * but in Iceberg 1.12.0 a DATE literal converts only to DATE: a DATE
default moves to midnight of its day in the
+ * promoted TIMESTAMP or TIMESTAMP_NS, and a date outside that type's
range throws, as the format requires.
+ */
+ private static Literal<?> promotedDefault(Literal<?> value, Type from,
Type to) {
+ if (value == null || from.typeId() != Type.TypeID.DATE ||
to.equals(from)) {
+ return value;
+ }
+ return Literal.of(DateTimeUtil.microsFromTimestamp(
+ DateTimeUtil.dateFromDays((Integer)
value.value()).atStartOfDay())).to(to);
+ }
+
+ private static void addHistoricalEqualityFields(List<NestedField> fields,
Set<Integer> missingFieldIds,
+ Schema historicalSchema, List<Schema> retainedSchemas) {
Map<Integer, NestedField> historicalFields =
TypeUtil.indexById(historicalSchema.asStruct());
Set<Integer> selectedFieldIds = new HashSet<>();
for (Integer fieldId : missingFieldIds) {
@@ -2345,7 +2458,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return;
}
Schema selectedSchema = TypeUtil.select(historicalSchema,
selectedFieldIds);
- mergeHistoricalEqualityFields(fields, selectedSchema.columns());
+ mergeHistoricalEqualityFields(fields,
withRetainedRelaxations(selectedSchema.columns(), retainedSchemas));
missingFieldIds.removeAll(selectedFieldIds);
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index b0a6866de3a..d997e826e05 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -50,6 +50,7 @@ import org.apache.doris.thrift.schema.external.TFieldPtr;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
@@ -62,6 +63,7 @@ import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotSummary;
@@ -88,6 +90,7 @@ import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.metrics.ScanReport;
import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.junit.jupiter.api.Assertions;
@@ -108,6 +111,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
@@ -1925,7 +1929,8 @@ public class IcebergScanPlanProviderTest {
@Test
public void equalityCarrierKeepsLatestDefinitionOfDroppedEqualityKey()
throws Exception {
// BE fills an equality key that an old data file lacks from the key's
carrier entry, so a dropped key must
- // resolve to its latest historical definition: its last name, its
optionality and its initial default.
+ // resolve to the latest definition on the scan's lineage: its last
name, its optionality and its initial
+ // default. Each schema change is followed by a write, as on a table
that keeps ingesting.
Schema schema = new Schema(
Types.NestedField.required(1, "id", Types.IntegerType.get()),
Types.NestedField.optional("k").withId(2).ofType(Types.IntegerType.get())
@@ -1938,7 +1943,11 @@ public class IcebergScanPlanProviderTest {
table.newRowDelta().addDeletes(equalityDeleteFile(
"s3://b/db/renamed_dropped_eq_key/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
table.updateSchema().renameColumn("k", "k2").commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/renamed_dropped_eq_key/f2.parquet", 1024, null,
null)).commit();
table.updateSchema().deleteColumn("k2").commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/renamed_dropped_eq_key/f3.parquet", 1024, null,
null)).commit();
IcebergScanPlanProvider provider = providerOver(table);
Map<String, String> props = provider.getScanNodeProperties(
@@ -2069,6 +2078,531 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals(table.schema().columns(), carrier);
}
+ @Test
+ public void equalityCarrierResolvesDroppedKeyWhenSchemaIdIsReused() throws
Exception {
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ Table table = createTable("reused_schema_id_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ int initialSchemaId = table.schema().schemaId();
+ table.updateSchema().addColumn("k", Types.StringType.get()).commit();
+ int keyId = table.schema().findField("k").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/reused_schema_id_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/reused_schema_id_eq_key/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+ // Dropping k makes the schema equal to the initial one, so Iceberg
reuses the initial schema ID, which the
+ // metadata list holds before the schema that still defines the
equality key. The next snapshot carries the
+ // reused ID; only its parent records the schema with k.
+ table.updateSchema().deleteColumn("k").commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/reused_schema_id_eq_key/f2.parquet", 1024, null,
null)).commit();
+ Assertions.assertEquals(initialSchemaId, table.schema().schemaId());
+ Assertions.assertEquals(initialSchemaId,
table.currentSnapshot().schemaId());
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(keyId, carrier.get(1).fieldId());
+ Assertions.assertEquals("k", carrier.get(1).name());
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(
+ null, new IcebergTableHandle("db1", "reused_schema_id_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
+ TFileScanRangeParams params = new TFileScanRangeParams();
+ provider.populateScanLevelParams(params, props);
+ List<TFieldPtr> fields =
params.getHistorySchemaInfo().get(0).getRootField().getFields();
+ Assertions.assertTrue(fields.stream().anyMatch(field ->
field.getFieldPtr().getId() == keyId),
+ "the scan dictionary must carry the dropped equality key after
the schema ID is reused");
+ }
+
+ @Test
+ public void
equalityCarrierKeepsPromotedTypeOfKeyDroppedBeforeItsDeleteCommits() throws
Exception {
+ // A writer started after a schema-only promotion of k to LONG writes
LONG tombstones, and k is dropped before
+ // its delete commits. The delete snapshot records the schema without
k and its parent the INT one, so only
+ // the metadata list knows LONG; an INT carrier would narrow
2147483648 to NULL and match NULL keys.
+ Schema schema = new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.IntegerType.get()));
+ Table table = createTable("promoted_dropped_eq_key", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/promoted_dropped_eq_key/f1.parquet", 1024, null,
null)).commit();
+ int intSchemaId = table.currentSnapshot().schemaId();
+ table.updateSchema().updateColumn("k", Types.LongType.get()).commit();
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/promoted_dropped_eq_key/eq.parquet",
FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+
Assertions.assertNull(table.schemas().get(table.currentSnapshot().schemaId()).findField(2));
+ Assertions.assertEquals(intSchemaId,
table.snapshot(table.currentSnapshot().parentId()).schemaId());
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(2, carrier.get(1).fieldId());
+ Assertions.assertEquals("k", carrier.get(1).name());
+ Assertions.assertEquals(Types.LongType.get(), carrier.get(1).type());
+ }
+
+ @Test
+ public void equalityCarrierKeepsLineageTimestampOfPromotedDateKey() throws
Exception {
+ // v3 promotes DATE to an unzoned TIMESTAMP, which
TypeUtil.isPromotionAllowed does not know. The delete
+ // snapshot records the TIMESTAMP schema; a DATE carrier would
truncate a noon tombstone to its date and
+ // delete the midnight row.
+ Table table = createTable("promoted_date_lineage_eq_key", new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.DateType.get())),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/promoted_date_lineage_eq_key/f1.parquet", 1024,
null, null)).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampType.withoutZone())), 2);
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/promoted_date_lineage_eq_key/eq.parquet",
FileFormat.PARQUET, 2)).commit();
+ table.updateSchema().deleteColumn("k").commit();
+ Assertions.assertEquals(Types.TimestampType.withoutZone(),
+
table.schemas().get(table.currentSnapshot().schemaId()).findType(2));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(Types.TimestampType.withoutZone(),
carrier.get(1).type());
+ }
+
+ @Test
+ public void equalityCarrierWidensDateKeyPromotedWithoutASnapshot() throws
Exception {
+ // A writer started after a schema-only v3 promotion of k to
TIMESTAMP_NS writes TIMESTAMP_NS tombstones,
+ // and k is dropped before its delete commits, so only the metadata
list knows the promotion.
+ Table table = createTable("promoted_date_dropped_eq_key", new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.DateType.get())),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/promoted_date_dropped_eq_key/f1.parquet", 1024,
null, null)).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampNanoType.withoutZone())), 2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/promoted_date_dropped_eq_key/eq.parquet",
FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(Types.DateType.get(), table.schemas()
+
.get(table.snapshot(table.currentSnapshot().parentId()).schemaId()).findType(2));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(Types.TimestampNanoType.withoutZone(),
carrier.get(1).type());
+ }
+
+ @Test
+ public void equalityCarrierFollowsUnknownKeyPromotedWithoutASnapshot()
throws Exception {
+ // v3 lets UNKNOWN become any type, which TypeUtil.isPromotionAllowed
does not know. k is promoted to DATE and
+ // then TIMESTAMP_NS without snapshots, and a pending append and a
pending delete commit after k is dropped,
+ // so the only snapshot that records k has it as UNKNOWN. An UNKNOWN
carrier is sent as STRING, and V2 would
+ // compare 2024-01-02 with 2024-01-02 00:00:00.000000000 and keep the
deleted row.
+ Table table = createTable("promoted_unknown_dropped_eq_key", new
Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.UnknownType.get())),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/promoted_unknown_dropped_eq_key/f1.parquet", 1024,
null, null)).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.DateType.get())), 2);
+ AppendFiles pendingAppend =
table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/promoted_unknown_dropped_eq_key/f2.parquet", 1024,
null, null));
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampNanoType.withoutZone())), 2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/promoted_unknown_dropped_eq_key/eq.parquet",
FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingAppend.commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(Types.UnknownType.get(),
table.schemas().get(table.snapshot(table.snapshot(
+
table.currentSnapshot().parentId()).parentId()).schemaId()).findType(2));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(Types.TimestampNanoType.withoutZone(),
carrier.get(1).type());
+ Assertions.assertTrue(carrier.get(1).isOptional(), carrier.toString());
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(null,
+ new IcebergTableHandle("db1",
"promoted_unknown_dropped_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
+ Assertions.assertEquals(TPrimitiveType.TIMESTAMP_NS,
+ dictFieldById(dictRootFields(provider, props),
2).getType().getType());
+ }
+
+ @Test
+ public void equalityCarrierConvertsDefaultsOfDateKeyPromotedToTimestamp()
throws Exception {
+ assertPromotedDateKeyDefaults("promoted_date_default_eq_key",
Types.TimestampType.withoutZone());
+ }
+
+ @Test
+ public void
equalityCarrierConvertsDefaultsOfDateKeyPromotedToTimestampNs() throws
Exception {
+ assertPromotedDateKeyDefaults("promoted_date_nanos_default_eq_key",
Types.TimestampNanoType.withoutZone());
+ }
+
+ /**
+ * Gives a v3 DATE key both defaults, promotes it to {@code promotedType}
without a snapshot and drops it while an
+ * equality delete on it is pending. The lineage selects the DATE
definition and the carrier rebuilds it in the
+ * promoted type, which casts its initial default; a DATE literal converts
only to DATE, so an unconverted
+ * initial default fails planning with "Cannot cast default value". The
rebuilt field has no write default.
+ */
+ private static void assertPromotedDateKeyDefaults(String name, Type
promotedType) {
+ Literal<Integer> date =
Literal.of("2024-01-02").to(Types.DateType.get());
+ Literal<Long> midnight =
Literal.of("2024-01-02T00:00:00").to(promotedType);
+ Table table = createTable(name, new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+
Types.NestedField.optional("k").withId(2).ofType(Types.DateType.get())
+
.withInitialDefault(date).withWriteDefault(date).build()),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/" +
name + "/f1.parquet", 1024, null, null))
+ .commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional("k").withId(2).ofType(promotedType)
+
.withInitialDefault(midnight).withWriteDefault(midnight).build()), 2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/" + name + "/eq.parquet", FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(Types.DateType.get(), table.schemas()
+
.get(table.snapshot(table.currentSnapshot().parentId()).schemaId()).findType(2));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(promotedType, carrier.get(1).type());
+ Assertions.assertEquals(midnight,
carrier.get(1).initialDefaultLiteral());
+ Assertions.assertNull(carrier.get(1).writeDefaultLiteral());
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(null, new
IcebergTableHandle("db1", name),
+ Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
+ Assertions.assertEquals("2024-01-02 00:00:00",
+ dictFieldById(dictRootFields(provider, props),
2).getInitialDefaultValue());
+ }
+
+ @Test
+ public void equalityCarrierDropsStaleWriteDefaultOfPromotedDateKey()
throws Exception {
+ // The snapshot that records k has a DATE write default outside the
TIMESTAMP_NS range. The write default is
+ // cleared and k is promoted to TIMESTAMP_NS without snapshots, then k
is dropped while a delete on it is
+ // pending. No reader uses a write default, so the carrier must not
convert the stale one, which overflows.
+ Literal<Integer> outOfRange =
Literal.of("1000-01-01").to(Types.DateType.get());
+ Table table = createTable("stale_write_default_eq_key", new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+
Types.NestedField.optional("k").withId(2).ofType(Types.DateType.get())
+ .withWriteDefault(outOfRange).build()),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/stale_write_default_eq_key/f1.parquet", 1024, null,
null)).commit();
+ table.updateSchema().updateColumnDefault("k", null).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampNanoType.withoutZone())), 2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/stale_write_default_eq_key/eq.parquet",
FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(outOfRange.value(), table.schemas()
+
.get(table.snapshot(table.currentSnapshot().parentId()).schemaId()).findField(2).writeDefault());
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(Types.TimestampNanoType.withoutZone(),
carrier.get(1).type());
+ Assertions.assertNull(carrier.get(1).initialDefaultLiteral());
+ Assertions.assertNull(carrier.get(1).writeDefaultLiteral());
+
+ IcebergScanPlanProvider provider = providerOver(table);
+ Map<String, String> props = provider.getScanNodeProperties(null,
+ new IcebergTableHandle("db1", "stale_write_default_eq_key"),
+ Collections.singletonList(new IcebergColumnHandle("id", 1)),
Optional.empty());
+ Assertions.assertFalse(dictFieldById(dictRootFields(provider, props),
2).isSetInitialDefaultValue());
+ }
+
+ @Test
+ public void
equalityCarrierRejectsDateKeyPromotedToTimestampThenTimestampNs() throws
Exception {
+ assertUnrelatedDateKeyPromotionsFailPlanning("date_ts_then_ns_eq_key",
Types.TimestampType.withoutZone(),
+ Types.TimestampNanoType.withoutZone());
+ }
+
+ @Test
+ public void
equalityCarrierRejectsDateKeyPromotedToTimestampNsThenTimestamp() throws
Exception {
+ assertUnrelatedDateKeyPromotionsFailPlanning("date_ns_then_ts_eq_key",
Types.TimestampNanoType.withoutZone(),
+ Types.TimestampType.withoutZone());
+ }
+
+ /**
+ * Promotes a v3 DATE key to {@code first} without a snapshot, sets the
DATE schema current again (publishing the
+ * identical schema reuses its ID), promotes the key to {@code second} and
drops it while an equality delete on
+ * it is pending. TIMESTAMP and TIMESTAMP_NS do not promote to each other
and the metadata does not record which
+ * one the delete was written in, so planning must fail in either order: a
TIMESTAMP carrier would round the
+ * tombstone 2024-01-02T00:00:00.000000001 to midnight and delete the DATE
row.
+ */
+ private static void assertUnrelatedDateKeyPromotionsFailPlanning(String
name, Type first, Type second) {
+ Schema dateSchema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.DateType.get()));
+ Table table = createTable(name, dateSchema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/" +
name + "/f1.parquet", 1024, null, null))
+ .commit();
+ int dateSchemaId = table.schema().schemaId();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", first)), 2);
+ publishSchema(table, dateSchema, 2);
+ Assertions.assertEquals(dateSchemaId, table.schema().schemaId());
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", second)), 2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/" + name + "/eq.parquet", FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(Types.DateType.get(), table.schemas()
+
.get(table.snapshot(table.currentSnapshot().parentId()).schemaId()).findType(2));
+
+ IllegalStateException error =
Assertions.assertThrows(IllegalStateException.class, () ->
+
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(table,
table.newScan(), table.schema()));
+ Assertions.assertEquals("Iceberg equality-delete field 2 (k) was
promoted to both " + first + " and "
+ + second + " in schema history", error.getMessage());
+ IcebergScanPlanProvider provider = providerOver(table);
+ Assertions.assertThrows(IllegalStateException.class, () ->
provider.getScanNodeProperties(null,
+ new IcebergTableHandle("db1", name),
Collections.singletonList(new IcebergColumnHandle("id", 1)),
+ Optional.empty()));
+ }
+
+ @Test
+ public void equalityCarrierKeepsLineagePromotionBesideAnAbandonedOne()
throws Exception {
+ // The schemas of the rollback history above, but the delete commits
while TIMESTAMP_NS is current, so the
+ // lineage records TIMESTAMP_NS. The abandoned TIMESTAMP schema is not
a promotion of that definition and
+ // must not fail planning.
+ Schema dateSchema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.DateType.get()));
+ Table table = createTable("date_ns_lineage_beside_ts_eq_key",
dateSchema, PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/date_ns_lineage_beside_ts_eq_key/f1.parquet", 1024,
null, null)).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampType.withoutZone())), 2);
+ publishSchema(table, dateSchema, 2);
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k",
Types.TimestampNanoType.withoutZone())), 2);
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/date_ns_lineage_beside_ts_eq_key/eq.parquet",
FileFormat.PARQUET, 2)).commit();
+ table.updateSchema().deleteColumn("k").commit();
+ Assertions.assertEquals(Types.TimestampNanoType.withoutZone(),
+
table.schemas().get(table.currentSnapshot().schemaId()).findType(2));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(Types.TimestampNanoType.withoutZone(),
carrier.get(1).type());
+ }
+
+ @Test
+ public void equalityCarrierRejectsUnknownKeyPromotedToUnrelatedTypes()
throws Exception {
+ // v3 lets UNKNOWN become any type. k is promoted to INT, set back to
UNKNOWN and promoted to STRING without
+ // snapshots, so INT and STRING are both promotions of the lineage's
UNKNOWN. An INT carrier would turn a
+ // non-numeric STRING tombstone into NULL, which matches every row
written while k was UNKNOWN.
+ Schema unknownSchema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.UnknownType.get()));
+ Table table = createTable("unknown_int_then_string_eq_key",
unknownSchema, PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/unknown_int_then_string_eq_key/f1.parquet", 1024,
null, null)).commit();
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.IntegerType.get())),
2);
+ publishSchema(table, unknownSchema, 2);
+ publishSchema(table, new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "k", Types.StringType.get())),
2);
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/unknown_int_then_string_eq_key/eq.parquet",
FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ pendingDelete.commit();
+ Assertions.assertEquals(Types.UnknownType.get(), table.schemas()
+
.get(table.snapshot(table.currentSnapshot().parentId()).schemaId()).findType(2));
+
+ IllegalStateException error =
Assertions.assertThrows(IllegalStateException.class, () ->
+
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(table,
table.newScan(), table.schema()));
+ Assertions.assertEquals("Iceberg equality-delete field 2 (k) was
promoted to both int and string"
+ + " in schema history", error.getMessage());
+ }
+
+ @Test
+ public void equalityCarrierKeepsSchemaOnlyOptionalityOfDroppedKey() throws
Exception {
+ // k becomes optional without a snapshot, and both a pending delete
and an append without k commit after k
+ // is dropped, so every snapshot on the lineage records either no k or
the required one. A required carrier
+ // would make both readers reject the appended file for a missing
required field.
+ Table table = droppedRequiredKeyTable("optional_dropped_eq_key", true);
+
Assertions.assertTrue(table.schemas().get(table.snapshot(table.snapshot(
+
table.currentSnapshot().parentId()).parentId()).schemaId()).findField(2).isRequired());
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertTrue(carrier.get(1).isOptional(), carrier.toString());
+ Assertions.assertEquals(Types.IntegerType.get(),
carrier.get(1).type());
+ }
+
+ @Test
+ public void equalityCarrierKeepsAlwaysRequiredDroppedKeyRequired() throws
Exception {
+ Table table = droppedRequiredKeyTable("required_dropped_eq_key",
false);
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertTrue(carrier.get(1).isRequired(), carrier.toString());
+ }
+
+ /**
+ * Appends a file with a required k, optionally makes k optional without a
snapshot, then drops k while an
+ * equality delete on it is pending and commits an append without k before
the delete.
+ */
+ private static Table droppedRequiredKeyTable(String name, boolean
makeOptional) {
+ Table table = createTable(name, new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "k", Types.IntegerType.get())),
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/" +
name + "/f1.parquet", 1024, null, null))
+ .commit();
+ if (makeOptional) {
+ table.updateSchema().makeColumnOptional("k").commit();
+ }
+ RowDelta pendingDelete =
table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/" + name + "/eq.parquet", FileFormat.PARQUET, 2));
+ table.updateSchema().deleteColumn("k").commit();
+ table.newAppend().appendFile(dataFile(table.spec(), "s3://b/db/" +
name + "/f2.parquet", 1024, null, null))
+ .commit();
+ pendingDelete.commit();
+
Assertions.assertNull(table.schemas().get(table.currentSnapshot().schemaId()).findField(2));
+ return table;
+ }
+
+ /** Publishes a schema directly: SchemaUpdate in Iceberg 1.12.0 rejects
the v3 DATE promotions. */
+ private static void publishSchema(Table table, Schema schema, int
lastColumnId) {
+ org.apache.iceberg.TableOperations operations =
((org.apache.iceberg.HasTableOperations) table).operations();
+ org.apache.iceberg.TableMetadata base = operations.current();
+ operations.commit(base,
org.apache.iceberg.TableMetadata.buildFrom(base)
+ .setCurrentSchema(schema, lastColumnId).build());
+ table.refresh();
+ }
+
+ @Test
+ public void
equalityCarrierResolvesReusedSchemaIdAfterParentSnapshotsExpire() throws
Exception {
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ Table table = createTable("reused_schema_id_expired_parent", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.updateSchema().addColumn("k", Types.StringType.get()).commit();
+ int keyId = table.schema().findField("k").fieldId();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/reused_schema_id_expired_parent/f1.parquet", 1024,
null, null)).commit();
+ long appendSnapshotId = table.currentSnapshot().snapshotId();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/reused_schema_id_expired_parent/eq.parquet",
FileFormat.PARQUET, keyId)).commit();
+ long deleteSnapshotId = table.currentSnapshot().snapshotId();
+ table.updateSchema().deleteColumn("k").commit();
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/reused_schema_id_expired_parent/f2.parquet", 1024,
null, null)).commit();
+ // With the snapshots that recorded k expired, the scan lineage ends
at the reused-ID snapshot, so the key
+ // can only come from the metadata schema list, where it sits after
the reused schema.
+
table.expireSnapshots().expireSnapshotId(appendSnapshotId).expireSnapshotId(deleteSnapshotId)
+ .cleanExpiredFiles(false).commit();
+
Assertions.assertNull(table.snapshot(table.currentSnapshot().parentId()));
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, table.newScan(), table.schema());
+ Assertions.assertEquals(2, carrier.size(), "the dropped equality key
must stay resolvable: " + carrier);
+ Assertions.assertEquals(keyId, carrier.get(1).fieldId());
+ Assertions.assertEquals("k", carrier.get(1).name());
+ }
+
+ @Test
+ public void
equalityCarrierPrefersScanLineageOverLaterRenameForTimeTravel() {
+ // The metadata list also holds schemas committed after a time-travel
target. A later schema that renames
+ // the key must not win over the definition recorded on the target's
own lineage.
+ NestedField id = Types.NestedField.required(1, "id",
Types.LongType.get());
+ Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id,
intKeyWithDefault("k")));
+ Schema timeTravelSchema = new Schema(1, Collections.singletonList(id));
+ Schema laterSchema = new Schema(2, Arrays.asList(id,
intKeyWithDefault("k2")));
+ Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L,
timeTravelSchema);
+ Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey,
timeTravelSchema, laterSchema),
+ snapshotOf(1000L, null, schemaWithEqualityKey),
timeTravelSnapshot);
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, timeTravelSnapshot, timeTravelSchema,
Collections.singleton(7));
+ Assertions.assertEquals(2, carrier.size());
+ Assertions.assertEquals(7, carrier.get(1).fieldId());
+ Assertions.assertEquals("k", carrier.get(1).name());
+ }
+
+ @Test
+ public void
equalityCarrierKeepsDefaultWhenTimeTravelParentExpiredBeforeLaterRename() {
+ // Once the lineage that recorded the key has expired, the metadata
list is the only source left. Its newest
+ // definition may come from a later rename, but the field ID and its
initial default stay the same.
+ NestedField id = Types.NestedField.required(1, "id",
Types.LongType.get());
+ Schema schemaWithEqualityKey = new Schema(100, Arrays.asList(id,
intKeyWithDefault("k")));
+ Schema timeTravelSchema = new Schema(1, Collections.singletonList(id));
+ Schema laterSchema = new Schema(2, Arrays.asList(id,
intKeyWithDefault("k2")));
+ Snapshot timeTravelSnapshot = snapshotOf(1001L, 1000L,
timeTravelSchema);
+ Table table = tableWithHistory(Arrays.asList(schemaWithEqualityKey,
timeTravelSchema, laterSchema),
+ timeTravelSnapshot);
+
+ List<NestedField> carrier =
IcebergScanPlanProvider.schemaForPotentialEqualityDeletes(
+ table, timeTravelSnapshot, timeTravelSchema,
Collections.singleton(7));
+ Assertions.assertEquals(2, carrier.size());
+ Assertions.assertEquals(7, carrier.get(1).fieldId());
+ Assertions.assertEquals(7, carrier.get(1).initialDefault());
+ }
+
+ private static NestedField intKeyWithDefault(String name) {
+ return
Types.NestedField.optional(name).withId(7).ofType(Types.IntegerType.get())
+ .withInitialDefault(7).build();
+ }
+
+ /** A snapshot double exposing only its lineage and the ID of the schema
it was written with. */
+ private static Snapshot snapshotOf(long snapshotId, Long parentId, Schema
schema) {
+ return (Snapshot)
Proxy.newProxyInstance(Snapshot.class.getClassLoader(), new Class<?>[]
{Snapshot.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "snapshotId":
+ return snapshotId;
+ case "parentId":
+ return parentId;
+ case "schemaId":
+ return schema.schemaId();
+ default:
+ throw new
UnsupportedOperationException(method.getName());
+ }
+ });
+ }
+
+ /**
+ * A table double exposing only schema history, in metadata-list order,
and the retained snapshots. It allows
+ * schema histories that the SDK cannot build, such as a later schema that
renames a dropped field ID.
+ */
+ private static Table tableWithHistory(List<Schema> schemas, Snapshot...
snapshots) {
+ Map<Integer, Schema> schemasById = new LinkedHashMap<>();
+ for (Schema schema : schemas) {
+ schemasById.put(schema.schemaId(), schema);
+ }
+ Map<Long, Snapshot> snapshotsById = new HashMap<>();
+ for (Snapshot snapshot : snapshots) {
+ snapshotsById.put(snapshot.snapshotId(), snapshot);
+ }
+ return (Table) Proxy.newProxyInstance(Table.class.getClassLoader(),
new Class<?>[] {Table.class},
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "schemas":
+ return schemasById;
+ case "snapshot":
+ return snapshotsById.get((Long) args[0]);
+ default:
+ throw new
UnsupportedOperationException(method.getName());
+ }
+ });
+ }
+
@Test
public void
partitionPrunedEqualityDeleteConservativelyRequiresCurrentBackendSemantics() {
PartitionSpec spec =
PartitionSpec.builderFor(PART_SCHEMA).identity("p").build();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
