vishalmore90 opened a new issue, #41052:
URL: https://github.com/apache/beam/issues/41052
### What happened?
When joining three or more PCollections in Beam YAML using `type: Join`,
`_SqlJoinTransform` in `sdks/python/apache_beam/yaml/yaml_join.py` generates
invalid SQL queries whenever the join graph topology or the dictionary key
ordering in `input` does not strictly follow a linear chain originating at
`tables[0]`.
Depending on the input key ordering and equality relationships,
`_SqlJoinTransform` exhibits two fatal failure modes:
1. **Missing `ON` clause (`INNER JOIN <table>` without `ON`)**: When an
intermediate table joins to a subsequent table rather than the first table, its
join condition is never populated. `_SqlJoinTransform` produces raw SQL
containing `... INNER JOIN tableB INNER JOIN tableC ON ...`, omitting the
required `ON` clause for `tableB`. Calcite's SQL parser rejects this
immediately with `SqlParseException: Encountered "INNER" ... Was expecting one
of: "ON" ... "USING" ...`.
2. **Forward table reference in `ON` clause**: When a table is joined before
one of its dependency tables, `_SqlJoinTransform` generates an `ON` clause
referencing a table that has not yet been introduced in the `FROM`/`JOIN`
sequence (e.g. `FROM A INNER JOIN B ON C.id = B.id INNER JOIN C ON ...`). In
Calcite/standard SQL, relation `C` is not yet in scope when joining `B`,
causing query validation to fail with `Table 'C' not found` or `Column 'C.id'
not found`.
#### Code Pointers / Root Cause
In `sdks/python/apache_beam/yaml/yaml_join.py` (lines 263–301):
```python
prev_table = tables[0]
join_conditions = {}
for i in range(1, len(tables)):
curr_table = tables[i]
join_type = generate_join_type(prev_table, curr_table)
join_conditions[curr_table] = f' {join_type} JOIN {curr_table}'
prev_table = curr_table
for equality in equalities_in_pairs:
left, right = equality.keys()
if left in conditioned and right in conditioned:
t = tables[max(tables.index(left), tables.index(right))]
join_conditions[t] = (
f'{join_conditions[t]} '
f'AND {left}.{equality[left]} = {right}.{equality[right]}')
elif left in conditioned:
join_conditions[right] = (
f'{join_conditions[right]} '
f'ON {left}.{equality[left]} = {right}.{equality[right]}')
conditioned.append(right)
elif right in conditioned:
join_conditions[left] = (
f'{join_conditions[left]} '
f'ON {left}.{equality[left]} = {right}.{equality[right]}')
conditioned.append(left)
else:
t = tables[max(tables.index(left), tables.index(right))]
join_conditions[t] = (
f'{join_conditions[t]} '
f'ON {left}.{equality[left]} = {right}.{equality[right]}')
conditioned.append(t)
if fields:
selects = ', '.join(_parse_fields(tables, fields))
else:
selects = '*'
query = f'SELECT {selects} FROM {first_table}'
query += ' '.join(condition for condition in join_conditions.values())
return pcolls | sql_transform_constructor(query)
```
1. `tables = list(pcolls.keys())` assumes tables can be joined in the
arbitrary order defined by Python `dict` iteration order of `input`.
2. `join_conditions` is pre-populated in fixed `tables[1:]` order with `'
{join_type} JOIN {curr_table}'`.
3. If an equality touches `left` and `right` where neither is in
`conditioned` (`[tables[0]]`), it appends `ON` only to `max(tables.index(left),
tables.index(right))`. The other table is never given an `ON` clause if
subsequent conditions also attach to `max(...)`.
4. As a result, `join_conditions.values()` contains entries without any `ON`
clause, or references tables that appear later in `tables`.
#### Steps to Reproduce
Consider three related tables: `products`, `orders`, and `line_items`, where
`orders` links to `line_items` via `order_id`, and `line_items` links to
`products` via `product_id`.
```yaml
pipeline:
transforms:
- type: Create
name: Products
config:
elements:
- product_id: 101
product_name: "Widget"
- type: Create
name: Orders
config:
elements:
- order_id: 1
customer: "Alice"
- type: Create
name: LineItems
config:
elements:
- order_id: 1
product_id: 101
qty: 2
- type: Join
name: JoinOrdersAndProducts
input:
products: Products
orders: Orders
line_items: LineItems
config:
type: inner
equalities:
- orders: order_id
line_items: order_id
- line_items: product_id
products: product_id
```
Run this YAML pipeline (or evaluate `_SqlJoinTransform` with these inputs):
**Failure Mode 1 Output:**
Because `input` keys are `['products', 'orders', 'line_items']`:
- Initial `conditioned = ['products']`.
- Condition 1 (`orders.order_id = line_items.order_id`): neither `orders`
nor `line_items` is in `conditioned`. Hits `else:`, attaches `ON` to
`line_items` (index 2). `conditioned` becomes `['products', 'line_items']`.
- Condition 2 (`line_items.product_id = products.product_id`): both are in
`conditioned`. Hits `left in conditioned and right in conditioned:`, attaches
`AND` to `line_items`.
- `join_conditions['orders']` is never assigned an `ON` clause!
Generated SQL:
```sql
SELECT * FROM products INNER JOIN orders INNER JOIN line_items ON
orders.order_id = line_items.order_id AND line_items.product_id =
products.product_id
```
Notice `INNER JOIN orders` has no `ON` clause, causing:
```
org.apache.beam.vendor.calcite.v1_28_0.org.apache.calcite.sql.parser.SqlParseException:
Encountered "INNER" at ... Was expecting one of: "ON" ... "USING" ...
```
**Failure Mode 2 Output:**
If the input dictionary order is `['orders', 'products', 'line_items']`:
Generated SQL:
```sql
SELECT * FROM orders INNER JOIN products ON line_items.product_id =
products.product_id INNER JOIN line_items ON orders.order_id =
line_items.order_id
```
Notice `INNER JOIN products ON line_items.product_id = products.product_id`
references `line_items` before `line_items` has been joined.
#### Proposed Solution
Instead of assuming `tables = list(pcolls.keys())` defines a valid SQL join
order:
1. Build an adjacency graph of tables and join conditions from
`equalities_in_pairs`.
2. Traverse the graph starting from `tables[0]` (e.g. via Breadth-First
Search / topological traversal) to discover tables in an order where every
newly joined table has an equality predicate connecting it to at least one
already joined table.
3. Emit each `JOIN <table_i> ON <conditions connecting table_i to previously
joined tables>` in that traversal order.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [x] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
