#126: Add support for Map type

src/main/scala/za/co/absa/standardization/SchemaValidator.scala

@@ -110,6 +110,7 @@ object SchemaValidator {
         field.dataType match {
           case s: StructType => fields ++= flattenStruct(s, prefix + field.name)
           case a: ArrayType => fields ++= flattenArray(field, a, prefix + field.name + "[]")
+          case m: MapType => fields ++= flattenMap(field, m, prefix + field.name + "[]")
           case _ =>
             val prefixedField = FlatField(structPath, field)
             fields += prefixedField
@@ -123,13 +124,27 @@ object SchemaValidator {
       arr.elementType match {
         case stuctInArray: StructType => arrayFields ++= flattenStruct(stuctInArray, structPath)
         case arrayType: ArrayType => arrayFields ++= flattenArray(field, arrayType, structPath + "[]")
+        case mapType: MapType => arrayFields ++= flattenMap(field, mapType, structPath + "[]")
         case _ =>
           val prefixedField = FlatField(structPath, field)
           arrayFields += prefixedField
       }
       arrayFields.toSeq //has to be here for Scala 2.13 compatibility
     }
 
+    def flattenMap(field: StructField, map: MapType, structPath: String): Seq[FlatField] = {
+      val mapFields = new ListBuffer[FlatField]
+      map.valueType match {
+        case structInMap: StructType => mapFields ++= flattenStruct(structInMap, structPath)
+        case arrayType: ArrayType => mapFields ++= flattenArray(field, arrayType, structPath + "[]")
+        case mapType: MapType => mapFields ++= flattenMap(field, mapType, structPath + "[]")
+        case _ =>
+          val prefixedField = FlatField(structPath, field)
+          mapFields += prefixedField
+      }
+      mapFields.toSeq //has to be here for Scala 2.13 compatibility
+    }
+
     flattenStruct(schema, "")
   }
 

src/main/scala/za/co/absa/standardization/stages/PlainSchemaGenerator.scala

@@ -40,6 +40,7 @@ object PlainSchemaGenerator {
     inp match {
       case st: StructType => StructType(structTypeFieldsConversion(st.fields))
       case at: ArrayType  => ArrayType(inputSchemaAsStringTypes(at.elementType), containsNull = true)
+      case mt: MapType    => MapType(mt.keyType, inputSchemaAsStringTypes(mt.valueType), valueContainsNull = true)
       case _: DataType    => StringType
     }
   }

src/main/scala/za/co/absa/standardization/stages/TypeParser.scala

@@ -49,6 +49,7 @@ import scala.util.{Random, Try}
   * Class hierarchy:
   *   TypeParser
   *     ArrayParser !
+  *     MapParser !
   *     StructParser !
   *     PrimitiveParser
   *       ScalarParser
@@ -106,9 +107,13 @@ sealed trait TypeParser[T] {
 
     (origType, fieldType) match {
       case (ArrayType(_, _), ArrayType(_, _)) => None
+      case (origMapType: MapType, targetMapType: MapType) if origMapType.keyType == targetMapType.keyType => None
+      case (_: MapType, _: MapType) => noCastingPossible
       case (StructType(_), StructType(_)) => None
       case (ArrayType(_, _), _) => noCastingPossible
       case (_, ArrayType(_, _)) => noCastingPossible
+      case (MapType(_, _, _), _) => noCastingPossible
+      case (_, MapType(_, _, _)) => noCastingPossible
       case (StructType(_), _) => noCastingPossible
       case (_, StructType(_)) => noCastingPossible
       case _ => None
@@ -166,6 +171,7 @@ object TypeParser {
                    (implicit defaults: TypeDefaults): TypeParser[_] = {
     val parserClass: (String, Column, DataType, Boolean, Boolean) => TypeParser[_] = field.dataType match {
       case _: ArrayType     => ArrayParser(TypedStructField.asArrayTypeStructField(field), _, _, _, _, _)
+      case _: MapType       => MapParser(TypedStructField.asMapTypeStructField(field), _, _, _, _, _)
       case _: StructType    => StructParser(TypedStructField.asStructTypeStructField(field), _, _, _, _, _)
       case _: ByteType      =>
         IntegralParser(TypedStructField.asNumericTypeStructField[Byte](field), _, _, _, _, _, Set(ShortType, IntegerType, LongType))
@@ -221,6 +227,44 @@ object TypeParser {
     }
   }
 
+  private final case class MapParser(override val field: MapTypeStructField,
+                                     path: String,
+                                     column: Column,
+                                     origType: DataType,
+                                     failOnInputNotPerSchema: Boolean,
+                                     isArrayElement: Boolean)
+                                    (implicit defaults: TypeDefaults) extends TypeParser[Any] {
+
+    override def fieldType: MapType = {
+      field.dataType
+    }
+
+    override protected def standardizeAfterCheck(stdConfig: StandardizationConfig)(implicit logger: Logger): ParseOutput = {
+      logger.info(s"Creating standardization plan for Map $inputFullPathName")
+      val origMapType = origType.asInstanceOf[MapType] // this should never throw an exception because of `checkSetupForFailure`
+      val valueField = StructField(fieldInputName, fieldType.valueType, fieldType.valueContainsNull, field.structField.metadata)
+      val lambda = (forCol: Column) => TypeParser(valueField, path, forCol, origMapType.valueType, failOnInputNotPerSchema, isArrayElement = true)
+        .standardize(stdConfig)
+
+      val lambdaErrCols = lambda.andThen(_.errors)
+      val lambdaStdCols = lambda.andThen(_.stdCol)
+      val nullErrCond = column.isNull and lit(!field.nullable)
+
+      val valueErrs = map_values(transform_values(column, (_: Column, valueCol: Column) => lambdaErrCols(valueCol)))
+      val finalErrs = when(nullErrCond,
+        array(typedLit(StandardizationErrorMessage.stdNullErr(inputFullPathName)(stdConfig.errorCodes))))
+        .when(column.isNull,
+          typedLit(Seq.empty[ErrorMessage])
+        )
+        .otherwise(
+          flatten(valueErrs)
+        )
+      val stdCols = transform_values(column, (_: Column, valueCol: Column) => lambdaStdCols(valueCol))
+      logger.info(s"Finished standardization plan for Map $inputFullPathName")
+      ParseOutput(stdCols as (fieldOutputName, metadata), finalErrs)
+    }
+  }
+
   private final case class StructParser(override val field: StructTypeStructField,
                                         path: String,
                                         column: Column,

src/main/scala/za/co/absa/standardization/types/TypedStructField.scala

@@ -148,6 +148,7 @@ object TypedStructField {
       case _: TimestampType => new TimestampTypeStructField(structField)
       case _: DateType      => new DateTypeStructField(structField)
       case at: ArrayType    => new ArrayTypeStructField(structField, at)
+      case mt: MapType      => new MapTypeStructField(structField, mt)
       case st: StructType   => new StructTypeStructField(structField, st)
       case _                => new GeneralTypeStructField(structField)
     }
@@ -159,6 +160,8 @@ object TypedStructField {
     TypedStructField(structField).asInstanceOf[DateTimeTypeStructField[T]]
   def asArrayTypeStructField(structField: StructField)(implicit defaults: TypeDefaults): ArrayTypeStructField =
     TypedStructField(structField).asInstanceOf[ArrayTypeStructField]
+  def asMapTypeStructField(structField: StructField)(implicit defaults: TypeDefaults): MapTypeStructField =
+    TypedStructField(structField).asInstanceOf[MapTypeStructField]
   def asStructTypeStructField(structField: StructField)(implicit defaults: TypeDefaults): StructTypeStructField =
     TypedStructField(structField).asInstanceOf[StructTypeStructField]
   def asBinaryTypeStructField(structField: StructField)(implicit defaults: TypeDefaults): BinaryTypeStructField =
@@ -474,6 +477,24 @@ object TypedStructField {
     }
   }
 
+  final class MapTypeStructField private[TypedStructField](structField: StructField, override val dataType: MapType)
+                                                           (implicit defaults: TypeDefaults)
+    extends TypedStructFieldTagged[Any](structField) with WeakSupport[Any] {
+
+    override def validate(): Seq[ValidationIssue] = {
+      val typedValueField = Try {
+        val valueField = StructField(name, dataType.valueType, dataType.valueContainsNull, structField.metadata)
+        TypedStructField(valueField)
+      }
+
+      super.validate() ++ FieldValidator.tryToValidationIssues(typedValueField.map(_.validate()))
+    }
+
+    override def ownDefaultValue: Try[Option[Option[Any]]] = {
+      Success(None) // map type doesn't have own default value, if defined it is to be applied to value type
+    }
+  }
+
   final class StructTypeStructField(structField: StructField, override val dataType: StructType)(implicit defaults: TypeDefaults)
     extends TypedStructFieldTagged[Any](structField) with WeakSupport[Any]
 

src/test/scala/za/co/absa/standardization/SchemaValidationSuite.scala

@@ -225,4 +225,28 @@ class SchemaValidationSuite extends AnyFunSuite with LoggerTestBase {
     assert(failures(3).fieldName == "matrix[][].foo.bar")
   }
 
+  test("Nested map fields should be validated") {
+    val schema = StructType(
+      Array(
+        StructField("id", StringType),
+        StructField("dates", MapType(StringType, DateType), nullable = true, Metadata.fromJson(""" { "pattern": "Fubar" } """)),
+        StructField("payments", MapType(StringType, StructType(Array(
+          StructField("due", DateType, nullable = false, Metadata.fromJson(""" { "pattern": "dd-MM-yyyy'T'HH:mm:ss" } """)),
+          StructField("made", TimestampType, nullable = false, Metadata.fromJson(""" { "pattern": "dd-MM-yyyy" } """))
+        ))))
+      ))
+    val failures = SchemaValidator.validateSchema(schema)(CommonTypeDefaults)
+    if (failures.lengthCompare(3) != 0) {
+      logger.error("Schema:\n")
+      logger.error(schema.prettyJson)
+      logger.error("Validation errors:")
+      logger.error(failures.mkString("\n"))
+    }
+
+    assert(failures.lengthCompare(3) == 0)
+    assert(failures(0).fieldName == "dates[].dates")
+    assert(failures(1).fieldName == "payments[].due")
+    assert(failures(2).fieldName == "payments[].made")
+  }
+
 }

src/test/scala/za/co/absa/standardization/interpreter/StandardizationInterpreter_MapSuite.scala

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed 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 za.co.absa.standardization.interpreter
+
+import com.github.mrpowers.spark.fast.tests.DatasetComparer
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types._
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import za.co.absa.spark.commons.implicits.DataFrameImplicits.DataFrameEnhancements
+import za.co.absa.spark.commons.test.SparkTestBase
+import za.co.absa.standardization.DataFrameTestUtils.RowSeqToDf
+import za.co.absa.standardization.RecordIdGeneration.IdType.NoId
+import za.co.absa.standardization.config.{BasicMetadataColumnsConfig, BasicStandardizationConfig}
+import za.co.absa.standardization.types.{CommonTypeDefaults, TypeDefaults}
+import za.co.absa.standardization.udf.UDFLibrary
+import za.co.absa.standardization.{ErrorMessageFactory, LoggerTestBase, Standardization}
+
+class StandardizationInterpreter_MapSuite extends AnyFunSuite with SparkTestBase with LoggerTestBase with Matchers with DatasetComparer {
+  private val stdConfig = BasicStandardizationConfig
+    .fromDefault()
+    .copy(metadataColumns = BasicMetadataColumnsConfig
+      .fromDefault()
+      .copy(recordIdStrategy = NoId
+      )
+    )
+  private implicit val udfLib: UDFLibrary = new UDFLibrary(stdConfig)
+  private implicit val defaults: TypeDefaults = CommonTypeDefaults
+
+  private val fieldName = "mapField"
+
+  test("Map of integers") {
+    val sourceSchema = StructType(Seq(
+      StructField(fieldName, MapType(StringType, StringType, valueContainsNull = true), nullable = true)
+    ))
+    val sourceData = Seq(
+      Row(Map("one" -> "1", "two" -> "2")),
+      Row(Map("good" -> "7", "bad" -> "x")),
+      Row(null) // scalastyle:ignore null
+    )
+    val src = spark.createDataFrame(spark.sparkContext.parallelize(sourceData), sourceSchema)
+
+    val desiredSchema = StructType(Seq(
+      StructField(fieldName, MapType(StringType, IntegerType, valueContainsNull = true), nullable = true)
+    ))
+
+    val expectedSchema = ErrorMessageFactory.attachErrColToSchemaPrint(
+      "root\n" +
+        " |-- mapField: map (nullable = true)\n" +
+        " |    |-- key: string\n" +
+        " |    |-- value: integer (valueContainsNull = true)"
+    )
+
+    val stdDF = Standardization.standardize(src, desiredSchema, stdConfig).cacheIfNotCachedYet()
+    assert(stdDF.schema.treeString == expectedSchema)
+
+    val expectedData = Seq(
+      Row(Map("one" -> 1, "two" -> 2), Seq()),
+      Row(Map("good" -> 7, "bad" -> null), Seq(
+        Row("stdCastError", "E00000", "Cast from 'string' to 'integer'", "mapField[*]", Seq("x"), Seq())
+      )),
+      Row(null, Seq()) // scalastyle:ignore null
+    )
+    val expectedDF = expectedData.toDfWithSchema(stdDF.schema)
+    assertSmallDatasetEquality(stdDF, expectedDF)
+  }
+}
+

src/test/scala/za/co/absa/standardization/interpreter/stages/PlainSchemaGeneratorSuite.scala

@@ -34,7 +34,12 @@ class PlainSchemaGeneratorSuite extends AnyFunSuite with SparkTestBase {
           StructField("i", IntegerType, nullable = false, new MetadataBuilder().putString("sourcecolumn", "override_i").build)
         ))))
       )))
