Rebuild bloom filter index when fpp config changes

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/bloom/OnHeapGuavaBloomFilterCreator.java

@@ -44,11 +44,24 @@ public class OnHeapGuavaBloomFilterCreator implements BloomFilterCreator {
   private static final Logger LOGGER = LoggerFactory.getLogger(OnHeapGuavaBloomFilterCreator.class);
 
   public static final int TYPE_VALUE = 1;
+  /** V1 format (current write format): {@code [TYPE_VALUE=1 (int)][VERSION=1 (int)][Guava bytes...]}. */
   public static final int VERSION = 1;
+  /**
+   * V2 format: {@code [TYPE_VALUE=1 (int)][VERSION_V2=2 (int)][effective FPP (double)][Guava bytes...]}.
+   * This format was introduced to embed the effective FPP in the header for change detection, but it broke
+   * backward compatibility because release-1.5.0 and older servers throw
+   * {@code IllegalStateException: Unsupported bloom filter type value: 1 and version: 2} on load.
+   * New segments are written in V1 format; this constant is kept so the reader can still handle any
+   * V2 segments that were created during the short window when V2 was the write format.
+   */
+  public static final int VERSION_V2 = 2;
+  /** Byte offset of the fpp field in a v2 bloom filter file (after TYPE_VALUE + VERSION_V2). */
+  public static final int FPP_OFFSET = 8;
 
   private final File _bloomFilterFile;
   private final BloomFilter<String> _bloomFilter;
   private final FieldSpec.DataType _dataType;
+  private final double _effectiveFpp;
 
   // TODO: This method is here for compatibility reasons, should be removed in future PRs
   //  exit_criteria: Not needed in Apache Pinot once #10184 is merged
@@ -69,6 +82,7 @@ public OnHeapGuavaBloomFilterCreator(File indexDir, String columnName, int cardi
       double minFpp = GuavaBloomFilterReaderUtils.computeFPP(maxSizeInBytes, cardinality);
       fpp = Math.max(fpp, minFpp);
     }
+    _effectiveFpp = fpp;
     LOGGER.info("Creating bloom filter with cardinality: {}, fpp: {}", cardinality, fpp);
     _bloomFilter = BloomFilter.create(Funnels.stringFunnel(StandardCharsets.UTF_8), cardinality, fpp);
   }

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/bloomfilter/BloomFilterHandler.java

@@ -24,9 +24,11 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.commons.io.FileUtils;
+import org.apache.pinot.segment.local.segment.creator.impl.bloom.OnHeapGuavaBloomFilterCreator;
 import org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
 import org.apache.pinot.segment.local.segment.index.loader.BaseIndexHandler;
 import org.apache.pinot.segment.local.segment.index.loader.LoaderUtils;
+import org.apache.pinot.segment.local.segment.index.readers.bloom.GuavaBloomFilterReaderUtils;
 import org.apache.pinot.segment.spi.ColumnMetadata;
 import org.apache.pinot.segment.spi.V1Constants;
 import org.apache.pinot.segment.spi.creator.IndexCreationContext;
