beetle0915 commented on code in PR #29513:
URL: https://github.com/apache/flink/pull/29513#discussion_r4338796452
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -3456,6 +3586,90 @@ def _prepare_join(
)
+def _asof_field_index(expression) -> Optional[int]:
+ if expression.getClass().getSimpleName() == "FieldReferenceExpression":
+ return expression.getFieldIndex()
+ if expression.getClass().getSimpleName() == "RexNodeExpression":
+ node = expression.getRexNode()
+ if node.getKind().name() == "INPUT_REF":
+ return node.getIndex()
+ return None
+
+
+def _prepare_asof_time(table: Table, on, right_columns: List[str]) ->
Tuple[Table, str]:
+ expression = table_col(on) if isinstance(on, str) else on
+ selected = table.select(expression)
+ projection = selected._j_table.getQueryOperation()
+ if not
projection.getChildren().get(0).equals(table._j_table.getQueryOperation()):
+ raise ValueError("on must be a row-wise time attribute, not an
aggregation")
+ schema = selected._j_table.getResolvedSchema()
+ checks =
get_gateway().jvm.org.apache.flink.table.types.logical.utils.LogicalTypeChecks
+ if schema.getColumnCount() != 1 or not checks.isTimeAttribute(
+ schema.getColumnDataTypes().get(0).getLogicalType()
+ ):
+ raise ValueError("on must be a left processing-time or event-time
attribute")
+ resolved = projection.getProjectList().get(0)
+ columns = list(table.get_resolved_schema().get_column_names())
+ _asof_expression_info(resolved, -1, len(columns))
+ field_index = _asof_field_index(resolved)
+ if field_index is not None:
+ return table, columns[field_index]
+ name = _unique_name("__pf_asof_time", set(columns) | set(right_columns))
+ return table.add_columns(expression.alias(name)), name
+
+
+def _asof_expression_info(expression, time_index: int, left_arity: int) ->
Tuple[Set[int], bool]:
+ class_name = expression.getClass().getSimpleName()
+ if class_name == "RexNodeExpression":
+ return _asof_expression_info(expression.getRexNode(), time_index,
left_arity)
+ if class_name == "FieldReferenceExpression":
Review Comment:
Updated the validation to restrict time attributes only in equality keys.
Residual predicates remain in the join condition, preserving LEFT JOIN
semantics. Added regression coverage for time-based residual predicates and
unmatched left rows.
--
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]
