class BigQueryFilter implements BeamSqlTableFilter {

  private Pair<Boolean, Integer> isSupported(RexNode node) {
    int numberOfInputRefs = 0;
    boolean isSupported = true;

    if (node instanceof RexCall) {
      RexCall compositeNode = (RexCall) node;

      // Only support comparisons in a predicate, some sql functions such as:
      //  CAST, TRIM? and REVERSE? should be supported as well.
      if (!node.getKind().belongsTo(SUPPORTED_OPS)) {
        isSupported = false;
      } else {
        for (RexNode operand : compositeNode.getOperands()) {
          // All operands must be supported for a parent node to be supported.
          Pair<Boolean, Integer> childSupported = isSupported(operand);
          // BigQuery supports complex combinations of both conjunctions (AND) and disjunctions
          // (OR).
          if (!node.getKind().belongsTo(ImmutableSet.of(AND, OR))) {
            numberOfInputRefs += childSupported.getRight();
          }
          // Predicate functions, where more than one field is involved are unsupported.
          isSupported = numberOfInputRefs < 2 && childSupported.getLeft();
        }
      }
    } else if (node instanceof RexInputRef) {
      numberOfInputRefs = 1;
    } else if (node instanceof RexLiteral) {
      // RexLiterals are expected, but no action is needed.
    } else {
      throw new UnsupportedOperationException(
          "Encountered an unexpected node type: " + node.getClass().getSimpleName());
    }

    return Pair.of(isSupported, numberOfInputRefs);
  }

  public  BigQueryFilter(List<RexNode> predicateCNF);

}