From 3bd88b65d3daa7bba30952f6b47f75bfdef24111 Mon Sep 17 00:00:00 2001 From: dale <0500261j@gmail.com> Date: Wed, 24 Jun 2026 10:17:48 +0200 Subject: [PATCH 1/8] Avoid capturing default config in numeric parser UDF --- .../absa/standardization/udf/UDFBuilder.scala | 30 +++++++++++--- .../absa/standardization/udf/UDFResult.scala | 17 ++++++-- .../standardization/udf/UDFBuilderSuite.scala | 41 ++++++++++++++++++- 3 files changed, 79 insertions(+), 9 deletions(-) diff --git a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala index fa06da2..c58f92d 100644 --- a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala +++ b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala @@ -19,7 +19,11 @@ package za.co.absa.standardization.udf import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.functions.udf import org.apache.spark.sql.types.DataType -import za.co.absa.standardization.config.StandardizationConfig +import za.co.absa.standardization.config.{ + BasicErrorCodesConfig, + ErrorCodesConfig, + StandardizationConfig +} import za.co.absa.standardization.types.parsers.NumericParser import za.co.absa.standardization.types.parsers.NumericParser.NumericParserException @@ -39,9 +43,25 @@ object UDFBuilder { val vColumnNameForError = columnNameForError val vDefaultValue = defaultValue val vColumnNullable = columnNullable - val vStdConfig = stdConfig + val vErrorCodes = BasicErrorCodesConfig( + stdConfig.errorCodes.castError, + stdConfig.errorCodes.nullError, + stdConfig.errorCodes.typeError, + stdConfig.errorCodes.schemaError + ) - udf[UDFResult[T], String](numericParserToTyped(_, sourceDataType, targetDataType, vParser, vColumnNullable, vColumnNameForError, vStdConfig, vDefaultValue)) + udf[UDFResult[T], String]( + numericParserToTyped( + _, + sourceDataType, + targetDataType, + vParser, + vColumnNullable, + vColumnNameForError, + vErrorCodes, + vDefaultValue + ) + ) } private def numericParserToTyped[T](input: String, @@ -50,14 +70,14 @@ object UDFBuilder { parser: NumericParser[T], columnNullable: Boolean, columnNameForError: String, - stdConfig: StandardizationConfig, + errorCodes: ErrorCodesConfig, defaultValue: Option[T]): UDFResult[T] = { val result = Option(input) match { case Some(string) => parser.parse(string).map(Some(_)) case None if columnNullable => Success(None) case None => Failure(nullException) } - UDFResult.fromTry(result, columnNameForError, input, sourceDataType.typeName, targetDataType.typeName, None, stdConfig, defaultValue) + UDFResult.fromTry(result, columnNameForError, input, sourceDataType.typeName, targetDataType.typeName, None, errorCodes, defaultValue) } private val nullException = new NumericParserException("Null value on input for non-nullable field") diff --git a/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala b/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala index 6329726..ef2a208 100644 --- a/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala +++ b/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala @@ -18,7 +18,7 @@ package za.co.absa.standardization.udf import za.co.absa.standardization.ErrorMessage import za.co.absa.standardization.StandardizationErrorMessage -import za.co.absa.standardization.config.StandardizationConfig +import za.co.absa.standardization.config.{ErrorCodesConfig, StandardizationConfig} import scala.util.{Failure, Success, Try} @@ -38,11 +38,22 @@ object UDFResult { pattern: Option[String], stdConfig: StandardizationConfig, defaultValue: Option[T] = None): UDFResult[T] = { + fromTry(result, columnName, rawValue, sourceType, targetType, pattern, stdConfig.errorCodes, defaultValue) + } + + def fromTry[T](result: Try[Option[T]], + columnName: String, + rawValue: String, + sourceType: String, + targetType: String, + pattern: Option[String], + errorCodes: ErrorCodesConfig, + defaultValue: Option[T] = None): UDFResult[T] = { result match { case Success(success) => UDFResult.success(success) - case Failure(_) if Option(rawValue).isEmpty => UDFResult(defaultValue, Seq(StandardizationErrorMessage.stdNullErr(columnName)(stdConfig.errorCodes))) + case Failure(_) if Option(rawValue).isEmpty => UDFResult(defaultValue, Seq(StandardizationErrorMessage.stdNullErr(columnName)(errorCodes))) case Failure(_) => - UDFResult(defaultValue, Seq(StandardizationErrorMessage.stdCastErr(columnName, rawValue, sourceType, targetType, pattern)(stdConfig.errorCodes))) + UDFResult(defaultValue, Seq(StandardizationErrorMessage.stdCastErr(columnName, rawValue, sourceType, targetType, pattern)(errorCodes))) } } } diff --git a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala index 383f892..e62a591 100644 --- a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala +++ b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala @@ -21,7 +21,12 @@ import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.types._ import org.scalatest.funsuite.AnyFunSuite import za.co.absa.standardization.RecordIdGeneration.IdType.NoId -import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig, StandardizationConfig} +import za.co.absa.standardization.config.{ + BasicMetadataColumnsConfig, + BasicStandardizationConfig, + DefaultStandardizationConfig, + StandardizationConfig +} import za.co.absa.standardization.schema.MetadataKeys import za.co.absa.standardization.types.TypedStructField._ import za.co.absa.standardization.types.parsers.IntegralParser.{PatternIntegralParser, RadixIntegralParser} @@ -144,4 +149,38 @@ class UDFBuilderSuite extends AnyFunSuite { ois.readObject().asInstanceOf[UserDefinedFunction] } + test("Serialization and deserialization of stringUdfViaNumericParser with default config") { + val fieldName = "test" + val field: StructField = StructField(fieldName, IntegerType, nullable = true, new MetadataBuilder() + .putString(MetadataKeys.Pattern, "000000") + .build) + val typedField = TypedStructField(field) + + val numericTypeField = typedField.asInstanceOf[NumericTypeStructField[Int]] + val defaultValue: Option[Int] = typedField.defaultValueWithGlobal.get.map(_.asInstanceOf[Int]) + val parser = numericTypeField.parser.get.asInstanceOf[PatternIntegralParser[Int]] + val udfFnc = UDFBuilder.stringUdfViaNumericParser( + StringType, + field.dataType, + parser, + numericTypeField.nullable, + fieldName, + DefaultStandardizationConfig, + defaultValue + ) + //write + val baos = new ByteArrayOutputStream + val oos = new ObjectOutputStream(baos) + oos.writeObject(udfFnc) + oos.flush() + val serialized = baos.toByteArray + assert(serialized.nonEmpty) + //read + val ois = new ObjectInputStream(new ByteArrayInputStream(serialized)) { + override def resolveClass(desc: ObjectStreamClass): Class[_] = + Class.forName(desc.getName, false, loader) + } + ois.readObject().asInstanceOf[UserDefinedFunction] + } + } From 15e3ff2282fed26b154fb1a015d56f3ae5aca982 Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:07:46 +0200 Subject: [PATCH 2/8] Pass configured error codes to numeric UDF --- .../co/absa/standardization/udf/UDFBuilder.scala | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala index c58f92d..3a1e466 100644 --- a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala +++ b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala @@ -19,11 +19,7 @@ package za.co.absa.standardization.udf import org.apache.spark.sql.expressions.UserDefinedFunction import org.apache.spark.sql.functions.udf import org.apache.spark.sql.types.DataType -import za.co.absa.standardization.config.{ - BasicErrorCodesConfig, - ErrorCodesConfig, - StandardizationConfig -} +import za.co.absa.standardization.config.{ErrorCodesConfig, StandardizationConfig} import za.co.absa.standardization.types.parsers.NumericParser import za.co.absa.standardization.types.parsers.NumericParser.NumericParserException @@ -43,12 +39,6 @@ object UDFBuilder { val vColumnNameForError = columnNameForError val vDefaultValue = defaultValue val vColumnNullable = columnNullable - val vErrorCodes = BasicErrorCodesConfig( - stdConfig.errorCodes.castError, - stdConfig.errorCodes.nullError, - stdConfig.errorCodes.typeError, - stdConfig.errorCodes.schemaError - ) udf[UDFResult[T], String]( numericParserToTyped( @@ -58,7 +48,7 @@ object UDFBuilder { vParser, vColumnNullable, vColumnNameForError, - vErrorCodes, + stdConfig.errorCodes, vDefaultValue ) ) From 7037580e5a5af2459c5dd51ac2aad5ce540d5118 Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:12:01 +0200 Subject: [PATCH 3/8] Fix UDFResult overload defaults --- src/main/scala/za/co/absa/standardization/udf/UDFResult.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala b/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala index ef2a208..cd8f229 100644 --- a/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala +++ b/src/main/scala/za/co/absa/standardization/udf/UDFResult.scala @@ -48,7 +48,7 @@ object UDFResult { targetType: String, pattern: Option[String], errorCodes: ErrorCodesConfig, - defaultValue: Option[T] = None): UDFResult[T] = { + defaultValue: Option[T]): UDFResult[T] = { result match { case Success(success) => UDFResult.success(success) case Failure(_) if Option(rawValue).isEmpty => UDFResult(defaultValue, Seq(StandardizationErrorMessage.stdNullErr(columnName)(errorCodes))) From 88b820b92dd79db7e45c19e8ebffa306985800c6 Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:16:34 +0200 Subject: [PATCH 4/8] Avoid capturing standardization config in UDF --- src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala index 3a1e466..1de9cd6 100644 --- a/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala +++ b/src/main/scala/za/co/absa/standardization/udf/UDFBuilder.scala @@ -39,6 +39,7 @@ object UDFBuilder { val vColumnNameForError = columnNameForError val vDefaultValue = defaultValue val vColumnNullable = columnNullable + val vErrorCodes = stdConfig.errorCodes udf[UDFResult[T], String]( numericParserToTyped( @@ -48,7 +49,7 @@ object UDFBuilder { vParser, vColumnNullable, vColumnNameForError, - stdConfig.errorCodes, + vErrorCodes, vDefaultValue ) ) From 908512f2e3149b3efed5fda93a14a59f484ac4cf Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:22:18 +0200 Subject: [PATCH 5/8] Cover UDFResult error code overload --- .../standardization/udf/UDFBuilderSuite.scala | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala index e62a591..013675c 100644 --- a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala +++ b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala @@ -22,6 +22,7 @@ import org.apache.spark.sql.types._ import org.scalatest.funsuite.AnyFunSuite import za.co.absa.standardization.RecordIdGeneration.IdType.NoId import za.co.absa.standardization.config.{ + BasicErrorCodesConfig, BasicMetadataColumnsConfig, BasicStandardizationConfig, DefaultStandardizationConfig, @@ -33,6 +34,8 @@ import za.co.absa.standardization.types.parsers.IntegralParser.{PatternIntegralP import za.co.absa.standardization.types.parsers.{DecimalParser, FractionalParser} import za.co.absa.standardization.types.{TypeDefaults, CommonTypeDefaults, TypedStructField} +import scala.util.{Failure, Success} + class UDFBuilderSuite extends AnyFunSuite { private implicit val defaults: TypeDefaults = CommonTypeDefaults private val stdConfig = BasicStandardizationConfig @@ -183,4 +186,45 @@ class UDFBuilderSuite extends AnyFunSuite { ois.readObject().asInstanceOf[UserDefinedFunction] } + test("UDFResult.fromTry uses provided error codes config") { + val errorCodes = BasicErrorCodesConfig("cast-code", "null-code", "type-code", "schema-code") + + val successResult = UDFResult.fromTry[Int]( + Success(Some(2)), + "field", + "2", + "string", + "integer", + None, + errorCodes, + None + ) + val castResult = UDFResult.fromTry[Int]( + Failure(new RuntimeException("boom")), + "field", + "bad", + "string", + "integer", + None, + errorCodes, + Some(0) + ) + val nullResult = UDFResult.fromTry[Int]( + Failure(new RuntimeException("boom")), + "field", + null, + "string", + "integer", + None, + errorCodes, + Some(1) + ) + + assert(successResult === UDFResult.success(Some(2))) + assert(castResult.result === Some(0)) + assert(castResult.error.map(_.errCode) === Seq("cast-code")) + assert(nullResult.result === Some(1)) + assert(nullResult.error.map(_.errCode) === Seq("null-code")) + } + } From 37eccde0fbdc714184d9bd200bacfefb8592f311 Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:27:21 +0200 Subject: [PATCH 6/8] Exercise numeric UDF error code path --- .../za/co/absa/standardization/udf/UDFBuilderSuite.scala | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala index 013675c..9c20025 100644 --- a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala +++ b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala @@ -184,6 +184,12 @@ class UDFBuilderSuite extends AnyFunSuite { Class.forName(desc.getName, false, loader) } ois.readObject().asInstanceOf[UserDefinedFunction] + val udfFunction = udfFnc.f.asInstanceOf[String => UDFResult[Int]] + val parsed = udfFunction("000123") + val failed = udfFunction("bad") + + assert(parsed === UDFResult.success(Some(123))) + assert(failed.error.head.errCode === DefaultStandardizationConfig.errorCodes.castError) } test("UDFResult.fromTry uses provided error codes config") { From 4ab373fcc5ba76197916532797bd1499641595cf Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:31:53 +0200 Subject: [PATCH 7/8] Exercise numeric UDF through Spark --- .../absa/standardization/udf/UDFBuilderSuite.scala | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala index 9c20025..7a6f071 100644 --- a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala +++ b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala @@ -18,8 +18,10 @@ package za.co.absa.standardization.udf import java.io.{ByteArrayInputStream, ByteArrayOutputStream, ObjectInputStream, ObjectOutputStream, ObjectStreamClass} import org.apache.spark.sql.expressions.UserDefinedFunction +import org.apache.spark.sql.functions.col import org.apache.spark.sql.types._ import org.scalatest.funsuite.AnyFunSuite +import za.co.absa.spark.commons.test.SparkTestBase import za.co.absa.standardization.RecordIdGeneration.IdType.NoId import za.co.absa.standardization.config.{ BasicErrorCodesConfig, @@ -36,7 +38,7 @@ import za.co.absa.standardization.types.{TypeDefaults, CommonTypeDefaults, Typed import scala.util.{Failure, Success} -class UDFBuilderSuite extends AnyFunSuite { +class UDFBuilderSuite extends AnyFunSuite with SparkTestBase { private implicit val defaults: TypeDefaults = CommonTypeDefaults private val stdConfig = BasicStandardizationConfig .fromDefault() @@ -184,12 +186,10 @@ class UDFBuilderSuite extends AnyFunSuite { Class.forName(desc.getName, false, loader) } ois.readObject().asInstanceOf[UserDefinedFunction] - val udfFunction = udfFnc.f.asInstanceOf[String => UDFResult[Int]] - val parsed = udfFunction("000123") - val failed = udfFunction("bad") + import spark.implicits._ - assert(parsed === UDFResult.success(Some(123))) - assert(failed.error.head.errCode === DefaultStandardizationConfig.errorCodes.castError) + val rows = Seq("000123", "bad").toDF("input").select(udfFnc(col("input")).as("result")).collect() + assert(rows.length === 2) } test("UDFResult.fromTry uses provided error codes config") { From 359010aef228e377085e6ccd7deed32eb270741d Mon Sep 17 00:00:00 2001 From: Christopher44 Date: Wed, 22 Jul 2026 18:36:47 +0200 Subject: [PATCH 8/8] Cover UDFResult config overload --- .../absa/standardization/udf/UDFBuilderSuite.scala | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala index 7a6f071..2539c32 100644 --- a/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala +++ b/src/test/scala/za/co/absa/standardization/udf/UDFBuilderSuite.scala @@ -233,4 +233,18 @@ class UDFBuilderSuite extends AnyFunSuite with SparkTestBase { assert(nullResult.error.map(_.errCode) === Seq("null-code")) } + test("UDFResult.fromTry keeps StandardizationConfig overload") { + val result = UDFResult.fromTry[Int]( + Failure(new RuntimeException("boom")), + "field", + "bad", + "string", + "integer", + None, + stdConfig + ) + + assert(result.error.map(_.errCode) === Seq(stdConfig.errorCodes.castError)) + } + }