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

paimon-common/src/main/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndex.java

@@ -95,12 +95,28 @@ public FileIndexReader createReader(SeekableInputStream inputStream, int start,
                             ? BitSliceIndexRoaringBitmap.map(input)
                             : BitSliceIndexRoaringBitmap.EMPTY;
 
-            return new Reader(dataType, rowNumber, positive, negative);
+            Reader reader = new Reader(dataType, rowNumber, positive, negative);
+            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 indexed value.
+     */
+    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 Function<Object, Long> valueMapper;
@@ -358,6 +374,32 @@ public FileIndexResult visitBetween(FieldRef fieldRef, Object from, Object to) {
         }
     }
 
+    /**
+     * Reader for a column whose values the mapper truncated, so comparing a literal against the
+     * indexed value cannot answer the predicate: {@code ts <> '...000000000'} would drop a row
+     * whose nanoseconds differ, and {@code ts = '...'} would select it. 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);
+        }
+    }
+
     public static Function<Object, Long> getValueMapper(DataType dataType) {
         return dataType.accept(
                 new DataTypeDefaultVisitor<Function<Object, Long>>() {

paimon-common/src/test/java/org/apache/paimon/fileindex/bsi/BitSliceIndexBitmapFileIndexTest.java

@@ -18,13 +18,16 @@
 
 package org.apache.paimon.fileindex.bsi;
 
+import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fileindex.FileIndexResult;
 import org.apache.paimon.fileindex.FileIndexWriter;
 import org.apache.paimon.fileindex.bitmap.BitmapIndexResult;
 import org.apache.paimon.fs.ByteArraySeekableStream;
 import org.apache.paimon.predicate.FieldRef;
 import org.apache.paimon.types.BigIntType;
 import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.TimestampType;
 import org.apache.paimon.utils.RoaringBitmap32;
 
 import org.junit.jupiter.api.Test;
@@ -331,4 +334,65 @@ public void testWriterCannotHandleLongMinValue() {
                 .hasCauseInstanceOf(IllegalArgumentException.class)
                 .hasRootCauseMessage("values should be non-negative");
     }
+
+    @Test
+    public void testSubMicrosecondTimestampIndexAnswersNoValuePredicate() {
+        // The value mapper stores micros, so these two rows share one indexed value.
+        Timestamp second = Timestamp.fromEpochMillis(1000, 0);
+        Timestamp secondAndHalfMicro = Timestamp.fromEpochMillis(1000, 500);
+
+        TimestampType nanos = new TimestampType(9);
+        FieldRef fieldRef = new FieldRef(0, "", nanos);
+        BitSliceIndexBitmapFileIndex bsiFileIndex = new BitSliceIndexBitmapFileIndex(nanos);
+        FileIndexWriter writer = bsiFileIndex.createWriter();
+        for (Object o : new Object[] {second, secondAndHalfMicro, null}) {
+            writer.write(o);
+        }
+        byte[] bytes = writer.serializedBytes();
+        FileIndexReader reader =
+                bsiFileIndex.createReader(new ByteArraySeekableStream(bytes), 0, bytes.length);
+
+        // Answering these from the index would drop row 1 from the <> result and select it for
+        // the =, since the bitmap is the 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);
+        assertThat(reader.visitLessThan(fieldRef, secondAndHalfMicro))
+                .isSameAs(FileIndexResult.REMAIN);
+        assertThat(reader.visitGreaterThan(fieldRef, second)).isSameAs(FileIndexResult.REMAIN);
+        assertThat(reader.visitBetween(fieldRef, second, secondAndHalfMicro))
+                .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() {
+        // 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);
+
+        TimestampType micros = new TimestampType(6);
+        FieldRef fieldRef = new FieldRef(0, "", micros);
+        BitSliceIndexBitmapFileIndex bsiFileIndex = new BitSliceIndexBitmapFileIndex(micros);
+        FileIndexWriter writer = bsiFileIndex.createWriter();
+        for (Object o : new Object[] {second, secondAndMicro}) {
+            writer.write(o);
+        }
+        byte[] bytes = writer.serializedBytes();
+        FileIndexReader reader =
+                bsiFileIndex.createReader(new ByteArraySeekableStream(bytes), 0, bytes.length);
+
+        assertThat(((BitmapIndexResult) reader.visitEqual(fieldRef, second)).get())
+                .isEqualTo(RoaringBitmap32.bitmapOf(0));
+        assertThat(((BitmapIndexResult) reader.visitNotEqual(fieldRef, second)).get())
+                .isEqualTo(RoaringBitmap32.bitmapOf(1));
+    }
 }