fix: check fields by name when we can

In some cases we do not care about the order of the fields
as long as the name and type matches.

core/src/main/scala/no/nrk/bigquery/conforms.scala

@@ -32,7 +32,7 @@ object conforms {
       case other => BQSchema.of(other)
     }
 
-    apply(
+    onlyTypes(
       actualSchema.copy(fields = actualSchema.fields.map(anonymize)),
       givenSchema
     )
@@ -60,11 +60,10 @@ object conforms {
       case other => BQSchema.of(other)
     }
 
-    apply(actualSchema, givenSchema)
+    typesAndName(actualSchema, givenSchema)
   }
 
-  /* this is a stronger comparison than `onlyTypes`, because it takes into column names as well */
-  def apply(
+  def onlyTypes(
       actualSchema: BQSchema,
       givenSchema: BQSchema
   ): Option[List[String]] = {
@@ -105,4 +104,45 @@ object conforms {
       case reasons => Some(reasons)
     }
   }
+
+  /** this is a stronger comparison than `onlyTypes`, because it takes into column names as well */
+  def typesAndName(
+      actualSchema: BQSchema,
+      givenSchema: BQSchema
+  ): Option[List[String]] = {
+    val reasonsBuilder = List.newBuilder[String]
+
+    def go(
+        path: List[BQField],
+        actualFields: Seq[BQField],
+        givenFields: Seq[BQField]
+    ): Unit =
+      actualFields.foreach { actualField =>
+        val givenFieldOpt = givenFields.find(_.name == actualField.name)
+
+        // if we're inside structs, render the full path
+        def render(f: BQField) =
+          s"field `${(f :: path).reverse.map(_.name).mkString(".")}`"
+
+        givenFieldOpt match {
+          case Some(givenField) if givenField.tpe != actualField.tpe =>
+            reasonsBuilder += s"Expected ${render(actualField)} to have type ${actualField.tpe}, got ${givenField.tpe}"
+          case Some(givenField) if (givenField.mode == Mode.REPEATED) != (actualField.mode == Mode.REPEATED) =>
+            reasonsBuilder += s"Expected ${render(actualField)} to have mode ${actualField.mode}, got ${givenField.mode}"
+          case Some(givenField) if givenField.subFields.nonEmpty =>
+            go(givenField :: path, actualField.subFields, givenField.subFields)
+          case Some(ok @ _) =>
+            ()
+          case None =>
+            reasonsBuilder += s"Expected ${render(actualField)} to part of [${givenFields.map(_.name).mkString(", ")}]"
+        }
+      }
+
+    go(Nil, actualSchema.fields, givenSchema.fields)
+
+    reasonsBuilder.result() match {
+      case Nil => None
+      case reasons => Some(reasons)
+    }
+  }
 }

core/src/main/scala/no/nrk/bigquery/internal/TableUpdateOperation.scala

@@ -39,15 +39,17 @@ object TableUpdateOperation {
                 )
 
                 val illegalSchemaExtension: Option[UpdateOperation.IllegalSchemaExtension] =
-                  conforms(
-                    actualSchema = remoteAsTableDef.schema,
-                    givenSchema = localTableDef.schema
-                  ).map { reasons =>
-                    UpdateOperation.IllegalSchemaExtension(
-                      TableDefOperationMeta(existingRemoteTable, tableDef),
-                      reasons.mkString(", ")
+                  conforms
+                    .onlyTypes(
+                      actualSchema = remoteAsTableDef.schema,
+                      givenSchema = localTableDef.schema
                     )
-                  }
+                    .map { reasons =>
+                      UpdateOperation.IllegalSchemaExtension(
+                        TableDefOperationMeta(existingRemoteTable, tableDef),
+                        reasons.mkString(", ")
+                      )
+                    }
 
                 if (localTableDef == remoteAsTableDef)
                   UpdateOperation.Noop(TableDefOperationMeta(existingRemoteTable, localTableDef))

core/src/test/scala/no/nrk/bigquery/conformsTest.scala

@@ -0,0 +1,63 @@
+package no.nrk.bigquery
+
+import com.google.cloud.bigquery.{Field, StandardSQLTypeName}
+import munit.FunSuite
+
+class conformsTest extends FunSuite {
+  test("failing conform by type position") {
+    val violations = conforms.onlyTypes(
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      ),
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.INT64, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.STRING, Field.Mode.NULLABLE)
+      )
+    )
+    assertEquals(violations.map(_.length), Some(2), clue(violations.map(_.mkString(", "))))
+  }
+
+  test("passing conform by type position") {
+    val violations = conforms.onlyTypes(
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      ),
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      )
+    )
+    assertEquals(violations.map(_.length), None, clue(violations.map(_.mkString(", "))))
+  }
+
+  test("passing conform by name and type") {
+    val violations = conforms.typesAndName(
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      ),
+      BQSchema.of(
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE),
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE)
+      )
+    )
+    assertEquals(violations.map(_.length), None, clue(violations.map(_.mkString(", "))))
+  }
+
+  test("failing conform by name and type") {
+    val violations = conforms.typesAndName(
+      BQSchema.of(
+        BQField("one", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("two", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      ),
+      BQSchema.of(
+        BQField("two", StandardSQLTypeName.STRING, Field.Mode.NULLABLE),
+        BQField("one", StandardSQLTypeName.INT64, Field.Mode.NULLABLE)
+      )
+    )
+    assertEquals(violations.map(_.length), Some(2), clue(violations.map(_.mkString(", "))))
+  }
+
+}

testing/src/main/scala/no/nrk/bigquery/testing/BQSmokeTest.scala

@@ -291,7 +291,7 @@ object BQSmokeTest {
     def checkSchema(actualSchema: BQSchema): Unit =
       this match {
         case CheckType.Schema(expectedSchema) =>
-          conforms(actualSchema, expectedSchema) match {
+          conforms.onlyTypes(actualSchema, expectedSchema) match {
             case Some(reasons) =>
               fail(s"Failed because ${reasons.mkString(", ")}", TypeClue(expectedSchema, actualSchema))
             case None => assert(true)