@@ -54,6 +56,14 @@
 public class BloomFilterHandler extends BaseIndexHandler {
   private static final Logger LOGGER = LoggerFactory.getLogger(BloomFilterHandler.class);
 
+  private static final int TYPE_VALUE_OFFSET = 0;
+  private static final int VERSION_OFFSET = 4;
+  // In V1 format the Guava payload starts at byte 8. Within that payload the layout is:
+  //   [strategy (1 byte)][numHashFunctions (1 byte)][numLongs (int, 4 bytes)][bit array...]
+  // These absolute file offsets match BaseGuavaBloomFilterReader's relative offsets + 8.
+  private static final int V1_NUM_HASH_FUNCTIONS_OFFSET = 9;  // 4 (TYPE) + 4 (VERSION) + 1 (strategy); reads 1 byte
+  private static final int V1_NUM_LONGS_OFFSET = 10;          // + 1 (numHashFunctions); reads 4-byte int
+
   private final Map<String, BloomFilterConfig> _bloomFilterConfigs;
 
   public BloomFilterHandler(SegmentDirectory segmentDirectory, Map<String, FieldIndexConfigs> fieldIndexConfigs,
@@ -72,6 +82,11 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
       if (!columnsToAddBF.remove(column)) {
         LOGGER.info("Need to remove existing bloom filter from segment: {}, column: {}", segmentName, column);
         return true;
+      } else {
+        // Bloom filter exists for this column; check if fpp config changed by comparing numHashFunctions
+        if (isFppChanged(segmentReader, segmentName, column)) {
+          return true;
+        }
       }
     }
     // Check if any new bloomfilter need to be added.
@@ -85,6 +100,120 @@ public boolean needUpdateIndices(SegmentDirectory.Reader segmentReader) {
     return false;
   }
 
+  /**
+   * Checks whether the effective fpp config has changed for an existing bloom filter index.
+   *
+   * <p>V2 segments (VERSION_V2 = 2) store the effective fpp explicitly in the header; the comparison is exact.
+   * V2 segments were created during a short window before the format was rolled back for backward compatibility;
+   * V2 reading is retained so those segments are not permanently stuck.
+   *
+   * <p>V1 segments (VERSION = 1, the current write format) do not store fpp in the header. FPP change is
+   * detected structurally: the stored {@code numHashFunctions} and {@code numLongs} in the Guava payload
+   * are compared to the values that would be produced by the configured fpp + cardinality. These two fields
+   * jointly encode the filter capacity and are stable across Guava versions.
+   *
+   * <p>Accepts both Reader and Writer since Writer extends Reader, allowing this method to be called from
+   * both needUpdateIndices and updateIndices.
+   */
+  private boolean isFppChanged(SegmentDirectory.Reader segmentReader, String segmentName, String column) {
+    try {
+      PinotDataBuffer dataBuffer = segmentReader.getIndexFor(column, StandardIndexes.bloomFilter());
+      int version = dataBuffer.getInt(VERSION_OFFSET);
+      BloomFilterConfig config = _bloomFilterConfigs.get(column);
+      ColumnMetadata columnMetadata = _segmentDirectory.getSegmentMetadata().getColumnMetadataFor(column);
+
+      if (version == OnHeapGuavaBloomFilterCreator.VERSION_V2) {
+        // V2 header stores the effective fpp explicitly.
+        double storedFpp = dataBuffer.getDouble(OnHeapGuavaBloomFilterCreator.FPP_OFFSET);
+        double expectedFpp = computeEffectiveFpp(columnMetadata, config);
+        if (Double.compare(storedFpp, expectedFpp) != 0) {
+          LOGGER.info("Bloom filter fpp config changed for segment: {}, column: {}, stored fpp: {}, "
+                  + "expected fpp: {}. Index needs to be rebuilt.",
+              segmentName, column, storedFpp, expectedFpp);
+          return true;
+        }
+        return false;
+      }
+
+      if (version != OnHeapGuavaBloomFilterCreator.VERSION) {
+        // Unrecognized version — force rebuild rather than silently reading bytes at wrong offsets.
+        LOGGER.warn("Unrecognized bloom filter version {} for segment: {}, column: {}; forcing rebuild.",
+            version, segmentName, column);
+        return true;
+      }
+      // V1 format: detect fpp change via numHashFunctions + numLongs in the Guava payload.
+      // These values are at fixed offsets (see V1_NUM_HASH_FUNCTIONS_OFFSET / V1_NUM_LONGS_OFFSET).
+      int storedNumHashFunctions = dataBuffer.getByte(V1_NUM_HASH_FUNCTIONS_OFFSET) & 0xFF;
+      int storedNumLongs = dataBuffer.getInt(V1_NUM_LONGS_OFFSET);
+      int[] expected = computeExpectedNumHashFunctionsAndNumLongs(columnMetadata, config);
+      int expectedNumHashFunctions = expected[0];
+      int expectedNumLongs = expected[1];
+      if (storedNumHashFunctions != expectedNumHashFunctions || storedNumLongs != expectedNumLongs) {
+        LOGGER.info("Bloom filter fpp config changed for segment: {}, column: {}, "
+                + "stored numHashFunctions: {}, expected: {}, stored numLongs: {}, expected: {}. "
+                + "Index needs to be rebuilt.",
+            segmentName, column, storedNumHashFunctions, expectedNumHashFunctions, storedNumLongs, expectedNumLongs);
+        return true;
+      }
+      return false;
+    } catch (Exception e) {
+      LOGGER.warn("Failed to read existing bloom filter for segment: {}, column: {}", segmentName, column, e);
+      return false;
+    }
+  }
+
+  /**
+   * Computes the {@code numHashFunctions} and {@code numLongs} that Guava would allocate for the given
+   * fpp and cardinality, matching the formulas in Guava's {@code BloomFilter}.
+   *
+   * <p>{@code numLongs} uses {@code optimalNumOfBits(n, p) = (long)(-n * ln(p) / (ln 2)^2)}.
+   * {@code numHashFunctions} uses the direct formula {@code max(1, round(-ln(p) / ln(2)))},
+   * which is what current Guava (33+) writes. The older two-step path via {@code idealNumBits / n * ln(2)}
+   * diverges for small cardinalities due to long-cast truncation and must not be used.
+   *
+   * @return {@code int[]{numHashFunctions, numLongs}}
+   */
+  private int[] computeExpectedNumHashFunctionsAndNumLongs(ColumnMetadata columnMetadata,
+      BloomFilterConfig bloomFilterConfig) {
+    double fpp = computeEffectiveFpp(columnMetadata, bloomFilterConfig);
+    int cardinality = columnMetadata != null ? columnMetadata.getCardinality() : 1;
+    if (cardinality <= 0) {
+      cardinality = columnMetadata != null ? columnMetadata.getTotalNumberOfEntries() : 1;
+    }
+    if (cardinality <= 0) {
+      cardinality = 1;
+    }
+    // Guava: optimalNumOfBits = -n * ln(p) / (ln 2)^2, minimum 1
+    long idealNumBits = Math.max(1, (long) (-cardinality * Math.log(fpp) / (Math.log(2) * Math.log(2))));
+    // Guava: numLongs = ceil(idealNumBits / 64).
+    int numLongs = (int) ((idealNumBits + 63) / 64);
+    // Guava (33+): optimalNumOfHashFunctions(p) = max(1, round(-ln(p) / ln(2)))
+    // This is the direct formula Guava now uses; it avoids the long-cast quantization error
+    // in the old two-step path (via idealNumBits / n) that diverges for small cardinalities.
+    int numHashFunctions = Math.max(1, (int) Math.round(-Math.log(fpp) / Math.log(2)));
+    return new int[]{numHashFunctions, numLongs};
+  }
+
+  /**
+   * Computes the effective fpp for a new bloom filter: applies the {@code maxSizeInBytes} cap if set.
+   * This matches the fpp computation in {@link OnHeapGuavaBloomFilterCreator}.
+   */
+  private double computeEffectiveFpp(ColumnMetadata columnMetadata, BloomFilterConfig bloomFilterConfig) {
+    double fpp = bloomFilterConfig.getFpp();
+    int maxSizeInBytes = bloomFilterConfig.getMaxSizeInBytes();
+    if (maxSizeInBytes > 0 && columnMetadata != null) {
+      int cardinality = columnMetadata.getCardinality();
+      if (cardinality <= 0) {
+        cardinality = columnMetadata.getTotalNumberOfEntries();
+      }
+      if (cardinality > 0) {
+        double minFpp = GuavaBloomFilterReaderUtils.computeFPP(maxSizeInBytes, cardinality);
+        fpp = Math.max(fpp, minFpp);
+      }
+    }
+    return fpp;
+  }
+
   @Override
   public void updateIndices(SegmentDirectory.Writer segmentWriter)
       throws Exception {
@@ -97,6 +226,15 @@ public void updateIndices(SegmentDirectory.Writer segmentWriter)
         LOGGER.info("Removing existing bloom filter from segment: {}, column: {}", segmentName, column);
         segmentWriter.removeIndex(column, StandardIndexes.bloomFilter());
         LOGGER.info("Removed existing bloom filter from segment: {}, column: {}", segmentName, column);
+      } else {
+        // Bloom filter exists for this column; check if fpp config changed
+        if (isFppChanged(segmentWriter, segmentName, column)) {
+          LOGGER.info("Deleting existing bloom filter for segment: {}, column: {} before rebuilding.", segmentName,
+              column);
+          segmentWriter.removeIndex(column, StandardIndexes.bloomFilter());
+          LOGGER.info("Removed existing bloom filter from segment: {}, column: {}", segmentName, column);
+          columnsToAddBF.add(column);
+        }
       }
     }
     for (String column : columnsToAddBF) {

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/bloom/BloomFilterReaderFactory.java

@@ -30,15 +30,29 @@ private BloomFilterReaderFactory() {
 
   private static final int TYPE_VALUE_OFFSET = 0;
   private static final int VERSION_OFFSET = 4;
-  private static final int HEADER_SIZE = 8;
+  /** Byte offset at which Guava bloom filter bytes start in the legacy v1 format. */
+  private static final int GUAVA_PAYLOAD_OFFSET_V1 = 8;
+  /**
+   * Byte offset at which Guava bloom filter bytes start in the v2 format.
+   * V2 header: [TYPE_VALUE (int, 4)][VERSION_V2 (int, 4)][FPP (double, 8)] = 16 bytes.
+   */
+  private static final int GUAVA_PAYLOAD_OFFSET_V2 = 16;
 
   public static BloomFilterReader getBloomFilterReader(PinotDataBuffer dataBuffer, boolean onHeap) {
     int typeValue = dataBuffer.getInt(TYPE_VALUE_OFFSET);
     int version = dataBuffer.getInt(VERSION_OFFSET);
+    if (typeValue == OnHeapGuavaBloomFilterCreator.TYPE_VALUE
+        && version == OnHeapGuavaBloomFilterCreator.VERSION_V2) {
+      // V2 format: [TYPE_VALUE][VERSION_V2][FPP (double)][Guava bytes]
+      PinotDataBuffer bloomFilterDataBuffer = dataBuffer.view(GUAVA_PAYLOAD_OFFSET_V2, dataBuffer.size());
+      return onHeap ? new OnHeapGuavaBloomFilterReader(bloomFilterDataBuffer)
+          : new OffHeapGuavaBloomFilterReader(bloomFilterDataBuffer);
+    }
+    // Legacy v1 format: [TYPE_VALUE=1][VERSION][Guava bytes]
     Preconditions.checkState(
         typeValue == OnHeapGuavaBloomFilterCreator.TYPE_VALUE && version == OnHeapGuavaBloomFilterCreator.VERSION,
         "Unsupported bloom filter type value: %s and version: %s", typeValue, version);
-    PinotDataBuffer bloomFilterDataBuffer = dataBuffer.view(HEADER_SIZE, dataBuffer.size());
+    PinotDataBuffer bloomFilterDataBuffer = dataBuffer.view(GUAVA_PAYLOAD_OFFSET_V1, dataBuffer.size());
     return onHeap ? new OnHeapGuavaBloomFilterReader(bloomFilterDataBuffer)
         : new OffHeapGuavaBloomFilterReader(bloomFilterDataBuffer);
   }

pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessorTest.java

@@ -1747,6 +1747,98 @@ public void testH3IndexResolutionUpdate(SegmentVersion segmentVersion)
     runPreProcessor(_newColumnsSchemaWithH3Json);
   }
 
+  @Test(dataProvider = "bothV1AndV3")
+  public void testBloomFilterFppUpdate(SegmentVersion segmentVersion)
+      throws Exception {
+    buildSegment(segmentVersion);
+
+    // Create bloom filter with fpp 0.1
+    _bloomFilterConfigs = Map.of("column3", new BloomFilterConfig(0.1, 0, false));
+    runPreProcessor();
+
+    // Verify no processing needed with same config
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            createIndexLoadingConfig(_schema))) {
+      assertFalse(processor.needProcess());
+    }
+
+    // Update bloom filter fpp to 0.01
+    _bloomFilterConfigs = Map.of("column3", new BloomFilterConfig(0.01, 0, false));
+
+    // Verify that preprocessing is needed due to fpp change
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            createIndexLoadingConfig(_schema))) {
+      assertTrue(processor.needProcess());
+      processor.process(SEGMENT_OPERATIONS_THROTTLER);
+    }
+
+    // Verify no more processing is needed after rebuild
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            createIndexLoadingConfig(_schema))) {
+      assertFalse(processor.needProcess());
+    }
+
+    // Remove bloom filter config
+    resetIndexConfigs();
+    runPreProcessor();
+  }
+
+  /**
+   * Tests that the formula for computing the expected number of hash functions is correct for very small cardinalities.
+   * With cardinality=1 and fpp=0.01, the old two-step formula (via optimalNumOfBits with a long cast) computes k=6
+   * but Guava actually writes k=7, causing an infinite rebuild loop on every segment reload. The fixed direct formula
+   * gives k=7, matching Guava exactly, so the idempotency check below would fail without the fix.
+   */
+  @Test(dataProvider = "bothV1AndV3")
+  public void testBloomFilterFppUpdateSmallCardinality(SegmentVersion segmentVersion)
+      throws Exception {
+    // Build a segment with a cardinality-1 STRING column (single repeated value → exactly 1 unique entry in dict)
+    String[] singleValueArray = {"only_value", "only_value", "only_value", "only_value", "only_value"};
+    long[] longValues = {1L, 2L, 3L, 4L, 5L};
+    TableConfig smallCardTableConfig =
+        new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
+    Schema smallCardSchema =
+        new Schema.SchemaBuilder().addSingleValueDimension("stringCol", FieldSpec.DataType.STRING)
+            .addMetric("longCol", FieldSpec.DataType.LONG)
+            .build();
+    buildTestSegment(smallCardTableConfig, smallCardSchema, singleValueArray, longValues);
+
+    // Create bloom filter with fpp 0.1
+    IndexingConfig indexingConfig = smallCardTableConfig.getIndexingConfig();
+    indexingConfig.setBloomFilterConfigs(Map.of("stringCol", new BloomFilterConfig(0.1, 0, false)));
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            new IndexLoadingConfig(smallCardTableConfig, smallCardSchema))) {
+      processor.process(SEGMENT_OPERATIONS_THROTTLER);
+    }
+
+    // Verify no processing needed with same fpp
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            new IndexLoadingConfig(smallCardTableConfig, smallCardSchema))) {
+      assertFalse(processor.needProcess());
+    }
+
+    // Update fpp to 0.01 — should trigger rebuild (k changes: 0.1→k=3, 0.01→k=7)
+    indexingConfig.setBloomFilterConfigs(Map.of("stringCol", new BloomFilterConfig(0.01, 0, false)));
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            new IndexLoadingConfig(smallCardTableConfig, smallCardSchema))) {
+      assertTrue(processor.needProcess());
+      processor.process(SEGMENT_OPERATIONS_THROTTLER);
+    }
+
+    // Verify no more processing needed after rebuild — proves no infinite rebuild loop (the formula bug)
+    try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);
+        SegmentPreProcessor processor = new SegmentPreProcessor(segmentDirectory,
+            new IndexLoadingConfig(smallCardTableConfig, smallCardSchema))) {
+      assertFalse(processor.needProcess());
+    }
+  }
+
   private void verifyProcessNeeded()
       throws Exception {
     try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(INDEX_DIR, ReadMode.mmap);