feat(amber): Enable R UDF Runtime via Optional texera-rudf Plugin

amber/src/main/python/core/architecture/managers/executor_manager.py

@@ -132,13 +132,29 @@ class declaration.
         :param language: The language of the operator code.
         :return:
         """
-        assert language not in [
-            "r-tuple",
-            "r-table",
-        ], "R language is not supported by default. Please consult third party plugin."
-        executor: type(Operator) = self.load_executor_definition(code)
-        self.executor = executor()
-        self.executor.is_source = is_source
+        if language in ("r-tuple", "r-table"):
+            # R support is provided by an optional plugin (texera-rudf)
+            executor_type = "Tuple" if language == "r-tuple" else "Table"
+            try:
+                import texera_r
+
+                source_executor_class = getattr(
+                    texera_r, f"R{executor_type}SourceExecutor"
+                )
+                executor_class = getattr(texera_r, f"R{executor_type}Executor")
+            except ImportError as e:
+                raise RuntimeError(
+                    "R operators require the texera-rudf package.\n"
+                    "Install with: pip install git+https://github.com/Texera/texera-rudf.git\n"
+                    f"Import error: {e}"
+                )
+            self.executor = (
+                source_executor_class(code) if is_source else executor_class(code)
+            )
+        else:
+            executor: type(Operator) = self.load_executor_definition(code)
+            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."
         )

amber/src/main/python/core/architecture/managers/test_executor_manager.py

@@ -39,7 +39,7 @@ def produce(self) -> Iterator[Union[TupleLike, TableLike, None]]:
 
 
 class TestExecutorManager:
-    """Test suite for ExecutorManager, focusing on R UDF support removal."""
+    """Test suite for ExecutorManager, focusing on R UDF plugin support."""
 
     @pytest.fixture
     def executor_manager(self):
@@ -57,26 +57,26 @@ def test_initialization(self, executor_manager):
         assert executor_manager.executor_version == 0
 
     def test_reject_r_tuple_language(self, executor_manager):
-        """Test that 'r-tuple' language is rejected with AssertionError."""
-        with pytest.raises(AssertionError) as exc_info:
+        """Test that 'r-tuple' language is rejected with RuntimeError when plugin is not available."""
+        with pytest.raises(RuntimeError) as exc_info:
             executor_manager.initialize_executor(
                 code=SAMPLE_OPERATOR_CODE, is_source=False, language="r-tuple"
             )
 
-        # Verify the error message mentions R UDF support has been dropped
-        assert "not supported" in str(exc_info.value) or "dropped" in str(
+        # Verify the error message mentions R operators require the texera-rudf package
+        assert "texera-rudf" in str(exc_info.value) or "R operators require" in str(
             exc_info.value
         )
 
     def test_reject_r_table_language(self, executor_manager):
-        """Test that 'r-table' language is rejected with AssertionError."""
-        with pytest.raises(AssertionError) as exc_info:
+        """Test that 'r-table' language is rejected with RuntimeError when plugin is not available."""
+        with pytest.raises(RuntimeError) as exc_info:
             executor_manager.initialize_executor(
                 code=SAMPLE_OPERATOR_CODE, is_source=False, language="r-table"
             )
 
-        # Verify the error message mentions R UDF support has been dropped
-        assert "not supported" in str(exc_info.value) or "dropped" in str(
+        # Verify the error message mentions R operators require the texera-rudf package
+        assert "texera-rudf" in str(exc_info.value) or "R operators require" in str(
             exc_info.value
         )