mapleFU commented on code in PR #37400:
URL: https://github.com/apache/arrow/pull/37400#discussion_r1342279333
##########
cpp/src/parquet/column_writer.cc:
##########
@@ -2322,12 +2347,133 @@ Status
TypedColumnWriterImpl<FLBAType>::WriteArrowDense(
return Status::OK();
}
+template <typename DType>
+void TypedColumnWriterImpl<DType>::UpdateBloomFilter(const T* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, static_cast<int>(current_hash_batch_size),
+ hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<FLBAType>::UpdateBloomFilter(const FLBA* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, descr_->type_length(),
+ static_cast<int>(current_hash_batch_size),
hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<BooleanType>::UpdateBloomFilter(const bool*,
int64_t) {
+ DCHECK(bloom_filter_ == nullptr);
Review Comment:
Ah yeah, Unreachable seems better.
##########
cpp/src/parquet/bloom_filter_reader_writer_test.cc:
##########
@@ -0,0 +1,149 @@
+// 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.
+
+#include <arrow/testing/gtest_util.h>
+#include <gtest/gtest.h>
+
+#include "parquet/bloom_filter.h"
+#include "parquet/bloom_filter_builder.h"
+#include "parquet/bloom_filter_reader.h"
+#include "parquet/file_reader.h"
+#include "parquet/test_util.h"
+
+namespace parquet::test {
+
+TEST(BloomFilterReader, ReadBloomFilter) {
+ std::string dir_string(parquet::test::get_data_dir());
+ std::string path = dir_string + "/data_index_bloom_encoding_stats.parquet";
+ auto reader = ParquetFileReader::OpenFile(path, false);
+ auto file_metadata = reader->metadata();
+ EXPECT_FALSE(file_metadata->is_encryption_algorithm_set());
+ auto& bloom_filter_reader = reader->GetBloomFilterReader();
+ auto row_group_0 = bloom_filter_reader.RowGroup(0);
+ ASSERT_NE(nullptr, row_group_0);
+ EXPECT_THROW(bloom_filter_reader.RowGroup(1), ParquetException);
+ auto bloom_filter = row_group_0->GetColumnBloomFilter(0);
+ ASSERT_NE(nullptr, bloom_filter);
+ EXPECT_THROW(row_group_0->GetColumnBloomFilter(1), ParquetException);
+
+ // assert exists
+ {
+ std::string_view sv = "Hello";
+ ByteArray ba{sv};
+ EXPECT_TRUE(bloom_filter->FindHash(bloom_filter->Hash(&ba)));
+ }
+
+ // no exists
+ {
+ std::string_view sv = "NOT_EXISTS";
+ ByteArray ba{sv};
+ EXPECT_FALSE(bloom_filter->FindHash(bloom_filter->Hash(&ba)));
+ }
+}
+
+TEST(BloomFilterReader, FileNotHaveBloomFilter) {
+ // Can still get a BloomFilterReader and a RowGroupBloomFilter
+ // reader, but cannot get a non-null BloomFilter.
+ std::string dir_string(parquet::test::get_data_dir());
Review Comment:
I found other place also uses `std::string
dir_string(parquet::test::get_data_dir());` ...
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -5584,5 +5592,104 @@ TEST_F(ParquetPageIndexRoundTripTest, EnablePerColumn) {
/*null_counts=*/{0}}));
}
+class ParquetBloomFilterRoundTripTest : public ::testing::Test,
Review Comment:
@wgtmac we already has an issue https://github.com/apache/arrow/issues/28908
##########
cpp/src/parquet/column_writer.cc:
##########
@@ -2322,12 +2347,133 @@ Status
TypedColumnWriterImpl<FLBAType>::WriteArrowDense(
return Status::OK();
}
+template <typename DType>
+void TypedColumnWriterImpl<DType>::UpdateBloomFilter(const T* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, static_cast<int>(current_hash_batch_size),
+ hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<FLBAType>::UpdateBloomFilter(const FLBA* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, descr_->type_length(),
+ static_cast<int>(current_hash_batch_size),
hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<BooleanType>::UpdateBloomFilter(const bool*,
int64_t) {
+ DCHECK(bloom_filter_ == nullptr);
Review Comment:
Oh, I think assert bloom_filter_ == nullptr is proper here, because
`updateBloomFilter` will be called whenever it has filter.
##########
cpp/src/parquet/bloom_filter_builder.h:
##########
@@ -0,0 +1,67 @@
+// 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.
+
+#pragma once
+
+#include "arrow/io/type_fwd.h"
+#include "parquet/types.h"
+
+namespace parquet {
+
+class BloomFilter;
+class SchemaDescriptor;
+struct BloomFilterOptions;
+struct BloomFilterLocation;
+
+/// \brief Interface for collecting bloom filter of a parquet file.
+class PARQUET_EXPORT BloomFilterBuilder {
+ public:
+ /// \brief API convenience to create a BloomFilterBuilder.
+ static std::unique_ptr<BloomFilterBuilder> Make(const SchemaDescriptor*
schema,
+ const WriterProperties&
properties);
+
+ /// Append a new row group to host all incoming bloom filters.
+ ///
+ /// This method must be called before Finish.
Review Comment:
Because bloom filter is row-group based, it should be called before finish
and before writing to every RowGroup
##########
cpp/src/parquet/column_writer.cc:
##########
@@ -2322,12 +2347,133 @@ Status
TypedColumnWriterImpl<FLBAType>::WriteArrowDense(
return Status::OK();
}
+template <typename DType>
+void TypedColumnWriterImpl<DType>::UpdateBloomFilter(const T* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, static_cast<int>(current_hash_batch_size),
+ hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<FLBAType>::UpdateBloomFilter(const FLBA* values,
+ int64_t num_values) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ for (int64_t i = 0; i < num_values; i += kHashBatchSize) {
+ int64_t current_hash_batch_size = std::min(kHashBatchSize, num_values -
i);
+ bloom_filter_->Hashes(values, descr_->type_length(),
+ static_cast<int>(current_hash_batch_size),
hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+ static_cast<int>(current_hash_batch_size));
+ }
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<BooleanType>::UpdateBloomFilter(const bool*,
int64_t) {
+ DCHECK(bloom_filter_ == nullptr);
+}
+
+template <typename DType>
+void TypedColumnWriterImpl<DType>::UpdateBloomFilterSpaced(const T* values,
+ int64_t num_values,
+ const uint8_t*
valid_bits,
+ int64_t
valid_bits_offset) {
+ if (bloom_filter_) {
+ std::array<uint64_t, kHashBatchSize> hashes;
+ ::arrow::internal::VisitSetBitRunsVoid(
+ valid_bits, valid_bits_offset, num_values, [&](int64_t position,
int64_t length) {
+ for (int64_t i = 0; i < length; i += kHashBatchSize) {
+ auto current_hash_batch_size = std::min(kHashBatchSize, length -
i);
+ bloom_filter_->Hashes(values + i + position,
+ static_cast<int>(current_hash_batch_size),
+ hashes.data());
+ bloom_filter_->InsertHashes(hashes.data(),
+
static_cast<int>(current_hash_batch_size));
+ }
+ });
+ }
+}
+
+template <>
+void TypedColumnWriterImpl<BooleanType>::UpdateBloomFilterSpaced(const bool*,
int64_t,
+ const
uint8_t*,
+ int64_t) {
+ DCHECK(bloom_filter_ == nullptr);
Review Comment:
Here, `UpdateBloomFilter` will always called, and when `bloom_filter`
exists, it will update the bloom filter. however, for boolean, we should check
`bloom_filter_` is not exists.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
