public  String generatePushDownFilter(List<String> writtenPartitions, List<FieldSchema> partitionFields, HiveSyncConfig config) {
    PartitionValueExtractor partitionValueExtractor = ReflectionUtils
        .loadClass(config.getStringOrDefault(META_SYNC_PARTITION_EXTRACTOR_CLASS));

    List<Partition> partitions = writtenPartitions.stream().map(s -> {
      List<String> values = partitionValueExtractor.extractPartitionValuesInPath(s);

      if (values.size() != partitionFields.size()) {
        throw new HoodieHiveSyncException("Partition fields and values should be same length"
            + ", but got partitionFields: " + partitionFields + " with values: " + values);
      }

      return new Partition(values, null);
    }).collect(Collectors.toList());

    Expression filter;
    int estimateSize = partitionFields.size() * partitions.size();
    if (estimateSize > config.getIntOrDefault(HIVE_SYNC_FILTER_PUSHDOWN_MAX_SIZE)) {
      filter = buildMinMaxPartitionExpression(partitions, partitionFields);
    } else {
      filter = buildPartitionExpression(partitions, partitionFields);
    }

    if (filter != null) {
      return generateFilterString(filter);
    }

    return "";
  }
@Test
  public void testPushDownFilterIfExceedLimit() {
    Properties props = new Properties();
    props.put(HIVE_SYNC_FILTER_PUSHDOWN_MAX_SIZE.key(), "0");
    HiveSyncConfig config = new HiveSyncConfig(props);
    List<FieldSchema> partitionFieldSchemas = new ArrayList<>(4);
    partitionFieldSchemas.add(new FieldSchema("date", "date"));
    partitionFieldSchemas.add(new FieldSchema("year", "string"));
    partitionFieldSchemas.add(new FieldSchema("month", "int"));
    partitionFieldSchemas.add(new FieldSchema("day", "bigint"));

    List<String> writtenPartitions = new ArrayList<>();
    writtenPartitions.add("2022-09-01/2022/9/1");

    assertEquals("(((date = 2022-09-01 AND year = \"2022\") AND month = 9) AND day = 1)",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));

    writtenPartitions.add("2022-09-02/2022/9/2");
    writtenPartitions.add("2022-09-03/2022/9/2");
    writtenPartitions.add("2022-09-04/2022/9/2");
    assertEquals(
        "((((date >= 2022-09-01 AND date <= 2022-09-04) AND year = \"2022\") AND month = 9) AND (day >= 1 AND day <= 2))",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));

    // If there are incompatible types to convert as filters inside partition
    partitionFieldSchemas.clear();
    writtenPartitions.clear();
    partitionFieldSchemas.add(new FieldSchema("date", "date"));
    partitionFieldSchemas.add(new FieldSchema("finished", "boolean"));

    writtenPartitions.add("2022-09-01/true");
    assertEquals("date = 2022-09-01",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));
    writtenPartitions.add("2022-09-02/true");
    writtenPartitions.add("2022-09-03/false");
    writtenPartitions.add("2022-09-04/false");
    assertEquals("(date >= 2022-09-01 AND date <= 2022-09-04)",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));

    // If no compatible types matched to convert as filters
    partitionFieldSchemas.clear();
    writtenPartitions.clear();
    partitionFieldSchemas.add(new FieldSchema("finished", "boolean"));

    writtenPartitions.add("true");
    assertEquals("",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));
    writtenPartitions.add("false");
    writtenPartitions.add("false");
    writtenPartitions.add("false");
    assertEquals("",
        partitionFilterGenerator.generatePushDownFilter(writtenPartitions, partitionFieldSchemas, config));
  }