From f1d85fbbf6f300e51f21299a70bc86750edc1329 Mon Sep 17 00:00:00 2001
From: Xiaozhen Liu <xiaozl3@uci.edu>
Date: Sun, 28 Jun 2026 09:49:12 -0700
Subject: [PATCH 1/4] feat(dao): add operator_port_cache table

Adds the operator_port_cache table (texera_ddl.sql + Liquibase migration sql/updates/26.sql), keyed by (workflow_id, global_port_id, cache_key) with ON DELETE CASCADE to workflow. The cache read/write logic and its tests land with the cache service that uses it. Part of #5882.
---
 sql/changelog.xml  |  5 +++++
 sql/texera_ddl.sql | 19 +++++++++++++++++++
 sql/updates/26.sql | 47 ++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 71 insertions(+)
 create mode 100644 sql/updates/26.sql

diff --git a/sql/changelog.xml b/sql/changelog.xml
index e216caf3d00..0aebde4ddb2 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 97288298516..c5812dd184c 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,24 @@ 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 across executions, keyed by the cache
+-- key (a SHA-256 hash of the upstream sub-DAG). cache_key_json holds the JSON
+-- the cache key is computed from.
+CREATE TABLE operator_port_cache
+(
+    workflow_id         INT NOT NULL,
+    global_port_id      VARCHAR(200) NOT NULL,
+    cache_key           CHAR(64) NOT NULL,
+    cache_key_json      TEXT NOT NULL,
+    result_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),
+    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 00000000000..5e0c2e85f71
--- /dev/null
+++ b/sql/updates/26.sql
@@ -0,0 +1,47 @@
+-- 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 across executions, keyed by the cache
+-- key (a SHA-256 hash of the upstream sub-DAG). cache_key_json holds the JSON
+-- the cache key is computed from.
+CREATE TABLE IF NOT EXISTS operator_port_cache
+(
+    workflow_id         INT NOT NULL,
+    global_port_id      VARCHAR(200) NOT NULL,
+    cache_key           CHAR(64) NOT NULL,
+    cache_key_json      TEXT NOT NULL,
+    result_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),
+    FOREIGN KEY (workflow_id) REFERENCES workflow(wid) ON DELETE CASCADE
+);
+
+COMMIT;

From 8a6554e7ebeea134747927f91f913a9d3c7a3b3a Mon Sep 17 00:00:00 2001
From: Xiaozhen Liu <xiaozl3@uci.edu>
Date: Sun, 28 Jun 2026 23:46:22 -0700
Subject: [PATCH 2/4] feat(dao): rename operator_port_cache result_uri to
 storage_uri

Address review: result implies a direction, storage_uri is clearer. tuple_count is kept (immutable per row, populated at materialization, read by the coordinator alongside the cache lookup so cached-region stats need no extra storage round-trip).
---
 sql/texera_ddl.sql | 2 +-
 sql/updates/26.sql | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index c5812dd184c..43df9b8cab1 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -382,7 +382,7 @@ CREATE TABLE operator_port_cache
     global_port_id      VARCHAR(200) NOT NULL,
     cache_key           CHAR(64) NOT NULL,
     cache_key_json      TEXT NOT NULL,
-    result_uri          TEXT NOT NULL,
+    storage_uri         TEXT NOT NULL,
     tuple_count         BIGINT,
     source_execution_id BIGINT,
     updated_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
diff --git a/sql/updates/26.sql b/sql/updates/26.sql
index 5e0c2e85f71..228238de522 100644
--- a/sql/updates/26.sql
+++ b/sql/updates/26.sql
@@ -36,7 +36,7 @@ CREATE TABLE IF NOT EXISTS operator_port_cache
     global_port_id      VARCHAR(200) NOT NULL,
     cache_key           CHAR(64) NOT NULL,
     cache_key_json      TEXT NOT NULL,
-    result_uri          TEXT NOT NULL,
+    storage_uri         TEXT NOT NULL,
     tuple_count         BIGINT,
     source_execution_id BIGINT,
     updated_at          TIMESTAMPTZ NOT NULL DEFAULT now(),

From a89dbd424ef5c08538597d7d18eb1ce41f2a9274 Mon Sep 17 00:00:00 2001
From: Xiaozhen Liu <xiaozl3@uci.edu>
Date: Mon, 29 Jun 2026 22:50:50 -0700
Subject: [PATCH 3/4] docs(dao): document operator_port_cache columns and key
 semantics

Address review: spell out that cache_key is the hash/lookup key and cache_key_json is the JSON it was computed from (collision check); that a changed upstream computation (e.g. operator version) yields a new cache_key and a new row rather than overwriting; and why tuple_count is kept.
---
 sql/texera_ddl.sql | 14 +++++++++++---
 sql/updates/26.sql | 14 +++++++++++---
 2 files changed, 22 insertions(+), 6 deletions(-)

diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index 43df9b8cab1..d35e700af55 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -373,9 +373,17 @@ CREATE TABLE operator_port_executions
 );
 
 -- operator_port_cache
--- Caches a materialized output port result across executions, keyed by the cache
--- key (a SHA-256 hash of the upstream sub-DAG). cache_key_json holds the JSON
--- the cache key is computed from.
+-- 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), where cache_key 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 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 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,
diff --git a/sql/updates/26.sql b/sql/updates/26.sql
index 228238de522..7253aeb5f7f 100644
--- a/sql/updates/26.sql
+++ b/sql/updates/26.sql
@@ -27,9 +27,17 @@ SET search_path TO texera_db;
 -- ============================================
 BEGIN;
 
--- Caches a materialized output port result across executions, keyed by the cache
--- key (a SHA-256 hash of the upstream sub-DAG). cache_key_json holds the JSON
--- the cache key is computed from.
+-- 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), where cache_key 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 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 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,

From 8e71ebb830057866517137b2de273b6a27dafbc1 Mon Sep 17 00:00:00 2001
From: Xiaozhen Liu <xiaozl3@uci.edu>
Date: Tue, 30 Jun 2026 09:15:42 -0700
Subject: [PATCH 4/4] refactor(dao): rename operator_port_cache cache_key
 column to cache_key_hash

Address review (Carlos, Yicong): make the hash explicit. cache_key_hash is the SHA-256 hash / lookup key; cache_key_json stays as the JSON it was computed from.
---
 sql/texera_ddl.sql | 24 ++++++++++++------------
 sql/updates/26.sql | 24 ++++++++++++------------
 2 files changed, 24 insertions(+), 24 deletions(-)

diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index d35e700af55..65e03879c0a 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -374,27 +374,27 @@ CREATE TABLE operator_port_executions
 
 -- 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), where cache_key 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 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 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.
+-- 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           CHAR(64) 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),
+    PRIMARY KEY (workflow_id, global_port_id, cache_key_hash),
     FOREIGN KEY (workflow_id) REFERENCES workflow(wid) ON DELETE CASCADE
 );
 
diff --git a/sql/updates/26.sql b/sql/updates/26.sql
index 7253aeb5f7f..495a8bf5f09 100644
--- a/sql/updates/26.sql
+++ b/sql/updates/26.sql
@@ -28,27 +28,27 @@ SET search_path TO texera_db;
 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), where cache_key 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 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 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.
+-- 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           CHAR(64) 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),
+    PRIMARY KEY (workflow_id, global_port_id, cache_key_hash),
     FOREIGN KEY (workflow_id) REFERENCES workflow(wid) ON DELETE CASCADE
 );