[BEAM-480] Move insertAll() from BigQueryTableInserter to BigQueryServices

[peihe]

No new code in this PR, just a refactoring.

[peihe]

R: @dhalperi

[dhalperi]

LGTM

[gadaldo]

If you move BigQueryTableInserter in the BigqueryServices, how can I insert the whole List<TableRow> into a Bigquery table? I could use com.google.api.services.bigquery.Bigquery service rather then the old BigQueryTableInserte, but I can't access getDatasetService(options) from that object so I can't use insertAll() method, so how can I replace the following code now?

Bigquery bigqueryService = ... // building bigQueryService with google credential
BigQueryTableInserter bigQueryTableInserter = new BigQueryTableInserter(bigqueryService);
   bigQueryTableInserter.getOrCreateTable(tableReference,
   BigQueryIO.Write.WriteDisposition.WRITE_APPEND,
   BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED,
   tableSchema);
bigQueryTableInserter.insertAll(tableReference, rows);

I am struggling with it.

Thank you

[peihe]

Could you explain to me the context of your usage of BigQueryTableInserter?

Could you use BigQueryIO.Write?
https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java#L185

BigqueryServices is the implementation details of the BigQueryIO (similarly we have BigtableServices). If we make the visibility public, we have to guarantee the backward compatibility from now on. And, we need to review the exposed methods signatures, and their use cases.

[gadaldo]

In terms of backward compatibility, I think that that feature (BigQueryTableInserter) was there and was working in production and now we can't see any alternative to it, so... not that great.

I can't use BigQueryIO.Write otherwise I would have done it. Mainly because of the nature of our dataflows; I'll try to explain the scenario:
We are receiving batch of Json events from pub sub in base64 algorithm. We have to store those into Bigquey tables, so we translate these records into GenericRecord and then into TableRows in a processElement() method. In this method we retrieve the table schema and the name of the table at runtime (from the header of the message) so we can't output() a List<TableRow> otherwise we lose the table name externally and the schema as well, that's why we can't use BigQueryIO.Write and we have to store these records within the processElement() method itself using the object BigQueryTableInserter.

If you have any other idea to apply so, it's welcomed.
Thank you