bugfix for not null field removal

snowplow · Feb 20, 2023 · b1173ca · b1173ca
1 parent fc2ce93
commit b1173ca
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 8 deletions.
diff --git a/modules/core/src/main/scala/com.snowplowanalytics/iglu.schemaddl/redshift/ShredModel.scala b/modules/core/src/main/scala/com.snowplowanalytics/iglu.schemaddl/redshift/ShredModel.scala
@@ -108,21 +108,28 @@ object ShredModel {
      *         perform a merge.
      */
     def merge(that: GoodModel): Either[RecoveryModel, GoodModel] = {
-      val baseLookup = entries.map { e => (e.columnName, e) }.toMap
+      val thisLookup = entries.map { e => (e.columnName, e) }.toMap
+      val thatLookup = that.entries.map { e => (e.columnName, e) }.toMap
       val additions: List[ShredModelEntry] =
         that.entries
-          .filter(col => !baseLookup.contains(col.columnName))
+          .filter(col => !thisLookup.contains(col.columnName))
           .map(entry => (entry.ptr, entry.subSchema))
           .foldLeft(Set.empty[(Pointer.SchemaPointer, Schema)])((acc, s) => acc + s)
           // this fold, toList preserves the order as it was in the older library versions < 0.18.0
           .toList
           .map { case (ptr, subSchema) => ShredModelEntry(ptr, subSchema, isLateAddition = true) }
       val additionsMigration: List[ColumnAddition] = additions.map(ColumnAddition.apply)
+      val removals: Either[NonEmptyList[Breaking], List[NonBreaking]] = entries
+        .filter(col => !thatLookup.contains(col.columnName))
+        .parTraverse {
+          case s if !s.isNullable => NullableRequired(s).asLeft.toEitherNel
+          case _ => NoChanges.asRight
+        }
       val modifications: Either[NonEmptyList[Breaking], List[NonBreaking]] =
         that.entries
-          .filter(col => baseLookup.contains(col.columnName))
+          .filter(col => thisLookup.contains(col.columnName))
           .parTraverse(newCol => {
-            val oldCol = baseLookup(newCol.columnName)
+            val oldCol = thisLookup(newCol.columnName)
             val (newType, newNullability, newEncoding) = (newCol.columnType, newCol.isNullable, newCol.compressionEncoding)
             val (oldType, oldNullability, oldEncoding) = (oldCol.columnType, oldCol.isNullable, oldCol.compressionEncoding)
             if (!oldNullability & newNullability)
@@ -139,10 +146,15 @@ object ShredModel {
               case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
             }
           })
-
+      val allChanges: Either[NonEmptyList[Breaking], List[NonBreaking]] = (modifications, removals) match {
+        case (Right(x), Right(y)) => (x ++ y).asRight[NonEmptyList[Breaking]]
+        case (Right(_), l@Left(_)) => l
+        case (l@Left(_), Right(_)) => l
+        case (Left(x), Left(y)) => (x ::: y).asLeft[List[NonBreaking]]
+      }
       (for {
-        mods <- modifications
-        extensions = mods.collect { case e: VarcharExtension => e }
+        changes <- allChanges
+        extensions = changes.collect { case e: VarcharExtension => e }
         modifedEntries = entries.map(
           entry => extensions.collectFirst {
             case s if s.old == entry => s.newEntry
@@ -169,7 +181,6 @@ object ShredModel {
     val tableName = s"${baseTableName}_${schemaKey.version.addition}_${schemaKey.version.revision}_recovered_${abs(entries.show.hashCode())}"
   }
 
-
   def good(s: IgluSchema): GoodModel = good(s.self.schemaKey, s.schema)
 
   def good(k: SchemaKey, s: Schema): GoodModel = new GoodModel(FlatSchema.extractProperties(s), k, Migrations.empty(k))

diff --git a/...es/core/src/test/scala/com/snowplowanalytics/iglu/schemaddl/redshift/ShredModelSpec.scala b/...es/core/src/test/scala/com/snowplowanalytics/iglu/schemaddl/redshift/ShredModelSpec.scala
@@ -251,6 +251,51 @@ class ShredModelSpec extends Specification {
       )
     }
 
+    "should make a recovery model when not null field is removed" in {
+      val s1 = ShredModel.good(dummyKey,
+        json"""{
+               "type": "object",
+               "properties": {
+                 "foo": {
+                   "type": "string",
+                   "maxLength": 20
+                 }},
+              "required": ["foo"]
+              }""".schema)
+      val s2 = ShredModel.good(dummyKey1,
+        json"""{
+               "type": "object",
+               "properties": {
+                 "foo1": {
+                   "type": "number"
+                 }}
+              }""".schema)
+
+      s1.merge(s2).toTestString must beLeft(
+        """CREATE TABLE IF NOT EXISTS s.com_acme_example_1_1_0_recovered_1202144068 (
+          |  "schema_vendor"  VARCHAR(128)     ENCODE ZSTD NOT NULL,
+          |  "schema_name"    VARCHAR(128)     ENCODE ZSTD NOT NULL,
+          |  "schema_format"  VARCHAR(128)     ENCODE ZSTD NOT NULL,
+          |  "schema_version" VARCHAR(128)     ENCODE ZSTD NOT NULL,
+          |  "root_id"        CHAR(36)         ENCODE RAW  NOT NULL,
+          |  "root_tstamp"    TIMESTAMP        ENCODE ZSTD NOT NULL,
+          |  "ref_root"       VARCHAR(255)     ENCODE ZSTD NOT NULL,
+          |  "ref_tree"       VARCHAR(1500)    ENCODE ZSTD NOT NULL,
+          |  "ref_parent"     VARCHAR(255)     ENCODE ZSTD NOT NULL,
+          |  "foo1"           DOUBLE PRECISION ENCODE RAW,
+          |  FOREIGN KEY (root_id) REFERENCES s.events(event_id)
+          |)
+          |DISTSTYLE KEY
+          |DISTKEY (root_id)
+          |SORTKEY (root_tstamp);
+          |
+          |COMMENT ON TABLE s.com_acme_example_1_1_0_recovered_1202144068 IS 'iglu:com.acme/example/jsonschema/1-0-1';
+          |
+          |Making required column nullable foo""".stripMargin
+      )
+    }
+
+
     "should make a ignore varchar narrowing" in {
       val s1 = ShredModel.good(dummyKey,
         json"""{