comphead commented on code in PR #7363:
URL: https://github.com/apache/datafusion-comet/pull/7363#discussion_r4459280729
##########
native/core/src/execution/operators/iceberg_common.rs:
##########
@@ -149,11 +149,30 @@ impl FileIoCacheKey {
properties: sorted_properties(catalog_properties),
})
}
+
+ /// Whether `other` is a `FileIO` for the same table, access mode and
catalog built from
+ /// different properties, which this key's `FileIO` supersedes. A catalog
that re-vends
+ /// credentials, or a commit that moves the metadata file, gives a table
new properties or a new
+ /// reference path, so the table is compared by its root: the reference
path up to `/metadata/`.
+ fn supersedes(&self, other: &Self) -> bool {
+ self.access_mode == other.access_mode
+ && self.catalog_name == other.catalog_name
+ && table_root(&self.reference_path) ==
table_root(&other.reference_path)
+ && self != other
+ }
+}
+
+/// The table location of a metadata file path, or the path itself when it
isn't one.
+fn table_root(reference_path: &str) -> &str {
+ reference_path
+ .rfind("/metadata/")
+ .map_or(reference_path, |i| &reference_path[..i])
Review Comment:
Iceberg writes every metadata file of a table into one directory:
`write.metadata.path` when it is set, and `<location>/metadata` otherwise
(`BaseMetastoreTableOperations.metadataFileLocation`). Searching for the last
`/metadata/` gets both custom layouts wrong. With
`write.metadata.path=s3://b/meta/db/t1` there is no `/metadata/` segment, so
the root is the whole file path. Each commit then gets a new root, and the old
`FileIO` and its provider stay until the LRU bound evicts them, which is the
#7390 case again. With `write.metadata.path=s3://b/metadata/db/t1` the root is
`s3://b` for every table laid out that way. Two such tables read in one query
through one catalog then evict each other on every insert, and each task
rebuilds its `FileIO`.
Could the read key compare the metadata file's parent directory instead,
while the write key keeps comparing its data location as it is? The doc comment
above and the design doc would change to match. I checked the two paths against
a small model of `table_root`, not in Comet.
##########
spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialProvider.java:
##########
@@ -50,8 +50,10 @@ default void initialize(Map<String, String>
catalogProperties) {}
CometS3Credentials getCredentialsForPath(CometS3CredentialContext context)
throws Exception;
/**
- * Invoked from the dispatcher's best-effort JVM shutdown hook. Default
no-op suits stateless
- * providers; override to release HTTP clients, refresh executors, etc.
+ * Invoked once nothing reads through this instance any more, such as after
an Iceberg catalog
+ * re-vends credentials for a table, and from the dispatcher's best-effort
JVM shutdown hook for
+ * instances still in use then. Default no-op suits stateless providers;
override to release HTTP
+ * clients, refresh executors, etc.
*/
Review Comment:
Before this change, `close()` ran only from the shutdown hook. Now `release`
calls it inline from `ProviderHandle::drop`, so it runs on whichever thread
drops the last bridge. That can be a Tokio worker, because iceberg-rust keeps
`FileIO` clones in tasks it spawns on the runtime (for example the delete file
loader in `caching_delete_file_loader.rs`). It can also overlap `initialize` of
a new instance for the same key, which is the case the retry loop in
`ensureInitialized` handles. For the shipped provider,
`VendedCredentialsProvider.close()` calls `RefreshingAuthManager.close()`,
which waits up to a minute for a token refresh that is in flight.
Could this Javadoc, and the Lifecycle section of
`s3-credential-providers.md`, tell provider authors that `close()` can run
while the job is running, on a native thread, that it should not block, and
that it must not release state other instances share? If a blocking `close()`
has to be supported, handing it to a dispatcher-owned thread in `release` would
keep the wait off the runtime. I have not run this.
##########
spark/src/test/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcherTest.java:
##########
@@ -267,4 +272,105 @@ public void closeAllSwallowsProviderExceptions() {
// Both close() invocations ran even though one (the first to be visited)
threw.
assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
}
+
+ @Test
+ public void releasingTheLastReferenceClosesTheProvider() throws Exception {
+ long h1 = init();
+ long h2 = init();
+ assertEquals(h1, h2);
+
+ CometS3CredentialDispatcher.release(h1);
+ assertEquals(0, TestCometS3CredentialProvider.closeCount.get());
+ CometS3CredentialDispatcher.getCredentialsForPath(h2, "b", "k", READ);
+
+ CometS3CredentialDispatcher.release(h2);
+ assertEquals(1, TestCometS3CredentialProvider.closeCount.get());
+ assertEquals(0, CometS3CredentialDispatcher.registeredCount());
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () -> CometS3CredentialDispatcher.getCredentialsForPath(h2, "b",
"k", READ));
+ assertTrue(thrown.getMessage().contains("not initialized"));
+ }
+
+ @Test
+ public void aRetiredKeyGetsANewProvider() {
+ long retired = init();
+ CometS3CredentialDispatcher.release(retired);
+
+ long fresh = init();
+
+ assertNotEquals(retired, fresh);
+ assertEquals(2, TestCometS3CredentialProvider.initCount.get());
+ assertEquals(1, CometS3CredentialDispatcher.registeredCount());
+ }
+
+ /** An extra release, or one for a handle closed at shutdown, changes
nothing. */
+ @Test
+ public void unknownOrRepeatedReleasesAreIgnored() {
+ CometS3CredentialDispatcher.release(Long.MAX_VALUE);
+ long handle = init();
+ CometS3CredentialDispatcher.release(handle);
+ CometS3CredentialDispatcher.release(handle);
+
+ assertEquals(1, TestCometS3CredentialProvider.closeCount.get());
+ long other = init();
+ CometS3CredentialDispatcher.release(handle);
+ assertEquals(1, CometS3CredentialDispatcher.registeredCount());
+ CometS3CredentialDispatcher.release(other);
+ assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
+ }
+
+ /**
+ * A REST catalog that re-vends credentials for a table changes its FileIO
properties, so each
+ * snapshot registers its own provider. Once native code releases a
snapshot's handle, that
+ * provider is closed, and only the snapshots still in use stay registered.
+ */
+ @Test
+ public void supersededSnapshotsAreClosedOnceReleased() {
+ long[] handles = new long[3];
+ for (int i = 0; i < handles.length; i++) {
+ Map<String, String> props = new HashMap<>();
+ props.put("s3.session-token", "token-" + i);
+ handles[i] =
CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, DK, props);
+ }
+ assertEquals(3, CometS3CredentialDispatcher.registeredCount());
+
+ CometS3CredentialDispatcher.release(handles[0]);
+ CometS3CredentialDispatcher.release(handles[1]);
+
+ assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
+ assertEquals(1, CometS3CredentialDispatcher.registeredCount());
+ }
Review Comment:
This test registers three keys that differ only in their properties and
releases two of them. `distinctCatalogPropertiesIsolateInstances` already shows
that distinct properties get distinct instances, and
`releasingTheLastReferenceClosesTheProvider` shows that the last release closes
an instance. Which snapshot gets superseded is decided in the native
`FileIoCache`, which this test cannot see. Could it be replaced by the
`registeredCount()` check after two snapshots in `CometS3CredentialBridgeSuite`
that was asked for earlier?
--
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]
