feat: add processor plugin support (#299)

* feat: add processor plugin support

Add PluginType.PROCESSOR to the plugin system, enabling third-party
processor plugins via entry points. Includes a demo plugin package
with RegexFilterProcessor (process_before_batch) and
SemanticDedupProcessor (process_after_generation).

- Add PluginType.PROCESSOR with processor_type discriminator
- Create processor_types.py for ProcessorConfigT with plugin injection
- Register plugin processors in engine ProcessorRegistry
- Use RLock in PluginRegistry to prevent deadlocks during discovery
- Add demo package: data-designer-demo-processors
- Update processor and plugin documentation

* test: add processor plugin registration test

Verify that processor plugins from PluginRegistry are picked up
by create_default_processor_registry and registered correctly.

* test: simplify processor plugin registration test

* move ProcessorConfig to base and convert demo to e2e test

- Move ProcessorConfig from processors.py to config.base to guard
  against circular deps (alongside SingleColumnConfig)
- Delete demo/ directory with regex_filter and semantic_dedup plugins
- Add regex_filter as an e2e processor plugin test in tests_e2e/

* move plan to plans/299/

Author: andreatgretel

docs/concepts/processors.md

@@ -138,6 +138,36 @@ builder.add_processor(
 
 Processors execute in the order they're added. Plan accordingly when one processor's output affects another.
 
+## Processor Plugins
+
+You can extend Data Designer with custom processors via the [plugin system](../plugins/overview.md). A processor plugin is a Python package that provides:
+
+- A **config class** inheriting from `ProcessorConfig` with a `processor_type: Literal["your-type"]` discriminator
+- An **implementation class** inheriting from `Processor` that overrides the desired callback methods
+- A **`Plugin` instance** connecting the two
+
+Once installed, plugin processors are automatically discovered and can be used with `add_processor()` like built-in processors.
+
+```python
+from my_processor_plugin.config import MyProcessorConfig
+
+builder.add_processor(
+    MyProcessorConfig(
+        name="my_processor",
+        # ... plugin-specific parameters ...
+    )
+)
+```
+
+**Entry point configuration** in `pyproject.toml`:
+
+```toml
+[project.entry-points."data_designer.plugins"]
+my-processor = "my_plugin.plugin:my_processor_plugin"
+```
+
+See the [plugins overview](../plugins/overview.md) for the full guide on creating plugins.
+
 ## Configuration Parameters
 
 ### Common Parameters

docs/plugins/overview.md

@@ -7,12 +7,11 @@
 
 Plugins are Python packages that extend Data Designer's capabilities without modifying the core library. Similar to [VS Code extensions](https://marketplace.visualstudio.com/vscode) and [Pytest plugins](https://docs.pytest.org/en/stable/reference/plugin_list.html), the plugin system empowers you to build specialized extensions for your specific use cases and share them with the community.
 
-**Current capabilities**: Data Designer supports two plugin types:
+**Current capabilities**: Data Designer supports three plugin types:
 
 - **Column Generator Plugins**: Custom column types you pass to the config builder's [add_column](../code_reference/config_builder.md#data_designer.config.config_builder.DataDesignerConfigBuilder.add_column) method.
 - **Seed Reader Plugins**: Custom seed dataset readers that let you load data from new sources (e.g., databases, cloud storage, custom formats).
-
-**Coming soon**: Plugin support for processors, validators, and more!
+- **Processor Plugins**: Custom processors that transform data before batches, after batches, or after generation completes. Pass them to the config builder's [add_processor](../code_reference/config_builder.md#data_designer.config.config_builder.DataDesignerConfigBuilder.add_processor) method.
 
 ## How do you use plugins?
 
@@ -39,9 +38,11 @@ Each plugin has three components, and we recommend organizing them into separate
 - **`config.py`** -- Configuration class defining user-facing parameters
     - Column generator plugins: inherit from `SingleColumnConfig` with a `column_type` discriminator
     - Seed reader plugins: inherit from `SeedSource` with a `seed_type` discriminator
+    - Processor plugins: inherit from `ProcessorConfig` with a `processor_type` discriminator
 - **`impl.py`** -- Implementation class containing the core logic
     - Column generator plugins: inherit from `ColumnGeneratorFullColumn` or `ColumnGeneratorCellByCell`
     - Seed reader plugins: inherit from `SeedReader`
+    - Processor plugins: inherit from `Processor` and override callback methods (`process_before_batch`, `process_after_batch`, `process_after_generation`)
 - **`plugin.py`** -- A `Plugin` instance that connects the config and implementation classes
 
 ### 2. Package Your Plugin
@@ -61,4 +62,23 @@ Each plugin has three components, and we recommend organizing them into separate
 - Publish to PyPI or another package index to make it installable by anyone via `pip install`
 - This step is only needed if you want others outside your environment to use the plugin
 
+**Example entry point for a processor plugin:**
+
+```toml
+[project.entry-points."data_designer.plugins"]
+my-processor = "my_plugin.plugin:my_processor_plugin"
+```
+
+Where `my_processor_plugin` is a `Plugin` instance with `plugin_type=PluginType.PROCESSOR`:
+
+```python
+from data_designer.plugins.plugin import Plugin, PluginType
+
+my_processor_plugin = Plugin(
+    config_qualified_name="my_plugin.config.MyProcessorConfig",
+    impl_qualified_name="my_plugin.impl.MyProcessor",
+    plugin_type=PluginType.PROCESSOR,
+)
+```
+
 **Ready to get started?** See the [Example Plugin](example.md) for a complete walkthrough of creating a column generator plugin.

packages/data-designer-config/src/data_designer/config/base.py

@@ -8,7 +8,7 @@
 
 from abc import ABC, abstractmethod
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class ConfigBase(BaseModel):
@@ -66,3 +66,20 @@ def side_effect_columns(self) -> list[str]:
             List of column names that this column will create as a side effect. Empty list
             indicates no side effect columns. Override in subclasses to specify side effects.
         """
+
+
+class ProcessorConfig(ConfigBase, ABC):
+    """Abstract base class for all processor configuration types.
+
+    Processors are transformations that run at different stages of the generation
+    pipeline. They can modify, reshape, or augment the dataset.
+
+    Attributes:
+        name: Unique name of the processor, used to identify the processor in results
+            and to name output artifacts on disk.
+    """
+
+    name: str = Field(
+        description="The name of the processor, used to identify the processor in the results and to write the artifacts to disk.",
+    )
+    processor_type: str

packages/data-designer-config/src/data_designer/config/config_builder.py

@@ -28,7 +28,8 @@
 from data_designer.config.exportable_config import ExportableConfigBase
 from data_designer.config.mcp import ToolConfig
 from data_designer.config.models import ModelConfig, load_model_configs
-from data_designer.config.processors import ProcessorConfigT, ProcessorType, get_processor_config_from_kwargs
+from data_designer.config.processor_types import ProcessorConfigT
+from data_designer.config.processors import ProcessorType, get_processor_config_from_kwargs
 from data_designer.config.sampler_constraints import (
     ColumnConstraintT,
     ColumnInequalityConstraint,

packages/data-designer-config/src/data_designer/config/data_designer_config.py

@@ -12,7 +12,7 @@
 from data_designer.config.exportable_config import ExportableConfigBase
 from data_designer.config.mcp import ToolConfig
 from data_designer.config.models import ModelConfig
-from data_designer.config.processors import ProcessorConfigT
+from data_designer.config.processor_types import ProcessorConfigT
 from data_designer.config.sampler_constraints import ColumnConstraintT
 from data_designer.config.seed import SeedConfig
 

packages/data-designer-config/src/data_designer/config/processor_types.py

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from typing_extensions import TypeAlias
+
+from data_designer.config.processors import DropColumnsProcessorConfig, SchemaTransformProcessorConfig
+from data_designer.plugin_manager import PluginManager
+
+plugin_manager = PluginManager()
+
+ProcessorConfigT: TypeAlias = DropColumnsProcessorConfig | SchemaTransformProcessorConfig
+ProcessorConfigT = plugin_manager.inject_into_processor_config_type_union(ProcessorConfigT)

packages/data-designer-config/src/data_designer/config/processors.py

@@ -4,14 +4,12 @@
 from __future__ import annotations
 
 import json
-from abc import ABC
 from enum import Enum
 from typing import Any, Literal
 
 from pydantic import Field, field_validator
-from typing_extensions import TypeAlias
 
-from data_designer.config.base import ConfigBase
+from data_designer.config.base import ProcessorConfig
 from data_designer.config.errors import InvalidConfigError
 
 
@@ -27,26 +25,6 @@ class ProcessorType(str, Enum):
     SCHEMA_TRANSFORM = "schema_transform"
 
 
-class ProcessorConfig(ConfigBase, ABC):
-    """Abstract base class for all processor configuration types.
-
-    Processors are transformations that run at different stages of the generation
-    pipeline. They can modify, reshape, or augment the dataset.
-
-    The processor implementation determines which stages it handles by overriding
-    the appropriate callback methods (process_before_batch, process_after_batch, process_after_generation).
-
-    Attributes:
-        name: Unique name of the processor, used to identify the processor in results
-            and to name output artifacts on disk.
-    """
-
-    name: str = Field(
-        description="The name of the processor, used to identify the processor in the results and to write the artifacts to disk.",
-    )
-    processor_type: str
-
-
 def get_processor_config_from_kwargs(processor_type: ProcessorType, **kwargs: Any) -> ProcessorConfig:
     """Create a processor configuration from a processor type and keyword arguments.
 
@@ -129,6 +107,3 @@ def validate_template(cls, v: dict[str, Any]) -> dict[str, Any]:
             if "not JSON serializable" in str(e):
                 raise InvalidConfigError("Template must be JSON serializable")
         return v
-
-
-ProcessorConfigT: TypeAlias = DropColumnsProcessorConfig | SchemaTransformProcessorConfig

packages/data-designer-config/src/data_designer/plugin_manager.py

@@ -76,3 +76,14 @@ def inject_into_seed_source_type_union(self, seed_source_type: type[TypeAlias])
         """
         seed_source_type = self._plugin_registry.add_plugin_types_to_union(seed_source_type, PluginType.SEED_READER)
         return seed_source_type
+
+    def inject_into_processor_config_type_union(self, processor_config_type: type[TypeAlias]) -> type[TypeAlias]:
+        """Inject plugins into the processor config type.
+
+        Args:
+            processor_config_type: The processor config type to inject plugins into.
+
+        Returns:
+            The processor config type with plugins injected.
+        """
+        return self._plugin_registry.add_plugin_types_to_union(processor_config_type, PluginType.PROCESSOR)

packages/data-designer-config/src/data_designer/plugins/plugin.py

@@ -20,13 +20,16 @@
 class PluginType(str, Enum):
     COLUMN_GENERATOR = "column-generator"
     SEED_READER = "seed-reader"
+    PROCESSOR = "processor"
 
     @property
     def discriminator_field(self) -> str:
         if self == PluginType.COLUMN_GENERATOR:
             return "column_type"
         elif self == PluginType.SEED_READER:
             return "seed_type"
+        elif self == PluginType.PROCESSOR:
+            return "processor_type"
         else:
             raise ValueError(f"Invalid plugin type: {self.value}")
 

packages/data-designer-config/src/data_designer/plugins/registry.py

@@ -23,7 +23,7 @@
 class PluginRegistry:
     _instance = None
     _plugins_discovered = False
-    _lock = threading.Lock()
+    _lock = threading.RLock()
 
     _plugins: dict[str, Plugin] = {}