@SuppressWarnings({"checkstyle:CyclomaticComplexity", "checkstyle:MethodLength"})
  public static Expression convert(Predicate predicate) {
    Operation op = FILTERS.get(predicate.name());
    if (op != null) {
      switch (op) {
        case TRUE:
          return Expressions.alwaysTrue();

        case FALSE:
          return Expressions.alwaysFalse();

        case IS_NULL:
          return isRef(child(predicate)) ? isNull(SparkUtil.toColumnName(child(predicate))) : null;

        case NOT_NULL:
          return isRef(child(predicate)) ? notNull(SparkUtil.toColumnName(child(predicate))) : null;

        case LT:
          if (isRef(leftChild(predicate)) && isLiteral(rightChild(predicate))) {
            String columnName = SparkUtil.toColumnName(leftChild(predicate));
            return lessThan(columnName, convertLiteral(rightChild(predicate)));
          } else if (isRef(rightChild(predicate)) && isLiteral(leftChild(predicate))) {
            String columnName = SparkUtil.toColumnName(rightChild(predicate));
            return greaterThan(columnName, convertLiteral(leftChild(predicate)));
          } else {
            return null;
          }

        case LT_EQ:
          if (isRef(leftChild(predicate)) && isLiteral(rightChild(predicate))) {
            String columnName = SparkUtil.toColumnName(leftChild(predicate));
            return lessThanOrEqual(columnName, convertLiteral(rightChild(predicate)));
          } else if (isRef(rightChild(predicate)) && isLiteral(leftChild(predicate))) {
            String columnName = SparkUtil.toColumnName(rightChild(predicate));
            return greaterThanOrEqual(columnName, convertLiteral(leftChild(predicate)));
          } else {
            return null;
          }

        case GT:
          if (isRef(leftChild(predicate)) && isLiteral(rightChild(predicate))) {
            String columnName = SparkUtil.toColumnName(leftChild(predicate));
            return greaterThan(columnName, convertLiteral(rightChild(predicate)));
          } else if (isRef(rightChild(predicate)) && isLiteral(leftChild(predicate))) {
            String columnName = SparkUtil.toColumnName(rightChild(predicate));
            return lessThan(columnName, convertLiteral(leftChild(predicate)));
          } else {
            return null;
          }

        case GT_EQ:
          if (isRef(leftChild(predicate)) && isLiteral(rightChild(predicate))) {
            String columnName = SparkUtil.toColumnName(leftChild(predicate));
            return greaterThanOrEqual(columnName, convertLiteral(rightChild(predicate)));
          } else if (isRef(rightChild(predicate)) && isLiteral(leftChild(predicate))) {
            String columnName = SparkUtil.toColumnName(rightChild(predicate));
            return lessThanOrEqual(columnName, convertLiteral(leftChild(predicate)));
          } else {
            return null;
          }

        case EQ: // used for both eq and null-safe-eq
          Object value;
          String columnName;
          if (isRef(leftChild(predicate)) && isLiteral(rightChild(predicate))) {
            columnName = SparkUtil.toColumnName(leftChild(predicate));
            value = convertLiteral(rightChild(predicate));
          } else if (isRef(rightChild(predicate)) && isLiteral(leftChild(predicate))) {
            columnName = SparkUtil.toColumnName(rightChild(predicate));
            value = convertLiteral(leftChild(predicate));
          } else {
            return null;
          }

          if (predicate.name().equals(EQ)) {
            // comparison with null in normal equality is always null. this is probably a mistake.
            Preconditions.checkNotNull(
                value, "Expression is always false (eq is not null-safe): %s", predicate);
            return handleEqual(columnName, value);
          } else { // "<=>"
            if (value == null) {
              return isNull(columnName);
            } else {
              return handleEqual(columnName, value);
            }
          }

        case IN:
          if (isSupportedInPredicate(predicate)) {
            return in(
                SparkUtil.toColumnName(childAtIndex(predicate, 0)),
                Arrays.stream(predicate.children())
                    .skip(1)
                    .map(val -> convertLiteral(((Literal<?>) val)))
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList()));
          } else {
            return null;
          }

        case NOT:
          Not notPredicate = (Not) predicate;
          Predicate childPredicate = notPredicate.child();
          if (childPredicate.name().equals(IN) && isSupportedInPredicate(childPredicate)) {
            // infer an extra notNull predicate for Spark NOT IN filters
            // as Iceberg expressions don't follow the 3-value SQL boolean logic
            // col NOT IN (1, 2) in Spark is equal to notNull(col) && notIn(col, 1, 2) in Iceberg
            Expression notIn =
                notIn(
                    SparkUtil.toColumnName(childAtIndex(childPredicate, 0)),
                    Arrays.stream(childPredicate.children())
                        .skip(1)
                        .map(val -> convertLiteral(((Literal<?>) val)))
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList()));
            return and(notNull(SparkUtil.toColumnName(childAtIndex(childPredicate, 0))), notIn);
          } else if (hasNoInFilter(childPredicate)) {
            Expression child = convert(childPredicate);
            if (child != null) {
              return not(child);
            }
          }
          return null;

        case AND:
          {
            And andPredicate = (And) predicate;
            Expression left = convert(andPredicate.left());
            Expression right = convert(andPredicate.right());
            if (left != null && right != null) {
              return and(left, right);
            }
            return null;
          }

        case OR:
          {
            Or orPredicate = (Or) predicate;
            Expression left = convert(orPredicate.left());
            Expression right = convert(orPredicate.right());
            if (left != null && right != null) {
              return or(left, right);
            }
            return null;
          }

        case STARTS_WITH:
          String colName = SparkUtil.toColumnName(leftChild(predicate));
          return startsWith(colName, convertLiteral(rightChild(predicate)).toString());
      }
    }

    return null;
  }
