diff --git a/docs/content/concepts/spec/fileindex.md b/docs/content/concepts/spec/fileindex.md
index f8cdd97505df..06a21a114ddd 100644
--- a/docs/content/concepts/spec/fileindex.md
+++ b/docs/content/concepts/spec/fileindex.md
@@ -3,7 +3,7 @@ title: "File Index"
 weight: 7
 type: docs
 aliases:
-- /concepts/spec/fileindex.html
+  - /concepts/spec/fileindex.html
 ---
 <!--
 Licensed to the Apache Software Foundation (ASF) under one
@@ -83,15 +83,15 @@ redundant bytes:                  var bytes (for compatibility with later versio
 BODY:                             column index bytes + column index bytes + column index bytes + .......
 </pre>
 
-## Index: BloomFilter 
+## Index: BloomFilter
 
 Define `'file-index.bloom-filter.columns'`.
 
-Content of bloom filter index is simple: 
+Content of bloom filter index is simple:
 - numHashFunctions 4 bytes int, BIG_ENDIAN
 - bloom filter bytes
 
-This class use (64-bits) long hash. Store the num hash function (one integer) and bit set bytes only. Hash bytes type 
+This class use (64-bits) long hash. Store the num hash function (one integer) and bit set bytes only. Hash bytes type
 (like varchar, binary, etc.) using xx hash, hash numeric type by [specified number hash](http://web.archive.org/web/20071223173210/http://www.concentric.net/~Ttwang/tech/inthash.htm).
 
 ## Index: Bitmap
@@ -261,4 +261,4 @@ BSI only support the following data type:
       <td>false</td>
     </tr>
     </tbody>
-</table>
+</table>
\ No newline at end of file
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndex.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndex.java
new file mode 100644
index 000000000000..b03ca5158bce
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndex.java
@@ -0,0 +1,168 @@
+/*
+ * 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.fileindex.dynamicbloomfilter;
+
+import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fileindex.FileIndexResult;
+import org.apache.paimon.fileindex.FileIndexWriter;
+import org.apache.paimon.fileindex.FileIndexer;
+import org.apache.paimon.fileindex.bloomfilter.FastHash;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.BloomFilter64;
+import org.apache.paimon.utils.DynamicBloomFilter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import static org.apache.paimon.fileindex.FileIndexResult.REMAIN;
+import static org.apache.paimon.fileindex.FileIndexResult.SKIP;
+
+/** implementation of dynamic bloom filter file index. */
+public class DynamicBloomFilterFileIndex implements FileIndexer {
+
+    public static final int VERSION_1 = 1;
+
+    private static final int DEFAULT_ITEMS = 1_000_000;
+    private static final double DEFAULT_FPP = 0.1;
+    private static final int DEFAULT_MAX_ITEMS = 10_000_000;
+
+    private static final String ITEMS = "items";
+    private static final String FPP = "fpp";
+    private static final String MAX_ITEMS = "max_items";
+
+    private final DataType dataType;
+    private final int items;
+    private final double fpp;
+
+    private final int maxItems;
+
+    public DynamicBloomFilterFileIndex(DataType dataType, Options options) {
+        this.dataType = dataType;
+        this.items = options.getInteger(ITEMS, DEFAULT_ITEMS);
+        this.fpp = options.getDouble(FPP, DEFAULT_FPP);
+        this.maxItems = options.getInteger(MAX_ITEMS, DEFAULT_MAX_ITEMS);
+    }
+
+    @Override
+    public FileIndexWriter createWriter() {
+        return new DynamicBloomFilterFileIndex.Writer(dataType, items, fpp, maxItems);
+    }
+
+    @Override
+    public FileIndexReader createReader(SeekableInputStream inputStream, int start, int length) {
+        try {
+            inputStream.seek(start);
+            DataInput input = new DataInputStream(inputStream);
+            byte version = input.readByte();
+            if (version > VERSION_1) {
+                throw new RuntimeException(
+                        String.format(
+                                "read dynamicBloomFilter index file fail, "
+                                        + "your plugin version is lower than %d",
+                                version));
+            }
+            int numBloomFilters = input.readInt();
+            int vectorSize = input.readInt();
+            int numHashFunctions = input.readInt();
+
+            BloomFilter64[] bloomFilter64s = new BloomFilter64[numBloomFilters];
+            for (int i = 0; i < numBloomFilters; i++) {
+                byte[] serializedBytes = new byte[vectorSize / Byte.SIZE];
+                input.readFully(serializedBytes);
+                BloomFilter64.BitSet bitSet = new BloomFilter64.BitSet(serializedBytes, 0);
+                bloomFilter64s[i] = new BloomFilter64(numHashFunctions, bitSet);
+            }
+            return new DynamicBloomFilterFileIndex.Reader(dataType, bloomFilter64s);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static class Reader extends FileIndexReader {
+
+        private final DynamicBloomFilter dynamicBloomFilter;
+        private final FastHash hashFunction;
+
+        public Reader(DataType type, BloomFilter64[] bloomFilter64s) {
+            this.dynamicBloomFilter = new DynamicBloomFilter(bloomFilter64s);
+            this.hashFunction = FastHash.getHashFunction(type);
+        }
+
+        @Override
+        public FileIndexResult visitEqual(FieldRef fieldRef, Object key) {
+            return key == null || dynamicBloomFilter.testHash(hashFunction.hash(key))
+                    ? REMAIN
+                    : SKIP;
+        }
+    }
+
+    private static class Writer extends FileIndexWriter {
+
+        private final DynamicBloomFilter filter;
+        private final FastHash hashFunction;
+
+        public Writer(DataType type, int items, double fpp, int maxItems) {
+            this.filter = new DynamicBloomFilter(items, fpp, maxItems);
+            this.hashFunction = FastHash.getHashFunction(type);
+        }
+
+        @Override
+        public void write(Object key) {
+            if (key != null) {
+                long hash = hashFunction.hash(key);
+                if (!filter.testHash(hash)) {
+                    filter.addHash(hash);
+                }
+            }
+        }
+
+        @Override
+        public byte[] serializedBytes() {
+            try {
+                ByteArrayOutputStream bos = new ByteArrayOutputStream();
+                DataOutput out = new DataOutputStream(bos);
+                BloomFilter64[] bloomFilterMatrix = filter.matrix();
+
+                // 1. write meta
+                out.writeByte(VERSION_1);
+                out.writeInt(bloomFilterMatrix.length);
+                // each bloom filter has same num of hashFunction and bitSet
+                out.writeInt(bloomFilterMatrix[0].getBitSet().bitSize());
+                out.writeInt(bloomFilterMatrix[0].getNumHashFunctions());
+
+                // 2. write each filter's bitset
+                for (BloomFilter64 filterMatrix : bloomFilterMatrix) {
+                    byte[] serialized = new byte[filterMatrix.getBitSet().bitSize() / Byte.SIZE];
+                    filterMatrix.getBitSet().toByteArray(serialized, 0, serialized.length);
+                    out.write(serialized);
+                }
+                return bos.toByteArray();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexFactory.java b/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexFactory.java
new file mode 100644
index 000000000000..8b569307c161
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.fileindex.dynamicbloomfilter;
+
+import org.apache.paimon.fileindex.FileIndexer;
+import org.apache.paimon.fileindex.FileIndexerFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataType;
+
+/** Factory to create {@link DynamicBloomFilterFileIndexFactory}. */
+public class DynamicBloomFilterFileIndexFactory implements FileIndexerFactory {
+
+    public static final String DYNAMIC_BLOOM_FILTER = "dynamic-bloom-filter";
+
+    @Override
+    public String identifier() {
+        return DYNAMIC_BLOOM_FILTER;
+    }
+
+    @Override
+    public FileIndexer create(DataType dataType, Options options) {
+        return new DynamicBloomFilterFileIndex(dataType, options);
+    }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/DynamicBloomFilter.java b/paimon-common/src/main/java/org/apache/paimon/utils/DynamicBloomFilter.java
new file mode 100644
index 000000000000..2dfbe3884dd9
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/DynamicBloomFilter.java
@@ -0,0 +1,108 @@
+/*
+ * 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;
+
+/* This file is based on source code from the hudi Project (http://hudi.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Dynamic Bloom Filter. This is largely based of
+ * org.apache.hudi.common.bloom.InternalDynamicBloomFilter.
+ */
+public class DynamicBloomFilter {
+
+    /** Threshold for the maximum number of key to record in a dynamic Bloom filter row. */
+    private int items;
+
+    /** The number of keys recorded in the current standard active Bloom filter. */
+    private int currentNbRecord;
+
+    private int maxItems;
+    private boolean reachedMax = false;
+    private int curMatrixIndex = 0;
+    private double fpp;
+
+    /** The matrix of Bloom filter. */
+    private BloomFilter64[] matrix;
+
+    public DynamicBloomFilter(BloomFilter64[] bloomFilter64s) {
+        this.matrix = bloomFilter64s;
+    }
+
+    public DynamicBloomFilter(int items, double fpp, int maxItems) {
+        this.items = items;
+        this.currentNbRecord = 0;
+        this.maxItems = maxItems;
+        this.fpp = fpp;
+        matrix = new BloomFilter64[1];
+        matrix[0] = new BloomFilter64(items, fpp);
+    }
+
+    public void addHash(long hash64) {
+        BloomFilter64 bf = getActiveStandardBF();
+        if (bf == null) {
+            addRow();
+            bf = matrix[matrix.length - 1];
+            currentNbRecord = 0;
+        }
+        bf.addHash(hash64);
+        currentNbRecord++;
+    }
+
+    /** Adds a new row to <i>this</i> dynamic Bloom filter. */
+    private void addRow() {
+        BloomFilter64[] tmp = new BloomFilter64[matrix.length + 1];
+        System.arraycopy(matrix, 0, tmp, 0, matrix.length);
+        tmp[tmp.length - 1] = new BloomFilter64(items, fpp);
+        matrix = tmp;
+    }
+
+    public boolean testHash(long key) {
+        for (BloomFilter64 bloomFilter : matrix) {
+            if (bloomFilter.testHash(key)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Returns the active standard Bloom filter in <i>this</i> dynamic Bloom filter.
+     *
+     * @return BloomFilter64 The active standard Bloom filter. <code>Null</code> otherwise.
+     */
+    private BloomFilter64 getActiveStandardBF() {
+        if (reachedMax) {
+            return matrix[curMatrixIndex++ % matrix.length];
+        }
+
+        if (currentNbRecord >= items && (matrix.length * items) < maxItems) {
+            return null;
+        } else if (currentNbRecord >= items && (matrix.length * items) >= maxItems) {
+            reachedMax = true;
+            return matrix[0];
+        }
+        return matrix[matrix.length - 1];
+    }
+
+    public BloomFilter64[] matrix() {
+        return matrix;
+    }
+}
diff --git a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory
index 30c908c72381..b2f8c9dde230 100644
--- a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory
+++ b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.fileindex.FileIndexerFactory
@@ -15,4 +15,5 @@
 
 org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory
 org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory
-org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory
\ No newline at end of file
+org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory
+org.apache.paimon.fileindex.dynamicbloomfilter.DynamicBloomFilterFileIndexFactory
\ No newline at end of file
diff --git a/paimon-common/src/test/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexTest.java b/paimon-common/src/test/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexTest.java
new file mode 100644
index 000000000000..10c4335fb12d
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/fileindex/dynamicbloomfilter/DynamicBloomFilterFileIndexTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.fileindex.dynamicbloomfilter;
+
+import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fileindex.FileIndexWriter;
+import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndex;
+import org.apache.paimon.fs.ByteArraySeekableStream;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.DynamicBloomFilter;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Random;
+
+/** Tests for {@link DynamicBloomFilterFileIndex}. */
+public class DynamicBloomFilterFileIndexTest {
+
+    private static final Random RANDOM = new Random();
+
+    @Test
+    public void testAddFindByRandom() {
+        DynamicBloomFilterFileIndex filter =
+                new DynamicBloomFilterFileIndex(
+                        DataTypes.BYTES(),
+                        new Options(
+                                new HashMap<String, String>() {
+                                    {
+                                        put("items", "10000");
+                                        put("fpp", "0.02");
+                                        put("max_items", "50000");
+                                    }
+                                }));
+        FileIndexWriter writer = filter.createWriter();
+        List<byte[]> testData = new ArrayList<>();
+
+        for (int i = 0; i < 10000; i++) {
+            testData.add(random());
+        }
+
+        // test empty bytes
+        testData.add(new byte[0]);
+
+        testData.forEach(writer::write);
+
+        byte[] serializedBytes = writer.serializedBytes();
+        FileIndexReader reader =
+                filter.createReader(
+                        new ByteArraySeekableStream(serializedBytes), 0, serializedBytes.length);
+
+        for (byte[] bytes : testData) {
+            Assertions.assertThat(reader.visitEqual(null, bytes).remain()).isTrue();
+        }
+
+        int errorCount = 0;
+        int num = 1000000;
+        for (int i = 0; i < num; i++) {
+            byte[] ra = random();
+            if (reader.visitEqual(null, ra).remain()) {
+                errorCount++;
+            }
+        }
+
+        // ffp should be less than 0.03
+        Assertions.assertThat((double) errorCount / num).isLessThan(0.03);
+    }
+
+    @Test
+    public void testAddFindByRandomLong() {
+        DynamicBloomFilterFileIndex filter =
+                new DynamicBloomFilterFileIndex(
+                        DataTypes.BIGINT(),
+                        new Options(
+                                new HashMap<String, String>() {
+                                    {
+                                        put("items", "10000");
+                                        put("fpp", "0.02");
+                                        put("max_items", "50000");
+                                    }
+                                }));
+        FileIndexWriter writer = filter.createWriter();
+        List<Long> testData = new ArrayList<>();
+
+        for (int i = 0; i < 10000; i++) {
+            testData.add(RANDOM.nextLong());
+        }
+
+        testData.forEach(writer::write);
+
+        byte[] serializedBytes = writer.serializedBytes();
+        FileIndexReader reader =
+                filter.createReader(
+                        new ByteArraySeekableStream(serializedBytes), 0, serializedBytes.length);
+
+        for (Long value : testData) {
+            Assertions.assertThat(reader.visitEqual(null, value).remain()).isTrue();
+        }
+
+        int errorCount = 0;
+        int num = 1000000;
+        for (int i = 0; i < num; i++) {
+            Long ra = RANDOM.nextLong();
+            if (reader.visitEqual(null, ra).remain()) {
+                errorCount++;
+            }
+        }
+
+        // ffp should be less than 0.03
+        Assertions.assertThat((double) errorCount / num).isLessThan(0.03);
+    }
+
+    @Test
+    public void testCompareFppWithBloomFilter() {
+        // When the amount of data written to the filter exceeds the number of item,
+        // the dynamic bloom filter has a smaller error rate than the bloom filter.
+
+        DynamicBloomFilterFileIndex dynamicFilter =
+                new DynamicBloomFilterFileIndex(
+                        DataTypes.BIGINT(),
+                        new Options(
+                                new HashMap<String, String>() {
+                                    {
+                                        put("items", "10000");
+                                        put("fpp", "0.02");
+                                        put("max_items", "50000");
+                                    }
+                                }));
+
+        BloomFilterFileIndex filter =
+                new BloomFilterFileIndex(
+                        DataTypes.BIGINT(),
+                        new Options(
+                                new HashMap<String, String>() {
+                                    {
+                                        put("items", "10000");
+                                        put("fpp", "0.02");
+                                    }
+                                }));
+
+        FileIndexWriter bloomFilterWriter = filter.createWriter();
+        FileIndexWriter dynamicBloomFilterWriter = dynamicFilter.createWriter();
+
+        List<Long> testData = new ArrayList<>();
+
+        for (int i = 0; i < 20000; i++) {
+            testData.add(RANDOM.nextLong());
+        }
+        testData.forEach(bloomFilterWriter::write);
+        testData.forEach(dynamicBloomFilterWriter::write);
+
+        byte[] bloomFilterSerializedBytes = bloomFilterWriter.serializedBytes();
+        FileIndexReader filterReader =
+                filter.createReader(
+                        new ByteArraySeekableStream(bloomFilterSerializedBytes),
+                        0,
+                        bloomFilterSerializedBytes.length);
+
+        byte[] dynamicSerializedBytes = dynamicBloomFilterWriter.serializedBytes();
+        FileIndexReader dynamicFilterReader =
+                dynamicFilter.createReader(
+                        new ByteArraySeekableStream(dynamicSerializedBytes),
+                        0,
+                        dynamicSerializedBytes.length);
+
+        int errorCount = 0;
+        int num = 1000000;
+        for (int i = 0; i < num; i++) {
+            Long ra = RANDOM.nextLong();
+            if (filterReader.visitEqual(null, ra).remain()) {
+                errorCount++;
+            }
+        }
+        double bloomFilterError = (double) errorCount / num;
+
+        errorCount = 0;
+        for (int i = 0; i < num; i++) {
+            Long ra = RANDOM.nextLong();
+            if (dynamicFilterReader.visitEqual(null, ra).remain()) {
+                errorCount++;
+            }
+        }
+        double dynamicBloomFilterError = (double) errorCount / num;
+
+        // ffp should be less than 0.03
+        Assertions.assertThat(dynamicBloomFilterError).isLessThan(bloomFilterError);
+    }
+
+    @Test
+    public void testDynamicBloomFilterBoundSize() {
+
+        int maxItems = 1000;
+        int items = 100;
+        int batchSize = 100;
+        int count = 0;
+        int curFilterNum = 0;
+
+        DynamicBloomFilter dynamicBloomFilter = new DynamicBloomFilter(items, 0.001, maxItems);
+        for (int i = 0; i < 15; i++) {
+            for (int j = 0; j < batchSize; j++) {
+                dynamicBloomFilter.addHash(RANDOM.nextLong());
+                count++;
+            }
+            // When the amount of inserted data exceeds the maxItems, the number of bloom filters
+            // will not change.
+            if (count > maxItems) {
+                Assertions.assertThat(curFilterNum).isEqualTo(dynamicBloomFilter.matrix().length);
+            } else {
+                Assertions.assertThat(curFilterNum).isLessThan(dynamicBloomFilter.matrix().length);
+                curFilterNum = dynamicBloomFilter.matrix().length;
+            }
+        }
+    }
+
+    private byte[] random() {
+        byte[] b = new byte[Math.abs(RANDOM.nextInt(400) + 1)];
+        RANDOM.nextBytes(b);
+        return b;
+    }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlyFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlyFileStoreTableTest.java
index 85ab7484f184..099c822158b8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlyFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/AppendOnlyFileStoreTableTest.java
@@ -29,6 +29,7 @@
 import org.apache.paimon.fileindex.bitmap.BitmapFileIndexFactory;
 import org.apache.paimon.fileindex.bloomfilter.BloomFilterFileIndexFactory;
 import org.apache.paimon.fileindex.bsi.BitSliceIndexBitmapFileIndexFactory;
+import org.apache.paimon.fileindex.dynamicbloomfilter.DynamicBloomFilterFileIndexFactory;
 import org.apache.paimon.fs.FileIOFinder;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
@@ -588,6 +589,154 @@ public void testBloomFilterInDisk() throws Exception {
         reader.forEachRemaining(row -> assertThat(row.getString(1).toString()).isEqualTo("b"));
     }
 
+    @Test
+    public void testDynamicBloomFilterInDisk() throws Exception {
+        RowType rowType =
+                RowType.builder()
+                        .field("id", DataTypes.INT())
+                        .field("index_column", DataTypes.STRING())
+                        .field("index_column2", DataTypes.INT())
+                        .field("index_column3", DataTypes.BIGINT())
+                        .build();
+        // in unaware-bucket mode, we split files into splits all the time
+        FileStoreTable table =
+                createUnawareBucketFileStoreTable(
+                        rowType,
+                        options -> {
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + "."
+                                            + CoreOptions.COLUMNS,
+                                    "index_column, index_column2, index_column3");
+                            options.set(FILE_INDEX_IN_MANIFEST_THRESHOLD.key(), "50 B");
+                        });
+
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+        List<CommitMessage> result = new ArrayList<>();
+        write.write(GenericRow.of(1, BinaryString.fromString("a"), 2, 3L));
+        write.write(GenericRow.of(1, BinaryString.fromString("c"), 2, 3L));
+        result.addAll(write.prepareCommit(true, 0));
+        write.write(GenericRow.of(1, BinaryString.fromString("b"), 2, 3L));
+        result.addAll(write.prepareCommit(true, 0));
+        commit.commit(0, result);
+        result.clear();
+
+        TableScan.Plan plan =
+                table.newScan()
+                        .withFilter(
+                                new PredicateBuilder(rowType)
+                                        .equal(1, BinaryString.fromString("b")))
+                        .plan();
+        List<DataFileMeta> metas =
+                plan.splits().stream()
+                        .flatMap(split -> ((DataSplit) split).dataFiles().stream())
+                        .collect(Collectors.toList());
+        assertThat(metas.size()).isEqualTo(2);
+
+        RecordReader<InternalRow> reader =
+                table.newRead()
+                        .withFilter(
+                                new PredicateBuilder(rowType)
+                                        .equal(1, BinaryString.fromString("b")))
+                        .createReader(plan.splits());
+        reader.forEachRemaining(row -> assertThat(row.getString(1).toString()).isEqualTo("b"));
+    }
+
+    @Test
+    public void testDynamicBloomFilterInMemory() throws Exception {
+        RowType rowType =
+                RowType.builder()
+                        .field("id", DataTypes.INT())
+                        .field("index_column", DataTypes.STRING())
+                        .field("index_column2", DataTypes.INT())
+                        .field("index_column3", DataTypes.BIGINT())
+                        .build();
+        // in unaware-bucket mode, we split files into splits all the time
+        FileStoreTable table =
+                createUnawareBucketFileStoreTable(
+                        rowType,
+                        options -> {
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + "."
+                                            + CoreOptions.COLUMNS,
+                                    "index_column, index_column2, index_column3");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column.items",
+                                    "150");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column.max_items",
+                                    "150");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column2.items",
+                                    "150");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column2.max_items",
+                                    "150");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column3.items",
+                                    "150");
+                            options.set(
+                                    FileIndexOptions.FILE_INDEX
+                                            + "."
+                                            + DynamicBloomFilterFileIndexFactory
+                                                    .DYNAMIC_BLOOM_FILTER
+                                            + ".index_column3.max_items",
+                                    "150");
+                            options.set(FILE_INDEX_IN_MANIFEST_THRESHOLD.key(), "500 B");
+                        });
+
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+        List<CommitMessage> result = new ArrayList<>();
+        write.write(GenericRow.of(1, BinaryString.fromString("a"), 2, 3L));
+        write.write(GenericRow.of(1, BinaryString.fromString("c"), 2, 3L));
+        result.addAll(write.prepareCommit(true, 0));
+        write.write(GenericRow.of(1, BinaryString.fromString("b"), 2, 3L));
+        result.addAll(write.prepareCommit(true, 0));
+        commit.commit(0, result);
+        result.clear();
+
+        TableScan.Plan plan =
+                table.newScan()
+                        .withFilter(
+                                new PredicateBuilder(rowType)
+                                        .equal(1, BinaryString.fromString("b")))
+                        .plan();
+        List<DataFileMeta> metas =
+                plan.splits().stream()
+                        .flatMap(split -> ((DataSplit) split).dataFiles().stream())
+                        .collect(Collectors.toList());
+        assertThat(metas.size()).isEqualTo(1);
+    }
+
     @Test
     public void testBSIAndBitmapIndexInMemory() throws Exception {
         RowType rowType =
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
index 9427ed61c9b9..70bef4f7e6e4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
@@ -777,6 +777,82 @@ public void testWithShardDeletionVectors() throws Exception {
         innerTestWithShard(table);
     }
 
+    @Test
+    public void testDeletionVectorsWithDynamicBloomFileIndexInMeta() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(
+                        conf -> {
+                            conf.set(BUCKET, 1);
+                            conf.set(DELETION_VECTORS_ENABLED, true);
+                            conf.set(TARGET_FILE_SIZE, MemorySize.ofBytes(1));
+                            conf.set("file-index.dynamic-bloom-filter.columns", "b");
+                            conf.set("file-index.dynamic-bloom-filter.b.items", "20");
+                            conf.set("file-index.dynamic-bloom-filter.b.max_items", "20");
+                        });
+
+        StreamTableWrite write =
+                table.newWrite(commitUser).withIOManager(new IOManagerImpl(tempDir.toString()));
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowData(1, 1, 300L));
+        write.write(rowData(1, 2, 400L));
+        write.write(rowData(1, 3, 200L));
+        write.write(rowData(1, 4, 500L));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        write.write(rowData(1, 5, 100L));
+        write.write(rowData(1, 6, 600L));
+        write.write(rowData(1, 7, 400L));
+        commit.commit(1, write.prepareCommit(true, 1));
+
+        PredicateBuilder builder = new PredicateBuilder(ROW_TYPE);
+        Predicate predicate = builder.equal(2, 300L);
+
+        List<Split> splits =
+                toSplits(table.newSnapshotReader().withFilter(predicate).read().dataSplits());
+
+        assertThat(((DataSplit) splits.get(0)).dataFiles().size()).isEqualTo(1);
+    }
+
+    @Test
+    public void testDeletionVectorsWithDynamicBloomFileIndexInFile() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(
+                        conf -> {
+                            conf.set(BUCKET, 1);
+                            conf.set(DELETION_VECTORS_ENABLED, true);
+                            conf.set(TARGET_FILE_SIZE, MemorySize.ofBytes(1));
+                            conf.set("file-index.dynamic-bloom-filter.columns", "b");
+                        });
+
+        StreamTableWrite write =
+                table.newWrite(commitUser).withIOManager(new IOManagerImpl(tempDir.toString()));
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowData(1, 1, 300L));
+        write.write(rowData(1, 2, 400L));
+        write.write(rowData(1, 3, 200L));
+        write.write(rowData(1, 4, 500L));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        write.write(rowData(1, 5, 100L));
+        write.write(rowData(1, 6, 600L));
+        write.write(rowData(1, 7, 400L));
+        commit.commit(1, write.prepareCommit(true, 1));
+
+        PredicateBuilder builder = new PredicateBuilder(ROW_TYPE);
+        List<Split> splits = toSplits(table.newSnapshotReader().read().dataSplits());
+        assertThat(((DataSplit) splits.get(0)).dataFiles().size()).isEqualTo(2);
+        TableRead read = table.newRead().withFilter(builder.equal(2, 300L));
+        assertThat(getResult(read, splits, BATCH_ROW_TO_STRING))
+                .hasSameElementsAs(
+                        Arrays.asList(
+                                "1|1|300|binary|varbinary|mapKey:mapVal|multiset",
+                                "1|2|400|binary|varbinary|mapKey:mapVal|multiset",
+                                "1|3|200|binary|varbinary|mapKey:mapVal|multiset",
+                                "1|4|500|binary|varbinary|mapKey:mapVal|multiset"));
+    }
+
     @Test
     public void testDeletionVectorsWithFileIndexInFile() throws Exception {
         FileStoreTable table =
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkFileIndexTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkFileIndexTest.java
new file mode 100644
index 000000000000..a2873a7ef5dc
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkFileIndexTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.flink;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fileindex.FileIndexFormat;
+import org.apache.paimon.fileindex.FileIndexReader;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** ITCase for FileIndex. */
+public class FlinkFileIndexTest extends CatalogITCaseBase {
+
+    @Test
+    public void testDynamicBloomFileIndex() throws Catalog.TableNotExistException, IOException {
+        sql(
+                "CREATE TABLE T ("
+                        + " k INT,"
+                        + " v STRING,"
+                        + " hh INT,"
+                        + " dt STRING"
+                        + ") PARTITIONED BY (dt, hh) WITH ("
+                        + " 'write-only' = 'true',"
+                        + " 'bucket' = '-1',"
+                        + " 'file-index.dynamic-bloom-filter.columns'='k,v'"
+                        + ")");
+
+        sql(
+                "INSERT INTO T VALUES (1, '100', 15, '20221208'), (1, '100', 16, '20221208'), (1, '100', 15, '20221209')");
+
+        FileStoreTable table = paimonTable("T");
+        List<ManifestEntry> list = table.store().newScan().plan().files();
+        for (ManifestEntry entry : list) {
+            List<String> extraFiles =
+                    entry.file().extraFiles().stream()
+                            .filter(s -> s.endsWith(DataFilePathFactory.INDEX_PATH_SUFFIX))
+                            .collect(Collectors.toList());
+
+            Assertions.assertThat(extraFiles.size()).isEqualTo(1);
+
+            String file = extraFiles.get(0);
+
+            Path indexFilePath =
+                    table.store()
+                            .pathFactory()
+                            .createDataFilePathFactory(entry.partition(), entry.bucket())
+                            .toAlignedPath(file, entry.file());
+            try (FileIndexFormat.Reader reader =
+                    FileIndexFormat.createReader(
+                            table.fileIO().newInputStream(indexFilePath), table.rowType())) {
+                Set<FileIndexReader> readerSetK = reader.readColumnIndex("k");
+                Assertions.assertThat(readerSetK.size()).isEqualTo(1);
+
+                Predicate predicateK = new PredicateBuilder(table.rowType()).equal(0, 1);
+                for (FileIndexReader fileIndexReader : readerSetK) {
+                    Assertions.assertThat(predicateK.visit(fileIndexReader).remain()).isTrue();
+                }
+
+                predicateK = new PredicateBuilder(table.rowType()).equal(0, 4);
+                for (FileIndexReader fileIndexReader : readerSetK) {
+                    Assertions.assertThat(predicateK.visit(fileIndexReader).remain()).isFalse();
+                }
+
+                Set<FileIndexReader> readerSetV = reader.readColumnIndex("v");
+                Assertions.assertThat(readerSetV.size()).isEqualTo(1);
+
+                Predicate predicateV =
+                        new PredicateBuilder(table.rowType())
+                                .equal(1, BinaryString.fromString("100"));
+                for (FileIndexReader fileIndexReader : readerSetV) {
+                    Assertions.assertThat(predicateV.visit(fileIndexReader).remain()).isTrue();
+                }
+
+                predicateV =
+                        new PredicateBuilder(table.rowType())
+                                .equal(1, BinaryString.fromString("101"));
+                for (FileIndexReader fileIndexReader : readerSetV) {
+                    Assertions.assertThat(predicateV.visit(fileIndexReader).remain()).isFalse();
+                }
+            }
+        }
+    }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
index fb8bee5d5962..f1b2ffb37ce3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/UnawareBucketAppendOnlyTableITCase.java
@@ -365,11 +365,21 @@ public void testLimit() {
     @Test
     public void testFileIndex() {
         batchSql(
-                "INSERT INTO index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'c', 'BBB'), (3, 'c', 'BBB')");
+                "INSERT INTO bloom_index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'c', 'BBB'), (3, 'c', 'BBB')");
         batchSql(
-                "INSERT INTO index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'd', 'BBB'), (3, 'd', 'BBB')");
+                "INSERT INTO bloom_index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'd', 'BBB'), (3, 'd', 'BBB')");
+        assertThat(
+                        batchSql(
+                                "SELECT * FROM bloom_index_table WHERE indexc = 'c' and (id = 2 or id = 3)"))
+                .containsExactlyInAnyOrder(Row.of(2, "c", "BBB"), Row.of(3, "c", "BBB"));
 
-        assertThat(batchSql("SELECT * FROM index_table WHERE indexc = 'c' and (id = 2 or id = 3)"))
+        batchSql(
+                "INSERT INTO dynamic_bloom_index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'c', 'BBB'), (3, 'c', 'BBB')");
+        batchSql(
+                "INSERT INTO dynamic_bloom_index_table VALUES (1, 'a', 'AAA'), (1, 'a', 'AAA'), (2, 'd', 'BBB'), (3, 'd', 'BBB')");
+        assertThat(
+                        batchSql(
+                                "SELECT * FROM dynamic_bloom_index_table WHERE indexc = 'c' and (id = 2 or id = 3)"))
                 .containsExactlyInAnyOrder(Row.of(2, "c", "BBB"), Row.of(3, "c", "BBB"));
     }
 
@@ -480,7 +490,8 @@ protected List<String> ddl() {
                 "CREATE TABLE IF NOT EXISTS append_table (id INT, data STRING) WITH ('bucket' = '-1')",
                 "CREATE TABLE IF NOT EXISTS part_table (id INT, data STRING, dt STRING) PARTITIONED BY (dt) WITH ('bucket' = '-1')",
                 "CREATE TABLE IF NOT EXISTS complex_table (id INT, data MAP<INT, INT>) WITH ('bucket' = '-1')",
-                "CREATE TABLE IF NOT EXISTS index_table (id INT, indexc STRING, data STRING) WITH ('bucket' = '-1', 'file-index.bloom-filter.columns'='indexc', 'file-index.bloom-filter.indexc.items' = '500')");
+                "CREATE TABLE IF NOT EXISTS bloom_index_table (id INT, indexc STRING, data STRING) WITH ('bucket' = '-1', 'file-index.bloom-filter.columns'='indexc', 'file-index.bloom-filter.indexc.items' = '500')",
+                "CREATE TABLE IF NOT EXISTS dynamic_bloom_index_table (id INT, indexc STRING, data STRING) WITH ('bucket' = '-1', 'file-index.bloom-filter.columns'='indexc', 'file-index.bloom-filter.indexc.items' = '500','file-index.bloom-filter.indexc.max_items' = '1000')");
     }
 
     @Override