[core][python] Support null bitmap in Python BTree

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java

@@ -42,7 +42,6 @@
 public class LazyFilteredBTreeReader implements GlobalIndexReader {
 
     private final BTreeFileMetaSelector fileSelector;
-    private final List<GlobalIndexIOMeta> files;
     private final Map<Path, GlobalIndexReader> readerCache;
     private final KeySerializer keySerializer;
     private final CacheManager cacheManager;
@@ -58,7 +57,6 @@ public LazyFilteredBTreeReader(
         this.cacheManager = cacheManager;
         this.fileReader = fileReader;
         this.keySerializer = keySerializer;
-        this.files = files;
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/utils/MutableObjectIteratorAdapter.java

@@ -42,9 +42,9 @@ public class MutableObjectIteratorAdapter<I extends E, E> implements Iterator<E>
 
     private final MutableObjectIterator<I> delegate;
     private final I instance;
+
     private E nextElement;
-    private boolean hasNext = false;
-    private boolean initialized = false;
+    private boolean prefetched = false;
 
     /**
      * Creates a new adapter wrapping the given {@link MutableObjectIterator}.
@@ -58,33 +58,26 @@ public MutableObjectIteratorAdapter(MutableObjectIterator<I> delegate, I instanc
 
     @Override
     public boolean hasNext() {
-        if (!initialized) {
+        if (!prefetched) {
             prefetch();
         }
-        return hasNext;
+        return nextElement != null;
     }
 
     @Override
     public E next() {
         if (!hasNext()) {
             throw new NoSuchElementException();
         }
-        E result = nextElement;
-        prefetch();
-        return result;
+        prefetched = false;
+        return nextElement;
     }
 
-    /**
-     * Prefetches the next element from the delegate iterator.
-     *
-     * <p>This method reads ahead one element to support the {@link #hasNext()} check required by
-     * the standard {@link Iterator} interface.
-     */
+    /** Prefetches the next element from the delegate iterator. */
     private void prefetch() {
         try {
             nextElement = delegate.next(instance);
-            hasNext = (nextElement != null);
-            initialized = true;
+            prefetched = true;
         } catch (IOException e) {
             throw new RuntimeException("Failed to read next element from MutableObjectIterator", e);
         }

paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java

@@ -416,6 +416,7 @@ public void testBtreeIndexWrite() throws Exception {
         testBtreeIndexWriteInt();
         testBtreeIndexWriteBigInt();
         testBtreeIndexWriteLarge();
+        testBtreeIndexWriteNull();
     }
 
     private void testBtreeIndexWriteString() throws Exception {
@@ -436,7 +437,7 @@ private void testBtreeIndexWriteBigInt() throws Exception {
                 DataTypes.BIGINT(), "test_btree_index_bigint", 1000L, 2000L, 3000L);
     }
 
-    private <T> void testBtreeIndexWriteGeneric(
+    private void testBtreeIndexWriteGeneric(
             DataType keyType, String tableName, Object key1, Object key2, Object key3)
             throws Exception {
         // create table
@@ -569,6 +570,74 @@ private void testBtreeIndexWriteLarge() throws Exception {
         assertThat(result).containsOnly("v2");
     }
 
+    private void testBtreeIndexWriteNull() throws Exception {
+        // create table
+        RowType rowType =
+                RowType.of(
+                        new DataType[] {DataTypes.STRING(), DataTypes.STRING()},
+                        new String[] {"k", "v"});
+        Options options = new Options();
+        Path tablePath = new Path(warehouse.toString() + "/default.db/test_btree_index_null");
+        options.set(PATH, tablePath.toString());
+        options.set(ROW_TRACKING_ENABLED, true);
+        options.set(DATA_EVOLUTION_ENABLED, true);
+        options.set(GLOBAL_INDEX_ENABLED, true);
+        TableSchema tableSchema =
+                SchemaUtils.forceCommit(
+                        new SchemaManager(LocalFileIO.create(), tablePath),
+                        new Schema(
+                                rowType.getFields(),
+                                Collections.emptyList(),
+                                Collections.emptyList(),
+                                options.toMap(),
+                                ""));
+        AppendOnlyFileStoreTable table =
+                new AppendOnlyFileStoreTable(
+                        FileIOFinder.find(tablePath),
+                        tablePath,
+                        tableSchema,
+                        CatalogEnvironment.empty());
+
+        // write data
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            write.write(
+                    GenericRow.of(BinaryString.fromString("k1"), BinaryString.fromString("v1")));
+            write.write(
+                    GenericRow.of(BinaryString.fromString("k2"), BinaryString.fromString("v2")));
+            write.write(GenericRow.of(null, BinaryString.fromString("v3")));
+            write.write(
+                    GenericRow.of(BinaryString.fromString("k4"), BinaryString.fromString("v4")));
+            write.write(GenericRow.of(null, BinaryString.fromString("v5")));
+            commit.commit(write.prepareCommit());
+        }
+
+        // build index
+        BTreeGlobalIndexBuilder builder =
+                new BTreeGlobalIndexBuilder(table).withIndexType("btree").withIndexField("k");
+        try (BatchTableCommit commit = writeBuilder.newCommit()) {
+            commit.commit(builder.build(builder.scan(), IOManager.create(warehouse.toString())));
+        }
+
+        // assert index
+        List<IndexManifestEntry> indexEntries =
+                table.indexManifestFileReader().read(table.latestSnapshot().get().indexManifest);
+        assertThat(indexEntries)
+                .singleElement()
+                .matches(entry -> entry.indexFile().rowCount() == 5);
+
+        // read index is null
+        PredicateBuilder predicateBuilder = new PredicateBuilder(table.rowType());
+        ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicateBuilder.isNull(0));
+        List<String> result = new ArrayList<>();
+        readBuilder
+                .newRead()
+                .createReader(readBuilder.newScan().plan())
+                .forEachRemaining(r -> result.add(r.getString(1).toString()));
+        assertThat(result).containsExactlyInAnyOrder("v3", "v5");
+    }
+
     // Helper method from TableTestBase
     protected Identifier identifier(String tableName) {
         return new Identifier(database, tableName);

paimon-core/src/test/java/org/apache/paimon/utils/MutableObjectIteratorAdapterTest.java

@@ -0,0 +1,217 @@
+/*
+ * 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.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link MutableObjectIteratorAdapter}. */
+class MutableObjectIteratorAdapterTest {
+
+    @Test
+    public void testBasicIteration() {
+        List<Integer> values = Arrays.asList(1, 2, 3, 4, 5);
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        List<Integer> result = new ArrayList<>();
+        while (adapter.hasNext()) {
+            result.add(adapter.next());
+        }
+
+        assertThat(result).containsExactly(1, 2, 3, 4, 5);
+    }
+
+    @Test
+    public void testEmptyIterator() {
+        List<Integer> values = Collections.emptyList();
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        assertThat(adapter.hasNext()).isFalse();
+        assertThatThrownBy(adapter::next).isInstanceOf(NoSuchElementException.class);
+    }
+
+    @Test
+    public void testSingleElement() {
+        List<Integer> values = Collections.singletonList(42);
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.next()).isEqualTo(42);
+        assertThat(adapter.hasNext()).isFalse();
+    }
+
+    @Test
+    public void testMultipleHasNextCalls() {
+        List<Integer> values = Arrays.asList(1, 2);
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        // Call hasNext multiple times should not advance the iterator
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.hasNext()).isTrue();
+
+        assertThat(adapter.next()).isEqualTo(1);
+
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.hasNext()).isTrue();
+
+        assertThat(adapter.next()).isEqualTo(2);
+        assertThat(adapter.hasNext()).isFalse();
+    }
+
+    @Test
+    public void testNextWithoutHasNext() {
+        List<Integer> values = Arrays.asList(1, 2, 3);
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        // Call next() without calling hasNext() should still work
+        assertThat(adapter.next()).isEqualTo(1);
+        assertThat(adapter.next()).isEqualTo(2);
+        assertThat(adapter.next()).isEqualTo(3);
+
+        assertThatThrownBy(adapter::next).isInstanceOf(NoSuchElementException.class);
+    }
+
+    @Test
+    public void testIOExceptionHandling() {
+        MutableObjectIterator<Integer> failingDelegate =
+                new MutableObjectIterator<Integer>() {
+                    private int count = 0;
+
+                    @Override
+                    public Integer next(Integer reuse) throws IOException {
+                        count++;
+                        if (count == 1) {
+                            return 1;
+                        }
+                        throw new IOException("Test exception");
+                    }
+
+                    @Override
+                    public Integer next() throws IOException {
+                        return next(null);
+                    }
+                };
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(failingDelegate, 0);
+
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.next()).isEqualTo(1);
+
+        assertThatThrownBy(adapter::hasNext)
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("Failed to read next element from MutableObjectIterator")
+                .hasCauseExactlyInstanceOf(IOException.class);
+    }
+
+    @Test
+    public void testNullElements() {
+        List<Integer> values = Arrays.asList(1, null, 3);
+        MutableObjectIterator<Integer> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Integer, Integer> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0);
+
+        assertThat(adapter.hasNext()).isTrue();
+        assertThat(adapter.next()).isEqualTo(1);
+
+        // Null element should be treated as end of iteration
+        assertThat(adapter.hasNext()).isFalse();
+    }
+
+    @Test
+    public void testStringType() {
+        List<String> values = Arrays.asList("hello", "world", "test");
+        MutableObjectIterator<String> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<String, String> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, "");
+
+        List<String> result = new ArrayList<>();
+        while (adapter.hasNext()) {
+            result.add(adapter.next());
+        }
+
+        assertThat(result).containsExactly("hello", "world", "test");
+    }
+
+    @Test
+    public void testLongType() {
+        List<Long> values = Arrays.asList(100L, 200L, 300L);
+        MutableObjectIterator<Long> delegate = createTestIterator(values);
+
+        MutableObjectIteratorAdapter<Long, Long> adapter =
+                new MutableObjectIteratorAdapter<>(delegate, 0L);
+
+        List<Long> result = new ArrayList<>();
+        while (adapter.hasNext()) {
+            result.add(adapter.next());
+        }
+
+        assertThat(result).containsExactly(100L, 200L, 300L);
+    }
+
+    /**
+     * Creates a test implementation of MutableObjectIterator that returns elements from the given
+     * list.
+     */
+    private <T> MutableObjectIterator<T> createTestIterator(List<T> values) {
+        return new MutableObjectIterator<T>() {
+            private int index = 0;
+
+            @Override
+            public T next(T reuse) {
+                if (index >= values.size()) {
+                    return null;
+                }
+                return values.get(index++);
+            }
+
+            @Override
+            public T next() {
+                return next(null);
+            }
+        };
+    }
+}