Yicong-Huang commented on code in PR #5441: URL: https://github.com/apache/texera/pull/5441#discussion_r3369740844
########## amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala: ########## @@ -0,0 +1,256 @@ +/* + * 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.workflow + +import org.apache.commons.vfs2.FileNotFoundException +import org.apache.texera.amber.core.virtualidentity.OperatorIdentity +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.web.model.websocket.request.LogicalPlanPojo +import org.scalatest.flatspec.AnyFlatSpec + +import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters.IteratorHasAsScala + +class LogicalPlanSpec extends AnyFlatSpec { + + // --------------------------------------------------------------------------- + // Fixture: a small operator chain `csv → keyword` (and richer variants below) + // --------------------------------------------------------------------------- + + private def csv(): CSVScanSourceOpDesc = TestOperators.headerlessSmallCsvScanOpDesc() + private def keyword(): org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc = + TestOperators.keywordSearchOpDesc("column-1", "Asia") Review Comment: for logical plan, I am expecting testing with a bit more complex plan. currently only two operator chain is too simple. Test with Join, Union, UDF with multiple input output ports, different source operators, etc. ########## amber/src/test/scala/org/apache/texera/workflow/LogicalPlanSpec.scala: ########## @@ -0,0 +1,256 @@ +/* + * 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.workflow + +import org.apache.commons.vfs2.FileNotFoundException +import org.apache.texera.amber.core.virtualidentity.OperatorIdentity +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.web.model.websocket.request.LogicalPlanPojo +import org.scalatest.flatspec.AnyFlatSpec + +import scala.collection.mutable.ArrayBuffer +import scala.jdk.CollectionConverters.IteratorHasAsScala + +class LogicalPlanSpec extends AnyFlatSpec { + + // --------------------------------------------------------------------------- + // Fixture: a small operator chain `csv → keyword` (and richer variants below) + // --------------------------------------------------------------------------- + + private def csv(): CSVScanSourceOpDesc = TestOperators.headerlessSmallCsvScanOpDesc() + private def keyword(): org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc = + TestOperators.keywordSearchOpDesc("column-1", "Asia") + + private def link( + from: OperatorIdentity, + to: OperatorIdentity, + fromPort: Int = 0, + toPort: Int = 0 + ): LogicalLink = + LogicalLink(from, PortIdentity(fromPort), to, PortIdentity(toPort)) + + // --------------------------------------------------------------------------- + // Construction + // --------------------------------------------------------------------------- + + "LogicalPlan" should "expose the operators and links it was constructed with" in { + val a = csv() + val b = keyword() + val l = link(a.operatorIdentifier, b.operatorIdentifier) + val plan = LogicalPlan(List(a, b), List(l)) + assert(plan.operators == List(a, b)) + assert(plan.links == List(l)) + } + + "LogicalPlan.apply(LogicalPlanPojo)" should + "lift the POJO's operators and links, ignoring opsToViewResult / opsToReuseResult" in { + val a = csv() + val b = keyword() + val l = link(a.operatorIdentifier, b.operatorIdentifier) + val pojo = LogicalPlanPojo( + operators = List(a, b), + links = List(l), + // `operatorId` is private — use the public OperatorIdentity wrapper. + opsToViewResult = List(b.operatorIdentifier.id), // intentionally non-empty + opsToReuseResult = List(a.operatorIdentifier.id) + ) + val plan = LogicalPlan(pojo) + assert(plan.operators == List(a, b)) + assert(plan.links == List(l)) + } + + // --------------------------------------------------------------------------- + // getTopologicalOpIds + // --------------------------------------------------------------------------- + + "LogicalPlan.getTopologicalOpIds" should "yield a topological order on a linear chain" in { + val a = csv() + val b = keyword() + val plan = LogicalPlan(List(a, b), List(link(a.operatorIdentifier, b.operatorIdentifier))) + val order = plan.getTopologicalOpIds.asScala.toList + assert(order == List(a.operatorIdentifier, b.operatorIdentifier)) + } + + it should "respect edge directionality across a fan-out shape (a → b, a → c)" in { + val a = csv() + val b = keyword() + val c = TestOperators.keywordSearchOpDesc("column-1", "Africa") + val plan = LogicalPlan( + List(a, b, c), + List( + link(a.operatorIdentifier, b.operatorIdentifier), + link(a.operatorIdentifier, c.operatorIdentifier) + ) + ) + val order = plan.getTopologicalOpIds.asScala.toList + assert(order.head == a.operatorIdentifier, s"expected a first, got $order") + assert(order.tail.toSet == Set(b.operatorIdentifier, c.operatorIdentifier)) + } + + // --------------------------------------------------------------------------- + // getOperator + // --------------------------------------------------------------------------- + + "LogicalPlan.getOperator" should "return the operator with the requested identifier" in { + val a = csv() + val b = keyword() + val plan = LogicalPlan(List(a, b), List.empty) + assert(plan.getOperator(a.operatorIdentifier) eq a) + assert(plan.getOperator(b.operatorIdentifier) eq b) + } + + it should "throw NoSuchElementException for an unknown operator id" in { + val a = csv() + val plan = LogicalPlan(List(a), List.empty) + intercept[NoSuchElementException] { + plan.getOperator(OperatorIdentity("not-in-plan")) + } + } + + // --------------------------------------------------------------------------- + // getTerminalOperatorIds + // --------------------------------------------------------------------------- + + "LogicalPlan.getTerminalOperatorIds" should "return the single sink in a linear chain" in { + val a = csv() + val b = keyword() + val plan = LogicalPlan(List(a, b), List(link(a.operatorIdentifier, b.operatorIdentifier))) + val sinks = plan.getTerminalOperatorIds + assert(sinks == List(b.operatorIdentifier)) + } + + it should "return every operator with out-degree 0 in a fan-out plan" in { + val a = csv() + val b = keyword() + val c = TestOperators.keywordSearchOpDesc("column-1", "Africa") + val plan = LogicalPlan( + List(a, b, c), + List( + link(a.operatorIdentifier, b.operatorIdentifier), + link(a.operatorIdentifier, c.operatorIdentifier) + ) + ) + assert(plan.getTerminalOperatorIds.toSet == Set(b.operatorIdentifier, c.operatorIdentifier)) + } + + it should "return every operator when there are no links" in { + // An isolated set of operators with no edges — every operator has + // out-degree 0 and is therefore terminal. + val a = csv() + val b = keyword() + val plan = LogicalPlan(List(a, b), List.empty) + assert(plan.getTerminalOperatorIds.toSet == Set(a.operatorIdentifier, b.operatorIdentifier)) + } + + it should "return an empty list for an empty plan" in { + val plan = LogicalPlan(List.empty, List.empty) + assert(plan.getTerminalOperatorIds.isEmpty) + } + + // --------------------------------------------------------------------------- + // getUpstreamLinks + // --------------------------------------------------------------------------- + + "LogicalPlan.getUpstreamLinks" should "return every link whose toOpId matches" in { + val a = csv() + val b = keyword() + val c = TestOperators.keywordSearchOpDesc("column-1", "Africa") + val ab = link(a.operatorIdentifier, b.operatorIdentifier) + val cb = link(c.operatorIdentifier, b.operatorIdentifier, fromPort = 0, toPort = 1) + val plan = LogicalPlan(List(a, b, c), List(ab, cb)) + val upstreams = plan.getUpstreamLinks(b.operatorIdentifier) + assert(upstreams.size == 2) + assert(upstreams.toSet == Set(ab, cb)) + } + + it should "preserve construction order when multiple links flow into the same target" in { + val a = csv() + val b = keyword() + val c = TestOperators.keywordSearchOpDesc("column-1", "Africa") + val ab = link(a.operatorIdentifier, b.operatorIdentifier) + val cb = link(c.operatorIdentifier, b.operatorIdentifier, fromPort = 0, toPort = 1) + val plan = LogicalPlan(List(a, b, c), List(ab, cb)) + assert(plan.getUpstreamLinks(b.operatorIdentifier) == List(ab, cb)) + } + + it should "return an empty list when nothing flows into the requested target" in { + val a = csv() + val plan = LogicalPlan(List(a), List.empty) + assert(plan.getUpstreamLinks(a.operatorIdentifier).isEmpty) + } + + // --------------------------------------------------------------------------- + // resolveScanSourceOpFileName + // --------------------------------------------------------------------------- + // + // A successful resolution of a real file path is environment-dependent + // (resolved through FileResolver, which can reach LakeFS / dataset + // service), so we exclusively pin the FAILURE behavior here — that + // matches the production contract worth pinning: + // 1. Some(errorList) — failures are appended to errorList by opId + // 2. None — the first failure rethrows Review Comment: is this a bug? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
