aglinxinyuan commented on code in PR #5900: URL: https://github.com/apache/texera/pull/5900#discussion_r3488229295
########## amber/src/test/python/core/architecture/packaging/test_state_materialization_e2e.py: ########## @@ -74,203 +68,198 @@ ) [email protected] -class TestStateMaterializationE2E: - @pytest.fixture(autouse=True, scope="class") - def _init_storage_config(self): - """Initialize StorageConfig + IcebergCatalogInstance for the real - postgres-backed catalog in the `amber-integration` CI job. - - Critical detail: the Scala integration tests that run earlier in - the same job connect to the iceberg catalog DB as user - `postgres/postgres` (the storage.conf default for - `STORAGE_ICEBERG_CATALOG_POSTGRES_USERNAME/PASSWORD`). pyiceberg - creates the catalog's `iceberg_tables` metadata table on first - use, owned by whoever wrote first — so it ends up owned by - `postgres`. We MUST connect as the same user, otherwise we hit - `permission denied for table iceberg_tables`. +# Module-level scratch dir for the sqlite catalog + iceberg warehouse. +# We don't initialize `StorageConfig` here: other test modules (e.g. +# test_iceberg_document.py) also call `StorageConfig.initialize` at +# import time, and the class rejects re-initialization with +# RuntimeError. Whichever module gets collected first wins; we adopt +# its namespaces below. +_WAREHOUSE_DIR = tempfile.mkdtemp(prefix="texera-state-e2e-warehouse-") - Why the reset: `test_iceberg_document.py` also calls - `StorageConfig.initialize` at module import time (with a - different `texera/password` user that works for it because no - Scala writes first in the `pyamber` job where it runs). pytest - imports every test module during collection, even ones whose - tests will be deselected by `-m integration`, so that - initialization happens here too. We force-reset the singletons - and re-init with the prod-correct credentials; safe because - test_iceberg_document's tests are deselected from this run. - All catalog + S3 settings read the same `STORAGE_*` env vars - the production code consumes (via storage.conf), so the test - matches whichever identity the Scala side uses in the same job - and stays aligned with the bucket / endpoint the workflow - provisions. Defaults mirror storage.conf so a local sbt run - without those vars exported still works. [email protected](scope="module", autouse=True) +def sqlite_iceberg_catalog(): + """Inject a sqlite-backed SqlCatalog so the test runs without external Review Comment: Confirmed intentional. Re-added a docstring note on the fixture explaining the divergence: the other iceberg tests use postgres/REST to mirror production, but this e2e deliberately uses a hermetic sqlite catalog so the writer→storage→reader join can run infra-free — the materialization logic it exercises is catalog-agnostic, so the sqlite backend drives the same code path. (74eef1d) _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## amber/src/test/python/core/storage/runnables/test_input_port_materialization_reader_runnable.py: ########## @@ -60,16 +60,27 @@ def runnable(self, me): return instance def test_state_rows_are_emitted_as_state_frames(self, runnable): - state_a = State({"loop_counter": 0}) - state_b = State({"loop_counter": 1}) + state_a = State({"i": 0}) + state_b = State({"i": 1}) - # The state document yields opaque tuples; from_tuple deserializes - # them. Patch from_tuple so we don't have to wire a real - # serialization. + # The state document yields opaque multi-column tuples. State.from_tuple + # (patched) deserializes the content column; the reader reads the + # loop-control columns directly off the row and carries them onto the + # emitted StateFrame envelope. + row_a = { Review Comment: Done. The e2e now writes non-default values for all three loop columns through the real Tuple/Schema/iceberg path and asserts they come back: `test_state_written_by_output_manager_is_replayed_by_reader` sets `loop_counter=7` / `loop_start_id="outer-loop"` / `loop_start_state_uri="vfs:///wf/outer-state"` and asserts each on the replayed `StateFrame`; `test_state_table_persists_across_writer_close` does the same against the raw iceberg row. Left the mocked reader unit test focused on the queue/partitioning wiring it actually exercises. (74eef1d) _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## common/workflow-core/src/test/scala/org/apache/texera/amber/core/state/StateSpec.scala: ########## @@ -98,21 +98,21 @@ class StateSpec extends AnyFlatSpec { it should "tuple-round-trip" in { Review Comment: Done. Extended the `tuple-round-trip` test to set non-default values (`5L`, `"outer-loop"`, `"vfs:///outer"`) and assert all three off the raw tuple via `getField` — `fromTuple` intentionally surfaces only `content`, so the loop columns are checked directly on the tuple. (74eef1d) _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## amber/src/main/python/core/models/state.py: ########## @@ -25,13 +25,41 @@ class State(dict): CONTENT = "content" - SCHEMA = Schema(raw_schema={CONTENT: "STRING"}) + # Loop-control bookkeeping owned by the worker runtime, NOT user state -- it + # never appears in the content JSON. In memory it rides on the StateFrame + # envelope; it is materialized/serialized as its own column (parallel to + # content) by to_tuple(...). from_tuple() returns the bare State; callers + # that need these values read the corresponding columns off the tuple. + LOOP_COUNTER = "loop_counter" + LOOP_START_ID = "loop_start_id" + LOOP_START_STATE_URI = "loop_start_state_uri" + SCHEMA = Schema( + raw_schema={ + CONTENT: "STRING", + LOOP_COUNTER: "LONG", + LOOP_START_ID: "STRING", + LOOP_START_STATE_URI: "STRING", + } + ) def to_json(self) -> str: return json.dumps(_to_json_value(self), separators=(",", ":")) - def to_tuple(self) -> Tuple: - return Tuple({State.CONTENT: self.to_json()}, schema=State.SCHEMA) + def to_tuple( + self, + loop_counter: int = 0, + loop_start_id: str = "", + loop_start_state_uri: str = "", + ) -> Tuple: + return Tuple( + { + State.CONTENT: self.to_json(), Review Comment: Done. Added `State.to_columns`, the single column-name → value mapping, and routed both `to_tuple` and the network sender's `StateFrame` branch through it, so adding a column later is a one-line change in one place. (`save_state_to_storage_if_needed` already builds its tuple via `to_tuple`, and `emit_state` produces a `StateFrame` whose serialization also flows through `to_columns` in the sender — so there's now a single source for the mapping.) Also switched the positional call sites to keyword args so the loop values aren't magic numbers. (74eef1d) _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## amber/src/main/python/core/models/state.py: ########## @@ -25,13 +25,41 @@ class State(dict): CONTENT = "content" - SCHEMA = Schema(raw_schema={CONTENT: "STRING"}) + # Loop-control bookkeeping owned by the worker runtime, NOT user state -- it + # never appears in the content JSON. In memory it rides on the StateFrame + # envelope; it is materialized/serialized as its own column (parallel to + # content) by to_tuple(...). from_tuple() returns the bare State; callers + # that need these values read the corresponding columns off the tuple. + LOOP_COUNTER = "loop_counter" + LOOP_START_ID = "loop_start_id" + LOOP_START_STATE_URI = "loop_start_state_uri" Review Comment: Good question — and you're right it differs from the offline sketch. As currently wired in the loop branch, the URI isn't derivable from `loop_start_id` alone: - `loop_start_id` is just LoopStart's *logical* op id — it's what `JumpToOperatorRegionRequest(OperatorIdentity(loop_start_id))` jumps to. - `loop_start_state_uri` is `VFSURIFactory.state_uri(readers[0].uri)` — the *physical* iceberg URI of LoopStart's **input port**, which also encodes the execution id, the physical layer name, and which input port/reader. LoopEnd writes the next iteration's state straight into that URI, so it needs the full path, not just the logical id. So the real choice is: (a) carry the resolved URI on the state — current approach, LoopEnd writes back with no controller round-trip; or (b) carry only `loop_start_id` and have the controller resolve LoopStart's input-port materialization URI during the jump and hand it back. (b) is the leaner payload but couples LoopEnd to a controller resolution step. Happy to go with (b) if that's what we landed on offline. This PR is only the dormant 4-column format — the consumer logic (and this decision) actually lands in the later loop PR, so flagging it here so we can settle it before then. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- 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]
