Phase 13.21.ADF v1.1: dematerialize(drop=, keep=) + remove drop_materialized

Adds dematerialize() public API for memory reclamation:
  dematerialize(drop=['dyp_I2','ddxp_I2']) -- drop named aliases
  dematerialize(keep=['dy_I3','dz_I3'])    -- keep named, drop rest
  dematerialize()                          -- drop all materialized
Raw columns always protected. Mutually exclusive drop/keep (ValueError).

Removes drop_materialized() — dematerialize(drop=...) is a strict
superset. 3 internal call sites updated. No external callers existed
(verified: zero test references, zero external imports).

Composes with Phase 13.21 join caching: after dematerialization,
re-materialization reuses cached join indices (index columns unchanged).

Tests J1_6..J1_10: drop+recover, keep, drop-all, raw-column protection,
mutual-exclusion error.

Author: miranov25

UTILS/dfextensions/AliasDataFrame/AliasDataFrame.py

@@ -9878,20 +9878,80 @@ def _get_materialized_aliases(self):
         aliases = self.aliases  # {name: expr}
         return {name for name in aliases if name in self.df.columns}
 
-    def drop_materialized(self, aliases):
+    # drop_materialized removed in Phase 13.21.ADF v1.1
+    # Use dematerialize(drop=...) instead — strict superset.
+
+    def dematerialize(self, drop=None, keep=None):
         """
-        Drop materialized alias columns from DataFrame.
-        
-        Args:
-            aliases: Alias names to drop (silently ignores non-existent)
-        
-        Note:
-            Only drops columns that are aliases, never physical columns.
+        Drop materialized alias columns to reclaim memory.
+
+        Aliases and subframes are preserved — dropped columns can be
+        re-materialized on demand via materialize_aliases().
+
+        Raw columns (those without a matching alias) are never dropped,
+        regardless of the drop/keep parameters.
+
+        Phase 13.21.ADF v1.1: composable with join index caching.
+        After dematerialization, re-materialization reuses cached join
+        indices (index columns unchanged), so the round-trip is cheap.
+
+        Parameters
+        ----------
+        drop : list of str, optional
+            Column names to drop. Only alias-backed columns are dropped;
+            raw columns in this list are silently ignored.
+            Mutually exclusive with keep.
+        keep : list of str, optional
+            Column names to preserve. All OTHER materialized alias columns
+            are dropped. Raw columns are always preserved regardless.
+            Mutually exclusive with drop.
+
+        If neither drop nor keep is given, drops ALL materialized alias
+        columns.
+
+        Returns
+        -------
+        list of str
+            Names of columns actually dropped.
+
+        Examples
+        --------
+        >>> # Drop specific columns you know are no longer needed:
+        >>> adf.dematerialize(drop=['dyp_I2', 'ddxp_I2', 'ddzp_I2'])
+        ['dyp_I2', 'ddxp_I2', 'ddzp_I2']
+
+        >>> # Keep only what the next step needs:
+        >>> adf.dematerialize(keep=['dy_I3', 'dz_I3'])
+        ['dyp_I2', 'ddxp_I2', ..., 'weight_trackI1', ...]
+
+        >>> # Drop everything materialized (back to raw + subframes):
+        >>> adf.dematerialize()
+        ['dy_I0', 'dz_I0', 'dyp_I2', ..., 'isNotEdge', ...]
         """
-        alias_names = set(self.aliases.keys())
-        to_drop = [a for a in aliases if a in alias_names and a in self.df.columns]
+        import gc
+
+        if drop is not None and keep is not None:
+            raise ValueError("Specify drop or keep, not both")
+
+        # Materialized alias columns = columns that DO have a matching alias
+        materialized = [c for c in self.df.columns if c in self.aliases]
+
+        if drop is not None:
+            # Drop only the named columns, only if they are alias-backed
+            to_drop = [c for c in drop if c in materialized]
+        elif keep is not None:
+            # Drop all materialized EXCEPT keep
+            keep_set = set(keep)
+            to_drop = [c for c in materialized if c not in keep_set]
+        else:
+            # Drop all materialized
+            to_drop = materialized
+
         if to_drop:
             self.df = self.df.drop(columns=to_drop)
