Xiao-zhen-Liu commented on code in PR #5900: URL: https://github.com/apache/texera/pull/5900#discussion_r3488202319
########## 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: This is the only sqlite-backed iceberg test in the repo — the other four (`test_iceberg_document.py`, the REST-catalog test, the utils-catalog test, `test_large_binary_manager.py`) use postgres or REST. #5682 moved this test onto postgres specifically to match production, and this PR reverts that and removes the comment that explained it. If running without external infra is the goal, that's reasonable — but it diverges from the convention. Please confirm it's intended, and re-add a note saying why this test uses sqlite. ########## 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: This round-trip test, and the Python-compatibility test below it, assert only the `content` column. With the format now four columns, nothing tests that `loop_counter` / `loop_start_id` / `loop_start_state_uri` survive a Python <-> Scala round-trip. Suggest extending one of these to set non-default values for the three columns and assert they come back. ########## 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: Design question, not a request: does the saved state need to carry the storage URI, or is `loop_start_id` enough for the controller to resolve where Loop End writes the next iteration? Raising it partly because it looks different from what we discussed offline. ########## 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: The four-column mapping is written by hand here, again in the network sender, and again in the two `OutputManager` methods. Adding a fifth column later means editing all of them, with no compiler help if one is missed. Worth building the column-to-value mapping once and reusing it. (Minor: the loop args are passed positionally at call sites, e.g. `save_state_to_storage_if_needed(state, 7)`, which reads as a magic number.) ########## 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: This test patches `State.from_tuple` and feeds the reader bare dicts (these rows), so it mostly checks that a dict returns what was put in it — it can't catch a wrong column name or a real schema mismatch. Separately, `loop_start_id` and `loop_start_state_uri` are never written then read through real storage with non-default values (the e2e only varies `loop_counter`). Suggest the e2e write non-default values for all three and assert they come back, which exercises the real Tuple/Schema/iceberg path instead of the mocked one. -- 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]
