Add implementation for bloom filters (predicate pushdown)

.gitignore

@@ -9,3 +9,4 @@ BenchmarkDotNet.Artifacts/
 Parquet.sln.DotSettings.user
 launchSettings.json
 .DS_Store
+.vscode

src/Parquet.Test/Bloom/BloomCollectorTest.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Linq;
+using Parquet.Bloom;
+using Xunit;
+using Encoding = System.Text.Encoding;
+
+namespace Parquet.Test.Bloom {
+    /// <summary>
+    /// Verifies that <see cref="BloomCollector"/> inserts PLAIN-encoded bytes
+    /// (without variable-length prefixes) into a <see cref="SplitBlockBloomFilter"/>
+    /// for various Parquet physical types.
+    /// </summary>
+    public sealed class BloomCollectorTest {
+        /// <summary>
+        /// Ensures that <see cref="BloomCollector.AddString"/> uses UTF-8 bytes of the string
+        /// (without any length prefix), per the Parquet bloom filter spec.
+        /// </summary>
+        [Fact]
+        public void AddString_Uses_Utf8_Bytes_NoLength() {
+            int blocks = 64;
+            var viaCollector = new BloomCollector(blocks);
+            var manual = new SplitBlockBloomFilter(blocks);
+
+            string s = "héłło world 👋";
+            byte[] utf8 = Encoding.UTF8.GetBytes(s);
+
+            viaCollector.AddString(s);
+            manual.Insert(utf8);
+
+            Assert.True(viaCollector.Filter.MightContain(utf8));
+            Assert.True(manual.MightContain(utf8));
+        }
+
+        /// <summary>
+        /// Ensures that <see cref="BloomCollector.AddByteArray"/> inserts the byte content as-is
+        /// (no length prefix), matching manual insertion.
+        /// </summary>
+        [Fact]
+        public void AddByteArray_AsIs_NoLength() {
+            int blocks = 32;
+            var viaCollector = new BloomCollector(blocks);
+            var manual = new SplitBlockBloomFilter(blocks);
+
+            byte[] payload = Enumerable.Range(0, 16).Select(i => (byte)((i * 7) + 3)).ToArray();
+
+            viaCollector.AddByteArray(payload);
+            manual.Insert(payload);
+
+            Assert.True(viaCollector.Filter.MightContain(payload));
+            Assert.True(manual.MightContain(payload));
+        }
+
+        /// <summary>
+        /// Verifies that <see cref="BloomCollector.AddFixed"/> inserts bytes as-is, appropriate for
+        /// FIXED_LEN_BYTE_ARRAY columns.
+        /// </summary>
+        [Fact]
+        public void AddFixed_AsIs() {
+            int blocks = 32;
+            var viaCollector = new BloomCollector(blocks);
+            var manual = new SplitBlockBloomFilter(blocks);
+
+            byte[] fixedBytes = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF, 0xAA, 0x55 };
+
+            viaCollector.AddFixed(fixedBytes);
+            manual.Insert(fixedBytes);
+
+            Assert.True(viaCollector.Filter.MightContain(fixedBytes));
+            Assert.True(manual.MightContain(fixedBytes));
+        }
+
+        /// <summary>
+        /// Confirms that null inputs are ignored by the collector and do not modify the filter.
+        /// Compares the bitset before and after a series of null insertions.
+        /// </summary>
+        [Fact]
+        public void Nulls_Are_Ignored() {
+            int blocks = 16;
+            var viaCollector = new BloomCollector(blocks);
+
+            byte[] before = viaCollector.Filter.ToByteArray();
+
+            viaCollector.AddBoolean(null);
+            viaCollector.AddInt32(null);
+            viaCollector.AddInt64(null);
+            viaCollector.AddFloat(null);
+            viaCollector.AddDouble(null);
+            viaCollector.AddString(null);
+            viaCollector.AddByteArray(null);
+            viaCollector.AddFixed(null);
+
+            byte[] after = viaCollector.Filter.ToByteArray();
+
+            Assert.Equal(before, after);
+        }
+    }
+}

src/Parquet.Test/Bloom/BloomFilterIO_ReaderTests.cs

