Support BATCH message stream aliasing via stream maps

tap_bigquery/client.py

@@ -16,7 +16,7 @@
 from google.cloud import bigquery
 from singer_sdk import SQLStream
 from singer_sdk.helpers import types
-from singer_sdk.helpers._batch import JSONLinesEncoding
+from singer_sdk.helpers._batch import JSONLinesEncoding, SDKBatchMessage
 
 from tap_bigquery.connector import BigQueryConnector
 
@@ -136,3 +136,14 @@ def get_records(self, partition: dict | None) -> Iterable[dict[str, Any]]:
             return None
 
         yield from super().get_records(partition)
+
+    def _write_batch_message(self, encoding, manifest):
+        for stream_map in self.stream_maps:
+            self._tap.write_message(
+                SDKBatchMessage(
+                    stream=stream_map.stream_alias,
+                    encoding=encoding,
+                    manifest=manifest,
+                ),
+            )
+        self._is_state_flushed = False