Added parquet support for filters parameter

python/lsst/daf/butler/formatters/parquet.py

@@ -131,6 +131,8 @@ def _read_parquet(
         component: str | None = None,
         expected_size: int = -1,
     ) -> Any:
+        import pyarrow.dataset as ds
+
         with generic_open(path, fs) as handle:
             schema = pq.read_schema(handle)
 
@@ -156,6 +158,7 @@ def _read_parquet(
             return len(temp_table[schema.names[0]])
 
         par_columns = None
+        par_filters = None
         if self.file_descriptor.parameters:
             par_columns = self.file_descriptor.parameters.pop("columns", None)
             if par_columns:
@@ -190,6 +193,27 @@ def _read_parquet(
                         par_columns,
                     )
 
+            par_filters = self.file_descriptor.parameters.pop("filters", None)
+            if par_filters:
+                # Validate basic filter structure
+                if not isinstance(par_filters, ds.Expression):
+                    if not isinstance(par_filters, list):
+                        raise TypeError("Filters must be a list or a pyarrow.dataset.Expression.")
+
+                    # Lists must contain tuples or lists of tuples
+                    if all(isinstance(item, tuple) for item in par_filters):
+                        par_filters = [par_filters]  # convert to DNF for column checking
+                    elif not all(isinstance(item, list) for item in par_filters):
+                        raise TypeError("List filters must contain tuples or lists of tuples.")
+
+                    # Ensure requested columns are in schema
+                    for predicate in par_filters:
+                        for col, _, _ in predicate:
+                            if col not in schema.names:
+                                raise ValueError(
+                                    f"Column {col} specified in filters not available in parquet file."
+                                )
+
             if len(self.file_descriptor.parameters):
                 raise ValueError(
                     f"Unsupported parameters {self.file_descriptor.parameters} in ArrowTable read."
@@ -202,6 +226,7 @@ def _read_parquet(
                 columns=par_columns,
                 use_threads=False,
                 use_pandas_metadata=(b"pandas" in metadata),
+                filters=par_filters,
             )
 
         return arrow_table

tests/test_parquet.py

@@ -408,6 +408,12 @@ def testSingleIndexDataFrame(self):
         # Passing an unrecognized column should be a ValueError.
         with self.assertRaises(ValueError):
             self.butler.get(self.datasetType, dataId={}, parameters={"columns": ["e"]})
+        # Filter predicates with unrecognized column should be a ValueError.
+        # FIXME: ValueErrorWeirdness1. This raises an AssertionError,
+        #        but Ctrl+f "ValueErrorWeirdness2" below.
+        # with self.assertRaises(ValueError):
+        #     self.butler.get(self.datasetType, dataId={},
+        #                     parameters={"filters": [("e", ">", 1)]})
 
     def testSingleIndexDataFrameWithLists(self):
         df1, allColumns = _makeSingleIndexDataFrame(include_lists=True)
@@ -455,6 +461,11 @@ def testMultiIndexDataFrame(self):
         # Passing an unrecognized column should be a ValueError.
         with self.assertRaises(ValueError):
             self.butler.get(self.datasetType, dataId={}, parameters={"columns": ["d"]})
+        # Filter predicates with unrecognized column should be a ValueError.
+        # FIXME: ValueErrorWeirdness2. This raises a ValueError, but Ctrl+f
+        #        "ValueErrorWeirdness1" above.
+        # df7 = self.butler.get(self.datasetType, dataId={},
+        #                       parameters={"filters": [("d", ">", 1)]})
 
     def testSingleIndexDataFrameEmptyString(self):
         """Test persisting a single index dataframe with empty strings."""