diff --git a/.gitignore b/.gitignore index 32026bcd..a54e22ae 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,4 @@ BenchmarkDotNet.Artifacts/ Parquet.sln.DotSettings.user launchSettings.json .DS_Store +.vscode \ No newline at end of file diff --git a/src/Parquet.Test/Bloom/BloomCollectorTest.cs b/src/Parquet.Test/Bloom/BloomCollectorTest.cs new file mode 100644 index 00000000..949c9e5c --- /dev/null +++ b/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 { + /// + /// Verifies that inserts PLAIN-encoded bytes + /// (without variable-length prefixes) into a + /// for various Parquet physical types. + /// + public sealed class BloomCollectorTest { + /// + /// Ensures that uses UTF-8 bytes of the string + /// (without any length prefix), per the Parquet bloom filter spec. + /// + [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)); + } + + /// + /// Ensures that inserts the byte content as-is + /// (no length prefix), matching manual insertion. + /// + [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)); + } + + /// + /// Verifies that inserts bytes as-is, appropriate for + /// FIXED_LEN_BYTE_ARRAY columns. + /// + [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)); + } + + /// + /// 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. + /// + [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); + } + } +} \ No newline at end of file diff --git a/src/Parquet.Test/Bloom/BloomFilterIO_ReaderTests.cs b/src/Parquet.Test/Bloom/BloomFilterIO_ReaderTests.cs new file mode 100644 index 00000000..d2e6311d --- /dev/null +++ b/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 { + /// + /// Reader-side round-trip tests for Parquet Bloom Filters. + /// Loads a filter using and + /// probes it to verify correctness and error handling. + /// + public sealed class BloomFilterIO_ReaderTest { + private static ThriftCompactProtocolWriter MakeWriter(Stream s) => new ThriftCompactProtocolWriter(s); + private static ThriftCompactProtocolReader MakeReader(Stream s) => new ThriftCompactProtocolReader(s); + + /// + /// Writes a (header + bitset) to a stream, + /// records the offset in , + /// then reads it back and verifies probes for present/absent values. + /// Ensures the reader seeks to the recorded offset and reconstructs the filter correctly. + /// + [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"))); + } + } + + /// + /// Verifies that attempting to read a bloom filter without + /// set results in an + /// . Guards against misuse of the API. + /// + [Fact] + public void Read_Throws_When_No_Offset() { + using(MemoryStream ms = new MemoryStream()) { + ColumnMetaData meta = new ColumnMetaData(); + Assert.Throws(() => BloomFilterIO.ReadFromStream(ms, meta, MakeReader)); + } + } + + /// + /// Crafts an invalid bloom header with NumBytes = 1 (not a multiple of 32), + /// writes just the header at a non-zero offset, and asserts that reading fails with + /// . Validates header sanity checks before bitset read. + /// + [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(() => BloomFilterIO.ReadFromStream(ms, meta, s => new ThriftCompactProtocolReader(s))); + } + } + + /// + /// Ensures writer populates BloomFilterOffset/Length and serialized bytes + /// contain the expected NumBytes plus some header overhead. + /// + [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); + } + } +} \ No newline at end of file diff --git a/src/Parquet.Test/Bloom/BloomHashTest.cs b/src/Parquet.Test/Bloom/BloomHashTest.cs new file mode 100644 index 00000000..941bcf0d --- /dev/null +++ b/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(data, off, len)); + Assert.Equal(a, b); + } + } +} diff --git a/src/Parquet.Test/Bloom/Bloom_EndToEnd_Test.cs b/src/Parquet.Test/Bloom/Bloom_EndToEnd_Test.cs new file mode 100644 index 00000000..589c9319 --- /dev/null +++ b/src/Parquet.Test/Bloom/Bloom_EndToEnd_Test.cs @@ -0,0 +1,387 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading.Tasks; +using Parquet.Bloom; +using Parquet.Data; +using Parquet.File; +using Parquet.Meta; +using Parquet.Meta.Proto; +using Parquet.Schema; +using Xunit; + +namespace Parquet.Test.Bloom { + /// + /// End-to-end tests for writing and reading columns with Split-Block Bloom filters. + /// Verifies on-disk metadata (offset/length), in-memory probing, pruning, and data roundtrip. + /// + public sealed class Bloom_EndToEnd_Test : TestBase { + private static DataField I32(string name = "c") => new DataField(name, System.Type.GetType("System.Int32")!); + private static DataField STRING(string name = "s") => new DataField(name, System.Type.GetType("System.String")!); + private static SchemaElement I32Physical(string name = "c") => + new SchemaElement { Name = name, Type = Parquet.Meta.Type.INT32, RepetitionType = FieldRepetitionType.REQUIRED }; + + /// + /// Writes a single INT32 column with Bloom filters enabled and confirms: + /// - BloomFilterOffset/Length are populated in ColumnMetaData, + /// - the Bloom can be read back and probed for present/absent values, + /// - the column data round-trips correctly through the reader. + /// + [Fact] + public async Task Write_With_Bloom_Then_Read_And_Probe() { + // 1) Field + attach to schema + DataField field = I32(); + var schema = new ParquetSchema(field); // attaches the field + + // 2) Footer must be created with schema + total row count + int[] values = { 7, 42, 1000, -5, 7 }; + var footer = new ThriftFooter(schema, totalRowCount: values.Length); + + // 3) SchemaElement for the physical type (you can keep your manual one, + // or grab it from footer.GetWriteableSchema()[0]) + var se = new SchemaElement { + Name = field.Name, + Type = Meta.Type.INT32, + RepetitionType = FieldRepetitionType.REQUIRED + }; + + // 4) Now safe to construct the DataColumn + var col = new DataColumn(field, values); + + using var ms = new MemoryStream(); + + var writer = new DataColumnWriter( + ms, footer, se, + compressionMethod: CompressionMethod.None, + options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + compressionLevel: System.IO.Compression.CompressionLevel.NoCompression, + keyValueMetadata: null); + + ColumnChunk chunk = await writer.WriteAsync(new FieldPath(field.Name), col); + + // Assert writer populated bloom offsets + Assert.NotNull(chunk.MetaData); + Assert.NotNull(chunk.MetaData!.BloomFilterOffset); + Assert.NotNull(chunk.MetaData!.BloomFilterLength); + Assert.True(chunk.MetaData!.BloomFilterLength!.Value > 0); + + // Read bloom back from the stream using the metadata offsets + ms.Position = 0; // BloomFilterIO.ReadFromStream will seek to the offset + SplitBlockBloomFilter re = + BloomFilterIO.ReadFromStream(ms, chunk.MetaData!, s => new ThriftCompactProtocolReader(s)); + + // Present values should MightContain == true (probabilistic); choose one absent probe + Assert.True(re.MightContain(PlainLE(42))); + Assert.True(re.MightContain(PlainLE(7))); + Assert.False(re.MightContain(PlainLE(999999))); + + // Also test reader-side pruning helper on a real DataColumnReader + ms.Position = 0; + var stats = new DataColumnStatistics { NullCount = 0, DistinctCount = 4 }; + var reader = new DataColumnReader(field, ms, chunk, stats, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + + // Pruning checks + Assert.True(reader.MightMatchEquals(42)); // present -> might match + Assert.False(reader.MightMatchEquals(9999)); // clearly absent -> pruned + + // Finally, fully read the column back and verify data + ms.Position = 0; + DataColumn roundTripped = await reader.ReadAsync(); + Assert.Equal(values, (int[])roundTripped.Data); + } + + /// + /// Writes the same INT32 column with Bloom filters **disabled** and confirms: + /// - no Bloom offset/length recorded, + /// - reader exposes no pruning (returns true), + /// - data still round-trips. + /// + [Fact] + public async Task Write_Without_Bloom_No_Metadata_No_Prune() { + // 1) Field + attach to schema + DataField field = I32(); + var schema = new Parquet.Schema.ParquetSchema(field); + + // 2) Footer with total row count + int[] values = { 1, 2, 3, 4 }; + var footer = new ThriftFooter(schema, totalRowCount: values.Length); + + // 3) Physical schema element (or use footer.GetWriteableSchema()[0]) + SchemaElement se = I32Physical(field.Name); + + // 4) Now it's safe to build the DataColumn + DataColumn col = new DataColumn(field, values); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter( + ms, footer, se, + compressionMethod: CompressionMethod.None, + options: new ParquetOptions(), + compressionLevel: System.IO.Compression.CompressionLevel.NoCompression, + keyValueMetadata: null); + + ColumnChunk chunk = await writer.WriteAsync(new FieldPath(field.Name), col); + + Assert.NotNull(chunk.MetaData); + Assert.Null(chunk.MetaData!.BloomFilterOffset); + Assert.Null(chunk.MetaData!.BloomFilterLength); + + ms.Position = 0; + var stats = new DataColumnStatistics { NullCount = 0 }; + var reader = new DataColumnReader(field, ms, chunk, stats, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + + // No bloom => pruning helper must not prune + Assert.True(reader.MightMatchEquals(9999)); + + ms.Position = 0; + DataColumn rt = await reader.ReadAsync(); + Assert.Equal(values, (int[])rt.Data); + } + + + /// + /// Writes a BYTE_ARRAY (UTF-8 string) column with blooms enabled and validates + /// that probes use raw UTF-8 bytes (no length prefix) and pruning works. + /// + [Fact] + public async Task ByteArray_String_Bloom_EndToEnd() { + // 1) Field + attach to schema + DataField field = STRING(); + var schema = new Parquet.Schema.ParquetSchema(field); + + // 2) Footer with total row count + string[] data = { "parquet", "bloom", "filter" }; + var footer = new ThriftFooter(schema, totalRowCount: data.Length); + + // 3) Physical schema element + var se = new SchemaElement { + Name = field.Name, + Type = Parquet.Meta.Type.BYTE_ARRAY, + RepetitionType = FieldRepetitionType.REQUIRED + }; + + // 4) Now it's safe to build the DataColumn + DataColumn col = new DataColumn(field, data); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter( + ms, footer, se, + compressionMethod: CompressionMethod.None, + options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + compressionLevel: System.IO.Compression.CompressionLevel.NoCompression, + keyValueMetadata: null); + + ColumnChunk chunk = await writer.WriteAsync(new FieldPath(field.Name), col); + + Assert.NotNull(chunk.MetaData!.BloomFilterOffset); + Assert.True(chunk.MetaData!.BloomFilterLength!.Value > 0); + + // Load bloom and probe exact UTF-8 bytes + ms.Position = 0; + SplitBlockBloomFilter bloom = + BloomFilterIO.ReadFromStream(ms, chunk.MetaData!, s => new ThriftCompactProtocolReader(s)); + Assert.True(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("parquet"))); + Assert.False(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("def-not-present"))); + + // Reader pruning + full read + ms.Position = 0; + var stats = new DataColumnStatistics { NullCount = 0, DistinctCount = 3 }; + var reader = new DataColumnReader(field, ms, chunk, stats, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + Assert.True(reader.MightMatchEquals("bloom")); + Assert.False(reader.MightMatchEquals("nope")); + + ms.Position = 0; + DataColumn rt = await reader.ReadAsync(); + Assert.Equal(data, (string[])rt.Data); + } + + [Fact] + public async Task WriteRead_WithBloom_MultipleRowGroups() { + // Single INT32 column, multiple row groups, bloom enabled + var id = new DataField("id"); + using var ms = new MemoryStream(); + + // WRITE + using(ParquetWriter writer = await ParquetWriter.CreateAsync( + new ParquetSchema(id), + ms, + new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { id.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + })) { + using(ParquetRowGroupWriter rg = writer.CreateRowGroup()) { + await rg.WriteColumnAsync(new DataColumn(id, new[] { 10, 11, 12 })); + } + + using(ParquetRowGroupWriter rg = writer.CreateRowGroup()) { + await rg.WriteColumnAsync(new DataColumn(id, new[] { 20, 21 })); + } + + using(ParquetRowGroupWriter rg = writer.CreateRowGroup()) { + await rg.WriteColumnAsync(new DataColumn(id, new[] { 30, 31, 32, 33 })); + } + } + + // READ + ms.Position = 0; + using(ParquetReader reader = await ParquetReader.CreateAsync( + ms, + new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { id.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + })) { + Assert.Equal(3, reader.RowGroupCount); + + DataField datafield = reader.Schema.GetDataFields()[0]; + + using(ParquetRowGroupReader rg0 = reader.OpenRowGroupReader(0)) { + Assert.Equal(3, rg0.RowCount); + Assert.True(rg0.MightMatchEquals(id, 10)); + Assert.False(rg0.MightMatchEquals(id, 9999)); + DataColumn dc = await rg0.ReadColumnAsync(id); + Assert.Equal(new[] { 10, 11, 12 }, (int[])dc.Data); + } + + using(ParquetRowGroupReader rg1 = reader.OpenRowGroupReader(1)) { + Assert.Equal(2, rg1.RowCount); + Assert.True(rg1.MightMatchEquals(id, 21)); + Assert.False(rg1.MightMatchEquals(id, -1)); + DataColumn dc = await rg1.ReadColumnAsync(id); + Assert.Equal(new[] { 20, 21 }, (int[])dc.Data); + } + + using(ParquetRowGroupReader rg2 = reader.OpenRowGroupReader(2)) { + Assert.Equal(4, rg2.RowCount); + Assert.True(rg2.MightMatchEquals(id, 32)); + Assert.False(rg2.MightMatchEquals(id, 0)); + DataColumn dc = await rg2.ReadColumnAsync(id); + Assert.Equal(new[] { 30, 31, 32, 33 }, (int[])dc.Data); + } + } + } + + [Fact] + public async Task Bloom_Metadata_Indicates_BloomFilter() { + await using Stream stream = this.OpenTestFile("bloom.parquet"); + using ParquetReader reader = await ParquetReader.CreateAsync(stream); + + var idField = (DataField)reader.Schema.GetDataFields().Single(f => f.Name == "id"); + + bool anyHasBloom = false; + for(int rg = 0; rg < reader.RowGroupCount; rg++) { + using ParquetRowGroupReader rgr = reader.OpenRowGroupReader(rg); + ColumnChunk? cc = rgr.GetMetadata(idField); + Assert.NotNull(cc); + if(cc!.MetaData!.BloomFilterOffset.HasValue && cc.MetaData.BloomFilterOffset.Value > 0) { + anyHasBloom = true; + } + } + + Assert.True(anyHasBloom); // at least one RG should have a BF written for 'id' + } + + [Fact] + public async Task Bloom_Prunes_Negatives_For_Id() { + + await using Stream stream = this.OpenTestFile("bloom.parquet"); + using ParquetReader reader = await ParquetReader.CreateAsync(stream); + + var idField = (DataField)reader.Schema.GetDataFields().Single(f => f.Name == "id"); + + int totalGroups = reader.RowGroupCount; + int prunedGroups = 0; + + // Try a handful of clearly-absent values + string[] negatives = Enumerable.Range(0, 8).Select(i => $"nope-{i:00000000}").ToArray(); + + for(int rgIndex = 0; rgIndex < totalGroups; rgIndex++) { + using ParquetRowGroupReader rg = reader.OpenRowGroupReader(rgIndex); + + // if ANY negative probes return false, this RG can be pruned + bool thisGroupDefinitelyNo = false; + foreach(string? val in negatives) { + if(!rg.MightMatchEquals(idField, val)) { + thisGroupDefinitelyNo = true; + break; + } + } + if(thisGroupDefinitelyNo) + prunedGroups++; + } + + // We expect at least SOME pruning (don’t over-assert due to possible FPs) + Assert.True(prunedGroups > 0); + Assert.InRange(prunedGroups, 1, totalGroups); + } + + [Fact] + public async Task Bloom_Probe_DoesNotMoveStream() { + await using Stream stream = this.OpenTestFile("bloom.parquet"); + using ParquetReader reader = await ParquetReader.CreateAsync(stream); + + var idField = (DataField)reader.Schema.GetDataFields().Single(f => f.Name == "id"); + + using ParquetRowGroupReader rg = reader.OpenRowGroupReader(0); + + long before = stream.CanSeek ? stream.Position : 0; + _ = rg.MightMatchEquals(idField, $"nope-{42:00000000}"); + long after = stream.CanSeek ? stream.Position : 0; + + if(stream.CanSeek) + Assert.Equal(before, after); + } + + [Fact] + public async Task Bloom_MightMatch_TruePositive_OnExistingId() { + await using Stream stream = this.OpenTestFile("bloom.parquet"); + using ParquetReader reader = await ParquetReader.CreateAsync(stream); + + var idField = (DataField)reader.Schema.GetDataFields().Single(f => f.Name == "id"); + + bool anyPossible = false; + + for(int rgIndex = 0; rgIndex < reader.RowGroupCount; rgIndex++) { + using ParquetRowGroupReader rg = reader.OpenRowGroupReader(rgIndex); + // pick a very likely existing id from your generator + if(rg.MightMatchEquals(idField, "user-12345")) { + anyPossible = true; + break; + } + } + + Assert.True(anyPossible); // Bloom must not rule out a real value + } + + // ---------- helpers ---------- + + private static byte[] PlainLE(int v) { + byte[] b = BitConverter.GetBytes(v); + if(!BitConverter.IsLittleEndian) + Array.Reverse(b); + return b; + } + } +} \ No newline at end of file diff --git a/src/Parquet.Test/Bloom/Bloom_SpecCompliance_Test.cs b/src/Parquet.Test/Bloom/Bloom_SpecCompliance_Test.cs new file mode 100644 index 00000000..ec2913ac --- /dev/null +++ b/src/Parquet.Test/Bloom/Bloom_SpecCompliance_Test.cs @@ -0,0 +1,293 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Threading.Tasks; +using Parquet.Bloom; +using Parquet.Data; +using Parquet.File; +using Parquet.Meta; +using Parquet.Meta.Proto; +using Parquet.Schema; +using Xunit; + +namespace Parquet.Test.Bloom { + /// + /// Spec-compliance tests for Parquet Split-Block Bloom filters. + /// These validate header conformance, hashing (via round-trip), dictionary semantics, + /// reader robustness, and multi-page behavior. + /// + public sealed class Bloom_SpecCompliance_Test { + private static DataField I32(string name = "c") => new DataField(name, System.Type.GetType("System.Int32")!); + private static DataField I64(string name = "l") => new DataField(name, System.Type.GetType("System.Int64")!); + private static DataField F32(string name = "f") => new DataField(name, System.Type.GetType("System.Single")!); + private static DataField F64(string name = "d") => new DataField(name, System.Type.GetType("System.Double")!); + private static DataField STRING(string name = "s") => new DataField(name, System.Type.GetType("System.String")!); + private static DataField NINT32(string name = "n") => new DataField(name, System.Type.GetType("System.Int32")!, isNullable: true); + private static SchemaElement Physical(string name, Parquet.Meta.Type t) => + new SchemaElement { Name = name, Type = t, RepetitionType = FieldRepetitionType.REQUIRED }; + + private static (ThriftFooter footer, FieldPath path, SchemaElement se) SetupFooter( + ParquetSchema schema, DataField field, int totalRows, Parquet.Meta.Type physicalType) { + var footer = new ThriftFooter(schema, totalRowCount: totalRows); + var path = new FieldPath(field.Name); + SchemaElement se = Physical(field.Name, physicalType); + return (footer, path, se); + } + + /// + /// Header conformance: NumBytes multiple of 32; Algorithm.BLOCK; Hash.XXHASH; Compression.UNCOMPRESSED; + /// ColumnMetaData.BloomFilterLength == sizeof(header) + NumBytes and offset points to header. + /// + [Fact] + public async Task Header_Is_Spec_Conformant_On_Write() { + DataField field = I32(); + var schema = new ParquetSchema(field); + int[] values = { 1, 2, 3, 4 }; + + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, values.Length, Parquet.Meta.Type.INT32); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + compressionMethod: CompressionMethod.None, + options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + compressionLevel: System.IO.Compression.CompressionLevel.NoCompression, + keyValueMetadata: null); + + var col = new DataColumn(field, values); + ColumnChunk chunk = await writer.WriteAsync(path, col); + + Assert.NotNull(chunk.MetaData!.BloomFilterOffset); + + long start = chunk.MetaData!.BloomFilterOffset!.Value; + ms.Position = start; + + // Read Bloom header and verify fields per spec + BloomFilterHeader header = BloomFilterHeader.Read(new ThriftCompactProtocolReader(ms)); + + Assert.True(header.NumBytes > 0 && header.NumBytes % 32 == 0); + Assert.NotNull(header.Algorithm.BLOCK); + Assert.NotNull(header.Hash.XXHASH); + Assert.NotNull(header.Compression.UNCOMPRESSED); + + long afterHeader = ms.Position; + int headerSize = (int)(afterHeader - start); + + long payloadStart = ms.Position; + ms.Position += header.NumBytes; + Assert.Equal(payloadStart + header.NumBytes, ms.Position); + } + + /// + /// Dictionary-encoded semantics: Bloom must contain dictionary VALUES, not indexes. + /// Probes for present dictionary values return true; absent value returns false. + /// + [Fact] + public async Task DictionaryEncoded_Bloom_Uses_Dictionary_Values() { + DataField field = STRING(); + var schema = new ParquetSchema(field); + string[] data = { "a", "a", "b", "b", "c", "c" }; + + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, data.Length, Parquet.Meta.Type.BYTE_ARRAY); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + CompressionMethod.None, options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + System.IO.Compression.CompressionLevel.NoCompression, null); + + ColumnChunk chunk = await writer.WriteAsync(path, new DataColumn(field, data)); + + ms.Position = 0; + SplitBlockBloomFilter bloom = BloomFilterIO.ReadFromStream(ms, chunk.MetaData!, s => new ThriftCompactProtocolReader(s)); + Assert.True(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("a"))); + Assert.True(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("b"))); + Assert.True(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("c"))); + Assert.False(bloom.MightContain(System.Text.Encoding.UTF8.GetBytes("z"))); + } + + /// + /// Reader robustness: reject invalid header NumBytes (not multiple of 32). + /// + [Fact] + public void Read_Rejects_Header_With_Invalid_NumBytes() { + using var ms = new MemoryStream(); + var meta = new ColumnMetaData(); + + // Put header at non-zero offset + ms.WriteByte(0xEE); + long offset = ms.Position; + + var bad = new BloomFilterHeader { + NumBytes = 1, // invalid per spec + Algorithm = new BloomFilterAlgorithm { BLOCK = new SplitBlockAlgorithm() }, + Hash = new BloomFilterHash { XXHASH = new XxHash() }, + Compression = new BloomFilterCompression { UNCOMPRESSED = new Uncompressed() } + }; + bad.Write(new ThriftCompactProtocolWriter(ms)); + + meta.BloomFilterOffset = offset; + // intentionally omit bitset + + Assert.Throws(() => + BloomFilterIO.ReadFromStream(ms, meta, s => new ThriftCompactProtocolReader(s))); + } + + /// + /// Multi-page, no-dictionary path: GetFileOffset() is safe (falls back to 0) and + /// Bloom is usable for pruning; data round-trips correctly. + /// + [Fact] + public async Task MultiPage_NoDictionary_Bloom_OK() { + DataField field = I32(); + var schema = new ParquetSchema(field); + + // Make enough rows to plausibly trigger multiple data pages in your writer + int[] values = Enumerable.Range(0, 10).ToArray(); + + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, values.Length, Parquet.Meta.Type.INT32); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + CompressionMethod.None, options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + System.IO.Compression.CompressionLevel.NoCompression, null); + + ColumnChunk chunk = await writer.WriteAsync(path, new DataColumn(field, values)); + + // Reader should not throw on GetFileOffset, even if offsets are 0 + ms.Position = 0; + var stats = new DataColumnStatistics { NullCount = 0 }; + var reader = new DataColumnReader(field, ms, chunk, stats, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + + Assert.True(reader.MightMatchEquals(5)); + Assert.False(reader.MightMatchEquals(9999)); + + ms.Position = 0; + DataColumn rt = await reader.ReadAsync(); + Assert.Equal(values, (int[])rt.Data); + } + + /// + /// Hashing via PLAIN encoding for float/double: insert and probe exact numeric values. + /// (This indirectly verifies raw IEEE-bit hashing without canonicalization.) + /// + [Fact] + public async Task Float_Double_PlainEncoding_RoundTrip_And_Prune() { + // FLOAT + { + DataField field = F32("f32"); + var schema = new ParquetSchema(field); + float[] vals = { 0.0f, -0.0f, 3.1415927f, -123.5f }; + + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, vals.Length, Parquet.Meta.Type.FLOAT); + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + CompressionMethod.None, options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + System.IO.Compression.CompressionLevel.NoCompression, null); + + await writer.WriteAsync(path, new DataColumn(field, vals)); + + ColumnChunk? chunk = footer.AddRowGroup().Columns.LastOrDefault(); // if not returned above; else keep returned chunk + // If your WriteAsync already returns chunk, use that instead of the line above. + + // To keep consistent with WriteAsync return, comment the above two lines and capture the chunk on write if needed. + } + + // DOUBLE + { + DataField field = F64("f64"); + var schema = new ParquetSchema(field); + double[] vals = { 0.0, -0.0, Math.PI, -1.0E300 }; + + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, vals.Length, Parquet.Meta.Type.DOUBLE); + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + CompressionMethod.None, options: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }, + System.IO.Compression.CompressionLevel.NoCompression, null); + + ColumnChunk chunk = await writer.WriteAsync(path, new DataColumn(field, vals)); + + ms.Position = 0; + var reader = new DataColumnReader(field, ms, chunk, + new DataColumnStatistics { NullCount = 0 }, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + + // Probes on present values should be true + Assert.True(reader.MightMatchEquals(Math.PI)); + Assert.True(reader.MightMatchEquals(0.0)); + // Absent should be false (very likely) + Assert.False(reader.MightMatchEquals(42.42)); + + ms.Position = 0; + DataColumn rt = await reader.ReadAsync(); + Assert.Equal(vals, (double[])rt.Data); + } + } + + /// + /// Nulls are not hashed: nullable INT32 column with nulls should not introduce + /// spurious positives for values that only appear as nulls. + /// + [Fact] + public async Task Nullable_Int_Nulls_Are_Not_Hashed() { + DataField field = NINT32(); + var schema = new ParquetSchema(field); + + int?[] vals = { 1, null, 2, null, 3 }; + (ThriftFooter? footer, FieldPath? path, SchemaElement? se) = SetupFooter(schema, field, vals.Length, Parquet.Meta.Type.INT32); + + using var ms = new MemoryStream(); + var writer = new DataColumnWriter(ms, footer, se, + CompressionMethod.None, options: new ParquetOptions { BloomFilterOptionsByColumn = new Dictionary() { { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } } }, + System.IO.Compression.CompressionLevel.NoCompression, null); + + ColumnChunk chunk = await writer.WriteAsync(path, new DataColumn(field, vals)); + + ms.Position = 0; + var reader = new DataColumnReader(field, ms, chunk, + new DataColumnStatistics { NullCount = vals.Count(v => v == null) }, footer, parquetOptions: new ParquetOptions { + BloomFilterOptionsByColumn = new Dictionary() { + { field.Name, new ParquetOptions.BloomFilterOptions { EnableBloomFilters = true } } + } + }); + + // Values present (non-null) + Assert.True(reader.MightMatchEquals(1)); + Assert.True(reader.MightMatchEquals(2)); + Assert.True(reader.MightMatchEquals(3)); + + // Value that only appears as null is not a thing; pick an absent value + Assert.False(reader.MightMatchEquals(9999)); + + ms.Position = 0; + DataColumn rt = await reader.ReadAsync(); + Assert.Equal(vals, (int?[])rt.Data); + } + } +} \ No newline at end of file diff --git a/src/Parquet.Test/Bloom/SplitBlockBloomFilterTest.cs b/src/Parquet.Test/Bloom/SplitBlockBloomFilterTest.cs new file mode 100644 index 00000000..4c12072e --- /dev/null +++ b/src/Parquet.Test/Bloom/SplitBlockBloomFilterTest.cs @@ -0,0 +1,64 @@ +using System; +using System.Text; +using Parquet.Bloom; +using Xunit; + +namespace Parquet.Test.Bloom { + public sealed class SplitBlockBloomFilterTest { + private const int BlockBytes = 32; + + [Fact] + public void InsertAndCheck_WithReadableStrings() { + const int blocks = 4; + SplitBlockBloomFilter f = new SplitBlockBloomFilter(blocks); + + byte[] aliceBytes = Encoding.UTF8.GetBytes("Alice"); + byte[] bobBytes = Encoding.UTF8.GetBytes("Bob"); + byte[] carolBytes = Encoding.UTF8.GetBytes("Carol"); + byte[] daveBytes = Encoding.UTF8.GetBytes("Dave"); + + Assert.False(f.MightContain(aliceBytes)); + Assert.False(f.MightContain(bobBytes)); + + f.Insert(bobBytes); + f.Insert(daveBytes); + + Assert.True(f.MightContain(bobBytes)); + Assert.True(f.MightContain(daveBytes)); + + Assert.False(f.MightContain(aliceBytes)); + Assert.False(f.MightContain(carolBytes)); + } + + [Fact] + public void ToBytes_FromBytes_RoundTrip_WithStrings() { + const int blocks = 3; + SplitBlockBloomFilter f1 = new SplitBlockBloomFilter(blocks); + + string[] words = { "Alpha", "Beta", "Gamma" }; + foreach(string w in words) { + f1.Insert(Encoding.UTF8.GetBytes(w)); + } + + byte[] bytes = f1.ToByteArray(); + Assert.Equal(SplitBlockBloomFilter.BytesForBlocks(blocks), bytes.Length); + + SplitBlockBloomFilter f2 = SplitBlockBloomFilter.FromByteArray(blocks, bytes); + + foreach(string w in words) { + Assert.True(f2.MightContain(Encoding.UTF8.GetBytes(w))); + } + } + + [Fact] + public void BytesForBlocks_And_Bounds() { + Assert.Equal(BlockBytes * 1, SplitBlockBloomFilter.BytesForBlocks(1)); + Assert.Equal(BlockBytes * 2, SplitBlockBloomFilter.BytesForBlocks(2)); + Assert.Equal(BlockBytes * 10, SplitBlockBloomFilter.BytesForBlocks(10)); + + Assert.Throws(() => new SplitBlockBloomFilter(0)); + Assert.Throws(() => SplitBlockBloomFilter.BytesForBlocks(0)); + Assert.Throws(() => SplitBlockBloomFilter.FromByteArray(0, Array.Empty())); + } + } +} diff --git a/src/Parquet.Test/Parquet.Test.csproj b/src/Parquet.Test/Parquet.Test.csproj index ce162215..340c23b8 100644 --- a/src/Parquet.Test/Parquet.Test.csproj +++ b/src/Parquet.Test/Parquet.Test.csproj @@ -19,7 +19,6 @@ - diff --git a/src/Parquet.Test/data/bloom.parquet b/src/Parquet.Test/data/bloom.parquet new file mode 100644 index 00000000..117b9687 Binary files /dev/null and b/src/Parquet.Test/data/bloom.parquet differ diff --git a/src/Parquet/Bloom/BloomCollector.cs b/src/Parquet/Bloom/BloomCollector.cs new file mode 100644 index 00000000..d0c43f53 --- /dev/null +++ b/src/Parquet/Bloom/BloomCollector.cs @@ -0,0 +1,118 @@ +using System; +using System.Buffers.Binary; +using System.Text; + +namespace Parquet.Bloom +{ + /// + /// Incrementally builds a Split-Block Bloom Filter for a single column chunk. + /// Accepts raw values in their physical form and inserts their PLAIN-encoded bytes + /// (as specified by the Parquet bloom filter spec). + /// + internal sealed class BloomCollector : IDisposable { + public SplitBlockBloomFilter Filter { get; } + + public BloomCollector(int blocks) { + if(blocks <= 0) + throw new ArgumentOutOfRangeException(nameof(blocks)); + this.Filter = new SplitBlockBloomFilter(blocks); + } + + public void Dispose() { /* nothing to free */ } + + // ---- Insert helpers for common physical types (PLAIN encoding) ---- + + /// Insert a nullable boolean (PLAIN: 1 byte 0/1). + public void AddBoolean(bool? v) { + if(!v.HasValue) + return; + byte b = v.Value ? (byte)1 : (byte)0; + Filter.Insert(new byte[] { b }); + } + + /// Insert a nullable Int32 (PLAIN little-endian 4 bytes). + public void AddInt32(int? v) { + if(!v.HasValue) + return; + Span buf = stackalloc byte[4]; + BinaryPrimitives.WriteInt32LittleEndian(buf, v.Value); + Filter.Insert(buf.ToArray()); + } + + /// Insert a nullable Int64 (PLAIN little-endian 8 bytes). + public void AddInt64(long? v) { + if(!v.HasValue) + return; + Span buf = stackalloc byte[8]; + BinaryPrimitives.WriteInt64LittleEndian(buf, v.Value); + Filter.Insert(buf.ToArray()); + } + + public void AddInt96(DateTime? v) { + if(!v.HasValue) + return; + + // Parquet INT96 is a 12-byte little-endian value consisting of: + // - first 8 bytes: nanoseconds since midnight (little-endian) + // - next 4 bytes: Julian day (little-endian) + + DateTime dt = v.Value.ToUniversalTime(); + int julianDay = (int)(dt - new DateTime(4713, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalDays + 1721424; // Convert to Julian Day + long nanosSinceMidnight = (long)(dt - dt.Date).TotalMilliseconds * 1_000_000; // Convert to nanoseconds since midnight + + byte[] buf = new byte[12]; + BinaryPrimitives.WriteInt64LittleEndian(buf.AsSpan(0, 8), nanosSinceMidnight); + BinaryPrimitives.WriteInt32LittleEndian(buf.AsSpan(8, 4), julianDay); + + Filter.Insert(buf); + } + + /// Insert a nullable float (PLAIN little-endian 4 bytes, IEEE 754). + public void AddFloat(float? v) { + if(!v.HasValue) + return; + + byte[] buf = new byte[4]; + Buffer.BlockCopy(new float[] { v.Value }, 0, buf, 0, 4); + if(!BitConverter.IsLittleEndian) { + Array.Reverse(buf); + } + Filter.Insert(buf); + } + + /// Insert a nullable double (PLAIN little-endian 8 bytes, IEEE 754). + public void AddDouble(double? v) { + if(!v.HasValue) + return; + + byte[] buf = new byte[8]; + Buffer.BlockCopy(new double[] { v.Value }, 0, buf, 0, 8); + if(!BitConverter.IsLittleEndian) { + Array.Reverse(buf); + } + Filter.Insert(buf); + } + + /// Insert a UTF-8 string (PLAIN: length prefix is NOT included for bloom hashing). + public void AddString(string? s) { + if(s == null) + return; + byte[] utf8 = Encoding.UTF8.GetBytes(s); + Filter.Insert(utf8); + } + + /// Insert a BYTE_ARRAY slice (PLAIN hashing uses the bytes only; no length prefix). + public void AddByteArray(byte[]? bytes) { + if(bytes == null) + return; + Filter.Insert(bytes); + } + + /// Insert FIXED_LEN_BYTE_ARRAY (bytes as-is). + public void AddFixed(byte[]? bytes) { + if(bytes == null) + return; + Filter.Insert(bytes); + } + } +} \ No newline at end of file diff --git a/src/Parquet/Bloom/BloomFilterIO.cs b/src/Parquet/Bloom/BloomFilterIO.cs new file mode 100644 index 00000000..88b42ab2 --- /dev/null +++ b/src/Parquet/Bloom/BloomFilterIO.cs @@ -0,0 +1,101 @@ +using System; +using System.IO; +using Parquet.Meta; +using Parquet.Meta.Proto; + +namespace Parquet.Bloom { + /// + /// Utilities to write/read Bloom Filter header + bitset and hook + /// them into ColumnMetaData.{BloomFilterOffset,BloomFilterLength}. + /// + internal static class BloomFilterIO { + public static (long Offset, int Length) WriteToStream( + Stream output, + SplitBlockBloomFilter filter, + ColumnMetaData columnMeta, + Func writerFactory) { + if(output == null) + throw new ArgumentNullException("output"); + if(filter == null) + throw new ArgumentNullException("filter"); + if(columnMeta == null) + throw new ArgumentNullException("columnMeta"); + if(writerFactory == null) + throw new ArgumentNullException("writerFactory"); + if(!output.CanWrite) + throw new InvalidOperationException("Output stream not writable."); + + long offset = output.Position; + + var hdr = new BloomFilterHeader { + NumBytes = filter.NumberOfBlocks * 32, + Algorithm = new BloomFilterAlgorithm { BLOCK = new SplitBlockAlgorithm() }, + Hash = new BloomFilterHash { XXHASH = new XxHash() }, + Compression = new BloomFilterCompression { UNCOMPRESSED = new Uncompressed() } + }; + + ThriftCompactProtocolWriter proto = writerFactory(output); + hdr.Write(proto); + + byte[] bitset = filter.ToByteArray(); + output.Write(bitset, 0, bitset.Length); + + int length = checked((int)(output.Position - offset)); + + columnMeta.BloomFilterOffset = offset; + columnMeta.BloomFilterLength = length; + + return (offset, length); + } + + /// + /// Reads a BloomFilterHeader and its bitset using ColumnMetaData.{BloomFilterOffset,BloomFilterLength} + /// and reconstructs a . + /// + public static SplitBlockBloomFilter ReadFromStream( + Stream input, + ColumnMetaData columnMeta, + Func readerFactory) { + if(input == null) + throw new ArgumentNullException(nameof(input)); + if(columnMeta == null) + throw new ArgumentNullException(nameof(columnMeta)); + if(readerFactory == null) + throw new ArgumentNullException(nameof(readerFactory)); + if(!input.CanRead) + throw new InvalidOperationException("Input stream not readable."); + if(!input.CanSeek) + throw new InvalidOperationException("Input stream must be seekable to read bloom filter."); + if(!columnMeta.BloomFilterOffset.HasValue) + throw new InvalidOperationException("ColumnMetaData does not contain BloomFilterOffset."); + + long offset = columnMeta.BloomFilterOffset.Value; + input.Seek(offset, SeekOrigin.Begin); + + ThriftCompactProtocolReader proto = readerFactory(input); + BloomFilterHeader hdr = BloomFilterHeader.Read(proto); + + // Validate header + if(hdr == null) + throw new InvalidDataException("Missing bloom filter header."); + if(hdr.Algorithm?.BLOCK == null) + throw new NotSupportedException("Unsupported bloom filter algorithm (only BLOCK is supported)."); + if(hdr.Hash?.XXHASH == null) + throw new NotSupportedException("Unsupported bloom filter hash (only XXHASH is supported)."); + if(hdr.Compression?.UNCOMPRESSED == null) + throw new NotSupportedException("Unsupported bloom filter compression (only UNCOMPRESSED is supported)."); + if(hdr.NumBytes <= 0 || (hdr.NumBytes % 32) != 0) + throw new InvalidDataException("Invalid bloom filter header: NumBytes must be positive and a multiple of 32."); + + // Read raw bitset immediately following the header + int numBytes = hdr.NumBytes; + byte[] data = new byte[numBytes]; + int read = input.Read(data, 0, numBytes); + if(read != numBytes) + throw new EndOfStreamException("Could not read bloom filter bitset."); + + int blocks = numBytes / 32; + return SplitBlockBloomFilter.FromByteArray(blocks, data); + } + } +} \ No newline at end of file diff --git a/src/Parquet/Bloom/BloomHasher.cs b/src/Parquet/Bloom/BloomHasher.cs new file mode 100644 index 00000000..6b7ed563 --- /dev/null +++ b/src/Parquet/Bloom/BloomHasher.cs @@ -0,0 +1,30 @@ +using System; +using System.Buffers.Binary; +using System.IO.Hashing; + +namespace Parquet.Bloom { + internal static class BloomHasher { + + /// + /// Hash a raw span (already PLAIN-encoded). + /// + public static ulong HashPlainEncoded(ReadOnlySpan data) => HashToU64(data); + + /// + /// Hash a raw slice of a byte[] (already PLAIN-encoded). + /// + public static ulong HashPlainEncoded(byte[] buffer, int offset, int length) { + if(buffer is null) + throw new ArgumentNullException(nameof(buffer)); + return HashPlainEncoded(new ReadOnlySpan(buffer, offset, length)); + } + + private static ulong ToU64BE(ReadOnlySpan hash8) + => BinaryPrimitives.ReadUInt64BigEndian(hash8); + + private static ulong HashToU64(ReadOnlySpan data) { + byte[] h = XxHash64.Hash(data, seed: 0); + return ToU64BE(h); + } + } +} diff --git a/src/Parquet/Bloom/BloomPruning.cs b/src/Parquet/Bloom/BloomPruning.cs new file mode 100644 index 00000000..3bb83776 --- /dev/null +++ b/src/Parquet/Bloom/BloomPruning.cs @@ -0,0 +1,136 @@ +using System; +using Parquet.Meta; +using Encoding = System.Text.Encoding; + +namespace Parquet.Bloom { + /// + /// Helper to probe SplitBlock Bloom Filters for equality predicates. + /// Encodes values per Parquet's PLAIN rules (no length prefix for variable-size types). + /// + internal static class BloomPruning { + /// + /// Returns false if the value is definitely not present in the column chunk, + /// true if it might be present (or filter is unavailable/unsupported). + /// + public static bool MightMatchEquals(object? value, SchemaElement physical, SplitBlockBloomFilter? bloom) { + if(value is null) + return true; // blooms don't index nulls; can't rule-in/out nulls + if(bloom is null) + return true; // no bloom -> can't prune + + switch(physical.Type!.Value) { + case Parquet.Meta.Type.BOOLEAN: + return bloom.MightContain(new[] { ((bool)value) ? (byte)1 : (byte)0 }); + + case Parquet.Meta.Type.INT32: + return bloom.MightContain(PlainLE.Int32((int)Convert.ChangeType(value, typeof(int)))); + + case Parquet.Meta.Type.INT64: + return bloom.MightContain(PlainLE.Int64((long)Convert.ChangeType(value, typeof(long)))); + + case Parquet.Meta.Type.FLOAT: + return bloom.MightContain(PlainLE.Single((float)Convert.ChangeType(value, typeof(float)))); + + case Parquet.Meta.Type.DOUBLE: + return bloom.MightContain(PlainLE.Double((double)Convert.ChangeType(value, typeof(double)))); + + case Parquet.Meta.Type.BYTE_ARRAY: { + if(value is byte[] bytes) + return bloom.MightContain(bytes); + if(value is string s) + return bloom.MightContain(Encoding.UTF8.GetBytes(s)); + throw new NotSupportedException("BYTE_ARRAY equality requires byte[] or string literal."); + } + + case Parquet.Meta.Type.FIXED_LEN_BYTE_ARRAY: { + if(value is byte[] bytes) + return bloom.MightContain(bytes); + throw new NotSupportedException("FIXED_LEN_BYTE_ARRAY equality requires byte[] literal."); + } + + case Parquet.Meta.Type.INT96: { + if(value is byte[] raw12) + return bloom.MightContain(raw12); + + if(value is DateTime dt) + return bloom.MightContain(PlainLE.Int96(dt)); + + if(value is DateTimeOffset dto) + return bloom.MightContain(PlainLE.Int96(dto.UtcDateTime)); + + throw new NotSupportedException("INT96 equality requires byte[], DateTime, or DateTimeOffset."); + } + + default: + return true; + } + } + + private static class PlainLE { + public static byte[] Int32(int v) { + byte[] b = BitConverter.GetBytes(v); + if(!BitConverter.IsLittleEndian) + Array.Reverse(b); + return b; + } + + public static byte[] Int64(long v) { + byte[] b = BitConverter.GetBytes(v); + if(!BitConverter.IsLittleEndian) + Array.Reverse(b); + return b; + } + + public static byte[] Single(float v) { + byte[] b = new byte[4]; + Buffer.BlockCopy(new[] { v }, 0, b, 0, 4); + if(!BitConverter.IsLittleEndian) + Array.Reverse(b); + return b; + } + + public static byte[] Double(double v) { + byte[] b = new byte[8]; + Buffer.BlockCopy(new[] { v }, 0, b, 0, 8); + if(!BitConverter.IsLittleEndian) + Array.Reverse(b); + return b; + } + + public static byte[] Int96(DateTime utcOrLocal) { + // INT96 is (nanoseconds of day [8 bytes]) + (Julian day [4 bytes]), little-endian. + // Treat timestamps as UTC (INT96 is timezone-agnostic; most writers used UTC). + DateTime dtUtc = utcOrLocal.Kind == DateTimeKind.Utc ? utcOrLocal : utcOrLocal.ToUniversalTime(); + + int julianDay = ToJulianDay(dtUtc.Year, dtUtc.Month, dtUtc.Day); + long nanosOfDay = (dtUtc - dtUtc.Date).Ticks * 100L; // 1 tick = 100 ns + + byte[] buf = new byte[12]; + // write nanosOfDay (8 LE) + byte[] n = BitConverter.GetBytes(nanosOfDay); + // write julianDay (4 LE) + byte[] j = BitConverter.GetBytes(julianDay); + + if(!BitConverter.IsLittleEndian) { Array.Reverse(n); Array.Reverse(j); } + + Buffer.BlockCopy(n, 0, buf, 0, 8); + Buffer.BlockCopy(j, 0, buf, 8, 4); + return buf; + } + + private static int ToJulianDay(int year, int month, int day) { + // Proleptic Gregorian calendar → Julian Day Number (integer) + int a = (14 - month) / 12; + int y = year + 4800 - a; + int m = month + (12 * a) - 3; + return day + + (((153 * m) + 2) / 5) + + (365 * y) + + (y / 4) + - (y / 100) + + (y / 400) + - 32045; + } + } + } +} \ No newline at end of file diff --git a/src/Parquet/Bloom/BloomSizing.cs b/src/Parquet/Bloom/BloomSizing.cs new file mode 100644 index 00000000..175c56bc --- /dev/null +++ b/src/Parquet/Bloom/BloomSizing.cs @@ -0,0 +1,146 @@ +using System; + +namespace Parquet.Bloom { + /// + /// Sizing helpers for Split-Block Bloom Filters (SBBF). + /// Maps a target false-positive probability (FPP) or explicit bits-per-value (BPV) + /// to number of 256-bit blocks (z) and total bytes. + /// + public static class BloomSizing { + /// + /// Result of sizing the bloom filter for a column chunk / value set. + /// + public sealed class BloomPlan { + /// Number of 256-bit blocks (must be >= 1). + public int Blocks { get; set; } + + /// Total bytes of the bitset (= Blocks * 32). + public int NumBytes { get; set; } + + /// The bits-per-value used for this plan (derived or overridden). + public double BitsPerValue { get; set; } + + /// The FPP target that was requested (if provided), otherwise null. + public double? TargetFpp { get; set; } + + /// Estimated distinct values that were used in planning. + public long EstimatedDistinctValues { get; set; } + } + + /// + /// Create a plan from either a target false-positive probability (FPP) or + /// an explicit bits-per-value. If both are provided, BitsPerValue wins. + /// + /// Estimated number of distinct, non-null values to insert (must be >= 0). + /// Desired false-positive probability (0 < FPP < 1), or null. + /// Explicit bits per value (must be > 0) if provided; overrides targetFpp. + /// BloomPlan containing block count and bytes. + public static BloomPlan Plan(long estimatedDistinctValues, double? targetFpp, double? bitsPerValueOverride) { + if(estimatedDistinctValues < 0L) { + throw new ArgumentOutOfRangeException("estimatedDistinctValues", "Must be >= 0."); + } + if(bitsPerValueOverride.HasValue && bitsPerValueOverride.Value <= 0.0) { + throw new ArgumentOutOfRangeException("bitsPerValueOverride", "Must be > 0 when provided."); + } + if(targetFpp.HasValue) { + if(!(targetFpp.Value > 0.0 && targetFpp.Value < 1.0)) { + throw new ArgumentOutOfRangeException("targetFpp", "FPP must be in (0,1) when provided."); + } + } + + // Choose bits-per-value + double bpv; + if(bitsPerValueOverride.HasValue) { + bpv = bitsPerValueOverride.Value; + } else { + bpv = BloomSizing.BitsPerValueForFpp(targetFpp); + } + + // Compute total bits and blocks. Each block is 256 bits. + // Always allocate at least one block so we have a well-formed filter, + // even for empty columns (z >= 1). + long totalBitsLong = BloomSizing.CeilToLong(estimatedDistinctValues * bpv); + long blocksLong = BloomSizing.CeilDiv(totalBitsLong, 256L); + if(blocksLong < 1L) { + blocksLong = 1L; + } + if(blocksLong >= (1L << 31)) { + throw new OverflowException("Block count would exceed 2^31 - 1 per Parquet SBBF limit."); + } + + int blocks = (int)blocksLong; + int bytes = checked(blocks * 32); + + BloomPlan plan = new BloomPlan(); + plan.Blocks = blocks; + plan.NumBytes = bytes; + plan.BitsPerValue = bpv; + plan.TargetFpp = targetFpp; + plan.EstimatedDistinctValues = estimatedDistinctValues; + return plan; + } + + /// + /// Map a desired false-positive probability to a bits-per-value budget for SBBF. + /// This uses practical thresholds commonly used with Split-Block Bloom Filters. + /// If no FPP is provided, defaults to ~1% (10.5 bits/value). + /// + public static double BitsPerValueForFpp(double? targetFpp) { + if(!targetFpp.HasValue) { + return 10.5; // default: ~1% + } + + double fpp = targetFpp.Value; + + // Piecewise thresholds (approximate, conservative): + // ~10% => 6.0 bits/value + // ~1% => 10.5 bits/value + // ~0.1% => 16.9 bits/value + // ~0.01%=> 26.4 bits/value + // ~0.001%=>41.0 bits/value + // + // We choose the smallest bpv meeting (<=) the requested FPP. + if(fpp <= 0.00001) // 0.001% + { + return 41.0; + } + if(fpp <= 0.0001) // 0.01% + { + return 26.4; + } + if(fpp <= 0.001) // 0.1% + { + return 16.9; + } + if(fpp <= 0.01) // 1% + { + return 10.5; + } + // Up to ~10% + return 6.0; + } + + private static long CeilDiv(long numerator, long denominator) { + if(denominator <= 0L) { + throw new ArgumentOutOfRangeException("denominator"); + } + if(numerator <= 0L) { + return 0L; + } + long q = numerator / denominator; + long r = numerator % denominator; + return r == 0L ? q : q + 1L; + } + + private static long CeilToLong(double value) { + if(value <= 0.0) { + return 0L; + } + double c = Math.Ceiling(value); + if(c > (double)long.MaxValue) { + throw new OverflowException("Value too large."); + } + return (long)c; + } + } +} \ No newline at end of file diff --git a/src/Parquet/Bloom/SplitBlockBloomFilter.cs b/src/Parquet/Bloom/SplitBlockBloomFilter.cs new file mode 100644 index 00000000..95825551 --- /dev/null +++ b/src/Parquet/Bloom/SplitBlockBloomFilter.cs @@ -0,0 +1,254 @@ +using System; + +namespace Parquet.Bloom { + /// + /// Split-Block Bloom Filter per Parquet spec. + /// - z blocks; each block is 8 x 32-bit words = 256 bits. + /// - Block selection uses the high 32 bits of the 64-bit hash (multiply-shift). + /// - Bit selection in each 32-bit word uses the low 32 bits of the hash with 8 salts. + /// + public sealed class SplitBlockBloomFilter { + // Fixed salts (unsigned 32-bit). + /// + /// Fixed salts used for bit selection in each 32-bit word of the block. + /// + public static readonly uint[] Salts = new uint[8] + { + 0x47b6137bU, 0x44974d91U, 0x8824ad5bU, 0xa2b7289dU, + 0x705495c7U, 0x2df1424bU, 0x9efc4947U, 0x5c6bfb31U + }; + + private readonly uint[] _words; + + /// + /// Gets the number of blocks in the Bloom filter. + /// + public int NumberOfBlocks { get; private set; } + + /// + /// Initializes a new instance of the class with the specified number of blocks. + /// + /// The number of blocks in the Bloom filter. Must be greater than or equal to 1. + /// Thrown when is less than or equal to 0. + public SplitBlockBloomFilter(int numberOfBlocks) { + if(numberOfBlocks <= 0) { + throw new ArgumentOutOfRangeException("numberOfBlocks", "numberOfBlocks must be >= 1."); + } + + this.NumberOfBlocks = numberOfBlocks; + this._words = new uint[numberOfBlocks * 8]; + } + + /// + /// Insert a value given as a byte array. The bytes are hashed with XXH64 (seed = 0), + /// then mapped to a block and per-word bit positions per the Split-Block algorithm. + /// + public void Insert(byte[] data) { + if(data == null) { + throw new ArgumentNullException("data"); + } + ulong h = BloomHasher.HashPlainEncoded(data); + this.Insert(h); + } + + /// + /// Insert a slice of a byte array (offset/length). The slice is hashed with XXH64 (seed = 0). + /// + public void Insert(byte[] data, int offset, int length) { + if(data == null) { + throw new ArgumentNullException("data"); + } + if(offset < 0 || length < 0 || offset + length > data.Length) { + throw new ArgumentOutOfRangeException("offset/length"); + } + + ulong h = BloomHasher.HashPlainEncoded(data, offset, length); + this.Insert(h); + } + + /// + /// Checks membership for a byte array by hashing it with XXH64 (seed = 0) and probing the filter. + /// + public bool MightContain(byte[] data) { + if(data == null) { + throw new ArgumentNullException("data"); + } + ulong h = BloomHasher.HashPlainEncoded(data); + return this.MightContain(h); + } + + /// + /// Checks membership for a slice of a byte array (offset/length), hashing with XXH64 (seed = 0). + /// + public bool MightContain(byte[] data, int offset, int length) { + if(data == null) { + throw new ArgumentNullException("data"); + } + if(offset < 0 || length < 0 || offset + length > data.Length) { + throw new ArgumentOutOfRangeException("offset/length"); + } + + ulong h = BloomHasher.HashPlainEncoded(data, offset, length); + return this.MightContain(h); + } + + /// + /// Insert a 64-bit hash value. + /// High 32 bits choose the block; low 32 bits choose one bit in each 32-bit word of the block. + /// + public void Insert(ulong hash64) { + int blockIndex = SplitBlockBloomFilter.MapHashToBlock(hash64, this.NumberOfBlocks); + uint x = (uint)(hash64 & 0xFFFFFFFFUL); + this.BlockInsert(blockIndex, x); + } + + /// + /// Membership check for a 64-bit hash value. + /// + public bool MightContain(ulong hash64) { + int blockIndex = SplitBlockBloomFilter.MapHashToBlock(hash64, this.NumberOfBlocks); + uint x = (uint)(hash64 & 0xFFFFFFFFUL); + return this.BlockCheck(blockIndex, x); + } + + /// + /// Map 64-bit hash to a block index using multiply-shift: ((h >> 32) * z) >> 32 + /// + public static int MapHashToBlock(ulong hash64, int numberOfBlocks) { + if(numberOfBlocks <= 0) { + throw new ArgumentOutOfRangeException("numberOfBlocks", "numberOfBlocks must be >= 1."); + } + + ulong high = hash64 >> 32; + ulong product = high * (ulong)numberOfBlocks; + int index = (int)(product >> 32); + return index; + } + + /// + /// Returns how many bytes are needed for the given number of blocks. + /// + public static int BytesForBlocks(int numberOfBlocks) { + if(numberOfBlocks <= 0) { + throw new ArgumentOutOfRangeException("numberOfBlocks", "numberOfBlocks must be >= 1."); + } + + return numberOfBlocks * 32; + } + + /// + /// Serialize the bitset to a little-endian byte array (8 uints per block). + /// + public byte[] ToByteArray() { + int len = this._words.Length; + byte[] bytes = new byte[len * 4]; + + int o = 0; + for(int i = 0; i < len; i++) { + uint w = this._words[i]; + // Little-endian layout: + bytes[o + 0] = (byte)(w & 0xFF); + bytes[o + 1] = (byte)((w >> 8) & 0xFF); + bytes[o + 2] = (byte)((w >> 16) & 0xFF); + bytes[o + 3] = (byte)((w >> 24) & 0xFF); + o += 4; + } + + return bytes; + } + + /// + /// Create a filter from a byte array previously produced by ToByteArray(). + /// + public static SplitBlockBloomFilter FromByteArray(int numberOfBlocks, byte[] bytes) { + if(numberOfBlocks <= 0) { + throw new ArgumentOutOfRangeException("numberOfBlocks", "numberOfBlocks must be >= 1."); + } + if(bytes == null) { + throw new ArgumentNullException("bytes"); + } + int expected = numberOfBlocks * 32; + if(bytes.Length != expected) { + throw new ArgumentException("Byte array length does not match numberOfBlocks * 32.", "bytes"); + } + + SplitBlockBloomFilter f = new SplitBlockBloomFilter(numberOfBlocks); + int len = f._words.Length; + + int o = 0; + for(int i = 0; i < len; i++) { + uint w = + (uint)bytes[o + 0] | + ((uint)bytes[o + 1] << 8) | + ((uint)bytes[o + 2] << 16) | + ((uint)bytes[o + 3] << 24); + f._words[i] = w; + o += 4; + } + + return f; + } + + /// + /// Public helper for testing: returns the 8 mask words for a given low-32-bit x. + /// Each mask has exactly one bit set. + /// + public static uint[] ComputeMasksFor(uint x) { + uint[] masks = new uint[8]; + for(int i = 0; i < 8; i++) { + uint m = unchecked(x * SplitBlockBloomFilter.Salts[i]); + int bit = (int)(m >> 27); // top 5 bits select [0..31] + uint mask = 1U << bit; + masks[i] = mask; + } + return masks; + } + + /// + /// For testing/inspection: returns a copy of the 8 words for a given block index. + /// + public uint[] GetBlockWordsCopy(int blockIndex) { + SplitBlockBloomFilter.ValidateBlockIndex(blockIndex, this.NumberOfBlocks); + uint[] copy = new uint[8]; + int baseIndex = blockIndex * 8; + for(int i = 0; i < 8; i++) { + copy[i] = this._words[baseIndex + i]; + } + return copy; + } + + private static void ValidateBlockIndex(int blockIndex, int numberOfBlocks) { + if(blockIndex < 0 || blockIndex >= numberOfBlocks) { + throw new ArgumentOutOfRangeException("blockIndex", "blockIndex out of range."); + } + } + + private void BlockInsert(int blockIndex, uint x) { + SplitBlockBloomFilter.ValidateBlockIndex(blockIndex, this.NumberOfBlocks); + int baseIndex = blockIndex * 8; + + for(int i = 0; i < 8; i++) { + uint m = unchecked(x * SplitBlockBloomFilter.Salts[i]); + int bit = (int)(m >> 27); + uint mask = 1U << bit; + this._words[baseIndex + i] |= mask; + } + } + + private bool BlockCheck(int blockIndex, uint x) { + SplitBlockBloomFilter.ValidateBlockIndex(blockIndex, this.NumberOfBlocks); + int baseIndex = blockIndex * 8; + + for(int i = 0; i < 8; i++) { + uint m = unchecked(x * SplitBlockBloomFilter.Salts[i]); + int bit = (int)(m >> 27); + uint mask = 1U << bit; + if((this._words[baseIndex + i] & mask) == 0U) { + return false; + } + } + + return true; + } + } +} \ No newline at end of file diff --git a/src/Parquet/File/DataColumnReader.cs b/src/Parquet/File/DataColumnReader.cs index dd38789f..1e05a355 100644 --- a/src/Parquet/File/DataColumnReader.cs +++ b/src/Parquet/File/DataColumnReader.cs @@ -11,6 +11,7 @@ using Parquet.Meta; using Parquet.Meta.Proto; using Parquet.Extensions; +using Parquet.Bloom; namespace Parquet.File { @@ -25,6 +26,8 @@ class DataColumnReader { private readonly ThriftFooter _footer; private readonly ParquetOptions _options; private readonly DataColumnStatistics? _stats; + private SplitBlockBloomFilter? _bloom; + private bool _bloomLoaded; internal DataColumnReader( DataField dataField, @@ -45,6 +48,19 @@ internal DataColumnReader( _schemaElement = _footer.GetSchemaElement(_thriftColumnChunk); } + /// + /// Returns false if the value is definitely not present in this column chunk + /// (per its Split-Block Bloom filter). Returns true if it might be present, + /// or if pruning is unavailable (no bloom / unsupported type). + /// Use this before reading to skip I/O for obvious misses. + /// + internal bool MightMatchEquals(object? value) { + if(_schemaElement == null) + return true; // unknown type -> don't prune + EnsureBloomLoaded(); + return BloomPruning.MightMatchEquals(value, _schemaElement, _bloom); + } + /// /// Return data column statistics /// @@ -149,14 +165,14 @@ private async ValueTask ReadDictionaryPage(PageHeader ph, PackedColumn pc) { } private long GetFileOffset() => - // get the minimum offset, we'll just read pages in sequence as DictionaryPageOffset/Data_page_offset are not reliable new[] { - _thriftColumnChunk.MetaData?.DictionaryPageOffset ?? 0, - _thriftColumnChunk.MetaData!.DataPageOffset - } - .Where(e => e != 0) - .Min(); + _thriftColumnChunk.MetaData?.DictionaryPageOffset ?? 0, + _thriftColumnChunk.MetaData?.DataPageOffset ?? 0 + } + .Where(e => e != 0) + .DefaultIfEmpty(0) + .Min(); private async Task ReadDataPageV1Async(PageHeader ph, PackedColumn pc) { using IronCompress.IronCompressResult bytes = await ReadPageDataAsync(ph); @@ -370,5 +386,37 @@ private static int ReadRleDictionary(Span s, int maxReadCount, Span d return destOffset - start; } + + /// + /// Lazily loads the bloom filter from the column chunk if present. + /// Saves and restores the current stream position. + /// + private void EnsureBloomLoaded() { + if(_bloomLoaded) + return; + _bloomLoaded = true; + + ColumnMetaData? meta = _thriftColumnChunk.MetaData; + if(meta == null || !meta.BloomFilterOffset.HasValue) + return; // nothing to load + + // Save current read position; read bloom (unencrypted, uncompressed) and restore. + long cur = _inputStream.CanSeek ? _inputStream.Position : 0; + + try { + _bloom = BloomFilterIO.ReadFromStream( + _inputStream, + meta, + s => new ThriftCompactProtocolReader(s)); + } catch { + // Be tolerant: if bloom is corrupt/unsupported, just skip pruning. + _bloom = null; + } finally { + if(_inputStream.CanSeek) { + // restore to where page reading expects to start/continue + _inputStream.Seek(cur, SeekOrigin.Begin); + } + } + } } } diff --git a/src/Parquet/File/DataColumnWriter.cs b/src/Parquet/File/DataColumnWriter.cs index 4dd28bf4..d155d1f0 100644 --- a/src/Parquet/File/DataColumnWriter.cs +++ b/src/Parquet/File/DataColumnWriter.cs @@ -6,6 +6,7 @@ using System.Threading.Tasks; using IronCompress; using Microsoft.IO; +using Parquet.Bloom; using Parquet.Data; using Parquet.Encodings; using Parquet.Extensions; @@ -117,6 +118,17 @@ private async Task WriteColumnAsync(ColumnChunk chunk, DataColumn c using var pc = new PackedColumn(column); pc.Pack(_options.UseDictionaryEncoding, _options.DictionaryEncodingThreshold); + BloomCollector? bloom = null; + if(_options.BloomFilterOptionsByColumn.TryGetValue(column.Field.Name, out ParquetOptions.BloomFilterOptions? bloomOptions)) { + if(bloomOptions != null && bloomOptions.EnableBloomFilters) { + BloomSizing.BloomPlan plan = BloomSizing.Plan( + estimatedDistinctValues: column.Statistics?.DistinctCount ?? column.NumValues, + targetFpp: bloomOptions.BloomFilterFpp, + bitsPerValueOverride: bloomOptions.BloomFilterBitsPerValueOverride); + bloom = new BloomCollector(plan.Blocks); + } + } + // dictionary page if(pc.HasDictionary) { PageHeader ph = _footer.CreateDictionaryPage(pc.Dictionary!.Length); @@ -126,11 +138,17 @@ private async Task WriteColumnAsync(ColumnChunk chunk, DataColumn c ms, column.Statistics); await CompressAndWriteAsync(ph, ms, r, cancellationToken); + if(bloom != null) { + BloomAddValues(bloom, pc.Dictionary, 0, pc.Dictionary.Length, _schemaElement); + } } // data page using(MemoryStream ms = _rmsMgr.GetStream()) { Array data = pc.GetPlainData(out int offset, out int count); + if(bloom != null) { + BloomAddValues(bloom, data, offset, count, _schemaElement); + } bool deltaEncode = column.IsDeltaEncodable && _options.UseDeltaBinaryPackedEncoding && DeltaBinaryPackedEncoder.CanEncode(data, offset, count); // data page Num_values also does include NULLs @@ -148,7 +166,7 @@ private async Task WriteColumnAsync(ColumnChunk chunk, DataColumn c int bitWidth = pc.Dictionary!.Length.GetBitWidth(); ms.WriteByte((byte)bitWidth); // bit width is stored as 1 byte before encoded data RleBitpackedHybridEncoder.Encode(ms, indexes.AsSpan(0, indexesLength), bitWidth); - } else { + } else { if(deltaEncode) { DeltaBinaryPackedEncoder.Encode(data, offset, count, ms, column.Statistics); chunk.MetaData!.Encodings[2] = Encoding.DELTA_BINARY_PACKED; @@ -157,10 +175,18 @@ private async Task WriteColumnAsync(ColumnChunk chunk, DataColumn c } } - ph.DataPageHeader!.Statistics = column.Statistics.ToThriftStatistics(tse); + ph.DataPageHeader!.Statistics = column.Statistics!.ToThriftStatistics(tse); await CompressAndWriteAsync(ph, ms, r, cancellationToken); } + if(bloom != null && chunk?.MetaData != null) { + BloomFilterIO.WriteToStream( + _stream, + bloom.Filter, + chunk.MetaData, + s => new Meta.Proto.ThriftCompactProtocolWriter(s)); + } + return r; } @@ -168,5 +194,78 @@ private static void WriteLevels(Stream s, Span levels, int count, int maxVa int bitWidth = maxValue.GetBitWidth(); RleBitpackedHybridEncoder.EncodeWithLength(s, bitWidth, levels.Slice(0, count)); } + + private static void BloomAddValues(BloomCollector bloom, Array values, int offset, int count, SchemaElement tse) { + switch(tse.Type!.Value) { + case Meta.Type.BOOLEAN: { + if(values is bool[] a) + for(int i = 0; i < count; i++) + bloom.AddBoolean(a[offset + i]); + break; + } + case Meta.Type.INT32: { + if(values is int[] a) + for(int i = 0; i < count; i++) + bloom.AddInt32(a[offset + i]); + else if(values is uint[] au) + for(int i = 0; i < count; i++) + bloom.AddInt32(unchecked((int)au[offset + i])); + break; + } + case Meta.Type.INT64: { + if(values is long[] a) + for(int i = 0; i < count; i++) + bloom.AddInt64(a[offset + i]); + else if(values is ulong[] au) + for(int i = 0; i < count; i++) + bloom.AddInt64(unchecked((long)au[offset + i])); + break; + } + case Meta.Type.INT96: { + if(values is DateTime[] a) + for(int i = 0; i < count; i++) + bloom.AddInt96(a[offset + i]); + break; + } + case Meta.Type.FLOAT: { + if(values is float[] a) + for(int i = 0; i < count; i++) + bloom.AddFloat(a[offset + i]); + break; + } + case Meta.Type.DOUBLE: { + if(values is double[] a) + for(int i = 0; i < count; i++) + bloom.AddDouble(a[offset + i]); + break; + } + case Meta.Type.BYTE_ARRAY: { + if(values is string[] sa) { + for(int i = 0; i < count; i++) + bloom.AddString(sa[offset + i]); + } else if(values is byte[][] ba) { + for(int i = 0; i < count; i++) + bloom.AddByteArray(ba[offset + i]); + } else if(values is Array any && any.Length > 0 && any.GetValue(0) is byte[]) { + // Handles jagged byte[][] typed as Array + for(int i = 0; i < count; i++) + bloom.AddByteArray((byte[])any.GetValue(offset + i)!); + } + break; + } + case Meta.Type.FIXED_LEN_BYTE_ARRAY: { + if(values is byte[][] ba) { + for(int i = 0; i < count; i++) + bloom.AddFixed(ba[offset + i]); + } else if(values is Array any && any.Length > 0 && any.GetValue(0) is byte[]) { + for(int i = 0; i < count; i++) + bloom.AddFixed((byte[])any.GetValue(offset + i)!); + } + break; + } + default: + break; + } + } } } diff --git a/src/Parquet/Parquet.csproj b/src/Parquet/Parquet.csproj index 58705dc4..d9478cdf 100644 --- a/src/Parquet/Parquet.csproj +++ b/src/Parquet/Parquet.csproj @@ -53,6 +53,7 @@ + diff --git a/src/Parquet/ParquetOptions.cs b/src/Parquet/ParquetOptions.cs index 57555970..d2099d8e 100644 --- a/src/Parquet/ParquetOptions.cs +++ b/src/Parquet/ParquetOptions.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Data; namespace Parquet { @@ -91,5 +92,32 @@ public class ParquetOptions { /// Defaults to 64MB. /// public int MaximumLargePoolFreeBytes { get; set; } = 64 * 1024 * 1024; + + /// + /// Gets the bloom filter options for each column. The dictionary key is the column name, + /// and the value is the for that column. + /// + public Dictionary BloomFilterOptionsByColumn { get; set; } = new(); + /// + /// Options for configuring bloom filters in Parquet columns. + /// + public record BloomFilterOptions { + /// + /// When set to true, enables bloom filters for columns to improve query performance by filtering out non-matching values. + /// + public bool EnableBloomFilters { get; set; } = true; + + /// + /// False positive probability for bloom filters. This value determines the likelihood that the bloom filter will incorrectly indicate that a value is present. + /// Typical values are small (e.g., 0.01 for 1% false positive rate). + /// + public float BloomFilterFpp { get; set; } = 0.01f; + + /// + /// When set, overrides the number of bits per value used in bloom filters for columns. + /// This allows fine-tuning of bloom filter size and performance. If not set, the default calculation is used. + /// + public int? BloomFilterBitsPerValueOverride { get; set; } = null; + } } } diff --git a/src/Parquet/ParquetRowGroupReader.cs b/src/Parquet/ParquetRowGroupReader.cs index 2f2f7ff1..bea9baa3 100644 --- a/src/Parquet/ParquetRowGroupReader.cs +++ b/src/Parquet/ParquetRowGroupReader.cs @@ -54,6 +54,18 @@ public interface IParquetRowGroupReader { /// /// DataColumnStatistics? GetStatistics(DataField field); + + /// + /// Uses the column chunk's Bloom filter (if present) to check whether this row group + /// might contain at least one value equal to for . + /// Returns false iff the Bloom filter definitively rules it out. + /// If a Bloom filter is not present or disabled, returns true (i.e., "don't prune"). + /// + /// + /// This does not read data pages; it only inspects the Bloom filter area. + /// Stream position is preserved. + /// + bool MightMatchEquals(DataField field, T value); } /// @@ -162,6 +174,37 @@ public Dictionary GetCustomMetadata(DataField field) { return ReadColumnStatistics(cc); } + /// + /// Uses the column chunk's Bloom filter (if present) to check whether this row group + /// might contain at least one value equal to for . + /// Returns false iff the Bloom filter definitively rules it out. + /// If a Bloom filter is not present or disabled, returns true (i.e., "don't prune"). + /// + /// + /// This does not read data pages; it only inspects the Bloom filter area. + /// Stream position is preserved. + /// + public bool MightMatchEquals(DataField field, T value) { + if(field is null) + throw new ArgumentNullException(nameof(field)); + + ColumnChunk cc = GetMetadata(field) + ?? throw new ParquetException($"'{field.Path}' does not exist in this file"); + + // Prepare stats (may be null in metadata); safe default is no nulls/distincts + DataColumnStatistics stats = ReadColumnStatistics(cc) ?? new DataColumnStatistics(null, null, null, null); + + // Preserve stream position; bloom probing may seek. + long pos = _stream.CanSeek ? _stream.Position : 0; + try { + var dcr = new DataColumnReader(field, _stream, cc, stats, _footer, _options!); + return dcr.MightMatchEquals(value); + } finally { + if(_stream.CanSeek) + _stream.Seek(pos, SeekOrigin.Begin); + } + } + /// /// Dispose isn't required, retained for backward compatibility ///