From 288ce3e663093628d00db56e9f5e875023629450 Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 20:51:17 +0100 Subject: [PATCH 1/6] Handle Option correctly Signed-off-by: Hongxin Liang --- .../flytekitscala/SdkScalaTypeTest.scala | 51 ++++++++++++++--- .../flyte/flytekitscala/SdkLiteralTypes.scala | 56 +++++++++---------- 2 files changed, 69 insertions(+), 38 deletions(-) diff --git a/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala b/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala index 72000200..11d67010 100644 --- a/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala +++ b/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala @@ -49,7 +49,12 @@ import org.flyte.flytekitscala.SdkLiteralTypes.{ } // The constructor is reflectedly invoked so it cannot be an inner class -case class ScalarNested(foo: String, bar: String) +case class ScalarNested( + foo: String, + bar: Option[String], + nestedNested: Option[ScalarNestedNested] +) +case class ScalarNestedNested(foo: String, bar: Option[String]) class SdkScalaTypeTest { @@ -178,7 +183,15 @@ class SdkScalaTypeTest { Struct.of( Map( "foo" -> Struct.Value.ofStringValue("foo"), - "bar" -> Struct.Value.ofStringValue("bar") + "bar" -> Struct.Value.ofStringValue("bar"), + "nestedNested" -> Struct.Value.ofStructValue( + Struct.of( + Map( + "foo" -> Struct.Value.ofStringValue("foo"), + "bar" -> Struct.Value.ofStringValue("bar") + ).asJava + ) + ) ).asJava ) ) @@ -196,7 +209,11 @@ class SdkScalaTypeTest { blob = SdkBindingDataFactory.of(blob), generic = SdkBindingDataFactory.of( SdkLiteralTypes.generics(), - ScalarNested("foo", "bar") + ScalarNested( + "foo", + Some("bar"), + Some(ScalarNestedNested("foo", Some("bar"))) + ) ) ) @@ -218,7 +235,11 @@ class SdkScalaTypeTest { blob = SdkBindingDataFactory.of(blob), generic = SdkBindingDataFactory.of( SdkLiteralTypes.generics(), - ScalarNested("foo", "bar") + ScalarNested( + "foo", + Some("bar"), + Some(ScalarNestedNested("foo", Some("bar"))) + ) ) ) @@ -245,7 +266,15 @@ class SdkScalaTypeTest { Struct.of( Map( "foo" -> Struct.Value.ofStringValue("foo"), - "bar" -> Struct.Value.ofStringValue("bar") + "bar" -> Struct.Value.ofStringValue("bar"), + "nestedNested" -> Struct.Value.ofStructValue( + Struct.of( + Map( + "foo" -> Struct.Value.ofStringValue("foo"), + "bar" -> Struct.Value.ofStringValue("bar") + ).asJava + ) + ) ).asJava ) ) @@ -285,7 +314,11 @@ class SdkScalaTypeTest { blob = SdkBindingDataFactory.of(blob), generic = SdkBindingDataFactory.of( SdkLiteralTypes.generics(), - ScalarNested("foo", "bar") + ScalarNested( + "foo", + Some("bar"), + Some(ScalarNestedNested("foo", Some("bar"))) + ) ) ) @@ -301,7 +334,11 @@ class SdkScalaTypeTest { "blob" -> SdkBindingDataFactory.of(blob), "generic" -> SdkBindingDataFactory.of( SdkLiteralTypes.generics[ScalarNested](), - ScalarNested("foo", "bar") + ScalarNested( + "foo", + Some("bar"), + Some(ScalarNestedNested("foo", Some("bar"))) + ) ) ).asJava diff --git a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala index 6a36f352..be0f0ead 100644 --- a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala +++ b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala @@ -297,41 +297,35 @@ object SdkLiteralTypes { ): S = { val mirror = runtimeMirror(classTag[S].runtimeClass.getClassLoader) - def valueToParamValue(value: Any, param: Symbol): Any = { - def valueToParamValue0(value: Any, param: Symbol): Any = { - if (param.typeSignature =:= typeOf[Byte]) { - value.asInstanceOf[Double].toByte - } else if (param.typeSignature =:= typeOf[Short]) { - value.asInstanceOf[Double].toShort - } else if (param.typeSignature =:= typeOf[Int]) { - value.asInstanceOf[Double].toInt - } else if (param.typeSignature =:= typeOf[Long]) { - value.asInstanceOf[Double].toLong - } else if (param.typeSignature =:= typeOf[Float]) { - value.asInstanceOf[Double].toFloat - } else if (param.typeSignature <:< typeOf[Product]) { - val typeTag = createTypeTag(param.typeSignature) - val classTag = ClassTag( - typeTag.mirror.runtimeClass(param.typeSignature) - ) - mapToProduct(value.asInstanceOf[Map[String, Any]])( - typeTag, - classTag - ) - } else { - value - } - } - - if (param.typeSignature <:< typeOf[Option[Any]]) { + def valueToParamValue(value: Any, tpe: Type): Any = { + if (tpe =:= typeOf[Byte]) { + value.asInstanceOf[Double].toByte + } else if (tpe =:= typeOf[Short]) { + value.asInstanceOf[Double].toShort + } else if (tpe =:= typeOf[Int]) { + value.asInstanceOf[Double].toInt + } else if (tpe =:= typeOf[Long]) { + value.asInstanceOf[Double].toLong + } else if (tpe =:= typeOf[Float]) { + value.asInstanceOf[Double].toFloat + } else if (tpe <:< typeOf[Option[Any]]) { // this has to be before Product check Some( - valueToParamValue0( + valueToParamValue( value, - param.typeSignature.dealias.typeArgs.head.typeSymbol + tpe.dealias.typeArgs.head ) ) + } else if (tpe <:< typeOf[Product]) { + val typeTag = createTypeTag(tpe) + val classTag = ClassTag( + typeTag.mirror.runtimeClass(tpe) + ) + mapToProduct(value.asInstanceOf[Map[String, Any]])( + typeTag, + classTag + ) } else { - valueToParamValue0(value, param) + value } } @@ -371,7 +365,7 @@ object SdkLiteralTypes { s"Map is missing required parameter named $paramName" ) ) - valueToParamValue(value, param) + valueToParamValue(value, param.typeSignature.dealias) }) constructorMirror(constructorArgs: _*).asInstanceOf[S] From e53f3610b2ea549f7fc82f0b7ebe86081d8fb4a0 Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 21:12:59 +0100 Subject: [PATCH 2/6] Clearer comment Signed-off-by: Hongxin Liang --- .../main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala index be0f0ead..e7cc445a 100644 --- a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala +++ b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala @@ -308,7 +308,7 @@ object SdkLiteralTypes { value.asInstanceOf[Double].toLong } else if (tpe =:= typeOf[Float]) { value.asInstanceOf[Double].toFloat - } else if (tpe <:< typeOf[Option[Any]]) { // this has to be before Product check + } else if (tpe <:< typeOf[Option[Any]]) { // this has to be before Product check because Option is a Product Some( valueToParamValue( value, From 57a581c7c2d4af696c3e2ab74e0b9382e009b686 Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 21:13:03 +0100 Subject: [PATCH 3/6] IT Signed-off-by: Hongxin Liang --- .../examples/flytekitscala/LaunchPlanRegistry.scala | 11 ++++++++--- .../flyte/examples/flytekitscala/NestedIOTask.scala | 2 +- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/LaunchPlanRegistry.scala b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/LaunchPlanRegistry.scala index df5c3b43..ae9e19ac 100644 --- a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/LaunchPlanRegistry.scala +++ b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/LaunchPlanRegistry.scala @@ -73,14 +73,19 @@ class LaunchPlanRegistry extends SimpleSdkLaunchPlanRegistry { 6.toDouble, "hello", List("1", "2"), - List(NestedNested(7.toDouble, NestedNestedNested("world"))), + List(NestedNested(7.toDouble, Some(NestedNestedNested("world")))), Map("1" -> "1", "2" -> "2"), - Map("foo" -> NestedNested(7.toDouble, NestedNestedNested("world"))), + Map( + "foo" -> NestedNested( + 7.toDouble, + Some(NestedNestedNested("world")) + ) + ), Some(false), None, Some(List("3", "4")), Some(Map("3" -> "3", "4" -> "4")), - NestedNested(7.toDouble, NestedNestedNested("world")) + NestedNested(7.toDouble, Some(NestedNestedNested("world"))) ) ) ) diff --git a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala index ef4d6124..d69b5242 100644 --- a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala +++ b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala @@ -24,7 +24,7 @@ import org.flyte.flytekitscala.{ } case class NestedNestedNested(string: String) -case class NestedNested(double: Double, nested: NestedNestedNested) +case class NestedNested(double: Double, nested: Option[NestedNestedNested]) case class Nested( boolean: Boolean, byte: Byte, From 51c95e8e8ba4de5958d014378816c5f607291081 Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 21:36:28 +0100 Subject: [PATCH 4/6] Early return of None Signed-off-by: Hongxin Liang --- .../org/flyte/flytekitscala/SdkScalaTypeTest.scala | 4 ++-- .../org/flyte/flytekitscala/SdkLiteralTypes.scala | 14 +++++++++----- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala b/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala index 11d67010..2424c823 100644 --- a/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala +++ b/flytekit-scala-tests/src/test/scala/org/flyte/flytekitscala/SdkScalaTypeTest.scala @@ -183,7 +183,7 @@ class SdkScalaTypeTest { Struct.of( Map( "foo" -> Struct.Value.ofStringValue("foo"), - "bar" -> Struct.Value.ofStringValue("bar"), + "bar" -> Struct.Value.ofNullValue(), "nestedNested" -> Struct.Value.ofStructValue( Struct.of( Map( @@ -211,7 +211,7 @@ class SdkScalaTypeTest { SdkLiteralTypes.generics(), ScalarNested( "foo", - Some("bar"), + None, Some(ScalarNestedNested("foo", Some("bar"))) ) ) diff --git a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala index e7cc445a..517ec24d 100644 --- a/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala +++ b/flytekit-scala_2.13/src/main/scala/org/flyte/flytekitscala/SdkLiteralTypes.scala @@ -309,12 +309,16 @@ object SdkLiteralTypes { } else if (tpe =:= typeOf[Float]) { value.asInstanceOf[Double].toFloat } else if (tpe <:< typeOf[Option[Any]]) { // this has to be before Product check because Option is a Product - Some( - valueToParamValue( - value, - tpe.dealias.typeArgs.head + if (value == None) { // None is used to represent Struct.Value.Kind.NULL_VALUE when converting struct to map + None + } else { + Some( + valueToParamValue( + value, + tpe.dealias.typeArgs.head + ) ) - ) + } } else if (tpe <:< typeOf[Product]) { val typeTag = createTypeTag(tpe) val classTag = ClassTag( From bd80eab8057e456e436e595cd18c829ade3ba04a Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 22:35:40 +0100 Subject: [PATCH 5/6] Enrich IT workflow Signed-off-by: Hongxin Liang --- .../examples/flytekitscala/NestedIOTask.scala | 23 ++++++++++--------- .../flytekitscala/NestedIOWorkflow.scala | 3 ++- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala index d69b5242..6f6c165a 100644 --- a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala +++ b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOTask.scala @@ -57,9 +57,6 @@ case class NestedIOTaskOutput( generic: SdkBindingData[Nested] ) -/** Example Flyte task that takes a name as the input and outputs a simple - * greeting message. - */ class NestedIOTask extends SdkRunnableTask[ NestedIOTaskInput, @@ -69,17 +66,21 @@ class NestedIOTask SdkScalaType[NestedIOTaskOutput] ) { - /** Defines task behavior. This task takes a name as the input, wraps it in a - * welcome message, and outputs the message. - * - * @param input - * the name of the person to be greeted - * @return - * the welcome message - */ override def run(input: NestedIOTaskInput): NestedIOTaskOutput = NestedIOTaskOutput( input.name, input.generic ) } + +class NestedIOTaskNoop + extends SdkRunnableTask[ + NestedIOTaskOutput, + NestedIOTaskOutput + ]( + SdkScalaType[NestedIOTaskOutput], + SdkScalaType[NestedIOTaskOutput] + ) { + + override def run(input: NestedIOTaskOutput): NestedIOTaskOutput = input +} diff --git a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOWorkflow.scala b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOWorkflow.scala index dfe99665..bd926873 100644 --- a/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOWorkflow.scala +++ b/flytekit-examples-scala/src/main/scala/org/flyte/examples/flytekitscala/NestedIOWorkflow.scala @@ -32,6 +32,7 @@ class NestedIOWorkflow builder: SdkScalaWorkflowBuilder, input: NestedIOTaskInput ): Unit = { - builder.apply(new NestedIOTask(), input) + val output = builder.apply(new NestedIOTask(), input) + builder.apply(new NestedIOTaskNoop(), output.getOutputs) } } From a9014a6e05ba9b10250e83e70a862bb82628df3e Mon Sep 17 00:00:00 2001 From: Hongxin Liang Date: Wed, 22 Nov 2023 22:51:54 +0100 Subject: [PATCH 6/6] Resource Signed-off-by: Hongxin Liang --- .../META-INF/services/org.flyte.flytekit.SdkRunnableTask | 1 + 1 file changed, 1 insertion(+) diff --git a/flytekit-examples-scala/src/main/resources/META-INF/services/org.flyte.flytekit.SdkRunnableTask b/flytekit-examples-scala/src/main/resources/META-INF/services/org.flyte.flytekit.SdkRunnableTask index 508e6cb5..201af9a9 100644 --- a/flytekit-examples-scala/src/main/resources/META-INF/services/org.flyte.flytekit.SdkRunnableTask +++ b/flytekit-examples-scala/src/main/resources/META-INF/services/org.flyte.flytekit.SdkRunnableTask @@ -4,3 +4,4 @@ org.flyte.examples.flytekitscala.GreetTask org.flyte.examples.flytekitscala.AddQuestionTask org.flyte.examples.flytekitscala.NoInputsTask org.flyte.examples.flytekitscala.NestedIOTask +org.flyte.examples.flytekitscala.NestedIOTaskNoop