damccorm commented on code in PR #41069:
URL: https://github.com/apache/beam/pull/41069#discussion_r4512602360
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java:
##########
@@ -345,12 +346,14 @@ private Table loadOrCreateTable(IcebergDestination
destination, Schema dataSchem
// Create namespace if it does not exist yet
if (!namespace.isEmpty() && catalog instanceof SupportsNamespaces) {
SupportsNamespaces supportsNamespaces = (SupportsNamespaces) catalog;
- if (!supportsNamespaces.namespaceExists(namespace)) {
- try {
- supportsNamespaces.createNamespace(namespace);
- LOG.info("Created new namespace '{}'.", namespace);
- } catch (AlreadyExistsException ignored) {
- // race condition: another worker already created this namespace
+ synchronized (NAMESPACE_LOCK) {
Review Comment:
Will this have significant performance implications?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java:
##########
@@ -345,12 +346,14 @@ private Table loadOrCreateTable(IcebergDestination
destination, Schema dataSchem
// Create namespace if it does not exist yet
if (!namespace.isEmpty() && catalog instanceof SupportsNamespaces) {
SupportsNamespaces supportsNamespaces = (SupportsNamespaces) catalog;
- if (!supportsNamespaces.namespaceExists(namespace)) {
- try {
- supportsNamespaces.createNamespace(namespace);
- LOG.info("Created new namespace '{}'.", namespace);
- } catch (AlreadyExistsException ignored) {
- // race condition: another worker already created this namespace
+ synchronized (NAMESPACE_LOCK) {
Review Comment:
It should not have a meaningful runtime impact because `loadOrCreateTable`
is only invoked as the cache-miss loader inside
`TableCache.getAndRefreshIfStale(catalogConfig, identifier, () ->
loadOrCreateTable(...))` (`RecordWriterManager.java:323`).
`TableCache.TABLES` is a process-wide Guava cache (`expireAfterAccess(1,
TimeUnit.HOURS)`) that already synchronizes per `(catalogConfig,
tableIdentifier)` on initial table load, and subsequent periodic refreshes
(`DEFAULT_REFRESH_INTERVAL = 2m`) call `table.refresh()` directly without
calling `loadOrCreateTable`. Thus `NAMESPACE_LOCK` is only entered once per
cold table-cache miss per unique table identifier on a worker (when multiple
threads simultaneously cold-load *different* tables in the same new namespace).
That said, we can make this even lighter-weight by using double-checked
locking (`if (!supportsNamespaces.namespaceExists(namespace)) { synchronized
(NAMESPACE_LOCK) { if (!supportsNamespaces.namespaceExists(namespace)) { ... }
} }`) so existing namespaces don't acquire the lock at all, or by also catching
`RuntimeException` with `Duplicate`/`AlreadyExists` from
`BigQueryMetastoreCatalog`. Happy to update to double-checked locking if that
sounds good!
--
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]
