Xiao-zhen-Liu commented on code in PR #8769:
URL: https://github.com/apache/texera/pull/8769#discussion_r4200293285
##########
amber/src/main/python/core/runnables/data_processor.py:
##########
@@ -130,11 +137,26 @@ def _set_output_tuple(self, output_iterator:
Iterator[Optional[TupleLike]]) -> N
"""
for output in output_iterator:
# output could be a None, a TupleLike, or a TableLike.
+ if isinstance(output, dict) and any(
Review Comment:
This breaks a case that works on main: `yield {"result": df}` into a BINARY
column, which is pickled today. The check should run only when the schema check
fails (catching `TypeError` too), which also avoids checking every output row.
##########
amber/src/main/python/core/runnables/data_processor.py:
##########
@@ -94,7 +97,11 @@ def process_tuple(self) -> None:
while not finished_current.is_set():
with self._executor_session() as (executor, port_id):
tuple_ =
self._context.tuple_processing_manager.get_input_tuple()
- self._set_output_tuple(executor.process_tuple(tuple_, port_id))
+ self._set_output_tuple(
+ require_yielded(
Review Comment:
`on_finish` (line 82) has the same problem and is not checked, so `return
{...}` there still gives the pampy error. It needs the same check.
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -29,6 +29,20 @@
import base64
+def require_yielded(result, method_name: str):
+ """
+ Check that a user method such as process_table produced an iterator.
Review Comment:
Nit: this rejects only None, Tuple, dict and DataFrame, so e.g. a returned
Series still passes. The docstring should say so.
##########
amber/src/main/python/core/runnables/data_processor.py:
##########
@@ -130,11 +137,26 @@ def _set_output_tuple(self, output_iterator:
Iterator[Optional[TupleLike]]) -> N
"""
for output in output_iterator:
# output could be a None, a TupleLike, or a TableLike.
+ if isinstance(output, dict) and any(
+ isinstance(value, pandas.DataFrame) for value in
output.values()
+ ):
+ raise TypeError(
+ "A dict of DataFrames was yielded: yield the DataFrame
itself, "
+ "not a dict of DataFrames."
+ )
for output_tuple in all_output_to_tuple(output):
if output_tuple is not None:
- output_tuple.finalize(
- self._context.output_manager.get_port().get_schema()
- )
+ try:
+ output_tuple.finalize(
+
self._context.output_manager.get_port().get_schema()
+ )
+ except KeyError as error:
Review Comment:
A wrong column type raises `TypeError` and gets no hint, although the type
is set in the same panel. It should get the same hint.
##########
amber/src/main/python/core/models/operator.py:
##########
@@ -270,14 +293,21 @@ def __init__(self):
self._Operator__internal_is_source: bool = False
self.__table_data: Mapping[int, List[Tuple]] = defaultdict(list)
+ def _get_table_data(self) -> Mapping[int, List[Tuple]]:
Review Comment:
Nit: `__init__` (line 294) still creates the buffer as well, so it is
created in two places. Creating it only here would be cleaner.
--
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]
