fix wrapper

Author: fresh-borzoni

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

@@ -50,7 +50,6 @@
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.predicate.Predicate;
-import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.shaded.guava32.com.google.common.collect.Lists;
 import org.apache.fluss.types.RowType;
 import org.apache.fluss.utils.ExceptionUtils;
@@ -564,7 +563,11 @@ private List<PartitionInfo> applyPartitionFilter(List<PartitionInfo> partitionIn
                             .filter(
                                     partition ->
                                             partitionFilters.test(
-                                                    toInternalRow(partition, partitionRowType)))
+                                                    PartitionUtils.toPartitionRow(
+                                                            partition
+                                                                    .getResolvedPartitionSpec()
+                                                                    .getPartitionValues(),
+                                                            partitionRowType)))
                             .collect(Collectors.toList());
 
             int filteredSize = filteredPartitionInfos.size();
@@ -587,12 +590,6 @@ private List<PartitionInfo> applyPartitionFilter(List<PartitionInfo> partitionIn
         }
     }
 
-    private static InternalRow toInternalRow(
-            PartitionInfo partitionInfo, RowType partitionRowType) {
-        return PartitionUtils.toPartitionRow(
-                partitionInfo.getResolvedPartitionSpec().getPartitionValues(), partitionRowType);
-    }
-
     /** Init the splits for Fluss. */
     private void checkPartitionChanges(Set<PartitionInfo> partitionInfos, Throwable t) {
         if (closed) {