feat: validate airbyte metadata added to iceberg schema #48604
Conversation
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
octavia-squidington-iii
added
area/connectors
|
this looks a lot like https://github.com/airbytehq/airbyte/blob/master/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/AirbyteTypeToAirbyteTypeWithMeta.kt ? which we can maybe reuse in some way |
| } | ||
| } | ||
|
|
||
| private fun createAirbyteMetadataFields(): List<NestedField> { |
There was a problem hiding this comment.
Most of these columns are being added to the schema using the mapper architecture. if you take a look at
val schema = pipeline.finalSchema.withAirbyteMeta(true).toIcebergSchema(primaryKeys)
In IcebergV2Writer. The withAirbyteMeta method is adding a bunch of em. If we want to add more like the generation id , etc. we should use the same design.
...-iceberg-v2/src/main/kotlin/io/airbyte/integrations/destination/iceberg/v2/io/IcebergUtil.kt
Outdated
Show resolved
Hide resolved
@edgao It does and I wasn't sure how to reuse that given that it returns Airbyte types and not Iceberg types. I suspect what is really being pointed out here is some refactor to this in the CDK to make the common parts more common. |
jdpgrailsdev
changed the title
|
PR has been updated to validate that the metadata is added to the schema. I have removed the code that I originally added to do this, as the existing logic already adds the fields. This is confirmed via new tests added to this PR. |
| // the deletion status. This should be revisited when there is a cheaper way to do | ||
| // this, such | ||
| // as after a protocol change that explicitly states the operation. | ||
| // if (record.data.toJson().get(AIRBYTE_CDC_DELETE_COLUMN) != null) { |
There was a problem hiding this comment.
can we do something like this to dodge the serialize?
(record.data as ObjectValue).properties[DELETE_COLUMN] != null
There was a problem hiding this comment.
@edgao Sure? I'll test it out and update the PR if it works
There was a problem hiding this comment.
@edgao That appeared to work. I will update the PR to include the DELETE operation detection based on your suggestion.
What
How
withMetadata()is calledIcebergUtilto a Micronaut singleton to better facilitate testingReview guide
IcebergUtil.ktIcebergUtilTest.ktIcebergV2WriterTest.ktAirbyteTypeToAirbyteTypeWithMetaTest.ktCan this PR be safely reverted and rolled back?