[Destination MS SQL V2] Correct Part Size, No buffered input stream (#55252)

johnny-schmidt · web-flow · commit 600587baa92f · 2025-03-10T17:35:24.000+02:00
diff --git a/airbyte-cdk/bulk/toolkits/load-azure-blob-storage/src/main/kotlin/io/airbyte/cdk/load/file/azureBlobStorage/AzureBlobStreamingUpload.kt b/airbyte-cdk/bulk/toolkits/load-azure-blob-storage/src/main/kotlin/io/airbyte/cdk/load/file/azureBlobStorage/AzureBlobStreamingUpload.kt
@@ -9,7 +9,6 @@ import io.airbyte.cdk.load.command.azureBlobStorage.AzureBlobStorageConfiguratio
 import io.airbyte.cdk.load.file.object_storage.StreamingUpload
 import io.airbyte.cdk.load.util.setOnce
 import io.github.oshai.kotlinlogging.KotlinLogging
-import java.io.BufferedInputStream
 import java.nio.ByteBuffer
 import java.util.Base64
 import java.util.concurrent.ConcurrentSkipListMap
@@ -42,14 +41,16 @@ class AzureBlobStreamingUpload(
 
         // The stageBlock call can be done asynchronously or blocking.
         // Here we use the blocking call in a coroutine context.
-        BufferedInputStream(part.inputStream()).use {
+        part.inputStream().use {
             blockBlobClient.stageBlock(
                 blockId,
                 it,
                 part.size.toLong(),
             )
         }
 
+        log.info { "Staged block #$index => $rawBlockId (encoded = $blockId)" }
+
         // Keep track of the blocks in the order they arrived (or the index).
         blockIds[index] = blockId
     }
@@ -77,6 +78,8 @@ class AzureBlobStreamingUpload(
                     blockBlobClient.setMetadata(filteredMetadata)
                 }
             }
+        } else {
+            log.warn { "Complete called multiple times for ${blockBlobClient.blobName}" }
         }
 
         return AzureBlob(blockBlobClient.blobName, config)
