From 35f97a65a00732662343cf93b46b615541964ae8 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 13 Oct 2026 16:35:48 -0700 Subject: [PATCH 1/2] fix(workflow-operator): keep the rows Distinct's engine keeps for NaN and signed zero The export called drop_duplicates, which counts two NaN values as equal and 0.0 as equal to -0.0. The engine's LinkedHashSet compares rows with Scala's ==, under which NaN equals nothing, and hashes doubles by their bits, so it keeps all of them. Select rows by the key the set operators already use. Closes #8974 Co-Authored-By: Claude Opus 5.5 --- .../operator/distinct/DistinctOpDesc.scala | 11 ++- .../distinct/DistinctOpDescSpec.scala | 92 ++++++++++++++++++- 2 files changed, 94 insertions(+), 9 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala index 17b646bfa10..61683b4d7b5 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala @@ -22,7 +22,7 @@ package org.apache.texera.amber.operator.distinct import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{HashPartition, InputPort, OutputPort, PhysicalOp} -import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StandaloneHelpers} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} class DistinctOpDesc extends LogicalOp with StandaloneCodeGenerator { @@ -55,8 +55,11 @@ class DistinctOpDesc extends LogicalOp with StandaloneCodeGenerator { ) override def generateStandaloneCode(): String = { - // JVM op uses LinkedHashSet to preserve first-occurrence order; - // pandas drop_duplicates does the same by default. - "out1df = in1df.drop_duplicates(ignore_index=True)" + // The engine's LinkedHashSet keeps the first row of each key in input + // order, by the rows' own equality: NaN equals nothing, and 0.0 and -0.0 + // hash apart. drop_duplicates counts both pairs as equal. + "out1df = _texera_set_select(in1df, lambda key: True)" } + + override def standaloneHelpers(): Seq[String] = Seq(StandaloneHelpers.SetRowKeys) } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala index 15a9cf5d64e..bd222cb61f7 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala @@ -19,15 +19,27 @@ package org.apache.texera.amber.operator.distinct +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{HashPartition, SinglePartition, UnknownPartition} import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.operator.tags.IntegrationTest +import org.scalatest.Tag import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.nio.charset.StandardCharsets +import java.nio.file.Files +import java.util.concurrent.TimeUnit +import scala.io.Source +import scala.util.Try + class DistinctOpDescSpec extends AnyFlatSpec with Matchers { + private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName) + private val workflowId = WorkflowIdentity(1L) private val executionId = ExecutionIdentity(1L) @@ -107,11 +119,81 @@ class DistinctOpDescSpec extends AnyFlatSpec with Matchers { a.operatorIdentifier should not equal b.operatorIdentifier } - // The JVM operator keeps the first occurrence of a duplicate, which is what - // drop_duplicates does by default, so the emitted line says nothing about order. - "DistinctOpDesc.generateStandaloneCode" should "drop duplicates in place" in { - (new DistinctOpDesc).generateStandaloneCode() shouldBe - "out1df = in1df.drop_duplicates(ignore_index=True)" + // The engine's LinkedHashSet compares rows with Scala's ==, under which NaN + // equals nothing, and hashes them by each double's bits, which tell 0.0 from + // -0.0. A repeated finite value and a repeated null are the controls, and the + // rows are compared in order, since the first of each is the one kept. + "DistinctOpDesc.generateStandaloneCode" should + "keep the rows the engine keeps for NaN and signed zero" taggedAs NeedsPythonPackages in { + val python = resolvePython().getOrElse(cancel("No runnable python executable")) + if (!canImportPandas(python)) cancel(s"'$python' cannot import pandas") + + val values: Seq[java.lang.Double] = + Seq(1.0, Double.NaN, 0.0, -0.0, null, 1.0, Double.NaN, null, -0.0) + val schema = Schema().add("v", AttributeType.DOUBLE) + + val exec = new DistinctOpExec + exec.open() + values.foreach(v => exec.processTuple(Tuple(schema, Array[Any](v)), 0)) + val engine = exec + .onFinish(0) + .map(_.getFields.head match { + case null => "None" + case d: java.lang.Double if d.isNaN => "nan" + case d => d.toString + }) + .toList + exec.close() + + val desc = new DistinctOpDesc + val pyValues = + values.map(v => if (v == null) "None" else if (v.isNaN) "np.nan" else v.toString) + val driver = + s"""import pandas as pd, numpy as np + |vals = [${pyValues.mkString(", ")}] + |mask = np.array([v is None for v in vals]) + |in1df = pd.DataFrame({"v": pd.arrays.FloatingArray( + | np.array([0.0 if v is None else v for v in vals]), mask)}) + |${(desc.standaloneHelpers() :+ desc.generateStandaloneCode()).mkString("\n")} + |for v in out1df["v"]: + | print("None" if v is pd.NA else repr(float(v))) + |""".stripMargin + val script = Files.createTempFile("distinct-nan-zero-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start() + val out = Source.fromInputStream(process.getInputStream, "UTF-8").mkString + process.waitFor(120, TimeUnit.SECONDS) + + withClue(s"python said:\n$out\nscript:\n$driver") { + process.exitValue() shouldBe 0 + out.linesIterator.toList shouldBe engine + } + } + + private def resolvePython(): Option[String] = { + def fromConfig: Option[String] = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + + def runnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable) } + private def canImportPandas(python: String): Boolean = + Try( + new ProcessBuilder(python, "-c", "import pandas, numpy").redirectErrorStream(true).start() + ).toOption.exists { p => + if (!p.waitFor(60, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } } From f6ac981b079e745eb1ccb32a86aef1e197c6920b Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 21 Oct 2026 04:58:12 -0700 Subject: [PATCH 2/2] test(workflow-compiling-service): expect Distinct's set select in the served script Distinct now keeps the rows the engine's HashSet keeps through _texera_set_select rather than drop_duplicates, so the export endpoint's spec looks for that call. Co-Authored-By: Claude Opus 5.5 --- .../texera/service/resource/WorkflowToPythonResourceSpec.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala index e9bc1c96fef..0005fc1ce91 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala @@ -166,7 +166,7 @@ class WorkflowToPythonResourceSpec extends AnyFlatSpec with BeforeAndAfterAll { // Both operators of the chain, and the import every script carries: enough // to show the payload is the translated plan rather than an empty string. assert(code.contains("import pandas as pd")) - assert(code.contains("drop_duplicates")) + assert(code.contains("_texera_set_select")) assert(code.contains("head(5)")) }