Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Destinations CDK: Plumbing related to airbyte_meta from protocol to r…
…aw table (#35944) Co-authored-by: Edward Gao <edward.gao@airbyte.io>
- Loading branch information
Showing
30 changed files
with
291 additions
and
98 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
.../src/main/java/io/airbyte/cdk/integrations/destination_async/deser/DeserializationUtil.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
| package io.airbyte.cdk.integrations.destination_async.deser | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting | ||
| import io.airbyte.cdk.integrations.destination_async.partial_messages.PartialAirbyteMessage | ||
| import io.airbyte.commons.json.Jsons | ||
| import io.airbyte.protocol.models.v0.AirbyteMessage | ||
|
|
||
| object DeserializationUtil { | ||
| /** | ||
| * Deserializes to a [PartialAirbyteMessage] which can represent both a Record or a State | ||
| * Message | ||
| * | ||
| * PartialAirbyteMessage holds either: | ||
| * * entire serialized message string when message is a valid State Message | ||
| * * serialized AirbyteRecordMessage when message is a valid Record Message | ||
| * | ||
| * @param messageString the string to deserialize | ||
| * @return PartialAirbyteMessage if the message is valid, empty otherwise | ||
| */ | ||
| @JvmStatic | ||
| @VisibleForTesting | ||
| fun deserializeAirbyteMessage( | ||
| messageString: String?, | ||
| dataTransformer: StreamAwareDataTransformer | ||
| ): PartialAirbyteMessage { | ||
| // TODO: This is doing some sketchy assumptions by deserializing either the whole or the | ||
| // partial based on type. | ||
| // Use JsonSubTypes and extend StdDeserializer to properly handle this. | ||
| // Make immutability a first class citizen in the PartialAirbyteMessage class. | ||
| val partial = | ||
| Jsons.tryDeserializeExact(messageString, PartialAirbyteMessage::class.java) | ||
| .orElseThrow { RuntimeException("Unable to deserialize PartialAirbyteMessage.") } | ||
|
|
||
| val msgType = partial.type | ||
| if (AirbyteMessage.Type.RECORD == msgType && partial.record.data != null) { | ||
| // Transform data provided by destination. | ||
| val transformedData = | ||
| dataTransformer.transform( | ||
| partial.record.streamDescriptor, | ||
| partial.record.data, | ||
| partial.record.meta | ||
| ) | ||
| // store serialized json & meta | ||
| partial.withSerialized(Jsons.serialize(transformedData.getLeft())) | ||
| partial.record.meta = transformedData.getRight() | ||
| // The connector doesn't need to be able to access to the record value. We can serialize | ||
| // it here and | ||
| // drop the json | ||
| // object. Having this data stored as a string is slightly more optimal for the memory | ||
| // usage. | ||
| partial.record.data = null | ||
| } else if (AirbyteMessage.Type.STATE == msgType) { | ||
| partial.withSerialized(messageString) | ||
| } else { | ||
| throw RuntimeException(String.format("Unsupported message type: %s", msgType)) | ||
| } | ||
|
|
||
| return partial | ||
| } | ||
| } |
24 changes: 24 additions & 0 deletions
24
...ain/java/io/airbyte/cdk/integrations/destination_async/deser/IdentityDataTransformer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.integrations.destination_async.deser; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.protocol.models.v0.AirbyteRecordMessageMeta; | ||
| import io.airbyte.protocol.models.v0.StreamDescriptor; | ||
| import org.apache.commons.lang3.tuple.ImmutablePair; | ||
|
|
||
| /** | ||
| * Identity transformer which echoes back the original data and meta. | ||
| */ | ||
| public class IdentityDataTransformer implements StreamAwareDataTransformer { | ||
|
|
||
| @Override | ||
| public ImmutablePair<JsonNode, AirbyteRecordMessageMeta> transform(StreamDescriptor streamDescriptor, | ||
| JsonNode data, | ||
| AirbyteRecordMessageMeta meta) { | ||
| return ImmutablePair.of(data, meta); | ||
| } | ||
|
|
||
| } |
26 changes: 26 additions & 0 deletions
26
.../java/io/airbyte/cdk/integrations/destination_async/deser/StreamAwareDataTransformer.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| /* | ||
| * Copyright (c) 2024 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.cdk.integrations.destination_async.deser; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.protocol.models.v0.AirbyteRecordMessageMeta; | ||
| import io.airbyte.protocol.models.v0.StreamDescriptor; | ||
| import org.apache.commons.lang3.tuple.ImmutablePair; | ||
|
|
||
| public interface StreamAwareDataTransformer { | ||
|
|
||
| /** | ||
| * Transforms the input data by applying destination limitations and populating | ||
| * {@link AirbyteRecordMessageMeta}. The returned pair contains the transformed data and the merged | ||
| * meta information from upstream. | ||
| * | ||
| * @param streamDescriptor | ||
| * @param data | ||
| * @param meta | ||
| * @return | ||
| */ | ||
| ImmutablePair<JsonNode, AirbyteRecordMessageMeta> transform(StreamDescriptor streamDescriptor, JsonNode data, AirbyteRecordMessageMeta meta); | ||
|
|
||
| } |
Oops, something went wrong.