Support Analyze and batch duration metrics (#162)

Author: george-zubrienko

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" % "1.1.9",
+    libraryDependencies += "com.sneaksanddata" % "arcane-framework_3" % "1.2.0",
     libraryDependencies += "io.netty" % "netty-tcnative-boringssl-static" % "2.0.65.Final",
 
     // bugfix for upgrade header

src/main/scala/main.scala

@@ -3,7 +3,7 @@ package com.sneaksanddata.arcane.microsoft_synapse_link
 import models.app.{AzureConnectionSettings, GraphExecutionSettings, MicrosoftSynapseLinkStreamContext}
 
 import com.azure.storage.common.StorageSharedKeyCredential
-import com.sneaksanddata.arcane.framework.excpetions.StreamFailException
+import com.sneaksanddata.arcane.framework.exceptions.StreamFailException
 import com.sneaksanddata.arcane.framework.logging.ZIOLogAnnotations.zlog
 import com.sneaksanddata.arcane.framework.models.app.StreamContext
 import com.sneaksanddata.arcane.framework.models.settings.{GroupingSettings, VersionedDataGraphBuilderSettings}

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

@@ -95,6 +95,13 @@ case class MicrosoftSynapseLinkStreamContext(spec: StreamSpec)
       }
     )
 
+    override val targetAnalyzeSettings: Option[AnalyzeSettings] = Some(
+      new AnalyzeSettings {
+        override val batchThreshold: Int          = spec.sinkSettings.analyzeSettings.batchThreshold
+        override val includedColumns: Seq[String] = spec.sinkSettings.analyzeSettings.includedColumns
+      }
+    )
+
   override val endpoint: String  = sys.env("ARCANE_FRAMEWORK__STORAGE_ENDPOINT")
   override val container: String = sys.env("ARCANE_FRAMEWORK__STORAGE_CONTAINER")
   override val account: String   = sys.env("ARCANE_FRAMEWORK__STORAGE_ACCOUNT")

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

@@ -35,13 +35,16 @@ case class SnapshotExpirationSettingsSpec(batchThreshold: Int, retentionThreshol
   */
 case class OrphanFilesExpirationSettings(batchThreshold: Int, retentionThreshold: String) derives ReadWriter
 
+case class AnalyzeSettings(batchThreshold: Int, includedColumns: Seq[String]) derives ReadWriter
+
 /** The configuration of Iceberg sink.
   */
 case class SinkSettings(
     targetTableName: String,
     optimizeSettings: OptimizeSettingsSpec,
     snapshotExpirationSettings: SnapshotExpirationSettingsSpec,
-    orphanFilesExpirationSettings: OrphanFilesExpirationSettings
+    orphanFilesExpirationSettings: OrphanFilesExpirationSettings,
+    analyzeSettings: AnalyzeSettings
 ) derives ReadWriter
 
 /** The configuration of the stream source.