[core][mosaic] Preserve projected raw reads

paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java

@@ -124,6 +124,9 @@ public SplitRead<InternalRow> withIOManager(@Nullable IOManager ioManager) {
 
     @Override
     public SplitRead<InternalRow> withReadType(RowType readRowType) {
+        if (!this.readRowType.equals(readRowType)) {
+            formatReaderMappings.clear();
+        }
         this.readRowType = readRowType;
         return this;
     }
@@ -194,7 +197,9 @@ public RecordReader<InternalRow> createReader(
                 pathFactory.createDataFilePathFactory(partition, bucket);
         List<ReaderSupplier<InternalRow>> suppliers = new ArrayList<>();
 
-        Builder formatReaderMappingBuilder = createFormatReaderMappingBuilder();
+        RowType outputRowType = readRowType;
+        Builder formatReaderMappingBuilder =
+                createFormatReaderMappingBuilder(outputRowType, topN, limit);
 
         for (DataFileMeta file : files) {
             suppliers.add(
@@ -203,6 +208,7 @@ public RecordReader<InternalRow> createReader(
                             dataFilePathFactory,
                             file,
                             formatReaderMappingBuilder,
+                            outputRowType,
                             dvFactories,
                             null));
         }
@@ -221,28 +227,26 @@ FileRecordReader<InternalRow> createFileReader(
                 dataFile.fileName(), () -> dvFactory.create(dataFile.fileName()).orElse(null));
         DataFilePathFactory dataFilePathFactory =
                 pathFactory.createDataFilePathFactory(dataSplit.partition(), dataSplit.bucket());
+        RowType outputRowType = readRowType;
         return (FileRecordReader<InternalRow>)
                 createFileReader(
                                 dataSplit.partition(),
                                 dataFilePathFactory,
                                 dataFile,
                                 // The caller has already selected the rows. Applying a regular
                                 // TopN or limit before position filtering can drop hits.
-                                createFormatReaderMappingBuilder(null, null),
+                                createFormatReaderMappingBuilder(outputRowType, null, null),
+                                outputRowType,
                                 dvFactories,
                                 selectedPositions)
                         .get();
     }
 
-    private Builder createFormatReaderMappingBuilder() {
-        return createFormatReaderMappingBuilder(topN, limit);
-    }
-
     private Builder createFormatReaderMappingBuilder(
-            @Nullable TopN pushDownTopN, @Nullable Integer pushDownLimit) {
+            RowType outputRowType, @Nullable TopN pushDownTopN, @Nullable Integer pushDownLimit) {
         return new Builder(
                 formatDiscover,
-                readRowType.getFields(),
+                outputRowType.getFields(),
                 schema -> {
                     if (rowTrackingEnabled) {
                         // maybe file has no row id and sequence number, but in manifest entry
@@ -261,6 +265,7 @@ private ReaderSupplier<InternalRow> createFileReader(
             DataFilePathFactory dataFilePathFactory,
             DataFileMeta file,
             Builder formatBuilder,
+            RowType outputRowType,
             @Nullable Map<String, IOExceptionSupplier<DeletionVector>> dvFactories,
             @Nullable RoaringBitmap32 selectedPositions) {
         String formatIdentifier = DataFilePathFactory.formatIdentifier(file.fileName());
@@ -285,6 +290,7 @@ private ReaderSupplier<InternalRow> createFileReader(
                         file,
                         dataFilePathFactory,
                         formatReaderMapping,
+                        outputRowType,
                         dvFactory,
                         selectedPositions);
     }
@@ -294,6 +300,7 @@ private FileRecordReader<InternalRow> createFileReader(
             DataFileMeta file,
             DataFilePathFactory dataFilePathFactory,
             FormatReaderMapping formatReaderMapping,
+            RowType outputRowType,
             IOExceptionSupplier<DeletionVector> dvFactory,
             @Nullable RoaringBitmap32 selectedPositions)
             throws IOException {
@@ -334,7 +341,7 @@ private FileRecordReader<InternalRow> createFileReader(
                         fileIO, dataFilePathFactory.toPath(file), file.fileSize(), selection);
         FileRecordReader<InternalRow> fileRecordReader =
                 new DataFileRecordReader(
-                        schema.logicalRowType(),
+                        outputRowType,
                         formatReaderMapping.getReaderFactory(),
                         formatReaderContext,
                         ignoreCorruptFiles,

paimon-core/src/test/java/org/apache/paimon/operation/RawFileSplitReadTest.java

@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.FlushingFileFormat;
+import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.SchemaUtils;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RawFileSplitRead}. */
+class RawFileSplitReadTest {
+
+    @TempDir java.nio.file.Path tempDir;
+
+    @Test
+    void testReaderMappingIsNotSharedBetweenReadTypes() throws Exception {
+        FileStoreTable table = createTable("mapping-cache");
+        DataSplit split = singleSplit(table);
+        InnerTableRead read = table.newRead();
+
+        RowType firstProjection = table.rowType().project("first");
+        read.withReadType(firstProjection);
+        try (RecordReader<InternalRow> reader = read.createReader(split)) {
+            RecordReader.RecordIterator<InternalRow> batch = reader.readBatch();
+            assertThat(batch).isNotNull();
+            assertThat(batch.next().getString(0).toString()).isEqualTo("value");
+            assertThat(batch.next()).isNull();
+            batch.releaseBatch();
+        }
+
+        RowType secondProjection = table.rowType().project("second");
+        read.withReadType(secondProjection);
+        try (RecordReader<InternalRow> reader = read.createReader(split)) {
+            RecordReader.RecordIterator<InternalRow> batch = reader.readBatch();
+            assertThat(batch).isNotNull();
+            InternalRow row = batch.next();
+            assertThat(row).isNotNull();
+            assertThat(row.getFieldCount()).isEqualTo(1);
+            assertThat(row.getInt(0)).isEqualTo(42);
+            assertThat(batch.next()).isNull();
+            batch.releaseBatch();
+        }
+    }
+
+    @Test
+    void testEqualReadTypeReusesFormatReaderMapping() throws Exception {
+        FileStoreTable table = createTable("equal-mapping-cache");
+        AtomicInteger readerFactoryCreations = new AtomicInteger();
+        FileFormat countingFormat =
+                new FlushingFileFormat(table.coreOptions().fileFormatString()) {
+                    @Override
+                    public FormatReaderFactory createReaderFactory(
+                            RowType dataSchemaRowType,
+                            RowType projectedRowType,
+                            List<Predicate> filters) {
+                        readerFactoryCreations.incrementAndGet();
+                        return super.createReaderFactory(
+                                dataSchemaRowType, projectedRowType, filters);
+                    }
+                };
+        RawFileSplitRead read =
+                new RawFileSplitRead(
+                        table.fileIO(),
+                        table.schemaManager(),
+                        table.schema(),
+                        table.rowType(),
+                        ignored -> countingFormat,
+                        table.store().pathFactory(),
+                        table.coreOptions());
+        DataSplit split = singleSplit(table);
+
+        RowType firstProjection = table.rowType().project("first");
+        read.withReadType(firstProjection);
+        try (RecordReader<InternalRow> ignored = read.createReader(split)) {
+            assertThat(readerFactoryCreations).hasValue(1);
+        }
+
+        RowType equalFirstProjection = table.rowType().project("first");
+        assertThat(equalFirstProjection).isEqualTo(firstProjection).isNotSameAs(firstProjection);
+        read.withReadType(equalFirstProjection);
+        try (RecordReader<InternalRow> ignored = read.createReader(split)) {
+            assertThat(readerFactoryCreations).hasValue(1);
+        }
+
+        read.withReadType(table.rowType().project("second"));
+        try (RecordReader<InternalRow> ignored = read.createReader(split)) {
+            assertThat(readerFactoryCreations).hasValue(2);
+        }
+    }
+
+    private FileStoreTable createTable(String directory) throws Exception {
+        Path tablePath = new Path(tempDir.resolve(directory).toUri());
+        Options options = new Options();
+        options.set(CoreOptions.PATH, tablePath.toString());
+        options.set(CoreOptions.BUCKET, 1);
+        options.set(CoreOptions.BUCKET_KEY, "first");
+        Schema schema =
+                Schema.newBuilder()
+                        .column("first", DataTypes.STRING())
+                        .column("second", DataTypes.INT())
+                        .options(options.toMap())
+                        .build();
+        TableSchema tableSchema =
+                SchemaUtils.forceCommit(new SchemaManager(LocalFileIO.create(), tablePath), schema);
+        FileStoreTable table =
+                FileStoreTableFactory.create(LocalFileIO.create(), tablePath, tableSchema);
+
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            write.write(GenericRow.of(BinaryString.fromString("value"), 42));
+            commit.commit(write.prepareCommit());
+        }
+        return table;
+    }
+
+    private static DataSplit singleSplit(FileStoreTable table) {
+        List<DataSplit> splits = table.newSnapshotReader().read().dataSplits();
+        assertThat(splits).hasSize(1);
+        return splits.get(0);
+    }
+}

paimon-mosaic/pom.xml

@@ -75,6 +75,13 @@ under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.paimon</groupId>
+            <artifactId>paimon-format</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.paimon</groupId>
             <artifactId>paimon-core</artifactId>