|
| 1 | +/* |
| 2 | + * Copyright 2026-present Alibaba Inc. |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | + * you may not use this file except in compliance with the License. |
| 6 | + * You may obtain a copy of the License at |
| 7 | + * |
| 8 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + * |
| 10 | + * Unless required by applicable law or agreed to in writing, software |
| 11 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + * See the License for the specific language governing permissions and |
| 14 | + * limitations under the License. |
| 15 | + */ |
| 16 | + |
| 17 | +#include "paimon/common/file_index/rangebitmap/range_bitmap.h" |
| 18 | + |
| 19 | +#include <algorithm> |
| 20 | + |
| 21 | +#include "fmt/format.h" |
| 22 | +#include "paimon/common/file_index/rangebitmap/dictionary/chunked_dictionary.h" |
| 23 | +#include "paimon/common/file_index/rangebitmap/dictionary/key_factory.h" |
| 24 | +#include "paimon/common/file_index/rangebitmap/utils/literal_serialization_utils.h" |
| 25 | +#include "paimon/common/io/data_output_stream.h" |
| 26 | +#include "paimon/common/io/memory_segment_output_stream.h" |
| 27 | +#include "paimon/common/memory/memory_segment_utils.h" |
| 28 | +#include "paimon/common/utils/field_type_utils.h" |
| 29 | +#include "paimon/io/data_input_stream.h" |
| 30 | +#include "paimon/memory/bytes.h" |
| 31 | + |
| 32 | +namespace paimon { |
| 33 | + |
| 34 | +Result<std::unique_ptr<RangeBitmap>> RangeBitmap::Create( |
| 35 | + const std::shared_ptr<InputStream>& input_stream, const int64_t offset, |
| 36 | + const FieldType field_type, const std::shared_ptr<MemoryPool>& pool) { |
| 37 | + PAIMON_RETURN_NOT_OK(input_stream->Seek(offset, SeekOrigin::FS_SEEK_SET)); |
| 38 | + const auto data_in = std::make_shared<DataInputStream>(input_stream); |
| 39 | + PAIMON_ASSIGN_OR_RAISE(int32_t header_length, data_in->ReadValue<int32_t>()); |
| 40 | + PAIMON_ASSIGN_OR_RAISE(int8_t version, data_in->ReadValue<int8_t>()); |
| 41 | + if (version != kCurrentVersion) { |
| 42 | + return Status::Invalid(fmt::format("RangeBitmap unsupported version: {}", version)); |
| 43 | + } |
| 44 | + PAIMON_ASSIGN_OR_RAISE(int32_t rid, data_in->ReadValue<int32_t>()); |
| 45 | + PAIMON_ASSIGN_OR_RAISE(int32_t cardinality, data_in->ReadValue<int32_t>()); |
| 46 | + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeyFactory> shared_key_factory, |
| 47 | + KeyFactory::Create(field_type)); |
| 48 | + PAIMON_ASSIGN_OR_RAISE(LiteralSerDeUtils::Deserializer key_deserializer, |
| 49 | + LiteralSerDeUtils::CreateValueReader(field_type)); |
| 50 | + auto min = Literal{field_type}; |
| 51 | + auto max = Literal{field_type}; |
| 52 | + if (cardinality > 0) { |
| 53 | + PAIMON_ASSIGN_OR_RAISE(min, key_deserializer(data_in, pool.get())); |
| 54 | + PAIMON_ASSIGN_OR_RAISE(max, key_deserializer(data_in, pool.get())); |
| 55 | + } |
| 56 | + PAIMON_ASSIGN_OR_RAISE(int32_t dictionary_length, data_in->ReadValue<int32_t>()); |
| 57 | + auto dictionary_offset = static_cast<int32_t>(offset + sizeof(int32_t) + header_length); |
| 58 | + int32_t bsi_offset = dictionary_offset + dictionary_length; |
| 59 | + return std::unique_ptr<RangeBitmap>(new RangeBitmap(rid, cardinality, dictionary_offset, |
| 60 | + bsi_offset, min, max, shared_key_factory, |
| 61 | + input_stream, pool)); |
| 62 | +} |
| 63 | + |
| 64 | +Status RangeBitmap::Not(RoaringBitmap32* out) { |
| 65 | + out->Flip(0, rid_); |
| 66 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 is_not_null, this->IsNotNull()); |
| 67 | + *out &= is_not_null; |
| 68 | + return Status::OK(); |
| 69 | +} |
| 70 | + |
| 71 | +Result<RoaringBitmap32> RangeBitmap::Eq(const Literal& key) { |
| 72 | + if (cardinality_ <= 0) { |
| 73 | + return RoaringBitmap32(); |
| 74 | + } |
| 75 | + PAIMON_ASSIGN_OR_RAISE(int32_t min_compare, key.CompareTo(min_)); |
| 76 | + PAIMON_ASSIGN_OR_RAISE(int32_t max_compare, key.CompareTo(max_)); |
| 77 | + PAIMON_ASSIGN_OR_RAISE(auto* bit_slice_ptr, this->GetBitSliceIndex()); |
| 78 | + if (min_compare == 0 && max_compare == 0) { |
| 79 | + return bit_slice_ptr->IsNotNull({}); |
| 80 | + } |
| 81 | + if (min_compare < 0 || max_compare > 0) { |
| 82 | + return RoaringBitmap32(); |
| 83 | + } |
| 84 | + PAIMON_ASSIGN_OR_RAISE(auto* dictionary, this->GetDictionary()); |
| 85 | + PAIMON_ASSIGN_OR_RAISE(int32_t code, dictionary->Find(key)); |
| 86 | + if (code < 0) { |
| 87 | + return RoaringBitmap32(); |
| 88 | + } |
| 89 | + return bit_slice_ptr->Eq(code); |
| 90 | +} |
| 91 | + |
| 92 | +Result<RoaringBitmap32> RangeBitmap::Neq(const Literal& key) { |
| 93 | + if (cardinality_ <= 0) { |
| 94 | + return RoaringBitmap32(); |
| 95 | + } |
| 96 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 result, Eq(key)); |
| 97 | + PAIMON_RETURN_NOT_OK(Not(&result)); |
| 98 | + return result; |
| 99 | +} |
| 100 | + |
| 101 | +Result<RoaringBitmap32> RangeBitmap::Lt(const Literal& key) { |
| 102 | + if (cardinality_ <= 0) { |
| 103 | + return RoaringBitmap32(); |
| 104 | + } |
| 105 | + PAIMON_ASSIGN_OR_RAISE(int32_t min_compare, key.CompareTo(min_)); |
| 106 | + PAIMON_ASSIGN_OR_RAISE(int32_t max_compare, key.CompareTo(max_)); |
| 107 | + if (max_compare > 0) { |
| 108 | + return IsNotNull(); |
| 109 | + } |
| 110 | + if (min_compare <= 0) { |
| 111 | + return RoaringBitmap32(); |
| 112 | + } |
| 113 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 result, Gte(key)); |
| 114 | + PAIMON_RETURN_NOT_OK(Not(&result)); |
| 115 | + return result; |
| 116 | +} |
| 117 | + |
| 118 | +Result<RoaringBitmap32> RangeBitmap::Lte(const Literal& key) { |
| 119 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 lt_result, Lt(key)); |
| 120 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 eq_result, Eq(key)); |
| 121 | + lt_result |= eq_result; |
| 122 | + return lt_result; |
| 123 | +} |
| 124 | + |
| 125 | +Result<RoaringBitmap32> RangeBitmap::Gt(const Literal& key) { |
| 126 | + if (cardinality_ <= 0) { |
| 127 | + return RoaringBitmap32(); |
| 128 | + } |
| 129 | + PAIMON_ASSIGN_OR_RAISE(int32_t max_compare, key.CompareTo(max_)); |
| 130 | + if (max_compare >= 0) { |
| 131 | + return RoaringBitmap32(); |
| 132 | + } |
| 133 | + PAIMON_ASSIGN_OR_RAISE(int32_t min_compare, key.CompareTo(min_)); |
| 134 | + if (min_compare < 0) { |
| 135 | + return IsNotNull(); |
| 136 | + } |
| 137 | + PAIMON_ASSIGN_OR_RAISE(auto* dictionary, this->GetDictionary()); |
| 138 | + PAIMON_ASSIGN_OR_RAISE(int32_t code, dictionary->Find(key)); |
| 139 | + PAIMON_ASSIGN_OR_RAISE(auto* bit_slice_ptr, this->GetBitSliceIndex()); |
| 140 | + if (code >= 0) { |
| 141 | + return bit_slice_ptr->Gt(code); |
| 142 | + } |
| 143 | + return bit_slice_ptr->Gte(-code - 1); |
| 144 | +} |
| 145 | + |
| 146 | +Result<RoaringBitmap32> RangeBitmap::Gte(const Literal& key) { |
| 147 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 gt_result, Gt(key)); |
| 148 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 eq_result, Eq(key)); |
| 149 | + gt_result |= eq_result; |
| 150 | + return gt_result; |
| 151 | +} |
| 152 | + |
| 153 | +Result<RoaringBitmap32> RangeBitmap::In(const std::vector<Literal>& keys) { |
| 154 | + if (cardinality_ <= 0) { |
| 155 | + return RoaringBitmap32(); |
| 156 | + } |
| 157 | + RoaringBitmap32 result{}; |
| 158 | + for (const auto& key : keys) { |
| 159 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 bitmap, Eq(key)); |
| 160 | + result |= bitmap; |
| 161 | + } |
| 162 | + return result; |
| 163 | +} |
| 164 | + |
| 165 | +Result<RoaringBitmap32> RangeBitmap::NotIn(const std::vector<Literal>& keys) { |
| 166 | + if (cardinality_ <= 0) { |
| 167 | + return RoaringBitmap32(); |
| 168 | + } |
| 169 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 result, In(keys)); |
| 170 | + PAIMON_RETURN_NOT_OK(Not(&result)); |
| 171 | + return result; |
| 172 | +} |
| 173 | + |
| 174 | +Result<RoaringBitmap32> RangeBitmap::IsNull() { |
| 175 | + if (cardinality_ <= 0) { |
| 176 | + if (rid_ > 0) { |
| 177 | + RoaringBitmap32 result; |
| 178 | + result.AddRange(0, rid_); |
| 179 | + return result; |
| 180 | + } |
| 181 | + return RoaringBitmap32(); |
| 182 | + } |
| 183 | + |
| 184 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 non_null_bitmap, IsNotNull()); |
| 185 | + non_null_bitmap.Flip(0, rid_); |
| 186 | + return non_null_bitmap; |
| 187 | +} |
| 188 | + |
| 189 | +Result<RoaringBitmap32> RangeBitmap::IsNotNull() { |
| 190 | + if (cardinality_ <= 0) { |
| 191 | + return RoaringBitmap32(); |
| 192 | + } |
| 193 | + PAIMON_ASSIGN_OR_RAISE(auto* bit_slice_ptr, this->GetBitSliceIndex()); |
| 194 | + PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 result, bit_slice_ptr->IsNotNull({})); |
| 195 | + return result; |
| 196 | +} |
| 197 | + |
| 198 | +RangeBitmap::RangeBitmap(const int32_t rid, const int32_t cardinality, |
| 199 | + const int32_t dictionary_offset, const int32_t bsi_offset, |
| 200 | + const Literal& min, const Literal& max, |
| 201 | + const std::shared_ptr<KeyFactory>& key_factory, |
| 202 | + const std::shared_ptr<InputStream>& input_stream, |
| 203 | + const std::shared_ptr<MemoryPool>& pool) |
| 204 | + : pool_(pool), |
| 205 | + rid_(rid), |
| 206 | + cardinality_(cardinality), |
| 207 | + bsi_offset_(bsi_offset), |
| 208 | + dictionary_offset_(dictionary_offset), |
| 209 | + min_(min), |
| 210 | + max_(max), |
| 211 | + key_factory_(key_factory), |
| 212 | + input_stream_(input_stream), |
| 213 | + bsi_(nullptr), |
| 214 | + dictionary_(nullptr) {} |
| 215 | + |
| 216 | +Result<BitSliceIndexBitmap*> RangeBitmap::GetBitSliceIndex() { |
| 217 | + if (bsi_ == nullptr) { |
| 218 | + PAIMON_ASSIGN_OR_RAISE(bsi_, |
| 219 | + BitSliceIndexBitmap::Create(input_stream_, bsi_offset_, pool_)); |
| 220 | + } |
| 221 | + return bsi_.get(); |
| 222 | +} |
| 223 | + |
| 224 | +Result<Dictionary*> RangeBitmap::GetDictionary() { |
| 225 | + if (dictionary_ == nullptr) { |
| 226 | + PAIMON_ASSIGN_OR_RAISE( |
| 227 | + dictionary_, ChunkedDictionary::Create(key_factory_->GetFieldType(), input_stream_, |
| 228 | + dictionary_offset_, pool_)); |
| 229 | + } |
| 230 | + return dictionary_.get(); |
| 231 | +} |
| 232 | + |
| 233 | +Result<std::unique_ptr<RangeBitmap::Appender>> RangeBitmap::Appender::Create( |
| 234 | + const std::shared_ptr<KeyFactory>& factory, int64_t limited_serialized_size_in_bytes, |
| 235 | + const std::shared_ptr<MemoryPool>& pool) { |
| 236 | + return std::unique_ptr<Appender>(new Appender(factory, limited_serialized_size_in_bytes, pool)); |
| 237 | +} |
| 238 | + |
| 239 | +RangeBitmap::Appender::Appender(const std::shared_ptr<KeyFactory>& factory, |
| 240 | + const int64_t limited_serialized_size_in_bytes, |
| 241 | + const std::shared_ptr<MemoryPool>& pool) |
| 242 | + : pool_(pool), |
| 243 | + rid_(0), |
| 244 | + bitmaps_(LiteralComparator(factory)), |
| 245 | + factory_(factory), |
| 246 | + chunk_size_bytes_limit_(limited_serialized_size_in_bytes) {} |
| 247 | + |
| 248 | +void RangeBitmap::Appender::Append(const Literal& key) { |
| 249 | + if (!key.IsNull()) { |
| 250 | + bitmaps_[key].Add(rid_); |
| 251 | + } |
| 252 | + rid_++; |
| 253 | +} |
| 254 | + |
| 255 | +Result<PAIMON_UNIQUE_PTR<Bytes>> RangeBitmap::Appender::Serialize() const { |
| 256 | + int32_t code = 0; |
| 257 | + const int32_t max_code = bitmaps_.empty() ? 0 : static_cast<int32_t>(bitmaps_.size() - 1); |
| 258 | + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BitSliceIndexBitmap::Appender> bsi_appender, |
| 259 | + BitSliceIndexBitmap::Appender::Create(0, max_code, pool_)); |
| 260 | + if (chunk_size_bytes_limit_ >= std::numeric_limits<int32_t>::max()) { |
| 261 | + return Status::Invalid(fmt::format( |
| 262 | + "Chunk size cannot be larger than 2GB, current bytes: {}", chunk_size_bytes_limit_)); |
| 263 | + } |
| 264 | + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ChunkedDictionary::Appender> dictionary_apender, |
| 265 | + ChunkedDictionary::Appender::Create( |
| 266 | + factory_, static_cast<int32_t>(chunk_size_bytes_limit_), pool_)); |
| 267 | + for (const auto& [key, bitmap] : bitmaps_) { |
| 268 | + PAIMON_RETURN_NOT_OK(dictionary_apender->AppendSorted(key, code)); |
| 269 | + for (auto it = bitmap.Begin(); it != bitmap.End(); ++it) { |
| 270 | + PAIMON_RETURN_NOT_OK(bsi_appender->Append(*it, code)); |
| 271 | + } |
| 272 | + code++; |
| 273 | + } |
| 274 | + PAIMON_ASSIGN_OR_RAISE(LiteralSerDeUtils::Serializer serializer, |
| 275 | + LiteralSerDeUtils::CreateValueWriter(factory_->GetFieldType())); |
| 276 | + Literal min{factory_->GetFieldType()}; |
| 277 | + Literal max{factory_->GetFieldType()}; |
| 278 | + if (!bitmaps_.empty()) { |
| 279 | + min = bitmaps_.begin()->first; |
| 280 | + max = bitmaps_.rbegin()->first; |
| 281 | + } |
| 282 | + PAIMON_ASSIGN_OR_RAISE(int32_t min_size, LiteralSerDeUtils::GetSerializedSizeInBytes(min)); |
| 283 | + PAIMON_ASSIGN_OR_RAISE(int32_t max_size, LiteralSerDeUtils::GetSerializedSizeInBytes(max)); |
| 284 | + int32_t header_size = 0; |
| 285 | + header_size += sizeof(int8_t); // version |
| 286 | + header_size += sizeof(int32_t); // rid |
| 287 | + header_size += sizeof(int32_t); // cardinality |
| 288 | + header_size += min.IsNull() ? 0 : min_size; // min literal size |
| 289 | + header_size += max.IsNull() ? 0 : max_size; // max literal size |
| 290 | + header_size += sizeof(int32_t); // dictionary length |
| 291 | + PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> dictionary_bytes, |
| 292 | + dictionary_apender->Serialize()); |
| 293 | + auto dictionary_length = static_cast<int32_t>(dictionary_bytes->size()); |
| 294 | + PAIMON_ASSIGN_OR_RAISE(PAIMON_UNIQUE_PTR<Bytes> bsi_bytes, bsi_appender->Serialize()); |
| 295 | + size_t bsi_length = bsi_bytes->size(); |
| 296 | + const auto data_output_stream = std::make_shared<MemorySegmentOutputStream>( |
| 297 | + MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool_); |
| 298 | + data_output_stream->WriteValue<int32_t>(header_size); |
| 299 | + data_output_stream->WriteValue<int8_t>(kCurrentVersion); |
| 300 | + data_output_stream->WriteValue<int32_t>(rid_); |
| 301 | + data_output_stream->WriteValue<int32_t>(static_cast<int32_t>(bitmaps_.size())); |
| 302 | + if (!min.IsNull()) { |
| 303 | + PAIMON_RETURN_NOT_OK(serializer(data_output_stream, min)); |
| 304 | + } |
| 305 | + if (!max.IsNull()) { |
| 306 | + PAIMON_RETURN_NOT_OK(serializer(data_output_stream, max)); |
| 307 | + } |
| 308 | + data_output_stream->WriteValue<int32_t>(dictionary_length); |
| 309 | + data_output_stream->Write(dictionary_bytes->data(), dictionary_length); |
| 310 | + data_output_stream->Write(bsi_bytes->data(), bsi_length); |
| 311 | + return MemorySegmentUtils::CopyToBytes(data_output_stream->Segments(), 0, |
| 312 | + static_cast<int32_t>(data_output_stream->CurrentSize()), |
| 313 | + pool_.get()); |
| 314 | +} |
| 315 | +} // namespace paimon |
0 commit comments