This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new bd703c9c9c4 [FLINK-40856][python] Support strict zip-style
multi-column DataFrame explode
bd703c9c9c4 is described below
commit bd703c9c9c4b3a6f6b4e088d9e876030448204f2
Author: beetle0915 <[email protected]>
AuthorDate: Sun Oct 11 09:47:39 2026 +0800
[FLINK-40856][python] Support strict zip-style multi-column DataFrame
explode
This closes #29457.
---
.../docs/reference/pyflink.dataframe/dataframe.rst | 8 ++
flink-python/pyflink/dataframe/dataframe.py | 140 ++++++++++++++++--
.../pyflink/dataframe/tests/test_dataframe.py | 157 ++++++++++++++++++++-
.../utils/python/ExplodeOrdinalityFunction.java | 47 ++++++
.../python/ExplodeOrdinalityFunctionTest.java | 104 ++++++++++++++
5 files changed, 447 insertions(+), 9 deletions(-)
diff --git a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
index ec7a33281e4..4a443034676 100644
--- a/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
+++ b/flink-python/docs/reference/pyflink.dataframe/dataframe.rst
@@ -39,6 +39,14 @@ Example::
... .filter(pf.col("id") > 0)
>>> names = df.select(df.name)
+Pass a list to :meth:`DataFrame.explode` to expand multiple ARRAY columns
together by position.
+The arrays must have equal lengths in each row; empty and null arrays count as
length zero.
+Unequal lengths fail during execution. If all arrays are empty or null, the
default retains
+one row with null expanded values; ``ignore_empty_and_null=True`` drops that
row::
+
+ >>> df = pf.from_dict({"items": [["apple", "pear"]], "counts": [[2, 3]]})
+ >>> result = df.explode(["items", "counts"])
+
DataFrame
---------
diff --git a/flink-python/pyflink/dataframe/dataframe.py
b/flink-python/pyflink/dataframe/dataframe.py
index 66152fb4ce9..e81bfee1d95 100644
--- a/flink-python/pyflink/dataframe/dataframe.py
+++ b/flink-python/pyflink/dataframe/dataframe.py
@@ -758,7 +758,7 @@ class DataFrame:
@PublicEvolving()
def explode(
self,
- column: Union[str, Expression],
+ column: Union[str, Expression, List[Union[str, Expression]]],
*,
output_column: Optional[Union[str, List[str]]] = None,
ignore_empty_and_null: bool = False,
@@ -773,15 +773,24 @@ class DataFrame:
fields. Empty and null collections produce a row with null output
fields unless
``ignore_empty_and_null`` is true.
+ A list of multiple columns or expressions expands ARRAYs together by
position. Their lengths
+ must match in each input row, treating empty and null arrays as length
zero; otherwise
+ execution fails. Null elements count toward the length. When all
arrays are empty or
+ null, ``ignore_empty_and_null`` controls whether to drop the row or
retain one row with
+ null expanded fields. Each array produces one output column, including
ARRAYs of ROWs.
+ A one-element list follows the single-input behavior.
+
.. warning::
- Due to FLINK-40658, actual null ROW elements in an ARRAY are not
handled correctly
- and may be silently dropped or cause execution to fail. This
limitation is independent
- of ``ignore_empty_and_null``. A non-null ROW whose fields are all
null is not affected.
+ Single-input expansion of an ARRAY containing null ROW elements
may drop those
+ elements or fail during execution, regardless of
``ignore_empty_and_null``.
+ A non-null ROW whose fields are all null is not affected.
- :param column: Collection column name or row-wise expression to expand.
+ :param column: Collection column name or row-wise expression to
expand, or a non-empty
+ list of ARRAY column names or row-wise expressions to expand
together.
:param output_column: Output name or list of names. Required for
multiple fields;
a single field defaults to the selected column name. Names must be
unique and must
- not conflict with retained input columns.
+ not conflict with retained input columns. For multiple ARRAYs,
names default to
+ the resolved input names and each ARRAY requires one output name.
:param ignore_empty_and_null: Whether to drop rows with empty or null
collections.
:return: A new DataFrame with the expanded rows.
:raises TypeError: If an argument has an unsupported type or the input
is not a collection.
@@ -811,14 +820,28 @@ class DataFrame:
>>> flattened.columns
['id', 'items$name', 'items$quantity']
+ Expand multiple ARRAYs together, matching their elements by position::
+
+ >>> df = pf.from_dict({"items": [["apple", "pear"]], "counts":
[[2, 3]]})
+ >>> result = df.explode(["items", "counts"])
+
.. versionadded:: 2.4.0
"""
- if not isinstance(column, (str, Expression)):
- raise TypeError("column must be a column name or expression")
+ if isinstance(column, list):
+ if not column:
+ raise ValueError("column must not be empty")
+ if not all(isinstance(value, (str, Expression)) for value in
column):
+ raise TypeError("column must contain only column names or
expressions")
+ if len(column) == 1:
+ column = column[0]
+ elif not isinstance(column, (str, Expression)):
+ raise TypeError("column must be a column name, expression, or list
of these")
if not isinstance(ignore_empty_and_null, bool):
raise TypeError("ignore_empty_and_null must be a boolean")
if output_column is not None and not isinstance(output_column, (str,
list)):
raise TypeError("output_column must be a string or list of
strings")
+ if isinstance(column, list):
+ return self._explode_columns(column, output_column,
ignore_empty_and_null)
expression = table_col(column) if isinstance(column, str) else column
selected = self._table.select(expression)
@@ -929,6 +952,107 @@ class DataFrame:
)
return DataFrame(table._t_env.sql_query(query))
+ def _explode_columns(
+ self,
+ columns: List[Union[str, Expression]],
+ output_column: Optional[Union[str, List[str]]],
+ ignore_empty_and_null: bool,
+ ) -> "DataFrame":
+ table = self._table
+ input_names = table.get_resolved_schema().get_column_names()
+ expressions: List[Expression] = []
+ for value in columns:
+ if isinstance(value, str):
+ expressions.append(table_col(value))
+ else:
+ expressions.append(value)
+ resolved_names = []
+ input_indexes = []
+ for expression in expressions:
+ selected = table.select(expression)
+ schema = selected.get_resolved_schema()
+ if len(schema.get_column_names()) != 1:
+ raise ValueError("each column must select a single column")
+ projection = selected._j_table.getQueryOperation()
+ if not
projection.getChildren().get(0).equals(table._j_table.getQueryOperation()):
+ raise ValueError("column must be a row-wise expression, not an
aggregation")
+ if not isinstance(schema.get_column_data_types()[0], ArrayType):
+ raise TypeError("multiple columns must all have an ARRAY type")
+ resolved_names.append(schema.get_column_names()[0])
+ resolved = projection.getProjectList().get(0)
+ index: Optional[int] = None
+ if resolved.getClass().getSimpleName() ==
"FieldReferenceExpression":
+ index = resolved.getFieldIndex()
+ if index is not None and index in input_indexes:
+ raise ValueError("column must not contain duplicate input
columns")
+ input_indexes.append(index)
+
+ if output_column is None:
+ output_names = resolved_names
+ elif isinstance(output_column, str):
+ output_names = [output_column]
+ else:
+ output_names = output_column
+ if not all(isinstance(name, str) for name in output_names):
+ raise TypeError("output_column must contain only strings")
+ if len(output_names) != len(columns):
+ raise ValueError("output_column must contain %d name(s)" %
len(columns))
+ if any(not name for name in output_names) or len(set(output_names)) !=
len(columns):
+ raise ValueError("output_column names must be non-empty and
unique")
+ retained = []
+ for index, name in enumerate(input_names):
+ if index not in input_indexes:
+ retained.append(name)
+ if set(output_names).intersection(retained):
+ raise ValueError("output_column names conflict with retained input
columns")
+
+ taken = set(input_names) | set(output_names)
+ array_names = []
+ computed = []
+ for expression, index in zip(expressions, input_indexes):
+ if index is None:
+ name = _unique_name("__pf_explode", taken)
+ taken.add(name)
+ computed.append(expression.alias(name))
+ else:
+ name = input_names[index]
+ array_names.append(name)
+ if computed:
+ table = table.add_columns(*computed)
+ ordinal_name = _unique_name("__pf_explode_ordinality", taken)
+ gateway = get_gateway()
+ # Validate before producing ordinals, including when the first array
is empty.
+ function =
gateway.jvm.org.apache.flink.table.utils.python.ExplodeOrdinalityFunction()
+ lengths = to_jarray(
+ gateway.jvm.Object,
+ [_get_java_expression(table_col(name).cardinality) for name in
array_names],
+ )
+ ordinals: Expression = Expression(
+ gateway.jvm.Expressions.call(function, lengths)
+ ).alias(ordinal_name)
+ if ignore_empty_and_null:
+ table = table.join_lateral(ordinals)
+ else:
+ table = table.left_outer_join_lateral(ordinals)
+ expanded = [
+ table_col(name).at(table_col(ordinal_name)).alias(output)
+ for name, output in zip(array_names, output_names)
+ ]
+ replacements = {}
+ for index, value in zip(input_indexes, expanded):
+ if index is not None:
+ replacements[index] = value
+ projections: List[Expression] = []
+ for index, name in enumerate(input_names):
+ if index in replacements:
+ projections.append(replacements[index])
+ else:
+ projections.append(table_col(name))
+ for index, value in zip(input_indexes, expanded):
+ if index is None:
+ projections.append(value)
+ return DataFrame(table.select(*projections))
+
# ======================== Joins ========================
@PublicEvolving()
diff --git a/flink-python/pyflink/dataframe/tests/test_dataframe.py
b/flink-python/pyflink/dataframe/tests/test_dataframe.py
index 5305802a406..ce906b405aa 100644
--- a/flink-python/pyflink/dataframe/tests/test_dataframe.py
+++ b/flink-python/pyflink/dataframe/tests/test_dataframe.py
@@ -416,7 +416,7 @@ class DataFrameExplodeTests(PyFlinkDataFrameUTTestCase):
)
def test_explode_rejects_invalid_arguments(self):
- for column in [None, 1, ["items"]]:
+ for column in [None, 1, ["items", 1]]:
with self.subTest(column=column):
with self.assertRaisesRegex(TypeError, "column"):
self.df.explode(column)
@@ -464,6 +464,80 @@ class DataFrameExplodeTests(PyFlinkDataFrameUTTestCase):
df.explode("items", output_column=["number", "text"])
+class DataFrameMultiColumnExplodeTests(PyFlinkDataFrameUTTestCase):
+ def setUp(self):
+ super().setUp()
+ self.df = pf.from_table(self.t_env.sql_query(
+ "SELECT 1 AS id, ARRAY[1, 2] AS items, 'x' AS label, ARRAY['a',
'b'] AS tags"))
+
+ def test_explode_is_lazy_and_replaces_columns_in_place(self):
+ table = self.df.to_table()
+ with patch.object(Table, "execute") as execute:
+ result = self.df.explode(
+ [pf.col("tags").alias("renamed"), "items"],
+ output_column=["tag", "item"])
+ execute.assert_not_called()
+ self.assertIs(self.df.to_table(), table)
+ self.assertIs(result.to_table()._t_env, self.t_env)
+ self.assert_dataframe_schema(result, ["id", "item", "label", "tag"])
+
+ def test_explode_defaults_and_computed_expression(self):
+ from pyflink.table.expressions import array
+
+ self.assert_dataframe_schema(
+ self.df.explode(["items", "tags"]), ["id", "items", "label",
"tags"])
+ self.assert_dataframe_schema(
+ self.df.explode([array(pf.col("id"), pf.lit(3)).alias("pair"),
"tags"]),
+ ["id", "items", "label", "tags", "pair"])
+ self.assert_dataframe_schema(self.df.explode(["items"]),
self.df.columns)
+ mapping = pf.from_table(self.t_env.sql_query("SELECT MAP['a', 1] AS
items"))
+ self.assert_dataframe_schema(
+ mapping.explode(["items"], output_column=["key", "value"]),
["key", "value"])
+
+ def test_explode_quotes_names_and_preserves_row_type(self):
+ df = pf.from_table(self.t_env.sql_query(
+ "SELECT ARRAY[CAST(NULL AS ROW<n INT, s STRING>)] AS `a``b`, "
+ "ARRAY[1] AS __pf_explode_ordinality"))
+ result = df.explode(
+ ["a`b", "__pf_explode_ordinality"], output_column=["select", "new`
name"])
+ self.assert_dataframe_schema(result, ["select", "new` name"], [
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("n", TableDataTypes.INT()),
+ TableDataTypes.FIELD("s", TableDataTypes.STRING()),
+ ]), TableDataTypes.INT(),
+ ])
+
+ def test_explode_rejects_invalid_inputs(self):
+ for columns, error, message in [
+ ([], ValueError, "empty"),
+ (["items", "id"], TypeError, "ARRAY"),
+ (["items", "items"], ValueError, "duplicate"),
+ (["items", pf.col("items").alias("other")], ValueError,
"duplicate"),
+ (["items", pf.col("*")], ValueError, "single column"),
+ (["items", pf.col("id").collect], ValueError, "row-wise"),
+ ]:
+ with self.subTest(columns=str(columns)):
+ with self.assertRaisesRegex(error, message):
+ self.df.explode(columns)
+ for collection in ["MAP['a', 1]", "MULTISET[1, 2]"]:
+ df = pf.from_table(self.t_env.sql_query(
+ "SELECT ARRAY[1] AS a, %s AS b" % collection))
+ with self.assertRaisesRegex(TypeError, "ARRAY"):
+ df.explode(["a", "b"])
+
+ def test_explode_rejects_invalid_names_and_flags(self):
+ for output, error in [
+ ([], ValueError), (["item"], ValueError), ("item", ValueError),
+ (["item", "item"], ValueError), (["item", ""], ValueError),
+ (["item", "id"], ValueError), (["item", 1], TypeError),
+ ]:
+ with self.subTest(output=output):
+ with self.assertRaisesRegex(error, "output_column"):
+ self.df.explode(["items", "tags"], output_column=output)
+ with self.assertRaisesRegex(TypeError, "ignore_empty_and_null"):
+ self.df.explode(["items", "tags"], ignore_empty_and_null=1)
+
+
class DataFrameSortingTests(PyFlinkDataFrameUTTestCase):
def setUp(self):
super().setUp()
@@ -3800,6 +3874,66 @@ class DataFrameExplodeITTests(PyFlinkITTestCase):
self.assertEqual(result.columns, ["__pf_explode", "value"])
self.assertCountEqual(result.collect(), [Row(1, 1), Row(1, 2)])
+ def test_explode_multiple_arrays(self):
+ from pyflink.table.expressions import call_sql
+
+ df = pf.from_table(self.t_env.from_elements(
+ [
+ (1, [Row(2), None, Row(None)], "x", [4, 5, None]),
+ (2, [], "y", []),
+ (3, None, "z", []),
+ (4, [], "n", None),
+ (5, None, "m", None),
+ (6, [Row(2), Row(2)], "d", [4, 4]),
+ ],
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("id", TableDataTypes.INT()),
+ TableDataTypes.FIELD("items",
TableDataTypes.ARRAY(TableDataTypes.ROW([
+ TableDataTypes.FIELD("n", TableDataTypes.INT()),
+ ]))),
+ TableDataTypes.FIELD("label", TableDataTypes.STRING()),
+ TableDataTypes.FIELD("tags",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ ])))
+ results = []
+ expected_rows = []
+ for ignore in [False, True]:
+ with self.subTest(ignore=ignore):
+ result = df.explode(
+ ["tags", pf.col("items").alias("item"),
+ call_sql("ARRAY_REVERSE(tags)").alias("reversed")],
+ ignore_empty_and_null=ignore)
+ self.assertEqual(result.columns, ["id", "item", "label",
"tags", "reversed"])
+ expected = [
+ Row(1, False, 2, "x", 4, None),
+ Row(1, True, None, "x", 5, 5),
+ Row(1, False, None, "x", None, 4),
+ Row(6, False, 2, "d", 4, 4),
+ Row(6, False, 2, "d", 4, 4),
+ ]
+ if not ignore:
+ expected += [Row(i, True, None, label, None, None)
+ for i, label in [(2, "y"), (3, "z"), (4,
"n"), (5, "m")]]
+ results.append(result.select(
+ pf.lit(ignore).alias("ignore"), "id",
pf.col("item").is_null,
+ pf.col("item").get("n"), "label", "tags", "reversed"))
+ expected_rows.extend(Row(ignore, *row) for row in expected)
+ self.assertCountEqual(results[0].union_all(results[1]).collect(),
expected_rows)
+
+ def test_explode_multiple_arrays_rejects_mismatches_at_execution(self):
+ for ignore, first, second in [
+ (False, [1], [2, 3]),
+ (True, [], [2]),
+ ]:
+ with self.subTest(ignore=ignore):
+ df = pf.from_table(self.t_env.from_elements(
+ [(first, second)], TableDataTypes.ROW([
+ TableDataTypes.FIELD("a",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ TableDataTypes.FIELD("b",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ ])))
+ result = df.explode(["a", "b"], ignore_empty_and_null=ignore)
+ with self.assertRaisesRegex(Exception, "same length"):
+ result.collect()
+
class DataFrameExplodeStreamITTests(PyFlinkStreamDataFrameTestCase):
def test_explode_in_streaming_mode(self):
@@ -3811,6 +3945,27 @@ class
DataFrameExplodeStreamITTests(PyFlinkStreamDataFrameTestCase):
])))
self.assertCountEqual(df.explode("items").collect(), [Row(1, 2),
Row(1, 3)])
+ def test_explode_multiple_arrays_in_streaming_mode(self):
+ df = pf.from_table(self.t_env.from_elements(
+ [(1, [2, None, 2], [4, 5, 4]), (2, [], None)],
+ TableDataTypes.ROW([
+ TableDataTypes.FIELD("id", TableDataTypes.INT()),
+ TableDataTypes.FIELD("a",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ TableDataTypes.FIELD("b",
TableDataTypes.ARRAY(TableDataTypes.INT())),
+ ])))
+ results = []
+ expected_rows = []
+ for ignore in [False, True]:
+ with self.subTest(ignore=ignore):
+ expected = [Row(1, 2, 4), Row(1, None, 5), Row(1, 2, 4)]
+ if not ignore:
+ expected.append(Row(2, None, None))
+ results.append(df.explode(
+ ["a", "b"], ignore_empty_and_null=ignore).select(
+ pf.lit(ignore).alias("ignore"), "id", "a", "b"))
+ expected_rows.extend(Row(ignore, *row) for row in expected)
+ self.assertCountEqual(results[0].union_all(results[1]).collect(),
expected_rows)
+
class DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase):
@classmethod
diff --git
a/flink-python/src/main/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunction.java
b/flink-python/src/main/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunction.java
new file mode 100644
index 00000000000..28573a3993f
--- /dev/null
+++
b/flink-python/src/main/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunction.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.utils.python;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.functions.TableFunction;
+
+import java.util.Arrays;
+
+/** Emits one-based ordinals for equally sized arrays, treating null lengths
as zero. */
+@Internal
+public class ExplodeOrdinalityFunction extends TableFunction<Integer> {
+
+ private static final long serialVersionUID = 1L;
+
+ public void eval(Integer... lengths) {
+ if (lengths.length == 0) {
+ return;
+ }
+ final int length = lengths[0] == null ? 0 : lengths[0];
+ for (final Integer otherLength : lengths) {
+ if ((otherLength == null ? 0 : otherLength) != length) {
+ throw new IllegalArgumentException(
+ "Arrays to explode must have the same length: " +
Arrays.toString(lengths));
+ }
+ }
+ for (int i = 0; i < length; i++) {
+ collect(i + 1);
+ }
+ }
+}
diff --git
a/flink-python/src/test/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunctionTest.java
b/flink-python/src/test/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunctionTest.java
new file mode 100644
index 00000000000..3d8d57129ca
--- /dev/null
+++
b/flink-python/src/test/java/org/apache/flink/table/utils/python/ExplodeOrdinalityFunctionTest.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.utils.python;
+
+import org.apache.flink.util.Collector;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ExplodeOrdinalityFunctionTest {
+
+ private final List<Integer> output = new ArrayList<>();
+ private final ExplodeOrdinalityFunction function = new
ExplodeOrdinalityFunction();
+
+ ExplodeOrdinalityFunctionTest() {
+ function.setCollector(
+ new Collector<Integer>() {
+ @Override
+ public void collect(Integer record) {
+ output.add(record);
+ }
+
+ @Override
+ public void close() {}
+ });
+ }
+
+ @Test
+ void testEqualLengths() {
+ function.eval(1);
+ assertThat(output).containsExactly(1);
+
+ output.clear();
+ function.eval(3, 3);
+ assertThat(output).containsExactly(1, 2, 3);
+
+ output.clear();
+ function.eval(4, 4, 4, 4);
+ assertThat(output).containsExactly(1, 2, 3, 4);
+ }
+
+ @Test
+ void testEmptyLengths() {
+ function.eval();
+ function.eval(0, 0, 0);
+ function.eval(null, null, null);
+ function.eval(null, 0, null, 0);
+ function.eval(0, null, 0, null);
+
+ assertThat(output).isEmpty();
+ }
+
+ @Test
+ void testMismatchedLengthsEmitNothing() {
+ final Integer[][] mismatches = {
+ {0, 1},
+ {null, 1},
+ {1, 0},
+ {1, null},
+ {2, 3},
+ {2, 2, 3},
+ {0, null, 1},
+ {3, 3, 3, 0}
+ };
+ for (final Integer[] lengths : mismatches) {
+ assertThatThrownBy(() -> function.eval(lengths))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("same length")
+ .hasMessageContaining(Arrays.toString(lengths));
+ assertThat(output).isEmpty();
+ }
+ }
+
+ @Test
+ void testRepeatedCalls() {
+ function.eval(2, 2, 2);
+ function.eval(0, null);
+ function.eval(1, 1);
+
+ assertThat(output).containsExactly(1, 2, 1);
+ }
+}
