[common] Stop answering value predicates from a truncated bitmap index

paimon-common/src/main/java/org/apache/paimon/fileindex/bitmap/BitmapFileIndex.java

@@ -71,12 +71,28 @@ public FileIndexWriter createWriter() {
     public FileIndexReader createReader(
             SeekableInputStream seekableInputStream, int start, int length) {
         try {
-            return new Reader(seekableInputStream, start, options);
+            Reader reader = new Reader(seekableInputStream, start, options);
+            return valuesAreTruncated(dataType) ? new TruncatedValueReader(reader) : reader;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
 
+    /**
+     * Whether the value mapper loses information for this type. TIMESTAMP above microsecond
+     * precision is mapped with {@link Timestamp#toMicros()}, so two values that differ only below a
+     * microsecond share one bitmap key.
+     */
+    private static boolean valuesAreTruncated(DataType dataType) {
+        if (dataType instanceof TimestampType) {
+            return ((TimestampType) dataType).getPrecision() > 6;
+        }
+        if (dataType instanceof LocalZonedTimestampType) {
+            return ((LocalZonedTimestampType) dataType).getPrecision() > 6;
+        }
+        return false;
+    }
+
     private static class Writer extends FileIndexWriter {
 
         private final int version;
@@ -312,6 +328,33 @@ private void readInternalMeta(DataType dataType) {
         }
     }
 
+    /**
+     * Reader for a column whose values the mapper truncated, so looking a literal up by its bitmap
+     * key cannot answer the predicate: {@code visitNotIn} flips the matched rows over the whole row
+     * count, so {@code ts <> '...000000000'} would drop every row in the same microsecond, and
+     * {@code ts = '...'} would select them. Inheriting {@link FileIndexReader}'s {@code REMAIN} for
+     * those leaves the rows to be read and filtered. Null-ness survives truncation, so those two
+     * questions still come from the index.
+     */
+    private static class TruncatedValueReader extends FileIndexReader {
+
+        private final Reader reader;
+
+        public TruncatedValueReader(Reader reader) {
+            this.reader = reader;
+        }
+
+        @Override
+        public FileIndexResult visitIsNull(FieldRef fieldRef) {
+            return reader.visitIsNull(fieldRef);
+        }
+
+        @Override
+        public FileIndexResult visitIsNotNull(FieldRef fieldRef) {
+            return reader.visitIsNotNull(fieldRef);
+        }
+    }
+
     // Currently, it is mainly used to convert timestamps to long
     public static Function<Object, Object> getValueMapper(DataType dataType) {
         return dataType.accept(

paimon-common/src/test/java/org/apache/paimon/fileindex/bitmapindex/BitmapFileIndexTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.fileindex.bitmapindex;
 
 import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.fileindex.FileIndexReader;
 import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fileindex.FileIndexWriter;
@@ -32,6 +33,8 @@
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.LocalZonedTimestampType;
+import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.types.VarCharType;
 import org.apache.paimon.utils.RoaringBitmap32;
 import org.apache.paimon.utils.StringUtils;
@@ -45,6 +48,8 @@
 import java.util.Arrays;
 import java.util.function.Consumer;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /** test for {@link BitmapFileIndex}. */
 public class BitmapFileIndexTest {
 
@@ -184,6 +189,70 @@ public void testV2EntryLargerThanBlockSize() throws Exception {
                 .equals(RoaringBitmap32.bitmapOf(1));
     }
 
+    @Test
+    public void testSubMicrosecondTimestampIndexAnswersNoValuePredicate() throws Exception {
+        // The value mapper stores micros, so these two rows share one bitmap key.
+        Timestamp second = Timestamp.fromEpochMillis(1000, 0);
+        Timestamp secondAndHalfMicro = Timestamp.fromEpochMillis(1000, 500);
+        Object[] dataColumn = {second, secondAndHalfMicro, null};
+
+        for (DataType nanos :
+                new DataType[] {new TimestampType(9), new LocalZonedTimestampType(9)}) {
+            FieldRef fieldRef = new FieldRef(0, "", nanos);
+            FileIndexReader reader =
+                    createTestReaderOnWriter(
+                            BitmapFileIndex.VERSION_2,
+                            null,
+                            nanos,
+                            writer -> {
+                                for (Object o : dataColumn) {
+                                    writer.write(o);
+                                }
+                            });
+
+            // Answering these from the index would select row 1 for the = and drop it from the
+            // <>, since the bitmap is the exact row set the scan reads.
+            assertThat(reader.visitEqual(fieldRef, second)).isSameAs(FileIndexResult.REMAIN);
+            assertThat(reader.visitNotEqual(fieldRef, second)).isSameAs(FileIndexResult.REMAIN);
+            assertThat(reader.visitIn(fieldRef, Arrays.asList(second, secondAndHalfMicro)))
+                    .isSameAs(FileIndexResult.REMAIN);
+            assertThat(reader.visitNotIn(fieldRef, Arrays.asList(second)))
+                    .isSameAs(FileIndexResult.REMAIN);
+
+            // Null-ness does not depend on the truncated digits, so it still prunes.
+            assertThat(((BitmapIndexResult) reader.visitIsNull(fieldRef)).get())
+                    .isEqualTo(RoaringBitmap32.bitmapOf(2));
+            assertThat(((BitmapIndexResult) reader.visitIsNotNull(fieldRef)).get())
+                    .isEqualTo(RoaringBitmap32.bitmapOf(0, 1));
+        }
+    }
+
+    @Test
+    public void testMicrosecondTimestampIndexStillAnswersValuePredicates() throws Exception {
+        // Precision 6 is exactly what the mapper stores, so nothing is given up there.
+        Timestamp second = Timestamp.fromEpochMillis(1000, 0);
+        Timestamp secondAndMicro = Timestamp.fromEpochMillis(1000, 1000);
+        Object[] dataColumn = {second, secondAndMicro};
+
+        TimestampType micros = new TimestampType(6);
+        FieldRef fieldRef = new FieldRef(0, "", micros);
+        FileIndexReader reader =
+                createTestReaderOnWriter(
+                        BitmapFileIndex.VERSION_2,
+                        null,
+                        micros,
+                        writer -> {
+                            for (Object o : dataColumn) {
+                                writer.write(o);
+                            }
+                        });
+
+        assertThat(((BitmapIndexResult) reader.visitEqual(fieldRef, second)).get())
+                .isEqualTo(RoaringBitmap32.bitmapOf(0));
+        assertThat(((BitmapIndexResult) reader.visitNotEqual(fieldRef, second)).get())
+                .isEqualTo(RoaringBitmap32.bitmapOf(1));
+    }
+
     private void testStringType(int version) throws Exception {
         FieldRef fieldRef = new FieldRef(0, "", DataTypes.STRING());
         BinaryString a = BinaryString.fromString("a");

paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionFileIndexTest.java

@@ -633,6 +633,70 @@ private static Map<String, String> bsiOptions(String column) {
         return options;
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"parquet", "orc"})
+    public void testSubMicrosecondTimestampBitmapMatchesUnindexed(String format) throws Exception {
+        // A bitmap index maps TIMESTAMP through toMicros(), so on a TIMESTAMP(9) column two values
+        // in the same microsecond share one bitmap key. visitNotIn flips the matched rows over the
+        // whole row count, so answering <> or NOT IN from that bitmap drops rows the residual
+        // filter can no longer recover. The fix makes the indexed read (executeFilter) return
+        // exactly what an unindexed full scan does.
+        FileStoreTable table = createTimestampBitmapTable("bitmap_ts9_" + format, format);
+
+        long base = 1_704_067_200_000L;
+        Timestamp tsA = Timestamp.fromEpochMillis(base, 123_000); // micro bucket base*1000+123
+        Timestamp tsB = Timestamp.fromEpochMillis(base, 123_400); // same bucket as tsA
+        Timestamp tsC = Timestamp.fromEpochMillis(base, 999_000); // a different bucket
+        write(
+                table,
+                GenericRow.of(0, tsA),
+                GenericRow.of(1, tsB),
+                GenericRow.of(2, tsC),
+                GenericRow.of(3, null));
+
+        PredicateBuilder b = new PredicateBuilder(table.rowType());
+
+        // Guard: the sub-microsecond nanos must survive the write/read round trip on this format,
+        // otherwise tsA and tsB collapse and the comparison below would pass vacuously.
+        List<Timestamp> stored = new ArrayList<>();
+        for (InternalRow row : fullScanFiltered(table, b.isNotNull(1))) {
+            stored.add(row.getTimestamp(1, 9));
+        }
+        assertThat(stored).contains(tsA, tsB);
+
+        // notEqual / NOT IN are the ones that lose rows to the micro-bucket collision;
+        // isNull/isNotNull still come from the index; equal / IN only over-select and are
+        // already corrected by the residual filter.
+        List<Predicate> predicates =
+                Arrays.asList(
+                        b.notEqual(1, tsA),
+                        b.notIn(1, Arrays.asList(tsA, tsC)),
+                        b.isNull(1),
+                        b.isNotNull(1),
+                        b.equal(1, tsA),
+                        b.in(1, Arrays.asList(tsA, tsC)),
+                        PredicateBuilder.and(b.notEqual(1, tsA), b.notEqual(1, tsC)),
+                        PredicateBuilder.or(b.equal(1, tsA), b.equal(1, tsC)));
+        for (Predicate p : predicates) {
+            assertThat(query(table, p))
+                    .containsExactlyInAnyOrderElementsOf(fullScanFiltered(table, p));
+        }
+    }
+
+    private FileStoreTable createTimestampBitmapTable(String name, String format) throws Exception {
+        Schema.Builder builder =
+                Schema.newBuilder()
+                        .column("f0", DataTypes.INT())
+                        .column("f1", DataTypes.TIMESTAMP(9))
+                        .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+                        .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+                        .option(CoreOptions.FILE_FORMAT.key(), format);
+        bitmapOptions("f1").forEach(builder::option);
+        Identifier identifier = identifier(name);
+        catalog.createTable(identifier, builder.build(), false);
+        return getTable(identifier);
+    }
+
     private void writeAllColumns(FileStoreTable table, int count) throws Exception {
         BatchWriteBuilder builder = table.newBatchWriteBuilder();
         try (BatchTableWrite write = builder.newWrite();