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..cbba6b8ee6e 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.extractdatetime.ExtractDateTimeOpDesc 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[ExtractDateTimeOpDesc], name = "ExtractDateTime"), 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/extractdatetime/DateTimeField.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/DateTimeField.java new file mode 100644 index 00000000000..1133d910d4f --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/DateTimeField.java @@ -0,0 +1,64 @@ +/* + * 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.extractdatetime; + +import com.fasterxml.jackson.annotation.JsonValue; + +/** + * A field of a timestamp that can be read out of it as a whole number. + * + *

Read as ISO-8601 states them, which is what lets the engine and the exported + * Python agree: a Monday is 1, and pandas counts weekdays from 0. + */ +public enum DateTimeField { + + YEAR("year"), + + QUARTER("quarter"), + + MONTH("month"), + + DAY("day"), + + DAY_OF_WEEK("day of week"), + + DAY_OF_YEAR("day of year"), + + WEEK_OF_YEAR("week of year"), + + HOUR("hour"), + + MINUTE("minute"), + + SECOND("second"); + + private final String name; + + DateTimeField(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/extractdatetime/ExtractDateTimeOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpDesc.scala new file mode 100644 index 00000000000..44a2a54e9ab --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpDesc.scala @@ -0,0 +1,143 @@ +/* + * 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.extractdatetime + +import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} +import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow._ +import org.apache.texera.amber.operator.StandaloneCodeGenerator +import org.apache.texera.amber.operator.map.MapOpDesc +import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral +import org.apache.texera.amber.util.JSONUtils.objectMapper + +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "attribute": { + "enum": ["timestamp"] + } + } +} +""") +class ExtractDateTimeOpDesc extends MapOpDesc with StandaloneCodeGenerator { + + @JsonProperty(required = true) + @JsonSchemaTitle("Attribute") + @JsonPropertyDescription("timestamp column to read") + @AutofillAttributeName + var attribute: String = _ + + @JsonProperty(required = true) + @JsonSchemaTitle("Fields") + @JsonPropertyDescription("parts of the timestamp to add as columns") + var fields: List[DateTimeField] = List.empty + + override def operatorInfo: OperatorInfo = + OperatorInfo( + userFriendlyName = "Extract Date/Time Fields", + operatorDescription = + "Read the year, month, weekday or another whole-number part out of a timestamp column", + operatorGroupName = OperatorGroupConstants.CLEANING_GROUP, + inputPorts = List(InputPort()), + outputPorts = List(OutputPort()) + ) + + /** The fields asked for, with the empty and the null cases answered once. */ + private def asked: List[DateTimeField] = + Option(fields).getOrElse(List.empty).filter(_ != null).distinct + + /** What a field is called once it is a column of its own: the source column and + * the field, so reading two timestamp columns names four distinct results and a + * reader can see which came from where. + */ + private def columnFor(field: DateTimeField): String = + s"${attribute}_${field.getName.replace(' ', '_')}" + + override def getPhysicalOp( + workflowId: WorkflowIdentity, + executionId: ExecutionIdentity + ): PhysicalOp = + PhysicalOp + .oneToOnePhysicalOp( + workflowId, + executionId, + operatorIdentifier, + OpExecWithClassName( + "org.apache.texera.amber.operator.extractdatetime.ExtractDateTimeOpExec", + objectMapper.writeValueAsString(this) + ) + ) + .withInputPorts(operatorInfo.inputPorts) + .withOutputPorts(operatorInfo.outputPorts) + .withPropagateSchema( + SchemaPropagationFunc { inputSchemas: Map[PortIdentity, Schema] => + // Every field reads as a whole number, so the added columns are INTEGER + // whichever fields were asked for. `add` refuses a name the input already + // carries, which is how a collision is reported before the operator runs. + val outputSchema = asked.foldLeft(inputSchemas.values.head) { (schema, field) => + schema.add(columnFor(field), AttributeType.INTEGER) + } + Map(operatorInfo.outputPorts.head.id -> outputSchema) + } + ) + + override def generateStandaloneCode(): String = { + if (asked.isEmpty) return "out1df = in1df.copy()" + val source = pyStringLiteral(attribute) + val lines = scala.collection.mutable.ArrayBuffer[String]( + "out1df = in1df.copy()", + // A no-op where the source already parsed its input, which is the usual case; + // made anyway for one that handed the column over as text. NOT coerced: the + // engine reads a real moment here, so a cell Python cannot is a disagreement. + s"""_texera_ts = pd.to_datetime(out1df[$source])""" + ) + asked.foreach { field => + val target = pyStringLiteral(columnFor(field)) + // Int64 rather than int64: a NaT has no year, and only the nullable dtype + // can hold the hole the engine leaves there. + lines += s"""out1df[$target] = ${expressionFor(field)}.astype("Int64")""" + } + lines.mkString("\n") + } + + /** The pandas reading of one field, stated in ISO terms where pandas does not. + * + * Weekday is the one that has to be said: pandas counts Monday as 0, ISO and + * `java.time.DayOfWeek` count it as 1. + */ + private def expressionFor(field: DateTimeField): String = + field match { + case DateTimeField.YEAR => "_texera_ts.dt.year" + case DateTimeField.QUARTER => "_texera_ts.dt.quarter" + case DateTimeField.MONTH => "_texera_ts.dt.month" + case DateTimeField.DAY => "_texera_ts.dt.day" + case DateTimeField.DAY_OF_WEEK => "(_texera_ts.dt.dayofweek + 1)" + case DateTimeField.DAY_OF_YEAR => "_texera_ts.dt.dayofyear" + case DateTimeField.WEEK_OF_YEAR => "_texera_ts.dt.isocalendar().week" + case DateTimeField.HOUR => "_texera_ts.dt.hour" + case DateTimeField.MINUTE => "_texera_ts.dt.minute" + case DateTimeField.SECOND => "_texera_ts.dt.second" + } +} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpExec.scala new file mode 100644 index 00000000000..e18700f10de --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpExec.scala @@ -0,0 +1,64 @@ +/* + * 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.extractdatetime + +import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.texera.amber.operator.map.MapOpExec +import org.apache.texera.amber.util.JSONUtils.objectMapper + +import java.sql.Timestamp +import java.time.LocalDateTime +import java.time.temporal.IsoFields + +class ExtractDateTimeOpExec(descString: String) extends MapOpExec { + + private val desc: ExtractDateTimeOpDesc = + objectMapper.readValue(descString, classOf[ExtractDateTimeOpDesc]) + + this.setMapFunc(extract) + + private def extract(tuple: Tuple): TupleLike = { + val moment = Option(tuple.getField[Timestamp](desc.attribute)).map(_.toLocalDateTime) + // A null timestamp has no fields, so every column this operator adds is empty + // for that row rather than the row being dropped: the operator adds columns and + // says nothing about which rows belong. + val added = Option(desc.fields) + .getOrElse(List.empty) + .filter(_ != null) + .distinct + .map(field => moment.map(m => Int.box(read(m, field))).orNull) + TupleLike(tuple.getFields ++ added) + } + + /** One field of a moment, in the ISO reading the exported Python also states. */ + private def read(moment: LocalDateTime, field: DateTimeField): Int = + field match { + case DateTimeField.YEAR => moment.getYear + case DateTimeField.QUARTER => moment.get(IsoFields.QUARTER_OF_YEAR) + case DateTimeField.MONTH => moment.getMonthValue + case DateTimeField.DAY => moment.getDayOfMonth + case DateTimeField.DAY_OF_WEEK => moment.getDayOfWeek.getValue + case DateTimeField.DAY_OF_YEAR => moment.getDayOfYear + case DateTimeField.WEEK_OF_YEAR => moment.get(IsoFields.WEEK_OF_WEEK_BASED_YEAR) + case DateTimeField.HOUR => moment.getHour + case DateTimeField.MINUTE => moment.getMinute + case DateTimeField.SECOND => moment.getSecond + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpDescSpec.scala new file mode 100644 index 00000000000..bfb5062665e --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/extractdatetime/ExtractDateTimeOpDescSpec.scala @@ -0,0 +1,182 @@ +/* + * 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.extractdatetime + +import org.apache.texera.amber.core.executor.OpExecWithClassName +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.sql.Timestamp + +class ExtractDateTimeOpDescSpec extends AnyFlatSpec with Matchers { + + private val workflowId = WorkflowIdentity(1L) + private val executionId = ExecutionIdentity(1L) + + private val inputSchema = new Schema( + new Attribute("id", AttributeType.INTEGER), + new Attribute("ts", AttributeType.TIMESTAMP) + ) + + private def desc(fields: DateTimeField*): ExtractDateTimeOpDesc = { + val d = new ExtractDateTimeOpDesc + d.attribute = "ts" + d.fields = fields.toList + d + } + + private def outputSchema(d: ExtractDateTimeOpDesc): Schema = + d.getPhysicalOp(workflowId, executionId) + .propagateSchema + .func(Map(PortIdentity() -> inputSchema))(PortIdentity()) + + private def rowsOf(d: ExtractDateTimeOpDesc, moments: Option[String]*): Seq[Seq[Any]] = { + val exec = new ExtractDateTimeOpExec(objectMapper.writeValueAsString(d)) + exec.open() + val out = moments.zipWithIndex.map { + case (moment, i) => + val b = Tuple.builder(inputSchema) + b.add(inputSchema.getAttribute("id"), Int.box(i)) + b.add(inputSchema.getAttribute("ts"), moment.map(Timestamp.valueOf).orNull) + exec.processTuple(b.build(), 0).next().getFields.toSeq + } + exec.close() + out + } + + "ExtractDateTimeOpDesc.operatorInfo" should "advertise the name and the Cleaning group" in { + val info = (new ExtractDateTimeOpDesc).operatorInfo + info.userFriendlyName shouldBe "Extract Date/Time Fields" + info.operatorGroupName shouldBe OperatorGroupConstants.CLEANING_GROUP + info.inputPorts should have length 1 + info.outputPorts should have length 1 + } + + "ExtractDateTimeOpDesc.getPhysicalOp" should "wire ExtractDateTimeOpExec" in { + (new ExtractDateTimeOpDesc) + .getPhysicalOp(workflowId, executionId) + .opExecInitInfo match { + case OpExecWithClassName(className, _) => + className shouldBe "org.apache.texera.amber.operator.extractdatetime.ExtractDateTimeOpExec" + case other => fail(s"unexpected executor: $other") + } + } + + "The output schema" should "name each added column after the source and the field" in { + val schema = outputSchema(desc(DateTimeField.YEAR, DateTimeField.DAY_OF_WEEK)) + schema.getAttributeNames shouldBe List("id", "ts", "ts_year", "ts_day_of_week") + schema.getAttribute("ts_year").getType shouldBe AttributeType.INTEGER + schema.getAttribute("ts_day_of_week").getType shouldBe AttributeType.INTEGER + } + + it should "keep the input untouched when no field is asked for" in { + outputSchema(desc()).getAttributeNames shouldBe List("id", "ts") + } + + // The same field twice names one column, so the repeat is dropped rather than + // reaching the schema as a duplicate. + it should "add one column for a field asked for twice" in { + outputSchema(desc(DateTimeField.YEAR, DateTimeField.YEAR)).getAttributeNames shouldBe List( + "id", + "ts", + "ts_year" + ) + } + + it should "refuse a derived name the input already carries" in { + val d = new ExtractDateTimeOpDesc + d.attribute = "ts" + d.fields = List(DateTimeField.YEAR) + val clashing = new Schema( + new Attribute("ts", AttributeType.TIMESTAMP), + new Attribute("ts_year", AttributeType.INTEGER) + ) + a[RuntimeException] should be thrownBy + d.getPhysicalOp(workflowId, executionId) + .propagateSchema + .func(Map(PortIdentity() -> clashing)) + } + + // 2024-03-05 14:09:07 is a Tuesday in ISO week 10 of Q1, day 65 of the year. + "The executor" should "read every field the way ISO-8601 states it" in { + val d = desc( + DateTimeField.YEAR, + DateTimeField.QUARTER, + DateTimeField.MONTH, + DateTimeField.DAY, + DateTimeField.DAY_OF_WEEK, + DateTimeField.DAY_OF_YEAR, + DateTimeField.WEEK_OF_YEAR, + DateTimeField.HOUR, + DateTimeField.MINUTE, + DateTimeField.SECOND + ) + rowsOf(d, Some("2024-03-05 14:09:07")).head.drop(2) shouldBe + Seq(2024, 1, 3, 5, 2, 65, 10, 14, 9, 7) + } + + it should "count Monday as 1 and Sunday as 7" in { + val week = Seq( + "2024-03-04", + "2024-03-05", + "2024-03-06", + "2024-03-07", + "2024-03-08", + "2024-03-09", + "2024-03-10" + ).map(day => Some(s"$day 00:00:00")) + rowsOf(desc(DateTimeField.DAY_OF_WEEK), week: _*).map(_.last) shouldBe Seq(1, 2, 3, 4, 5, 6, 7) + } + + // The operator adds columns; it says nothing about which rows belong, so a row + // whose timestamp is empty keeps its place with the added columns empty. + it should "leave the added columns empty for an empty timestamp, and keep the row" in { + val rows = rowsOf( + desc(DateTimeField.YEAR, DateTimeField.MONTH), + Some("2024-03-05 14:09:07"), + None + ) + rows should have length 2 + rows(1).drop(2) shouldBe Seq(null, null) + } + + "The generated Python" should "state the ISO weekday, which pandas does not" in { + desc(DateTimeField.DAY_OF_WEEK).generateStandaloneCode() should + include("_texera_ts.dt.dayofweek + 1") + } + + it should "hold the source and the derived name as escaped literals" in { + val d = new ExtractDateTimeOpDesc + d.attribute = "a\"b" + d.fields = List(DateTimeField.YEAR) + val code = d.generateStandaloneCode() + code should include("""out1df["a\"b"]""") + code should include("""out1df["a\"b_year"]""") + } + + it should "copy the frame through when no field is asked for" in { + desc().generateStandaloneCode() shouldBe "out1df = in1df.copy()" + } +} diff --git a/frontend/src/assets/operator_images/ExtractDateTime.png b/frontend/src/assets/operator_images/ExtractDateTime.png new file mode 100644 index 00000000000..0f2eea28cb8 Binary files /dev/null and b/frontend/src/assets/operator_images/ExtractDateTime.png differ