Allow type narrowing from VARCHAR to other types in redshift (#184)

Allow type narrowing from VARCHAR to other types in redshift (close #183)
snowplow · Jun 14, 2023 · a67049a · a67049a
1 parent c15fe6d
commit a67049a
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 5 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
@@ -145,29 +145,54 @@ object ShredModel {
                 case ColumnType.RedshiftInteger => NoChanges.asRight
                 case ColumnType.RedshiftBigInt => NoChanges.asRight
                 case ColumnType.RedshiftDouble => NoChanges.asRight
-                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(0) - scale.getOrElse(0) > 5 => NoChanges.asRight
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 5 => NoChanges.asRight
+                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(18) - scale.getOrElse(0) >= 5 => NoChanges.asRight
                 case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
               }
               case ColumnType.RedshiftInteger => oldType match {
                 case ColumnType.RedshiftInteger => NoChanges.asRight
                 case ColumnType.RedshiftBigInt => NoChanges.asRight
                 case ColumnType.RedshiftDouble => NoChanges.asRight
-                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(0) - scale.getOrElse(0) > 10 => NoChanges.asRight
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 10 => NoChanges.asRight
+                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(18) - scale.getOrElse(0) >= 10 => NoChanges.asRight
                 case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
               }
               case ColumnType.RedshiftBigInt => oldType match {
                 case ColumnType.RedshiftBigInt => NoChanges.asRight
                 case ColumnType.RedshiftDouble => NoChanges.asRight
-                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(0) - scale.getOrElse(0) > 19 => NoChanges.asRight
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 19 => NoChanges.asRight
+                case ColumnType.RedshiftDecimal(precision, scale) if precision.getOrElse(18) - scale.getOrElse(0) >= 19 => NoChanges.asRight
                 case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
               }
               case ColumnType.RedshiftDecimal(new_precision, new_scale) => oldType match {
                 case ColumnType.RedshiftDouble => NoChanges.asRight
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= new_precision.getOrElse(18) => NoChanges.asRight
                 case ColumnType.RedshiftDecimal(old_precision, old_scale)
-                  if old_precision.getOrElse(0) >= new_precision.getOrElse(0) &
+                  if old_precision.getOrElse(18) >= new_precision.getOrElse(18) &
                     old_scale.getOrElse(0) >= new_scale.getOrElse(0) => NoChanges.asRight
                 case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
               }
+              case ColumnType.RedshiftChar(newSize) => oldType match {
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= newSize => NoChanges.asRight
+                case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
+              }
+              case ColumnType.RedshiftDate => oldType match {
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 10 => NoChanges.asRight
+                case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
+              }
+              case ColumnType.RedshiftTimestamp => oldType match {
+                // DD-MM-YYYY HH:mm:ss.ttttt
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 25 => NoChanges.asRight
+                case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
+              }
+              case ColumnType.RedshiftBoolean => oldType match {
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= 5 => NoChanges.asRight
+                case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
+              }
+              case ColumnType.RedshiftChar(newSize) => oldType match {
+                case ColumnType.RedshiftVarchar(oldSize) if oldSize >= newSize => NoChanges.asRight
+                case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
+              }
               case _ if newType == oldType => NoChanges.asRight
               case _ => IncompatibleTypes(oldCol, newCol).asLeft.toEitherNel
             }

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
@@ -204,7 +204,7 @@ class ShredModelSpec extends Specification {
               "$$schema": "http://iglucentral.com/schemas/com.snowplowanalytics.self-desc/schema/jsonschema/1-0-0#"
             }""".schema)
 
-      isRedshiftMigrationBreaking(List(s1), s2) must beTrue
+      isRedshiftMigrationBreaking(List(s2), s1) must beTrue
     }
 
     "should make a recovery model when incompatible encodings are merged" in {