Add basic predicate-pushdown optimization

.github/docker-compose.yaml

@@ -11,5 +11,7 @@ services:
         container_name: dask-worker
         image: daskdev/dask:latest
         command: dask-worker dask-scheduler:8786
+        environment:
+            EXTRA_CONDA_PACKAGES: "pyarrow>1.0.0"  # required for parquet IO
         volumes:
             - /tmp:/tmp

dask_sql/physical/rel/logical/filter.py

@@ -1,12 +1,14 @@
 import logging
 from typing import TYPE_CHECKING, Union
 
+import dask.config as dask_config
 import dask.dataframe as dd
 import numpy as np
 
 from dask_sql.datacontainer import DataContainer
 from dask_sql.physical.rel.base import BaseRelPlugin
 from dask_sql.physical.rex import RexConverter
+from dask_sql.physical.utils.filter import attempt_predicate_pushdown
 
 if TYPE_CHECKING:
     import dask_sql
@@ -31,7 +33,11 @@ def filter_or_scalar(df: dd.DataFrame, filter_condition: Union[np.bool_, dd.Seri
 
     # In SQL, a NULL in a boolean is False on filtering
     filter_condition = filter_condition.fillna(False)
-    return df[filter_condition]
+    out = df[filter_condition]
+    if dask_config.get("sql.predicate_pushdown"):
+        return attempt_predicate_pushdown(out)
+    else:
+        return out
 
 
 class DaskFilterPlugin(BaseRelPlugin):