code: def apply_filters(parts, statistics, filters):
      """Apply filters onto parts/statistics pairs
  
      Parameters
      ----------
      parts: list
          Tokens corresponding to row groups to read in the future
      statistics: List[dict]
          List of statistics for each part, including min and max values
      filters: Union[List[Tuple[str, str, Any]], List[List[Tuple[str, str, Any]]]]
          List of filters to apply, like ``[[('x', '=', 0), ...], ...]``. This
          implements partition-level (hive) filtering only, i.e., to prevent the
          loading of some row-groups and/or files.
  
          Predicates can be expressed in disjunctive normal form (DNF). This means
          that the innermost tuple describes a single column predicate. These
          inner predicates are combined with an AND conjunction into a larger
          predicate. The outer-most list then combines all of the combined
          filters with an OR disjunction.
  
          Predicates can also be expressed as a List[Tuple]. These are evaluated
          as an AND conjunction. To express OR in predicates, one must use the
          (preferred) List[List[Tuple]] notation.
  
          Note that the "fastparquet" engine does not currently support DNF for
          the filtering of partitioned columns (List[Tuple] is required).
      Returns
      -------
      parts, statistics: the same as the input, but possibly a subset
      """
  
      # Supported predicate operators
      _supported_operators = {
          "=",
          "==",
          "!=",
          "<",
          "<=",
          ">",
          ">=",
          "is",
          "is not",
          "in",
          "not in",
      }
  
      def apply_conjunction(parts, statistics, conjunction):
          for column, operator, value in conjunction:
              if operator not in _supported_operators:
                  # Use same error message as `_filters_to_expression`
                  raise ValueError(
                      f'"{(column, operator, value)}" is not a valid operator in predicates.'
                  )
              elif operator in ("in", "not in") and not isinstance(
                  value, (list, set, tuple)
              ):
                  raise TypeError(
                      "Value of 'in' filter must be a list, set, or tuple."
                  )
              out_parts = []
              out_statistics = []
              for part, stats in zip(parts, statistics):
                  if "filter" in stats and stats["filter"]:
                      continue  # Filtered by engine
                  try:
                      c = toolz.groupby("name", stats["columns"])[
                          column
                      ][0]
                      min = c["min"]
                      max = c["max"]
                      null_count = c.get("null_count", None)
                  except KeyError:
                      out_parts.append(part)
                      out_statistics.append(stats)
                  else:
                      if (
                          # Must allow row-groups with "missing" stats
                          (
                              min is None
                              and max is None
                              and not null_count
                          )
                          # Check "is" and "is not" fiters first
                          or operator == "is"
                          and null_count
                          or operator == "is not"
                          and (not pd.isna(min) or not pd.isna(max))
                          # Allow all-null row-groups if not fitering out nulls
                          or operator != "is not"
                          and min is None
                          and max is None
                          and null_count
                          # Start conventional (non-null) fitering
                          # (main/max cannot be None for remaining checks)
                          or operator in ("==", "=")
                          and min <= value <= max
                          or operator == "!="
                          and (
                              null_count or min != value or max != value
                          )
                          or operator == "<"
                          and min < value
                          or operator == "<="
                          and min <= value
                          or operator == ">"
                          and max > value
                          or operator == ">="
                          and max >= value
                          or operator == "in"
                          and any(min <= item <= max for item in value)
                          or operator == "not in"
                          and not any(
                              min == max == item for item in value
                          )
                      ):
                          out_parts.append(part)
                          out_statistics.append(stats)
  
              parts, statistics = out_parts, out_statistics
  
          return parts, statistics
  
      conjunction, *disjunction = (
          filters if isinstance(filters[0], list) else [filters]
      )
  
      out_parts, out_statistics = apply_conjunction(
          parts, statistics, conjunction
      )
      for conjunction in disjunction:
          for part, stats in zip(
              *apply_conjunction(parts, statistics, conjunction)
          ):
              if part not in out_parts:
                  out_parts.append(part)
                  out_statistics.append(stats)
  
      return out_parts, out_statistics
  
imports: from __future__ import annotations
  import contextlib
  import math
  import warnings
  from typing import Literal
  import pandas
  import tlz
  from fsspec.core import get_fs_token_paths
  from fsspec.utils import stringify_path
  import dask
  from dask.base import tokenize
  from dask.blockwise import BlockIndex
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.core import DataFrame, Scalar
  from dask.dataframe.io.io import from_map
  from dask.dataframe.io.parquet.utils import Engine, _sort_and_analyze_paths, _split_user_options
  from dask.dataframe.io.utils import DataFrameIOFunction, _is_local_fs
  from dask.dataframe.methods import concat
  from dask.delayed import Delayed
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import DataFrameIOLayer
  from dask.utils import apply, import_required, natural_sort_key, parse_bytes
label_desc: Use image node_id 8 for calling a global function with example usage: apply_filters(parts, statistics, filters) and returns: out_parts,  out_statistics, parts,  statistics