[flink]Support partition pushdown in Flink connector

Author: Alibaba-HZY

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java

@@ -1153,39 +1153,6 @@ private List<String> writeRowsToTwoPartition(TablePath tablePath, Collection<Str
     @Test
     void testStreamingReadPartitionPushDownWithInExpr() throws Exception {
 
-        //        tEnv.executeSql(
-        //                "create table partitioned_table"
-        //                        + " (a int not null, b varchar, c string, primary key (a, c) NOT
-        // ENFORCED) partitioned by (c) "
-        //                        + "with ('table.auto-partition.enabled' = 'true',
-        // 'table.auto-partition.time-unit' = 'year')");
-        //        TablePath tablePath = TablePath.of(DEFAULT_DB, "partitioned_table");
-        //
-        //        // write data into partitions and wait snapshot is done
-        //        Map<Long, String> partitionNameById =
-        //                waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(),
-        // tablePath);
-        //        List<String> expectedRowValues =
-        //                writeRowsToPartition(tablePath, partitionNameById.values()).stream()
-        //                        .filter(s -> s.contains("2025"))
-        //                        .collect(Collectors.toList());
-        //        waitUtilAllBucketFinishSnapshot(admin, tablePath, partitionNameById.values());
-        //
-        //        org.apache.flink.util.CloseableIterator<Row> rowIter =
-        //                tEnv.executeSql("select * from partitioned_table where c in(
-        // '2000','2001','2025')")
-        //                        .collect();
-        //
-        //        assertResultsIgnoreOrder(rowIter, expectedRowValues, false);
-        //
-        //        // then create some new partitions, and write rows to the new partitions
-        //        List<String> newPartitions = Arrays.asList("2000", "2001");
-        //        FlinkTestBase.createPartitions(
-        //                FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), tablePath, newPartitions);
-        //        // write data to the new partitions
-        //        expectedRowValues = writeRowsToPartition(tablePath, newPartitions);
-        //        assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
-
         tEnv.executeSql(
                 "create table partitioned_table_in"
                         + " (a int not null, b varchar, c string, primary key (a, c) NOT ENFORCED) partitioned by (c) ");