Add Dynamic Schema support to StorageWriteToBigQuery - #39236

apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
org.apache.beam.sdk.util.UserCodeException: java.lang.RuntimeException: Append to stream projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ failed with Status Code INVALID_ARGUMENT. The stream may not exist.
	at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn$DoFnInvoker$bf057275.invokeFinishBundle(Unknown Source)
	at org.apache.beam.fn.harness.FnApiDoFnRunner.finishBundle(FnApiDoFnRunner.java:1317)
	at org.apache.beam.fn.harness.data.PTransformFunctionRegistry.lambda$register$0(PTransformFunctionRegistry.java:115)
	at org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:567)
	at org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:150)
	at org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:115)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:164)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.lang.RuntimeException: Append to stream projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ failed with Status Code INVALID_ARGUMENT. The stream may not exist.
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn$DestinationState.lambda$flush$9(StorageApiWriteUnshardedRecords.java:856)
	at org.apache.beam.sdk.io.gcp.bigquery.RetryManager.await(RetryManager.java:319)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.flushAll(StorageApiWriteUnshardedRecords.java:1069)
	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.finishBundle(StorageApiWriteUnshardedRecords.java:1257)
Caused by: com.google.cloud.bigquery.storage.v1.Exceptions$SchemaMismatchedException: INVALID_ARGUMENT: Input schema has more fields than BigQuery schema, extra fields: 'name' Entity: projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ
	at com.google.cloud.bigquery.storage.v1.Exceptions.toStorageException(Exceptions.java:200)
	at com.google.cloud.bigquery.storage.v1.Exceptions.toStorageException(Exceptions.java:222)
	at com.google.cloud.bigquery.storage.v1.ConnectionWorker.doneCallback(ConnectionWorker.java:1785)
	at com.google.cloud.bigquery.storage.v1.ConnectionWorker.access$900(ConnectionWorker.java:73)
	at com.google.cloud.bigquery.storage.v1.ConnectionWorker$3.run(ConnectionWorker.java:739)
	at com.google.cloud.bigquery.storage.v1.StreamConnection$1.onError(StreamConnection.java:86)
	at com.google.api.gax.tracing.TracedResponseObserver.onError(TracedResponseObserver.java:104)
	at com.google.api.gax.grpc.ExceptionResponseObserver.onErrorImpl(ExceptionResponseObserver.java:84)
	at com.google.api.gax.rpc.StateCheckingResponseObserver.onError(StateCheckingResponseObserver.java:84)
	at com.google.api.gax.grpc.GrpcDirectStreamController$ResponseObserverAdapter.onClose(GrpcDirectStreamController.java:148)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:651)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at com.google.api.gax.grpc.GrpcLoggingInterceptor$1$1.onClose(GrpcLoggingInterceptor.java:98)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at io.grpc.census.CensusStatsModule$StatsClientInterceptor$1$1.onClose(CensusStatsModule.java:814)
	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
	at io.grpc.census.CensusTracingModule$TracingClientInterceptor$1$1.onClose(CensusTracingModule.java:494)
	at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:500)
	at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:464)
	at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:497)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:566)
	at io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:72)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:734)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:715)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
