[Bug]: [Java BQ FILE_LOADS] When streaming to dynamic destinations with copy jobs and CREATE_IF_NEEDED, only the first destination's table is created

[ahmedabu98]

What happened?

Was testing FILE_LOADS streaming writes and found that when dynamic destinations are set and copy jobs are used (ie. large data) and CREATE_IF_NEEDED is set, only the first table is created. For example, if I'm writing to two tables A and B, it becomes a race condition on which copy job is seen first in the pipeline. If copy job to table A is performed first, then table A will be created and all subsequent copy jobs to table B will fail with an error similar to the following:

WARNING: Load job beam_bq_job_COPY_testpipelineahmedabualsaud0905154941c26eb941_17a94e2694554455aa31cca9f9389b49_4e2479b4160d04b56a8075645f4974e1_00003-0 failed, will retry: {
  "errorResult" : {
    "message" : "Not found: Table <project>:<dataset>.mytable_B",
    "reason" : "notFound"
  },
  "errors" : [ {
    "message" : "Not found: Table <project>:<dataset>.mytable_B",
    "reason" : "notFound"
  } ],
  "state" : "DONE"
}. Next job id beam_bq_job_COPY_testpipelineahmedabualsaud0905154941c26eb941_17a94e2694554455aa31cca9f9389b49_4e2479b4160d04b56a8075645f4974e1_00003-1

What we would expect instead is for all tables to be created.

P.S. not seeing this behavior in batch mode

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

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

[ahmedabu98]

This behavior is likely due to these lines:

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteRename.java

Lines 193 to 198 in 4c66866

	boolean isFirstPane =
	firstTempTable != null && firstTempTable.isFirstPane() && c.pane().isFirst();
	WriteDisposition writeDisposition =
	isFirstPane ? firstPaneWriteDisposition : WriteDisposition.WRITE_APPEND;
	CreateDisposition createDisposition =
	isFirstPane ? firstPaneCreateDisposition : CreateDisposition.CREATE_NEVER;

The general idea is after the first pane, we set appropriate create and write dispositions so that subsequent jobs don't overwrite previous data. However here, c.pane().isFirst() in streaming is only true for the first copy job. Subsequent copy jobs seem to appear in different panes (maybe because of this GBK). This results in Beam setting CREATE_NEVER disposition on everything after the first copy job, even if its the first job for a particular destination. BigQuery tries to copy into a non-existent table and instead of creating the table it throws the error mentioned above.

[Abacn]

This was identical to https://issues.apache.org/jira/browse/BEAM-7195 - was the fix in #14238 no longer effective?

[ahmedabu98]

I don't think a test was ever created for the changes in that PR, so I can't tell

But I see that the solution in that PR was not fully extended to the multiple partitions path. I can try implementing it there as well. Thanks @Abacn!

[ahmedabu98]

Fixed in #28312