[vishalmore90]

Fixes #41238

Description

In Beam YAML, YamlProviders.Flatten._merge_schemas() handles schema unification when flattening multiple PCollections. Previously:

1. Field names were collected using set().union(*(s.keys() for s in schemas)), causing non-deterministic schema field ordering across Python runs.
2. Mixing optional and non-optional types of the same inner type (e.g. int and Optional[int]) was treated as a type collision and degraded the unified field type to Optional[Any].
3. Fields were unconditionally typed as Optional[...], even when present and required across all input schemas.
4. Input PCollections with identical schemas were still subjected to synthetic schema creation and per-element row conversion via _unify_element_with_schema.

This PR fixes these issues by:

• Returning None early from _merge_schemas if all PCollections share identical schemas, allowing beam.Flatten to execute natively without row conversion overhead.
• Using dict.fromkeys(...) to preserve deterministic field ordering.
• Unwrapping optional types using _match_is_optional and extract_optional_type to unify types with compatible bases (e.g., int + Optional[int] -> Optional[int]).
• Keeping fields non-optional when they are present and non-optional across all input schemas.

Checklist

• Description of change and rationale included above.
• Mentioned the appropriate issue (Fixes #41238).
• Updated CHANGES.md under ## Bugfixes.
• Added unit tests covering identical schemas, deterministic ordering, optional unwrapping, required field preservation, and conflicting type fallback.
• Verified code formatting with yapf and pycodestyle.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.75%. Comparing base (bc9e036) to head (82fcde3).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41239      +/-   ##
============================================
- Coverage     57.77%   57.75%   -0.03%     
  Complexity     2617     2617              
============================================
  Files          1202     1202              
  Lines        171710   171800      +90     
  Branches       2163     2163              
============================================
+ Hits          99204    99221      +17     
- Misses        69845    69918      +73     
  Partials       2661     2661              

Flag	Coverage Δ
python	80.50% <100.00%> (-0.07%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @derrickaw for label yaml.
Backup reviewer: chamikaramj (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[derrickaw]

/gemini review

[github-actions]

📋 Review Summary

This PR fixes a bug in YAML Flatten schema unification where fields already present across all input schemas with identical types were unnecessarily wrapped in Optional, and where schemas with mixed Optional[T] and T for the same field collapsed to Optional[Any] (a non-deterministic coder) instead of Optional[T]. This prevented downstream transforms such as Combine / GroupByKey from using those fields as grouping keys.

🔍 General Feedback

• The schema merging logic in _merge_schemas cleanly uses native_type_compatibility.extract_optional_type to normalize base types and track field presence/nullability across input PCollections.
• The added unit and E2E tests thoroughly cover non-nullable preservation, mixed optionality, missing fields, conflicting types, and the Flatten -> Combine grouping scenario.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[vishalmore90]

PTAL @derrickaw