diff --git a/airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/file/object_storage/ObjectStorageFormattingWriter.kt b/airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/file/object_storage/ObjectStorageFormattingWriter.kt
@@ -117,7 +117,6 @@ class MSSQLCSVFormattingWriter(
     private val printer = finalSchema.toCsvPrinterWithHeader(outputStream)
     private val mssqlRowValidator = MSSQLCsvRowValidator(validateValuesPreLoad)
     override fun accept(record: DestinationRecordAirbyteValue) {
-
         printer.printRecord(
             mssqlRowValidator
                 .validate(record, this.finalSchema)
diff --git a/airbyte-integrations/connectors/destination-mssql-v2/metadata.yaml b/airbyte-integrations/connectors/destination-mssql-v2/metadata.yaml
@@ -16,7 +16,7 @@ data:
             type: GSM
   connectorType: destination
   definitionId: 37a928c1-2d5c-431a-a97d-ae236bd1ea0c
-  dockerImageTag: 0.1.14
+  dockerImageTag: 0.1.15
   dockerRepository: airbyte/destination-mssql-v2
   documentationUrl: https://docs.airbyte.com/integrations/destinations/mssql-v2
   githubIssueLabel: destination-mssql-v2
diff --git a/airbyte-integrations/connectors/destination-mssql-v2/src/main/kotlin/io/airbyte/integrations/destination/mssql/v2/config/MSSQLConfiguration.kt b/airbyte-integrations/connectors/destination-mssql-v2/src/main/kotlin/io/airbyte/integrations/destination/mssql/v2/config/MSSQLConfiguration.kt
@@ -8,6 +8,7 @@ import io.airbyte.cdk.ConfigErrorException
 import io.airbyte.cdk.command.FeatureFlag
 import io.airbyte.cdk.load.command.DestinationConfiguration
 import io.airbyte.cdk.load.command.DestinationConfigurationFactory
+import io.airbyte.cdk.load.command.object_storage.ObjectStorageUploadConfiguration
 import io.micronaut.context.annotation.Factory
 import jakarta.inject.Singleton
 
@@ -25,6 +26,7 @@ data class MSSQLConfiguration(
     override val numProcessRecordsWorkers = 1
     override val numProcessBatchWorkers: Int = 1
     override val processEmptyFiles: Boolean = true
+    override val recordBatchSizeBytes = ObjectStorageUploadConfiguration.DEFAULT_PART_SIZE_BYTES
 }
 
 @Singleton
diff --git a/docs/integrations/destinations/mssql-v2.md b/docs/integrations/destinations/mssql-v2.md
@@ -99,22 +99,23 @@ You’ll need to supply:
 <details>
   <summary>Expand to review</summary>
 
-| Version | Date       | Pull Request                                              | Subject                                                |
-|:--------|:-----------|:----------------------------------------------------------|:-------------------------------------------------------|
-| 0.1.14  | 2025-03-05 | [54159](https://github.com/airbytehq/airbyte/pull/54159)  | RC12: Support For Bulk Insert Using Azure Blob Storage |
-| 0.1.13  | 2025-03-04 | [55193](https://github.com/airbytehq/airbyte/pull/55193)  | RC11: Increase decimal precision                       |
-| 0.1.12  | 2025-02-24 | [54648](https://github.com/airbytehq/airbyte/pull/54648)  | RC10: Fix index column names with hyphens              |
-| 0.1.11  | 2025-02-21 | [54197](https://github.com/airbytehq/airbyte/pull/54197)  | RC9: Fix index column names with invalid characters    |
-| 0.1.10  | 2025-02-20 | [54186](https://github.com/airbytehq/airbyte/pull/54186)  | RC8: Fix String support                                |
-| 0.1.9   | 2025-02-11 | [53364](https://github.com/airbytehq/airbyte/pull/53364)  | RC7: Revert deletion change                            |
-| 0.1.8   | 2025-02-11 | [53364](https://github.com/airbytehq/airbyte/pull/53364)  | RC6: Break up deletes into loop to reduce locking      |
-| 0.1.7   | 2025-02-07 | [53236](https://github.com/airbytehq/airbyte/pull/53236)  | RC5: Use rowlock hint                                  |
-| 0.1.6   | 2025-02-06 | [53192](https://github.com/airbytehq/airbyte/pull/53192)  | RC4: Fix config, timehandling, performance tweak       |
-| 0.1.5   | 2025-02-04 | [53174](https://github.com/airbytehq/airbyte/pull/53174)  | RC3: Fix metadata.yaml for publish                     |
-| 0.1.4   | 2025-02-04 | [52704](https://github.com/airbytehq/airbyte/pull/52704)  | RC2: Performance improvement                           |
-| 0.1.3   | 2025-01-24 | [52096](https://github.com/airbytehq/airbyte/pull/52096)  | Release candidate                                      |
-| 0.1.2   | 2025-01-10 | [51508](https://github.com/airbytehq/airbyte/pull/51508)  | Use a non-root base image                              |
-| 0.1.1   | 2024-12-18 | [49870](https://github.com/airbytehq/airbyte/pull/49870)  | Use a base image: airbyte/java-connector-base:1.0.0    |
-| 0.1.0   | 2024-12-16 | [\#49460](https://github.com/airbytehq/airbyte/pull/49460)| Initial commit                                         |
+| Version | Date       | Pull Request                                               | Subject                                                |
+|:--------|:-----------|:-----------------------------------------------------------|:-------------------------------------------------------|
+| 0.1.15  | 2025-03-07 | [55252](https://github.com/airbytehq/airbyte/pull/55252)   | RC13: Bugfix for OOM on Bulk Load                      |
+| 0.1.14  | 2025-03-05 | [54159](https://github.com/airbytehq/airbyte/pull/54159)   | RC12: Support For Bulk Insert Using Azure Blob Storage |
+| 0.1.13  | 2025-03-04 | [55193](https://github.com/airbytehq/airbyte/pull/55193)   | RC11: Increase decimal precision                       |
+| 0.1.12  | 2025-02-24 | [54648](https://github.com/airbytehq/airbyte/pull/54648)   | RC10: Fix index column names with hyphens              |
+| 0.1.11  | 2025-02-21 | [54197](https://github.com/airbytehq/airbyte/pull/54197)   | RC9: Fix index column names with invalid characters    |
+| 0.1.10  | 2025-02-20 | [54186](https://github.com/airbytehq/airbyte/pull/54186)   | RC8: Fix String support                                |
+| 0.1.9   | 2025-02-11 | [53364](https://github.com/airbytehq/airbyte/pull/53364)   | RC7: Revert deletion change                            |
+| 0.1.8   | 2025-02-11 | [53364](https://github.com/airbytehq/airbyte/pull/53364)   | RC6: Break up deletes into loop to reduce locking      |
+| 0.1.7   | 2025-02-07 | [53236](https://github.com/airbytehq/airbyte/pull/53236)   | RC5: Use rowlock hint                                  |
+| 0.1.6   | 2025-02-06 | [53192](https://github.com/airbytehq/airbyte/pull/53192)   | RC4: Fix config, timehandling, performance tweak       |
+| 0.1.5   | 2025-02-04 | [53174](https://github.com/airbytehq/airbyte/pull/53174)   | RC3: Fix metadata.yaml for publish                     |
+| 0.1.4   | 2025-02-04 | [52704](https://github.com/airbytehq/airbyte/pull/52704)   | RC2: Performance improvement                           |
+| 0.1.3   | 2025-01-24 | [52096](https://github.com/airbytehq/airbyte/pull/52096)   | Release candidate                                      |
+| 0.1.2   | 2025-01-10 | [51508](https://github.com/airbytehq/airbyte/pull/51508)   | Use a non-root base image                              |
+| 0.1.1   | 2024-12-18 | [49870](https://github.com/airbytehq/airbyte/pull/49870)   | Use a base image: airbyte/java-connector-base:1.0.0    |
+| 0.1.0   | 2024-12-16 | [\#49460](https://github.com/airbytehq/airbyte/pull/49460) | Initial commit                                         |
 
 </details>

Original file line number	Diff line number	Diff line change
`@@ -9,7 +9,6 @@ import io.airbyte.cdk.load.command.azureBlobStorage.AzureBlobStorageConfiguratio`
`9`	`9`	`import io.airbyte.cdk.load.file.object_storage.StreamingUpload`
`10`	`10`	`import io.airbyte.cdk.load.util.setOnce`
`11`	`11`	`import io.github.oshai.kotlinlogging.KotlinLogging`
`12`		`-import java.io.BufferedInputStream`
`13`	`12`	`import java.nio.ByteBuffer`
`14`	`13`	`import java.util.Base64`
`15`	`14`	`import java.util.concurrent.ConcurrentSkipListMap`
`@@ -42,14 +41,16 @@ class AzureBlobStreamingUpload(`
`42`	`41`
`43`	`42`	`// The stageBlock call can be done asynchronously or blocking.`
`44`	`43`	`// Here we use the blocking call in a coroutine context.`
`45`		`- BufferedInputStream(part.inputStream()).use {`
	`44`	`+ part.inputStream().use {`
`46`	`45`	`blockBlobClient.stageBlock(`
`47`	`46`	`blockId,`
`48`	`47`	`it,`
`49`	`48`	`part.size.toLong(),`
`50`	`49`	`)`
`51`	`50`	`}`
`52`	`51`
	`52`	`+ log.info { "Staged block #$index => $rawBlockId (encoded = $blockId)" }`
	`53`	`+`
`53`	`54`	`// Keep track of the blocks in the order they arrived (or the index).`
`54`	`55`	`blockIds[index] = blockId`
`55`	`56`	`}`
`@@ -77,6 +78,8 @@ class AzureBlobStreamingUpload(`
`77`	`78`	`blockBlobClient.setMetadata(filteredMetadata)`
`78`	`79`	`}`
`79`	`80`	`}`
	`81`	`+ } else {`
	`82`	`+ log.warn { "Complete called multiple times for ${blockBlobClient.blobName}" }`
`80`	`83`	`}`
`81`	`84`
`82`	`85`	`return AzureBlob(blockBlobClient.blobName, config)`