@Test
  public void testV2Filters() {
    Map<String, String> attrMap = Maps.newHashMap();
    attrMap.put("id", "id");
    attrMap.put("`i.d`", "i.d");
    attrMap.put("`i``d`", "i`d");
    attrMap.put("`d`.b.`dd```", "d.b.dd`");
    attrMap.put("a.`aa```.c", "a.aa`.c");

    attrMap.forEach(
        (quoted, unquoted) -> {
          NamedReference namedReference = FieldReference.apply(quoted);
          org.apache.spark.sql.connector.expressions.Expression[] attrOnly =
              new org.apache.spark.sql.connector.expressions.Expression[] {namedReference};

          LiteralValue value = new LiteralValue(1, DataTypes.IntegerType);
          org.apache.spark.sql.connector.expressions.Expression[] attrAndValue =
              new org.apache.spark.sql.connector.expressions.Expression[] {namedReference, value};
          org.apache.spark.sql.connector.expressions.Expression[] valueAndAttr =
              new org.apache.spark.sql.connector.expressions.Expression[] {value, namedReference};

          Predicate isNull = new Predicate("IS_NULL", attrOnly);
          Expression expectedIsNull = Expressions.isNull(unquoted);
          Expression actualIsNull = SparkV2Filters.convert(isNull);
          Assert.assertEquals(
              "IsNull must match", expectedIsNull.toString(), actualIsNull.toString());

          Predicate isNotNull = new Predicate("IS_NOT_NULL", attrOnly);
          Expression expectedIsNotNull = Expressions.notNull(unquoted);
          Expression actualIsNotNull = SparkV2Filters.convert(isNotNull);
          Assert.assertEquals(
              "IsNotNull must match", expectedIsNotNull.toString(), actualIsNotNull.toString());

          Predicate lt1 = new Predicate("<", attrAndValue);
          Expression expectedLt1 = Expressions.lessThan(unquoted, 1);
          Expression actualLt1 = SparkV2Filters.convert(lt1);
          Assert.assertEquals("LessThan must match", expectedLt1.toString(), actualLt1.toString());

          Predicate lt2 = new Predicate("<", valueAndAttr);
          Expression expectedLt2 = Expressions.greaterThan(unquoted, 1);
          Expression actualLt2 = SparkV2Filters.convert(lt2);
          Assert.assertEquals("LessThan must match", expectedLt2.toString(), actualLt2.toString());

          Predicate ltEq1 = new Predicate("<=", attrAndValue);
          Expression expectedLtEq1 = Expressions.lessThanOrEqual(unquoted, 1);
          Expression actualLtEq1 = SparkV2Filters.convert(ltEq1);
          Assert.assertEquals(
              "LessThanOrEqual must match", expectedLtEq1.toString(), actualLtEq1.toString());

          Predicate ltEq2 = new Predicate("<=", valueAndAttr);
          Expression expectedLtEq2 = Expressions.greaterThanOrEqual(unquoted, 1);
          Expression actualLtEq2 = SparkV2Filters.convert(ltEq2);
          Assert.assertEquals(
              "LessThanOrEqual must match", expectedLtEq2.toString(), actualLtEq2.toString());

          Predicate gt1 = new Predicate(">", attrAndValue);
          Expression expectedGt1 = Expressions.greaterThan(unquoted, 1);
          Expression actualGt1 = SparkV2Filters.convert(gt1);
          Assert.assertEquals(
              "GreaterThan must match", expectedGt1.toString(), actualGt1.toString());

          Predicate gt2 = new Predicate(">", valueAndAttr);
          Expression expectedGt2 = Expressions.lessThan(unquoted, 1);
          Expression actualGt2 = SparkV2Filters.convert(gt2);
          Assert.assertEquals(
              "GreaterThan must match", expectedGt2.toString(), actualGt2.toString());

          Predicate gtEq1 = new Predicate(">=", attrAndValue);
          Expression expectedGtEq1 = Expressions.greaterThanOrEqual(unquoted, 1);
          Expression actualGtEq1 = SparkV2Filters.convert(gtEq1);
          Assert.assertEquals(
              "GreaterThanOrEqual must match", expectedGtEq1.toString(), actualGtEq1.toString());

          Predicate gtEq2 = new Predicate(">=", valueAndAttr);
          Expression expectedGtEq2 = Expressions.lessThanOrEqual(unquoted, 1);
          Expression actualGtEq2 = SparkV2Filters.convert(gtEq2);
          Assert.assertEquals(
              "GreaterThanOrEqual must match", expectedGtEq2.toString(), actualGtEq2.toString());

          Predicate eq1 = new Predicate("=", attrAndValue);
          Expression expectedEq1 = Expressions.equal(unquoted, 1);
          Expression actualEq1 = SparkV2Filters.convert(eq1);
          Assert.assertEquals("EqualTo must match", expectedEq1.toString(), actualEq1.toString());

          Predicate eq2 = new Predicate("=", valueAndAttr);
          Expression expectedEq2 = Expressions.equal(unquoted, 1);
          Expression actualEq2 = SparkV2Filters.convert(eq2);
          Assert.assertEquals("EqualTo must match", expectedEq2.toString(), actualEq2.toString());

          Predicate eqNullSafe1 = new Predicate("<=>", attrAndValue);
          Expression expectedEqNullSafe1 = Expressions.equal(unquoted, 1);
          Expression actualEqNullSafe1 = SparkV2Filters.convert(eqNullSafe1);
          Assert.assertEquals(
              "EqualNullSafe must match",
              expectedEqNullSafe1.toString(),
              actualEqNullSafe1.toString());

          Predicate eqNullSafe2 = new Predicate("<=>", valueAndAttr);
          Expression expectedEqNullSafe2 = Expressions.equal(unquoted, 1);
          Expression actualEqNullSafe2 = SparkV2Filters.convert(eqNullSafe2);
          Assert.assertEquals(
              "EqualNullSafe must match",
              expectedEqNullSafe2.toString(),
              actualEqNullSafe2.toString());

          LiteralValue str =
              new LiteralValue(UTF8String.fromString("iceberg"), DataTypes.StringType);
          org.apache.spark.sql.connector.expressions.Expression[] attrAndStr =
              new org.apache.spark.sql.connector.expressions.Expression[] {namedReference, str};
          Predicate startsWith = new Predicate("STARTS_WITH", attrAndStr);
          Expression expectedStartsWith = Expressions.startsWith(unquoted, "iceberg");
          Expression actualStartsWith = SparkV2Filters.convert(startsWith);
          Assert.assertEquals(
              "StartsWith must match", expectedStartsWith.toString(), actualStartsWith.toString());

          Predicate in = new Predicate("IN", attrAndValue);
          Expression expectedIn = Expressions.in(unquoted, 1);
          Expression actualIn = SparkV2Filters.convert(in);
          Assert.assertEquals("In must match", expectedIn.toString(), actualIn.toString());

          Predicate and = new And(lt1, eq1);
          Expression expectedAnd = Expressions.and(expectedLt1, expectedEq1);
          Expression actualAnd = SparkV2Filters.convert(and);
          Assert.assertEquals("And must match", expectedAnd.toString(), actualAnd.toString());

          org.apache.spark.sql.connector.expressions.Expression[] attrAndAttr =
              new org.apache.spark.sql.connector.expressions.Expression[] {
                namedReference, namedReference
              };
          Predicate invalid = new Predicate("<", attrAndAttr);
          Predicate andWithInvalidLeft = new And(invalid, eq1);
          Expression convertedAnd = SparkV2Filters.convert(andWithInvalidLeft);
          Assert.assertEquals("And must match", convertedAnd, null);

          Predicate or = new Or(lt1, eq1);
          Expression expectedOr = Expressions.or(expectedLt1, expectedEq1);
          Expression actualOr = SparkV2Filters.convert(or);
          Assert.assertEquals("Or must match", expectedOr.toString(), actualOr.toString());

          Predicate orWithInvalidLeft = new Or(invalid, eq1);
          Expression convertedOr = SparkV2Filters.convert(orWithInvalidLeft);
          Assert.assertEquals("Or must match", convertedOr, null);

          Predicate not = new Not(lt1);
          Expression expectedNot = Expressions.not(expectedLt1);
          Expression actualNot = SparkV2Filters.convert(not);
          Assert.assertEquals("Not must match", expectedNot.toString(), actualNot.toString());
        });
  }