+            gc.collect()
+
+        return to_drop
 
     def _resolve_draw_param(self, param_value, param_name: str):
         """
@@ -10298,7 +10358,7 @@ def draw(self,
         if cleanup_needed:
             we_added = self._get_materialized_aliases() - already_materialized
             if we_added:
-                self.drop_materialized(we_added)
+                self.dematerialize(drop=list(we_added))
 
         return result
 
@@ -11249,7 +11309,7 @@ def draw_batch(self,
             if we_added:
                 if verbose:
                     print(f"Clearing {len(we_added)} materialized aliases")
-                self.drop_materialized(we_added)
+                self.dematerialize(drop=list(we_added))
 
         return results
 
@@ -11537,7 +11597,7 @@ def draw_figures(
             if we_added:
                 if verbose:
                     print(f"[draw_figures] Clearing {len(we_added)} materialized aliases")
-                self.drop_materialized(we_added)
+                self.dematerialize(drop=list(we_added))
 
         return results
 

UTILS/dfextensions/AliasDataFrame/tests/test_J1_join_cache.py

@@ -270,6 +270,125 @@ def test_J1_5_multi_subframe_pipeline(self):
             err_msg="J1_5: multi-subframe pipeline result mismatch"
         )
 
+    @pytest.mark.invariance
+    def test_J1_6_dematerialize_drop_and_recover(self):
+        """
+        J1_6: dematerialize(drop=[...]) drops the column;
+        re-materialize recovers identical values.
+        """
+        adf = _build_main()
+        sf = _build_coeff_sf()
+        adf.register_subframe('Coeff', sf, index_columns=['sec'])
+        adf.add_alias('correction', 'Coeff.c0 + x', dtype=np.float32)
+        adf.add_alias('scaled', 'Coeff.c1 * y', dtype=np.float32)
+
+        adf.materialize_aliases(names=['correction', 'scaled'])
+        original_correction = adf.df['correction'].values.copy()
+        original_scaled = adf.df['scaled'].values.copy()
+
+        # Drop only 'correction', keep 'scaled'
+        dropped = adf.dematerialize(drop=['correction'])
+        assert dropped == ['correction'], f"Expected ['correction'], got {dropped}"
+        assert 'correction' not in adf.df.columns, "correction should be dropped"
+        assert 'scaled' in adf.df.columns, "scaled should survive"
+
+        # Raw columns must survive
+        assert 'x' in adf.df.columns, "raw column 'x' must survive"
+        assert 'sec' in adf.df.columns, "raw column 'sec' must survive"
+
+        # Re-materialize and verify bit-exact recovery
+        adf.materialize_aliases(names=['correction'])
+        np.testing.assert_array_equal(
+            adf.df['correction'].values, original_correction,
+            err_msg="J1_6: re-materialized values differ from original"
+        )
+
+    @pytest.mark.invariance
+    def test_J1_7_dematerialize_keep(self):
+        """
+        J1_7: dematerialize(keep=[...]) drops all OTHER materialized aliases;
+        raw columns survive; re-materialization recovers values.
+        """
+        adf = _build_main()
+        sf = _build_coeff_sf()
+        adf.register_subframe('Coeff', sf, index_columns=['sec'])
+        adf.add_alias('a1', 'Coeff.c0', dtype=np.float32)
+        adf.add_alias('a2', 'Coeff.c1', dtype=np.float32)
+        adf.add_alias('a3', 'x**2', dtype=np.float32)
+
+        adf.materialize_aliases(names=['a1', 'a2', 'a3'])
+        original_a2 = adf.df['a2'].values.copy()
+        original_a3 = adf.df['a3'].values.copy()
+
+        # Keep only a1; drop a2 and a3
+        dropped = adf.dematerialize(keep=['a1'])
+        assert 'a1' in adf.df.columns, "a1 should survive (in keep list)"
+        assert 'a2' not in adf.df.columns, "a2 should be dropped"
+        assert 'a3' not in adf.df.columns, "a3 should be dropped"
+        assert set(dropped) == {'a2', 'a3'}, f"Expected {{'a2','a3'}}, got {set(dropped)}"
+
+        # Raw columns always survive
+        assert 'x' in adf.df.columns
+        assert 'sec' in adf.df.columns
+
+        # Re-materialize dropped aliases
+        adf.materialize_aliases(names=['a2', 'a3'])
+        np.testing.assert_array_equal(
+            adf.df['a2'].values, original_a2,
+            err_msg="J1_7: re-materialized a2 differs"
+        )
+        np.testing.assert_array_equal(
+            adf.df['a3'].values, original_a3,
+            err_msg="J1_7: re-materialized a3 differs"
+        )
+
+    @pytest.mark.invariance
+    def test_J1_8_dematerialize_all(self):
+        """
+        J1_8: dematerialize() with no args drops ALL materialized alias columns.
+        Raw columns survive.
+        """
+        adf = _build_main()
+        adf.add_alias('x2', 'x**2', dtype=np.float32)
+        adf.add_alias('y2', 'y**2', dtype=np.float32)
+        adf.materialize_aliases(names=['x2', 'y2'])
+
+        raw_cols_before = [c for c in adf.df.columns if c not in adf.aliases]
+
+        dropped = adf.dematerialize()
+        assert 'x2' not in adf.df.columns
+        assert 'y2' not in adf.df.columns
+
+        # All raw columns still present
+        for col in raw_cols_before:
+            assert col in adf.df.columns, f"raw column '{col}' was dropped"
+
+    @pytest.mark.invariance
+    def test_J1_9_dematerialize_ignores_raw_columns(self):
+        """
+        J1_9: dematerialize(drop=['raw_col']) silently ignores raw columns.
+        No crash, no drop.
+        """
+        adf = _build_main()
+        adf.add_alias('a1', 'x**2', dtype=np.float32)
+        adf.materialize_aliases(names=['a1'])
+
+        # Try to drop a raw column — should be silently ignored
+        dropped = adf.dematerialize(drop=['x', 'sec', 'a1'])
+        assert 'a1' not in adf.df.columns, "alias a1 should be dropped"
+        assert 'x' in adf.df.columns, "raw column x must survive"
+        assert 'sec' in adf.df.columns, "raw column sec must survive"
+        assert dropped == ['a1'], f"Only alias columns in drop list: {dropped}"
+
+    @pytest.mark.invariance
+    def test_J1_10_dematerialize_drop_keep_mutual_exclusion(self):
+        """
+        J1_10: specifying both drop and keep raises ValueError.
+        """
+        adf = _build_main()
+        with pytest.raises(ValueError, match="drop or keep, not both"):
+            adf.dematerialize(drop=['x'], keep=['y'])
+
 
 class TestJ2JoinCachePerformance:
     """J2 — cache hit count verification."""