[core] Handle query auth wrappers in split consumers

paimon-common/src/main/java/org/apache/paimon/reader/ScoreRecordReader.java

@@ -18,14 +18,53 @@
 
 package org.apache.paimon.reader;
 
+import org.apache.paimon.utils.Filter;
+
 import javax.annotation.Nullable;
 
 import java.io.IOException;
+import java.util.function.Function;
 
 /** A {@link RecordReader} whose records expose vector-search scores and row identifiers. */
 public interface ScoreRecordReader<T> extends RecordReader<T> {
 
     @Nullable
     @Override
     ScoreRecordIterator<T> readBatch() throws IOException;
+
+    @Override
+    default <R> ScoreRecordReader<R> transform(Function<T, R> function) {
+        ScoreRecordReader<T> thisReader = this;
+        return new ScoreRecordReader<R>() {
+            @Nullable
+            @Override
+            public ScoreRecordIterator<R> readBatch() throws IOException {
+                ScoreRecordIterator<T> iterator = thisReader.readBatch();
+                return iterator == null ? null : iterator.transform(function);
+            }
+
+            @Override
+            public void close() throws IOException {
+                thisReader.close();
+            }
+        };
+    }
+
+    @Override
+    default ScoreRecordReader<T> filter(Filter<T> filter) {
+        ScoreRecordReader<T> thisReader = this;
+        return new ScoreRecordReader<T>() {
+            @Nullable
+            @Override
+            public ScoreRecordIterator<T> readBatch() throws IOException {
+                ScoreRecordIterator<T> iterator = thisReader.readBatch();
+                return iterator == null ? null : iterator.filter(filter);
+            }
+
+            @Override
+            public void close() throws IOException {
+                thisReader.close();
+            }
+        };
+    }
 }

paimon-common/src/test/java/org/apache/paimon/reader/ScoreRecordReaderTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.reader;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ScoreRecordReader}. */
+public class ScoreRecordReaderTest {
+
+    @Test
+    public void testFilterAndTransformPreserveScoreMetadata() throws IOException {
+        ScoreRecordReader<Integer> reader =
+                new ScoreRecordReader<Integer>() {
+                    private boolean emitted;
+
+                    @Nullable
+                    @Override
+                    public ScoreRecordIterator<Integer> readBatch() {
+                        if (emitted) {
+                            return null;
+                        }
+                        emitted = true;
+                        return new TestingScoreRecordIterator();
+                    }
+
+                    @Override
+                    public void close() {}
+                };
+
+        ScoreRecordReader<String> transformed =
+                reader.filter(value -> value == 1).transform(String::valueOf);
+        ScoreRecordIterator<String> iterator = transformed.readBatch();
+
+        assertThat(iterator).isNotNull();
+        assertThat(iterator.next()).isEqualTo("1");
+        assertThat(iterator.returnedRowId()).isEqualTo(11L);
+        assertThat(iterator.returnedScore()).isEqualTo(0.75f);
+        assertThat(iterator.next()).isNull();
+        assertThat(transformed.readBatch()).isNull();
+    }
+
+    private static class TestingScoreRecordIterator implements ScoreRecordIterator<Integer> {
+
+        private final int[] values = {0, 1};
+        private final long[] rowIds = {10L, 11L};
+        private final float[] scores = {0.25f, 0.75f};
+        private int index = -1;
+
+        @Nullable
+        @Override
+        public Integer next() {
+            index++;
+            return index < values.length ? values[index] : null;
+        }
+
+        @Override
+        public float returnedScore() {
+            return scores[index];
+        }
+
+        @Override
+        public long returnedRowId() {
+            return rowIds[index];
+        }
+
+        @Override
+        public void releaseBatch() {}
+    }
+}

paimon-core/src/main/java/org/apache/paimon/crosspartition/IndexBootstrap.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.JoinedRow;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
@@ -41,12 +42,12 @@
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.paimon.CoreOptions.QUERY_AUTH_ENABLED;
 import static org.apache.paimon.CoreOptions.SCAN_MODE;
 import static org.apache.paimon.CoreOptions.StartupMode.LATEST;
 import static org.apache.paimon.io.SplitsParallelReadUtil.parallelExecute;