self = <apache_beam.io.external.xlang_bigqueryio_it_test.BigQueryXlangStorageWriteIT testMethod=test_write_to_dynamic_destinations_with_dynamic_schema>

    def test_write_to_dynamic_destinations_with_dynamic_schema(self):
      base_table_spec = '{}.dynamic_dest_dyn_schema_'.format(self.dataset_id)
      spec_with_project = '{}:{}'.format(self.project, base_table_spec)
      table_id_a = 'dynamic_dest_dyn_schema_users'
      table_id_b = 'dynamic_dest_dyn_schema_scores'
      table_a = base_table_spec + 'users'
      table_b = base_table_spec + 'scores'
    
      schema_a = "id:INTEGER,name:STRING"
      schema_b = "id:INTEGER,score:INTEGER,active:BOOLEAN"
    
      # Pre-create destination tables with their distinct specific schemas prior
      # to pipeline execution to ensure tables only contain their specific fields.
      self.bigquery_client.get_or_create_table(
          project_id=self.project,
          dataset_id=self.dataset_id,
          table_id=table_id_a,
          schema=bigquery_tools.get_table_schema_from_string(schema_a),
          create_disposition='CREATE_IF_NEEDED',
          write_disposition='WRITE_APPEND')
      self.bigquery_client.get_or_create_table(
          project_id=self.project,
          dataset_id=self.dataset_id,
          table_id=table_id_b,
          schema=bigquery_tools.get_table_schema_from_string(schema_b),
          create_disposition='CREATE_IF_NEEDED',
          write_disposition='WRITE_APPEND')
    
      elements_a = [
          {
              'id': 1, 'name': 'alice'
          },
          {
              'id': 2, 'name': 'bob'
          },
      ]
      elements_b = [
          {
              'id': 101, 'score': 95, 'active': True
          },
          {
              'id': 102, 'score': 80, 'active': False
          },
      ]
      elements = elements_a + elements_b
    
      schema_map = {
          spec_with_project + 'users': schema_a,
          spec_with_project + 'scores': schema_b,
      }
    
      bq_matchers = [
          BigqueryFullResultMatcher(
              project=self.project,
              query="SELECT * FROM %s" % table_a,
              data=self.parse_expected_data(elements_a)),
          BigqueryFullResultMatcher(
              project=self.project,
              query="SELECT * FROM %s" % table_b,
              data=self.parse_expected_data(elements_b)),
      ]
    
      def get_destination(record):
        if 'name' in record:
          return spec_with_project + 'users'
        return spec_with_project + 'scores'
    
      def get_schema_raw(dest, side_map):
        return side_map[dest]
    
      get_schema = bigquery.dynamic_schema(
          get_schema_raw,
          union_schema="id:INTEGER,name:STRING,score:INTEGER,active:BOOLEAN")
    
>     with beam.Pipeline(argv=self.args) as p:
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

apache_beam/io/external/xlang_bigqueryio_it_test.py:562: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/pipeline.py:652: in __exit__
    self.result = self.run()
                  ^^^^^^^^^^
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:629: in _run_internal
    return self.runner.run_pipeline(self, self._options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/dataflow/test_dataflow_runner.py:66: in run_pipeline
    self.result.wait_until_finish(duration=wait_duration)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <DataflowPipelineResult id: "2026-08-12_12_19_54-2769580264293846910"
project_id: "apache-beam-testing"
name: "beamapp...y_name: "Apache Beam Python 3.14 SDK"
  }
}
start_time {
  seconds: 1786562395
  nanos: 221277000
}
 at 0x7b4f59a2bf00>
duration = None

    def wait_until_finish(self, duration=None):
      if not self.is_in_terminal_state():
        if not self.has_job:
          raise IOError('Failed to get the Dataflow job id.')
        gcp_options = self._options.view_as(GoogleCloudOptions)
        consoleUrl = (
            "Console URL: https://console.cloud.google.com/"
            f"dataflow/jobs/{gcp_options.region}/{self.job_id()}"
            f"?project={gcp_options.project}")
        thread = threading.Thread(
            target=DataflowRunner.poll_for_job_completion,
            args=(self._runner, self, duration))
    
        # Mark the thread as a daemon thread so a keyboard interrupt on the main
        # thread will terminate everything. This is also the reason we will not
        # use thread.join() to wait for the polling thread.
        thread.daemon = True
        thread.start()
        while thread.is_alive():
          time.sleep(5.0)
    
        # TODO: Merge the termination code in poll_for_job_completion and
        # is_in_terminal_state.
        terminated = self.is_in_terminal_state()
        assert duration or terminated, (
            'Job did not reach to a terminal state after waiting indefinitely. '
            '{}'.format(consoleUrl))
    
        if terminated and self.state != PipelineState.DONE:
          # TODO(BEAM-1290): Consider converting this to an error log based on
          # theresolution of the issue.
          _LOGGER.error(consoleUrl)
