zjw1111 commented on code in PR #444: URL: https://github.com/apache/paimon-cpp/pull/444#discussion_r4320895632
########## src/paimon/common/utils/dense_hash_table.h: ########## @@ -0,0 +1,257 @@ +/* + * 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 <algorithm> +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <optional> +#include <type_traits> +#include <utility> + +#include "paimon/common/memory/bytes_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" +#include "paimon/result.h" + +namespace paimon { + +template <typename Key, typename Value> +class DenseHashTableFileReader; + +/// A fixed-capacity, linear-probing hash table for already-hashed unsigned keys. +/// Adapted from: +/// https://github.com/alibaba/havenask/blob/main/aios/storage/indexlib/index/common/hash_table/DenseHashTable.h +/// All key and value bit patterns are supported. +/// Mutation requires external synchronization; mounted hash tables are read-only. +/// +/// The native layout is [uint64 bucket_count, uint64 key_count], followed by fixed-size +/// [Key, Value] buckets and two special-key buckets. Integers use host byte order and fields have +/// no padding. Dumped buffers require matching key/value types and endianness when read. This is +/// a container layout, not a versioned Paimon index format. +template <typename Key, typename Value> +class DenseHashTable { + static_assert(std::is_same_v<Key, uint32_t> || std::is_same_v<Key, uint64_t>); + static_assert(std::is_integral_v<Value> && !std::is_same_v<Value, bool>); + + public: + static Result<std::unique_ptr<DenseHashTable>> Create(uint64_t capacity, int32_t occupancy_pct, + const std::shared_ptr<MemoryPool>& pool) { + if (!pool) { + return Status::Invalid("dense hash table requires a memory pool"); + } + PAIMON_ASSIGN_OR_RAISE(uint64_t bucket_count, BucketCount(capacity, occupancy_pct)); + PAIMON_ASSIGN_OR_RAISE(uint64_t bytes, RequiredBytes(bucket_count)); + auto hash_table = std::unique_ptr<DenseHashTable>(new DenseHashTable()); + hash_table->data_ = AllocateBytesKeepingPoolAlive(bytes, pool); + hash_table->header_ = Header{bucket_count, 0}; + hash_table->capacity_ = capacity; + hash_table->writable_ = true; + hash_table->WriteHeader(); + for (uint64_t index = 0; index < bucket_count; ++index) { + hash_table->WriteBucket(index, Bucket{kEmptyKey, Value{}}); + } + hash_table->WriteBucket(bucket_count, Bucket{0, Value{}}); + hash_table->WriteBucket(bucket_count + 1, Bucket{0, Value{}}); + return hash_table; + } + + /// Mounts a shared buffer without copying. Its bytes must not be mutated while mounted. + /// The returned hash table retains the buffer; its allocator must also outlive the buffer. + static Result<std::unique_ptr<DenseHashTable>> MountForRead(std::shared_ptr<const Bytes> data) { Review Comment: why not const T& ########## src/paimon/common/utils/dense_hash_table.h: ########## @@ -0,0 +1,257 @@ +/* + * 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 <algorithm> +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <optional> +#include <type_traits> +#include <utility> + +#include "paimon/common/memory/bytes_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" +#include "paimon/result.h" + +namespace paimon { + +template <typename Key, typename Value> +class DenseHashTableFileReader; + +/// A fixed-capacity, linear-probing hash table for already-hashed unsigned keys. +/// Adapted from: +/// https://github.com/alibaba/havenask/blob/main/aios/storage/indexlib/index/common/hash_table/DenseHashTable.h Review Comment: also add in `LICENSE` file ########## src/paimon/common/utils/hash_map_test.cpp: ########## @@ -0,0 +1,132 @@ +/* + * 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 "paimon/common/utils/hash_map.h" + +#include <cstdint> +#include <limits> +#include <map> +#include <memory> +#include <random> + +#include "gtest/gtest.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { + +using Map = HashMap<uint64_t, int64_t>; Review Comment: need using? ########## src/paimon/common/utils/hash_map.h: ########## @@ -0,0 +1,207 @@ +/* + * 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 <cstddef> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <optional> +#include <type_traits> +#include <utility> +#include <vector> + +#include "paimon/common/memory/bytes_utils.h" +#include "paimon/macros.h" +#include "paimon/result.h" + +namespace paimon { + +/// A pool-backed hash map which grows by adding blocks without rehashing existing entries. +/// Adapted from indexlib/util/HashMap.h in https://github.com/alibaba/havenask. Review Comment: move to line 19, format it ########## src/paimon/common/utils/dense_hash_table_file_reader.h: ########## @@ -0,0 +1,186 @@ +/* + * 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 <algorithm> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <optional> +#include <string> +#include <utility> + +#include "paimon/common/io/cache/cache_manager.h" +#include "paimon/common/utils/dense_hash_table.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" +#include "paimon/memory/memory_segment.h" +#include "paimon/result.h" + +namespace paimon { + +/// Reads the native DenseHashTable layout using positional reads and a shared CacheManager. +/// Opening reads only the header, not the bucket array. Find is safe for concurrent callers if +/// the input stream supports concurrent positional reads. Shared streams are not closed here. +/// File paths identify immutable contents and must be unique within the shared cache. Destroying +/// a reader does not invalidate cached pages, which retain their allocation's memory pool. +template <typename Key, typename Value> +class DenseHashTableFileReader { + public: + static Result<std::unique_ptr<DenseHashTableFileReader>> Create( + uint32_t page_size, const std::shared_ptr<CacheManager>& cache_manager, + const std::shared_ptr<FileSystem>& fs, const std::string& path, + const std::shared_ptr<MemoryPool>& pool) { + if (!fs) { + return Status::Invalid("dense hash table file system is null"); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input, fs->Open(path)); + return Create(page_size, cache_manager, path, input, pool); + } + + static Result<std::unique_ptr<DenseHashTableFileReader>> Create( + uint32_t page_size, const std::shared_ptr<CacheManager>& cache_manager, + const std::string& file_path, const std::shared_ptr<InputStream>& input, + const std::shared_ptr<MemoryPool>& pool) { + if (!input || !pool || !cache_manager || file_path.empty() || page_size == 0 || + page_size > static_cast<uint32_t>(std::numeric_limits<int32_t>::max())) { + return Status::Invalid("invalid dense hash table reader input, pool or cache options"); + } + PAIMON_ASSIGN_OR_RAISE(int64_t length, input->Length()); + if (length < static_cast<int64_t>(HashTable::kHeaderSize)) { + return Status::Invalid("dense hash table file is shorter than its header"); + } + typename HashTable::Header header; + PAIMON_RETURN_NOT_OK( + ReadAt(input.get(), reinterpret_cast<char*>(&header), HashTable::kHeaderSize, 0)); + PAIMON_RETURN_NOT_OK(HashTable::ValidateHeader(header, length)); + return std::unique_ptr<DenseHashTableFileReader>(new DenseHashTableFileReader( + page_size, cache_manager, file_path, input, header, length, pool)); + } + + DenseHashTableFileReader(const DenseHashTableFileReader&) = delete; + DenseHashTableFileReader& operator=(const DenseHashTableFileReader&) = delete; + + Result<std::optional<Value>> Find(Key key) const { + std::optional<Page> page; + if (key == HashTable::kEmptyKey || key == HashTable::kOtherSpecialKey) { + PAIMON_ASSIGN_OR_RAISE( + typename HashTable::Bucket bucket, + ReadBucket(header_.bucket_count + (key == HashTable::kOtherSpecialKey), &page)); + if (bucket.key > 1) { + return Status::Invalid("invalid dense hash table special bucket"); + } + return bucket.key == 0 ? std::nullopt : std::optional<Value>(bucket.value); + } + uint64_t index = key % header_.bucket_count; + for (uint64_t probe = 0; probe < header_.bucket_count; ++probe) { + PAIMON_ASSIGN_OR_RAISE(typename HashTable::Bucket bucket, ReadBucket(index, &page)); + if (bucket.key == HashTable::kEmptyKey) { + return std::optional<Value>(); + } + if (bucket.key == HashTable::kOtherSpecialKey) { + return Status::Invalid("invalid dense hash table ordinary bucket"); + } + if (bucket.key == key) { + return std::optional<Value>(bucket.value); + } + index = (index + 1) % header_.bucket_count; + } + return std::optional<Value>(); + } + + uint64_t Size() const { + return header_.key_count; + } + + private: + using HashTable = DenseHashTable<Key, Value>; + + struct Page { + uint64_t index; + MemorySegment segment; + }; + + DenseHashTableFileReader(uint32_t page_size, std::shared_ptr<CacheManager> cache_manager, + std::string file_path, std::shared_ptr<InputStream> input, + typename HashTable::Header header, uint64_t length, + std::shared_ptr<MemoryPool> pool) + : input_(std::move(input)), + pool_(std::move(pool)), + cache_manager_(std::move(cache_manager)), + file_path_(std::move(file_path)), + header_(header), + length_(length), + page_size_(page_size) {} + + static Status ReadAt(InputStream* input, char* data, int64_t size, int64_t offset) { Review Comment: why InputStream*, not shared_ptr? ########## src/paimon/common/utils/dense_hash_table.h: ########## @@ -0,0 +1,257 @@ +/* + * 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 <algorithm> +#include <cstddef> +#include <cstdint> +#include <cstring> +#include <limits> +#include <memory> +#include <optional> +#include <type_traits> +#include <utility> + +#include "paimon/common/memory/bytes_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" +#include "paimon/result.h" + +namespace paimon { + +template <typename Key, typename Value> +class DenseHashTableFileReader; + +/// A fixed-capacity, linear-probing hash table for already-hashed unsigned keys. +/// Adapted from: +/// https://github.com/alibaba/havenask/blob/main/aios/storage/indexlib/index/common/hash_table/DenseHashTable.h Review Comment: move to line 19 ########## src/paimon/core/io/data_file_key_offset_index_writer.cpp: ########## @@ -0,0 +1,169 @@ +/* + * 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 "paimon/core/io/data_file_key_offset_index_writer.h" + +#include <utility> + +#include "arrow/api.h" +#include "arrow/compute/cast.h" +#include "paimon/common/global_index/key_serializer.h" +#include "paimon/common/io/data_output_stream.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/dense_hash_table.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/core/casting/casting_utils.h" +#include "paimon/core/io/key_offset_index_utils.h" +#include "paimon/fs/file_system.h" +#include "paimon/macros.h" + +namespace paimon { +Result<std::unique_ptr<DataFileKeyOffsetIndexWriter>> DataFileKeyOffsetIndexWriter::Create( + const std::shared_ptr<arrow::Field>& key_field, const std::shared_ptr<FileSystem>& file_system, + const std::shared_ptr<MemoryPool>& memory_pool) { + if (!key_field || !file_system || !memory_pool) { + return Status::Invalid("key-offset writer is missing a required dependency"); + } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> serializer, + KeySerializer::Create(key_field->type(), memory_pool)); + using OffsetMap = HashMap<uint64_t, int64_t>; + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<OffsetMap> offsets, + OffsetMap::Create(/*initial_capacity=*/1024, memory_pool)); + return std::unique_ptr<DataFileKeyOffsetIndexWriter>(new DataFileKeyOffsetIndexWriter( + key_field, file_system, std::move(serializer), std::move(offsets), memory_pool)); Review Comment: these param, not move, and not const& ########## src/paimon/core/io/data_file_writer_factory.cpp: ########## @@ -73,4 +76,24 @@ Result<std::unique_ptr<DataFileIndexWriter>> DataFileWriterFactory::CreateFileIn path_factory, pool_); } +Result<std::unique_ptr<DataFileKeyOffsetIndexWriter>> +DataFileWriterFactory::CreateOffsetIndexWriter( + const std::shared_ptr<arrow::Schema>& logical_schema) const { + const std::vector<std::string>& key_fields = options_.GetRealtimeDeduplicateKeyFields(); + if (key_fields.empty()) { + return std::unique_ptr<DataFileKeyOffsetIndexWriter>(); + } + if (!logical_schema || key_fields.size() != 1) { + return Status::Invalid("offset index requires exactly one deduplicate key field"); + } + std::shared_ptr<arrow::Field> key_field = logical_schema->GetFieldByName(key_fields.front()); + std::shared_ptr<arrow::Field> offset_field = + logical_schema->GetFieldByName(SpecialFields::RealtimeOffset().Name()); + if (!key_field || !offset_field || offset_field->type()->id() != arrow::Type::INT64) { Review Comment: is it necessary to validate offset_field here? ########## src/paimon/core/io/data_file_key_offset_index_writer.cpp: ########## @@ -0,0 +1,169 @@ +/* + * 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 "paimon/core/io/data_file_key_offset_index_writer.h" Review Comment: move the key-offset's writer、reader、utils to a single dir? -- 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]
