fix(compiler): support columns_added and columns_removed in processor config (#394)

[ManoharPaturi]

Description

Fixes #394.

Jinja2 {{ }} references in downstream prompt templates fail when referencing columns created by PRE_BATCH processors because the compiler previously validated templates against the raw seed dataset schema, which did not include columns added or removed at runtime by processors.

This PR adds:

• columns_added and columns_removed fields to ProcessorConfig with default empty lists.
• _apply_processor_column_modifications in compile_data_designer_config which:
  • Removes columns specified in processor.columns_removed from config.columns (raising an InvalidConfigError if a non-existent column is targeted for removal).
  • Appends SeedDatasetColumnConfig entries for columns declared in processor.columns_added to config.columns (validating that a seed dataset is configured and raising InvalidConfigError on collisions).
• Downstream template validation and execution DAG resolution now see the updated column schema.
• Comprehensive unit tests in packages/data-designer-config/tests/config/test_processors.py and packages/data-designer-engine/tests/engine/test_compiler.py.

Testing

• uv run pytest packages/data-designer-config/tests (659 passed)
• uv run pytest packages/data-designer-engine/tests (2,262 passed)
• uv run ruff check and uv run ruff format --check (all checks passed)

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[github-actions]

Linked Issue Check

Issue #394 has not been triaged yet. A maintainer needs to review
the issue and add the triaged label for this check to pass.

You can continue working on the PR in the meantime. The check will
re-run automatically once the issue is triaged.

[greptile-apps]

Confidence Score: 4/5

The PR is not yet safe to merge because processor declarations from post-batch and after-generation stages are still applied to the generation-time schema.

Findings

1.  Processor stages are ignored ▶
2.  Repeated additions bypass collision checks ▶

Fix with agent prompt

### Issue 1
packages/data-designer-engine/src/data_designer/engine/compiler.py:undefined-29
This applies column declarations from every processor before generation, although processor stages are determined by their runtime implementations. If a POST_BATCH or AFTER_GENERATION processor declares an added column, generation-time templates can reference it successfully during validation even though it will not exist until after generation, causing a runtime failure. Likewise, columns declared as removed by a later-stage processor are hidden from validation before that processor actually runs. Restrict these schema changes to PRE_BATCH processors or make the stage part of the configuration contract.

### Issue 2
packages/data-designer-engine/src/data_designer/engine/compiler.py:45-51
`existing_columns` is captured only once before this loop. With `columns_added=["state", "state"]`, both checks pass and two columns named `state` are appended. Static validation does not reject the duplicate, so execution-graph construction later tries to register `state` twice and raises `ValueError`.

```suggestion
            existing_columns = {col.name for col in config.columns}
            for col_name in processor.columns_added:
                if col_name in existing_columns:
                    raise InvalidConfigError(
                        f"🛑 Processor '{processor.name}' adds column '{col_name}' which collides with an existing column."
                    )
                config.columns.append(SeedDatasetColumnConfig(name=col_name))
                existing_columns.add(col_name)
```

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

Summary

Adds processor-declared column additions and removals to the compiled dataset schema so downstream template validation and dependency resolution can account for processor-produced columns.

• Extends ProcessorConfig with serialized columns_added and columns_removed fields.
• Applies declared schema changes after resolving seed columns and before static validation.
• Fixes duplicate additions by updating collision-tracking state after each appended column.
• Adds configuration and compiler coverage for defaults, serialization, additions, removals, collisions, duplicates, and missing seed datasets.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
    A[Resolve seed columns] --> B[Apply processor column declarations]
    B --> C[Add internal row ID if needed]
    C --> D[Validate references]
    D --> E[Build execution plan]

Loading

_{Reviews (2) · Last reviewed commit: "fix(compiler): prevent duplicate columns..."}

[greptile-apps]

Processor stages are ignored

This applies column declarations from every processor before generation, although processor stages are determined by their runtime implementations. If a POST_BATCH or AFTER_GENERATION processor declares an added column, generation-time templates can reference it successfully during validation even though it will not exist until after generation, causing a runtime failure. Likewise, columns declared as removed by a later-stage processor are hidden from validation before that processor actually runs. Restrict these schema changes to PRE_BATCH processors or make the stage part of the configuration contract.

Knowledge Base Used:

• Validation and processing
• Workflow compilation and execution

Prompt To Fix With AI

This is a comment left during a code review.
Path: packages/data-designer-engine/src/data_designer/engine/compiler.py
Line: 29

Comment:
**Processor stages are ignored**

This applies column declarations from every processor before generation, although processor stages are determined by their runtime implementations. If a POST_BATCH or AFTER_GENERATION processor declares an added column, generation-time templates can reference it successfully during validation even though it will not exist until after generation, causing a runtime failure. Likewise, columns declared as removed by a later-stage processor are hidden from validation before that processor actually runs. Restrict these schema changes to PRE_BATCH processors or make the stage part of the configuration contract.

**Knowledge Base Used:**
- [Validation and processing](https://app.greptile.com/nvidia-public-github/-/custom-context/knowledge-base/nvidia-nemo/datadesigner/-/docs/validation-and-processing.md)
- [Workflow compilation and execution](https://app.greptile.com/nvidia-public-github/-/custom-context/knowledge-base/nvidia-nemo/datadesigner/-/docs/workflow-compilation-execution.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

[greptile-apps]

Repeated additions bypass collision checks

existing_columns is captured only once before this loop. With columns_added=["state", "state"], both checks pass and two columns named state are appended. Static validation does not reject the duplicate, so execution-graph construction later tries to register state twice and raises ValueError.

Suggested change

	existing_columns = {col.name for col in config.columns}
	for col_name in processor.columns_added:
	if col_name in existing_columns:
	raise InvalidConfigError(
	f"🛑 Processor '{processor.name}' adds column '{col_name}' which collides with an existing column."
	)
	config.columns.append(SeedDatasetColumnConfig(name=col_name))
	existing_columns = {col.name for col in config.columns}
	for col_name in processor.columns_added:
	if col_name in existing_columns:
	raise InvalidConfigError(
	f"🛑 Processor '{processor.name}' adds column '{col_name}' which collides with an existing column."
	)
	config.columns.append(SeedDatasetColumnConfig(name=col_name))
	existing_columns.add(col_name)

Knowledge Base Used: Workflow compilation and execution

Prompt To Fix With AI

This is a comment left during a code review.
Path: packages/data-designer-engine/src/data_designer/engine/compiler.py
Line: 45-51

Comment:
**Repeated additions bypass collision checks**

`existing_columns` is captured only once before this loop. With `columns_added=["state", "state"]`, both checks pass and two columns named `state` are appended. Static validation does not reject the duplicate, so execution-graph construction later tries to register `state` twice and raises `ValueError`.

```suggestion
            existing_columns = {col.name for col in config.columns}
            for col_name in processor.columns_added:
                if col_name in existing_columns:
                    raise InvalidConfigError(
                        f"🛑 Processor '{processor.name}' adds column '{col_name}' which collides with an existing column."
                    )
                config.columns.append(SeedDatasetColumnConfig(name=col_name))
                existing_columns.add(col_name)
```

**Knowledge Base Used:** [Workflow compilation and execution](https://app.greptile.com/nvidia-public-github/-/custom-context/knowledge-base/nvidia-nemo/datadesigner/-/docs/workflow-compilation-execution.md)

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.