beetle0915 commented on code in PR #29513:
URL: https://github.com/apache/flink/pull/29513#discussion_r4338799920
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -3389,13 +3494,38 @@ def _prepare_join(
raise ValueError("join() requires on or both left_on and right_on")
if left_on is None or right_on is None:
raise ValueError("left_on and right_on must be provided together")
- left_keys = _normalize_join_keys(left_on, "left_on")
- right_keys = _normalize_join_keys(right_on, "right_on")
+ left_keys = _normalize_join_keys(
+ left_on, "left_on", allow_expressions=asof_time is not None)
+ right_keys = _normalize_join_keys(
+ right_on, "right_on", allow_expressions=asof_time is not None)
if len(left_keys) != len(right_keys):
raise ValueError("left_on and right_on must have the same number
of keys")
_validate_join_columns(left_keys, left_columns, "left_on")
_validate_join_columns(right_keys, right_columns, "right_on")
+ if asof_time is not None:
+ # Preserve direct field references so temporal joins retain the right
primary key.
+ left_keys, right_keys = list(left_keys), list(right_keys)
+ for table, keys, columns in (
+ (left_table, left_keys, left_columns), (right_table, right_keys,
right_columns)
+ ):
+ for index, key in enumerate(keys):
+ if isinstance(key, Expression):
+ selected = table.select(key)
+ if len(selected.get_resolved_schema().get_column_names())
!= 1:
+ raise ValueError("each join key must select a single
column")
+ projection = selected._j_table.getQueryOperation()
+ if not projection.getChildren().get(0).equals(
+ table._j_table.getQueryOperation()
+ ):
+ raise ValueError("join keys must be row-wise
expressions, not aggregations")
+ resolved = projection.getProjectList().get(0)
+ time_index = left_columns.index(asof_time) if table is
left_table else -1
+ _asof_expression_info(resolved, time_index, len(columns))
+ field_index = _asof_field_index(resolved)
Review Comment:
Updated the paired-key handling to preserve the original field reference
through no-op casts. Both forms in your example now plan successfully. Added
regression coverage, while retaining casts that change the type or materialize
a time attribute.
--
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]
