partitioned_filter_block.cc 18.1 KB
Newer Older
Maysam Yabandeh's avatar
Maysam Yabandeh committed
1
//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
Siying Dong's avatar
Siying Dong committed
2
3
4
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).
Maysam Yabandeh's avatar
Maysam Yabandeh committed
5

6
#include "table/block_based/partitioned_filter_block.h"
Maysam Yabandeh's avatar
Maysam Yabandeh committed
7

Maysam Yabandeh's avatar
Maysam Yabandeh committed
8
9
#include <utility>

10
#include "monitoring/perf_context_imp.h"
11
#include "port/malloc.h"
Maysam Yabandeh's avatar
Maysam Yabandeh committed
12
#include "port/port.h"
Maysam Yabandeh's avatar
Maysam Yabandeh committed
13
#include "rocksdb/filter_policy.h"
14
15
#include "table/block_based/block.h"
#include "table/block_based/block_based_table_reader.h"
16
#include "test_util/testharness.h"
Maysam Yabandeh's avatar
Maysam Yabandeh committed
17
18
#include "util/coding.h"

19
namespace ROCKSDB_NAMESPACE {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
20

21
PartitionedFilterBlockBuilder::PartitionedFilterBlockBuilder(
Yanqin Jin's avatar
Yanqin Jin committed
22
    const SliceTransform* _prefix_extractor, bool whole_key_filtering,
23
    FilterBitsBuilder* filter_bits_builder, int index_block_restart_interval,
24
    const bool use_value_delta_encoding,
25
26
    PartitionedIndexBuilder* const p_index_builder,
    const uint32_t partition_size)
Yanqin Jin's avatar
Yanqin Jin committed
27
    : FullFilterBlockBuilder(_prefix_extractor, whole_key_filtering,
28
                             filter_bits_builder),
29
30
31
32
33
34
      index_on_filter_block_builder_(index_block_restart_interval,
                                     true /*use_delta_encoding*/,
                                     use_value_delta_encoding),
      index_on_filter_block_builder_without_seq_(index_block_restart_interval,
                                                 true /*use_delta_encoding*/,
                                                 use_value_delta_encoding),
Maysam Yabandeh's avatar
Maysam Yabandeh committed
35
      p_index_builder_(p_index_builder),
36
37
      keys_added_to_partition_(0) {
  keys_per_partition_ =
38
39
      filter_bits_builder_->CalculateNumEntry(partition_size);
}
40
41

PartitionedFilterBlockBuilder::~PartitionedFilterBlockBuilder() {}
Maysam Yabandeh's avatar
Maysam Yabandeh committed
42

43
44
void PartitionedFilterBlockBuilder::MaybeCutAFilterBlock(
    const Slice* next_key) {
45
  // Use == to send the request only once
46
  if (keys_added_to_partition_ == keys_per_partition_) {
47
48
49
50
    // Currently only index builder is in charge of cutting a partition. We keep
    // requesting until it is granted.
    p_index_builder_->RequestPartitionCut();
  }
51
52
  if (!p_index_builder_->ShouldCutFilterBlock()) {
    return;
Maysam Yabandeh's avatar
Maysam Yabandeh committed
53
  }
54
  filter_gc.push_back(std::unique_ptr<const char[]>(nullptr));
55
56
57
58
59
60
61
62
63
64

  // Add the prefix of the next key before finishing the partition. This hack,
  // fixes a bug with format_verison=3 where seeking for the prefix would lead
  // us to the previous partition.
  const bool add_prefix =
      next_key && prefix_extractor() && prefix_extractor()->InDomain(*next_key);
  if (add_prefix) {
    FullFilterBlockBuilder::AddPrefix(*next_key);
  }

65
66
67
  Slice filter = filter_bits_builder_->Finish(&filter_gc.back());
  std::string& index_key = p_index_builder_->GetPartitionKey();
  filters.push_back({index_key, filter});
68
  keys_added_to_partition_ = 0;
69
  Reset();
70
71
}

72
73
74
75
76
void PartitionedFilterBlockBuilder::Add(const Slice& key) {
  MaybeCutAFilterBlock(&key);
  FullFilterBlockBuilder::Add(key);
}

77
void PartitionedFilterBlockBuilder::AddKey(const Slice& key) {
78
79
  FullFilterBlockBuilder::AddKey(key);
  keys_added_to_partition_++;
Maysam Yabandeh's avatar
Maysam Yabandeh committed
80
81
}

82
83
84
85
86
Slice PartitionedFilterBlockBuilder::Finish(
    const BlockHandle& last_partition_block_handle, Status* status) {
  if (finishing_filters == true) {
    // Record the handle of the last written filter block in the index
    FilterEntry& last_entry = filters.front();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
87
88
    std::string handle_encoding;
    last_partition_block_handle.EncodeTo(&handle_encoding);
89
    std::string handle_delta_encoding;
90
91
92
93
    PutVarsignedint64(
        &handle_delta_encoding,
        last_partition_block_handle.size() - last_encoded_handle_.size());
    last_encoded_handle_ = last_partition_block_handle;
94
95
96
    const Slice handle_delta_encoding_slice(handle_delta_encoding);
    index_on_filter_block_builder_.Add(last_entry.key, handle_encoding,
                                       &handle_delta_encoding_slice);
97
98
    if (!p_index_builder_->seperator_is_key_plus_seq()) {
      index_on_filter_block_builder_without_seq_.Add(
99
100
          ExtractUserKey(last_entry.key), handle_encoding,
          &handle_delta_encoding_slice);
101
    }
102
    filters.pop_front();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
103
  } else {
104
    MaybeCutAFilterBlock(nullptr);
Maysam Yabandeh's avatar
Maysam Yabandeh committed
105
  }
106
107
108
109
110
  // If there is no filter partition left, then return the index on filter
  // partitions
  if (UNLIKELY(filters.empty())) {
    *status = Status::OK();
    if (finishing_filters) {
111
112
113
114
115
      if (p_index_builder_->seperator_is_key_plus_seq()) {
        return index_on_filter_block_builder_.Finish();
      } else {
        return index_on_filter_block_builder_without_seq_.Finish();
      }
116
117
118
119
120
121
122
123
124
125
    } else {
      // This is the rare case where no key was added to the filter
      return Slice();
    }
  } else {
    // Return the next filter partition in line and set Incomplete() status to
    // indicate we expect more calls to Finish
    *status = Status::Incomplete();
    finishing_filters = true;
    return filters.front().filter;
Maysam Yabandeh's avatar
Maysam Yabandeh committed
126
127
128
  }
}

Maysam Yabandeh's avatar
Maysam Yabandeh committed
129
PartitionedFilterBlockReader::PartitionedFilterBlockReader(
130
131
    const BlockBasedTable* t, CachableEntry<Block>&& filter_block)
    : FilterBlockReaderCommon(t, std::move(filter_block)) {}
Maysam Yabandeh's avatar
Maysam Yabandeh committed
132

133
134
135
136
137
138
139
140
141
142
143
std::unique_ptr<FilterBlockReader> PartitionedFilterBlockReader::Create(
    const BlockBasedTable* table, FilePrefetchBuffer* prefetch_buffer,
    bool use_cache, bool prefetch, bool pin,
    BlockCacheLookupContext* lookup_context) {
  assert(table);
  assert(table->get_rep());
  assert(!pin || prefetch);

  CachableEntry<Block> filter_block;
  if (prefetch || !use_cache) {
    const Status s = ReadFilterBlock(table, prefetch_buffer, ReadOptions(),
144
145
                                     use_cache, nullptr /* get_context */,
                                     lookup_context, &filter_block);
146
    if (!s.ok()) {
147
      IGNORE_STATUS_IF_ERROR(s);
148
149
150
151
152
153
      return std::unique_ptr<FilterBlockReader>();
    }

    if (use_cache && !pin) {
      filter_block.Reset();
    }
Maysam Yabandeh's avatar
Maysam Yabandeh committed
154
  }
155
156
157

  return std::unique_ptr<FilterBlockReader>(
      new PartitionedFilterBlockReader(table, std::move(filter_block)));
Maysam Yabandeh's avatar
Maysam Yabandeh committed
158
159
160
}

bool PartitionedFilterBlockReader::KeyMayMatch(
161
    const Slice& key, const SliceTransform* prefix_extractor,
162
    uint64_t block_offset, const bool no_io, const Slice* const const_ikey_ptr,
163
    GetContext* get_context, BlockCacheLookupContext* lookup_context) {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
164
165
  assert(const_ikey_ptr != nullptr);
  assert(block_offset == kNotValid);
166
  if (!whole_key_filtering()) {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
167
168
    return true;
  }
169
170
171
172

  return MayMatch(key, prefix_extractor, block_offset, no_io, const_ikey_ptr,
                  get_context, lookup_context,
                  &FullFilterBlockReader::KeyMayMatch);
Maysam Yabandeh's avatar
Maysam Yabandeh committed
173
174
}

175
176
177
178
179
180
181
182
183
184
185
186
187
void PartitionedFilterBlockReader::KeysMayMatch(
    MultiGetRange* range, const SliceTransform* prefix_extractor,
    uint64_t block_offset, const bool no_io,
    BlockCacheLookupContext* lookup_context) {
  assert(block_offset == kNotValid);
  if (!whole_key_filtering()) {
    return;  // Any/all may match
  }

  MayMatch(range, prefix_extractor, block_offset, no_io, lookup_context,
           &FullFilterBlockReader::KeysMayMatch);
}

Maysam Yabandeh's avatar
Maysam Yabandeh committed
188
bool PartitionedFilterBlockReader::PrefixMayMatch(
189
    const Slice& prefix, const SliceTransform* prefix_extractor,
190
    uint64_t block_offset, const bool no_io, const Slice* const const_ikey_ptr,
191
    GetContext* get_context, BlockCacheLookupContext* lookup_context) {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
192
193
  assert(const_ikey_ptr != nullptr);
  assert(block_offset == kNotValid);
194
  if (!table_prefix_extractor() && !prefix_extractor) {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
195
196
    return true;
  }
197
198
199
200

  return MayMatch(prefix, prefix_extractor, block_offset, no_io, const_ikey_ptr,
                  get_context, lookup_context,
                  &FullFilterBlockReader::PrefixMayMatch);
Maysam Yabandeh's avatar
Maysam Yabandeh committed
201
202
}

203
204
205
206
207
208
209
210
211
212
213
214
215
void PartitionedFilterBlockReader::PrefixesMayMatch(
    MultiGetRange* range, const SliceTransform* prefix_extractor,
    uint64_t block_offset, const bool no_io,
    BlockCacheLookupContext* lookup_context) {
  assert(block_offset == kNotValid);
  if (!table_prefix_extractor() && !prefix_extractor) {
    return;  // Any/all may match
  }

  MayMatch(range, prefix_extractor, block_offset, no_io, lookup_context,
           &FullFilterBlockReader::PrefixesMayMatch);
}

216
BlockHandle PartitionedFilterBlockReader::GetFilterPartitionHandle(
217
    const CachableEntry<Block>& filter_block, const Slice& entry) const {
Maysam Yabandeh's avatar
Maysam Yabandeh committed
218
  IndexBlockIter iter;
219
  const InternalKeyComparator* const comparator = internal_comparator();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
220
  Statistics* kNullStats = nullptr;
221
  filter_block.GetValue()->NewIndexIterator(
222
223
224
      comparator, comparator->user_comparator(),
      table()->get_rep()->get_global_seqno(BlockType::kFilter), &iter,
      kNullStats, true /* total_order_seek */, false /* have_first_key */,
225
      index_key_includes_seq(), index_value_is_full());
Maysam Yabandeh's avatar
Maysam Yabandeh committed
226
227
  iter.Seek(entry);
  if (UNLIKELY(!iter.Valid())) {
228
229
230
231
232
233
    // entry is larger than all the keys. However its prefix might still be
    // present in the last partition. If this is called by PrefixMayMatch this
    // is necessary for correct behavior. Otherwise it is unnecessary but safe.
    // Assuming this is an unlikely case for full key search, the performance
    // overhead should be negligible.
    iter.SeekToLast();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
234
235
  }
  assert(iter.Valid());
236
  BlockHandle fltr_blk_handle = iter.value().handle;
237
  return fltr_blk_handle;
Maysam Yabandeh's avatar
Maysam Yabandeh committed
238
239
}

240
241
242
243
Status PartitionedFilterBlockReader::GetFilterPartitionBlock(
    FilePrefetchBuffer* prefetch_buffer, const BlockHandle& fltr_blk_handle,
    bool no_io, GetContext* get_context,
    BlockCacheLookupContext* lookup_context,
244
    CachableEntry<ParsedFullFilterBlock>* filter_block) const {
245
246
247
248
249
250
251
252
253
254
255
  assert(table());
  assert(filter_block);
  assert(filter_block->IsEmpty());

  if (!filter_map_.empty()) {
    auto iter = filter_map_.find(fltr_blk_handle.offset());
    // This is a possible scenario since block cache might not have had space
    // for the partition
    if (iter != filter_map_.end()) {
      filter_block->SetUnownedValue(iter->second.GetValue());
      return Status::OK();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
256
257
    }
  }
258
259
260
261
262
263
264
265
266

  ReadOptions read_options;
  if (no_io) {
    read_options.read_tier = kBlockCacheTier;
  }

  const Status s =
      table()->RetrieveBlock(prefetch_buffer, read_options, fltr_blk_handle,
                             UncompressionDict::GetEmptyDict(), filter_block,
267
268
                             BlockType::kFilter, get_context, lookup_context,
                             /* for_compaction */ false, /* use_cache */ true);
269
270
271
272
273
274
275
276
277
278
279
280
281

  return s;
}

bool PartitionedFilterBlockReader::MayMatch(
    const Slice& slice, const SliceTransform* prefix_extractor,
    uint64_t block_offset, bool no_io, const Slice* const_ikey_ptr,
    GetContext* get_context, BlockCacheLookupContext* lookup_context,
    FilterFunction filter_function) const {
  CachableEntry<Block> filter_block;
  Status s =
      GetOrReadFilterBlock(no_io, get_context, lookup_context, &filter_block);
  if (UNLIKELY(!s.ok())) {
282
    IGNORE_STATUS_IF_ERROR(s);
283
284
285
286
287
288
289
290
291
292
293
294
    return true;
  }

  if (UNLIKELY(filter_block.GetValue()->size() == 0)) {
    return true;
  }

  auto filter_handle = GetFilterPartitionHandle(filter_block, *const_ikey_ptr);
  if (UNLIKELY(filter_handle.size() == 0)) {  // key is out of range
    return false;
  }

295
  CachableEntry<ParsedFullFilterBlock> filter_partition_block;
296
297
298
299
  s = GetFilterPartitionBlock(nullptr /* prefetch_buffer */, filter_handle,
                              no_io, get_context, lookup_context,
                              &filter_partition_block);
  if (UNLIKELY(!s.ok())) {
300
    IGNORE_STATUS_IF_ERROR(s);
301
302
303
304
305
306
307
308
    return true;
  }

  FullFilterBlockReader filter_partition(table(),
                                         std::move(filter_partition_block));
  return (filter_partition.*filter_function)(
      slice, prefix_extractor, block_offset, no_io, const_ikey_ptr, get_context,
      lookup_context);
Maysam Yabandeh's avatar
Maysam Yabandeh committed
309
310
}

311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
void PartitionedFilterBlockReader::MayMatch(
    MultiGetRange* range, const SliceTransform* prefix_extractor,
    uint64_t block_offset, bool no_io, BlockCacheLookupContext* lookup_context,
    FilterManyFunction filter_function) const {
  CachableEntry<Block> filter_block;
  Status s = GetOrReadFilterBlock(no_io, range->begin()->get_context,
                                  lookup_context, &filter_block);
  if (UNLIKELY(!s.ok())) {
    IGNORE_STATUS_IF_ERROR(s);
    return;  // Any/all may match
  }

  if (UNLIKELY(filter_block.GetValue()->size() == 0)) {
    return;  // Any/all may match
  }

  auto start_iter_same_handle = range->begin();
  BlockHandle prev_filter_handle = BlockHandle::NullBlockHandle();

  // For all keys mapping to same partition (must be adjacent in sorted order)
  // share block cache lookup and use full filter multiget on the partition
  // filter.
  for (auto iter = start_iter_same_handle; iter != range->end(); ++iter) {
    // TODO: re-use one top-level index iterator
    BlockHandle this_filter_handle =
        GetFilterPartitionHandle(filter_block, iter->ikey);
    if (!prev_filter_handle.IsNull() &&
        this_filter_handle != prev_filter_handle) {
      MultiGetRange subrange(*range, start_iter_same_handle, iter);
      MayMatchPartition(&subrange, prefix_extractor, block_offset,
                        prev_filter_handle, no_io, lookup_context,
                        filter_function);
      range->AddSkipsFrom(subrange);
      start_iter_same_handle = iter;
    }
    if (UNLIKELY(this_filter_handle.size() == 0)) {  // key is out of range
      // Not reachable with current behavior of GetFilterPartitionHandle
      assert(false);
      range->SkipKey(iter);
      prev_filter_handle = BlockHandle::NullBlockHandle();
    } else {
      prev_filter_handle = this_filter_handle;
    }
  }
  if (!prev_filter_handle.IsNull()) {
    MultiGetRange subrange(*range, start_iter_same_handle, range->end());
    MayMatchPartition(&subrange, prefix_extractor, block_offset,
                      prev_filter_handle, no_io, lookup_context,
                      filter_function);
    range->AddSkipsFrom(subrange);
  }
}

void PartitionedFilterBlockReader::MayMatchPartition(
    MultiGetRange* range, const SliceTransform* prefix_extractor,
    uint64_t block_offset, BlockHandle filter_handle, bool no_io,
    BlockCacheLookupContext* lookup_context,
    FilterManyFunction filter_function) const {
  CachableEntry<ParsedFullFilterBlock> filter_partition_block;
  Status s = GetFilterPartitionBlock(
      nullptr /* prefetch_buffer */, filter_handle, no_io,
      range->begin()->get_context, lookup_context, &filter_partition_block);
  if (UNLIKELY(!s.ok())) {
    IGNORE_STATUS_IF_ERROR(s);
    return;  // Any/all may match
  }

  FullFilterBlockReader filter_partition(table(),
                                         std::move(filter_partition_block));
  (filter_partition.*filter_function)(range, prefix_extractor, block_offset,
                                      no_io, lookup_context);
}

Maysam Yabandeh's avatar
Maysam Yabandeh committed
384
size_t PartitionedFilterBlockReader::ApproximateMemoryUsage() const {
385
  size_t usage = ApproximateFilterBlockMemoryUsage();
386
#ifdef ROCKSDB_MALLOC_USABLE_SIZE
387
  usage += malloc_usable_size(const_cast<PartitionedFilterBlockReader*>(this));
388
389
390
391
392
#else
  usage += sizeof(*this);
#endif  // ROCKSDB_MALLOC_USABLE_SIZE
  return usage;
  // TODO(myabandeh): better estimation for filter_map_ size
Maysam Yabandeh's avatar
Maysam Yabandeh committed
393
394
}

395
// TODO(myabandeh): merge this with the same function in IndexReader
396
397
398
399
400
401
void PartitionedFilterBlockReader::CacheDependencies(bool pin) {
  assert(table());

  const BlockBasedTable::Rep* const rep = table()->get_rep();
  assert(rep);

402
  BlockCacheLookupContext lookup_context{TableReaderCaller::kPrefetch};
403
404
405
406
407
408
409
410
411
412

  CachableEntry<Block> filter_block;

  Status s = GetOrReadFilterBlock(false /* no_io */, nullptr /* get_context */,
                                  &lookup_context, &filter_block);
  if (!s.ok()) {
    ROCKS_LOG_WARN(rep->ioptions.info_log,
                   "Error retrieving top-level filter block while trying to "
                   "cache filter partitions: %s",
                   s.ToString().c_str());
413
    IGNORE_STATUS_IF_ERROR(s);
414
415
416
417
418
419
    return;
  }

  // Before read partitions, prefetch them to avoid lots of IOs
  assert(filter_block.GetValue());

Maysam Yabandeh's avatar
Maysam Yabandeh committed
420
  IndexBlockIter biter;
421
  const InternalKeyComparator* const comparator = internal_comparator();
Maysam Yabandeh's avatar
Maysam Yabandeh committed
422
  Statistics* kNullStats = nullptr;
423
  filter_block.GetValue()->NewIndexIterator(
424
425
      comparator, comparator->user_comparator(),
      rep->get_global_seqno(BlockType::kFilter), &biter, kNullStats,
426
427
      true /* total_order_seek */, false /* have_first_key */,
      index_key_includes_seq(), index_value_is_full());
428
429
430
  // Index partitions are assumed to be consecuitive. Prefetch them all.
  // Read the first block offset
  biter.SeekToFirst();
431
  BlockHandle handle = biter.value().handle;
432
433
434
435
  uint64_t prefetch_off = handle.offset();

  // Read the last block's offset
  biter.SeekToLast();
436
  handle = biter.value().handle;
437
438
439
  uint64_t last_off = handle.offset() + handle.size() + kBlockTrailerSize;
  uint64_t prefetch_len = last_off - prefetch_off;
  std::unique_ptr<FilePrefetchBuffer> prefetch_buffer;
440

441
  prefetch_buffer.reset(new FilePrefetchBuffer());
442
  s = prefetch_buffer->Prefetch(rep->file.get(), prefetch_off,
443
                                static_cast<size_t>(prefetch_len));
444
445

  // After prefetch, read the partitions one by one
446
447
  ReadOptions read_options;
  for (biter.SeekToFirst(); biter.Valid(); biter.Next()) {
448
    handle = biter.value().handle;
449

450
    CachableEntry<ParsedFullFilterBlock> block;
451
452
453
454
455
456
457
458
459
460
461
462
463
    // TODO: Support counter batch update for partitioned index and
    // filter blocks
    s = table()->MaybeReadBlockAndLoadToCache(
        prefetch_buffer.get(), read_options, handle,
        UncompressionDict::GetEmptyDict(), &block, BlockType::kFilter,
        nullptr /* get_context */, &lookup_context, nullptr /* contents */);

    assert(s.ok() || block.GetValue() == nullptr);
    if (s.ok() && block.GetValue() != nullptr) {
      if (block.IsCached()) {
        if (pin) {
          filter_map_[handle.offset()] = std::move(block);
        }
464
465
      }
    }
466
    IGNORE_STATUS_IF_ERROR(s);
467
468
469
  }
}

470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
const InternalKeyComparator* PartitionedFilterBlockReader::internal_comparator()
    const {
  assert(table());
  assert(table()->get_rep());

  return &table()->get_rep()->internal_comparator;
}

bool PartitionedFilterBlockReader::index_key_includes_seq() const {
  assert(table());
  assert(table()->get_rep());

  return table()->get_rep()->index_key_includes_seq;
}

bool PartitionedFilterBlockReader::index_value_is_full() const {
  assert(table());
  assert(table()->get_rep());

  return table()->get_rep()->index_value_is_full;
}

492
}  // namespace ROCKSDB_NAMESPACE