Fix PQ empty part issue

Author: johnny-schmidt

airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/pipline/object_storage/ObjectLoaderPartToObjectAccumulator.kt

@@ -67,7 +67,9 @@ class ObjectLoaderPartToObjectAccumulator(
 
     override suspend fun accept(input: Part, state: State): Pair<State, ObjectResult?> {
         input.bytes?.let { state.streamingUpload.uploadPart(it, input.partIndex) }
-            ?: throw IllegalStateException("Empty part received: this should not happen")
+        if (input.bytes == null) {
+            throw IllegalStateException("Empty non-final part received: this should not happen")
+        }
         state.bookkeeper.add(input)
         if (state.bookkeeper.isComplete) {
             return Pair(state, finish(state))

airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/pipline/object_storage/ObjectLoaderRecordToPartAccumulator.kt

@@ -68,10 +68,10 @@ class ObjectLoaderRecordToPartAccumulator<T : OutputStream>(
         return State(stream, writerFactory.create(stream), partFactory)
     }
 
-    private fun makePart(state: State<T>): Part {
+    private fun makePart(state: State<T>, forceFinish: Boolean = false): Part {
         state.writer.flush()
         val newSize = state.partFactory.totalSize + state.writer.bufferSize
-        val isFinal =
+        val isFinal = forceFinish ||
             newSize >= objectSizeBytes ||
                 batchSizeOverride != null // HACK: This is a hack to force a flush
         val bytes =
@@ -109,6 +109,6 @@ class ObjectLoaderRecordToPartAccumulator<T : OutputStream>(
     }
 
     override suspend fun finish(state: State<T>): Part {
-        return makePart(state)
+        return makePart(state, true)
     }
 }

airbyte-integrations/connectors/destination-s3/src/test-integration/kotlin/io/airbyte/integrations/destination/s3_v2/S3V2PerformanceTest.kt

@@ -4,23 +4,35 @@
 
 package io.airbyte.integrations.destination.s3_v2
 
-import io.airbyte.cdk.load.command.Property
 import io.airbyte.cdk.load.write.BasicPerformanceTest
-import org.junit.jupiter.api.Disabled
 import org.junit.jupiter.api.Test
 
 //@Disabled("We don't want this to run in CI")
-class S3V2PerformanceTest :
+class S3V2JsonNoFrillsPerformanceTest :
     BasicPerformanceTest(
         configContents = S3V2TestUtils.getConfig(S3V2TestUtils.JSON_UNCOMPRESSED_CONFIG_PATH),
         configSpecClass = S3V2Specification::class.java,
         defaultRecordsToInsert = 1_000_000,
-        micronautProperties =
-            mapOf(
-                Property("airbyte.destination.aws.assume-role.access-key", "FOO") to "foo",
-                Property("airbyte.destination.aws.assume-role.secret-key", "BAR") to "bar",
-                Property("airbyte.destination.aws.assume-role.external-id", "BAZ") to "baz",
-            ),
+        micronautProperties = S3V2TestUtils.PERFORMANCE_TEST_MICRONAUT_PROPERTIES
+    ) {
+    @Test
+    override fun testInsertRecords() {
+        super.testInsertRecords()
+    }
+
+    @Test
+    override fun testRefreshingRecords() {
+        super.testRefreshingRecords()
+    }
+}
+
+class S3V2ParquetSnappyPerformanceTest :
+    BasicPerformanceTest(
+        configContents = S3V2TestUtils.getConfig(S3V2TestUtils.PARQUET_SNAPPY_CONFIG_PATH),
+        configSpecClass = S3V2Specification::class.java,
+        defaultRecordsToInsert = 1_000_000,
+        micronautProperties = S3V2TestUtils.PERFORMANCE_TEST_MICRONAUT_PROPERTIES
+,
     ) {
     @Test
     override fun testInsertRecords() {

airbyte-integrations/connectors/destination-s3/src/test-integration/kotlin/io/airbyte/integrations/destination/s3_v2/S3V2TestUtils.kt

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.destination.s3_v2
 
+import io.airbyte.cdk.load.command.Property
 import io.airbyte.cdk.load.command.aws.AwsAssumeRoleCredentials
 import io.airbyte.cdk.load.util.Jsons
 import java.nio.file.Files
@@ -52,4 +53,10 @@ object S3V2TestUtils {
                 assumeRoleExternalId,
             )
     }
+
+    val PERFORMANCE_TEST_MICRONAUT_PROPERTIES = mapOf(
+        Property("airbyte.destination.aws.assume-role.access-key", "FOO") to "foo",
+        Property("airbyte.destination.aws.assume-role.secret-key", "BAR") to "bar",
+        Property("airbyte.destination.aws.assume-role.external-id", "BAZ") to "baz",
+    )
 }