Worker-side IOperatorExecutor Creation by shengquan-ni · Pull Request #1772 · apache/texera

shengquan-ni · 2022-12-13T07:24:14Z

This PR fixes #1005.

For native operators, this PR moves the creation of operators to the worker side by changing the parameter from IOperatorExecutor to a lambda function () => IOperatorExecutor so that only the lambda function is serialized.
For python operators, this PR makes PythonWorkflowWorker directly initialize the operator by sending InitializeOperatorLogic to the python side. Earlier this control message is sent by the controller before starting the workflow. The payload of this control message is retrieved from the IOperatorExecutor. Since now we are moving the creation to the worker side, this logic needs to be updated.
Eailer we also use IOperatorExecutor on the controller side to determine if a worker is a python worker, this PR changed this part by passing the python operator information from the logical plan so we do not rely on IOperatorExecutor anymore.

Yicong-Huang · 2022-12-13T19:37:59Z

@shengquan-ni I see many changes regarding initializing python workers, can you elaborate in the PR description?

Yicong-Huang

It is not very clear to me what is the new life cycle of an operator exec. My understanding is that an operator would be 1) initialized with a function/lambda/python code, 2) opened, 3) process tuples, 4) process end of stream (on finish), 5) closed. An operator may also be updated during execution (i.e., UpdateOperatorLogic for python, and potentially also for java).

Can you elaborate more in the PR description and code comments, to mention how are those life cycles be managed in the new design? For example, it is not clear when an operator is initialized with a function, how can an operator update its function?

Yicong-Huang · 2022-12-16T19:49:51Z

+  pythonProxyClient.enqueueCommand(
+    ControlInvocation(
+      IgnoreReply,
+      InitializeOperatorLogic(
+        operatorExecutor.asInstanceOf[PythonUDFOpExecV2].getCode,
+        allUpstreamLinkIds.toArray,
+        operatorExecutor.isInstanceOf[ISourceOperatorExecutor],
+        operatorExecutor.asInstanceOf[PythonUDFOpExecV2].getOutputSchema
+      )
+    ),
+    SELF
+  )


this change is a bit weird to me. During the construction of PythonWorkflowWorker, the proxy client and proxy server have not been constructed and the connections have not been established yet. Is it a good idea to Enqueue a command now? Can we do it after all the components are initialized?

I think it is a minor concern because this message will stay in the queue if the proxy is not initialized yet. But I can move it to preStart() call if you want.

I think it may be better to move it after the proxy layer gets initialized. I am not sure if preStart is the correct location either.

