Xiao-zhen-Liu commented on code in PR #8768:
URL: https://github.com/apache/texera/pull/8768#discussion_r4200292698
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
return executors[0]
+ # Each pytexera base class and the one method a subclass must implement.
+ _BASE_CLASS_METHODS = {
Review Comment:
Nit: this table repeats the method each base class declares abstract, so it
can go stale without any test failing. A test that compares it with the classes
(e.g. `UDFOperatorV2.__abstractmethods__`) would catch that.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
Review Comment:
This message is wrong in two cases: a class without a base class (e.g.
`class ProcessTupleOperator:`) is told that no class was found, and for the
1-out UDF, following the example fails with "Please use SourceOperator API". It
should name the expected base classes and give a source example for source
operators.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
+ "`from pytexera import *` as the first line."
Review Comment:
Nit: the import is not the first line in our templates; "at the top", as on
line 229, is accurate.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
Review Comment:
When the user has one class that misses its method, this opening sentence
reads as if they defined zero or two classes. I'd keep it only for those two
cases.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
+ "`from pytexera import *` as the first line."
+ )
+
+ explanations = []
+ for abstract_class in abstract_classes:
+ base_names = [
+ base.__name__
+ for base in abstract_class.__mro__
+ if base.__name__ in cls._BASE_CLASS_METHODS
+ ]
+ base_name = base_names[0] if base_names else "Operator"
+ missing = ", ".join(sorted(abstract_class.__abstractmethods__))
+ explanation = (
+ f"{abstract_class.__name__} subclasses {base_name} but does
not "
+ f"implement {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)
+ return (
+ " ".join(explanations)
+ + f" Each base class needs its own method:
{cls._base_class_pairing()}."
+ )
+
+ @classmethod
+ def _with_user_code_hint(cls, error: Exception, file_path: str) ->
Exception:
+ """
+ Rebuild an error raised while importing user code so its message says
+ which line of the UDF failed and, for common mistakes, how to fix it.
+ Returns the original error if it cannot be rebuilt with a new message.
+ """
+ message = str(error)
+ user_lines = [
+ frame.lineno
+ for frame in traceback.extract_tb(error.__traceback__)
+ if frame.filename == file_path
+ ]
+ if user_lines:
+ message += f" (line {user_lines[-1]} of the UDF code)"
Review Comment:
With UI parameters, a method of 3 or more lines is injected after the first
class, so line numbers after that class are off by 3 or more.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
+ "`from pytexera import *` as the first line."
+ )
+
+ explanations = []
+ for abstract_class in abstract_classes:
+ base_names = [
+ base.__name__
+ for base in abstract_class.__mro__
+ if base.__name__ in cls._BASE_CLASS_METHODS
+ ]
+ base_name = base_names[0] if base_names else "Operator"
+ missing = ", ".join(sorted(abstract_class.__abstractmethods__))
+ explanation = (
+ f"{abstract_class.__name__} subclasses {base_name} but does
not "
+ f"implement {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)
+ return (
+ " ".join(explanations)
+ + f" Each base class needs its own method:
{cls._base_class_pairing()}."
+ )
+
+ @classmethod
+ def _with_user_code_hint(cls, error: Exception, file_path: str) ->
Exception:
+ """
+ Rebuild an error raised while importing user code so its message says
+ which line of the UDF failed and, for common mistakes, how to fix it.
+ Returns the original error if it cannot be rebuilt with a new message.
+ """
+ message = str(error)
+ user_lines = [
+ frame.lineno
+ for frame in traceback.extract_tb(error.__traceback__)
+ if frame.filename == file_path
+ ]
+ if user_lines:
+ message += f" (line {user_lines[-1]} of the UDF code)"
+ if isinstance(error, NameError) and error.name in pytexera_names():
+ message += ". Add `from pytexera import *` at the top of the UDF."
+ elif "No super class method found" in message:
+ message += (
+ f". The method must match the base class:
{cls._base_class_pairing()}."
Review Comment:
If `@overrides` is on a helper method, the fix is to remove it, but this
hint says to rename the method. The hint should mention both.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
Review Comment:
Nit: when no hint is added, this raises the error with itself as its cause.
A plain `raise` is enough there.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
+ "`from pytexera import *` as the first line."
+ )
+
+ explanations = []
+ for abstract_class in abstract_classes:
+ base_names = [
+ base.__name__
+ for base in abstract_class.__mro__
+ if base.__name__ in cls._BASE_CLASS_METHODS
+ ]
+ base_name = base_names[0] if base_names else "Operator"
+ missing = ", ".join(sorted(abstract_class.__abstractmethods__))
+ explanation = (
+ f"{abstract_class.__name__} subclasses {base_name} but does
not "
+ f"implement {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)
+ return (
+ " ".join(explanations)
+ + f" Each base class needs its own method:
{cls._base_class_pairing()}."
+ )
+
+ @classmethod
+ def _with_user_code_hint(cls, error: Exception, file_path: str) ->
Exception:
+ """
+ Rebuild an error raised while importing user code so its message says
+ which line of the UDF failed and, for common mistakes, how to fix it.
+ Returns the original error if it cannot be rebuilt with a new message.
+ """
+ message = str(error)
+ user_lines = [
+ frame.lineno
+ for frame in traceback.extract_tb(error.__traceback__)
+ if frame.filename == file_path
+ ]
+ if user_lines:
+ message += f" (line {user_lines[-1]} of the UDF code)"
Review Comment:
The UI shows only the error text, not its type, so a failed `assert x > 0`
shows nothing but "(line 3 of the UDF code)". The text should include the error
type.
##########
amber/src/main/python/core/architecture/managers/executor_manager.py:
##########
@@ -119,15 +127,117 @@ def load_executor_definition(self, code: str) ->
type(Operator):
# 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:
+ # SyntaxError already carries the user's line number.
+ raise
+ except Exception as error:
+ raise self._with_user_code_hint(error, str(file_path)) from error
self.operator_module_name = module_name
executors = list(
filter(self.is_concrete_operator,
executor_module.__dict__.values())
)
- assert len(executors) == 1, "There should be one and only one Operator
defined"
+ assert len(executors) == 1, (
+ "There should be one and only one Operator defined. "
+ + self._explain_operator_count(executors, executor_module)
+ )
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) -> str:
+ """
+ Explain why a UDF module does not define exactly one concrete Operator,
+ naming the classes involved and how to fix them.
+ """
+ if len(executors) > 1:
+ names = ", ".join(executor.__name__ for executor in executors)
+ return (
+ f"Found {len(executors)}: {names}. Keep one and delete or
comment "
+ "out the others. Helper classes must not subclass an Operator."
+ )
+
+ abstract_classes = [
+ value
+ for value in module.__dict__.values()
+ if inspect.isclass(value)
+ and issubclass(value, Operator)
+ and value.__module__ == module.__name__
+ ]
+ if not abstract_classes:
+ return (
+ "No Operator class was found. Uncomment or write exactly one
class, "
+ "e.g. `class ProcessTupleOperator(UDFOperatorV2):`, and keep "
+ "`from pytexera import *` as the first line."
+ )
+
+ explanations = []
+ for abstract_class in abstract_classes:
+ base_names = [
+ base.__name__
+ for base in abstract_class.__mro__
+ if base.__name__ in cls._BASE_CLASS_METHODS
+ ]
+ base_name = base_names[0] if base_names else "Operator"
+ missing = ", ".join(sorted(abstract_class.__abstractmethods__))
+ explanation = (
+ f"{abstract_class.__name__} subclasses {base_name} but does
not "
+ f"implement {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)
+ return (
+ " ".join(explanations)
+ + f" Each base class needs its own method:
{cls._base_class_pairing()}."
+ )
+
+ @classmethod
+ def _with_user_code_hint(cls, error: Exception, file_path: str) ->
Exception:
+ """
+ Rebuild an error raised while importing user code so its message says
+ which line of the UDF failed and, for common mistakes, how to fix it.
+ Returns the original error if it cannot be rebuilt with a new message.
+ """
+ message = str(error)
+ user_lines = [
+ frame.lineno
+ for frame in traceback.extract_tb(error.__traceback__)
+ if frame.filename == file_path
+ ]
+ if user_lines:
+ message += f" (line {user_lines[-1]} of the UDF code)"
+ if isinstance(error, NameError) and error.name in pytexera_names():
+ message += ". Add `from pytexera import *` at the top of the UDF."
+ elif "No super class method found" in message:
+ message += (
+ f". The method must match the base class:
{cls._base_class_pairing()}."
+ )
+ if message == str(error):
+ return error
+ try:
+ return type(error)(message)
Review Comment:
Rebuilding the error with `type(error)(message)` garbles some messages: a
`KeyError` now shows as `"'missing' (line 3 of the UDF code)"`, with extra
quotes.
--
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]
