address gemini comment

Author: derrickaw

sdks/python/apache_beam/yaml/yaml_mapping.py

@@ -243,25 +243,29 @@ def __init__(self, fields, original_fields, input_schema):
     self.passthrough_fields = []
 
     script = []
-    for name, expr in fields.items():
+    for i, (name, expr) in enumerate(fields.items()):
       if isinstance(expr, str) and expr in input_schema:
         self.passthrough_fields.append((name, expr))
         continue
 
       if isinstance(expr, str):
         expr = {'expression': expr}
 
+      # We use numeric indexing (func_{i}) instead of reusing the output field
+      # name to prevent syntax errors if output names contain spaces or hyphens.
+      # We also use bracket notation for robustness against input field names
+      # that aren't compliant dot-access identifiers.
       if 'expression' in expr:
         e = expr['expression']
-        code = f"var func_{name} = (__row__) => {{ " + " ".join(
-            [f"const {n} = __row__.{n};"
+        code = f"var func_{i} = (__row__) => {{ " + " ".join(
+            [f"const {n} = __row__['{n}'];"
              for n in original_fields if n in e]) + f" return ({e}); }}"
         script.append(code)
-        self.field_funcs[name] = f"func_{name}"
+        self.field_funcs[name] = f"func_{i}"
       elif 'callable' in expr:
-        code = f"var func_{name} = {expr['callable']}"
+        code = f"var func_{i} = {expr['callable']}"
         script.append(code)
-        self.field_funcs[name] = f"func_{name}"
+        self.field_funcs[name] = f"func_{i}"
       elif 'path' in expr and 'name' in expr:
         path = expr['path']
         func_name = expr['name']
@@ -320,8 +324,11 @@ def _get_javascript_udf_code(
     udf_code = FileSystems.open(path).read().decode()
     return udf_code, name
   elif expression:
+    # We use bracket notation for robustness against field names that
+    # aren't compliant dot-access identifiers.
     udf_code = f"var {function_name} = (__row__) => {{ " + " ".join([
-        f"const {n} = __row__.{n};" for n in original_fields if n in expression
+        f"const {n} = __row__['{n}'];"
+        for n in original_fields if n in expression
     ]) + f" return ({expression}); }}"
     return udf_code, function_name
   elif callable:

sdks/python/apache_beam/yaml/yaml_udf_test.py

@@ -397,7 +397,37 @@ def test_map_to_fields_js_date(self):
       expected_date = datetime.datetime(
           2026, 4, 17, 18, tzinfo=datetime.timezone.utc)
 
-      assert_that(result | as_rows(), equal_to([beam.Row(date=expected_date)]))
+  @unittest.skipIf(MiniRacer is None, 'py_mini_racer not installed.')
+  def test_map_to_fields_js_special_names(self):
+    with beam.Pipeline(options=beam.options.pipeline_options.PipelineOptions(
+        pickle_library='cloudpickle', yaml_experimental_features=['javascript'
+                                                                  ])) as p:
+      elements = p | beam.Create([beam.Row(label='test')])
+      result = elements | YamlTransform(
+          '''
+      type: MapToFields
+      config:
+        language: javascript
+        fields:
+          'weird output-name':
+            expression: "label + '-ok'"
+      ''')
+
+      # Verify that it yields a single row with the transformed output.
+      # We use as_dict here because comparing typed Row with spaces can fail
+      # downstream assertions if mapped back to a python tuple.
+      def row_to_dict(r):
+        return {
+            k: getattr(r, k)
+            for k in dir(r)
+            if not k.startswith('_') and not callable(getattr(r, k))
+        }
+
+      assert_that(
+          result | beam.Map(lambda r: dict(r._asdict())),
+          equal_to([{
+              'weird output-name': 'test-ok'
+          }]))
 
 
 if __name__ == '__main__':