feat: Add flexible missing column handling to compress_columns

Add on_missing and return_summary parameters to compress_columns method
for more flexible handling of missing columns during compression workflows.

Changes:
- Add on_missing parameter with three modes:
  * 'warn' (default): Skip missing columns with warning
  * 'error': Raise KeyError if any column missing (strict mode)
  * 'ignore': Skip missing columns silently

- Add return_summary parameter to return compression results:
  * Returns dict with 'compressed' and 'skipped' column lists
  * Maintains backward compatibility (default returns self)

- Implement smart column availability checking:
  * Checks DataFrame columns, aliases, and compression_info
  * Handles compressed columns that become aliases
  * Preserves state validation for already-tracked columns
  * Filtering now applies uniformly across all modes (including selective)

- Update test_selective_mode_validates_column_exists:
  * Now uses on_missing='error' for strict validation
  * Changed expected exception from ValueError to KeyError
  * Reflects new filtering behavior

- Add comprehensive test suite (TestCompressionOnMissing):
  * 9 new tests covering all on_missing modes
  * Tests for return_summary functionality
  * Tests for method chaining and backward compatibility
  * Tests for edge cases (all missing, partial missing)

This enables progressive compression workflows where some columns may not
exist yet, while maintaining strict validation when needed for production
use cases.

Breaking change: Default behavior now warns and skips missing columns
instead of raising an error. Use on_missing='error' for strict validation.

All tests passing (70/70).

Author: miranov25

UTILS/dfextensions/AliasDataFrame/AliasDataFrame.py

@@ -835,49 +835,46 @@ def compress_columns(self, compression_spec=None, columns=None, suffix='_c', dro
             )
 
         # === NEW: Filter columns based on on_missing parameter ===
-        # Special handling for selective mode to preserve original validation behavior
-        if schema_mode != 'selective':
-            import warnings
-
-            # Check which columns exist in DataFrame, aliases, OR are already tracked
-            existing_in_df = set(self.df.columns)
-            existing_in_aliases = set(self.aliases.keys())
-            tracked_in_schema = set(self.compression_info.keys()) - {'__meta__'}
-
-            # A column is "available" if:
-            # 1. It exists physically in the DataFrame, OR
-            # 2. It exists as an alias (compressed columns become aliases), OR
-            # 3. It's already tracked in compression_info (for state validation)
-            available_cols = [col for col in cols_to_process
-                              if col in existing_in_df or
-                              col in existing_in_aliases or
-                              col in tracked_in_schema]
-
-            # A column is "missing" only if it's nowhere: not in df, not an alias, not tracked
-            missing_cols = [col for col in cols_to_process
-                            if col not in existing_in_df and
-                            col not in existing_in_aliases and
-                            col not in tracked_in_schema]
-
-            # Handle missing columns according to on_missing mode
-            if missing_cols:
-                if on_missing == 'error':
-                    raise KeyError(
-                        f"Missing columns: {missing_cols}\n"
-                        f"Available in DataFrame: {list(existing_in_df)[:20]}...\n"
-                        f"Available as aliases: {list(existing_in_aliases)[:20]}..."
-                    )
-                elif on_missing == 'warn':
-                    warnings.warn(
-                        f"Skipping missing columns: {missing_cols}\n"
-                        f"Hint: Use columns= to restrict, or on_missing='error' for strict mode."
-                    )
-                # else: on_missing == 'ignore', do nothing
+        import warnings
+
+        # Check which columns exist in DataFrame, aliases, OR are already tracked
+        existing_in_df = set(self.df.columns)
+        existing_in_aliases = set(self.aliases.keys())
+        tracked_in_schema = set(self.compression_info.keys()) - {'__meta__'}
+
+        # A column is "available" if:
+        # 1. It exists physically in the DataFrame, OR
+        # 2. It exists as an alias (compressed columns become aliases), OR
+        # 3. It's already tracked in compression_info (for state validation)
+        available_cols = [col for col in cols_to_process
+                          if col in existing_in_df or
+                          col in existing_in_aliases or
+                          col in tracked_in_schema]
+
+        # A column is "missing" only if it's nowhere: not in df, not an alias, not tracked
+        missing_cols = [col for col in cols_to_process
+                        if col not in existing_in_df and
+                        col not in existing_in_aliases and
+                        col not in tracked_in_schema]
+
+        # Handle missing columns according to on_missing mode
+        if missing_cols:
+            if on_missing == 'error':
+                raise KeyError(
+                    f"Missing columns: {missing_cols}\n"
+                    f"Available in DataFrame: {list(existing_in_df)[:20]}...\n"
+                    f"Available as aliases: {list(existing_in_aliases)[:20]}..."
+                )
+            elif on_missing == 'warn':
+                warnings.warn(
+                    f"Skipping missing columns: {missing_cols}\n"
+                    f"Hint: Use columns= to restrict, or on_missing='error' for strict mode."
+                )
+            # else: on_missing == 'ignore', do nothing
 
