[common] Reject out-of-range bloom filter items and fpp options

paimon-common/src/main/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndex.java

@@ -36,6 +36,7 @@
 
 import static org.apache.paimon.fileindex.FileIndexResult.REMAIN;
 import static org.apache.paimon.fileindex.FileIndexResult.SKIP;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /**
  * Bloom filter for file index.
@@ -86,6 +87,12 @@ private static class Writer extends FileIndexWriter {
         private final FastHash hashFunction;
 
         public Writer(DataType type, int items, double fpp) {
+            checkArgument(
+                    fpp > 0 && fpp < 1,
+                    "Bloom filter '" + FPP + "' must be in range (0, 1), but was %s.",
+                    fpp);
+            checkArgument(
+                    items > 0, "Bloom filter '" + ITEMS + "' must be positive, but was %s.", items);
             this.filter = new BloomFilter64(items, fpp);
             this.hashFunction = FastHash.getHashFunction(type);
         }

paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter64.java

@@ -26,7 +26,17 @@ public final class BloomFilter64 {
     private final int numHashFunctions;
 
     public BloomFilter64(long items, double fpp) {
-        int nb = (int) (-items * Math.log(fpp) / (Math.log(2) * Math.log(2)));
+        Preconditions.checkArgument(
+                items > 0, "Bloom filter items must be positive, but was %s.", items);
+        long numBitsEstimate = (long) (-items * Math.log(fpp) / (Math.log(2) * Math.log(2)));
+        Preconditions.checkArgument(
+                numBitsEstimate >= 0 && numBitsEstimate <= Integer.MAX_VALUE - Byte.SIZE,
+                "Bloom filter needs %s bits for items=%s and fpp=%s, which is out of the "
+                        + "supported range; reduce items or increase fpp.",
+                numBitsEstimate,
+                items,
+                fpp);
+        int nb = (int) numBitsEstimate;
         this.numBits = nb + (Byte.SIZE - (nb % Byte.SIZE));
         this.numHashFunctions =
                 Math.max(1, (int) Math.round((double) numBits / items * Math.log(2)));

paimon-common/src/test/java/org/apache/paimon/fileindex/bloomfilter/BloomFilterFileIndexTest.java

@@ -126,6 +126,47 @@ public void testAddFindByRandomLong() {
         Assertions.assertThat((double) errorCount / num).isLessThan(0.03);
     }
 
+    @Test
+    public void testRejectsInvalidOptions() {
+        // fpp must be a probability in (0, 1): a percentage-shaped value, zero, or >= 1 is rejected
+        // at write time with a message naming the option, instead of a bare
+        // NegativeArraySizeException
+        // or a silently useless one-byte filter.
+        Assertions.assertThatThrownBy(() -> createWriter("10000", "10"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("'fpp'");
+        Assertions.assertThatThrownBy(() -> createWriter("10000", "0"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("'fpp'");
+        Assertions.assertThatThrownBy(() -> createWriter("10000", "1.0"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("'fpp'");
+        Assertions.assertThatThrownBy(() -> createWriter("10000", "-0.1"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("'fpp'");
+
+        // items must be positive.
+        Assertions.assertThatThrownBy(() -> createWriter("0", "0.1"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("'items'");
+
+        // a huge items count would overflow the bit-set size; reject instead of allocating a
+        // negative-length array.
+        Assertions.assertThatThrownBy(() -> createWriter(String.valueOf(Integer.MAX_VALUE), "0.1"))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("bits");
+
+        // a valid configuration still builds.
+        createWriter("10000", "0.02");
+    }
+
+    private static FileIndexWriter createWriter(String items, String fpp) {
+        Options options = new Options();
+        options.set("items", items);
+        options.set("fpp", fpp);
+        return new BloomFilterFileIndex(DataTypes.BYTES(), options).createWriter();
+    }
+
     private byte[] random() {
         byte[] b = new byte[Math.abs(RANDOM.nextInt(400) + 1)];
         RANDOM.nextBytes(b);