[Bug]: BigQuerySourceBase does not propagate a Coder to AvroSource

[piter75]

What happened?

Since #22718 our Spotify's Scio based streaming pipelines on Google Cloud Dataflow are failing with the AvroCodec exception while reading data from BigQuery (with TypedRead).
The last released version of Beam that works properly is 2.42.0 and we cannot upgrade some of our pipelines further because of the issue.

We are reading GenericRecords from temporary BigQuery table and apply parseFn function to it to create arbitrary (non-avro) types which is effectively the Case 3 from the table described in the AvroSource.Mode.

// pseudo code

BigQueryIO
        .read(SerializableFunction[SchemaAndRecord, T] parseFn)
        .withCoder(Coder[T] coder)

I analysed the issue. It is complex but the gist of it is that:

1. Support custom avro DatumReader when reading from BigQuery #22718 adds the ability to use a custom AvroSource.DatumReaderFactory implementation for reading from BigQuery;
2. it creates its "default" / "backwards compatibility" implementation and uses it in BigQueryIO.read;
3. this "default" implementation is in fact using the parseFn function (supplied to BigQueryIO.read) to actually return the parsed type from custom DatumReader;
4. however, it does not (and cannot) propagate the output Coder to the AvroSource used for reading the data;
5. Dataflow (in the streaming mode) is wrapping the AvroSource in UnboundedReadFromBoundedSource wrapper to use it as UnboundedSource;
6. on the way it tries to get the output Coder from the underlying AvroSource to use it as CheckpointCoder for checkpointing;
7. AvroSource does not have a clue about parseFn being actually used and it returns the AvroCoder instance which of course cannot encode arbitrary (non-avro) types

The biggest issue I see is that the contract between using parseFn in the process and supplying the output Coder that AvroSource enforces is broken by moving the responsibility of applying the parseFn into GenericDatumTransformer.

I am thinking about contributing a fix and I am pondering on the following solution:

1. removal of the BigQueryIO.GenericDatumTransformer
2. bringing back the parseFn to BigQueryBaseSource hierarchy
3. simplifying the datumReaderFactory type to AvroSource.DatumReaderFactory<T> and stop applying parseFn in it
4. adding validation that only one of parseFn or datumReaderFactory is used - I believe that the purpose of custom DatumReader is to actually read SpecificRecords and output them without the need for additional parsing.
5. creating AvroSources accordingly to which param was actually provided in BigQuerySourceBase.createSources

This will of course add more complexity to the already complex process but will keep the backwards compatibility in more scenarios.

CC: @steveniemitz @kkdoon

Issue Priority

Priority: 1 (data loss / total loss of function)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[kkdoon]

Hi @piter75,

Thanks for the detailed explanation. Is there a way to reproduce the bug via a unit-test? Also, curious that how is your streaming job reading from a bounded source. Would you be able to share more details of your job graph?

I can also take a look at the issue and try to debug it at my end. But overall your approach makes sense to me.

Thanks!

[piter75]

Thanks for responding @kkdoon.

Our use case for BoundedSource in streaming job stems from the fact that we want to have a hot start whenever we start/re-start the job.

Part of our pipeline depends on the join between streams with different event frequency. On one side there is a stream that delivers thousands of messages per second and on the other is one that may not have matching messages for a day.
We solved this frequency mismatch by loading the history of the "slow moving" stream from BigQuery and then making a union with a stream of messages that come straight from the PubSub topic.
This way we have a union that is both complete from the start and still unbounded during the pipeline run.

Unit testing this issue may be difficult because it is triggered when used on a specific runner in specific run mode and I wouldn't like to test the behaviour of the BigQuery source with the runner tests.

I will try to come up with some reproduction example.

What runner are you using by the way?

[kennknowles]

Was the comment at #22718 (comment) addressed?

[kennknowles]

It looks like #23594 was a response to it?

CC @steveniemitz

[kkdoon]

@kennknowles yeah correct, #23594 was in response to the comment and the issue #23541.

@piter75 we are using dataflow runner.

[RustedBones]

@piter75 are you still working on that? If not I can give it a shot.

[piter75]

@RustedBones We did introduce a workaround in our pipelines, and then I got caught up by some other tasks :(
I don't foresee having time to address this issue in the next 2-3 weeks.
I would be greatful If you can take over.

[RustedBones]

@kennknowles since you were involved in the conversation, can you have a look to the associated PR ?

[liferoad]

@Abacn FYI.