and the default compression is gzip apparently?

airbytehq · Nov 25, 2024 · e3f0bb4 · e3f0bb4
1 parent 114a17d
commit e3f0bb4
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/command/DestinationCatalog.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/command/DestinationCatalog.kt
@@ -26,7 +26,7 @@ data class DestinationCatalog(val streams: List<DestinationStream> = emptyList()
                 "Catalog must have at least one stream: check that files are in the correct location."
             )
         }
-        log.info { "Destination catalog initialized: $streams"}
+        log.info { "Destination catalog initialized: $streams" }
     }
 
     fun getStream(name: String, namespace: String?): DestinationStream {

diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
@@ -435,7 +435,11 @@ class DestinationMessageFactory(
                         namespace = status.streamDescriptor.namespace,
                         name = status.streamDescriptor.name,
                     )
-                if ((message.trace.type ?: AirbyteTraceMessage.Type.STREAM_STATUS) == AirbyteTraceMessage.Type.STREAM_STATUS) {
+                if (
+                    (message.trace.type
+                        ?: AirbyteTraceMessage.Type.STREAM_STATUS) ==
+                        AirbyteTraceMessage.Type.STREAM_STATUS
+                ) {
                     when (status.status) {
                         AirbyteStreamStatus.COMPLETE ->
                             if (fileTransferEnabled) {

diff --git a/...ain/kotlin/io/airbyte/cdk/load/command/object_storage/ObjectStorageFormatSpecification.kt b/...ain/kotlin/io/airbyte/cdk/load/command/object_storage/ObjectStorageFormatSpecification.kt
@@ -129,7 +129,8 @@ class CSVFormatSpecification(
     ObjectStorageCompressionSpecificationProvider {
     override val flattening: FlatteningSpecificationProvider.Flattening =
         FlatteningSpecificationProvider.Flattening.NO_FLATTENING
-    override val compression: ObjectStorageCompressionSpecification? = NoCompressionSpecification()
+    override val compression: ObjectStorageCompressionSpecification? =
+        GZIPCompressionSpecification()
 }
 
 /** JSONL */
@@ -144,7 +145,8 @@ class JsonFormatSpecification(
     ObjectStorageCompressionSpecificationProvider {
     override val flattening: FlatteningSpecificationProvider.Flattening? =
         FlatteningSpecificationProvider.Flattening.NO_FLATTENING
-    override val compression: ObjectStorageCompressionSpecification? = NoCompressionSpecification()
+    override val compression: ObjectStorageCompressionSpecification? =
+        GZIPCompressionSpecification()
 }
 
 /** AVRO */