This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5967-1a584332af50eb21e715df91a108a3d7c81736cc in repository https://gitbox.apache.org/repos/asf/texera.git
commit c2fe6c99513c2ee8c7bbfaa8d82eb75289c6d071 Author: Xiaozhen Liu <[email protected]> AuthorDate: Tue Jun 30 09:26:04 2026 -0700 feat(dao): add operator_port_cache table (#5967) ### What changes were proposed in this PR? Adds the `operator_port_cache` table that records a materialized output port result so it can be reused across executions. It is keyed by `(workflow_id, global_port_id, cache_key)` and stores the JSON the cache key was computed from, the result location, an optional tuple count and source execution id, and a database-managed `updated_at`. The foreign key to `workflow(wid)` is `ON DELETE CASCADE`. The stored JSON (`cache_key_json`) lets a lookup confirm a hash match by comparing the full JSON, so a hash collision never reuses the wrong result. The change is additive: a new table in `sql/texera_ddl.sql` (fresh installs) plus a Liquibase migration `sql/updates/26.sql` registered in `sql/changelog.xml` (existing deployments). No code reads or writes the table yet; the cache read/write logic and its tests land with the cache service that uses it, following the convention of testing a table through its consumer (as `feedback` is tested via `FeedbackResourceSpec`). ### Any related issues, documentation, discussions? Closes #5969. Part of the storage foundation #5882 (umbrella #5881). Design discussion: #5880. ### How was this PR tested? Verified the schema directly against Postgres: the migration applies cleanly, the columns and primary key `(workflow_id, global_port_id, cache_key)` are correct, the foreign key's delete rule is `CASCADE`, the schema file and the migration define identical columns/keys, and `changelog.xml` is well-formed and registers `26.sql`. The generated jOOQ classes build from the table. The table's runtime behavior is exercised by the cache service tests in the follow-up PR. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 (Claude Code) --- sql/changelog.xml | 5 +++++ sql/texera_ddl.sql | 27 +++++++++++++++++++++++++++ sql/updates/26.sql | 55 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+) diff --git a/sql/changelog.xml b/sql/changelog.xml index e216caf3d0..0aebde4ddb 100644 --- a/sql/changelog.xml +++ b/sql/changelog.xml @@ -38,6 +38,11 @@ <sqlFile path="sql/updates/25.sql"/> </changeSet> + <!-- Add operator_port_cache table --> + <changeSet id="26" author="Xiao-zhen-Liu"> + <sqlFile path="sql/updates/26.sql"/> + </changeSet> + <!-- example changeSet <changeSet id="1" author="author"> <sqlFile path="sql/updates/1.sql"/> diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql index 9728829851..65e03879c0 100644 --- a/sql/texera_ddl.sql +++ b/sql/texera_ddl.sql @@ -48,6 +48,7 @@ SET search_path TO texera_db, public; -- ============================================ DROP TABLE IF EXISTS operator_executions CASCADE; DROP TABLE IF EXISTS operator_port_executions CASCADE; +DROP TABLE IF EXISTS operator_port_cache CASCADE; DROP TABLE IF EXISTS workflow_user_access CASCADE; DROP TABLE IF EXISTS workflow_of_user CASCADE; DROP TABLE IF EXISTS user_config CASCADE; @@ -371,6 +372,32 @@ CREATE TABLE operator_port_executions FOREIGN KEY (workflow_execution_id) REFERENCES workflow_executions(eid) ON DELETE CASCADE ); +-- operator_port_cache +-- Caches a materialized output port result so it can be reused across executions. +-- A row is identified by (workflow_id, global_port_id, cache_key_hash), where +-- cache_key_hash is a SHA-256 hash of the upstream sub-DAG that produces the port (its +-- operators, their parameters and exec info, schemas, and wiring). cache_key_hash is the +-- lookup key; cache_key_json is the JSON the hash was computed from, kept so a hash match +-- can be confirmed against the full content (collision safety). A different upstream +-- computation (for example an operator parameter or version change) produces a different +-- cache_key_hash and therefore a new row, so existing entries are never overwritten: each +-- row is the result of one specific computation of one port. tuple_count is the result's +-- row count, kept so the coordinator can report a reused region's output stats without a +-- second query to the Iceberg catalog. +CREATE TABLE operator_port_cache +( + workflow_id INT NOT NULL, + global_port_id VARCHAR(200) NOT NULL, + cache_key_hash CHAR(64) NOT NULL, + cache_key_json TEXT NOT NULL, + storage_uri TEXT NOT NULL, + tuple_count BIGINT, + source_execution_id BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workflow_id, global_port_id, cache_key_hash), + FOREIGN KEY (workflow_id) REFERENCES workflow(wid) ON DELETE CASCADE +); + -- workflow_user_likes CREATE TABLE IF NOT EXISTS workflow_user_likes ( diff --git a/sql/updates/26.sql b/sql/updates/26.sql new file mode 100644 index 0000000000..495a8bf5f0 --- /dev/null +++ b/sql/updates/26.sql @@ -0,0 +1,55 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- ============================================ +-- 1. Connect to the texera_db database +-- ============================================ +\c texera_db + +SET search_path TO texera_db; + +-- ============================================ +-- 2. Add the operator_port_cache table +-- ============================================ +BEGIN; + +-- Caches a materialized output port result so it can be reused across executions. +-- A row is identified by (workflow_id, global_port_id, cache_key_hash), where +-- cache_key_hash is a SHA-256 hash of the upstream sub-DAG that produces the port (its +-- operators, their parameters and exec info, schemas, and wiring). cache_key_hash is the +-- lookup key; cache_key_json is the JSON the hash was computed from, kept so a hash match +-- can be confirmed against the full content (collision safety). A different upstream +-- computation (for example an operator parameter or version change) produces a different +-- cache_key_hash and therefore a new row, so existing entries are never overwritten: each +-- row is the result of one specific computation of one port. tuple_count is the result's +-- row count, kept so the coordinator can report a reused region's output stats without a +-- second query to the Iceberg catalog. +CREATE TABLE IF NOT EXISTS operator_port_cache +( + workflow_id INT NOT NULL, + global_port_id VARCHAR(200) NOT NULL, + cache_key_hash CHAR(64) NOT NULL, + cache_key_json TEXT NOT NULL, + storage_uri TEXT NOT NULL, + tuple_count BIGINT, + source_execution_id BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workflow_id, global_port_id, cache_key_hash), + FOREIGN KEY (workflow_id) REFERENCES workflow(wid) ON DELETE CASCADE +); + +COMMIT;
