[Schema Management] Ignore index changes when comparing schemas (#337)

Author: george-zubrienko

src/main/scala/models/schemas/ArcaneSchema.scala

@@ -39,6 +39,12 @@ trait ArcaneSchemaField:
   val name: String
   val fieldType: ArcaneType
 
+  /** Field comparison that ignored field indexes. Needed to prevent reordering from triggering schema changes.
+    * @return
+    */
+  final def identical(other: ArcaneSchemaField): Boolean =
+    other.name.toLowerCase() == name.toLowerCase() && other.fieldType == fieldType
+
 /** A field in the schema definition that carries index information from the source that can be re-applied when
   * converting to Iceberg
   */
@@ -132,11 +138,11 @@ object ArcaneSchema:
     */
   extension (targetSchema: ArcaneSchema)
     def getMissingFields(sourceSchema: ArcaneSchema): Seq[ArcaneSchemaField] =
-      sourceSchema.filter { sourceField => !targetSchema.contains(sourceField) }
+      sourceSchema.filter { sourceField => !targetSchema.exists(_.identical(sourceField)) }
 
   extension (targetSchema: ArcaneSchema)
     def getMissingFields(sourceFields: Seq[ArcaneSchemaField]): Seq[ArcaneSchemaField] =
-      sourceFields.filter { sourceField => !targetSchema.contains(sourceField) }
+      sourceFields.filter { sourceField => !targetSchema.exists(_.identical(sourceField)) }
 
 given NamedCell[ArcaneSchemaField] with
   extension (field: ArcaneSchemaField) def name: String = field.name

src/test/scala/tests/models/ArcaneSchemaTests.scala

@@ -2,7 +2,7 @@ package com.sneaksanddata.arcane.framework
 package tests.models
 
 import models.schemas.ArcaneType.{BigDecimalType, IntType, ListType, StringType}
-import models.schemas.{ArcaneSchema, IndexedField, MergeKeyField}
+import models.schemas.{ArcaneSchema, IndexedField, IndexedMergeKeyField, MergeKeyField}
 
 import org.scalatest.Inspectors.forAll
 import org.scalatest.flatspec.AnyFlatSpec
@@ -41,6 +41,20 @@ class ArcaneSchemaTests extends AnyFlatSpec with Matchers {
           ArcaneSchema(Seq(MergeKeyField)),
           ArcaneSchema(Seq(IndexedField("colA", StringType, 1), IndexedField("colB", IntType, 2))),
           Seq(MergeKeyField)
+        ),
+        (
+          ArcaneSchema(Seq(IndexedMergeKeyField(0), IndexedField("colA", StringType, 1))),
+          ArcaneSchema(
+            Seq(IndexedMergeKeyField(1), IndexedField("colA", StringType, 2), IndexedField("colB", IntType, 3))
+          ),
+          Seq()
+        ),
+        (
+          ArcaneSchema(
+            Seq(IndexedMergeKeyField(1), IndexedField("colA", StringType, 2), IndexedField("colB", IntType, 3))
+          ),
+          ArcaneSchema(Seq(IndexedMergeKeyField(0), IndexedField("colA", StringType, 1))),
+          Seq(IndexedField("colB", IntType, 3))
         )
       )
     ) { case (schemaA, schemaB, expected) =>