Yicong-Huang · 2022-12-16T19:58:46Z

    allUpstreamLinkIds: Set[LinkIdentity],
    supportFaultTolerance: Boolean
 ) extends WorkflowActor(actorId, parentNetworkCommunicationActorRef, supportFaultTolerance) {
+  lazy val operatorExecutor: IOperatorExecutor = opFn(workerIdx)


this could be a life cycle question that is fundamental to this PR. When exactly is the operatorExecutor initialized? Here you are using lazy val, it seems the operator exec instance will be implicitly initialized when it is first used. In the current implementation, the first usage is when the operator exec is opened. You are effectively pushing initializing right before opening.

The behavior could change as we change the implementation. Would it be better to explicitly initialize an operator instead of depending on implementation?

We are using a DI framework called macwire. It recommends using lazy val. But it's a minor thing, I can change it to val if needed.

Yicong-Huang · 2022-12-16T20:01:07Z

+      if (o.operatorInfo.operatorGroupName == OperatorGroupConstants.UDF_GROUP) {
+        // Currently, all UDFs are python operators.
+        // TODO: change this if we add UDFs for other languages.
+        pythonOperators.add(amberOperator)
+      }


I don't feel using the group name is a good way to determine if it is a python operator. We might easily move python operators into different groups. For example, we will introduce "native Python Operator", such as the RedditSource operator, which is written in Python but belongs to the source operator group, user will drag the operator and use it as all other native python operators.

So the reason for having this change is now we don't want to have any IOperatorExecutor on the controller side. Previously we check if IOperatorExecutor is a PythonUDFOpExecV2, but now we need another way of identifying python operators. I know this is not a good way to determine, so that's why I left a TODO there. We can refine it later.

we are having some "native python" executors, for example reddit search operator and lambda expression operator, they will have a java OpDesc but executed with a PythonOpExec. they are likely not to be put under the OperatorGroupConstants.UDF_GROUP group. So I think we must find a better way to determine them now.

Yicong-Huang · 2022-12-16T20:22:12Z


  def getLink(linkID: LinkIdentity): LinkStrategy = idToLink(linkID)

-  def getPythonWorkers: Iterable[ActorVirtualIdentity] =


I like the new filtering predicates you added. but can we keep this method? it is convenient if I want to send some control messages to all python workers.

As I said, the old one is not available anymore because we don't have the IOperatorExecutor on the controller side.

IOperatorExecutor is just one implementation for getPythonWorkers. I think it's better to keep the getPythonWorkers method but we can replace it with another implementation (but not through the operator group).

Yicong-Huang · 2022-12-16T20:24:23Z


 class Workflow(
    workflowId: WorkflowIdentity,
+    pythonOperators: mutable.Set[OpExecConfig],


I think this is a redundant information. operatorToOpExecConfig should already contain all OpExecConfigs. And whether it is python or not should be determined from operatorToOpExecConfig. I suggest we do not pass pythonOperators again.

By only looking at OpExecConfig, we cannot say whether an operator is a python operator or not. Previously we check the IOperatorExecutor but now we can't. So this change is necessary.

then isn't it better to add some information to OpExecConfig?

shengquan-ni · 2022-12-28T07:05:46Z

@Yicong-Huang the following diagram illustrates what's changed in this PR.

Yicong-Huang · 2023-01-05T21:46:22Z

@Yicong-Huang the following diagram illustrates what's changed in this PR.

Could you please edit the PR description to include this diagram? I think it is useful to readers.

shengquan-ni · 2023-01-31T22:51:52Z

The change in #1807 is also doing worker-side operator creation. Close this one.

update

2adf5b4

shengquan-ni self-assigned this Dec 13, 2022

shengquan-ni added fix engine labels Dec 13, 2022

shengquan-ni added 4 commits December 13, 2022 00:52

fix python worker creation

65b1d45

Merge branch 'master' into shengquan-operator-remote-creation

78e90cc

fix error

ebbf923

Update PythonWorkflowWorker.scala

87e2151

shengquan-ni requested review from Yicong-Huang and zuozhiw December 16, 2022 07:16

shengquan-ni added 2 commits December 15, 2022 23:30

better naming

473a0a5

Merge branch 'master' into shengquan-operator-remote-creation

f3030be

Yicong-Huang reviewed Dec 16, 2022

View reviewed changes

Update PythonUDFSourceOpDescV2.java

bd7e25f

shengquan-ni requested a review from Yicong-Huang December 29, 2022 06:21

Merge branch 'master' into shengquan-operator-remote-creation

eca32f1

shengquan-ni closed this Jan 31, 2023

aglinxinyuan deleted the shengquan-operator-remote-creation branch September 6, 2025 00:55


		def getLink(linkID: LinkIdentity): LinkStrategy = idToLink(linkID)

		def getPythonWorkers: Iterable[ActorVirtualIdentity] =

Uh oh!

Conversation

shengquan-ni commented Dec 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Yicong-Huang commented Dec 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Yicong-Huang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shengquan-ni Dec 28, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Yicong-Huang Dec 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Yicong-Huang Dec 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shengquan-ni Dec 28, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shengquan-ni commented Dec 28, 2022

Uh oh!

Yicong-Huang commented Jan 5, 2023

Uh oh!

shengquan-ni commented Jan 31, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

shengquan-ni commented Dec 13, 2022 •

edited

Loading

Yicong-Huang commented Dec 13, 2022 •

edited

Loading

shengquan-ni Dec 28, 2022 •

edited

Loading

Yicong-Huang Dec 16, 2022 •

edited

Loading

Yicong-Huang Dec 16, 2022 •

edited

Loading

shengquan-ni Dec 28, 2022 •

edited

Loading