huadongliu opened a new issue #2493:
URL: https://github.com/apache/iceberg/issues/2493
Metadata file missing exception was seen when expiring snapshots with
deleting metadata enabled. The test ran fine without upgrading to V2.
```
org.apache.iceberg.exceptions.ValidationException: Metadata file for version
1 is missing
at
org.apache.iceberg.hadoop.HadoopTableOperations.refresh(HadoopTableOperations.java:106)
at
org.apache.iceberg.hadoop.HadoopTableOperations.current(HadoopTableOperations.java:80)
at org.apache.iceberg.RemoveSnapshots.<init>(RemoveSnapshots.java:87)
at org.apache.iceberg.BaseTable.expireSnapshots(BaseTable.java:189)
public class TestRecord {
private String userId;
private Timestamp eventTime;
private Integer count;
public TestRecord(String userId, Timestamp eventTime, Integer count) {
this.userId = userId;
this.eventTime = eventTime;
this.count = count;
}
public String getUserId() {
return userId;
}
public Timestamp getEventTime() {
return eventTime;
}
public Integer getCount() {
return count;
}
}
public void testDeleteAndExpire() throws Exception {
Schema schema = new Schema(
optional(1, "userId", Types.StringType.get()),
optional(2, "eventTime", Types.TimestampType.withZone()),
optional(3, "count", Types.IntegerType.get())
);
PartitionSpec spec = PartitionSpec.builderFor(schema)
.day("eventTime")
.truncate("userId", 2)
.build();
Map<String, String> options = Maps.newHashMap();
options.put("write.metadata.delete-after-commit.enabled", "true");
options.put("write.metadata.previous-versions-max", "1");
Table table = TABLES.create(schema, spec,
SortOrder.builderFor(schema).withOrderId(1).asc("userId", NULLS_LAST).build(),
options, tableLocation);
TableOperations ops = ((BaseTable) table).operations();
TableMetadata meta = ops.current();
ops.commit(meta, meta.upgradeToFormatVersion(2));
List<TestRecord> records = Lists.newArrayList(
new TestRecord("112", Timestamp.valueOf("2021-04-15 00:00:00"),
112),
new TestRecord("111", Timestamp.valueOf("2021-04-15 00:00:00"),
111),
new TestRecord("114", Timestamp.valueOf("2021-04-16 00:00:00"),
114),
new TestRecord("113", Timestamp.valueOf("2021-04-16 00:00:00"),
113),
new TestRecord("222", Timestamp.valueOf("2021-04-15 00:00:00"),
222),
new TestRecord("221", Timestamp.valueOf("2021-04-15 00:00:00"),
221),
new TestRecord("224", Timestamp.valueOf("2021-04-16 00:00:00"),
224),
new TestRecord("223", Timestamp.valueOf("2021-04-16 00:00:00"),
223)
);
Dataset<Row> df = spark.createDataFrame(records, TestRecord.class);
df.sortWithinPartitions("userId").write().format("iceberg").mode("append").save(tableLocation);
List<TestRecord> records2 = Lists.newArrayList(
new TestRecord("116", Timestamp.valueOf("2021-04-15 00:00:00"),
116),
new TestRecord("115", Timestamp.valueOf("2021-04-15 00:00:00"),
115),
new TestRecord("118", Timestamp.valueOf("2021-04-16 00:00:00"),
118),
new TestRecord("117", Timestamp.valueOf("2021-04-16 00:00:00"),
117),
new TestRecord("226", Timestamp.valueOf("2021-04-15 00:00:00"),
226),
new TestRecord("225", Timestamp.valueOf("2021-04-15 00:00:00"),
225),
new TestRecord("228", Timestamp.valueOf("2021-04-16 00:00:00"),
228),
new TestRecord("227", Timestamp.valueOf("2021-04-16 00:00:00"),
227)
);
Dataset<Row> df2 = spark.createDataFrame(records2, TestRecord.class);
df2.sortWithinPartitions("userId").write().format("iceberg").mode("append").save(tableLocation);
long tsToExpire = System.currentTimeMillis();
table.expireSnapshots()
.expireOlderThan(tsToExpire)
.commit();
}
```
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
