Fokko commented on code in PR #13345:
URL: https://github.com/apache/iceberg/pull/13345#discussion_r2162391125
##########
core/src/test/java/org/apache/iceberg/jdbc/TestJdbcTableConcurrency.java:
##########
@@ -149,4 +190,1015 @@ public synchronized void testConcurrentConnections()
throws InterruptedException
assertThat(executorService.awaitTermination(3,
TimeUnit.MINUTES)).as("Timeout").isTrue();
assertThat(Iterables.size(icebergTable.snapshots())).isEqualTo(7);
}
+
+ @Test
+ public synchronized void testInitializeWithSlowConcurrentConnections()
Review Comment:
Appreciate the tests here 👍
##########
core/src/main/java/org/apache/iceberg/jdbc/JdbcUtil.java:
##########
@@ -123,7 +123,7 @@ enum SchemaVersion {
+ JdbcTableOperations.METADATA_LOCATION_PROP
+ " = ?";
static final String V0_CREATE_CATALOG_SQL =
- "CREATE TABLE "
+ "CREATE TABLE IF NOT EXISTS "
Review Comment:
It might also fail because of permissions. It can be that a user has read
privileges, but does not have create table privileges.
##########
core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java:
##########
@@ -156,48 +157,62 @@ public void initialize(String name, Map<String, String>
properties) {
closeableGroup.setSuppressCloseFailure(true);
}
- private void initializeCatalogTables() {
- LOG.trace("Creating database tables (if missing) to store iceberg
catalog");
- try {
- connections.run(
- conn -> {
- DatabaseMetaData dbMeta = conn.getMetaData();
- ResultSet tableExists =
- dbMeta.getTables(
- null /* catalog name */,
- null /* schemaPattern */,
- JdbcUtil.CATALOG_TABLE_VIEW_NAME /* tableNamePattern */,
- null /* types */);
- if (tableExists.next()) {
- return true;
- }
-
- LOG.debug(
- "Creating table {} to store iceberg catalog tables",
- JdbcUtil.CATALOG_TABLE_VIEW_NAME);
- return
conn.prepareStatement(JdbcUtil.V0_CREATE_CATALOG_SQL).execute();
- });
-
- connections.run(
- conn -> {
- DatabaseMetaData dbMeta = conn.getMetaData();
- ResultSet tableExists =
- dbMeta.getTables(
- null /* catalog name */,
- null /* schemaPattern */,
- JdbcUtil.NAMESPACE_PROPERTIES_TABLE_NAME /*
tableNamePattern */,
- null /* types */);
-
- if (tableExists.next()) {
+ private void atomicCreateTable(String tableName, String sqlCommand, String
reason)
+ throws SQLException, InterruptedException {
+ connections.run(
+ conn -> {
+ DatabaseMetaData dbMeta = conn.getMetaData();
+
+ // check the existence of a table name
+ Predicate<String> tableTest =
+ name -> {
+ try {
+ ResultSet result =
+ dbMeta.getTables(
+ null /* catalog name */,
+ null /* schemaPattern */,
+ name /* tableNamePattern */,
+ null /* types */);
+ return result.next();
+ } catch (SQLException e) {
+ return false;
+ }
+ };
+
+ // some databases force table name to upper case -- check that last.
+ Predicate<String> tableExists =
+ name -> tableTest.test(name) ||
tableTest.test(name.toUpperCase(Locale.ROOT));
+
+ if (tableExists.test(tableName)) {
+ return true;
+ }
+
+ LOG.debug("Creating table {} {}", tableName, reason);
+ try {
+ conn.prepareStatement(sqlCommand).execute();
+ return true;
+ } catch (SQLException e) {
+ // see if table was created by another thread.
Review Comment:
```suggestion
// see if table was created by another thread or process.
```
--
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: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
