feat(RDF): Modular API with collision handling

Phase 4 complete:
- setup_rdf_with_friends() - returns (rdf, file_handle)
- setup_chain_with_friends() - TChain for multiple files
- add_defines_to_rdf(on_collision='warn') - handles existing columns
- get_join_columns_for_snapshot() - index columns helper
- cache_to_snapshot() - convenience function
- export_tree(columns=...) - snapshot mode

20 RDF tests passing.

Co-authored-by: Claude (Architect)
Reviewed-by: GPT, Gemini, Claude

Author: miranov25

UTILS/dfextensions/AliasDataFrame/AliasDataFrameRDF.py

@@ -912,7 +912,7 @@ def setup_chain_with_friends(adf, file_patterns, treename="tree"):
     return rdf, chain, file_handles
 
 
-def add_defines_to_rdf(rdf, adf, target_aliases):
+def add_defines_to_rdf(rdf, adf, target_aliases, on_collision='warn'):
     """
     Add Define() chain to RDataFrame for requested aliases.
     
@@ -927,6 +927,12 @@ def add_defines_to_rdf(rdf, adf, target_aliases):
         AliasDataFrame with alias definitions
     target_aliases : list of str
         Alias names to define (dependencies are auto-resolved)
+    on_collision : str
+        How to handle when alias name matches existing column:
+        - 'error': Raise ValueError
+        - 'skip': Skip silently
+        - 'warn': Skip with UserWarning (default)
+        - 'redefine': Use Redefine() to overwrite
         
     Returns
     -------
@@ -946,16 +952,50 @@ def add_defines_to_rdf(rdf, adf, target_aliases):
     - Resolves alias dependencies using get_ordered_defines()
     - Converts Python expressions to C++ using to_cpp_expr()
     - Applies Define() calls in correct dependency order
+    - Handles column name collisions based on on_collision parameter
     """
+    import warnings
+    
+    # Get existing columns in RDataFrame
+    existing_columns = set(str(c) for c in rdf.GetColumnNames())
+    
     # Get ordered defines with C++ expressions
     defines = get_ordered_defines(target_aliases, aDF=adf)
+    skipped = []
 
     # Apply Define() chain
     for d in defines:
         name = d['name']
         cpp_expr = d['cpp_expr']
+        
+        if name in existing_columns:
+            if on_collision == 'error':
+                raise ValueError(
+                    f"add_defines_to_rdf: column '{name}' already exists. "
+                    f"Use on_collision='skip', 'warn', or 'redefine'."
+                )
+            elif on_collision == 'skip':
+                skipped.append(name)
+                continue
+            elif on_collision == 'warn':
+                skipped.append(name)
+                continue
+            elif on_collision == 'redefine':
+                rdf = rdf.Redefine(name, cpp_expr)
+                continue
+            else:
+                raise ValueError(f"Unknown on_collision: {on_collision!r}")
+        
         rdf = rdf.Define(name, cpp_expr)
 
+    # Single warning for all skipped columns
+    if skipped and on_collision == 'warn':
+        warnings.warn(
+            f"add_defines_to_rdf: {len(skipped)} column(s) already exist, "
+            f"using existing branch data: {skipped[:5]}{'...' if len(skipped) > 5 else ''}",
+            UserWarning
+        )
+    
     return rdf
 
 