>         raise DataflowRuntimeException(
              'Dataflow pipeline failed. State: %s, Error:\n%s' %
              (self.state, getattr(self._runner, 'last_error_msg', None)),
E             apache_beam.runners.dataflow.dataflow_runner.DataflowRuntimeException: Dataflow pipeline failed. State: FAILED, Error:
E             org.apache.beam.sdk.util.UserCodeException: java.lang.RuntimeException: Append to stream projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ failed with Status Code INVALID_ARGUMENT. The stream may not exist.
E             	at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
E             	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn$DoFnInvoker$bf057275.invokeFinishBundle(Unknown Source)
E             	at org.apache.beam.fn.harness.FnApiDoFnRunner.finishBundle(FnApiDoFnRunner.java:1317)
E             	at org.apache.beam.fn.harness.data.PTransformFunctionRegistry.lambda$register$0(PTransformFunctionRegistry.java:115)
E             	at org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:567)
E             	at org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:150)
E             	at org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:115)
E             	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
E             	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
E             	at org.apache.beam.sdk.util.UnboundedScheduledExecutorService$ScheduledFutureTask.run(UnboundedScheduledExecutorService.java:164)
E             	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
E             	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
E             	at java.base/java.lang.Thread.run(Thread.java:840)
E             Caused by: java.lang.RuntimeException: Append to stream projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ failed with Status Code INVALID_ARGUMENT. The stream may not exist.
E             	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn$DestinationState.lambda$flush$9(StorageApiWriteUnshardedRecords.java:856)
E             	at org.apache.beam.sdk.io.gcp.bigquery.RetryManager.await(RetryManager.java:319)
E             	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.flushAll(StorageApiWriteUnshardedRecords.java:1069)
E             	at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.finishBundle(StorageApiWriteUnshardedRecords.java:1257)
E             Caused by: com.google.cloud.bigquery.storage.v1.Exceptions$SchemaMismatchedException: INVALID_ARGUMENT: Input schema has more fields than BigQuery schema, extra fields: 'name' Entity: projects/apache-beam-testing/datasets/python_xlang_storage_write_1786562378_399e18/tables/dynamic_dest_dyn_schema_scores/streams/Cic2YjVhZGUwMC0wMDAwLTI3MDQtYWNiYi0yNDA1ODg3N2FhNTQ6czQ
E             	at com.google.cloud.bigquery.storage.v1.Exceptions.toStorageException(Exceptions.java:200)
E             	at com.google.cloud.bigquery.storage.v1.Exceptions.toStorageException(Exceptions.java:222)
E             	at com.google.cloud.bigquery.storage.v1.ConnectionWorker.doneCallback(ConnectionWorker.java:1785)
E             	at com.google.cloud.bigquery.storage.v1.ConnectionWorker.access$900(ConnectionWorker.java:73)
E             	at com.google.cloud.bigquery.storage.v1.ConnectionWorker$3.run(ConnectionWorker.java:739)
E             	at com.google.cloud.bigquery.storage.v1.StreamConnection$1.onError(StreamConnection.java:86)
E             	at com.google.api.gax.tracing.TracedResponseObserver.onError(TracedResponseObserver.java:104)
E             	at com.google.api.gax.grpc.ExceptionResponseObserver.onErrorImpl(ExceptionResponseObserver.java:84)
E             	at com.google.api.gax.rpc.StateCheckingResponseObserver.onError(StateCheckingResponseObserver.java:84)
E             	at com.google.api.gax.grpc.GrpcDirectStreamController$ResponseObserverAdapter.onClose(GrpcDirectStreamController.java:148)
E             	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
E             	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
E             	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
E             	at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:651)
E             	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
E             	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
E             	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
E             	at com.google.api.gax.grpc.GrpcLoggingInterceptor$1$1.onClose(GrpcLoggingInterceptor.java:98)
E             	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
E             	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
E             	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
E             	at io.grpc.census.CensusStatsModule$StatsClientInterceptor$1$1.onClose(CensusStatsModule.java:814)
E             	at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39)
E             	at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23)
E             	at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40)
E             	at io.grpc.census.CensusTracingModule$TracingClientInterceptor$1$1.onClose(CensusTracingModule.java:494)
E             	at io.grpc.internal.DelayedClientCall$DelayedListener$3.run(DelayedClientCall.java:500)
E             	at io.grpc.internal.DelayedClientCall$DelayedListener.delayOrExecute(DelayedClientCall.java:464)
E             	at io.grpc.internal.DelayedClientCall$DelayedListener.onClose(DelayedClientCall.java:497)
E             	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:566)
E             	at io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:72)
E             	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:734)
E             	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:715)
E             	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
E             	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
E             	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
E             	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
E             	at java.base/java.lang.Thread.run(Thread.java:840)

apache_beam/runners/dataflow/dataflow_runner.py:834: DataflowRuntimeException