S3 Dest emits V2 fields and captures failures #42409

johnny-schmidt · 2024-07-22T18:58:31Z

What

Accomplishes the field-renaming part of making S3 V2 compliant.

S3 Syncs now include all V2 metadata fields
_airbyte_meta.changes includes conversion failures for avro and parquet

How

fields are added separately by each conversion format when the converter is so configured (useV2Field[Name]s: bool is passed to each converter
v2 json-to-avro converter uses the FieldConversionFailureListener (added to the json-avro-converter artifact here), which
- nulls the field (by returning null for a new value)
- stages a post-processing change to insert a NULLED change object into the record (after it's completely processed)
!!!FOR THREAD SAFETY!!! the json-to-avro converter is no longer a constant (b/c the listener keeps state!). It is initialized by AvroRecordFactory, which is indirectly initialized on buffer creation (ie, at the start of each flush)
json-to-avrò schema converter also takes useV2Fields
record creation methods now take an optional generationId
setup code builds a map of StreamDescriptor -> generationId and passes it to the s3 flush worker, which in turn passes generationId to the converters
Tests:
- the existing retrieveRecords method stripped the meta fields, I changed this and added a shim so most test cases would still get the stripped rows
- added one test case that validates the new field names and some change capture scenarios

Review Guide

In addition to validating the above, please look out for implications for other destinations that use s3, avro/parquet, and/or the test framework

User Impact

This is a breaking change. However it will be bundled with two more PRs that also contain breaking changes

avrò and parquet schemas more sane (see proposed changes), fixing this issue
S3 no data downtime

Can this PR be safely reverted and rolled back?

YES 💚
NO ❌

vercel · 2024-07-22T18:58:33Z

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Aug 14, 2024 3:54pm

johnny-schmidt · 2024-07-22T18:58:46Z

S3 Dest emits V2 fields and captures failures #42409 : 2 dependent PRs (#42542 , #42579 ) 👈
S3 Destination Uses Async Framework #42405 : 2 other dependent PRs (#42543 , #42924 )
master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @johnny-schmidt and the rest of your teammates on Graphite

johnny-schmidt · 2024-07-22T21:59:15Z

...src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/csv/NoFlatteningSheetGenerator.kt

@@ -19,6 +32,8 @@ class NoFlatteningSheetGenerator : BaseSheetGenerator(), CsvSheetGenerator {

    /** When no flattening is needed, the record column is just one json blob. */
    override fun getRecordColumns(json: JsonNode): List<String> {
-        return listOf(Jsons.serialize(json))
+        val tmp = Jsons.serialize(json)
+        println("tmp: $tmp")


I'll remove this.

stephane-airbyte · 2024-07-23T16:09:09Z

...nations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/avro/AvroRecordFactory.kt

@@ -22,6 +25,29 @@ class AvroRecordFactory(private val schema: Schema?, private val converter: Json
    companion object {
        private val MAPPER: ObjectMapper = MoreMappers.initMapper()
        private val WRITER: ObjectWriter = MAPPER.writer()
+
+        fun createV1JsonToAvroConverter(): JsonAvroConverter {


should we mark this as deprecated?

Possibly. It's still in use by GCS at least, but should go away completely when that's done.

stephane-airbyte · 2024-07-23T16:10:00Z

...nations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/avro/AvroRecordFactory.kt

+
+        return converter!!.convertToGenericDataRecord(WRITER.writeValueAsBytes(jsonRecord), schema)
+    }
+
    @Throws(JsonProcessingException::class)


stephane-airbyte · 2024-07-23T16:19:36Z

...estFixtures/kotlin/io/airbyte/cdk/integrations/destination/s3/S3DestinationAcceptanceTest.kt

@@ -84,6 +91,7 @@ protected constructor(protected val outputFormat: FileUploadFormat) :
                DateTime.now(DateTimeZone.UTC),
                s3DestinationConfig.pathFormat!!,
            )
+        println("outputPrefix: $outputPrefix")


stephane-airbyte · 2024-07-23T16:20:28Z

airbyte-integrations/connectors/destination-s3/build.gradle

+}
+
+run {
+    standardInput = System.in


I don't understand this addition.

Also for testing, will remove. (It's necessary to make the running gradle run task accept input from stdin.)

stephane-airbyte

LGTM with minor comments

edgao

seems reasonable, had some comments/questions. The runtime code changes looked like mostly plumbing so I didn't read them in much detail; had some more questions in the test code. LGTM once everything is resolved

edgao · 2024-07-23T17:16:44Z

...rc/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/BaseSerializedBuffer.kt

@@ -42,7 +42,7 @@ protected constructor(private val bufferStorage: BufferStorage) : SerializableBu
     */
    @Deprecated("")
    @Throws(IOException::class)
-    protected abstract fun writeRecord(record: AirbyteRecordMessage)
+    protected abstract fun writeRecord(record: AirbyteRecordMessage, generationId: Long = 0)


weird thought: should we take generation ID via the constructor? since it's constant across the entire sync

I considered that, as well as changes necessary to make my code use the new methods.

I think that's the correct design. Threading things through is an anti pattern.

But the even more correct pattern is for this code to know nothing about airbyte fields, which I imagine is how it will work in a new CDK. So I punted on doing the work to rewrite anything.

Actually, is it constant per sync or per stream?

ah yeah, it's per stream. Isn't this object instanced per stream? (totally fair that this would be too much work for no payoff though)

...nations/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/avro/AvroRecordFactory.kt

edgao · 2024-07-24T15:34:37Z

...ions/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/avro/AvroSerializedBuffer.kt

                    )
-                AvroSerializedBuffer(createStorageFunction.call(), codecFactory, schema)
+                println("schema: $schema")


remove this println / use a logger instance

That's a stray debug statement, which I'll remove

I thought we should not log client schemas?

client data is bad to log, but we treat the schemas as non-sensitive (e.g. a lot of DB sources log out every column in every table for debugging purposes)

edgao · 2024-07-24T15:46:06Z

...src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/avro/JsonToAvroSchemaConverter.kt

+                    .name(JavaBaseConstants.COLUMN_NAME_AB_GENERATION_ID)
+                    .type(Schema.create(Schema.Type.LONG))
+                    .noDefault()
+                val changeSchema: Schema =


nit: this looks identical to the schema in AvroFieldConversionFailureListener, probably better to reuse that?

I'll try to migrate it to AvroConstants

edgao · 2024-07-24T16:01:16Z

...ns/src/main/kotlin/io/airbyte/cdk/integrations/destination/s3/jsonl/JsonLSerializedBuffer.kt

        json.put(JavaBaseConstants.COLUMN_NAME_AB_ID, UUID.randomUUID().toString())
        json.put(JavaBaseConstants.COLUMN_NAME_EMITTED_AT, record.emittedAt)
+    }
+
+    @Deprecated("Deprecated in Java")


.... I know you didn't write this, but I assume this should just be deprecated everywhere? I'm assuming this method was deprecated before we introduced kotlin >.>

...ures/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseCsvDestinationAcceptanceTest.kt

...es/kotlin/io/airbyte/cdk/integrations/destination/s3/S3BaseJsonlDestinationAcceptanceTest.kt

airbyte-integrations/connectors/destination-s3/build.gradle

...-destinations/src/testFixtures/resources/v0/users_with_generation_id_configured_catalog.json

edgao · 2024-07-24T16:22:39Z

...res/kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt

+            )
+
+        val jsons = MoreMappers.initMapper().createObjectNode()
+        // Iterate over every key value pair in the json node


checking my understanding: given this node:

{ "_airbyte_raw_id": "foo", "_airbyte_data": { "id": 42 } }

we would return this jsons:

{ "id": 42, "_airbyte_data": { "id": 42 } }

(assuming yes: are we intentionally promoting all the data fields, and also keeping the airbyte_data sub-object?)

... also, why not just this? is the idea to preserve non-airbyte top-level fields?

val newObjectNode = node["_airbyte_data"].deepCopy() newObject.put("_airbyte_data", node["_airbyte_data"])

You're correct, but that's a mistake. I'm missing an else near the end.

The intention is to remove airbyte fields AND promote anything under _airbyte_data to the top. I was trying to replicate the effective behavior of all the client implementations of the former retrieveRecords method.

Your example should produce.

{ "id": 42 }

gisripa · 2024-07-25T15:31:01Z

...rc/main/kotlin/io/airbyte/cdk/integrations/destination/record_buffer/BaseSerializedBuffer.kt

@@ -42,7 +42,7 @@ protected constructor(private val bufferStorage: BufferStorage) : SerializableBu
     */
    @Deprecated("")
    @Throws(IOException::class)
-    protected abstract fun writeRecord(record: AirbyteRecordMessage)
+    protected abstract fun writeRecord(record: AirbyteRecordMessage, generationId: Long = 0)

    /**
     * TODO: (ryankfu) move destination to use serialized record string instead of passing entire


[Re: line 62]

Other consumers using the Async rely on this method. We were unable to delete the method on L45 because of S3 not moving to Async and remnants of old calls. Any chance we can achieve that in this PR ?

See this comment inline on Graphite.

Same comment as above.

octavia-squidington-iii added area/connectors Connector related issues CDK Connector Development Kit labels Jul 22, 2024

johnny-schmidt mentioned this pull request Jul 22, 2024

S3 Destination Uses Async Framework #42405

Merged

1 task

octavia-squidington-iii added the connectors/destination/s3 label Jul 22, 2024

johnny-schmidt marked this pull request as ready for review July 22, 2024 20:00

johnny-schmidt requested review from a team and edgao July 22, 2024 20:00

johnny-schmidt commented Jul 22, 2024

View reviewed changes

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8554-s3-async-gt branch 2 times, most recently from 470eee5 to f5414ea Compare July 23, 2024 00:40

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch 2 times, most recently from 480dc23 to 285767f Compare July 23, 2024 00:41

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8554-s3-async-gt branch from f5414ea to 5c9ea03 Compare July 23, 2024 00:50

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 285767f to 92cca0f Compare July 23, 2024 00:52

stephane-airbyte reviewed Jul 23, 2024

View reviewed changes

stephane-airbyte approved these changes Jul 23, 2024

View reviewed changes

edgao approved these changes Jul 24, 2024

View reviewed changes

gisripa force-pushed the epic-8349-certify-s3/issue-8554-s3-async-gt branch from 5c9ea03 to 4d4e5db Compare July 24, 2024 17:53

gisripa reviewed Jul 25, 2024

View reviewed changes

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8554-s3-async-gt branch from 4d4e5db to b7615bb Compare July 25, 2024 17:16

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 2ed6b8e to 32a40ca Compare July 25, 2024 17:16

johnny-schmidt mentioned this pull request Jul 25, 2024

time types now logical numerics w/o strings #42542

Merged

2 tasks

gisripa force-pushed the epic-8349-certify-s3/issue-8554-s3-async-gt branch from b7615bb to 4d4e5db Compare July 25, 2024 18:43

gisripa mentioned this pull request Jul 25, 2024

Destination S3: Connector reacting to generationId and minGenId #42543

Merged

2 tasks

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from f97e80b to 80ef98f Compare August 12, 2024 18:16

gisripa force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch 2 times, most recently from 40277b3 to 51c23f8 Compare August 12, 2024 23:58

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 51c23f8 to 1844494 Compare August 13, 2024 15:34

gisripa force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 1844494 to 33e3e0d Compare August 13, 2024 21:26

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch 2 times, most recently from 1844494 to 194365d Compare August 13, 2024 22:16

octavia-squidington-iii added the area/documentation Improvements or additions to documentation label Aug 13, 2024

vercel bot deployed to Preview August 13, 2024 22:51 View deployment

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 39aa845 to 1585419 Compare August 13, 2024 22:53

vercel bot deployed to Preview August 13, 2024 22:58 View deployment

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 3acf1ff to 491ad7c Compare August 14, 2024 01:06

johnny-schmidt added 13 commits August 14, 2024 08:44

S3 Uses V2 Fields with Change Capture

3ba697c

upgrade to new json-avro-converter version

44d1ff9

addressed PR comments

64d5092

more formatting

1ade390

compile failures

2c0abcb

CDK test failures

dde8ce6

time types now logical numerics w/o strings

0d4612e

Test cases

1760458

Avro and Parquet Field Conversion Improvements

55dbd29

Object merges

5dd71eb

S3 1.0 Breaking Change Documentation

151e458

cdk version bump again

b6d2c65

cdk flip

a626bae

johnny-schmidt force-pushed the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch from 491ad7c to a626bae Compare August 14, 2024 15:54

johnny-schmidt merged commit e9a37d4 into master Aug 14, 2024
31 of 32 checks passed

johnny-schmidt deleted the epic-8349-certify-s3/issue-8556-s3-v2-fields-gt branch August 14, 2024 16:41

kev-datams mentioned this pull request Aug 19, 2024

Destination GCS: Failed to convert json to avro #14064

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

S3 Dest emits V2 fields and captures failures #42409

S3 Dest emits V2 fields and captures failures #42409

johnny-schmidt commented Jul 22, 2024 •

edited

Loading

vercel bot commented Jul 22, 2024 •

edited

Loading

johnny-schmidt commented Jul 22, 2024 •

edited

Loading

johnny-schmidt Jul 22, 2024

stephane-airbyte Jul 23, 2024

johnny-schmidt Jul 26, 2024

stephane-airbyte Jul 23, 2024

stephane-airbyte Jul 23, 2024

johnny-schmidt Jul 26, 2024

stephane-airbyte Jul 23, 2024

johnny-schmidt Jul 26, 2024

stephane-airbyte left a comment

edgao left a comment

edgao Jul 23, 2024

johnny-schmidt Jul 26, 2024

johnny-schmidt Jul 26, 2024

edgao Jul 26, 2024

edgao Jul 24, 2024

johnny-schmidt Jul 26, 2024 •

edited

Loading

edgao Jul 26, 2024

edgao Jul 24, 2024

johnny-schmidt Jul 26, 2024

edgao Jul 24, 2024

edgao Jul 24, 2024

johnny-schmidt Jul 26, 2024

gisripa Jul 25, 2024

johnny-schmidt Jul 26, 2024

S3 Dest emits V2 fields and captures failures #42409

S3 Dest emits V2 fields and captures failures #42409

Conversation

johnny-schmidt commented Jul 22, 2024 • edited Loading

What

How

Review Guide

User Impact

Can this PR be safely reverted and rolled back?

vercel bot commented Jul 22, 2024 • edited Loading

johnny-schmidt commented Jul 22, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephane-airbyte left a comment

Choose a reason for hiding this comment

edgao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johnny-schmidt Jul 26, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johnny-schmidt commented Jul 22, 2024 •

edited

Loading

vercel bot commented Jul 22, 2024 •

edited

Loading

johnny-schmidt commented Jul 22, 2024 •

edited

Loading

johnny-schmidt Jul 26, 2024 •

edited

Loading