-            # Update cols_to_process to only include available columns
-            cols_to_process = available_cols
-            # else: In selective mode, don't filter - let existing validation handle it
-            # === END NEW CODE ===
+        # Update cols_to_process to only include available columns
+        cols_to_process = available_cols
+        # === END NEW CODE ===
 
         for orig_col in cols_to_process:
             # Get config (from spec or existing schema)
@@ -901,16 +898,7 @@ def compress_columns(self, compression_spec=None, columns=None, suffix='_c', dro
                     )
                 compressed_col = f"{orig_col}{suffix}"
 
-            # For selective mode, validate column exists and handle schema updates
-            if schema_mode == 'selective':
-                # Validate column exists in DataFrame or aliases
-                if orig_col not in self.df.columns and orig_col not in self.aliases:
-                    available = list(self.df.columns)[:10]
-                    raise ValueError(
-                        f"Column '{orig_col}' not found in DataFrame or aliases. "
-                        f"Cannot compress non-existent column.\n"
-                        f"Available columns (first 10): {available}..."
-                    )
+            # For selective mode, nothing to be done
 
             # Check current state and validate transitions
             current_state = self.get_compression_state(orig_col)

…ons/AliasDataFrame/AliasDataFrameTest.py …s/AliasDataFrame/test_alias_dataframe.pyUTILS/dfextensions/AliasDataFrame/AliasDataFrameTest.py renamed to UTILS/dfextensions/AliasDataFrame/test_alias_dataframe.py UTILS/dfextensions/AliasDataFrame/AliasDataFrameTest.py renamed to UTILS/dfextensions/AliasDataFrame/test_alias_dataframe.py

@@ -1091,7 +1091,7 @@ def test_selective_mode_errors_on_schema_change_when_compressed(self):
         self.assertIn('decompress first', str(cm.exception).lower())
 
     def test_selective_mode_validates_column_exists(self):
-        """Test that selective mode validates column exists in DataFrame"""
+        """Test that selective mode with on_missing='error' validates column exists"""
         spec = {
             'nonexistent': {
                 'compress': 'round(nonexistent*10)',
@@ -1101,12 +1101,11 @@ def test_selective_mode_validates_column_exists(self):
             }
         }
 
-        with self.assertRaises(ValueError) as cm:
-            self.adf.compress_columns(spec, columns=['nonexistent'])
-
-        self.assertIn('not found in DataFrame', str(cm.exception))
-        self.assertIn('nonexistent', str(cm.exception))
+        # Use on_missing='error' for strict validation
+        with self.assertRaises(KeyError) as cm:
+            self.adf.compress_columns(spec, columns=['nonexistent'], on_missing='error')
 
+        self.assertIn("Missing columns", str(cm.exception))
     def test_selective_mode_validates_columns_in_spec(self):
         """Test that selective mode validates requested columns are in spec"""
         with self.assertRaises(ValueError) as cm:
@@ -1212,5 +1211,177 @@ def test_pattern1_pattern2_mixing(self):
         self.assertEqual(self.adf.compression_info['dz']['compress_expr'], self.spec['dz']['compress'])
 
 