-    ))))
+    )))),
+    StructField("j", MapType(StringType, IntegerType, valueContainsNull = false)),
+    StructField("k", MapType(StringType, StructType(Seq(
+      StructField("l", IntegerType, nullable = false),
+      StructField("m", IntegerType, nullable = false, new MetadataBuilder().putString("sourcecolumn", "override_m").build)
+    )), valueContainsNull = false))
   ))
 
   private val expectedSchemaSeq = Seq(
@@ -49,7 +54,12 @@ class PlainSchemaGeneratorSuite extends AnyFunSuite with SparkTestBase {
           StructField("override_i", StringType, nullable = true, new MetadataBuilder().putString("sourcecolumn", "override_i").build)
         ))))
       )))
-    ))))
+    )))),
+    StructField("j", MapType(StringType, StringType, valueContainsNull = true), nullable = true),
+    StructField("k", MapType(StringType, StructType(Seq(
+      StructField("l", StringType, nullable = true),
+      StructField("override_m", StringType, nullable = true, new MetadataBuilder().putString("sourcecolumn", "override_m").build)
+    )), valueContainsNull = true), nullable = true)
   )
 
   private val expectedSchema = StructType(expectedSchemaSeq)

src/test/scala/za/co/absa/standardization/types/TypedStructFieldSuite.scala

