Source Amazon Ads: schema typed incorrectly, "Could not evaluate union" exception

[animer3009]

Environment

• Airbyte version: 0.40.10
• OS Version / Instance: Ubuntu 22.04, GCP
• Deployment: Docker
• Source Connector and version: Amazon Ads 0.1.22
• Destination Connector and version: BigQuery (denormalized typed struct) 1.2.4
• Step where error happened: BigQuery (denormalized typed struct) Destination connector on time sync

Current Behavior

I am regularly getting errors like this.

2022-09-16 23:07:15 destination > Exception while accepting message
Stack Trace: tech.allegro.schema.json2avro.converter.AvroConversionException: Failed to convert JSON to Avro: Could not evaluate union, field portfolioId is expected to be one of these: NULL, STRING. If this is a complex type, check if offending field (path: portfolioId) adheres to schema: 130457359476581
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:129)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:118)
	at tech.allegro.schema.json2avro.converter.JsonAvroConverter.convertToGenericDataRecord(JsonAvroConverter.java:95)
	at io.airbyte.integrations.destination.s3.avro.AvroRecordFactory.getAvroRecord(AvroRecordFactory.java:44)
	at io.airbyte.integrations.destination.bigquery.BigQueryAvroSerializedBuffer.writeRecord(BigQueryAvroSerializedBuffer.java:46)
	at io.airbyte.integrations.destination.record_buffer.BaseSerializedBuffer.accept(BaseSerializedBuffer.java:92)
	at io.airbyte.integrations.destination.record_buffer.SerializedBufferingStrategy.addRecord(SerializedBufferingStrategy.java:61)
	at io.airbyte.integrations.destination.buffered_stream_consumer.BufferedStreamConsumer.acceptTracked(BufferedStreamConsumer.java:138)
	at io.airbyte.integrations.base.FailureTrackingAirbyteMessageConsumer.accept(FailureTrackingAirbyteMessageConsumer.java:47)
	at io.airbyte.integrations.base.IntegrationRunner.consumeMessage(IntegrationRunner.java:259)
	at io.airbyte.integrations.base.IntegrationRunner.consumeWriteStream(IntegrationRunner.java:172)
	at io.airbyte.integrations.base.IntegrationRunner.lambda$runConsumer$2(IntegrationRunner.java:178)
	at io.airbyte.integrations.base.IntegrationRunner.watchForOrphanThreads(IntegrationRunner.java:207)
	at io.airbyte.integrations.base.IntegrationRunner.runConsumer(IntegrationRunner.java:177)
	at io.airbyte.integrations.base.IntegrationRunner.runInternal(IntegrationRunner.java:146)
	at io.airbyte.integrations.base.IntegrationRunner.run(IntegrationRunner.java:97)
	at io.airbyte.integrations.destination.bigquery.BigQueryDenormalizedDestination.main(BigQueryDenormalizedDestination.java:200)
Caused by: org.apache.avro.AvroTypeException: Could not evaluate union, field portfolioId is expected to be one of these: NULL, STRING. If this is a complex type, check if offending field (path: portfolioId) adheres to schema: 130457359476581
	at tech.allegro.schema.json2avro.converter.AvroTypeExceptions.unionException(AvroTypeExceptions.java:28)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.readUnion(JsonGenericRecordReader.java:294)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:196)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:170)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.lambda$readRecord$0(JsonGenericRecordReader.java:152)
	at java.base/java.util.LinkedHashMap.forEach(LinkedHashMap.java:721)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.readRecord(JsonGenericRecordReader.java:141)
	at tech.allegro.schema.json2avro.converter.JsonGenericRecordReader.read(JsonGenericRecordReader.java:127)
	... 16 more

The full log is attached.
logs-10.txt

[sh4sh]

Here is the source of the exception: https://github.com/airbytehq/json-avro-converter/blob/95f40995354f31069c1db4c2504362c8ea8f80c5/converter/src/main/java/tech/allegro/schema/json2avro/converter/AvroTypeExceptions.java#L18

Stack Trace: tech.allegro.schema.json2avro.converter.AvroConversionException: Failed to convert JSON to Avro: Could not evaluate union, field portfolioId is expected to be one of these: NULL, STRING. If this is a complex type, check if offending field (path: portfolioId) adheres to schema: 130457359476581

This exception message says that the field path: portfolioId with the value 130457359476581 is expected to be either NULL or STRING according to the Amazon Ads source schema. I think the bug is here:

airbyte/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/schemas/sponsored_display.py

Line 19 in b4fd1ea

portfolioId: str = None

I wonder if it should be typed as Decimal to match the two other schemas that use portfolioId:

airbyte/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/schemas/sponsored_brands.py

Line 22 in 69f53eb

portfolioId: Decimal

airbyte/airbyte-integrations/connectors/source-amazon-ads/source_amazon_ads/schemas/sponsored_products.py

Line 22 in 69f53eb

portfolioId: Decimal

[sh4sh]

According to Amazon Ads docs, portfolioId is integer for all endpoints on V3 (Sponsored Brands, Sponsored Display) https://advertising.amazon.com/API/docs/en-us/reporting/v3/metrics#portfolioid

And for Sponsored Display, it is nullable as well https://advertising.amazon.com/API/docs/en-us/sponsored-display/3-0/openapi#/Campaigns/getCampaign

portfolioId
integer($int64)
nullable: true
example: null
Identifier of the portfolio that will be associated with the campaign. If null then the campaign will be disassociated from existing portfolio. Campaigns with CPC and vCPM costType are supported.

It's notable that Sponsored Brands, Sponsored Display are both on V3 whereas Sponsored Products is on V2 and for those calls portfolioId is of type number.

[animer3009]

@sh4sh FYI - This issue exists only on time use BigQuery (denormalized typed struct) destination connector.
On case stabile BigQuery destination connector it works fine.

[alafanechere]

@bmcgilli @oustynova @sh4sh I think we can't proceed in solving this issue without test data. We indeed miss test data for the stream have these portfolioId fields.
Changing the field type is a breaking change and I don't think we can safely introduce it without test data to check the actual type of the field and how setting it to integer impact existing syncs and their destination.
Can we launch a campaign on our test account to collect test data?
An other approach would be to ask @animer3009 to test out the change made in #25874 . @roman-yermilov-gl a pre-release should be published to make the connector docker image available. @animer3009 could then upgrade the connector on an existing connection and share how the new version behaves.