[core] Fix that cannot read binlog table with projection (#6417)

Author: yuzelin

paimon-core/src/main/java/org/apache/paimon/table/system/BinlogTable.java

@@ -131,7 +131,7 @@ public RecordReader<InternalRow> createReader(Split split) throws IOException {
                         (row1, row2) ->
                                 new AuditLogRow(
                                         readProjection, convertToArray(row1, row2, fieldGetters)),
-                        wrapped.rowType());
+                        this.wrappedReadType);
             } else {
                 return dataRead.createReader(split)
                         .transform(

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java

@@ -1172,6 +1172,26 @@ public void testBinlogTableStreamRead() throws Exception {
         iterator.close();
     }
 
+    @Test
+    public void testBinlogTableStreamReadWithProjection() throws Exception {
+        sql(
+                "CREATE TABLE T (a INT, b INT, primary key (a) NOT ENFORCED) with ('changelog-producer' = 'lookup', "
+                        + "'bucket' = '2')");
+        BlockingIterator<Row, Row> iterator =
+                streamSqlBlockIter(
+                        "SELECT rowkind, a FROM T$binlog /*+ OPTIONS('scan.mode' = 'latest') */");
+        sql("INSERT INTO T VALUES (1, 2)");
+        sql("INSERT INTO T VALUES (1, 3)");
+        sql("INSERT INTO T VALUES (2, 2)");
+        List<Row> rows = iterator.collect(3);
+        assertThat(rows)
+                .containsExactly(
+                        Row.of("+I", new Integer[] {1}),
+                        Row.of("+U", new Integer[] {1, 1}),
+                        Row.of("+I", new Integer[] {2}));
+        iterator.close();
+    }
+
     @Test
     public void testBinlogTableBatchRead() throws Exception {
         sql(