(texera) branch main updated: feat(dao): add operator_port_cache table (#5967)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git


The following commit(s) were added to refs/heads/main by this push:
     new 104bcc40cb feat(dao): add operator_port_cache table (#5967)
104bcc40cb is described below

commit 104bcc40cbda04621b37bcc5fb6f3ca58cb99f46
Author: Xiaozhen Liu <xi...@uci.edu>
AuthorDate: Tue Jun 30 23:34:40 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;