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-5966-1a584332af50eb21e715df91a108a3d7c81736cc in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2442e8f3ebd12353bd5c9cfa40873a1f980e7011 Author: Xiaozhen Liu <[email protected]> AuthorDate: Tue Jun 30 09:29:52 2026 -0700 feat(workflow-core): add operator output port cache key (#5966) ### What changes were proposed in this PR? Adds `CacheKeyUtil` (in workflow-core), which computes a deterministic cache key for an output port from the upstream sub-DAG that produces it. The sub-DAG is the port's operator plus its transitive upstream, obtained from a new `PhysicalPlan.getTransitiveUpstreamSubPlan(opId)`; `computeCacheKey` takes that sub-DAG. For a target output port, the key is built from: - the target output port's identity; - every operator in the sub-DAG and, for each: its physical and logical operator id, its execution init info (the concrete execution definition the engine runs), and the schema of each of its output ports; - the edges among those operators, including the specific ports each edge connects. These are written to a JSON document with sorted keys and a sorted operator and edge order, so the same upstream computation always serializes to the same bytes and hashes (SHA-256) to the same key. Any upstream change (a parameter edit, a rewiring, a schema change) produces a different key; changes elsewhere in the workflow do not. Matching is collision-safe: `isSameComputation` compares the hash first and, on a hash match, confirms with the full JSON, so a hash collision never reuses a result the port was not computed from. This is the first piece of the operator output port result cache. `CacheKeyUtil` is a pure function with no database or storage dependency, and nothing calls it yet, so it changes no existing behavior. The cache table (companion PR) and the service that uses both land separately. ### Any related issues, documentation, discussions? Closes #5968. Part of the storage foundation #5882 (umbrella #5881). Design discussion: #5880. ### How was this PR tested? New unit tests in workflow-core (built from `PhysicalPlan`s directly, no engine test helpers, so the logic is covered in its own module): `CacheKeyUtilSpec` covers a stable key, sensitivity to an upstream change, different ports, a source with no upstream, upstream-only scope, and the three `isSameComputation` cases; the new `PhysicalPlan.getTransitiveUpstreamSubPlan` is tested in `WorkflowCoreTypesSpec`. Run with `sbt "WorkflowCore/testOnly *CacheKeyUtilSpec *WorkflowCoreTypesSpec"` — all pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 (Claude Code) --- .../texera/amber/core/workflow/PhysicalPlan.scala | 16 ++ .../amber/core/workflow/cache/CacheKeyUtil.scala | 205 +++++++++++++++++++++ .../core/workflow/WorkflowCoreTypesSpec.scala | 63 +++++++ .../core/workflow/cache/CacheKeyUtilSpec.scala | 180 ++++++++++++++++++ 4 files changed, 464 insertions(+) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalPlan.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalPlan.scala index 5a2a2a61b2..7c92aca2b8 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalPlan.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalPlan.scala @@ -84,6 +84,22 @@ case class PhysicalPlan( PhysicalPlan(newOps, newLinks) } + /** + * Returns the sub-plan made up of the given operator and all of its transitive + * upstream operators (the connected component feeding it), with the links among them. + */ + def getTransitiveUpstreamSubPlan(physicalOpId: PhysicalOpIdentity): PhysicalPlan = { + @scala.annotation.tailrec + def upstreamClosure( + frontier: Set[PhysicalOpIdentity], + collected: Set[PhysicalOpIdentity] + ): Set[PhysicalOpIdentity] = { + val next = frontier.flatMap(getUpstreamPhysicalOpIds) -- collected + if (next.isEmpty) collected else upstreamClosure(next, collected ++ next) + } + getSubPlan(upstreamClosure(Set(physicalOpId), Set(physicalOpId))) + } + def getUpstreamPhysicalOpIds(physicalOpId: PhysicalOpIdentity): Set[PhysicalOpIdentity] = { dag.incomingEdgesOf(physicalOpId).asScala.map(e => dag.getEdgeSource(e)).toSet } diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtil.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtil.scala new file mode 100644 index 0000000000..d181f20138 --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtil.scala @@ -0,0 +1,205 @@ +/* + * 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. + */ + +package org.apache.texera.amber.core.workflow.cache + +import com.fasterxml.jackson.databind.SerializationFeature +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.core.executor.OpExecInitInfo +import org.apache.texera.amber.core.workflow.{ + GlobalPortIdentity, + PhysicalLink, + PhysicalOp, + PhysicalPlan +} +import org.apache.texera.amber.util.JSONUtils + +import java.nio.charset.StandardCharsets +import java.security.MessageDigest + +/** + * The cache key of an output port. + * + * @param json the JSON describing the upstream sub-DAG the key is computed from + * @param hash SHA-256 hash of `json`; cache lookups match on this hash and then + * confirm the match with `json` (see [[CacheKeyUtil.isSameComputation]]) + */ +case class StorageCacheKey(json: String, hash: String) + +/** + * Computes deterministic cache keys for an output port from its upstream sub-DAG. + * + * The sub-DAG is the port's operator together with its transitive upstream operators + * (see [[PhysicalPlan.getTransitiveUpstreamSubPlan]]); the caller passes it in. The JSON + * payload captures: + * - the target output port, + * - all operators in the sub-DAG (sorted), + * - their exec init info (proto string), + * - their output schemas (string form when available), + * - all edges between those operators (sorted). + * + * The payload is serialized with ordered keys and hashed with SHA-256. Identical sub-DAGs + * produce identical hashes; any change in structure or configuration changes the hash. Sort + * orders use the full port identity (id + internal flag) so the payload is stable regardless + * of Map/Set iteration order. + */ +object CacheKeyUtil { + + /** + * Compute the cache key of the given upstream `subDag` for the output port `target`. + * + * Contract: the caller provides `target` as an output port and `subDag` as the sub-DAG that + * produces it, that is `plan.getTransitiveUpstreamSubPlan(target.opId)`. This method does not + * re-validate that relationship; it hashes whatever sub-DAG it is handed. The payload uses + * sorted keys and is hashed with SHA-256. + */ + def computeCacheKey( + subDag: PhysicalPlan, + target: GlobalPortIdentity + ): StorageCacheKey = { + val payload = buildJSONPayload(subDag.operators, subDag.links, target) + val json = objectMapper.writeValueAsString(payload) + StorageCacheKey(json, sha256Hex(json)) + } + + /** + * Whether two cache keys identify the same upstream computation. + * + * The hash is compared first; on a hash match the full JSON is compared as well. This + * keeps the match safe against a hash collision: if two different computations ever + * produced the same hash, their JSON would still differ, so they are reported as different + * and a cached result is never reused for a port it was not computed from. + */ + def isSameComputation(a: StorageCacheKey, b: StorageCacheKey): Boolean = + a.hash == b.hash && a.json == b.json + + /** + * Build the JSON payload describing the sub-DAG: + * - target port + * - sorted nodes with exec info and schemas + * - sorted edges + */ + private def buildJSONPayload( + nodes: Set[PhysicalOp], + links: Set[PhysicalLink], + target: GlobalPortIdentity + ): ObjectNode = { + val root = objectMapper.createObjectNode() + // target.toString is used only as a stable discriminator inside the hash; it is + // NOT the GlobalPortIdentitySerde form stored in the operator_port_cache table. + root.put("targetPort", target.toString) + + val nodeArray = objectMapper.createArrayNode() + nodes.toList + .sortBy(_.id.toString) + .foreach(op => nodeArray.add(buildNode(op))) + root.set("nodes", nodeArray) + + val edgeArray = objectMapper.createArrayNode() + links.toList + .sortBy(link => + ( + link.fromOpId.toString, + link.fromPortId.id, + link.fromPortId.internal, + link.toOpId.toString, + link.toPortId.id, + link.toPortId.internal + ) + ) + .foreach(link => edgeArray.add(buildEdge(link))) + root.set("edges", edgeArray) + + root + } + + /** + * Serialize a physical operator into a deterministic JSON node. + * Captures IDs, exec init info, and output schemas. + */ + private def buildNode(op: PhysicalOp): ObjectNode = { + val node = objectMapper.createObjectNode() + node.put("physicalOpId", op.id.toString) + node.put("logicalOpId", op.id.logicalOpId.toString) + node.set("opExec", serializeOpExec(op.opExecInitInfo)) + + val schemaArray = objectMapper.createArrayNode() + op.outputPorts.toList + .sortBy(p => (p._1.id, p._1.internal)) + .foreach { + case (portId, (_, _, schemaEither)) => + val schemaNode = objectMapper.createObjectNode() + schemaNode.put("portId", portId.id) + schemaNode.put("internal", portId.internal) + schemaEither.toOption match { + case Some(schema) => + schemaNode.put("available", true) + schemaNode.put("schemaString", schema.toString) + case None => + schemaNode.put("available", false) + } + schemaArray.add(schemaNode) + } + node.set("outputSchemas", schemaArray) + + node + } + + /** + * Serialize a physical link into a deterministic JSON node. + */ + private def buildEdge(link: PhysicalLink): ObjectNode = { + val edge = objectMapper.createObjectNode() + edge.put("fromOpId", link.fromOpId.toString) + edge.put("fromPortId", link.fromPortId.id) + edge.put("fromInternal", link.fromPortId.internal) + edge.put("toOpId", link.toOpId.toString) + edge.put("toPortId", link.toPortId.id) + edge.put("toInternal", link.toPortId.internal) + edge + } + + // Derived from the shared JSONUtils.objectMapper so configuration stays in sync, with + // ORDER_MAP_ENTRIES_BY_KEYS added on top: the cache key must serialize map keys in a stable + // order so the same sub-DAG always produces the same hash. The shared mapper does not set + // this, so reusing it directly would risk non-deterministic keys. + private val objectMapper = + JSONUtils.objectMapper.copy().enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS) + + private def sha256Hex(value: String): String = { + val digest = MessageDigest.getInstance("SHA-256") + val bytes = digest.digest(value.getBytes(StandardCharsets.UTF_8)) + bytes.map("%02x".format(_)).mkString + } + + /** + * Serialize the operator's exec init info deterministically via its proto string. + * + * The cache key is computed at the physical-plan layer, where the operator Desc is not + * available on a `PhysicalOp`; only `opExecInitInfo` is. It is also the concrete execution + * definition the engine actually runs, so hashing it ties the key to exactly what produces a + * port's result, not to a higher-level description that could map to different executions. If + * what executes a port changes, the key changes, so a stale result is never reused. + */ + private def serializeOpExec(opExecInitInfo: OpExecInitInfo): ObjectNode = { + val n = objectMapper.createObjectNode() + n.put("protoString", opExecInitInfo.asMessage.toProtoString) + n + } +} diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/WorkflowCoreTypesSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/WorkflowCoreTypesSpec.scala index 2bf47489a3..7c876d26f6 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/WorkflowCoreTypesSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/WorkflowCoreTypesSpec.scala @@ -338,6 +338,69 @@ class WorkflowCoreTypesSpec extends AnyFlatSpec { assert(sub.links == Set(link("a", "b"))) } + "PhysicalPlan.getTransitiveUpstreamSubPlan" should "include the operator and all its transitive upstream, with the links between them" in { + val a = physicalOp("a") + val b = physicalOp("b") + val c = physicalOp("c") + val d = physicalOp("d") + // a -> b -> c -> d and a -> c; the upstream sub-DAG of c is {a, b, c} (d is downstream) + val plan = + PhysicalPlan( + Set(a, b, c, d), + Set(link("a", "b"), link("b", "c"), link("c", "d"), link("a", "c")) + ) + val sub = plan.getTransitiveUpstreamSubPlan(c.id) + assert(sub.operators.map(_.id) == Set(a.id, b.id, c.id)) + assert(sub.links == Set(link("a", "b"), link("b", "c"), link("a", "c"))) + } + + it should "return only the operator itself when it has no upstream" in { + val a = physicalOp("a") + val b = physicalOp("b") + val plan = PhysicalPlan(Set(a, b), Set(link("a", "b"))) + val sub = plan.getTransitiveUpstreamSubPlan(a.id) + assert(sub.operators.map(_.id) == Set(a.id)) + assert(sub.links.isEmpty) + } + + it should "include all branches when an operator has multiple inputs (join or union)" in { + val s1 = physicalOp("s1") + val s2 = physicalOp("s2") + val j = newPhysicalOp("j") + .withInputPorts(List(InputPort(PortIdentity(0)), InputPort(PortIdentity(1)))) + .withOutputPorts(List(OutputPort(PortIdentity(0)))) + val l1 = PhysicalLink(s1.id, PortIdentity(0), j.id, PortIdentity(0)) + val l2 = PhysicalLink(s2.id, PortIdentity(0), j.id, PortIdentity(1)) + val plan = PhysicalPlan(Set(s1, s2, j), Set(l1, l2)) + val sub = plan.getTransitiveUpstreamSubPlan(j.id) + assert(sub.operators.map(_.id) == Set(s1.id, s2.id, j.id)) + assert(sub.links == Set(l1, l2)) + } + + it should "follow only the target's upstream with multiple sources and sinks" in { + val a = physicalOp("a") + val b = physicalOp("b") + val c = newPhysicalOp("c") + .withInputPorts(List(InputPort(PortIdentity(0)), InputPort(PortIdentity(1)))) + .withOutputPorts(List(OutputPort(PortIdentity(0)))) + val d = physicalOp("d") + val e = physicalOp("e") + // sources a, b converge at c; c fans out to sinks d and e + val plan = PhysicalPlan( + Set(a, b, c, d, e), + Set( + PhysicalLink(a.id, PortIdentity(0), c.id, PortIdentity(0)), + PhysicalLink(b.id, PortIdentity(0), c.id, PortIdentity(1)), + link("c", "d"), + link("c", "e") + ) + ) + // the sub-DAG of sink d is {a, b, c, d}; the other sink e is excluded + val sub = plan.getTransitiveUpstreamSubPlan(d.id) + assert(sub.operators.map(_.id) == Set(a.id, b.id, c.id, d.id)) + assert(!sub.operators.map(_.id).contains(e.id)) + } + "PhysicalPlan.getPhysicalOpsOfLogicalOp" should "return every physical op sharing a logical id, in topological order" in { val a = physicalOp("a") val b = physicalOp("b") diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtilSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtilSpec.scala new file mode 100644 index 0000000000..e48229bf8f --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/cache/CacheKeyUtilSpec.scala @@ -0,0 +1,180 @@ +/* + * 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. + */ + +package org.apache.texera.amber.core.workflow.cache + +import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithCode} +import org.apache.texera.amber.core.virtualidentity.{ + ExecutionIdentity, + OperatorIdentity, + PhysicalOpIdentity, + WorkflowIdentity +} +import org.apache.texera.amber.core.workflow.{ + GlobalPortIdentity, + InputPort, + OutputPort, + PhysicalLink, + PhysicalOp, + PhysicalPlan, + PortIdentity +} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** + * Unit tests for [[CacheKeyUtil]]. These live in the workflow-core module, next + * to the code under test, and build a `PhysicalPlan` directly (no engine test + * helpers), so the cache-key logic is exercised and covered in its own module. + */ +class CacheKeyUtilSpec extends AnyFlatSpec with Matchers { + + private val workflowId = WorkflowIdentity(0L) + private val executionId = ExecutionIdentity(0L) + + private def opId(name: String): PhysicalOpIdentity = + PhysicalOpIdentity(OperatorIdentity(name), "main") + + private def physicalOp(name: String): PhysicalOp = + PhysicalOp + .oneToOnePhysicalOp(opId(name), workflowId, executionId, OpExecInitInfo.Empty) + .withInputPorts(List(InputPort(PortIdentity(0)))) + .withOutputPorts(List(OutputPort(PortIdentity(0)))) + + private def link(from: String, to: String): PhysicalLink = + PhysicalLink(opId(from), PortIdentity(0), opId(to), PortIdentity(0)) + + private def outputPort(name: String): GlobalPortIdentity = + GlobalPortIdentity(opId(name), PortIdentity(0), input = false) + + /** The cache key of the named operator's output port, computed from its upstream sub-DAG. */ + private def keyOf(plan: PhysicalPlan, name: String): StorageCacheKey = + CacheKeyUtil.computeCacheKey(plan.getTransitiveUpstreamSubPlan(opId(name)), outputPort(name)) + + /** a -> b -> c */ + private def linearPlan(): PhysicalPlan = + PhysicalPlan( + Set(physicalOp("a"), physicalOp("b"), physicalOp("c")), + Set(link("a", "b"), link("b", "c")) + ) + + "CacheKeyUtil.computeCacheKey" should "be stable for the same sub-DAG and port" in { + val plan = linearPlan() + val k1 = keyOf(plan, "b") + val k2 = keyOf(plan, "b") + k1.hash shouldEqual k2.hash + k1.json shouldEqual k2.json + k1.hash should have length 64 + } + + it should "produce different keys for different ports in the same plan" in { + val plan = linearPlan() + keyOf(plan, "b").hash should not equal keyOf(plan, "c").hash + } + + it should "include only upstream operators, not downstream ones" in { + val plan = linearPlan() + val key = keyOf(plan, "b") + key.json should include(opId("a").toString) + key.json should not include opId("c").toString + } + + it should "be stable for a source operator with no upstream" in { + val plan = linearPlan() + keyOf(plan, "a").hash shouldEqual keyOf(plan, "a").hash + } + + it should "change when the upstream structure changes" in { + val base = linearPlan() + // b gains a second upstream (x) on a new input port: the upstream sub-DAG of + // b's output port is now different, so the key must differ. + val b2 = physicalOp("b") + .withInputPorts(List(InputPort(PortIdentity(0)), InputPort(PortIdentity(1)))) + val widened = PhysicalPlan( + Set(physicalOp("a"), physicalOp("x"), b2, physicalOp("c")), + Set( + link("a", "b"), + PhysicalLink(opId("x"), PortIdentity(0), opId("b"), PortIdentity(1)), + link("b", "c") + ) + ) + keyOf(base, "b").hash should not equal keyOf(widened, "b").hash + } + + it should "change the cache key when an upstream operator's exec info changes" in { + def planWith(code: String): PhysicalPlan = + PhysicalPlan( + Set( + PhysicalOp + .oneToOnePhysicalOp(opId("a"), workflowId, executionId, OpExecWithCode(code, "python")) + .withInputPorts(List(InputPort(PortIdentity(0)))) + .withOutputPorts(List(OutputPort(PortIdentity(0)))) + ), + Set.empty + ) + keyOf(planWith("def f(t): return t"), "a").hash should not equal + keyOf(planWith("def f(t): return t + 1"), "a").hash + } + + it should "ignore output-port attributes that do not change the result (blocking, mode, reuseStorage)" in { + def planWith(out: OutputPort): PhysicalPlan = + PhysicalPlan( + Set( + PhysicalOp + .oneToOnePhysicalOp(opId("a"), workflowId, executionId, OpExecInitInfo.Empty) + .withInputPorts(List(InputPort(PortIdentity(0)))) + .withOutputPorts(List(out)) + ), + Set.empty + ) + // blocking (scheduling), reuseStorage (storage), and mode (how the stored result is + // presented to the UI) do not change the materialized data, so they are intentionally + // not part of the cache identity. + val plain = OutputPort(PortIdentity(0)) + val decorated = OutputPort( + PortIdentity(0), + blocking = true, + mode = OutputPort.OutputMode.SET_DELTA, + reuseStorage = true + ) + keyOf(planWith(plain), "a").hash shouldEqual keyOf(planWith(decorated), "a").hash + } + + "CacheKeyUtil.isSameComputation" should "treat two keys with the same hash and JSON as a match" in { + val plan = linearPlan() + CacheKeyUtil.isSameComputation(keyOf(plan, "b"), keyOf(plan, "b")) shouldBe true + } + + it should "reject a hash collision by comparing the full JSON" in { + // Two different computations that hash to the same value (fabricated, since a + // real SHA-256 collision is infeasible to construct): the JSON differs, so the + // match is rejected and a cached result is never reused for the wrong port. + CacheKeyUtil.isSameComputation( + StorageCacheKey("upstream-A", "same-hash"), + StorageCacheKey("upstream-B", "same-hash") + ) shouldBe false + } + + it should "reject keys with different hashes" in { + CacheKeyUtil.isSameComputation( + StorageCacheKey("upstream-A", "hash-1"), + StorageCacheKey("upstream-A", "hash-2") + ) shouldBe false + } +}
