kunwp1 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3776641942
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocument.scala:
##########
Review Comment:
Can you check if this line needs to be changed now to initialize it to None?
Your change adds a loadTableMetadata() in seekToUsableFile() so this line might
cause inefficiency.
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergDocumentSpec.scala:
##########
@@ -82,6 +84,29 @@ class IcebergDocumentSpec extends AnyFlatSpec with Matchers
with BeforeAndAfterA
new IcebergDocument[Tuple](tableNamespace, tableName, icebergSchema,
serde, deserde)
}
+ /**
+ * Delegates to the shared local catalog while counting `loadTable` calls
-- the
+ * discriminator for per-seek table re-resolution (#7290): a reader that
only
+ * refreshed a pinned Table would touch the catalog exactly once, at
iterator
+ * construction, no matter how long it polls.
+ */
+ private class CountingCatalog(delegate: Catalog) extends Catalog {
+ val loadTableCalls = new AtomicInteger()
+ override def name(): String = "counting"
+ override def loadTable(identifier: TableIdentifier): Table = {
+ loadTableCalls.incrementAndGet()
+ delegate.loadTable(identifier)
+ }
+ override def tableExists(identifier: TableIdentifier): Boolean =
+ delegate.tableExists(identifier)
+ override def listTables(namespace: Namespace):
java.util.List[TableIdentifier] =
+ delegate.listTables(namespace)
+ override def dropTable(identifier: TableIdentifier, purge: Boolean):
Boolean =
+ delegate.dropTable(identifier, purge)
+ override def renameTable(from: TableIdentifier, to: TableIdentifier): Unit
=
+ delegate.renameTable(from, to)
+ }
+
Review Comment:
This same code also appears in the other test file. Try to de-duplicate this.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,11 +136,35 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ getOrLoad(catalogs, cacheKey(name), () => createCatalog(name))
}
+ /**
+ * `Cache.get` wraps loader failures (`UncheckedExecutionException`,
`ExecutionException`,
+ * `ExecutionError`); unwrap them so `createCatalog` failures keep the
types they had
+ * before the cache existed. Package-private so the spec can pin the
unwrapping against
+ * an isolated cache with a throwing loader.
+ */
+ private[storage] def getOrLoad(
+ cache: Cache[String, Catalog],
+ key: String,
+ loader: () => Catalog
+ ): Catalog =
+ try {
+ // get(key, loader) locks per key, not globally: a cache miss's REST
config
+ // round trip no longer blocks lookups of other warehouses.
+ cache.get(
+ key,
+ new Callable[Catalog] {
+ override def call(): Catalog = loader()
+ }
+ )
+ } catch {
+ case e: UncheckedExecutionException => throw e.getCause
+ case e: ExecutionException => throw e.getCause
+ case e: ExecutionError => throw e.getCause
Review Comment:
Collapse these three identical clauses.
--
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]
