adityaksolves commented on code in PR #18517:
URL: https://github.com/apache/iceberg/pull/18517#discussion_r4409268627
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -239,4 +241,138 @@ void testCaseSensitiveCachingDoesNotMatch() {
assertThat(cache.schema(tableIdentifier, SCHEMA_MIXEDCASE))
.isEqualTo(TableMetadataCache.NOT_FOUND);
}
+
+ @Test
+ void testSchemaPrefersCurrentSchemaAfterDropAndReAddColumn() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int retiredFieldId = table.schema().findField("extra").fieldId();
+ table.updateSchema().deleteColumn("extra").commit();
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("extra").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
table.schema()).resolvedTableSchema();
+
+
assertThat(resolved.findField("extra").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaStillMatchesHistoricalSchemaWhenDroppedColumnIsNotReAdded() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ Schema withExtra = table.schema();
+ table.updateSchema().deleteColumn("extra").commit();
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
withExtra).resolvedTableSchema();
+
+ assertThat(resolved.findField("extra")).isNotNull();
+ }
+
+ @Test
+ void testSchemasAreOrderedCurrentFirstThenNewestToOldest() {
+ Map<Integer, Schema> schemas = Maps.newHashMap();
+ for (int id = 0; id < 4; id++) {
+ schemas.put(id, SCHEMA);
+ }
+
+ assertThat(TableMetadataCache.orderSchemas(schemas,
3).keySet()).containsExactly(3, 2, 1, 0);
+
+ // after a rollback the current schema is not the one with the highest id
+ assertThat(TableMetadataCache.orderSchemas(schemas,
1).keySet()).containsExactly(1, 3, 2, 0);
+ }
+
+ @Test
+ void testSchemaPrefersCurrentSchemaWhenColumnIsReAddedAfterAnotherChange() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("c", Types.StringType.get()).commit();
+ Schema oldShape = table.schema();
+ int retiredFieldId = oldShape.findField("c").fieldId();
+ table.updateSchema().deleteColumn("c").commit();
+ table.updateSchema().addColumn("d", Types.StringType.get()).commit();
+ table.updateSchema().addColumn("c", Types.StringType.get()).commit();
+ int currentFieldId = table.schema().findField("c").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
oldShape).resolvedTableSchema();
+
+ assertThat(resolved.findField("c").fieldId()).isEqualTo(currentFieldId);
+ }
+
+ @Test
+ void testSchemaPrefersCurrentSchemaWhenColumnIsReAddedWithWiderType() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA);
+ table.updateSchema().addColumn("c", Types.IntegerType.get()).commit();
+ Schema oldShape = table.schema();
+ int retiredFieldId = oldShape.findField("c").fieldId();
+ table.updateSchema().deleteColumn("c").commit();
+ table.updateSchema().addColumn("c", Types.LongType.get()).commit();
+ int currentFieldId = table.schema().findField("c").fieldId();
+ assertThat(currentFieldId).isNotEqualTo(retiredFieldId);
+
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+
+ Schema resolved = cache.schema(tableIdentifier,
oldShape).resolvedTableSchema();
+
+ assertThat(resolved.findField("c").fieldId()).isEqualTo(currentFieldId);
+ assertThat(resolved.findField("c").type()).isEqualTo(Types.LongType.get());
Review Comment:
Added `testWriteOldShapeRecordAfterColumnReAddedWithWiderType` to
`TestDynamicIcebergSink` in e67fff9. Column `c` (int) is dropped and re-added
as long, then an old-shape record is written and read back through the current
schema, where `c` has the value. It fails without the tiering change.
--
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]
