Improve performance when writing large batches (#127)

Author: george-zubrienko

.helm/templates/crd-microsoft-synapse.yaml

@@ -223,6 +223,9 @@ spec:
                 stagingDataSettings:
                   type: object
                   properties:
+                    maxRowsPerFile:
+                      type: integer
+                      default: 10000
                     dataLocation:
                       type: string
                     tableNamePrefix:

build.sbt

@@ -25,7 +25,7 @@ lazy val plugin = (project in file("."))
     name := "arcane-stream-microsoft-synapse-link",
     idePackagePrefix := Some("com.sneaksanddata.arcane.microsoft_synapse_link"),
 
-    libraryDependencies += "com.sneaksanddata" % "arcane-framework_3" % "0.7.3",
+    libraryDependencies += "com.sneaksanddata" % "arcane-framework_3" % "0.7.5",
     libraryDependencies += "io.netty" % "netty-tcnative-boringssl-static" % "2.0.65.Final",
 
     // bugfix for upgrade header

src/main/scala/models/app/MicrosoftSynapseLinkStreamContext.scala

@@ -129,6 +129,8 @@ case class MicrosoftSynapseLinkStreamContext(spec: StreamSpec) extends StreamCon
   override val baseLocation: String = spec.sourceSettings.baseLocation
   override val changeCaptureIntervalSeconds: Int = spec.sourceSettings.changeCaptureIntervalSeconds
 
+  override val maxRowsPerFile: Option[Int] = Some(spec.maxRowsPerFile)
+
   private def parseBackfillStartDate(str: String): Option[OffsetDateTime] =
     val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH.mm.ss'Z'").withZone(ZoneOffset.UTC)
     Try(OffsetDateTime.parse(str, formatter)) match

src/main/scala/models/app/contracts/StreamSpec.scala

@@ -16,7 +16,7 @@ case class CatalogSettings(namespace: String,
 /**
  * The configuration of staging data.
  */
-case class StagingDataSettings(tableNamePrefix: String, catalog: CatalogSettings, dataLocation: Option[String] = None) derives ReadWriter
+case class StagingDataSettings(tableNamePrefix: String, catalog: CatalogSettings, maxRowsPerFile: Int, dataLocation: Option[String] = None) derives ReadWriter
 
 /**
  * The configuration of Iceberg sink.
@@ -74,11 +74,13 @@ case class FieldSelectionRuleSpec(ruleType: String, fields: Array[String]) deriv
  * @param lookBackInterval             The look back interval in seconds
  */
 case class StreamSpec(sourceSettings: SourceSettings,
+                     
+                     // Staging settings
+                     maxRowsPerFile: Int,
 
                       // Grouping settings
                       rowsPerGroup: Int,
                       groupingIntervalSeconds: Int,
-                      groupsPerFile: Int,
                       lookBackInterval: Int,
 
                       // Iceberg settings