Failed to push down partition filters in flink batch source For primary key table

[luoyuxia]

Search before asking

• I searched in the issues and found nothing similar.

Description

    public Result applyFilters(List<ResolvedExpression> filters) {

        List<ResolvedExpression> acceptedFilters = new ArrayList<>();
        List<ResolvedExpression> remainingFilters = new ArrayList<>();

        // primary pushdown
        // (1) batch execution mode,
        // (2) default (full) startup mode,
        // (3) the table is a pk table,
        // (4) all filters are pk field equal expression
        if (!streaming
                && startupOptions.startupMode == FlinkConnectorOptions.ScanStartupMode.FULL
                && hasPrimaryKey()
                && **filters.size() == primaryKeyIndexes.length**) {

The filters.size() == primaryKeyIndexes.length is wrong which will cause some filter push down fall into the pk lookup code branch, then resuting wrong result.

Willingness to contribute

• I'm willing to submit a PR!

[luoyuxia]

cc @loserwang1024