sshiv012 commented on code in PR #8769:
URL: https://github.com/apache/texera/pull/8769#discussion_r4202050687
##########
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:
Good catch, and you're right. On main that row just gets pickled into the
BINARY column, and I'd turned it into an error. I moved the dict check so it
only runs after `finalize` actually fails, catching `TypeError` as well as
`KeyError`. Valid rows don't get checked at all now. I added a test for the
BINARY case so this doesn't come back.
##########
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:
Yep, missed that one. I only wrapped the call sites I thought of, not every
place that calls into the user's code. `on_finish` goes through the same check
now, and there's a test for it.
##########
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:
Makes sense, since it's the same panel. Type errors get the same hint now,
and I tweaked it to say the columns *and their types* have to match.
##########
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:
Fair point. I updated the docstring to say it only catches a returned None,
Tuple, dict or DataFrame, and that anything else, like a Series, passes through
unchecked.
##########
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:
Done. The buffers are only created in the getters now. I kept the
`BATCH_SIZE` check in `BatchOperator.__init__` so a bad size still fails right
when the operator is created.
--
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]
