Cherrypick #35789 for 2.67.0: Fix timestamp issue in a yaml pipeline that calls SQL transform (#35797)

Cherrypick #35789 for 2.67.0: Fix timestamp issue in a yaml pipeline that calls SQL transform

Author: Amar3tto

sdks/python/apache_beam/typehints/schemas.py

@@ -151,11 +151,21 @@ def named_fields_to_schema(
   if isinstance(names_and_types, dict):
     names_and_types = names_and_types.items()
 
+  _, cached_schema = schema_registry.by_id.get(schema_id, (None, None))
+  if cached_schema:
+    type_by_name_from_schema = {
+        field.name: field.type
+        for field in cached_schema.fields
+    }
+  else:
+    type_by_name_from_schema = {}
+
   schema = schema_pb2.Schema(
       fields=[
           schema_pb2.Field(
               name=name,
-              type=typing_to_runner_api(type),
+              type=type_by_name_from_schema.get(
+                  name, typing_to_runner_api(type)),
               options=[
                   option_to_runner_api(option_tuple)
                   for option_tuple in field_options.get(name, [])

sdks/python/apache_beam/yaml/tests/sql.yaml

@@ -76,3 +76,21 @@ pipelines:
                   - {a: "x", s: "2"}
                   - {a: "x", s: "3"}
                   - {a: "y", s: "10"}
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: Create
+          name: CreateSampleData
+          config:
+            elements:
+              - { id: 1, name: "John" }
+              - { id: 2, name: "Jane" }
+        - type: Sql
+          name: sql
+          config:
+            query: >
+              SELECT *, CURRENT_TIMESTAMP AS ingest_timestamp FROM PCOLLECTION
+        - type: PyTransform
+          config:
+            constructor: apache_beam.transforms.util.LogElements