From 2d6fb1aa88d82366914ceee5445e150d7caa85ba Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 9 Sep 2026 17:36:29 -0700 Subject: [PATCH 1/2] feat(operator): cut a numeric column into bins The same gap as the timestamp parts, on the other kind of column. Grouping by a continuous number puts almost every row in a group of its own, so counts per age bracket or per price band could not be asked for; a user wanting them had to write a Python UDF. Two cuts, because they answer different questions. Equal width divides the span into bins of one size; equal frequency cuts at the quantiles, so the bins hold about as many rows as each other. The second is why the operator reads the whole table before it emits: a quantile is not known until the last row has arrived. The bin reads as its own range, "(2.5, 5.0]", which is a label to group by rather than a number to do arithmetic on. Co-Authored-By: Claude Opus 5 (1M context) --- .../texera/amber/operator/LogicalOp.scala | 2 + .../amber/operator/binning/BinningMethod.java | 43 ++++++ .../operator/binning/BinningOpDesc.scala | 136 ++++++++++++++++++ .../operator/binning/BinningOpDescSpec.scala | 109 ++++++++++++++ .../src/assets/operator_images/Binning.png | Bin 0 -> 1532 bytes 5 files changed, 290 insertions(+) create mode 100644 common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java create mode 100644 common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala create mode 100644 common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala create mode 100644 frontend/src/assets/operator_images/Binning.png diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala index efa46144180..889cb1eca7e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala @@ -89,6 +89,7 @@ import org.apache.texera.amber.operator.source.sql.postgresql.PostgreSQLSourceOp import org.apache.texera.amber.operator.split.SplitOpDesc import org.apache.texera.amber.operator.substringSearch.SubstringSearchOpDesc import org.apache.texera.amber.operator.symmetricDifference.SymmetricDifferenceOpDesc +import org.apache.texera.amber.operator.binning.BinningOpDesc import org.apache.texera.amber.operator.typecasting.TypeCastingOpDesc import org.apache.texera.amber.operator.udf.java.JavaUDFOpDesc import org.apache.texera.amber.operator.udf.python._ @@ -220,6 +221,7 @@ trait StateTransferFunc new Type(value = classOf[PostgreSQLSourceOpDesc], name = "PostgreSQLSource"), new Type(value = classOf[AsterixDBSourceOpDesc], name = "AsterixDBSource"), new Type(value = classOf[TypeCastingOpDesc], name = "TypeCasting"), + new Type(value = classOf[BinningOpDesc], name = "Binning"), new Type(value = classOf[LimitOpDesc], name = "Limit"), new Type(value = classOf[SleepOpDesc], name = "Sleep"), new Type(value = classOf[LoopStartOpDesc], name = "LoopStart"), diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java new file mode 100644 index 00000000000..e0d9cbef61e --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningMethod.java @@ -0,0 +1,43 @@ +/* + * 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.binning; + +import com.fasterxml.jackson.annotation.JsonValue; + +/** How the range of a numeric column is cut into bins. */ +public enum BinningMethod { + + EQUAL_WIDTH("equal width"), + + EQUAL_FREQUENCY("equal frequency"); + + private final String name; + + BinningMethod(String name) { + this.name = name; + } + + // use the name string instead of enum string in JSON + @JsonValue + public String getName() { + return this.name; + } + +} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala new file mode 100644 index 00000000000..373e3300841 --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala @@ -0,0 +1,136 @@ +/* + * 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.binning + +import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} +import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{ + PythonTemplateBuilderStringContext, + pyStringLiteral +} + +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "attribute": { + "enum": ["integer", "long", "double"] + } + } +} +""") +class BinningOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerator { + + @JsonProperty(required = true) + @JsonSchemaTitle("Attribute") + @JsonPropertyDescription("numeric column to cut into bins") + @AutofillAttributeName + var attribute: EncodableString = "" + + @JsonProperty(required = true, defaultValue = "equal width") + @JsonSchemaTitle("Method") + @JsonPropertyDescription("how the bins are cut") + var method: BinningMethod = BinningMethod.EQUAL_WIDTH + + @JsonProperty(required = true, defaultValue = "4") + @JsonSchemaTitle("Number of bins") + @JsonPropertyDescription("how many bins to cut the column into") + @JsonSchemaInject(json = """{"minimum": 2, "maximum": 100}""") + var bins: Int = 4 + + override def operatorInfo: OperatorInfo = + OperatorInfo( + userFriendlyName = "Binning", + operatorDescription = + "Cut a numeric column into bins, so rows can be grouped by range rather than by value", + operatorGroupName = OperatorGroupConstants.CLEANING_GROUP, + inputPorts = List(InputPort()), + // Blocking: an equal-frequency cut is made at the column's quantiles, which + // are not known until the last row has arrived. + outputPorts = List(OutputPort(blocking = true)) + ) + + /** The bin a row fell in, named after the column it was cut from. */ + private def resultColumn: String = s"${attribute}_bin" + + override def getOutputSchemas( + inputSchemas: Map[PortIdentity, Schema] + ): Map[PortIdentity, Schema] = + Map( + operatorInfo.outputPorts.head.id -> + // The bin reads as its own range, "(2.5, 5.0]", rather than as a number: + // the point of binning is a label to group by, and a number would invite + // arithmetic on what is really a name. + inputSchemas.values.head.add(resultColumn, AttributeType.STRING) + ) + + /** `duplicates="drop"` because a quantile cut can put two edges in one place + * where the values repeat, and fewer bins beats raising. + */ + private def cutArgs: String = + method match { + case BinningMethod.EQUAL_WIDTH => s"bins=$bins" + case BinningMethod.EQUAL_FREQUENCY => s"""q=$bins, duplicates="drop"""" + } + + private def cutName: String = + method match { + case BinningMethod.EQUAL_WIDTH => "pd.cut" + case BinningMethod.EQUAL_FREQUENCY => "pd.qcut" + } + + /** An empty cell has no bin, and `astype(str)` would render its absence as the + * text "nan", so the hole is kept as one. + */ + private val labelSuffix: String = + """.astype("string").astype("object").where(lambda s: s.notna(), None)""" + + override def generatePythonCode(): String = { + val cut = cutName + val args = cutArgs + val suffix = labelSuffix + // The result column is named in PYTHON rather than here: joining it to + // `attribute` in Scala would hand `pyb` a plain string, and the value it was + // built to protect would be spliced into the template unguarded. + pyb"""from pytexera import * + |import pandas as pd + | + |class ProcessTableOperator(UDFTableOperator): + | + | @overrides + | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + | out = table.copy() + | _column = $attribute + | out[_column + "_bin"] = ${cut}(out[_column], ${args})${suffix} + | yield out""".encode + } + + override def generateStandaloneCode(): String = { + val column = pyStringLiteral(attribute) + val result = pyStringLiteral(resultColumn) + s"""out1df = in1df.copy() + |out1df[$result] = $cutName(out1df[$column], $cutArgs)$labelSuffix""".stripMargin + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala new file mode 100644 index 00000000000..f193578d02e --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala @@ -0,0 +1,109 @@ +/* + * 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.binning + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class BinningOpDescSpec extends AnyFlatSpec with Matchers { + + private val inputSchema = new Schema( + new Attribute("id", AttributeType.INTEGER), + new Attribute("age", AttributeType.DOUBLE) + ) + + private def desc(m: BinningMethod = BinningMethod.EQUAL_WIDTH, n: Int = 4): BinningOpDesc = { + val d = new BinningOpDesc + d.attribute = "age" + d.method = m + d.bins = n + d + } + + "BinningOpDesc.operatorInfo" should "advertise the name and the Cleaning group" in { + val info = (new BinningOpDesc).operatorInfo + info.userFriendlyName shouldBe "Binning" + info.operatorGroupName shouldBe OperatorGroupConstants.CLEANING_GROUP + info.inputPorts should have length 1 + info.outputPorts should have length 1 + } + + // A quantile cut is made at the column's own quantiles, which the last row can + // still move, so nothing may be emitted until the input ends. + it should "declare its output port blocking" in { + (new BinningOpDesc).operatorInfo.outputPorts.head.blocking shouldBe true + } + + "The output schema" should "append one STRING column named after the source" in { + val schema = desc().getOutputSchemas(Map(PortIdentity() -> inputSchema))(PortIdentity()) + schema.getAttributeNames shouldBe List("id", "age", "age_bin") + schema.getAttribute("age_bin").getType shouldBe AttributeType.STRING + } + + it should "refuse a derived name the input already carries" in { + val clashing = inputSchema.add("age_bin", AttributeType.STRING) + a[RuntimeException] should be thrownBy + desc().getOutputSchemas(Map(PortIdentity() -> clashing)) + } + + "The generated code" should "cut on the count of bins asked for" in { + desc(BinningMethod.EQUAL_WIDTH, 7).generateStandaloneCode() should include("bins=7") + } + + // A quantile cut can put two edges in one place where the values repeat, and + // dropping the duplicate yields fewer bins rather than raising. + it should "ask the quantile cut to drop a duplicate edge" in { + val code = desc(BinningMethod.EQUAL_FREQUENCY).generateStandaloneCode() + code should include("pd.qcut(") + code should include("""duplicates="drop"""") + code should include("q=4") + } + + it should "keep an empty cell empty rather than rendering it as text" in { + desc().generateStandaloneCode() should include("where(lambda s: s.notna(), None)") + } + + it should "hold the source and the derived name as escaped literals" in { + val d = desc() + d.attribute = "a\"b" + val code = d.generateStandaloneCode() + code should include("""in1df.copy()""") + code should include("""out1df["a\"b"]""") + code should include("""out1df["a\"b_bin"]""") + } + + // Both paths make the same call, on frames the two runtimes name differently, so + // a difference between them is a difference in this operator rather than in pandas. + "The two renderings" should "make the same pandas call" in { + val d = desc(BinningMethod.EQUAL_FREQUENCY, 5) + d.generateStandaloneCode() should include("""pd.qcut(out1df["age"], q=5, duplicates="drop")""") + d.generatePythonCode() should include("""q=5, duplicates="drop"""") + d.generatePythonCode() should include("pd.qcut(out[") + } + + "The platform code" should "be a table operator, since a quantile needs every row" in { + val code = desc().generatePythonCode() + code should include("class ProcessTableOperator(UDFTableOperator)") + code should include("def process_table(") + } +} diff --git a/frontend/src/assets/operator_images/Binning.png b/frontend/src/assets/operator_images/Binning.png new file mode 100644 index 0000000000000000000000000000000000000000..3f616cf97508f7c9185038ce69a2a433d75bb681 GIT binary patch literal 1532 zcmbW1eLNF*9LH@YBeIjDB-Rn8q&$_YW(o5!Pn*0f_=2=3K z%}C{W2ur9TY|d$$hnbLd>+a9%{W&Jho-8hK+{oViP^>da_@t(B5Cm{`)X@A#QZE zWJ#tB_uRlwu*a=3`Wh+VZgh679U_o2H|wW+5GlgPuQA1p0om+SJ^UDfD~hi_ZgTvt zvLwW}PFnR)#7Jy>PGvLTu-a)Bbiw<00;RFi)(I^Z$4~TJ{1E0nYpGR2YA}5FlJWM1 zjzX+e3kt^Q_8JRc-{enDRY})f{#esAnK{MXYs-0+nv{e&{Ic6?|$oE($Z;?qu`3&l8=qQ))`3^cR*n)NeLY$Fa zhEk(zItONd-qA%o02#JI+*9wi?G3!h{&;2Dy zJt$9iyoH=mZjJb(BDT971h(7K9iO|YT2og3>B!+hIzeSDux*ubJEbnY%EB>B93DZH{k5^xsKh z4w&O@jims&Swq_s^^~quoqSz;;_>Q1C7SpAk{Zzh=Ko2DDmHVnQ~_Xac=ra&H720>J))s$ZdgU-RQa zb)A6Vs0uH%MC^=n1YghXo9|NwN?2TdYLZH@c`0qbjO3|%X|9@NoWa{iTZD^h{Q8t$PzU4$eU^gK> z6!1aCXgQ!LXZ>n%Hd>pfdGS6-g2)N{syO+s-_4Lkk}#o=j?jl}RC&pHZ78b&i@q%n zCYTPw!=J?1IJGzK5s-BNMcJ3&gIZJ)B|2tnOM|%l?1hdT_$iw?_ojr}$%q#7G+pkW z02DR!zDL%bgb?j-JpIwy#OU95qCcq2K9*L^OMlWQW7!?n$v^#o^ZGM^)V2y=!OG4~CPXse5z`ELahCeg*6`()3EV?pvvb2CHy0~7yGU)| zjvaOvcAP)5Gx0-ZE83`4n{ z)lQw93!h9>cgY(gwTaY!TExK{%^k~5O(<=EY4Bt$|GSMk-l7w0jhUF%W3{`u(ag@b zkCU>kIAJugm~^$buRy@T8XnRe`OP=ppbvAyM>R@6<&7(bbjwU@@Um@YE80QJg>X(s zx10s_1!LJ&78)xfBR({EU E0_sPsf&c&j literal 0 HcmV?d00001 From f72755ccbc77cbf5e21a0a223eb3fd5bede8ffda Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 9 Sep 2026 23:50:52 -0700 Subject: [PATCH 2/2] fix(operator): bin a column that holds nothing rather than raising An equal-width cut has no edges to find when every cell of the column is empty, and none when no row arrived at all, so pandas raises before the suffix that keeps a hole a hole can run. Both renderings now skip the cut in that case and let the suffix turn the holes into the empty bin labels they already are. The equal-frequency cut does not raise on the same table, but it goes through the same guard: one shape of answer for a table with nothing in it, whichever way it was asked to cut. Two tests run the exported script over an all-empty column, an empty table and a column with one hole, since the defect is what pandas does at runtime rather than what the operator writes. Co-Authored-By: Claude Opus 5 (1M context) --- .../operator/binning/BinningOpDesc.scala | 15 +- .../operator/binning/BinningOpDescSpec.scala | 131 +++++++++++++++++- 2 files changed, 139 insertions(+), 7 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala index 373e3300841..58ff59ebfb0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/binning/BinningOpDesc.scala @@ -114,6 +114,11 @@ class BinningOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerato // The result column is named in PYTHON rather than here: joining it to // `attribute` in Scala would hand `pyb` a plain string, and the value it was // built to protect would be spliced into the template unguarded. + // + // The guard is there because edges cannot be found in a column that holds + // nothing: an equal-width cut raises both when every cell is empty and when + // no row arrived at all. Leaving those values uncut keeps the holes, and the + // suffix turns them into the empty bins they already are. pyb"""from pytexera import * |import pandas as pd | @@ -123,7 +128,10 @@ class BinningOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerato | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: | out = table.copy() | _column = $attribute - | out[_column + "_bin"] = ${cut}(out[_column], ${args})${suffix} + | _binned = out[_column] + | if _binned.notna().any(): + | _binned = ${cut}(_binned, ${args}) + | out[_column + "_bin"] = _binned${suffix} | yield out""".encode } @@ -131,6 +139,9 @@ class BinningOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerato val column = pyStringLiteral(attribute) val result = pyStringLiteral(resultColumn) s"""out1df = in1df.copy() - |out1df[$result] = $cutName(out1df[$column], $cutArgs)$labelSuffix""".stripMargin + |_binned = out1df[$column] + |if _binned.notna().any(): + | _binned = $cutName(_binned, $cutArgs) + |out1df[$result] = _binned$labelSuffix""".stripMargin } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala index f193578d02e..5cc0027c2f4 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/binning/BinningOpDescSpec.scala @@ -19,12 +19,18 @@ package org.apache.texera.amber.operator.binning +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.metadata.OperatorGroupConstants 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.util.Try + class BinningOpDescSpec extends AnyFlatSpec with Matchers { private val inputSchema = new Schema( @@ -92,13 +98,22 @@ class BinningOpDescSpec extends AnyFlatSpec with Matchers { code should include("""out1df["a\"b_bin"]""") } - // Both paths make the same call, on frames the two runtimes name differently, so - // a difference between them is a difference in this operator rather than in pandas. + // Edges cannot be found in a column that holds nothing, and an equal-width cut + // raises rather than handing back empty bins. Both renderings skip the cut, which + // leaves the holes for the suffix to turn into the empty labels they already are. + it should "leave a column with nothing in it uncut" in { + val guard = "if _binned.notna().any():" + desc().generateStandaloneCode() should include(guard) + desc().generatePythonCode() should include(guard) + } + + // Both paths make the same call, on a series both name the same, so a difference + // between them is a difference in this operator rather than in pandas. "The two renderings" should "make the same pandas call" in { val d = desc(BinningMethod.EQUAL_FREQUENCY, 5) - d.generateStandaloneCode() should include("""pd.qcut(out1df["age"], q=5, duplicates="drop")""") - d.generatePythonCode() should include("""q=5, duplicates="drop"""") - d.generatePythonCode() should include("pd.qcut(out[") + val call = """pd.qcut(_binned, q=5, duplicates="drop")""" + d.generateStandaloneCode() should include(call) + d.generatePythonCode() should include(call) } "The platform code" should "be a table operator, since a quantile needs every row" in { @@ -106,4 +121,110 @@ class BinningOpDescSpec extends AnyFlatSpec with Matchers { code should include("class ProcessTableOperator(UDFTableOperator)") code should include("def process_table(") } + + // Python executable resolution, following FilledAreaPlotOpDescSpec: + // udf.conf python.path (UDF_PYTHON_PATH), then python3 / python / py. + private def resolvePythonExecutable(): Option[String] = { + def fromConfig: Option[String] = { + val configOpt = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + configOpt + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + } + + def isRunnable(exe: String): Boolean = { + val pTry = Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()) + pTry.toOption.exists { p => + val finished = p.waitFor(5, TimeUnit.SECONDS) + if (!finished) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(isRunnable) + } + + private def canImportPandas(python: String): Boolean = { + val pTry = Try( + new ProcessBuilder(python, "-c", "import pandas").redirectErrorStream(true).start() + ) + pTry.toOption.exists { p => + val finished = p.waitFor(60, TimeUnit.SECONDS) + if (!finished) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + } + + // Runs the exported code as written over three frames, and reports the length of + // the bin column and how much of it was filled. The exported code is plain pandas, + // so nothing about the operator has to be stubbed for it to run. + private val runtimeDriverScript: String = + """import sys + |import pandas as pd + | + |code = open(sys.argv[1]).read() + |cases = { + | "allnull": pd.DataFrame({"age": pd.Series([None, None], dtype="float64")}), + | "empty": pd.DataFrame({"age": pd.Series([], dtype="float64")}), + | "mixed": pd.DataFrame({"age": pd.Series([1.0, 2.0, None])}), + |} + |for cid, frame in cases.items(): + | namespace = {"pd": pd, "in1df": frame} + | try: + | exec(code, namespace) + | labels = namespace["out1df"]["age_bin"] + | print("CASE %s OK:%d:%d" % (cid, len(labels), labels.notna().sum())) + | except Exception as error: + | print("CASE %s %s" % (cid, type(error).__name__)) + |""".stripMargin + + // An all-empty column used to raise on an equal-width cut, and so did an empty + // table, both before the suffix that keeps a hole a hole could run. + for (method <- BinningMethod.values) { + it should s"bin a column with nothing in it rather than raising, cutting by $method" in { + val python = resolvePythonExecutable().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandas(python)) { + cancel(s"'$python' cannot import pandas; skipping runtime verification") + } + + val moduleFile = Files.createTempFile("binning_op_", ".py") + val driverFile = Files.createTempFile("binning_driver_", ".py") + try { + val code = desc(method).generateStandaloneCode() + Files.write(moduleFile, code.getBytes(StandardCharsets.UTF_8)) + Files.write(driverFile, runtimeDriverScript.getBytes(StandardCharsets.UTF_8)) + + val process = new ProcessBuilder(python, driverFile.toString, moduleFile.toString) + .redirectErrorStream(true) + .start() + val finished = process.waitFor(120, TimeUnit.SECONDS) + if (!finished) { + process.destroyForcibly() + fail("Runtime verification driver timed out after 120s") + } + val output = new String(process.getInputStream.readAllBytes(), StandardCharsets.UTF_8) + withClue(s"Exported code:\n$code\nDriver output:\n$output\n") { + process.exitValue() shouldBe 0 + val verdicts = "CASE (\\S+) (\\S+)".r + .findAllMatchIn(output) + .map(m => m.group(1) -> m.group(2)) + .toMap + verdicts shouldBe Map( + "allnull" -> "OK:2:0", // two rows, neither of them in a bin + "empty" -> "OK:0:0", // no rows to put in one + "mixed" -> "OK:3:2" // the hole is the only row left unlabelled + ) + } + } finally { + Try(Files.deleteIfExists(moduleFile)) + Try(Files.deleteIfExists(driverFile)) + () + } + } + } }