@@ -1038,7 +1078,7 @@ def get_join_columns_for_snapshot(adf, target_aliases=None):
 
 def cache_to_snapshot(adf, input_file, output_file, target_aliases, 
                       treename="tree", output_treename="cache",
-                      include_join_columns=True):
+                      include_join_columns=True, on_collision='warn'):
     """
     Convenience function: compute aliases and save to ROOT file via RDataFrame.
     
@@ -1061,6 +1101,9 @@ def cache_to_snapshot(adf, input_file, output_file, target_aliases,
         Output tree name (default: "cache")
     include_join_columns : bool
         If True, include index columns for rejoining (default: True)
+    on_collision : str
+        How to handle column collisions (default: 'warn')
+        See add_defines_to_rdf() for options.
         
     Returns
     -------
@@ -1091,7 +1134,7 @@ def cache_to_snapshot(adf, input_file, output_file, target_aliases,
     rdf, file_handle = setup_rdf_with_friends(adf, input_file, treename)
 
     # Add defines
-    rdf = add_defines_to_rdf(rdf, adf, target_aliases)
+    rdf = add_defines_to_rdf(rdf, adf, target_aliases, on_collision=on_collision)
 
     # Determine columns to save
     columns_to_save = list(target_aliases)

UTILS/dfextensions/AliasDataFrame/tests/test_AliasDataFrameRDF.py

@@ -745,6 +745,71 @@ def test_cache_creates_file(self, tmp_path):
         assert result['time_s'] > 0
 
 
+@pytest.mark.skipif(not HAS_ROOT, reason="ROOT not available")
+class TestAddDefinesCollision:
+    """Test add_defines_to_rdf collision handling."""
+    
+    @pytest.fixture
+    def adf_with_collision(self, tmp_path):
+        """Create ADF with physical column that matches an alias name."""
+        df = pd.DataFrame({
+            'x': np.array([1, 2, 3], dtype=np.float32),
+            'y': np.array([4, 5, 6], dtype=np.float32),
+        })
+        adf = AliasDataFrame(df)
+        
+        # Export FIRST (so 'x' is a physical branch in tree)
+        filepath = str(tmp_path / "collision.root")
+        adf.export_tree(filepath, 'tree')
+        
+        # THEN add alias 'x' (collides with physical branch 'x')
+        adf.add_alias('x', 'y * 2')
+        
+        return adf, filepath
+    
+    def test_collision_error(self, adf_with_collision):
+        """Test on_collision='error' raises ValueError."""
+        from AliasDataFrameRDF import setup_rdf_with_friends, add_defines_to_rdf
+        adf, filepath = adf_with_collision
+        
+        rdf, f = setup_rdf_with_friends(adf, filepath)
+        with pytest.raises(ValueError, match="already exists"):
+            add_defines_to_rdf(rdf, adf, ['x'], on_collision='error')
+    
+    def test_collision_skip(self, adf_with_collision):
+        """Test on_collision='skip' skips silently."""
+        from AliasDataFrameRDF import setup_rdf_with_friends, add_defines_to_rdf
+        adf, filepath = adf_with_collision
+        
+        rdf, f = setup_rdf_with_friends(adf, filepath)
+        rdf = add_defines_to_rdf(rdf, adf, ['x'], on_collision='skip')
+        
+        # Should use original branch value (1,2,3), not alias (8,10,12)
+        mean = rdf.Mean('x').GetValue()
+        assert abs(mean - 2.0) < 0.1  # Original data mean
+    
+    def test_collision_warn(self, adf_with_collision):
+        """Test on_collision='warn' skips with warning."""
+        from AliasDataFrameRDF import setup_rdf_with_friends, add_defines_to_rdf
+        adf, filepath = adf_with_collision
+        
+        rdf, f = setup_rdf_with_friends(adf, filepath)
+        with pytest.warns(UserWarning, match="already exist"):
+            rdf = add_defines_to_rdf(rdf, adf, ['x'], on_collision='warn')
+    
+    def test_collision_redefine(self, adf_with_collision):
+        """Test on_collision='redefine' overwrites with alias."""
+        from AliasDataFrameRDF import setup_rdf_with_friends, add_defines_to_rdf
+        adf, filepath = adf_with_collision
+        
+        rdf, f = setup_rdf_with_friends(adf, filepath)
+        rdf = add_defines_to_rdf(rdf, adf, ['x'], on_collision='redefine')
+        
+        # Should use alias value (y*2 = 8,10,12), not original (1,2,3)
+        mean = rdf.Mean('x').GetValue()
+        assert abs(mean - 10.0) < 0.1  # Alias computation mean
+
+
 # =============================================================================
 # Summary Report
 # =============================================================================