Update MsSqlReader to emit indexed schema (#335)

Author: george-zubrienko

src/main/scala/services/base/SchemaProvider.scala

@@ -17,6 +17,8 @@ trait CanAdd[Schema]:
     */
   extension (a: Schema) def addField(fieldName: String, fieldType: ArcaneType): Schema
 
+  extension (a: Schema) def addIndexedField(fieldName: String, fieldType: ArcaneType, fieldId: Int): Schema
+
 /** Represents a provider of a schema for a data produced by Arcane.
   *
   * @tparam Schema

src/main/scala/services/mssql/SqlSchema.scala

@@ -1,35 +1,28 @@
 package com.sneaksanddata.arcane.framework
 package services.mssql
 
-import services.base.CanAdd
+import models.schemas.{ArcaneSchema, MergeKeyField, given_CanAdd_ArcaneSchema}
 import utils.SqlUtils.{JdbcFieldInfo, toArcaneType}
 
-import scala.annotation.tailrec
-import scala.util.{Failure, Success, Try}
-
 /** Represents the schema of a table in a Microsoft SQL Server database. The schema is represented as a sequence of
   * tuples, where each tuple contains the column name, type (java.sql.Types), precision, and scale.
   */
 type SqlSchema = Seq[(String, Int, Int, Int)]
 
-/** Companion object for [[SqlSchema]].
-  */
-object SqlSchema:
-
-  /** Converts a SQL schema to an Arcane schema and normalizes the field names.
-    *
-    * @param sqlSchema
-    *   The SQL schema to convert.
-    * @param schema
-    *   The Arcane schema to add the fields to.
-    * @return
-    *   The Arcane schema with the fields added.
-    */
-  @tailrec
-  def toSchema[Schema: CanAdd](sqlSchema: SqlSchema, schema: Schema): Try[Schema] =
-    sqlSchema match
-      case Seq() => Success(schema)
-      case (name, fieldType, precision, scale) +: xs =>
-        toArcaneType(new JdbcFieldInfo(name = name, typeId = fieldType, precision = precision, scale = scale)) match
-          case Success(arcaneType) => toSchema(xs, schema.addField("\\W+".r.replaceAllIn(name, ""), arcaneType))
-          case Failure(exception)  => Failure[Schema](exception)
+given Conversion[SqlSchema, ArcaneSchema]:
+  // assume that sqlSchema contains merge key and it always comes first
+  // check resources/get_select_delta_query.sql
+  override def apply(sqlSchema: SqlSchema): ArcaneSchema = sqlSchema
+    .foldLeft((ArcaneSchema.empty(), 0)) { case ((agg, fieldIndex), (name, fieldType, precision, scale)) =>
+      (
+        agg.addIndexedField(
+          fieldId = fieldIndex,
+          fieldName = "\\W+".r.replaceAllIn(name, ""),
+          // propagate failure by resolving Try
+          fieldType =
+            toArcaneType(new JdbcFieldInfo(name = name, typeId = fieldType, precision = precision, scale = scale)).get
+        ),
+        fieldIndex + 1
+      )
+    }
+    ._1

src/main/scala/services/mssql/base/MsSqlReader.scala

@@ -7,7 +7,7 @@ import models.schemas.{ArcaneSchema, DataRow, given_CanAdd_ArcaneSchema}
 import models.settings.mssql.MsSqlServerDatabaseSourceSettings
 import services.base.SchemaProvider
 import services.mssql.QueryProvider.{getBackfillQuery, getChangesQuery, getSchemaQuery}
-import services.mssql.SqlSchema.toSchema
+import services.mssql.given_Conversion_SqlSchema_ArcaneSchema
 import services.mssql.base.MsSqlReader.{closeSafe, executeQuerySafe}
 import services.mssql.query.LazyQueryResult.toDataRow
 import services.mssql.query.{LazyQueryResult, ScalarQueryResult}
@@ -196,10 +196,9 @@ class MsSqlReader(
     */
   override lazy val getSchema: Task[this.SchemaType] =
     for
-      query        <- this.getSchemaQuery
-      sqlSchema    <- getSqlSchema(query)
-      arcaneSchema <- ZIO.fromTry(toSchema(sqlSchema, empty))
-    yield arcaneSchema
+      query     <- this.getSchemaQuery
+      sqlSchema <- getSqlSchema(query)
+    yield sqlSchema
 
   private def getSqlSchema(query: String): Task[SqlSchema] =
     ZIO.scoped {

src/test/scala/tests/mssql/MsSqlReaderTests.scala

@@ -2,7 +2,7 @@ package com.sneaksanddata.arcane.framework
 package tests.mssql
 
 import models.schemas.ArcaneType.*
-import models.schemas.{ArcaneSchemaField, DataCell, Field, MergeKeyField}
+import models.schemas.{ArcaneSchemaField, DataCell, IndexedField, IndexedMergeKeyField}
 import models.settings.*
 import models.settings.mssql.MsSqlServerDatabaseSourceSettings
 import services.filters.ColumnSummaryFieldsFilteringService
@@ -287,17 +287,17 @@ object MsSqlReaderTests extends ZIOSpecDefault:
         )
         expected <- ZIO.succeed(
           List(
-            Field("x", IntType),
-            Field("SYS_CHANGE_VERSION", LongType),
-            Field("SYS_CHANGE_OPERATION", StringType),
-            Field("y", IntType),
-            Field("z", BigDecimalType(30, 6)),
-            Field("a", ByteArrayType),
-            Field("b", TimestampType),
-            Field("cd", IntType),
-            Field("e", FloatType),
-            Field("ChangeTrackingVersion", LongType),
-            MergeKeyField
+            IndexedField("x", IntType, 0),
+            IndexedField("SYS_CHANGE_VERSION", LongType, 1),
+            IndexedField("SYS_CHANGE_OPERATION", StringType, 2),
+            IndexedField("y", IntType, 3),
+            IndexedField("z", BigDecimalType(30, 6), 4),
+            IndexedField("a", ByteArrayType, 5),
+            IndexedField("b", TimestampType, 6),
+            IndexedField("cd", IntType, 7),
+            IndexedField("e", FloatType, 8),
+            IndexedField("ChangeTrackingVersion", LongType, 9),
+            IndexedMergeKeyField(10)
           )
         )
         schema <- connector.getSchema