normalize Spanner CDC

.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "revision": 3,
+  "revision": 4,
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

sdks/python/apache_beam/yaml/extended_tests/databases/spanner_cdc.yaml

@@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+fixtures:
+  - name: SPANNER_CDC
+    type: "apache_beam.yaml.integration_tests.temp_spanner_change_stream"
+    config:
+      project: "apache-beam-testing"
+      instance: "beam-test"
+      num_rows: 2
+
+pipelines:
+  - name: read_spanner_cdc_bounded
+    pipeline:
+      type: composite
+      transforms:
+        - type: ReadFromSpannerCDC
+          name: ReadCdc
+          config:
+            project: "{SPANNER_CDC[PROJECT]}"
+            instance: "{SPANNER_CDC[INSTANCE]}"
+            database: "{SPANNER_CDC[DATABASE]}"
+            table: "{SPANNER_CDC[TABLE]}"
+            change_stream: "{SPANNER_CDC[CHANGE_STREAM]}"
+            start_at: "{SPANNER_CDC[START_AT]}"
+            end_at: "{SPANNER_CDC[END_AT]}"
+        - type: MapToFields
+          name: SelectOperation
+          input: ReadCdc
+          config:
+            language: python
+            fields:
+              operation: operation
+        - type: AssertEqual
+          input: SelectOperation
+          config:
+            elements:
+              - {operation: INSERT}
+              - {operation: INSERT}
+              - {operation: UPDATE}
+              - {operation: UPDATE}
+              - {operation: DELETE}
+              - {operation: DELETE}
+        - type: AssertEqual
+          input: ReadCdc.errors
+          config:
+            elements: []
+    options:
+      project: "apache-beam-testing"
+      streaming: true

sdks/python/apache_beam/yaml/integration_tests.py

@@ -165,6 +165,102 @@ def temp_spanner_table(project, prefix='temp_spanner_db_'):
     spanner_client._delete_database()
 
 
+def _spanner_commit_timestamp_str(commit_ts):
+  """Format a Spanner commit timestamp for SchemaTransform config fields."""
+  text = commit_ts.isoformat()
+  if text.endswith('+00:00'):
+    return text[:-6] + 'Z'
+  return text
+
+
+@contextlib.contextmanager
+def temp_spanner_change_stream(
+    project='apache-beam-testing', instance='beam-test', num_rows=2):
+  """Context manager for a Spanner table with a change stream and seeded mutations.
+
+  Creates a temporary database, table, and change stream, then inserts, updates,
+  and deletes ``num_rows`` rows. Yields identifiers and inclusive ``start_at`` /
+  ``end_at`` timestamps suitable for a bounded ``ReadFromSpannerCDC`` YAML IT.
+
+  Args:
+    project (str): GCP project id.
+    instance (str): Spanner instance id.
+    num_rows (int): Number of rows to insert/update/delete.
+
+  Yields:
+    dict: Keys ``PROJECT``, ``INSTANCE``, ``DATABASE``, ``TABLE``,
+      ``CHANGE_STREAM``, ``START_AT``, and ``END_AT``.
+  """
+  from google.cloud import spanner
+
+  client = spanner.Client(project=project)
+  spanner_instance = client.instance(instance)
+  database_id = f'temp-{uuid.uuid4().hex[:10]}'
+  # Unquoted GSQL identifiers must start with a letter.
+  table = f'T{uuid.uuid4().hex[:10]}'
+  change_stream = f'C{uuid.uuid4().hex[:10]}'
+
+  database = spanner_instance.database(
+      database_id,
+      ddl_statements=[
+          f'''CREATE TABLE {table} (
+            SingerId INT64 NOT NULL,
+            FirstName STRING(1024),
+            LastName STRING(1024)
+          ) PRIMARY KEY (SingerId)'''
+      ])
+  logging.info(
+      'Creating Spanner CDC test database %s table %s', database_id, table)
+  database.create().result(timeout=600)
+
+  logging.info('Creating change stream %s for table %s', change_stream, table)
+  database.update_ddl([f'CREATE CHANGE STREAM {change_stream} FOR {table}'
+                       ]).result(timeout=600)
+
+  def _commit_mutation(build_batch_fn):
+    batch = database.batch()
+    build_batch_fn(batch)
+    return batch.commit()
+
+  start_at = None
+  for singer_id in range(1, num_rows + 1):
+    commit_ts = _commit_mutation(
+        lambda batch, singer_id=singer_id: batch.insert(
+            table=table, columns=('SingerId', 'FirstName', 'LastName'), values=[
+                (singer_id, f'First {singer_id}', f'Last {singer_id}'), ]))
+    if start_at is None:
+      start_at = commit_ts
+
+  for singer_id in range(1, num_rows + 1):
+    _commit_mutation(
+        lambda batch, singer_id=singer_id: batch.update(
+            table=table, columns=('SingerId', 'FirstName', 'LastName'), values=[
+                (
+                    singer_id,
+                    f'Updated First {singer_id}',
+                    f'Updated Last {singer_id}', ), ]))
+
+  end_at = None
+  for singer_id in range(1, num_rows + 1):
+    end_at = _commit_mutation(
+        lambda batch, singer_id=singer_id: batch.delete(
+            table=table, keyset=spanner.KeySet(keys=[[singer_id]])))
+
+  try:
+    yield {
+        'PROJECT': project,
+        'INSTANCE': instance,
+        'DATABASE': database_id,
+        'TABLE': table,
+        'CHANGE_STREAM': change_stream,
+        'START_AT': _spanner_commit_timestamp_str(start_at),
+        'END_AT': _spanner_commit_timestamp_str(end_at),
+    }
+  finally:
+    logging.info('Deleting Spanner CDC test database: %s', database_id)
+    database.drop()
+
+
 @contextlib.contextmanager
 def temp_bigquery_table(project, prefix='yaml_bq_it_'):
   """Context manager to create and clean up a temporary BigQuery dataset.

sdks/python/apache_beam/yaml/standard_io.yaml

@@ -514,6 +514,27 @@
       config:
         gradle_target: 'sdks:java:io:google-cloud-platform:expansion-service:shadowJar'
 
+# Spanner CDC
+- type: renaming
+  transforms:
+    'ReadFromSpannerCDC': 'ReadFromSpannerCDC'
+  config:
+    mappings:
+      'ReadFromSpannerCDC':
+        project: 'project_id'
+        instance: 'instance_id'
+        database: 'database_id'
+        table: 'table'
+        change_stream: 'change_stream_name'
+        start_at: 'start_at_timestamp'
+        end_at: 'end_at_timestamp'
+    underlying_provider:
+      type: beamJar
+      transforms:
+        'ReadFromSpannerCDC': 'beam:schematransform:org.apache.beam:spanner_cdc_read:v1'
+      config:
+        gradle_target: 'sdks:java:io:google-cloud-platform:expansion-service:shadowJar'
+
 # TFRecord
 - type: renaming
   transforms: