wombatu-kun commented on code in PR #18858:
URL: https://github.com/apache/iceberg/pull/18858#discussion_r4523464855
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableUpdater.java:
##########
@@ -170,6 +182,9 @@ private TableMetadataCache.ResolvedSchemaInfo
findOrCreateSchema(
if (newSchema.compareResult() !=
CompareSchemasVisitor.Result.SCHEMA_UPDATE_NEEDED) {
LOG.debug("Table {} schema updated concurrently to {}",
identifier, schema);
return newSchema;
+ } else if (attemptsLeft > 1) {
Review Comment:
No test covers a conflict on every attempt, so nothing checks that the
update gives up after `MAX_ATTEMPTS` and rethrows `CommitFailedException`
instead of retrying further. Letting `ConcurrentCommitCatalog` commit on every
load would cover it; not a blocker.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -514,4 +516,145 @@ void testTableCreatedForCatalogWithoutNamespaceSupport() {
assertThat(catalog.tableExists(tableIdentifier)).isTrue();
}
+
+ @Test
+ void specUpdateSucceedsDespiteUnrelatedConcurrentCommit() {
+ ConcurrentCommitCatalog catalog = concurrentCommitCatalog();
+ TableIdentifier tableIdentifier =
TableIdentifier.of(Namespace.of("myNamespace"), "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+ catalog.commitUnrelatedChangeOnNextLoad();
+
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).bucket("data",
10).build();
+ PartitionSpec updated =
+ tableUpdater.update(
+ tableIdentifier, SnapshotRef.MAIN_BRANCH, SCHEMA, spec,
TableCreator.DEFAULT)
+ .f1;
+
+ assertThat(PartitionSpecEvolution.checkCompatibility(updated,
spec)).isTrue();
+ assertThat(catalog.loadTable(tableIdentifier).spec()).isEqualTo(updated);
+ }
+
+ @Test
+ void schemaUpdateSucceedsDespiteUnrelatedConcurrentCommit() {
+ ConcurrentCommitCatalog catalog = concurrentCommitCatalog();
+ TableIdentifier tableIdentifier =
TableIdentifier.of(Namespace.of("myNamespace"), "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+ catalog.commitUnrelatedChangeOnNextLoad();
+
+ Schema updated =
+ tableUpdater
+ .update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA2,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT)
+ .f0
+ .resolvedTableSchema();
+
+ assertThat(updated.sameSchema(SCHEMA2)).isTrue();
+
assertThat(catalog.loadTable(tableIdentifier).schema().sameSchema(SCHEMA2)).isTrue();
+ }
+
+ @Test
+ void tableCreatedByAnotherWriterWhileCreatingIsLoaded() {
Review Comment:
This takes the same path as the existing `testTableAlreadyExists`, where
`createTable` also throws `AlreadyExistsException` and the table then loads.
Consider dropping it.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableUpdater.java:
##########
@@ -514,4 +516,145 @@ void testTableCreatedForCatalogWithoutNamespaceSupport() {
assertThat(catalog.tableExists(tableIdentifier)).isTrue();
}
+
+ @Test
+ void specUpdateSucceedsDespiteUnrelatedConcurrentCommit() {
+ ConcurrentCommitCatalog catalog = concurrentCommitCatalog();
+ TableIdentifier tableIdentifier =
TableIdentifier.of(Namespace.of("myNamespace"), "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+ catalog.commitUnrelatedChangeOnNextLoad();
+
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).bucket("data",
10).build();
+ PartitionSpec updated =
+ tableUpdater.update(
+ tableIdentifier, SnapshotRef.MAIN_BRANCH, SCHEMA, spec,
TableCreator.DEFAULT)
+ .f1;
+
+ assertThat(PartitionSpecEvolution.checkCompatibility(updated,
spec)).isTrue();
+ assertThat(catalog.loadTable(tableIdentifier).spec()).isEqualTo(updated);
+ }
+
+ @Test
+ void schemaUpdateSucceedsDespiteUnrelatedConcurrentCommit() {
+ ConcurrentCommitCatalog catalog = concurrentCommitCatalog();
+ TableIdentifier tableIdentifier =
TableIdentifier.of(Namespace.of("myNamespace"), "myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT);
+ catalog.commitUnrelatedChangeOnNextLoad();
+
+ Schema updated =
+ tableUpdater
+ .update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA2,
+ PartitionSpec.unpartitioned(),
+ TableCreator.DEFAULT)
+ .f0
+ .resolvedTableSchema();
+
+ assertThat(updated.sameSchema(SCHEMA2)).isTrue();
+
assertThat(catalog.loadTable(tableIdentifier).schema().sameSchema(SCHEMA2)).isTrue();
+ }
+
+ @Test
+ void tableCreatedByAnotherWriterWhileCreatingIsLoaded() {
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("myTable");
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, Long.MAX_VALUE, 10,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ TableUpdater tableUpdater = new TableUpdater(cache, catalog,
CASE_SENSITIVE, PRESERVE_COLUMNS);
+ // Another writer creates the table right before this one does
+ TableCreator losingTheRace =
+ (catalog1, identifier, schema, spec) -> {
+ catalog1.createTable(identifier, schema, spec);
+ throw new AlreadyExistsException("Table already exists: %s",
identifier);
+ };
+
+ Tuple2<TableMetadataCache.ResolvedSchemaInfo, PartitionSpec> result =
+ tableUpdater.update(
+ tableIdentifier,
+ SnapshotRef.MAIN_BRANCH,
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ losingTheRace);
+
+
assertThat(result.f0.compareResult()).isEqualTo(CompareSchemasVisitor.Result.SAME);
+ assertThat(cache.exists(tableIdentifier)).isTrue();
+ }
+
+ @Test
+ void tableCreationFailsWithAlreadyExistsWhenTableCannotBeLoaded() {
+ // Like a view with the same name: the table exists for create but not for
load
Review Comment:
A real view reproduces this without either stub: with a view of the same
name on a plain `InMemoryCatalog`, `TableCreator.DEFAULT` throws
`AlreadyExistsException` and `loadTable` throws `NoSuchTableException`.
Consider creating the view with `buildView` instead of overriding `loadTable`
and the creator.
--
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]
