You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/07/05 01:44:48 UTC
[PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
aglinxinyuan opened a new pull request, #6128:
URL: https://github.com/apache/texera/pull/6128
### What changes were proposed in this PR?
Adds unit-test coverage for the operator-descriptor base classes in `common/workflow-operator`, which Codecov reports as partly uncovered:
- **`LogicalOpSpec`** (new) — exercises the `LogicalOp` base directly: the `getPhysicalOp` `???` default (via a minimal subtype), the reflection-based `hashCode`/`equals`, `toString`, and the default `runtimeReconfiguration` `UnsupportedOperationException`.
- **`PythonOperatorDescriptorSpec`** (new) — exercises `PythonOperatorDescriptor.getPhysicalOp` through in-spec subtypes: the code-generation exception is embedded in the generated code rather than thrown, a source op gets `PreferController`, and a one-to-one op gets no location preference.
- **`PythonUDFSourceOpDescV2Spec`** (extended) — adds the blank virtual-environment-name rejection, the trimmed `pveName`, and the parallelizable/`suggestedWorkerNum` path when `workers > 1`.
No production code is changed; this is test-only.
### Any related issues, documentation, discussions?
Coverage gaps identified from the Codecov report for `apache/texera`.
### How was this PR tested?
New/extended ScalaTest specs, run locally:
```
sbt "WorkflowOperator/testOnly *LogicalOpSpec *PythonOperatorDescriptorSpec *PythonUDFSourceOpDescV2Spec"
```
All pass. `scalafmt` and `scalafixAll --check` are clean.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8 [1M context])
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6128:
URL: https://github.com/apache/texera/pull/6128#issuecomment-4884440280
<!-- texera-reviewer-suggestion -->
### Automated Reviewer Suggestions
Based on the `git blame` history of the changed files, we recommend the following reviewers:
- No candidates found from `git blame` history.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #6128:
URL: https://github.com/apache/texera/pull/6128#discussion_r3524177415
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.OpExecWithCode
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity, PreferController}
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PythonOperatorDescriptorSpec extends AnyFlatSpec with Matchers {
+
+ private val workflowId = WorkflowIdentity(1L)
+ private val executionId = ExecutionIdentity(1L)
+
+ private class OneToOnePyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = "print('hi')"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "t",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private class SourcePyOp extends PythonOperatorDescriptor {
+ override def asSource(): Boolean = true
+ override def parallelizable(): Boolean = true
+ override def generatePythonCode(): String = "yield None"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo("s", "d", OperatorGroupConstants.PYTHON_GROUP, List(), List(OutputPort()))
+ }
+
+ private class ThrowingPyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = throw new RuntimeException("boom")
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "x",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private def code(op: PythonOperatorDescriptor): String =
+ op.getPhysicalOp(workflowId, executionId).opExecInitInfo.asInstanceOf[OpExecWithCode].code
+
Review Comment:
Switched the `code` helper to a pattern match with a `fail(...)` fallback, so a shape change reports a clear message instead of a `ClassCastException` (c2800d0).
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.OpExecWithCode
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity, PreferController}
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PythonOperatorDescriptorSpec extends AnyFlatSpec with Matchers {
+
+ private val workflowId = WorkflowIdentity(1L)
+ private val executionId = ExecutionIdentity(1L)
+
+ private class OneToOnePyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = "print('hi')"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "t",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private class SourcePyOp extends PythonOperatorDescriptor {
+ override def asSource(): Boolean = true
+ override def parallelizable(): Boolean = true
+ override def generatePythonCode(): String = "yield None"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo("s", "d", OperatorGroupConstants.PYTHON_GROUP, List(), List(OutputPort()))
+ }
+
+ private class ThrowingPyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = throw new RuntimeException("boom")
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "x",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private def code(op: PythonOperatorDescriptor): String =
+ op.getPhysicalOp(workflowId, executionId).opExecInitInfo.asInstanceOf[OpExecWithCode].code
+
+ "PythonOperatorDescriptor.getPhysicalOp" should
+ "embed the generation exception in the code instead of throwing" in {
+ code(new ThrowingPyOp) shouldBe "#EXCEPTION DURING CODE GENERATION: boom"
+ }
+
+ it should "build a source PhysicalOp when asSource is true" in {
+ val physical = (new SourcePyOp).getPhysicalOp(workflowId, executionId)
+ physical.locationPreference shouldBe Some(PreferController)
+ physical.opExecInitInfo match {
+ case OpExecWithCode(c, language) =>
+ c shouldBe "yield None"
+ language shouldBe "python"
+ case other => fail(s"expected OpExecWithCode, got $other")
+ }
+ physical.isSourceOperator shouldBe true
+ }
+
+ it should "build a one-to-one PhysicalOp when asSource is false" in {
+ val physical = (new OneToOnePyOp).getPhysicalOp(workflowId, executionId)
+ physical.locationPreference shouldBe None
+ physical.opExecInitInfo.asInstanceOf[OpExecWithCode].code shouldBe "print('hi')"
Review Comment:
Same here — replaced the cast with a pattern match (c2800d0).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #6128:
URL: https://github.com/apache/texera/pull/6128#issuecomment-4884446978
## [Codecov](https://app.codecov.io/gh/apache/texera/pull/6128?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 59.11%. Comparing base ([`e0040dc`](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c6227c6`](https://app.codecov.io/gh/apache/texera/commit/c6227c65c393abf849ad9425b42084ad69fe749b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on main.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #6128 +/- ##
=========================================
Coverage 59.11% 59.11%
Complexity 3202 3202
=========================================
Files 1132 1132
Lines 43681 43681
Branches 4734 4734
=========================================
Hits 25821 25821
Misses 16430 16430
Partials 1430 1430
```
| [Flag](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | | *Carryforward flag |
|---|---|---|---|
| [access-control-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.00% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [agent-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `44.59% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [amber](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `63.12% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [computing-unit-managing-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [config-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `52.30% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [file-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `62.81% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [frontend](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `51.31% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [notebook-migration-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.57% <ø> (ø)` | | |
| [pyamber](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.19% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [workflow-compiling-service](https://app.codecov.io/gh/apache/texera/pull/6128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.14% <ø> (ø)` | | Carriedforward from [e0040dc](https://app.codecov.io/gh/apache/texera/commit/e0040dc322594aa52b7888689d1a1642e004071b?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
*This pull request uses carry forward flags. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/texera/pull/6128?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #6128:
URL: https://github.com/apache/texera/pull/6128#issuecomment-4884456119
<!-- texera-benchmarks-comment -->
## ⚠️ Benchmark changes need a look
🟢 0 better · 🔴 6 worse · ⚪ 9 noise (<±5%) · 0 without baseline
> Compared against main `3a12e62` benchmarked on **this same runner**, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.
[Dashboard](https://apache.github.io/texera/dev/bench/) · [Run](https://github.com/apache/texera/actions/runs/28726022769)
| | config | throughput | MB/s | latency | max Δ latest / 7d |
|---|---|---:|---:|---:|---:|
| 🔴 | bs=10 sw=10 sl=64 | 382 | 0.233 | 24,358/40,020/40,020 us | 🔴 +43.0% / 🔴 +164.3% |
| 🔴 | bs=100 sw=10 sl=64 | 799 | 0.488 | 123,648/139,289/139,289 us | 🔴 +5.2% / 🔴 +29.0% |
| ⚪ | bs=1000 sw=10 sl=64 | 927 | 0.566 | 1,079,514/1,103,068/1,103,068 us | ⚪ within ±5% / 🔴 +9.1% |
<details><summary>Baseline details</summary>
Latest main `3a12e62` from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---:|---:|---:|---:|---:|
| bs=10 sw=10 sl=64 | throughput | 382 tuples/sec | 443 tuples/sec | 776.36 tuples/sec | -13.8% | -50.8% |
| bs=10 sw=10 sl=64 | MB/s | 0.233 MB/s | 0.27 MB/s | 0.474 MB/s | -13.7% | -50.8% |
| bs=10 sw=10 sl=64 | p50 | 24,358 us | 23,287 us | 12,636 us | +4.6% | +92.8% |
| bs=10 sw=10 sl=64 | p95 | 40,020 us | 27,981 us | 15,143 us | +43.0% | +164.3% |
| bs=10 sw=10 sl=64 | p99 | 40,020 us | 27,981 us | 18,954 us | +43.0% | +111.1% |
| bs=100 sw=10 sl=64 | throughput | 799 tuples/sec | 841 tuples/sec | 985.33 tuples/sec | -5.0% | -18.9% |
| bs=100 sw=10 sl=64 | MB/s | 0.488 MB/s | 0.514 MB/s | 0.601 MB/s | -5.1% | -18.9% |
| bs=100 sw=10 sl=64 | p50 | 123,648 us | 117,545 us | 101,671 us | +5.2% | +21.6% |
| bs=100 sw=10 sl=64 | p95 | 139,289 us | 141,225 us | 107,939 us | -1.4% | +29.0% |
| bs=100 sw=10 sl=64 | p99 | 139,289 us | 141,225 us | 113,798 us | -1.4% | +22.4% |
| bs=1000 sw=10 sl=64 | throughput | 927 tuples/sec | 921 tuples/sec | 1,016 tuples/sec | +0.7% | -8.8% |
| bs=1000 sw=10 sl=64 | MB/s | 0.566 MB/s | 0.562 MB/s | 0.62 MB/s | +0.7% | -8.8% |
| bs=1000 sw=10 sl=64 | p50 | 1,079,514 us | 1,084,426 us | 989,693 us | -0.5% | +9.1% |
| bs=1000 sw=10 sl=64 | p95 | 1,103,068 us | 1,149,963 us | 1,028,327 us | -4.1% | +7.3% |
| bs=1000 sw=10 sl=64 | p99 | 1,103,068 us | 1,149,963 us | 1,059,969 us | -4.1% | +4.1% |
</details>
<details><summary>Raw CSV</summary>
```csv
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,523.15,200,128000,382,0.233,24358.42,40020.35,40020.35
1,100,10,64,20,2502.52,2000,1280000,799,0.488,123648.29,139288.90,139288.90
2,1000,10,64,20,21586.04,20000,12800000,927,0.566,1079513.75,1103067.94,1103067.94
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #6128:
URL: https://github.com/apache/texera/pull/6128#discussion_r3524137753
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.OpExecWithCode
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity, PreferController}
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PythonOperatorDescriptorSpec extends AnyFlatSpec with Matchers {
+
+ private val workflowId = WorkflowIdentity(1L)
+ private val executionId = ExecutionIdentity(1L)
+
+ private class OneToOnePyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = "print('hi')"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "t",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private class SourcePyOp extends PythonOperatorDescriptor {
+ override def asSource(): Boolean = true
+ override def parallelizable(): Boolean = true
+ override def generatePythonCode(): String = "yield None"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo("s", "d", OperatorGroupConstants.PYTHON_GROUP, List(), List(OutputPort()))
+ }
+
+ private class ThrowingPyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = throw new RuntimeException("boom")
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "x",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private def code(op: PythonOperatorDescriptor): String =
+ op.getPhysicalOp(workflowId, executionId).opExecInitInfo.asInstanceOf[OpExecWithCode].code
+
Review Comment:
`code(...)` uses `asInstanceOf[OpExecWithCode]`, so a mismatch in `opExecInitInfo` would fail with a `ClassCastException` and a poor error message. Pattern-match and `fail(...)` instead to make the test diagnostics clearer.
##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/PythonOperatorDescriptorSpec.scala:
##########
@@ -0,0 +1,98 @@
+/*
+ * 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.OpExecWithCode
+import org.apache.texera.amber.core.tuple.Schema
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity, PreferController}
+import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class PythonOperatorDescriptorSpec extends AnyFlatSpec with Matchers {
+
+ private val workflowId = WorkflowIdentity(1L)
+ private val executionId = ExecutionIdentity(1L)
+
+ private class OneToOnePyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = "print('hi')"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "t",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private class SourcePyOp extends PythonOperatorDescriptor {
+ override def asSource(): Boolean = true
+ override def parallelizable(): Boolean = true
+ override def generatePythonCode(): String = "yield None"
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo("s", "d", OperatorGroupConstants.PYTHON_GROUP, List(), List(OutputPort()))
+ }
+
+ private class ThrowingPyOp extends PythonOperatorDescriptor {
+ override def generatePythonCode(): String = throw new RuntimeException("boom")
+ override def getOutputSchemas(in: Map[PortIdentity, Schema]): Map[PortIdentity, Schema] =
+ Map.empty
+ override def operatorInfo: OperatorInfo =
+ OperatorInfo(
+ "x",
+ "d",
+ OperatorGroupConstants.PYTHON_GROUP,
+ List(InputPort()),
+ List(OutputPort())
+ )
+ }
+
+ private def code(op: PythonOperatorDescriptor): String =
+ op.getPhysicalOp(workflowId, executionId).opExecInitInfo.asInstanceOf[OpExecWithCode].code
+
+ "PythonOperatorDescriptor.getPhysicalOp" should
+ "embed the generation exception in the code instead of throwing" in {
+ code(new ThrowingPyOp) shouldBe "#EXCEPTION DURING CODE GENERATION: boom"
+ }
+
+ it should "build a source PhysicalOp when asSource is true" in {
+ val physical = (new SourcePyOp).getPhysicalOp(workflowId, executionId)
+ physical.locationPreference shouldBe Some(PreferController)
+ physical.opExecInitInfo match {
+ case OpExecWithCode(c, language) =>
+ c shouldBe "yield None"
+ language shouldBe "python"
+ case other => fail(s"expected OpExecWithCode, got $other")
+ }
+ physical.isSourceOperator shouldBe true
+ }
+
+ it should "build a one-to-one PhysicalOp when asSource is false" in {
+ val physical = (new OneToOnePyOp).getPhysicalOp(workflowId, executionId)
+ physical.locationPreference shouldBe None
+ physical.opExecInitInfo.asInstanceOf[OpExecWithCode].code shouldBe "print('hi')"
Review Comment:
Avoid `asInstanceOf[OpExecWithCode]` here as well; a small pattern match produces a clearer failure if `opExecInitInfo` changes shape.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] test(workflow-operator): add unit tests for LogicalOp and Python operator descriptor bases [texera]
Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #6128:
URL: https://github.com/apache/texera/pull/6128
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org