@@ -80,11 +81,12 @@ public RecordReader<InternalRow> bootstrap(int numAssigners, int assignId) throw
                         .mapToInt(Integer::intValue)
                         .toArray();
 
-        // force using the latest scan mode
+        // Force using the latest scan mode and bypass query auth for this internal index read.
+        Options bootstrapOptions = new Options();
+        bootstrapOptions.set(SCAN_MODE, LATEST);
+        bootstrapOptions.set(QUERY_AUTH_ENABLED, false);
         ReadBuilder readBuilder =
-                table.copy(Collections.singletonMap(SCAN_MODE.key(), LATEST.toString()))
-                        .newReadBuilder()
-                        .withProjection(keyProjection);
+                table.copy(bootstrapOptions.toMap()).newReadBuilder().withProjection(keyProjection);
 
         DataTableScan tableScan = (DataTableScan) readBuilder.newScan();
         List<Split> splits =
@@ -119,7 +121,7 @@ public RecordReader<InternalRow> bootstrap(int numAssigners, int assignId) throw
                 options.pageSize(),
                 options.crossPartitionUpsertBootstrapParallelism(),
                 split -> {
-                    DataSplit dataSplit = ((DataSplit) split);
+                    DataSplit dataSplit = (DataSplit) split;
                     int bucket = dataSplit.bucket();
                     return partBucketConverter.toGenericRow(
                             new JoinedRow(dataSplit.partition(), GenericRow.of(bucket)));

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java

@@ -37,6 +37,7 @@
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.QueryAuthSplit;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Filter;
@@ -332,12 +333,17 @@ private Optional<GlobalIndexResult> evalGlobalIndex() {
     public static Plan wrapToIndexSplits(
             List<Split> splits, RowRangeIndex rowRangeIndex, ScoreGetter scoreGetter) {
         List<Split> indexedSplits = new ArrayList<>();
-        Function<Split, List<IndexedSplit>> process =
-                split ->
-                        Collections.singletonList(
-                                split instanceof IndexedSplit
-                                        ? (IndexedSplit) split
-                                        : wrap((DataSplit) split, rowRangeIndex, scoreGetter));
+        Function<Split, List<Split>> process =
+                split -> {
+                    Split wrappedSplit = QueryAuthSplit.unwrap(split);
+                    if (wrappedSplit instanceof IndexedSplit) {
+                        return Collections.singletonList(split);
+                    }
+                    IndexedSplit indexedSplit =
+                            wrap((DataSplit) wrappedSplit, rowRangeIndex, scoreGetter);
+                    return Collections.singletonList(
+                            QueryAuthSplit.retainAuth(split, indexedSplit));
+                };
         randomlyExecuteSequentialReturn(process, splits, null).forEachRemaining(indexedSplits::add);
         return () -> indexedSplits;
     }

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

@@ -40,6 +40,8 @@
 import java.util.Map;
 import java.util.concurrent.ThreadPoolExecutor;
 
+import static org.apache.paimon.CoreOptions.QUERY_AUTH_ENABLED;
+
 /** List what data files recorded in manifests are missing from the filesystem. */
 public class ListUnexistingFiles {
 
@@ -48,11 +50,14 @@ public class ListUnexistingFiles {
     private final ThreadPoolExecutor executor;
 
     public ListUnexistingFiles(FileStoreTable table) {
-        this.table = table;
-        this.pathFactory = table.store().pathFactory();
+        this.table =
+                table.coreOptions().queryAuthEnabled()
+                        ? table.copy(Collections.singletonMap(QUERY_AUTH_ENABLED.key(), "false"))
+                        : table;
+        this.pathFactory = this.table.store().pathFactory();
         this.executor =
                 FileOperationThreadPool.getExecutorService(
-                        table.coreOptions().fileOperationThreadNum());
+                        this.table.coreOptions().fileOperationThreadNum());
     }
 
     public Map<Integer, Map<String, DataFileMeta>> list(BinaryRow partition) throws Exception {