aglinxinyuan commented on code in PR #5085:
URL: https://github.com/apache/texera/pull/5085#discussion_r3251716411
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala:
##########
@@ -103,6 +103,40 @@ object DocumentFactory {
}
}
+ /**
+ * Check whether a document exists at the given URI without opening it.
+ *
+ * Returns true iff the underlying storage already has an entry for this
+ * URI (e.g., an iceberg table at the resolved namespace + storage key).
+ * Useful for "create only if absent" flows that would otherwise have to
+ * call `openDocument` inside a try/catch to test existence.
+ */
+ def documentExists(uri: URI): Boolean = {
+ uri.getScheme match {
+ case VFS_FILE_URI_SCHEME =>
+ val (_, _, _, resourceType) = decodeURI(uri)
+ val storageKey = sanitizeURIPath(uri)
+
+ val namespace = resourceType match {
+ case RESULT => StorageConfig.icebergTableResultNamespace
+ case CONSOLE_MESSAGES =>
StorageConfig.icebergTableConsoleMessagesNamespace
+ case RUNTIME_STATISTICS =>
StorageConfig.icebergTableRuntimeStatisticsNamespace
+ case STATE => StorageConfig.icebergTableStateNamespace
+ case _ =>
+ throw new IllegalArgumentException(s"Resource type $resourceType
is not supported")
+ }
+
+ IcebergUtil
+ .loadTableMetadata(IcebergCatalogInstance.getInstance(), namespace,
storageKey)
+ .isDefined
+
Review Comment:
Switched to `IcebergUtil.tableExists`, which delegates to
`catalog.tableExists` and lets unexpected catalog errors propagate. Fixed in
`ceff249ee`.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala:
##########
@@ -103,6 +103,40 @@ object DocumentFactory {
}
}
+ /**
+ * Check whether a document exists at the given URI without opening it.
+ *
+ * Returns true iff the underlying storage already has an entry for this
+ * URI (e.g., an iceberg table at the resolved namespace + storage key).
+ * Useful for "create only if absent" flows that would otherwise have to
+ * call `openDocument` inside a try/catch to test existence.
+ */
+ def documentExists(uri: URI): Boolean = {
+ uri.getScheme match {
+ case VFS_FILE_URI_SCHEME =>
+ val (_, _, _, resourceType) = decodeURI(uri)
+ val storageKey = sanitizeURIPath(uri)
+
+ val namespace = resourceType match {
+ case RESULT => StorageConfig.icebergTableResultNamespace
+ case CONSOLE_MESSAGES =>
StorageConfig.icebergTableConsoleMessagesNamespace
+ case RUNTIME_STATISTICS =>
StorageConfig.icebergTableRuntimeStatisticsNamespace
+ case STATE => StorageConfig.icebergTableStateNamespace
+ case _ =>
+ throw new IllegalArgumentException(s"Resource type $resourceType
is not supported")
+ }
Review Comment:
Extracted a private `resolveNamespace(resourceType)` helper and routed
`createDocument`, `openDocument`, and `documentExists` through it so the
mapping lives in one place. Fixed in `ceff249ee`.
--
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]
