This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8768-5945cc81858e33db3d5280e9913e089f2035b3a6 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 57b7ae7b2c5c02a04e84a81645825422cd8780d4 Author: Suryaa Charan <[email protected]> AuthorDate: Thu Oct 15 18:48:07 2026 +0000 feat(pyamber): explain Python UDF load errors with cause and fix (#8768) ### What changes were proposed in this PR? When a Python UDF fails to load, the error now says what to fix. The existing "There should be one and only one Operator defined" sentence stays first, so anything matching on it still works. The line number and fix are attached as a note to the original exception, so its type, message and attributes stay as raised. The RPC server now includes exception notes in the `ControlError` message, which is what the Static Error tab shows. | Mistake | Message now adds | |---|---| | No class (e.g. the whole template left commented out) | `No Operator class was found. Uncomment or write exactly one class...` | | Method doesn't match the base class | `WrapperClass subclasses UDFOperatorV2 but does not implement process_tuple. If you meant process_table, subclass UDFTableOperator instead.` | | Two classes | `Found 2: FirstOperator, SecondOperator. Keep one and delete or comment out the others.` | | Missing `from pytexera import *` | `name 'UDFOperatorV2' is not defined (line 2 of the UDF code). Add from pytexera import * at the top of the UDF.` | | `@overrides` on the wrong method | the base class to method pairing | | A class with two pytexera bases | `X uses more than one base class (UDFOperatorV2, UDFTableOperator). Subclass exactly one.` | | Any other load error | the line number in the user's code | Before and after, in the Static Error tab. Before is the untouched template; after is a class with the wrong method: | Before | After | |---|---| | <img width="800" height="288" alt="load-before" src="https://github.com/user-attachments/assets/bd2ac22c-1953-435c-b3d5-0cf2e6f0ddda"; />| <img width="800" height="348" alt="load-after" src="https://github.com/user-attachments/assets/53985c6f-9ad3-4d93-aed1-0974d14e85ad"; /> | Missing import, after: <img width="800" height="303" alt="load-after-import" src="https://github.com/user-attachments/assets/c1ed1322-f7d0-4937-8aa2-848917a115ba"; /> ### Any related issues, documentation, discussions? Part of #8736. Split out of #8737 as the reviewer asked. ### How was this PR tested? New tests in `test_executor_manager.py` cover each case above, plus a syntax error, a `KeyError` and a custom exception at load time (both keep their own message and get the hint as a note), a class alias, and a `LoopEndOperator` subclass. `test_async_rpc_server.py` covers notes in `ControlError`, and loads a broken UDF inside a handler to check the message that gets sent back. ``` cd amber pytest -m "not integration" src/test/python/core/architecture/managers/test_executor_manager.py src/test/python/core/architecture/rpc/test_async_rpc_server.py pytest -m "not integration" ruff check src/main/python src/test/python && ruff format --check src/main/python src/test/python ``` The screenshots are from a local Texera built from this branch. The displayed text for those cases is the same after the follow-up commit. Known limitations: UI-parameter source-line mapping relies on the generated hook's reserved name and structure. Runtime traceback locations (e.g., errors inside process_tuple) are not remapped. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) --------- Co-authored-by: suryaacharan <[email protected]> Co-authored-by: Claude Opus 5.5 <[email protected]> --- .../core/architecture/managers/executor_manager.py | 244 +++++++++++- .../core/architecture/rpc/async_rpc_server.py | 11 +- .../architecture/managers/test_executor_manager.py | 419 +++++++++++++++++++++ .../core/architecture/rpc/test_async_rpc_server.py | 41 ++ 4 files changed, 706 insertions(+), 9 deletions(-) diff --git a/amber/src/main/python/core/architecture/managers/executor_manager.py b/amber/src/main/python/core/architecture/managers/executor_manager.py index 8daa0b9e64..a3dca1adfa 100644 --- a/amber/src/main/python/core/architecture/managers/executor_manager.py +++ b/amber/src/main/python/core/architecture/managers/executor_manager.py @@ -20,6 +20,7 @@ import inspect import itertools import sys import tempfile +import traceback from cached_property import cached_property from loguru import logger from pathlib import Path @@ -28,6 +29,13 @@ from typing import Tuple, Optional from core.models import Operator, SourceOperator +def pytexera_names() -> set: + """Names that `from pytexera import *` brings into a UDF.""" + import pytexera + + return set(pytexera.__all__) + + class ExecutorManager: # Process-wide monotonically increasing counter used to generate the # tmp module names ExecutorManager hands to importlib. Making this a @@ -94,7 +102,9 @@ class ExecutorManager: file_name = f"{module_name}.py" return module_name, file_name - def load_executor_definition(self, code: str) -> type(Operator): + def load_executor_definition(self, code: str, is_source: bool = False) -> type( + Operator + ): """ Load the given executor code in string into a class definition :param code: str, python code that defines an Operator, should contain one @@ -119,15 +129,221 @@ class ExecutorManager: # gen_module_file_name guarantees module_name is unique across # the process, so import_module will always cleanly load source # from the tmp directory we just wrote — no re-import / reload dance. - executor_module = importlib.import_module(module_name) + try: + executor_module = importlib.import_module(module_name) + except SyntaxError as error: + # SyntaxError carries its own line number; only correct it for the + # injected UI-parameter hook, keeping the error itself as raised. + self._correct_syntax_error_line(error, code) + raise + except Exception as error: + self._add_user_code_hint(error, str(file_path), code) + raise self.operator_module_name = module_name + # Deduplicate by identity: an alias such as `Y = X` is the same class. executors = list( - filter(self.is_concrete_operator, executor_module.__dict__.values()) + { + id(value): value + for value in executor_module.__dict__.values() + if self.is_concrete_operator(value) + }.values() + ) + assert len(executors) == 1, self._explain_operator_count( + executors, executor_module, is_source ) - assert len(executors) == 1, "There should be one and only one Operator defined" return executors[0] + # Each pytexera base class and the one method a subclass must implement. + _BASE_CLASS_METHODS = { + "UDFOperatorV2": "process_tuple", + "UDFBatchOperator": "process_batch", + "UDFTableOperator": "process_table", + "UDFSourceOperator": "produce", + } + + @classmethod + def _base_class_pairing(cls) -> str: + return ", ".join( + f"{base} -> {method}" for base, method in cls._BASE_CLASS_METHODS.items() + ) + + @classmethod + def _explain_operator_count(cls, executors: list, module, is_source: bool) -> str: + """ + Explain why a UDF module does not define exactly one concrete Operator, + naming the classes involved and how to fix them. + """ + opening = "There should be one and only one Operator defined. " + example = ( + "`class GenerateOperator(UDFSourceOperator):`" + if is_source + else "`class ProcessTupleOperator(UDFOperatorV2):`" + ) + if len(executors) > 1: + names = ", ".join(executor.__name__ for executor in executors) + return opening + ( + f"Found {len(executors)}: {names}. Keep one and delete or comment " + "out the others." + ) + + user_classes = [ + value + for value in module.__dict__.values() + if inspect.isclass(value) and value.__module__ == module.__name__ + ] + abstract_classes = [c for c in user_classes if issubclass(c, Operator)] + if not abstract_classes: + if user_classes: + names = ", ".join(c.__name__ for c in user_classes) + verb = "does not" if len(user_classes) == 1 else "do not" + bases = ( + "UDFSourceOperator" + if is_source + else "UDFOperatorV2, UDFBatchOperator or UDFTableOperator" + ) + return ( + f"{names} {verb} subclass a pytexera base class. " + f"Subclass {bases}, e.g. {example}" + ) + return opening + ( + f"No Operator class was found. Uncomment or write exactly one class, " + f"e.g. {example}, and keep `from pytexera import *` at the top." + ) + + explanations = [] + mentions_udf_bases = False + for abstract_class in abstract_classes: + name = abstract_class.__name__ + missing = sorted(abstract_class.__abstractmethods__) + udf_bases = [ + base.__name__ + for base in abstract_class.__mro__ + if base.__name__ in cls._BASE_CLASS_METHODS + ] + if len(udf_bases) > 1: + mentions_udf_bases = True + explanations.append( + f"{name} uses more than one base class ({', '.join(udf_bases)}). " + "Subclass exactly one." + ) + elif len(udf_bases) == 1: + mentions_udf_bases = True + base_name = udf_bases[0] + explanation = ( + f"{name} subclasses {base_name} but does not implement " + f"{', '.join(missing)}." + ) + for other_base, method in cls._BASE_CLASS_METHODS.items(): + if other_base != base_name and method in abstract_class.__dict__: + explanation += ( + f" If you meant {method}, subclass {other_base} instead." + ) + explanations.append(explanation) + else: + explanations.append(f"{name} does not implement {', '.join(missing)}.") + pairing = ( + f" Each base class needs its own method: {cls._base_class_pairing()}." + if mentions_udf_bases + else "" + ) + return " ".join(explanations) + pairing + + # The UI-parameter injector (PythonUdfUiParameterInjector) appends this hook + # method to the end of the first UDF class, shifting the lines after it. + # Known limitations: the hook is found by its reserved name and its body is + # assumed to be the lines indented under its header, matching what the + # injector generates today; nothing checks that the two stay in sync. + # Only load errors are corrected; tracebacks of errors raised while the UDF + # runs (e.g. in process_tuple) still use the loaded code's line numbers. + _INJECTED_HOOK_NAME = "_texera_injected_ui_parameters" + + @classmethod + def _correct_syntax_error_line(cls, error: SyntaxError, code: str) -> None: + """ + Map a SyntaxError's start and end lines past the injected UI-parameter + hook. Each is mapped on its own, since a range can start before the hook + and end after it; a line inside the hook is left as reported. + """ + if error.lineno is not None: + line = cls._user_line(code, error.lineno) + if line is not None: + error.lineno = line + if error.end_lineno is not None: + end_line = cls._user_line(code, error.end_lineno) + if end_line is not None: + error.end_lineno = end_line + + @classmethod + def _user_line(cls, code: str, line: int) -> Optional[int]: + """ + Map a line number in the code that was loaded back to the user's code by + skipping the injected UI-parameter hook (its decorator, header and body). + Returns None for a line inside the hook, which the user did not write. + """ + lines = code.splitlines() + header = next( + ( + i + for i, text in enumerate(lines) + if text.lstrip().startswith(f"def {cls._INJECTED_HOOK_NAME}(") + ), + None, + ) + if header is None: + return line + start = header - 1 if lines[header - 1].strip() == "@overrides" else header + indent = len(lines[header]) - len(lines[header].lstrip()) + end = header + 1 + # The hook's body is the run of lines indented deeper than its header. + while ( + end < len(lines) + and lines[end].strip() + and len(lines[end]) - len(lines[end].lstrip()) > indent + ): + end += 1 + injected = end - start + # Lines are 1-based; the hook occupies lines start+1 .. end. + if line > end: + return line - injected + return None if line > start else line + + @classmethod + def _add_user_code_hint( + cls, error: Exception, file_path: str, code: str = "" + ) -> None: + """ + Attach a note to an error raised while importing user code, saying + which line of the UDF failed and, for common mistakes, how to fix it. + The error itself is left unchanged, so its type, message and + attributes stay exactly as raised; the note is shown with it. + """ + location = "" + user_lines = [ + frame.lineno + for frame in traceback.extract_tb(error.__traceback__) + if frame.filename == file_path + ] + if user_lines: + line = cls._user_line(code, user_lines[-1]) + location = ( + f"({type(error).__name__} at line {line} of the UDF code)" + if line is not None + else f"({type(error).__name__} in the generated UI-parameter code)" + ) + fix = "" + if isinstance(error, NameError) and error.name in pytexera_names(): + fix = "Add `from pytexera import *` at the top of the UDF." + elif "No super class method found" in str(error): + fix = ( + f"The method must match the base class: {cls._base_class_pairing()}. " + "If it is a helper method, remove @overrides instead." + ) + if fix: + error.add_note(f"{location}. {fix}" if location else fix) + elif location: + error.add_note(location) + def close(self) -> None: """ Remove the tmp directory and release all resources created within it. @@ -165,6 +381,18 @@ class ExecutorManager: and not inspect.isabstract(cls) ) + @staticmethod + def _source_mismatch_message(is_source: bool) -> str: + if is_source: + return ( + "Please use SourceOperator API for source operators: this is a " + "1-out UDF, so subclass UDFSourceOperator and implement produce()." + ) + return ( + "This UDF reads input, so it cannot use UDFSourceOperator: subclass " + "UDFOperatorV2, UDFBatchOperator or UDFTableOperator instead." + ) + def initialize_executor(self, code: str, is_source: bool, language: str) -> None: """ Initialize the executor with the given code. The output schema is @@ -192,11 +420,11 @@ class ExecutorManager: ) self.executor = executor_class(code) else: - executor: type(Operator) = self.load_executor_definition(code) + executor: type(Operator) = self.load_executor_definition(code, is_source) self.executor = executor() self.executor.is_source = is_source assert isinstance(self.executor, SourceOperator) == self.executor.is_source, ( - "Please use SourceOperator API for source operators." + self._source_mismatch_message(is_source) ) def update_executor(self, code: str, is_source: bool) -> None: @@ -210,11 +438,11 @@ class ExecutorManager: :return: """ original_internal_state = self.executor.__dict__ - executor: type(Operator) = self.load_executor_definition(code) + executor: type(Operator) = self.load_executor_definition(code, is_source) self.executor = executor() self.executor.is_source = is_source assert isinstance(self.executor, SourceOperator) == self.executor.is_source, ( - "Please use SourceOperator API for source operators." + self._source_mismatch_message(is_source) ) # overwrite the internal state self.executor.__dict__ = original_internal_state diff --git a/amber/src/main/python/core/architecture/rpc/async_rpc_server.py b/amber/src/main/python/core/architecture/rpc/async_rpc_server.py index 49dc5f0547..b861a01df1 100644 --- a/amber/src/main/python/core/architecture/rpc/async_rpc_server.py +++ b/amber/src/main/python/core/architecture/rpc/async_rpc_server.py @@ -104,7 +104,16 @@ class AsyncRPCServer: control_return: ControlReturn = set_one_of( ControlReturn, ControlError( - error_message=str(exception), language=ErrorLanguage.PYTHON + # Include notes, which carry hints such as the UDF line. + error_message=" ".join( + part + for part in ( + str(exception), + *getattr(exception, "__notes__", ()), + ) + if part + ), + language=ErrorLanguage.PYTHON, ), ) diff --git a/amber/src/test/python/core/architecture/managers/test_executor_manager.py b/amber/src/test/python/core/architecture/managers/test_executor_manager.py index d7b9d039cd..f1bec1adb4 100644 --- a/amber/src/test/python/core/architecture/managers/test_executor_manager.py +++ b/amber/src/test/python/core/architecture/managers/test_executor_manager.py @@ -505,3 +505,422 @@ class TestUpdateExecutor: assert initialized_manager.executor.counter == 42 assert initialized_manager.executor.added_after_update is True + + +WRONG_METHOD_CODE = """ +from pytexera import * + +class WrapperClass(UDFOperatorV2): + def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + yield table +""" + +MISSING_IMPORT_CODE = """ +class ProcessTupleOperator(UDFOperatorV2): + def process_tuple(self, tuple_, port): + yield tuple_ +""" + +WRONG_METHOD_WITH_OVERRIDES_CODE = """ +from pytexera import * + +class ProcessTupleOperator(UDFOperatorV2): + @overrides + def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + yield table +""" + +SYNTAX_ERROR_CODE = """from pytexera import * + +class ProcessTupleOperator(UDFOperatorV2): + def process_tuple(self, tuple_: Tuple, port: int) + yield tuple_ +""" + +RUNTIME_ERROR_AT_LOAD_CODE = """from pytexera import * + +LOOKUP = {}["missing"] + +class ProcessTupleOperator(UDFOperatorV2): + def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]: + yield tuple_ +""" + + +def shown(error: BaseException) -> str: + """The error text as the RPC server sends it: the message plus any notes.""" + return " ".join( + part for part in (str(error), *getattr(error, "__notes__", ())) if part + ) + + +class TestActionableLoadErrors: + """Load errors must name the cause and the fix for the UDF author.""" + + @pytest.fixture + def executor_manager(self): + manager = ExecutorManager() + yield manager + manager.close() + + def test_no_class_says_to_keep_one_class(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(NO_OPERATOR_CODE) + message = str(exc_info.value) + assert "one and only one Operator" in message + assert "No Operator class was found" in message + + def test_abstract_class_names_missing_method_and_matching_base( + self, executor_manager + ): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(WRONG_METHOD_CODE) + message = str(exc_info.value) + # One class that misses its method: the zero/two-classes sentence would mislead. + assert "one and only one Operator" not in message + assert "WrapperClass subclasses UDFOperatorV2" in message + assert "does not implement process_tuple" in message + assert "If you meant process_table, subclass UDFTableOperator" in message + + def test_two_classes_are_named(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(TWO_OPERATORS_CODE) + message = str(exc_info.value) + assert "one and only one Operator" in message + assert "FirstOperator, SecondOperator" in message + assert "Keep one" in message + + def test_missing_import_suggests_pytexera_import(self, executor_manager): + with pytest.raises(NameError) as exc_info: + executor_manager.load_executor_definition(MISSING_IMPORT_CODE) + message = shown(exc_info.value) + assert "name 'UDFOperatorV2' is not defined" in message + assert "Add `from pytexera import *` at the top of the UDF" in message + assert "line 2" in message + + def test_non_pytexera_name_error_has_no_import_hint(self, executor_manager): + code = "from pytexera import *\n\nundefined_thing\n" + with pytest.raises(NameError) as exc_info: + executor_manager.load_executor_definition(code) + message = shown(exc_info.value) + assert "from pytexera import *" not in message + assert "line 3" in message + + def test_overrides_mismatch_explains_method_pairing(self, executor_manager): + with pytest.raises(TypeError) as exc_info: + executor_manager.load_executor_definition(WRONG_METHOD_WITH_OVERRIDES_CODE) + message = shown(exc_info.value) + assert "No super class method found" in message + assert "UDFTableOperator -> process_table" in message + assert "remove @overrides" in message + + def test_syntax_error_reports_user_line(self, executor_manager): + with pytest.raises(SyntaxError) as exc_info: + executor_manager.load_executor_definition(SYNTAX_ERROR_CODE) + assert exc_info.value.lineno == 4 + assert "line 4" in str(exc_info.value) + + def test_runtime_error_while_loading_reports_user_line(self, executor_manager): + with pytest.raises(KeyError) as exc_info: + executor_manager.load_executor_definition(RUNTIME_ERROR_AT_LOAD_CODE) + # KeyError quotes its message, so the line is attached as a note. + assert "line 3" in "".join(exc_info.value.__notes__) + + def test_error_without_user_line_or_hint_gets_no_note(self): + error = ValueError("boom") + ExecutorManager._add_user_code_hint(error, "/not/the/udf.py") + assert not hasattr(error, "__notes__") + + def test_error_that_cannot_be_rebuilt_is_kept_as_is(self, executor_manager): + # UnicodeDecodeError needs five constructor arguments, so the hint + # can't be added by rebuilding it; the original error must survive. + code = 'from pytexera import *\n\nb"\\xff".decode("utf-8")\n' + with pytest.raises(UnicodeDecodeError) as exc_info: + executor_manager.load_executor_definition(code) + assert exc_info.value.object == b"\xff" + + +class TestLoadErrorHintEdgeCases: + """Hints must not garble or misdescribe the user's original error.""" + + @pytest.fixture + def executor_manager(self): + manager = ExecutorManager() + yield manager + manager.close() + + def test_key_error_message_is_not_wrapped_in_its_quotes(self, executor_manager): + code = 'from pytexera import *\n\nLOOKUP = {}["missing"]\n' + with pytest.raises(KeyError) as exc_info: + executor_manager.load_executor_definition(code) + assert exc_info.value.args == ("missing",) + assert "line 3 of the UDF code" in "".join(exc_info.value.__notes__) + + def test_custom_exception_keeps_its_own_message(self, executor_manager): + code = ( + "class CodeError(Exception):\n" + " def __init__(self, code):\n" + " self.code = code\n" + " def __str__(self):\n" + " return f'code={self.code}'\n" + "\n" + "raise CodeError(5)\n" + ) + with pytest.raises(Exception) as exc_info: + executor_manager.load_executor_definition(code) + assert str(exc_info.value) == "code=5" + assert exc_info.value.code == 5 + + def test_name_error_keeps_its_name(self, executor_manager): + with pytest.raises(NameError) as exc_info: + executor_manager.load_executor_definition(MISSING_IMPORT_CODE) + assert exc_info.value.name == "UDFOperatorV2" + assert "Add `from pytexera import *`" in shown(exc_info.value) + + def test_empty_message_gets_no_leading_space(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition("assert False\n") + assert str(exc_info.value) == "" + assert shown(exc_info.value) == "(AssertionError at line 1 of the UDF code)" + + def test_unchanged_error_is_not_its_own_cause(self, executor_manager): + code = 'from pytexera import *\n\nb"\\xff".decode("utf-8")\n' + with pytest.raises(UnicodeDecodeError) as exc_info: + executor_manager.load_executor_definition(code) + assert exc_info.value.__cause__ is not exc_info.value + + def test_two_pytexera_bases_are_flagged(self, executor_manager): + code = ( + "from pytexera import *\n\n" + "class X(UDFOperatorV2, UDFTableOperator):\n" + " def process_tuple(self, tuple_, port):\n" + " yield tuple_\n" + ) + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(code) + message = str(exc_info.value) + assert "X uses more than one base class" in message + assert "If you meant" not in message + + def test_non_udf_operator_gets_no_udf_pairing_list(self, executor_manager): + code = "from pytexera import *\n\nclass MyLoop(LoopEndOperator):\n pass\n" + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(code) + message = str(exc_info.value) + assert "MyLoop does not implement" in message + assert "Each base class needs its own method" not in message + + def test_class_alias_is_not_counted_twice(self, executor_manager): + code = ( + "from pytexera import *\n\n" + "class X(UDFOperatorV2):\n" + " def process_tuple(self, tuple_, port):\n" + " yield tuple_\n\n" + "Y = X\n" + ) + assert executor_manager.load_executor_definition(code).__name__ == "X" + + def test_two_classes_message_does_not_forbid_helper_bases(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(TWO_OPERATORS_CODE) + message = str(exc_info.value) + assert "Keep one" in message + assert "Helper classes must not subclass" not in message + + def test_no_class_message_says_import_at_the_top(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(NO_OPERATOR_CODE) + assert "at the top" in str(exc_info.value) + assert "as the first line" not in str(exc_info.value) + + +class TestLoadErrorReviewFollowUps: + @pytest.fixture + def executor_manager(self): + manager = ExecutorManager() + yield manager + manager.close() + + def test_base_class_table_matches_the_pytexera_classes(self): + import pytexera + + for base, method in ExecutorManager._BASE_CLASS_METHODS.items(): + assert getattr(pytexera, base).__abstractmethods__ == {method} + + def test_class_without_a_base_is_named_and_told_which_bases_to_use( + self, executor_manager + ): + code = ( + "from pytexera import *\n\n" + "class ProcessTupleOperator:\n" + " def process_tuple(self, tuple_, port):\n" + " yield tuple_\n" + ) + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(code) + message = str(exc_info.value) + assert "ProcessTupleOperator does not subclass a pytexera base class" in message + assert "UDFOperatorV2" in message + assert "one and only one Operator" not in message + + def test_no_class_in_a_source_udf_gives_a_source_example(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.load_executor_definition(NO_OPERATOR_CODE, is_source=True) + message = str(exc_info.value) + assert "class GenerateOperator(UDFSourceOperator):" in message + assert "ProcessTupleOperator(UDFOperatorV2)" not in message + + def test_tuple_class_in_a_source_udf_says_what_to_subclass(self, executor_manager): + with pytest.raises(AssertionError) as exc_info: + executor_manager.initialize_executor( + code=SAMPLE_OPERATOR_CODE, is_source=True, language="python" + ) + message = str(exc_info.value) + assert "SourceOperator API" in message + assert "subclass UDFSourceOperator and implement produce()" in message + + def test_error_type_is_part_of_the_hint(self, executor_manager): + with pytest.raises(NameError) as exc_info: + executor_manager.load_executor_definition(MISSING_IMPORT_CODE) + assert "(NameError at line 2 of the UDF code)" in shown(exc_info.value) + + def test_line_numbers_skip_the_injected_ui_parameter_hook(self, executor_manager): + # The UI-parameter injector appends a hook method to the end of the first + # class; the user's line 7 becomes line 10 in the code that is loaded. + code = ( + "from pytexera import *\n" + "\n" + "class ProcessTupleOperator(UDFOperatorV2):\n" + " def process_tuple(self, tuple_, port):\n" + " yield tuple_\n" + "\n" + " @overrides\n" + " def _texera_injected_ui_parameters(self) -> Dict[str, Any]:\n" + ' return {"count": 1}\n' + 'VALUE = {}["k"]\n' + ) + with pytest.raises(KeyError) as exc_info: + executor_manager.load_executor_definition(code) + assert "(KeyError at line 7 of the UDF code)" in shown(exc_info.value) + + def test_source_class_in_a_udf_with_input_says_what_to_subclass( + self, executor_manager + ): + with pytest.raises(AssertionError) as exc_info: + executor_manager.initialize_executor( + code=SAMPLE_SOURCE_OPERATOR_CODE, is_source=False, language="python" + ) + assert "cannot use UDFSourceOperator" in str(exc_info.value) + + @pytest.mark.parametrize( + "loaded_line, decorated, expected", + [ + (4, True, 4), # before the hook: unchanged + (10, True, 7), # after a decorated hook (3 lines) + (9, False, 7), # after an undecorated hook (2 lines) + (11, True, 8), # a blank user line after the hook is not part of it + (8, True, None), # inside the hook: not a line the user wrote + ], + ) + def test_user_line_mapping(self, loaded_line, decorated, expected): + hook = ([" @overrides"] if decorated else []) + [ + " def _texera_injected_ui_parameters(self) -> Dict[str, Any]:", + " return {}", + ] + user = [ + "from pytexera import *", + "", + "class ProcessTupleOperator(UDFOperatorV2):", + " def process_tuple(self, tuple_, port):", + " yield tuple_", + "", + ] + code = "\n".join(user + hook + ["", 'VALUE = {}["k"]']) + "\n" + assert ExecutorManager._user_line(code, loaded_line) == expected + + def test_user_line_is_unchanged_without_a_hook(self): + assert ExecutorManager._user_line("x = 1\ny = 2\n", 2) == 2 + + +HOOK_LINES = ( + " @overrides\n" + " def _texera_injected_ui_parameters(self) -> Dict[str, Any]:\n" + ' return {"count": 1}\n' +) +CLASS_LINES = ( + "from pytexera import *\n" + "\n" + "class ProcessTupleOperator(UDFOperatorV2):\n" + " def process_tuple(self, tuple_, port):\n" + " yield tuple_\n" + "\n" +) + + +class TestInjectedHookLineNumbers: + @pytest.fixture + def executor_manager(self): + manager = ExecutorManager() + yield manager + manager.close() + + def test_syntax_error_after_the_hook_reports_the_user_line(self, executor_manager): + # User line 7 is `x = (`; after injection it is line 10. + with pytest.raises(SyntaxError) as exc_info: + executor_manager.load_executor_definition( + CLASS_LINES + HOOK_LINES + "x = (\n" + ) + assert exc_info.value.lineno == 7 + + def test_syntax_error_before_the_hook_is_unchanged(self, executor_manager): + code = ( + "from pytexera import *\n" + "\n" + "class ProcessTupleOperator(UDFOperatorV2):\n" + " def process_tuple(self, tuple_, port)\n" + " yield tuple_\n" + "\n" + HOOK_LINES + ) + with pytest.raises(SyntaxError) as exc_info: + executor_manager.load_executor_definition(code) + assert exc_info.value.lineno == 4 + + def test_error_inside_the_hook_is_not_given_a_user_line(self, executor_manager): + # The hook only runs at load time if something calls it; force that so + # the deepest frame in the UDF file is a generated line. + hook = HOOK_LINES.replace('{"count": 1}', '{"count": 1 / 0}') + code = ( + CLASS_LINES + + hook + + "ProcessTupleOperator()._texera_injected_ui_parameters()\n" + ) + with pytest.raises(ZeroDivisionError) as exc_info: + executor_manager.load_executor_definition(code) + text = shown(exc_info.value) + assert "(ZeroDivisionError in the generated UI-parameter code)" in text + assert "at line" not in text + + def test_syntax_error_without_a_line_number_is_left_alone(self): + error = SyntaxError("bad") + ExecutorManager._correct_syntax_error_line(error, CLASS_LINES + HOOK_LINES) + assert error.lineno is None + + def test_syntax_error_without_an_end_line_still_gets_its_line_corrected(self): + error = SyntaxError("bad", ("udf.py", 10, 1, "x = (\n")) + ExecutorManager._correct_syntax_error_line( + error, CLASS_LINES + HOOK_LINES + "x = (\n" + ) + assert (error.lineno, error.end_lineno) == (7, None) + + def test_syntax_error_spanning_the_hook_maps_each_end_separately(self): + # Starts at user line 4 (before the hook), ends at loaded line 10 (after it). + error = SyntaxError("bad", ("udf.py", 4, 1, "x\n", 10, 2)) + ExecutorManager._correct_syntax_error_line( + error, CLASS_LINES + HOOK_LINES + "x = (\n" + ) + assert (error.lineno, error.end_lineno) == (4, 7) + + def test_syntax_error_inside_the_hook_is_left_as_reported(self): + error = SyntaxError("bad", ("udf.py", 8, 1, "x\n", 9, 2)) + ExecutorManager._correct_syntax_error_line(error, CLASS_LINES + HOOK_LINES) + assert (error.lineno, error.end_lineno) == (8, 9) diff --git a/amber/src/test/python/core/architecture/rpc/test_async_rpc_server.py b/amber/src/test/python/core/architecture/rpc/test_async_rpc_server.py index 0613634b3f..d56c33a4d4 100644 --- a/amber/src/test/python/core/architecture/rpc/test_async_rpc_server.py +++ b/amber/src/test/python/core/architecture/rpc/test_async_rpc_server.py @@ -340,6 +340,47 @@ class TestReceiveError: assert error.error_message == "boom" assert error.language == ErrorLanguage.PYTHON + @pytest.mark.timeout(2) + def test_control_error_includes_exception_notes(self): + async def handler_func(stream): + error = ValueError("boom") + error.add_note("(line 2 of the UDF code)") + raise error + + server = _make_server({"explode": _make_handler(handler_func)}) + server.receive(_make_channel("A", "B"), _make_invocation("Explode", 6)) + + error = get_one_of( + _return_invocation_of( + _drain_single_element(server._output_queue) + ).return_value + ) + assert error.error_message == "boom (line 2 of the UDF code)" + + @pytest.mark.timeout(5) + def test_udf_load_error_hint_reaches_the_control_error(self): + # End to end on the Python side: a UDF that fails to load inside a + # control handler must show its hint in the message sent back. + from core.architecture.managers.executor_manager import ExecutorManager + + manager = ExecutorManager() + + async def handler_func(stream): + manager.load_executor_definition('from pytexera import *\n\nx = {}["k"]\n') + + server = _make_server({"init": _make_handler(handler_func)}) + try: + server.receive(_make_channel("A", "B"), _make_invocation("Init", 7)) + finally: + manager.close() + + error = get_one_of( + _return_invocation_of( + _drain_single_element(server._output_queue) + ).return_value + ) + assert error.error_message == "'k' (KeyError at line 3 of the UDF code)" + @pytest.mark.timeout(2) def test_unknown_method_replies_with_control_error_instead_of_raising(self): server = _make_server({})