+class TestCompressionOnMissing(unittest.TestCase):
+    """Test on_missing and return_summary parameters"""
+
+    def setUp(self):
+        """Create test DataFrame and compression spec"""
+        self.df = pd.DataFrame({
+            'dy': np.random.randn(10),
+            'dz': np.random.randn(10),
+            'y': np.random.randn(10) * 50,
+        })
+
+        # Spec with some columns that won't exist in df
+        self.spec = {
+            'dy': {
+                'compress': 'round(asinh(dy)*40)',
+                'decompress': 'sinh(dy_c/40.)',
+                'compressed_dtype': np.int16,
+                'decompressed_dtype': np.float16
+            },
+            'dz': {
+                'compress': 'round(asinh(dz)*40)',
+                'decompress': 'sinh(dz_c/40.)',
+                'compressed_dtype': np.int16,
+                'decompressed_dtype': np.float16
+            },
+            'y': {
+                'compress': 'round(y*(0x7fff/50.))',
+                'decompress': 'y_c*(50.0/(0x7fff))',
+                'compressed_dtype': np.int16,
+                'decompressed_dtype': np.float32
+            },
+            'dyC0': {  # This column doesn't exist in df
+                'compress': 'round(asinh(dyC0)*100)',
+                'decompress': 'sinh(dyC0_c/100.)',
+                'compressed_dtype': np.int16,
+                'decompressed_dtype': np.float16
+            }
+        }
+
+    def test_default_warn_mode(self):
+        """Test default on_missing='warn' behavior"""
+        adf = AliasDataFrame(self.df)
+
+        with self.assertWarns(UserWarning) as cm:
+            result = adf.compress_columns(self.spec, return_summary=True)
+
+        # Check warning message
+        self.assertIn("Skipping missing columns", str(cm.warning))
+        self.assertIn("dyC0", str(cm.warning))
+
+        # Check summary
+        self.assertEqual(set(result['compressed']), {'dy', 'dz', 'y'})
+        self.assertEqual(result['skipped'], ['dyC0'])
+
+        # Verify compression worked
+        self.assertIn('dy_c', adf.df.columns)
+        self.assertIn('dz_c', adf.df.columns)
+        self.assertIn('y_c', adf.df.columns)
+        self.assertNotIn('dyC0_c', adf.df.columns)
+
+    def test_strict_error_mode(self):
+        """Test on_missing='error' raises KeyError"""
+        adf = AliasDataFrame(self.df)
+
+        with self.assertRaises(KeyError) as cm:
+            adf.compress_columns(self.spec, on_missing='error')
+
+        # Check error message
+        self.assertIn("Missing columns", str(cm.exception))
+        self.assertIn("dyC0", str(cm.exception))
+
+    def test_silent_ignore_mode(self):
+        """Test on_missing='ignore' produces no warnings"""
+        adf = AliasDataFrame(self.df)
+
+        # Use warnings filter to catch any warnings
+        import warnings
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            result = adf.compress_columns(self.spec, on_missing='ignore', return_summary=True)
+
+        # No warnings should be raised
+        self.assertEqual(len(w), 0)
+
+        # But compression should still work
+        self.assertEqual(set(result['compressed']), {'dy', 'dz', 'y'})
+        self.assertEqual(result['skipped'], ['dyC0'])
+
+    def test_explicit_columns_subset(self):
+        """Test compression with explicit columns parameter"""
+        adf = AliasDataFrame(self.df)
+
+        # Only compress dy and dz
+        result = adf.compress_columns(
+            self.spec,
+            columns=['dy', 'dz'],
+            return_summary=True
+        )
+
+        self.assertEqual(set(result['compressed']), {'dy', 'dz'})
+        self.assertEqual(result['skipped'], [])  # All requested columns exist
+
+        # Verify only requested columns were compressed
+        self.assertIn('dy_c', adf.df.columns)
+        self.assertIn('dz_c', adf.df.columns)
+        self.assertNotIn('y_c', adf.df.columns)
+
+    def test_return_summary_default_false(self):
+        """Test that return_summary=False returns self (backward compatible)"""
+        adf = AliasDataFrame(self.df)
+
+        # Default should return self
+        result = adf.compress_columns({'dy': self.spec['dy']})
+        self.assertIs(result, adf)
+
+        # Explicit False should also return self
+        result = adf.compress_columns(
+            {'dz': self.spec['dz']},
+            return_summary=False
+        )
+        self.assertIs(result, adf)
+
+    def test_all_columns_missing_warn(self):
+        """Test behavior when all columns are missing"""
+        adf = AliasDataFrame(pd.DataFrame({'x': [1, 2, 3]}))
+
+        with self.assertWarns(UserWarning) as cm:
+            result = adf.compress_columns(self.spec, return_summary=True)
+
+        self.assertIn("Skipping missing columns", str(cm.warning))
+        self.assertEqual(result['compressed'], [])
+        self.assertEqual(set(result['skipped']), {'dy', 'dz', 'y', 'dyC0'})
+
+    def test_all_columns_missing_error(self):
+        """Test error mode when all columns are missing"""
+        adf = AliasDataFrame(pd.DataFrame({'x': [1, 2, 3]}))
+
+        with self.assertRaises(KeyError) as cm:
+            adf.compress_columns(self.spec, on_missing='error')
+
+        self.assertIn("Missing columns", str(cm.exception))
+
+    def test_partial_missing_with_columns_param(self):
+        """Test warning when some explicitly requested columns are missing"""
+        df = pd.DataFrame({'dy': np.random.randn(10)})
+        adf = AliasDataFrame(df)
+
+        with self.assertWarns(UserWarning) as cm:
+            result = adf.compress_columns(
+                self.spec,
+                columns=['dy', 'dyC0'],  # dyC0 doesn't exist
+                return_summary=True
+            )
+
+        self.assertEqual(result['compressed'], ['dy'])
+        self.assertEqual(result['skipped'], ['dyC0'])
+
+    def test_method_chaining_still_works(self):
+        """Test that method chaining still works with default parameters"""
+        adf = AliasDataFrame(self.df)
+
+        # Should be able to chain
+        result = (adf
+                  .compress_columns({'dy': self.spec['dy']})
+                  .compress_columns({'dz': self.spec['dz']}))
+
+        self.assertIs(result, adf)
+        self.assertIn('dy_c', adf.df.columns)
+        self.assertIn('dz_c', adf.df.columns)
+
+
+
 if __name__ == "__main__":
     unittest.main()