lxy-9602 commented on code in PR #234:
URL: https://github.com/apache/paimon-cpp/pull/234#discussion_r3842310329
##########
src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp:
##########
@@ -58,9 +66,85 @@ Result<std::shared_ptr<FileIndexReader>>
BloomFilterFileIndex::CreateReader(
return BloomFilterFileIndexReader::Create(arrow_type, bytes);
}
+Result<std::shared_ptr<FileIndexWriter>> BloomFilterFileIndex::CreateWriter(
+ ::ArrowSchema* c_arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
arrow_schema,
+ arrow::ImportSchema(c_arrow_schema));
+ if (arrow_schema->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid schema for BloomFilterFileIndexWriter, supposed to have
single field.");
+ }
+ return BloomFilterFileIndexWriter::Create(arrow_schema->field(0),
options_, pool);
+}
+
+Result<std::shared_ptr<BloomFilterFileIndexWriter>>
BloomFilterFileIndexWriter::Create(
+ const std::shared_ptr<arrow::Field>& field, const std::map<std::string,
std::string>& options,
+ const std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_ASSIGN_OR_RAISE(FastHash::HashFunction hash_function,
+ FastHash::GetHashFunction(field->type()));
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t items, OptionsUtils::GetValueFromMap<int32_t>(options,
BloomFilterFileIndex::kItems,
+
BloomFilterFileIndex::kDefaultItems));
+ PAIMON_ASSIGN_OR_RAISE(
+ double fpp, OptionsUtils::GetValueFromMap<double>(options,
BloomFilterFileIndex::kFpp,
+
BloomFilterFileIndex::kDefaultFpp));
+ std::shared_ptr<arrow::DataType> struct_type = arrow::struct_({field});
+ PAIMON_ASSIGN_OR_RAISE(BloomFilter64 filter, BloomFilter64::Create(items,
fpp, pool));
+ return std::shared_ptr<BloomFilterFileIndexWriter>(
+ new BloomFilterFileIndexWriter(struct_type, hash_function,
std::move(filter), pool));
+}
+
+BloomFilterFileIndexWriter::BloomFilterFileIndexWriter(
+ const std::shared_ptr<arrow::DataType>& struct_type,
+ const FastHash::HashFunction& hash_function, BloomFilter64&& filter,
+ const std::shared_ptr<MemoryPool>& pool)
+ : struct_type_(struct_type),
+ hash_function_(hash_function),
+ filter_(std::move(filter)),
+ pool_(pool) {}
+
+Status BloomFilterFileIndexWriter::AddBatch(::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(batch, struct_type_));
+ if (!array || array->type_id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ "invalid batch for BloomFilterFileIndexWriter, expected a struct
array");
+ }
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(array);
+ if (struct_array->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid batch for BloomFilterFileIndexWriter, expected a struct
array with exactly "
+ "one field");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<Literal> values,
+ LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0),
/*own_data=*/false));
+ for (const Literal& value : values) {
+ if (!value.IsNull()) {
+ filter_.AddHash(hash_function_(value));
+ }
+ }
+ return Status::OK();
+}
+
+Result<PAIMON_UNIQUE_PTR<Bytes>> BloomFilterFileIndexWriter::SerializedBytes()
const {
+ constexpr int32_t kHeaderLength = sizeof(int32_t);
+ const int32_t bit_set_length = filter_.GetBitSet().ByteLength();
+ PAIMON_UNIQUE_PTR<Bytes> bytes =
+ Bytes::AllocateBytes(kHeaderLength + bit_set_length, pool_.get());
+ const auto num_hash_functions =
static_cast<uint32_t>(filter_.GetNumHashFunctions());
+ bytes->data()[0] = static_cast<char>((num_hash_functions >> 24) & 0xff);
+ bytes->data()[1] = static_cast<char>((num_hash_functions >> 16) & 0xff);
+ bytes->data()[2] = static_cast<char>((num_hash_functions >> 8) & 0xff);
+ bytes->data()[3] = static_cast<char>(num_hash_functions & 0xff);
+ filter_.GetBitSet().ToByteArray(kHeaderLength, bit_set_length,
bytes->data());
Review Comment:
Could we use the built-in utility function here instead?
##########
src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp:
##########
@@ -58,9 +66,85 @@ Result<std::shared_ptr<FileIndexReader>>
BloomFilterFileIndex::CreateReader(
return BloomFilterFileIndexReader::Create(arrow_type, bytes);
}
+Result<std::shared_ptr<FileIndexWriter>> BloomFilterFileIndex::CreateWriter(
+ ::ArrowSchema* c_arrow_schema, const std::shared_ptr<MemoryPool>& pool)
const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema>
arrow_schema,
+ arrow::ImportSchema(c_arrow_schema));
+ if (arrow_schema->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid schema for BloomFilterFileIndexWriter, supposed to have
single field.");
+ }
+ return BloomFilterFileIndexWriter::Create(arrow_schema->field(0),
options_, pool);
+}
+
+Result<std::shared_ptr<BloomFilterFileIndexWriter>>
BloomFilterFileIndexWriter::Create(
+ const std::shared_ptr<arrow::Field>& field, const std::map<std::string,
std::string>& options,
+ const std::shared_ptr<MemoryPool>& pool) {
+ PAIMON_ASSIGN_OR_RAISE(FastHash::HashFunction hash_function,
+ FastHash::GetHashFunction(field->type()));
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t items, OptionsUtils::GetValueFromMap<int32_t>(options,
BloomFilterFileIndex::kItems,
+
BloomFilterFileIndex::kDefaultItems));
+ PAIMON_ASSIGN_OR_RAISE(
+ double fpp, OptionsUtils::GetValueFromMap<double>(options,
BloomFilterFileIndex::kFpp,
+
BloomFilterFileIndex::kDefaultFpp));
+ std::shared_ptr<arrow::DataType> struct_type = arrow::struct_({field});
+ PAIMON_ASSIGN_OR_RAISE(BloomFilter64 filter, BloomFilter64::Create(items,
fpp, pool));
+ return std::shared_ptr<BloomFilterFileIndexWriter>(
+ new BloomFilterFileIndexWriter(struct_type, hash_function,
std::move(filter), pool));
+}
+
+BloomFilterFileIndexWriter::BloomFilterFileIndexWriter(
+ const std::shared_ptr<arrow::DataType>& struct_type,
+ const FastHash::HashFunction& hash_function, BloomFilter64&& filter,
+ const std::shared_ptr<MemoryPool>& pool)
+ : struct_type_(struct_type),
+ hash_function_(hash_function),
+ filter_(std::move(filter)),
+ pool_(pool) {}
+
+Status BloomFilterFileIndexWriter::AddBatch(::ArrowArray* batch) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
+ arrow::ImportArray(batch, struct_type_));
+ if (!array || array->type_id() != arrow::Type::STRUCT) {
+ return Status::Invalid(
+ "invalid batch for BloomFilterFileIndexWriter, expected a struct
array");
+ }
+ std::shared_ptr<arrow::StructArray> struct_array =
+ checked_pointer_cast<arrow::StructArray>(array);
+ if (struct_array->num_fields() != 1) {
+ return Status::Invalid(
+ "invalid batch for BloomFilterFileIndexWriter, expected a struct
array with exactly "
+ "one field");
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<Literal> values,
+ LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0),
/*own_data=*/false));
+ for (const Literal& value : values) {
+ if (!value.IsNull()) {
+ filter_.AddHash(hash_function_(value));
Review Comment:
Could you please verify that the NaN handling is compatible with Java and
add corresponding tests?
--
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]
