From b005f4de6ba258ad5e7ed76c743a6ff4a151d5ae Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Mon, 25 May 2026 17:39:08 +0200 Subject: [PATCH 1/2] #100 Add null check when standardizing struct fields. This improves the performance of processing of multi-segment mainframe files. --- .../absa/standardization/stages/TypeParser.scala | 5 ++--- .../StandardizationInterpreterSuite.scala | 14 +++++++------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala b/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala index b5af030..92a2191 100644 --- a/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala +++ b/src/main/scala/za/co/absa/standardization/stages/TypeParser.scala @@ -26,7 +26,6 @@ import za.co.absa.spark.commons.implicits.StructFieldImplicits.StructFieldMetada import za.co.absa.spark.commons.implicits.StructTypeImplicits.StructTypeEnhancements import za.co.absa.spark.commons.utils.SchemaUtils import za.co.absa.spark.hofs.transform -import za.co.absa.standardization.{ErrorMessage, StandardizationErrorMessage} import za.co.absa.standardization.config.StandardizationConfig import za.co.absa.standardization.implicits.StdColumnImplicits.StdColumnEnhancements import za.co.absa.standardization.schema.StdSchemaUtils.FieldWithSource @@ -34,9 +33,9 @@ import za.co.absa.standardization.schema.{MetadataKeys, MetadataValues, StdSchem import za.co.absa.standardization.time.DateTimePattern import za.co.absa.standardization.typeClasses.{DoubleLike, LongLike} import za.co.absa.standardization.types.TypedStructField._ -import za.co.absa.standardization.types.parsers.DateTimeParser import za.co.absa.standardization.types.{ParseOutput, TypeDefaults, TypedStructField} import za.co.absa.standardization.udf.{UDFBuilder, UDFNames} +import za.co.absa.standardization.{ErrorMessage, StandardizationErrorMessage} import java.security.InvalidParameterException import java.sql.Timestamp @@ -257,7 +256,7 @@ object TypeParser { ) ) // rebuild the struct - val outputColumn = struct(cols: _*) as (fieldOutputName, metadata) + val outputColumn = when(column.isNull, lit(null)).otherwise(struct(cols: _*)) as (fieldOutputName, metadata) ParseOutput(outputColumn, errs1) } diff --git a/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala b/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala index bd8d31b..8a60f09 100644 --- a/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala +++ b/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala @@ -23,10 +23,10 @@ import za.co.absa.spark.commons.implicits.DataFrameImplicits.DataFrameEnhancemen import za.co.absa.spark.commons.test.SparkTestBase import za.co.absa.spark.commons.utils.JsonUtils import za.co.absa.standardization.RecordIdGeneration.IdType.NoId -import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig, DefaultStandardizationConfig, ErrorCodesConfig} +import za.co.absa.standardization._ +import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig, ErrorCodesConfig} import za.co.absa.standardization.types.{CommonTypeDefaults, TypeDefaults} import za.co.absa.standardization.udf.UDFLibrary -import za.co.absa.standardization._ import java.sql.{Date, Timestamp} import java.util.TimeZone @@ -97,7 +97,7 @@ class StandardizationInterpreterSuite extends AnyFunSuite with SparkTestBase wit StructType( Seq( StructField("subFieldA", IntegerType, nullable = true), - StructField("subFieldB", StringType, nullable = true))), nullable = false), + StructField("subFieldB", StringType, nullable = true))), nullable = true), StructField("rootStruct2", StructType( Seq( @@ -105,7 +105,7 @@ class StandardizationInterpreterSuite extends AnyFunSuite with SparkTestBase wit StructType( Seq( StructField("subSub2FieldA", IntegerType, nullable = true), - StructField("subSub2FieldB", StringType, nullable = true))), nullable = false))), nullable = false), + StructField("subSub2FieldB", StringType, nullable = true))), nullable = true))), nullable = true), StructField("rootArray", ArrayType( StructType( @@ -116,7 +116,7 @@ class StandardizationInterpreterSuite extends AnyFunSuite with SparkTestBase wit StructType( Seq( StructField("subFieldA", IntegerType, nullable = true), - StructField("subFieldB", StringType, nullable = true))), nullable = false))), containsNull = false + StructField("subFieldB", StringType, nullable = true))), nullable = true))), containsNull = true )))) test("Non-null errors produced for non-nullable attribute in a struct") { @@ -357,10 +357,10 @@ class StandardizationInterpreterSuite extends AnyFunSuite with SparkTestBase wit val expectedSchema = "root\n" + " |-- first_name: string (nullable = true)\n" + " |-- last_name: string (nullable = true)\n" + - " |-- body_stats: struct (nullable = false)\n" + + " |-- body_stats: struct (nullable = true)\n" + " | |-- height: integer (nullable = true)\n" + " | |-- weight: integer (nullable = true)\n" + - " | |-- miscellaneous: struct (nullable = false)\n" + + " | |-- miscellaneous: struct (nullable = true)\n" + " | | |-- eye_color: string (nullable = true)\n" + " | | |-- glasses: boolean (nullable = true)\n" + " | |-- temperature_measurements: array (nullable = true)\n" + From e25a567ff848c709af05b437d7d52520448efadc Mon Sep 17 00:00:00 2001 From: Ruslan Iushchenko Date: Tue, 26 May 2026 16:54:04 +0200 Subject: [PATCH 2/2] #100 Add test for null struct preservation during standardization. --- .../StandardizationInterpreterSuite.scala | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala b/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala index 8a60f09..46fa097 100644 --- a/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala +++ b/src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreterSuite.scala @@ -49,6 +49,9 @@ object StandardizationInterpreterSuite { case class subarrayCC(arrayFieldA: Integer, arrayFieldB: String, arrayStruct: subCC) case class rootCC(rootField: String, rootStruct: subCC, rootStruct2: sub1CC, rootArray: Array[subarrayCC]) + case class InnerStruct(innerA: String, innerB: Integer) + case class NullStructRecord(id: Long, name: Option[String], details: Option[InnerStruct]) + // used by the last test: // cannot use case class as the field names contain spaces therefore cast will happen into tuple type BodyStats = (Int, Int, (String, Option[Boolean]), Seq[Double]) @@ -225,6 +228,37 @@ class StandardizationInterpreterSuite extends AnyFunSuite with SparkTestBase wit assert(count == 0) } + test("Test standardization preserves null for a nullable nested struct") { + val schema = StructType( + Array( + StructField("id", LongType, nullable = false), + StructField("name", StringType, nullable = true), + StructField("details", StructType(Array( + StructField("innerA", StringType, nullable = false), + StructField("innerB", IntegerType, nullable = false) + )), nullable = true))) + + val sourceDF = spark.createDataFrame( + Array( + NullStructRecord(1, Some("Record 1"), Some(InnerStruct("hello", 42))), + NullStructRecord(2, Some("Record 2"), None), + NullStructRecord(3, None, None) + )) + + val standardizedDF = Standardization.standardize(sourceDF, schema, stdConfig) + + // No errors should be produced since the struct is nullable + val errorCount = standardizedDF.where(size(col("errCol")) > 0).count() + assert(errorCount == 0) + + // Verify that null struct values remain null in the output + val dfStandardized = standardizedDF.orderBy("id") + val detailsValues = dfStandardized.select("details").collect() + assert(detailsValues(0).getStruct(0) != null, "Record 1 should have a non-null details struct") + assert(detailsValues(1).isNullAt(0), "Record 2 should have a null details struct in the output") + assert(detailsValues(2).isNullAt(0), "Record 3 should have a null details struct in the output") + } + test ("Test standardization of Date and Timestamp fields with default value and pattern") { val schema = StructType( Seq(