@@ -92,6 +92,7 @@ class TypedStructFieldSuite extends AnyFunSuite {
       case TimestampType  => (field.isInstanceOf[TimestampTypeStructField], "TimestampTypeStructField")
       case _: DecimalType => (field.isInstanceOf[DecimalTypeStructField], "DecimalTypeStructField")
       case _: ArrayType   => (field.isInstanceOf[ArrayTypeStructField], "ArrayTypeStructField")
+      case _: MapType     => (field.isInstanceOf[MapTypeStructField], "MapTypeStructField")
       case _: StructType  => (field.isInstanceOf[StructTypeStructField], "StructTypeStructField")
       case _ => (field.isInstanceOf[GeneralTypeStructField], "GeneralTypeStructField")
     }
@@ -289,4 +290,21 @@ class TypedStructFieldSuite extends AnyFunSuite {
     val errMsg = "Illegal pattern character 'b'"
     checkField(typed, fieldType, Success(None), Success(None), nullable, Seq(ValidationError(errMsg)))
   }
+
+  test("Map type with long value data type and correct associated metadata") {
+    val fieldType = MapType(StringType, LongType)
+    val nullable = true
+    val field = createField(fieldType, nullable, Some("9999"), Map( MetadataKeys.MinusSign->"$"))
+    val typed = TypedStructField(field)
+    checkField(typed, fieldType, Success(None), Success(None), nullable)
+  }
+
+  test("Map type with date value data type and incorrect associated metadata") {
+    val fieldType = MapType(StringType, DateType)
+    val nullable = true
+    val field = createField(fieldType, nullable, None, Map( MetadataKeys.Pattern->"Fubar"))
+    val typed = TypedStructField(field)
+    val errMsg = "Illegal pattern character 'b'"
+    checkField(typed, fieldType, Success(None), Success(None), nullable, Seq(ValidationError(errMsg)))
+  }
 }