@@ -0,0 +1,115 @@
+using System;
+using System.IO;
+using Parquet.Bloom;
+using Parquet.Meta;
+using Parquet.Meta.Proto;
+using Xunit;
+using Encoding = System.Text.Encoding;
+
+namespace Parquet.Test.Bloom {
+    /// <summary>
+    /// Reader-side round-trip tests for Parquet Bloom Filters.
+    /// Loads a filter using <see cref="ColumnMetaData.BloomFilterOffset"/> and
+    /// probes it to verify correctness and error handling.
+    /// </summary>
+    public sealed class BloomFilterIO_ReaderTest {
+        private static ThriftCompactProtocolWriter MakeWriter(Stream s) => new ThriftCompactProtocolWriter(s);
+        private static ThriftCompactProtocolReader MakeReader(Stream s) => new ThriftCompactProtocolReader(s);
+
+        /// <summary>
+        /// Writes a <see cref="SplitBlockBloomFilter"/> (header + bitset) to a stream,
+        /// records the offset in <see cref="ColumnMetaData.BloomFilterOffset"/>,
+        /// then reads it back and verifies probes for present/absent values.
+        /// Ensures the reader seeks to the recorded offset and reconstructs the filter correctly.
+        /// </summary>
+        [Fact]
+        public void Read_RoundTrip_Probes_Work() {
+            SplitBlockBloomFilter f = new SplitBlockBloomFilter(16);
+            f.Insert(Encoding.ASCII.GetBytes("parquet"));
+            f.Insert(Encoding.ASCII.GetBytes("bloom"));
+            f.Insert(Encoding.ASCII.GetBytes("filter"));
+
+            ColumnMetaData meta = new ColumnMetaData();
+
+            using(MemoryStream ms = new MemoryStream()) {
+                ms.WriteByte(0x42); // padding to move offset
+
+                (long off, int len) = BloomFilterIO.WriteToStream(ms, f, meta, MakeWriter);
+
+                // Seek somewhere else to prove Read seeks to offset
+                ms.Seek(0, SeekOrigin.Begin);
+
+                SplitBlockBloomFilter re = BloomFilterIO.ReadFromStream(ms, meta, MakeReader);
+
+                Assert.True(re.MightContain(Encoding.ASCII.GetBytes("parquet")));
+                Assert.True(re.MightContain(Encoding.ASCII.GetBytes("bloom")));
+                Assert.True(re.MightContain(Encoding.ASCII.GetBytes("filter")));
+                Assert.False(re.MightContain(Encoding.ASCII.GetBytes("definitely-not-present")));
+            }
+        }
+
+        /// <summary>
+        /// Verifies that attempting to read a bloom filter without
+        /// <see cref="ColumnMetaData.BloomFilterOffset"/> set results in an
+        /// <see cref="InvalidOperationException"/>. Guards against misuse of the API.
+        /// </summary>
+        [Fact]
+        public void Read_Throws_When_No_Offset() {
+            using(MemoryStream ms = new MemoryStream()) {
+                ColumnMetaData meta = new ColumnMetaData();
+                Assert.Throws<InvalidOperationException>(() => BloomFilterIO.ReadFromStream(ms, meta, MakeReader));
+            }
+        }
+
+        /// <summary>
+        /// Crafts an invalid bloom header with <c>NumBytes = 1</c> (not a multiple of 32),
+        /// writes just the header at a non-zero offset, and asserts that reading fails with
+        /// <see cref="InvalidDataException"/>. Validates header sanity checks before bitset read.
+        /// </summary>
+        [Fact]
+        public void Read_Rejects_Invalid_Header_NumBytes() {
+            // Hand-craft a header with NumBytes = 1 (invalid), then try to read.
+            using(MemoryStream ms = new MemoryStream()) {
+                ColumnMetaData meta = new ColumnMetaData();
+
+                // write at non-zero position
+                ms.WriteByte(0x00);
+                long offset = ms.Position;
+
+                BloomFilterHeader bad = new BloomFilterHeader();
+                bad.NumBytes = 1; // invalid (not multiple of 32)
+                bad.Algorithm = new BloomFilterAlgorithm { BLOCK = new SplitBlockAlgorithm() };
+                bad.Hash = new BloomFilterHash { XXHASH = new XxHash() };
+                bad.Compression = new BloomFilterCompression { UNCOMPRESSED = new Uncompressed() };
+
+                ThriftCompactProtocolWriter writer = new ThriftCompactProtocolWriter(ms);
+                bad.Write(writer);
+
+                meta.BloomFilterOffset = offset;
+                // No bitset written
+
+                Assert.Throws<InvalidDataException>(() => BloomFilterIO.ReadFromStream(ms, meta, s => new ThriftCompactProtocolReader(s)));
+            }
+        }
+
+        /// <summary>
+        /// Ensures writer populates BloomFilterOffset/Length and serialized bytes
+        /// contain the expected NumBytes plus some header overhead.
+        /// </summary>
+        [Fact]
+        public void Writer_Populates_Meta_And_Writes_Bytes() {
+            var f = new SplitBlockBloomFilter(8);
+            f.Insert(Encoding.ASCII.GetBytes("alpha"));
+
+            var meta = new ColumnMetaData();
+            using var ms = new MemoryStream();
+            ms.WriteByte(0xCC);
+
+            (long Offset, int Length) _ = BloomFilterIO.WriteToStream(ms, f, meta, s => new ThriftCompactProtocolWriter(s));
+
+            Assert.True(meta.BloomFilterOffset.HasValue);
+            Assert.True(meta.BloomFilterLength.HasValue);
+            Assert.True(meta.BloomFilterLength.Value >= f.NumberOfBlocks * 32);
+        }
+    }
+}

src/Parquet.Test/Bloom/BloomHashTest.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Buffers.Binary;
+using Xunit;
+
+namespace Parquet.Test.Bloom {
+    public class BloomHashTest {
+        [Fact]
+        public void Hash_Int32_IsStable() {
+            byte[] bytes = new byte[4];
+            BinaryPrimitives.WriteInt32LittleEndian(bytes, 12345);
+            Assert.Equal(
+                Parquet.Bloom.BloomHasher.HashPlainEncoded(bytes),
+                Parquet.Bloom.BloomHasher.HashPlainEncoded(bytes));
+        }
+
+        [Fact]
+        public void HashPlainEncoded_Slice_Equals_SpanSlice() {
+            byte[] data = { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+            int off = 3, len = 4;
+            ulong a = Parquet.Bloom.BloomHasher.HashPlainEncoded(data, off, len);
+            ulong b = Parquet.Bloom.BloomHasher.HashPlainEncoded(new ReadOnlySpan<byte>(data, off, len));
+            Assert.Equal(a, b);
+        }
+    }
+}