diff --git a/flink-python/pyflink/dataframe/dataframe.py b/flink-python/pyflink/dataframe/dataframe.py index 426e3a5640d68c..b6f13da715fdf0 100644 --- a/flink-python/pyflink/dataframe/dataframe.py +++ b/flink-python/pyflink/dataframe/dataframe.py @@ -38,7 +38,7 @@ from pyflink.common import Row from pyflink.dataframe.datatype import _INT_MAX, DataType from pyflink.java_gateway import get_gateway -from pyflink.table.expression import Expression +from pyflink.table.expression import Expression, _get_java_expression from pyflink.table.expressions import ( and_, call_sql, @@ -47,6 +47,7 @@ ) from pyflink.table.table import Table from pyflink.util.api_stability_decorators import PublicEvolving +from pyflink.util.java_utils import to_jarray __all__ = ["DataFrame", "GroupedDataFrame", "col", "lit"] @@ -622,6 +623,119 @@ def top_n( distinct = drop_duplicates unique = drop_duplicates + # ======================== Joins ======================== + + @PublicEvolving() + def join( + self, + other: "DataFrame", + *, + on=None, + how: str = "inner", + left_on=None, + right_on=None, + ) -> "DataFrame": + """ + Join this DataFrame with another DataFrame. + + Use ``on`` when both sides share the same named join keys, or pass a boolean expression as + the complete join predicate. A predicate must contain at least one equality condition + between the inputs. Use ``left_on`` and ``right_on`` together when the key names differ. + Shared named keys occur once in the result; other duplicate column names must be renamed + before joining. ``semi`` and ``anti`` joins return only columns from this DataFrame, while + ``cross`` performs a Cartesian product and accepts no join keys. A ``semi`` join requires + shared named keys through ``on`` or keys through ``left_on``/``right_on`` instead of a + boolean predicate. + + :param other: DataFrame on the right side of the join. + :param on: Shared column name, list of shared column names, or a boolean join expression. + :param how: Join type: ``"inner"``, ``"left"``, ``"right"``, ``"full"``, ``"outer"``, + ``"semi"``, ``"anti"``, or ``"cross"``. + :param left_on: Column name, expression, or list of column names from this DataFrame. + :param right_on: Column name, expression, or list of column names from ``other``. + :return: A new DataFrame containing the join result. + :raises TypeError: If an argument has an unsupported type. + :raises ValueError: If the join type, keys, schemas, or argument combination is invalid. + + Example:: + + >>> import pyflink.dataframe as pf + >>> orders = pf.from_records([(1, 10)], schema=["customer_id", "amount"]) + >>> customers = pf.from_records([(1, "Alice")], schema=["customer_id", "name"]) + >>> orders.join(customers, on="customer_id") + >>> orders.join( + ... customers.rename_columns({"customer_id": "id"}), + ... left_on="customer_id", + ... right_on="id", + ... how="left", + ... ) + + .. versionadded:: 2.4.0 + """ + if not isinstance(other, DataFrame): + raise TypeError("other must be a pyflink.dataframe.DataFrame") + if self._table._t_env._j_tenv != other._table._t_env._j_tenv: + raise ValueError("DataFrames must belong to the same TableEnvironment") + + join_type = _normalize_join_type(how) + if join_type == "cross": + if on is not None or left_on is not None or right_on is not None: + raise ValueError("cross join does not accept on, left_on, or right_on") + _validate_join_column_conflicts(self.columns, other.columns, set()) + return DataFrame(self._table.join(other._table)) + + ( + left_table, + right_table, + predicate, + shared_keys, + left_key_names, + right_key_names, + ) = _prepare_join( + self._table, + other._table, + on, + left_on, + right_on, + validate_column_conflicts=join_type not in ("semi", "anti"), + ) + + if join_type == "semi": + if not left_key_names: + raise ValueError( + "semi join requires named keys through on or left_on/right_on" + ) + return DataFrame( + _build_semi_join_sql( + left_table, + right_table, + self.columns, + left_key_names, + right_key_names, + ) + ) + if join_type == "anti": + return DataFrame( + _build_anti_join_sql( + left_table, + right_table, + predicate, + self.columns, + ) + ) + + return DataFrame( + _build_regular_join_sql( + left_table, + right_table, + predicate, + self.columns, + other.columns, + shared_keys, + join_type, + ) + ) + # ======================== Filtering & Ordering ======================== @PublicEvolving() @@ -1303,6 +1417,327 @@ def agg(self, *aggs: Expression, **named_aggs: Expression) -> DataFrame: # ======================== Internal Helpers ======================== +def _normalize_join_type(how: str) -> str: + if not isinstance(how, str): + raise TypeError("how must be a string") + aliases = { + "inner": "inner", + "left": "left", + "right": "right", + "full": "full", + "outer": "full", + "semi": "semi", + "anti": "anti", + "cross": "cross", + } + if how not in aliases: + raise ValueError( + 'how must be one of "inner", "left", "right", "full", "outer", ' + '"semi", "anti", or "cross"' + ) + return aliases[how] + + +def _normalize_join_keys(value, parameter_name: str) -> List[Union[str, Expression]]: + if isinstance(value, (str, Expression)): + return [value] + if isinstance(value, list): + if not value: + raise ValueError("%s must not be empty" % parameter_name) + if not all(isinstance(key, str) for key in value): + raise TypeError( + "%s must be a string, an expression, or a list of strings" % parameter_name + ) + if len(set(value)) != len(value): + raise ValueError("%s must not contain duplicate column names" % parameter_name) + return value + raise TypeError( + "%s must be a string, an expression, or a list of strings" % parameter_name + ) + + +def _validate_join_columns( + keys: List[Union[str, Expression]], columns: List[str], parameter_name: str +) -> None: + for key in keys: + if isinstance(key, str) and key not in columns: + raise ValueError( + "%s column '%s' does not exist, available columns: %s" + % (parameter_name, key, columns) + ) + + +def _validate_join_column_conflicts( + left_columns: List[str], right_columns: List[str], shared_keys: Set[str] +) -> None: + conflicts = sorted((set(left_columns) & set(right_columns)) - shared_keys) + if conflicts: + raise ValueError( + "join() found duplicate non-key columns %s; rename them with rename_columns() " + "before joining" % conflicts + ) + + +def _prepare_join( + left_table: Table, + right_table: Table, + on, + left_on, + right_on, + *, + validate_column_conflicts: bool, +) -> Tuple[Table, Table, Expression, Dict[str, str], List[str], List[str]]: + left_columns = list(left_table.get_resolved_schema().get_column_names()) + right_columns = list(right_table.get_resolved_schema().get_column_names()) + + if on is not None: + if left_on is not None or right_on is not None: + raise ValueError("on cannot be combined with left_on or right_on") + if isinstance(on, Expression): + if validate_column_conflicts: + _validate_join_column_conflicts(left_columns, right_columns, set()) + return left_table, right_table, on, {}, [], [] + left_keys = _normalize_join_keys(on, "on") + right_keys = list(left_keys) + _validate_join_columns(left_keys, left_columns, "on") + _validate_join_columns(right_keys, right_columns, "on") + else: + if left_on is None and right_on is None: + 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") + 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") + + shared_names = { + left_key + for left_key, right_key in zip(left_keys, right_keys) + if isinstance(left_key, str) + and isinstance(right_key, str) + and left_key == right_key + } + if validate_column_conflicts: + _validate_join_column_conflicts(left_columns, right_columns, shared_names) + + taken = set(left_columns) | set(right_columns) + shared_keys: Dict[str, str] = {} + right_rename_expressions: List[Expression] = [] + for name in left_columns: + if name in shared_names: + temporary_name = _unique_name("__pf_join_right_%s" % name, taken) + taken.add(temporary_name) + shared_keys[name] = temporary_name + right_rename_expressions.append(table_col(name).alias(temporary_name)) + left_key_names: List[str] = [] + right_key_names: List[str] = [] + left_computed_keys: List[Expression] = [] + right_computed_keys: List[Expression] = [] + for index, (left_key, right_key) in enumerate(zip(left_keys, right_keys)): + if isinstance(left_key, str): + left_key_names.append(left_key) + else: + temporary_name = _unique_name("__pf_join_left_key_%d" % index, taken) + taken.add(temporary_name) + left_key_names.append(temporary_name) + left_computed_keys.append(left_key.alias(temporary_name)) + + if isinstance(right_key, str): + right_key_names.append(shared_keys.get(right_key, right_key)) + else: + temporary_name = _unique_name("__pf_join_right_key_%d" % index, taken) + taken.add(temporary_name) + right_key_names.append(temporary_name) + right_computed_keys.append(right_key.alias(temporary_name)) + + if left_computed_keys: + left_table = left_table.add_columns(*left_computed_keys) + if right_computed_keys: + right_table = right_table.add_columns(*right_computed_keys) + if right_rename_expressions: + right_table = right_table.rename_columns(*right_rename_expressions) + + conditions = [ + table_col(left_name) == table_col(right_name) + for left_name, right_name in zip(left_key_names, right_key_names) + ] + predicate = conditions[0] if len(conditions) == 1 else and_(*conditions) + return ( + left_table, + right_table, + predicate, + shared_keys, + left_key_names, + right_key_names, + ) + + +def _serialize_join_predicate( + left_table: Table, + right_table: Table, + predicate: Expression, +) -> Tuple[str, str, str]: + left_alias, right_alias = "__pf_join_left", "__pf_join_right" + operation_tree_builder = ( + left_table._j_table.getTableEnvironment().getOperationTreeBuilder() + ) + gateway = get_gateway() + query_operations = to_jarray( + gateway.jvm.org.apache.flink.table.operations.QueryOperation, + [ + left_table._j_table.getQueryOperation(), + right_table._j_table.getQueryOperation(), + ], + ) + resolved_predicate = operation_tree_builder.resolveExpression( + _get_java_expression(predicate), query_operations + ) + + aliases = gateway.jvm.java.util.HashMap() + aliases.put(0, left_alias) + aliases.put(1, right_alias) + operation_expression_utils = ( + gateway.jvm.org.apache.flink.table.operations.utils.OperationExpressionsUtils + ) + predicate_sql = operation_expression_utils.scopeReferencesWithAlias( + aliases, resolved_predicate + ).asSerializableString() + return left_alias, right_alias, predicate_sql + + +def _build_regular_join_sql( + left_table: Table, + right_table: Table, + predicate: Expression, + left_output_columns: List[str], + right_output_columns: List[str], + shared_keys: Dict[str, str], + join_type: str, +) -> Table: + left_alias, right_alias, predicate_sql = _serialize_join_predicate( + left_table, right_table, predicate + ) + left_alias_sql = _quote_identifier(left_alias) + right_alias_sql = _quote_identifier(right_alias) + + projections = [] + for name in left_output_columns: + left_field = "%s.%s" % (left_alias_sql, _quote_identifier(name)) + if name in shared_keys and join_type in ("right", "full"): + right_field = "%s.%s" % ( + right_alias_sql, + _quote_identifier(shared_keys[name]), + ) + expression = "COALESCE(%s, %s)" % (left_field, right_field) + else: + expression = left_field + projections.append("%s AS %s" % (expression, _quote_identifier(name))) + projections.extend( + "%s.%s AS %s" + % (right_alias_sql, _quote_identifier(name), _quote_identifier(name)) + for name in right_output_columns + if name not in shared_keys + ) + + join_keyword = { + "inner": "INNER JOIN", + "left": "LEFT OUTER JOIN", + "right": "RIGHT OUTER JOIN", + "full": "FULL OUTER JOIN", + }[join_type] + query = ( + "SELECT %s FROM %s AS %s %s %s AS %s ON %s" + % ( + ", ".join(projections), + _quote_identifier(str(left_table)), + left_alias_sql, + join_keyword, + _quote_identifier(str(right_table)), + right_alias_sql, + predicate_sql, + ) + ) + return left_table._t_env.sql_query(query) + + +def _build_semi_join_sql( + left_table: Table, + right_table: Table, + output_columns: List[str], + left_key_names: List[str], + right_key_names: List[str], +) -> Table: + left_alias, right_alias = "__pf_join_left", "__pf_join_right" + left_alias_sql = _quote_identifier(left_alias) + right_alias_sql = _quote_identifier(right_alias) + select_list = ", ".join( + "%s.%s" % (left_alias_sql, _quote_identifier(name)) for name in output_columns + ) + left_keys = ", ".join( + "%s.%s" % (left_alias_sql, _quote_identifier(name)) + for name in left_key_names + ) + right_keys = ", ".join( + "%s.%s" % (right_alias_sql, _quote_identifier(name)) + for name in right_key_names + ) + if len(left_key_names) > 1: + left_keys = "(%s)" % left_keys + query = ( + "SELECT %s FROM %s AS %s WHERE %s IN (" + "SELECT %s FROM %s AS %s)" + % ( + select_list, + _quote_identifier(str(left_table)), + left_alias_sql, + left_keys, + right_keys, + _quote_identifier(str(right_table)), + right_alias_sql, + ) + ) + return left_table._t_env.sql_query(query) + + +def _build_anti_join_sql( + left_table: Table, + right_table: Table, + predicate: Expression, + output_columns: List[str], +) -> Table: + left_alias, right_alias, predicate_sql = _serialize_join_predicate( + left_table, right_table, predicate + ) + left_alias_sql = _quote_identifier(left_alias) + right_alias_sql = _quote_identifier(right_alias) + select_list = ", ".join( + "%s.%s" % (left_alias_sql, _quote_identifier(name)) for name in output_columns + ) + right_columns = list(right_table.get_resolved_schema().get_column_names()) + match_marker = _unique_name("__pf_join_match", set(right_columns)) + match_marker_sql = _quote_identifier(match_marker) + query = ( + "SELECT %s FROM %s AS %s LEFT OUTER JOIN (" + "SELECT *, TRUE AS %s FROM %s" + ") AS %s ON %s WHERE %s.%s IS NULL" + % ( + select_list, + _quote_identifier(str(left_table)), + left_alias_sql, + match_marker_sql, + _quote_identifier(str(right_table)), + right_alias_sql, + predicate_sql, + right_alias_sql, + match_marker_sql, + ) + ) + return left_table._t_env.sql_query(query) + + def _normalize_subset( subset: Union[str, List[str], None], parameter_name: str = "subset" ) -> Optional[List[str]]: diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py b/flink-python/pyflink/dataframe/tests/test_dataframe.py index f104db9242f437..c1cc378bef1baf 100644 --- a/flink-python/pyflink/dataframe/tests/test_dataframe.py +++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py @@ -921,6 +921,240 @@ def test_rename_columns_rejects_invalid_arguments(self): invalid_call() +class DataFrameJoinTests(PyFlinkDataFrameUTTestCase): + def setUp(self): + super().setUp() + self.left = pf.from_records( + [(1, "left")], + schema=["id", "left_value"], + ) + self.right = pf.from_records( + [(1, "right")], + schema=["id", "right_value"], + ) + + def test_join_on_shared_key_keeps_key_once(self): + result = self.left.join(self.right, on="id") + + self.assert_dataframe_schema( + result, + ["id", "left_value", "right_value"], + [ + TableDataTypes.BIGINT(), + TableDataTypes.STRING(), + TableDataTypes.STRING(), + ], + ) + + def test_join_supports_multiple_shared_keys(self): + left = pf.from_records( + [(1, "A", "left")], + schema=["id", "category", "left_value"], + ) + right = pf.from_records( + [(1, "A", "right")], + schema=["id", "category", "right_value"], + ) + + result = left.join(right, on=["id", "category"]) + + self.assertEqual( + result.columns, + ["id", "category", "left_value", "right_value"], + ) + self.assertEqual( + left.join(right, on=["id", "category"], how="semi").columns, + ["id", "category", "left_value"], + ) + + def test_join_supports_different_and_computed_keys(self): + right = self.right.rename_columns({"id": "right_id"}) + + named_result = self.left.join( + right, + left_on="id", + right_on="right_id", + how="left", + ) + computed_result = self.left.join( + right, + left_on=pf.col("id") + 1, + right_on=pf.col("right_id"), + ) + + for result in (named_result, computed_result): + self.assertEqual( + result.columns, + ["id", "left_value", "right_id", "right_value"], + ) + self.assertFalse(any(name.startswith("__pf_join") for name in result.columns)) + + def test_join_supports_expression_predicate(self): + right = pf.from_records( + [(1, 0, 2, "right")], + schema=["right_id", "min_id", "max_id", "right_value"], + ) + + result = self.left.join( + right, + on=(pf.col("id") == pf.col("right_id")) + & (pf.col("id") >= pf.col("min_id")) + & (pf.col("id") < pf.col("max_id")), + ) + + self.assertEqual( + result.columns, + ["id", "left_value", "right_id", "min_id", "max_id", "right_value"], + ) + + def test_join_supports_semi_anti_and_cross(self): + for how in ("semi", "anti"): + with self.subTest(how=how): + self.assertEqual( + self.left.join(self.right, on="id", how=how).columns, + ["id", "left_value"], + ) + + right = self.right.rename_columns({"id": "right_id"}) + self.assertEqual( + self.left.join(right, how="cross").columns, + ["id", "left_value", "right_id", "right_value"], + ) + + def test_join_outer_alias_matches_full_schema(self): + full = self.left.join(self.right, on="id", how="full") + outer = self.left.join(self.right, on="id", how="outer") + + self.assertEqual(full.columns, outer.columns) + self.assertEqual( + full._table.get_resolved_schema(), + outer._table.get_resolved_schema(), + ) + + def test_join_rejects_invalid_argument_combinations(self): + invalid_calls = [ + ( + "other", + lambda: self.left.join(object(), on="id"), + TypeError, + "other must be", + ), + ( + "how_type", + lambda: self.left.join(self.right, on="id", how=1), + TypeError, + "how must be a string", + ), + ( + "how_value", + lambda: self.left.join(self.right, on="id", how="sideways"), + ValueError, + "how must be one of", + ), + ( + "missing_keys", + lambda: self.left.join(self.right), + ValueError, + "requires on or both", + ), + ( + "mixed_keys", + lambda: self.left.join( + self.right, + on="id", + left_on="id", + right_on="id", + ), + ValueError, + "on cannot be combined", + ), + ( + "missing_right_on", + lambda: self.left.join(self.right, left_on="id"), + ValueError, + "must be provided together", + ), + ( + "different_key_counts", + lambda: self.left.join( + self.right, + left_on=["id", "left_value"], + right_on=["id"], + ), + ValueError, + "same number of keys", + ), + ( + "empty_keys", + lambda: self.left.join(self.right, on=[]), + ValueError, + "on must not be empty", + ), + ( + "invalid_key_type", + lambda: self.left.join(self.right, on=1), + TypeError, + "on must be a string", + ), + ( + "missing_column", + lambda: self.left.join(self.right, on="missing"), + ValueError, + "on column 'missing' does not exist", + ), + ( + "cross_keys", + lambda: self.left.join(self.right, on="id", how="cross"), + ValueError, + "cross join does not accept", + ), + ( + "semi_predicate", + lambda: self.left.join( + self.right.rename_columns({"id": "right_id"}), + on=pf.col("id") == pf.col("right_id"), + how="semi", + ), + ValueError, + "semi join requires named keys", + ), + ] + for name, invalid_call, error, message in invalid_calls: + with self.subTest(name=name): + with self.assertRaisesRegex(error, message): + invalid_call() + + def test_join_rejects_duplicate_non_key_columns(self): + right = pf.from_records( + [(1, "duplicate")], + schema=["right_id", "left_value"], + ) + + with self.assertRaisesRegex(ValueError, "duplicate non-key columns.*rename_columns"): + self.left.join(right, left_on="id", right_on="right_id") + + for how in ("semi", "anti"): + with self.subTest(how=how): + self.assertEqual( + self.left.join( + right, + left_on="id", + right_on="right_id", + how=how, + ).columns, + ["id", "left_value"], + ) + + def test_join_rejects_different_table_environments(self): + other_environment = TableEnvironment.create(EnvironmentSettings.in_batch_mode()) + other = pf.from_table( + other_environment.sql_query("SELECT 1 AS id, 'right' AS right_value") + ) + + with self.assertRaisesRegex(ValueError, "same TableEnvironment"): + self.left.join(other, on="id") + + class DataFramePropertyTests(PyFlinkDataFrameUTTestCase): def test_schema_exposes_ordered_metadata(self): dataframe = pf.from_records( @@ -2241,6 +2475,19 @@ def _nullable_dataframe(self): ) return pf.from_table(table) + def _join_dataframes(self): + left = self.t_env.sql_query( + "SELECT * FROM (VALUES " + "(CAST(1 AS INT), 'L1'), (2, 'L2'), (CAST(NULL AS INT), 'LN')) " + "AS T(id, left_value)" + ) + right = self.t_env.sql_query( + "SELECT * FROM (VALUES " + "(CAST(2 AS INT), 'R2'), (3, 'R3'), (CAST(NULL AS INT), 'RN')) " + "AS T(id, right_value)" + ) + return pf.from_table(left), pf.from_table(right) + def test_sort_returns_rows_in_ascending_order(self): self.assertEqual( self._unsorted_dataframe().sort("id").collect(), @@ -2329,6 +2576,126 @@ def test_grouped_aggregation_with_batch_table_environment(self): [Row("engineering", 30, 2), Row("sales", 5, 1)], ) + def test_join_types_and_null_keys(self): + expected = { + "inner": [Row(2, "L2", "R2")], + "left": [ + Row(1, "L1", None), + Row(2, "L2", "R2"), + Row(None, "LN", None), + ], + "right": [ + Row(2, "L2", "R2"), + Row(3, None, "R3"), + Row(None, None, "RN"), + ], + "full": [ + Row(1, "L1", None), + Row(2, "L2", "R2"), + Row(3, None, "R3"), + Row(None, "LN", None), + Row(None, None, "RN"), + ], + "outer": [ + Row(1, "L1", None), + Row(2, "L2", "R2"), + Row(3, None, "R3"), + Row(None, "LN", None), + Row(None, None, "RN"), + ], + } + + for how, expected_rows in expected.items(): + with self.subTest(how=how): + left, right = self._join_dataframes() + self.assertCountEqual( + left.join(right, on="id", how=how).collect(), + expected_rows, + ) + + def test_semi_and_anti_join_preserve_left_multiplicity(self): + left = pf.from_table( + self.t_env.sql_query( + "SELECT * FROM (VALUES " + "(CAST(1 AS INT), 'A'), (1, 'B'), (2, 'C'), " + "(CAST(NULL AS INT), 'N')) AS T(id, left_value)" + ) + ) + right = pf.from_table( + self.t_env.sql_query( + "SELECT * FROM (VALUES " + "(CAST(1 AS INT), 'X'), (1, 'Y'), (CAST(NULL AS INT), 'Z')) " + "AS T(id, right_value)" + ) + ) + + self.assertCountEqual( + left.join(right, on="id", how="semi").collect(), + [Row(1, "A"), Row(1, "B")], + ) + self.assertCountEqual( + left.join(right, on="id", how="anti").collect(), + [Row(2, "C"), Row(None, "N")], + ) + + def test_join_with_different_names_computed_keys_and_expression(self): + left = pf.from_table( + self.t_env.sql_query( + "SELECT * FROM (VALUES (CAST(1 AS INT), 'L1'), (2, 'L2')) " + "AS T(id, left_value)" + ) + ) + right = pf.from_table( + self.t_env.sql_query( + "SELECT * FROM (VALUES " + "(CAST(2 AS INT), 1, 3, 'R2'), (3, 2, 4, 'R3')) " + "AS T(right_id, min_id, max_id, right_value)" + ) + ) + + self.assertCountEqual( + left.join(right, left_on=pf.col("id") + 1, right_on="right_id").collect(), + [ + Row(1, "L1", 2, 1, 3, "R2"), + Row(2, "L2", 3, 2, 4, "R3"), + ], + ) + predicate = ( + (pf.col("id") + 1 == pf.col("right_id")) + & (pf.col("id") >= pf.col("min_id")) + & (pf.col("id") < pf.col("max_id")) + ) + self.assertCountEqual( + left.join(right, on=predicate).collect(), + [ + Row(1, "L1", 2, 1, 3, "R2"), + Row(2, "L2", 3, 2, 4, "R3"), + ], + ) + self.assertEqual(left.join(right, on=predicate, how="anti").collect(), []) + self.assertCountEqual( + left.join( + right, + left_on=pf.col("id") + 1, + right_on="right_id", + how="semi", + ).collect(), + [Row(1, "L1"), Row(2, "L2")], + ) + + def test_cross_join(self): + left = pf.from_table( + self.t_env.sql_query("SELECT * FROM (VALUES (1), (2)) AS T(id)") + ) + right = pf.from_table( + self.t_env.sql_query("SELECT * FROM (VALUES ('S'), ('M')) AS T(size_name)") + ) + + self.assertCountEqual( + left.join(right, how="cross").collect(), + [Row(1, "S"), Row(1, "M"), Row(2, "S"), Row(2, "M")], + ) + class DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase): @classmethod