You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/05 08:04:07 UTC
(texera) 01/01: refactor(engine): remove dead CacheSourceOpExec and OpExecSource plumbing (#6126)
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-6126-44ec60d25e63c337424bea8be569ac19b4df3d92
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 95684ebc5a6d642e0680c14003cf77bb01c102d2
Author: Xinyuan Lin <xi...@uci.edu>
AuthorDate: Sun Jul 5 01:03:46 2026 -0700
refactor(engine): remove dead CacheSourceOpExec and OpExecSource plumbing (#6126)
### What changes were proposed in this PR?
Remove the dead `CacheSourceOpExec` executor and its `OpExecSource`
plumbing.
Cache source operators were replaced by input-port materialization
reader threads in #3425 (the symmetric follow-up to sink-operator
removal in #3312). The scheduler no longer inserts a cache-source read
op for a materialized link — it tags the downstream input port with the
URI directly:
```
Before (pre-#3425): upstream OutputPort(URI) ──► [CacheSourceOpExec] ──► downstream InputPort
After (#3425+): upstream OutputPort(URI) ─────────────────────────► downstream InputPort(URI)
```
Since then `CacheSourceOpExec` has been unreachable:
`SpecialPhysicalOpFactory.newSourcePhysicalOp` — the only producer of an
`OpExecSource` init-info, which is what constructed `CacheSourceOpExec`
in `DataProcessorRPCHandlerInitializer.setupExecutor` — has no
production caller (its only references were its own spec).
Removed:
| Item | Location |
| --- | --- |
| `CacheSourceOpExec` |
`common/workflow-operator/.../source/cache/CacheSourceOpExec.scala` |
| `SpecialPhysicalOpFactory` (+ spec) |
`common/workflow-operator/.../operator/SpecialPhysicalOpFactory.scala` |
| `OpExecSource` match arm + now-unused imports |
`amber/.../worker/DataProcessorRPCHandlerInitializer.scala` |
| `OpExecSource` message + `opExecSource = 3` oneof field |
`common/workflow-core/.../protobuf/.../executor.proto` |
The `OpExecInitInfo` oneof is an internal controller↔worker wire format
(built and consumed within the same deployment;
`preserve_unknown_fields: false`), so dropping the field is safe. The
now-unused `virtualidentity.proto` import is dropped.
### Any related issues, documentation, discussions?
Closes #6125. The dead code was discovered during review of the
test-coverage PR #6098.
**Depends on #6132** (multi-region e2e integration test). Per review
feedback, the multi-region path — which is exactly what the removed
cache sources used to serve — gets end-to-end coverage **first** in
#6132; this removal then rides on top, and its CI re-runs that test to
prove the removal doesn't regress the path. Merge #6132 before this PR.
### How was this PR tested?
No behavioral change — pure dead-code removal.
- `sbt "WorkflowOperator/compile" "WorkflowExecutionService/compile"` —
green (proto regenerates; the `OpExecInitInfo` match remains exhaustive
without the `OpExecSource` case)
- `sbt "WorkflowExecutionService/scalafmtCheck"
"WorkflowExecutionService/scalafixAll --check"` — clean
- The multi-region e2e test in #6132 exercises the materialization
read/write path end-to-end; once that merges, this branch is rebased on
it.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
---
.../DataProcessorRPCHandlerInitializer.scala | 6 --
.../org/apache/texera/amber/core/executor.proto | 9 +-
.../amber/operator/SpecialPhysicalOpFactory.scala | 67 --------------
.../operator/source/cache/CacheSourceOpExec.scala | 41 ---------
.../operator/SpecialPhysicalOpFactorySpec.scala | 100 ---------------------
5 files changed, 1 insertion(+), 222 deletions(-)
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala
index 6b0c62ac3f..f932b311fb 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorRPCHandlerInitializer.scala
@@ -23,7 +23,6 @@ import com.twitter.util.Future
import org.apache.texera.amber.core.executor.{
ExecFactory,
OpExecInitInfo,
- OpExecSource,
OpExecWithClassName,
OpExecWithCode
}
@@ -39,9 +38,6 @@ import org.apache.texera.amber.engine.architecture.rpc.workerservice.WorkerServi
import org.apache.texera.amber.engine.architecture.worker.promisehandlers._
import org.apache.texera.amber.engine.common.AmberLogging
import org.apache.texera.amber.engine.common.rpc.AsyncRPCHandlerInitializer
-import org.apache.texera.amber.operator.source.cache.CacheSourceOpExec
-
-import java.net.URI
class DataProcessorRPCHandlerInitializer(val dp: DataProcessor)
extends AsyncRPCHandlerInitializer(dp.asyncRPCClient, dp.asyncRPCServer)
@@ -89,8 +85,6 @@ class DataProcessorRPCHandlerInitializer(val dp: DataProcessor)
ExecFactory.newExecFromJavaClassName(className, descString, workerIdx, workerCount)
case OpExecWithCode(code, _) =>
ExecFactory.newExecFromJavaCode(code)
- case OpExecSource(storageUri, _) =>
- new CacheSourceOpExec(URI.create(storageUri))
case OpExecInitInfo.Empty =>
throw new IllegalArgumentException("Empty executor initialization info")
}
diff --git a/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/executor.proto b/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/executor.proto
index 20504c8c1c..aab34d7191 100644
--- a/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/executor.proto
+++ b/common/workflow-core/src/main/protobuf/org/apache/texera/amber/core/executor.proto
@@ -19,7 +19,6 @@ syntax = "proto3";
package org.apache.texera.amber.core;
-import "org/apache/texera/amber/core/virtualidentity.proto";
import "scalapb/scalapb.proto";
option (scalapb.options) = {
@@ -39,15 +38,9 @@ message OpExecWithClassName {
string descString = 2;
}
-message OpExecSource {
- string storageKey = 1;
- WorkflowIdentity workflowIdentity = 2 [(scalapb.field).no_box = true];
-}
-
message OpExecInitInfo {
oneof sealed_value {
OpExecWithClassName opExecWithClassName = 1;
- OpExecWithCode opExecWithCode = 2;
- OpExecSource opExecSource = 3;
+ OpExecWithCode opExecWithCode = 2;
}
}
\ No newline at end of file
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactory.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactory.scala
deleted file mode 100644
index 07d0d7e5d3..0000000000
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactory.scala
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.operator
-
-import org.apache.texera.amber.core.executor.OpExecSource
-import org.apache.texera.amber.core.storage.VFSURIFactory
-import org.apache.texera.amber.core.tuple.Schema
-import org.apache.texera.amber.core.virtualidentity.{
- ExecutionIdentity,
- PhysicalOpIdentity,
- WorkflowIdentity
-}
-import org.apache.texera.amber.core.workflow._
-
-import java.net.URI
-
-object SpecialPhysicalOpFactory {
- def newSourcePhysicalOp(
- workflowIdentity: WorkflowIdentity,
- executionIdentity: ExecutionIdentity,
- uri: URI,
- downstreamOperator: PhysicalOpIdentity,
- downstreamPort: PortIdentity,
- schema: Schema
- ): PhysicalOp = {
-
- val (_, _, globalPortIdOption, _) = VFSURIFactory.decodeURI(uri)
- val globalPortId = globalPortIdOption.get
- val outputPort = OutputPort()
- PhysicalOp
- .sourcePhysicalOp(
- PhysicalOpIdentity(
- globalPortId.opId.logicalOpId,
- s"${globalPortId.opId.layerName}_source_${globalPortId.portId.id}_${downstreamOperator.logicalOpId.id
- .replace('-', '_')}_${downstreamPort.id}"
- ),
- workflowIdentity,
- executionIdentity,
- OpExecSource(uri.toString, workflowIdentity)
- )
- .withInputPorts(List.empty)
- .withOutputPorts(List(outputPort))
- .withPropagateSchema(
- SchemaPropagationFunc(_ => Map(outputPort.id -> schema))
- )
- .propagateSchema()
-
- }
-
-}
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/cache/CacheSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/cache/CacheSourceOpExec.scala
deleted file mode 100644
index b1f76489eb..0000000000
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/cache/CacheSourceOpExec.scala
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.operator.source.cache
-
-import com.typesafe.scalalogging.LazyLogging
-import org.apache.texera.amber.core.executor.SourceOperatorExecutor
-import org.apache.texera.amber.core.storage.model.VirtualDocument
-import org.apache.texera.amber.core.storage.{DocumentFactory, VFSResourceType, VFSURIFactory}
-import org.apache.texera.amber.core.tuple.{Tuple, TupleLike}
-
-import java.net.URI
-
-class CacheSourceOpExec(storageUri: URI) extends SourceOperatorExecutor with LazyLogging {
- val (_, _, _, resourceType) = VFSURIFactory.decodeURI(storageUri)
- if (resourceType != VFSResourceType.RESULT) {
- throw new RuntimeException("The storage URI must point to a result storage")
- }
-
- private val storage =
- DocumentFactory.openDocument(storageUri)._1.asInstanceOf[VirtualDocument[Tuple]]
-
- override def produceTuple(): Iterator[TupleLike] = storage.get()
-
-}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala
deleted file mode 100644
index e5ceb37967..0000000000
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/SpecialPhysicalOpFactorySpec.scala
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.operator
-
-import org.apache.texera.amber.core.storage.VFSURIFactory
-import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
-import org.apache.texera.amber.core.virtualidentity.{
- ExecutionIdentity,
- OperatorIdentity,
- PhysicalOpIdentity,
- WorkflowIdentity
-}
-import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PortIdentity, PreferCoordinator}
-import org.scalatest.flatspec.AnyFlatSpec
-import org.scalatest.matchers.should.Matchers
-
-class SpecialPhysicalOpFactorySpec extends AnyFlatSpec with Matchers {
-
- private val workflowId = WorkflowIdentity(42L)
- private val executionId = ExecutionIdentity(7L)
- private val schema = Schema().add(new Attribute("col", AttributeType.STRING))
-
- private def portUri(opName: String, layer: String, portId: Int) = {
- val gpid = GlobalPortIdentity(
- PhysicalOpIdentity(OperatorIdentity(opName), layer),
- PortIdentity(portId),
- input = false
- )
- VFSURIFactory.resultURI(VFSURIFactory.createPortBaseURI(workflowId, executionId, gpid))
- }
-
- "SpecialPhysicalOpFactory.newSourcePhysicalOp" should
- "derive a source op that wires the decoded port identity, ports, and schema" in {
- val uri = portUri("srcOp", "layerA", 3)
- val downstream = PhysicalOpIdentity(OperatorIdentity("down-stream"), "main")
- val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
- workflowId,
- executionId,
- uri,
- downstream,
- PortIdentity(5),
- schema
- )
- op.id.logicalOpId shouldBe OperatorIdentity("srcOp")
- // layerName: "${layerName}_source_${portId.id}_${downstreamLogicalId}_${downstreamPort.id}"
- // with '-' in the downstream logical id replaced by '_'
- op.id.layerName shouldBe "layerA_source_3_down_stream_5"
- op.workflowId shouldBe workflowId
- op.executionId shouldBe executionId
- op.isSourceOperator shouldBe true
- op.inputPorts shouldBe empty
- op.outputPorts.keySet shouldBe Set(PortIdentity(0))
- op.locationPreference shouldBe Some(PreferCoordinator)
- op.parallelizable shouldBe false
- op.outputPorts(PortIdentity(0))._3 shouldBe Right(schema)
- }
-
- it should "replace every dash in the downstream operator id when building the layer name" in {
- val op = SpecialPhysicalOpFactory.newSourcePhysicalOp(
- workflowId,
- executionId,
- portUri("srcOp", "layerA", 3),
- PhysicalOpIdentity(OperatorIdentity("a-b-c"), "main"),
- PortIdentity(9),
- schema
- )
- op.id.layerName shouldBe "layerA_source_3_a_b_c_9"
- }
-
- it should "fail when the URI carries no global port identity" in {
- val noPortUri = VFSURIFactory.createRuntimeStatisticsURI(workflowId, executionId)
- intercept[NoSuchElementException] {
- SpecialPhysicalOpFactory.newSourcePhysicalOp(
- workflowId,
- executionId,
- noPortUri,
- PhysicalOpIdentity(OperatorIdentity("down"), "main"),
- PortIdentity(0),
- schema
- )
- }
- }
-}