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 2442e8f3eb feat(workflow-core): add operator output port cache key
(#5966)
2442e8f3eb is described below
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
+ }
+}
