feat(storage): add DocumentFactory.documentExists - #5085
Merged
Conversation
Enables "create only if absent" flows to test existence without catching exceptions from openDocument. Splits off from #4206. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a small storage helper to let callers check whether a VFS-backed document already exists at a URI, avoiding the current “probe by calling openDocument and catching exceptions” pattern. This fits into workflow-core’s storage abstractions around VFS/Iceberg-backed documents.
Changes:
- Added
DocumentFactory.documentExists(uri: URI): Booleanfor thevfsscheme. - Implemented existence probing via Iceberg namespace resolution +
IcebergUtil.loadTableMetadata. - Explicitly rejects unsupported URI schemes with
UnsupportedOperationException.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #5085 +/- ##
============================================
+ Coverage 43.12% 43.14% +0.01%
+ Complexity 2208 2207 -1
============================================
Files 1045 1045
Lines 40249 40260 +11
Branches 4252 4250 -2
============================================
+ Hits 17358 17369 +11
Misses 21817 21817
Partials 1074 1074
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
aglinxinyuan
marked this pull request as draft
May 16, 2026 00:51
- Extract resolveNamespace helper so createDocument, openDocument, and documentExists share one resourceType → namespace mapping. - Use catalog.tableExists via IcebergUtil instead of loadTableMetadata so transient catalog errors surface instead of becoming false negatives. - Tweak the unsupported-scheme message to mention "checking document existence" rather than "checking the document". - Add IcebergDocumentSpec cases for existing/fresh URIs and unsupported schemes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Drops the single-use IcebergUtil.tableExists wrapper and calls catalog.tableExists directly via TableIdentifier. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Mirrors the Scala documentExists helper: - Extract _resolve_namespace so create_document, open_document, and document_exists share one resource_type -> namespace mapping. - document_exists calls catalog.table_exists directly so transient catalog errors surface instead of becoming false negatives. - Add unit tests covering true/false catalog responses, unsupported resource type, and unsupported URI scheme. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Xinyuan Lin <xinyual3@uci.edu>
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
aglinxinyuan
marked this pull request as ready for review
May 16, 2026 01:36
aglinxinyuan
requested review from
Xiao-zhen-Liu and
mengw15
and removed request for
Xiao-zhen-Liu
May 16, 2026 01:36
mengw15
approved these changes
May 16, 2026
Contributor
|
Also could we make Codecov 100%? |
…ng asymmetry Address review feedback on #5085: - Document `@throws` cases on `DocumentFactory.documentExists` (Scala) and add a `Raises:` block on `DocumentFactory.document_exists` (Python), so callers gating a `createDocument` call know what to catch. - Note in Python `_resolve_namespace` that only RESULT/STATE are mapped because CONSOLE_MESSAGES and RUNTIME_STATISTICS are written exclusively from the Scala runtime; the asymmetry vs. the Scala helper is intentional. - Split the combined documentExists assert in `IcebergDocumentSpec` into two `it should` blocks ("true for a created URI" / "false for a never-created URI") so a regression names the failing scenario.
… of resolveNamespace The original `IcebergDocumentSpec` only exercised the RESULT and STATE branches of `DocumentFactory.resolveNamespace`. Codecov flagged CONSOLE_MESSAGES and RUNTIME_STATISTICS as missing patch coverage. Add `documentExists` cases for both URI types so all four mapped resource-type branches are now exercised. The defensive `case _ =>` branch in `resolveNamespace` remains unreachable from any well-formed VFS URI (`VFSURIFactory.decodeURI` already validates resource types before reaching `resolveNamespace`); exercising it would require either reflection-based test plumbing or restructuring the lookup. Keeping it as defensive code for future enum additions.
…reflection After covering CONSOLE_MESSAGES and RUNTIME_STATISTICS, the match in `resolveNamespace` was still flagged as partial by jacoco because the defensive `case _ =>` (unreachable from any well-formed VFS URI) was never taken. Reflect into the private method and pass `null` to exercise the wildcard branch, asserting it throws IllegalArgumentException as documented. Keeps the defensive code intact while bringing branch coverage on `resolveNamespace` to 100%.
aglinxinyuan
enabled auto-merge (squash)
May 16, 2026 04:31
yangzhang75
pushed a commit
to yangzhang75/texera
that referenced
this pull request
Jun 22, 2026
### What changes were proposed in this PR?
Adds a `documentExists`-style helper to `DocumentFactory` in both the
Scala and Python code paths, so callers can check whether an
iceberg-backed document already exists at a `vfs://` URI without
catching exceptions from `openDocument` / `open_document`.
- Scala: new `DocumentFactory.documentExists(uri: URI): Boolean`.
Resolves the `VFSResourceType` to its iceberg namespace, then probes the
catalog via
`IcebergCatalogInstance.getInstance().tableExists(TableIdentifier.of(namespace,
storageKey))`. Throws `UnsupportedOperationException` for non-`vfs` URI
schemes; `IllegalArgumentException` for unsupported resource types.
- Python: new `DocumentFactory.document_exists(uri: str) -> bool`. Same
shape: probes via `catalog.table_exists(f"{namespace}.{storage_key}")`;
raises `NotImplementedError` / `ValueError` symmetrically.
- Refactor: extracted a private `resolveNamespace` (Scala) and
`_resolve_namespace` (Python) so `createDocument`, `openDocument`, and
the new helper share one resource-type → namespace mapping in each
language.
- Why `Catalog.tableExists` rather than `loadTableMetadata`:
`loadTableMetadata` catches every exception and returns `None`, so a
transient catalog error would have surfaced as a false-negative "doesn't
exist" answer. `Catalog.tableExists` only returns `false` on actual
not-found, and lets unexpected errors propagate.
- The change in `open_document` from a hard-coded `"vfs"` literal to
`VFSURIFactory.VFS_FILE_URI_SCHEME` aligns the three methods on the same
scheme constant.
### Any related issues, documentation, discussions?
Closes: apache#5089
### How was this PR tested?
- `sbt "WorkflowCore/Test/compile"` — clean.
- `sbt "WorkflowCore/testOnly *IcebergDocumentSpec"` — 14/14 pass,
including two new cases asserting `documentExists` returns true after
`createDocument`, false on a fresh URI, and throws
`UnsupportedOperationException` for an unsupported scheme.
- `sbt "WorkflowCore/testOnly *IcebergUtilSpec"` — 13/13 pass (refactor
did not touch `IcebergUtil`).
- `pytest amber/src/test/python/core/storage/test_document_factory.py` —
11/11 pass, including four new cases covering `document_exists`
returning true/false based on `catalog.table_exists`, raising
`ValueError` on an unsupported resource type, and raising
`NotImplementedError` on an unsupported scheme.
- `ruff check` clean on `document_factory.py` and
`test_document_factory.py`.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7 in compliance with ASF.
---------
Signed-off-by: Xinyuan Lin <xinyual3@uci.edu>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Meng Wang <mengw15@uci.edu>
gupta-sahil01
pushed a commit
to gupta-sahil01/texera
that referenced
this pull request
Jul 18, 2026
> _Re-opened from my fork to satisfy the requirement that contributions come from a fork rather than a branch on the main repository (the prior PR [apache#4206](apache#4206) was on an `apache/texera` branch). The full review history — Copilot's and @Xiao-zhen-Liu's review threads and my replies — is preserved on apache#4206 for reference._ --- ### What changes were proposed in this PR? Adds two operators — **Loop Start** and **Loop End** — that let users write a for-loop over the rows of a pandas table inside a visual workflow. The user supplies four small Python snippets: | Field | Where | Example | |---|---|---| | `initialization` | Loop Start | `i = 0` | | `output` | Loop Start | `table.iloc[i]` — the row passed into the loop body each iteration | | `update` | Loop End | `i += 1` | | `condition` | Loop End | `i < len(table)` — keep looping while this is true | Operators between Loop Start and Loop End are the loop body and run once per iteration. While `condition` is true the runtime starts another iteration; when it turns false, downstream operators run on the accumulated output. #### How an iteration works ``` Upstream Table │ ▼ ┌──────────┐ loop variables: row i, ┌──────────┐ ┌─────────┐ │ Loop ├───counters, accumulators ─►│ loop ├──────────────►│ Loop │ │ Start │ (the loop's "state") │ body │ │ End │ └──────────┘ └──────────┘ └────┬────┘ ▲ │ │ (1) DCM: "schedule the Loop Start region again" │ │ (2) write the next iteration's state (i, accumulators, table) │ │ to the iceberg table that Loop Start reads its input from │ └──────────────────────────────────────────────────────────────────┘ when condition() == True ``` The Loop End → Loop Start arrow is **not a graph edge** — the region DAG stays acyclic. When an iteration ends, Loop End (1) sends a `jump_to_operator_region` DCM asking the coordinator to schedule the Loop Start region again, and (2) writes the next iteration's state into the iceberg state channel Loop Start reads from (the same cross-region channel from apache#4490, reused here as the back-edge). Details in the code comments on `MainLoop._jump_to_loop_start` and `_process_state_frame`. The write address is the same every iteration, so it is **config, not per-iteration data**: the scheduler resolves each Loop Start's input-port state URI and ships it to workers at setup (`InitializeExecutorRequest.loopStartStateUris`); a Loop End selects its entry by the `loop_start_id` stamped on the state it consumes. This is why the State format is only 3 columns (`content`/`loop_counter`/`loop_start_id`, from apache#5900) — see the proto field doc and `WorkflowExecutionManager.loopStartStateUris` for the rationale. Nested loops work with no extra machinery: a `loop_counter` on the state (+1 through an inner Loop Start, −1 through an inner Loop End, consumed at 0) routes each state to its owning loop — see `MainLoop._process_state_frame`. #### State serialization The loop state = the user's variables + the input `table`. Variables ride the State `content` column as JSON (raw `bytes` base64-encoded); the `table` is serialized as an **Apache Arrow IPC stream**, not pickle — the state is persisted to iceberg and read back by another worker, so `pickle.loads` would be a remote-code-execution surface. Arrow's nested-type slowness doesn't apply here (Texera tables are flat) and the channel is low-volume, so safety wins. See `table_to_ipc_bytes` in `core/models/table.py`. #### File map | Area | File | Purpose | |---|---|---| | Descriptors | `loop/{LoopOpDesc,LoopStartOpDesc,LoopEndOpDesc}.scala` | Code-gen the Python operator from the user expressions; Loop Start sets `isLoopStart`, Loop End sets output-port `reuseStorage` | | Runtime base | `core/models/operator.py` | `LoopStartOperator` / `LoopEndOperator` the generated code extends; guarded expression eval | | Physical plan | `PhysicalOp.scala` | Compile-time `isLoopStart` (loop-back resolution) + `requiresMaterializedExecution` (forces whole-plan MATERIALIZED) markers | | Loop-back resolution + delivery | `WorkflowExecutionManager.scala`, `controlcommands.proto`, `RegionExecutionManager.scala`, `initialize_executor_handler.py`, `context.py` | Derive `{Loop Start id → input-port state URI}` from the final schedule and ship it to workers at setup | | Worker runtime | `MainLoop` (`_process_state_frame`, `_jump_to_loop_start`) | `loop_counter` bookkeeping, nested pass-through, and the jump-DCM + state-write on Loop End completion | | Storage reuse | `OutputPort.reuseStorage` + `RegionExecutionManager`; `OutputManager.reset_output_storage` | A Loop End accumulates across its own iterations; an inner nested Loop End resets once per outer iteration | | Frontend | `LoopStart.png`, `LoopEnd.png` | Operator icons | ### Any related issues, documentation, discussions? Closes apache#4442. Builds on apache#4490 (cross-region state materialization) and apache#5085 (`DocumentFactory.documentExists`). Prerequisites split out of this PR per @Xiao-zhen-Liu's request, all merged: apache#5706 (worker-id helper), apache#5707 (state test harness), **apache#5900** (the 3-column State format, dormant on main until this PR activates it). ### How was this PR tested? - **Unit** — `test_main_loop.py` (loop-counter routing, nested pass-through, the full `_jump_to_loop_start` contract incl. DCM-before-write ordering and the fail-loud missing-URI case, and `complete()`'s error reporting for condition / loop-back-write failures); `test_loop_operators.py` (operator base classes, guarded eval, the generated-code shape via a base64 round-trip, reserved-`table` collision raising, a multi-iteration loop to completion); `test_output_manager.py` (`reset_output_storage` contract); `test_initialize_executor_handler.py` (proto field → `Context`); `LoopStartOpDescSpec` / `LoopEndOpDescSpec` (codegen, ports, JSON round-trip, flag pins). - **Integration** (`@IntegrationTest`, CI) — `LoopIntegrationSpec`: single loop (3 iterations) and nested 3×3 loop (9 inner iterations), asserted via materialized iceberg row counts. - `scalafmtCheckAll` + `scalafixAll --check` and Python `ruff format` clean; full Python unit suite passes. #### Manual workflows Input for both is a 3-row table from `TextInput("1\n2\n3")`; each condition is `i < len(table)`. | Workflow | Topology | Expected | |---|---|---| | [Loop.json](https://github.com/user-attachments/files/27985168/Loop.json) | `TextInput → LoopStart → LoopEnd` | 3 iterations, terminates. | | [Nested Loop.json](https://github.com/user-attachments/files/27985169/Nested.Loop.json) | `TextInput → OuterStart → InnerStart → InnerEnd → OuterEnd` | 3 outer × 3 inner = **9** inner iterations, terminates. | Basic Loop: <img width="1715" height="368" alt="loop" src="https://github.com/user-attachments/assets/4b9ea672-b5c9-4392-9ac3-764cc9cbb772" /> Nested Loop: <img width="1715" height="368" alt="nested" src="https://github.com/user-attachments/assets/240ea180-43ca-4815-aa6b-0ff82eb81d7b" /> ### Was this PR authored or co-authored using generative AI tooling? Co-authored with Claude (Opus 4.7, Opus 4.8) in compliance with ASF. --------- Signed-off-by: Xinyuan Lin <xinyual3@uci.edu> Co-authored-by: Xiaozhen Liu <xiaozl3@uci.edu> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Co-authored-by: Meng Wang <mengw15@uci.edu> Co-authored-by: probe <probe@x>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
Adds a
documentExists-style helper toDocumentFactoryin both the Scala and Python code paths, so callers can check whether an iceberg-backed document already exists at avfs://URI without catching exceptions fromopenDocument/open_document.DocumentFactory.documentExists(uri: URI): Boolean. Resolves theVFSResourceTypeto its iceberg namespace, then probes the catalog viaIcebergCatalogInstance.getInstance().tableExists(TableIdentifier.of(namespace, storageKey)). ThrowsUnsupportedOperationExceptionfor non-vfsURI schemes;IllegalArgumentExceptionfor unsupported resource types.DocumentFactory.document_exists(uri: str) -> bool. Same shape: probes viacatalog.table_exists(f"{namespace}.{storage_key}"); raisesNotImplementedError/ValueErrorsymmetrically.resolveNamespace(Scala) and_resolve_namespace(Python) socreateDocument,openDocument, and the new helper share one resource-type → namespace mapping in each language.Catalog.tableExistsrather thanloadTableMetadata:loadTableMetadatacatches every exception and returnsNone, so a transient catalog error would have surfaced as a false-negative "doesn't exist" answer.Catalog.tableExistsonly returnsfalseon actual not-found, and lets unexpected errors propagate.open_documentfrom a hard-coded"vfs"literal toVFSURIFactory.VFS_FILE_URI_SCHEMEaligns the three methods on the same scheme constant.Any related issues, documentation, discussions?
Closes: #5089
How was this PR tested?
sbt "WorkflowCore/Test/compile"— clean.sbt "WorkflowCore/testOnly *IcebergDocumentSpec"— 14/14 pass, including two new cases assertingdocumentExistsreturns true aftercreateDocument, false on a fresh URI, and throwsUnsupportedOperationExceptionfor an unsupported scheme.sbt "WorkflowCore/testOnly *IcebergUtilSpec"— 13/13 pass (refactor did not touchIcebergUtil).pytest amber/src/test/python/core/storage/test_document_factory.py— 11/11 pass, including four new cases coveringdocument_existsreturning true/false based oncatalog.table_exists, raisingValueErroron an unsupported resource type, and raisingNotImplementedErroron an unsupported scheme.ruff checkclean ondocument_factory.pyandtest_document_factory.py.Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7 in compliance with ASF.