YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.

//  This source code is licensed under the BSD-style license found in the

//  LICENSE file in the root directory of this source tree. An additional grant

//  of patent rights can be found in the PATENTS file in the same directory.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed 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.

//

// Copyright (c) 2011 The LevelDB Authors. All rights reserved.

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "yb/rocksdb/table/block_based_table_builder.h"

#include <assert.h>

#include <inttypes.h>

#include <stdio.h>

#include <map>

#include <memory>

#include <string>

#include <unordered_map>

#include <utility>

#include <glog/logging.h>

#include "yb/gutil/macros.h"

#include "yb/rocksdb/cache.h"

#include "yb/rocksdb/comparator.h"

#include "yb/rocksdb/db/dbformat.h"

#include "yb/rocksdb/env.h"

#include "yb/rocksdb/filter_policy.h"

#include "yb/rocksdb/flush_block_policy.h"

#include "yb/rocksdb/table.h"

#include "yb/rocksdb/table/block.h"

#include "yb/rocksdb/table/block_based_filter_block.h"

#include "yb/rocksdb/table/block_based_table_factory.h"

#include "yb/rocksdb/table/block_based_table_internal.h"

#include "yb/rocksdb/table/block_builder.h"

#include "yb/rocksdb/table/filter_block.h"

#include "yb/rocksdb/table/fixed_size_filter_block.h"

#include "yb/rocksdb/table/format.h"

#include "yb/rocksdb/table/full_filter_block.h"

#include "yb/rocksdb/table/index_builder.h"

#include "yb/rocksdb/table/meta_blocks.h"

#include "yb/rocksdb/table/table_builder.h"

#include "yb/rocksdb/util/coding.h"

#include "yb/rocksdb/util/compression.h"

#include "yb/rocksdb/util/crc32c.h"

#include "yb/rocksdb/util/file_reader_writer.h"

#include "yb/rocksdb/util/stop_watch.h"

#include "yb/rocksdb/util/xxhash.h"

#include "yb/util/mem_tracker.h"

#include "yb/util/status_log.h"

namespace rocksdb {

extern const char kHashIndexPrefixesBlock[];

extern const char kHashIndexPrefixesMetadataBlock[];

typedef FilterPolicy::FilterType FilterType;

// Without anonymous namespace here, we fail the warning -Wmissing-prototypes

namespace {

FilterType GetFilterType(const BlockBasedTableOptions& table_opt) {

  std::shared_ptr<const FilterPolicy> policy(table_opt.filter_policy);

  return policy != nullptr ? 
policy->GetFilterType()11.7k
 : 
FilterType::kNoFilter53.9k
;

}

// Create a filter builder based on its type.

FilterBlockBuilder* CreateFilterBlockBuilder(const ImmutableCFOptions& opt,

    const BlockBasedTableOptions& table_opt, FilterType filter_type) {

  switch (filter_type) {

    case FilterType::kBlockBasedFilter:

      return new BlockBasedFilterBlockBuilder(opt.prefix_extractor, table_opt);

    case FilterType::kFixedSizeFilter:

      return new FixedSizeFilterBlockBuilder(opt.prefix_extractor, table_opt);

    case FilterType::kFullFilter:

      return new FullFilterBlockBuilder(opt.prefix_extractor,

                                        table_opt.whole_key_filtering,

                                        table_opt.filter_policy->GetFilterBitsBuilder());

    case FilterType::kNoFilter:

      return nullptr;

  }

  RLOG(InfoLogLevel::FATAL_LEVEL, opt.info_log, "Corrupted filter_type: %d", filter_type);

  assert(false);

  return nullptr;

}

bool GoodCompressionRatio(size_t compressed_size, size_t raw_size) {

  // Check to see if compressed less than 12.5%

  return compressed_size < raw_size - (raw_size / 8u);

}

// format_version is the block format as defined in include/rocksdb/table.h

Slice CompressBlock(const Slice& raw,

                    const CompressionOptions& compression_options,

                    CompressionType* type, uint32_t format_version,

                    std::string* compressed_output) {

  if (*type == kNoCompression) {

    return raw;

  }

  // Will return compressed block contents if (1) the compression method is

  // supported in this platform and (2) the compression rate is "good enough".

  switch (*type) {

    case kSnappyCompression:

      if (Snappy_Compress(compression_options, raw.cdata(), raw.size(),

                          compressed_output) &&

          GoodCompressionRatio(compressed_output->size(), raw.size())) {

        return *compressed_output;

      }

      break;  // fall back to no compression.

    case kZlibCompression:

      if (Zlib_Compress(

              compression_options,

              GetCompressFormatForVersion(kZlibCompression, format_version),

              raw.cdata(), raw.size(), compressed_output) &&

          
GoodCompressionRatio(compressed_output->size(), raw.size())18.6k
) {

        return *compressed_output;

      }

      break;  // fall back to no compression.

    case kBZip2Compression:

      if (BZip2_Compress(

              compression_options,

              GetCompressFormatForVersion(kBZip2Compression, format_version),

              raw.cdata(), raw.size(), compressed_output) &&

          
GoodCompressionRatio(compressed_output->size(), raw.size())0
) {

        return *compressed_output;

      }

      break;  // fall back to no compression.

    case kLZ4Compression:

      if (LZ4_Compress(

              compression_options,

              GetCompressFormatForVersion(kLZ4Compression, format_version),

              raw.cdata(), raw.size(), compressed_output) &&

          GoodCompressionRatio(compressed_output->size(), raw.size())) {

        return *compressed_output;

      }

      break;  // fall back to no compression.

    case kLZ4HCCompression:

      if (LZ4HC_Compress(

              compression_options,

              GetCompressFormatForVersion(kLZ4HCCompression, format_version),

              raw.cdata(), raw.size(), compressed_output) &&

          GoodCompressionRatio(compressed_output->size(), raw.size())) {

        return *compressed_output;

      }

      break;     // fall back to no compression.

    case kZSTDNotFinalCompression:

      if (ZSTD_Compress(compression_options, raw.cdata(), raw.size(),

                        compressed_output) &&

          
GoodCompressionRatio(compressed_output->size(), raw.size())0
) {

        return *compressed_output;

      }

      break;     // fall back to no compression.

    
default: {}0
  // Do not recognize this compression type

  }

  // Compression method is not supported, or not good compression ratio, so just

  // fall back to uncompressed form.

  *type = kNoCompression;

  return raw;

}

}  // namespace

// kBlockBasedTableMagicNumber was picked by running

//    echo rocksdb.table.block_based | sha1sum

// and taking the leading 64 bits.

// Please note that kBlockBasedTableMagicNumber may also be accessed by other

// .cc files

// for that reason we declare it extern in the header but to get the space

// allocated

// it must be not extern in one place.

const uint64_t kBlockBasedTableMagicNumber = 0x88e241b785f4cff7ull;

// We also support reading and writing legacy block based table format (for

// backwards compatibility)

const uint64_t kLegacyBlockBasedTableMagicNumber = 0xdb4775248b80fb57ull;

// A collector that collects properties of interest to block-based table.

// For now this class looks heavy-weight since we only write one additional

// property.

// But in the foreseeable future, we will add more and more properties that are

// specific to block-based table.

class BlockBasedTableBuilder::BlockBasedTablePropertiesCollector

    : public IntTblPropCollector {

 public:

  explicit BlockBasedTablePropertiesCollector(

      BlockBasedTableBuilder::Rep* rep, IndexType index_type, bool whole_key_filtering,

      bool prefix_filtering, const KeyValueEncodingFormat key_value_encoding_format)

      : rep_(rep),

        index_type_(index_type),

        whole_key_filtering_(whole_key_filtering),

        prefix_filtering_(prefix_filtering),

        key_value_encoding_format_(key_value_encoding_format) {}

  virtual Status InternalAdd(const Slice& key, const Slice& value,

                             uint64_t file_size) override {

    // Intentionally left blank. Have no interest in collecting stats for

    // individual key/value pairs.

    return Status::OK();

  }

  Status Finish(UserCollectedProperties* properties) override;

  // The name of the properties collector can be used for debugging purpose.

  const char* Name() const override {

    return "BlockBasedTablePropertiesCollector";

  }

  UserCollectedProperties GetReadableProperties() const override {

    // Intentionally left blank.

    return UserCollectedProperties();

  }

 private:

  BlockBasedTableBuilder::Rep* rep_;

  IndexType index_type_;

  bool whole_key_filtering_;

  bool prefix_filtering_;

  KeyValueEncodingFormat key_value_encoding_format_;

};

// Originally following data was stored in BlockBasedTableBuilder::Rep and related to a single SST

// file. Since SST file is now split into two types of files - data file and metadata file,

// all file-related data was moved into dedicated structure for each file.

struct BlockBasedTableBuilder::FileWriterWithOffsetAndCachePrefix {

  // Pointer to file writer. BlockBasedTableBuilder constructor accepts raw pointers to

  // WritableFileWriter and it is responsibility of client code to delete writer instance after

  // usage.

  WritableFileWriter* writer = nullptr;

  // Current offset.

  uint64_t offset = 0;

  // BlockBasedTableBuilder uses compressed block cache passed to BlockBasedTableBuilder constructor

  // inside BlockBasedTableOptions instance. If that cache is set in options (off by default)

  // all data written to files is also put into compressed block cache to reduce number of file

  // read requests later during read operations.

  // File blocks are referred in cache by keys, which are composed from following data (see

  // BlockBasedTableBuilder::InsertBlockInCache):

  // - cache key prefix (unique for each file), generated by GenerateCachePrefix

  // - block offset within a file.

  block_based_table::CacheKeyPrefixBuffer compressed_cache_key_prefix;

};

struct BlockBasedTableBuilder::Rep {

  const ImmutableCFOptions ioptions;

  const BlockBasedTableOptions table_options;

  InternalKeyComparatorPtr internal_comparator;

  // When two file writers are passed to BlockBasedTableBuilder during creation - we

  // use separate writers for data and metadata. In case BlockBasedTableBuilder was created for

  // single file - both data_writer and metadata_writer will actually point to the same structure,

  // so data and metadata will both go into one file. As of 2017-03-10 we support both cases.

  // Actually it is only allowed to pass nullptr as data_writer at BlockBasedTableBuilder::Rep

  // constructor and higher levels in order to support one method signature (with default nullptr

  // value) for handling both cases. At the level of BlockBasedTableBuilder implementation both

  // writers (inside BlockBasedTableBuilder::Rep) are not null and refer to

  // the same file or separate files.

  std::shared_ptr<FileWriterWithOffsetAndCachePrefix> metadata_writer;

  std::shared_ptr<FileWriterWithOffsetAndCachePrefix> data_writer;

  Status status;

  FilterType filter_type;

  std::unique_ptr<FilterBlockBuilder> filter_block_builder;

  BlockBuilder data_block_builder;

  InternalKeySliceTransform internal_prefix_transform;

  const FilterPolicy::KeyTransformer* const filter_key_transformer;

  std::unique_ptr<IndexBuilder> data_index_builder;

  IndexBuilder::IndexBlocks data_index_blocks;

  BlockHandle last_index_block_handle;

  std::unique_ptr<IndexBuilder> filter_index_builder;

  std::string last_key;

  std::string last_filter_key;

  const CompressionType compression_type;

  const CompressionOptions compression_opts;

  TableProperties props;

  bool closed = false;  // Either Finish() or Abandon() has been called.

  BlockHandle data_pending_handle;    // Handle to add to data index block

  BlockHandle filter_pending_handle;  // Handle to add to filter index block

  std::string compressed_output;

  std::unique_ptr<FlushBlockPolicy> flush_block_policy;

  std::vector<std::unique_ptr<IntTblPropCollector>> table_properties_collectors;

  yb::MemTrackerPtr mem_tracker;

  bool TEST_skip_writing_key_value_encoding_format_ = false;

  Rep(const ImmutableCFOptions& _ioptions,

      const BlockBasedTableOptions& table_opt,

      const InternalKeyComparatorPtr& icomparator,

      const IntTblPropCollectorFactories& int_tbl_prop_collector_factories,

      uint32_t column_family_id,

      WritableFileWriter* metadata_file,

      WritableFileWriter* data_file,

      const CompressionType _compression_type,

      const CompressionOptions& _compression_opts,

      const bool skip_filters);

  bool is_split_sst() const { return data_writer != metadata_writer; }

};

Status BlockBasedTableBuilder::BlockBasedTablePropertiesCollector::Finish(

    UserCollectedProperties* properties) {

  std::string val;

  PutFixed32(&val, static_cast<uint32_t>(index_type_));

  properties->emplace(BlockBasedTablePropertyNames::kIndexType, val);

  properties->emplace(

      BlockBasedTablePropertyNames::kWholeKeyFiltering,

      ToBlockBasedTablePropertyValue(whole_key_filtering_));

  properties->emplace(

      BlockBasedTablePropertyNames::kPrefixFiltering,

      ToBlockBasedTablePropertyValue(prefix_filtering_));

  val.clear();

  PutFixed32(&val, rep_->data_index_builder->NumLevels());

  properties->emplace(BlockBasedTablePropertyNames::kNumIndexLevels, val);

  if (!rep_->TEST_skip_writing_key_value_encoding_format_) {

    val.clear();

    PutFixed8(&val, static_cast<uint8_t>(key_value_encoding_format_));

    properties->emplace(BlockBasedTablePropertyNames::kDataBlockKeyValueEncodingFormat, val);

  }

  return Status::OK();

}

BlockBasedTableBuilder::Rep::Rep(

    const ImmutableCFOptions& _ioptions,

    const BlockBasedTableOptions& table_opt,

    const InternalKeyComparatorPtr& icomparator,

    const IntTblPropCollectorFactories& int_tbl_prop_collector_factories,

    uint32_t column_family_id,

    WritableFileWriter* metadata_file,

    WritableFileWriter* data_file,

    const CompressionType _compression_type,

    const CompressionOptions& _compression_opts,

    const bool skip_filters)

    : ioptions(_ioptions),

      table_options(table_opt),

      internal_comparator(icomparator),

      filter_type(GetFilterType(table_options)),

      filter_block_builder(skip_filters ? nullptr : CreateFilterBlockBuilder(

          _ioptions, table_options, filter_type)),

      data_block_builder(

          table_options.block_restart_interval,

          table_options.data_block_key_value_encoding_format, table_options.use_delta_encoding),

      internal_prefix_transform(_ioptions.prefix_extractor),

      filter_key_transformer(table_opt.filter_policy ?

          table_opt.filter_policy->GetKeyTransformer() : nullptr),

      data_index_builder(

          IndexBuilder::CreateIndexBuilder(

              table_options.index_type, internal_comparator.get(), &internal_prefix_transform,

              table_options)),

      filter_index_builder(

          // Prefix_extractor is not used by binary search index which we use for bloom filter

          // blocks indexing.

          IndexBuilder::CreateIndexBuilder(

              IndexType::kBinarySearch, BytewiseComparator(),

              nullptr /* prefix_extractor */, table_options)),

      compression_type(_compression_type),

      compression_opts(_compression_opts),

      flush_block_policy(

          table_options.flush_block_policy_factory->NewFlushBlockPolicy(

              table_options, data_block_builder)) {

  if (_ioptions.mem_tracker) {

    mem_tracker = yb::MemTracker::FindOrCreateTracker(

        "BlockBasedTableBuilder", _ioptions.mem_tracker);

  }

  metadata_writer = std::make_shared<FileWriterWithOffsetAndCachePrefix>();

  metadata_writer->writer = metadata_file;

  if (data_file != nullptr) {

    data_writer = std::make_shared<FileWriterWithOffsetAndCachePrefix>();

    data_writer->writer = data_file;

  } else {

    data_writer = metadata_writer;

  }

  for (auto& collector_factories : int_tbl_prop_collector_factories) {

    table_properties_collectors.emplace_back(

        collector_factories->CreateIntTblPropCollector(column_family_id));

  }

  table_properties_collectors.emplace_back(new BlockBasedTablePropertiesCollector(

      this, table_options.index_type, table_options.whole_key_filtering,

      _ioptions.prefix_extractor != nullptr, table_options.data_block_key_value_encoding_format));

}

BlockBasedTableBuilder::BlockBasedTableBuilder(

    const ImmutableCFOptions& ioptions,

    const BlockBasedTableOptions& table_options,

    const InternalKeyComparatorPtr& internal_comparator,

    const IntTblPropCollectorFactories& int_tbl_prop_collector_factories,

    uint32_t column_family_id,

    WritableFileWriter* metadata_file,

    WritableFileWriter* data_file,

    const CompressionType compression_type,

    const CompressionOptions& compression_opts,

    const bool skip_filters) {

  BlockBasedTableOptions sanitized_table_options(table_options);

  if (sanitized_table_options.format_version == 0 &&

      
sanitized_table_options.checksum != kCRC32c0
) {

    RLOG(InfoLogLevel::WARN_LEVEL, ioptions.info_log,

        "Silently converting format_version to 1 because checksum is "

        "non-default");

    // silently convert format_version to 1 to keep consistent with current

    // behavior

    sanitized_table_options.format_version = 1;

  }

  rep_ = new Rep(ioptions, sanitized_table_options, internal_comparator,

                 int_tbl_prop_collector_factories, column_family_id, metadata_file, data_file,

                 compression_type, compression_opts, skip_filters);

  if (rep_->filter_block_builder != nullptr) {

    rep_->filter_block_builder->StartBlock(0);

  }

  if (table_options.block_cache_compressed.get() != nullptr) {

    GenerateCachePrefix(

        table_options.block_cache_compressed.get(), metadata_file->writable_file(),

        &rep_->metadata_writer->compressed_cache_key_prefix);

    if (rep_->is_split_sst()) {

      GenerateCachePrefix(

          table_options.block_cache_compressed.get(), data_file->writable_file(),

          &rep_->metadata_writer->compressed_cache_key_prefix);

    }

  }

}

BlockBasedTableBuilder::~BlockBasedTableBuilder() {

  assert(rep_->closed);  // Catch errors where caller forgot to call Finish()

  delete rep_;

}

void BlockBasedTableBuilder::Add(const Slice& key, const Slice& value) {

  Rep* const r = rep_;

  DCHECK(!r->closed);

  if (!ok()) 
return0
;

  if (r->props.num_entries > 0) {

    DCHECK_GT(r->internal_comparator->Compare(key, Slice(r->last_key)), 0)

        << "New key: " << key.ToDebugHexString()

        << ", last key: " << Slice(r->last_key).ToDebugHexString();

  }

  const auto should_flush_data = r->flush_block_policy->Update(key, value);

  if (should_flush_data) {

    DCHECK(!r->data_block_builder.empty());

    FlushDataBlock(key);

  }

  if (r->filter_block_builder != nullptr) {

    const Slice user_key = ExtractUserKey(key);

    const Slice filter_key = r->filter_key_transformer ?

        r->filter_key_transformer->Transform(user_key) : 
user_key1.75M
;

    if (!filter_key.empty() &&

        
(97.1M
r->props.num_entries == 097.1M
 ||

         
BytewiseComparator()->Compare(r->last_filter_key, filter_key) != 097.1M
)) {

      // No need to insert duplicate keys into Bloom filter.

      if (r->filter_block_builder->ShouldFlush()) {

        FlushFilterBlock(&filter_key);

      }

      r->filter_block_builder->Add(filter_key);

      r->last_filter_key.assign(filter_key.cdata(), filter_key.size());

    }

  }

  r->last_key.assign(key.cdata(), key.size());

  r->data_block_builder.Add(key, value);

  r->props.num_entries++;

  r->props.raw_key_size += key.size();

  r->props.raw_value_size += value.size();

  r->data_index_builder->OnKeyAdded(key);

  NotifyCollectTableCollectorsOnAdd(key, value, r->data_writer->offset,

      r->table_properties_collectors,

      r->ioptions.info_log);

}

void BlockBasedTableBuilder::FlushDataBlock(const Slice& next_block_first_key) {

  Rep* const r = rep_;

  assert(!r->closed);

  if (!ok()) 
return0
;

  size_t data_block_size = 0;

  if (
!r->data_block_builder.empty()2.86M
) {

    data_block_size = WriteBlock(&r->data_block_builder, &r->data_pending_handle,

        r->data_writer.get());

  }

  if (!ok()) 
return0
;

  
if (2.86M
!r->table_options.skip_table_builder_flush2.86M
) {

    r->status = r->data_writer->writer->Flush();

  }

  if (!ok()) 
return5
;

  if (r->filter_block_builder != nullptr && 
r->filter_type == FilterType::kBlockBasedFilter578k
) {

    // For FilterType::kBlockBasedFilter separate block of bloom filter is written per data block.

    r->filter_block_builder->StartBlock(r->data_writer->offset);

  }

  r->props.data_size += data_block_size;

  ++r->props.num_data_blocks;

  // Add item to index block.

  // We do not emit the index entry for a block until we have seen the

  // first key for the next data block.  This allows us to use shorter

  // keys in the index block.  For example, consider a block boundary

  // between the keys "the quick brown fox" and "the who".  We can use

  // "the r" as the key for the index block entry since it is >= all

  // entries in the first block and < all entries in subsequent

  // blocks.

  r->data_index_builder->AddIndexEntry(&r->last_key,

      next_block_first_key.empty() ? 
nullptr65.4k
 : 
&next_block_first_key2.79M
,

      r->data_pending_handle);

  while (r->data_index_builder->ShouldFlush()) {

    auto result = r->data_index_builder->FlushNextBlock(

        &r->data_index_blocks, r->last_index_block_handle);

    if (!result.ok()) {

      r->status = result.status();

      return;

    }

    DCHECK(result.get());

    WriteBlock(

        r->data_index_blocks.index_block_contents, &r->last_index_block_handle,

        r->metadata_writer.get());

    if (!ok()) 
return0
;

    ++r->props.num_data_index_blocks;

  }

}

void BlockBasedTableBuilder::FlushFilterBlock(const Slice* const next_block_first_filter_key) {

  Rep* const r = rep_;

  assert(!r->closed);

  assert(r->filter_block_builder != nullptr);

  if (!ok()) 
return0
;

  const size_t filter_block_size = WriteRawBlock(r->filter_block_builder->Finish(), kNoCompression,

      &r->filter_pending_handle, r->metadata_writer.get());

  if (!ok()) 
return0
;

  
if (14.3k
!r->table_options.skip_table_builder_flush14.3k
) {

    r->status = r->metadata_writer->writer->Flush();

  }

  if (!ok()) 
return0
;

  r->props.filter_size += filter_block_size;

  ++r->props.num_filter_blocks;

  const bool is_last_flush = next_block_first_filter_key == nullptr;

  if (!is_last_flush) {

    r->filter_block_builder->StartBlock(0);

  }

  // See explanation in BlockBasedTableBuilder::FlushDataBlock.

  r->filter_index_builder->AddIndexEntry(

      &r->last_filter_key, next_block_first_filter_key, r->filter_pending_handle);

}

size_t BlockBasedTableBuilder::WriteBlock(BlockBuilder* block,

                                          BlockHandle* handle,

                                          FileWriterWithOffsetAndCachePrefix* writer_info) {

  size_t block_size = WriteBlock(block->Finish(), handle, writer_info);

  block->Reset();

  return block_size;

}

size_t BlockBasedTableBuilder::WriteBlock(const Slice& raw_block_contents,

    BlockHandle* handle,

    FileWriterWithOffsetAndCachePrefix* writer_info) {

  // File format contains a sequence of blocks where each block has:

  //    block_data: uint8[n]

  //    type: uint8

  //    crc: uint32

  assert(ok());

  Rep* r = rep_;

  auto type = r->compression_type;

  Slice block_contents;

  if (raw_block_contents.size() < kCompressionSizeLimit) {

    block_contents =

        CompressBlock(raw_block_contents, r->compression_opts, &type,

                      r->table_options.format_version, &r->compressed_output);

  } else {

    RecordTick(r->ioptions.statistics, NUMBER_BLOCK_NOT_COMPRESSED);

    type = kNoCompression;

    block_contents = raw_block_contents;

  }

  size_t block_size = WriteRawBlock(block_contents, type, handle, writer_info);

  r->compressed_output.clear();

  return block_size;

}

size_t BlockBasedTableBuilder::WriteRawBlock(const Slice& block_contents,

                                             CompressionType type,

                                             BlockHandle* handle,

                                             FileWriterWithOffsetAndCachePrefix* writer_info) {

  Rep* r = rep_;

  StopWatch sw(r->ioptions.env, r->ioptions.statistics, WRITE_RAW_BLOCK_MICROS);

  const auto start_offset = writer_info->offset;

  handle->set_offset(writer_info->offset);

  handle->set_size(block_contents.size());

  r->status = writer_info->writer->Append(block_contents);

  if (
r->status.ok()3.10M
) {

    char trailer[kBlockTrailerSize];

    trailer[0] = type;

    char* trailer_without_type = trailer + 1;

    switch (r->table_options.checksum) {

      case kNoChecksum:

        // we don't support no checksum yet

        assert(false);

        // intentional fallthrough in release binary

        // We add a fallthrough annotation in release mode only -- otherwise we get a compile error

        // in debug mode ("fallthrough annotation in unreachable code").

#ifdef NDEBUG

        FALLTHROUGH_INTENDED;

#endif

      case kCRC32c: {

        auto crc = crc32c::Value(block_contents.data(), block_contents.size());

        crc = crc32c::Extend(crc, trailer, 1);  // Extend to cover block type

        EncodeFixed32(trailer_without_type, crc32c::Mask(crc));

        break;

      }

      case kxxHash: {

        void* xxh = XXH32_init(0);

        XXH32_update(xxh, block_contents.data(),

                     static_cast<uint32_t>(block_contents.size()));

        XXH32_update(xxh, trailer, 1);  // Extend  to cover block type

        EncodeFixed32(trailer_without_type, XXH32_digest(xxh));

        break;

      }

    }

    r->status = writer_info->writer->Append(Slice(trailer, kBlockTrailerSize));

    if (
r->status.ok()3.10M
) {

      r->status = InsertBlockInCache(block_contents, type, handle, writer_info);

    }

    if (
r->status.ok()3.10M
) {

      writer_info->offset += block_contents.size() + kBlockTrailerSize;

    }

  }

  return writer_info->offset - start_offset;

}

Status BlockBasedTableBuilder::status() const {

  return rep_->status;

}

static void DeleteCachedBlock(const Slice& key, void* value) {

  Block* block = reinterpret_cast<Block*>(value);

  delete block;

}

//

// Make a copy of the block contents and insert into compressed block cache

//

Status BlockBasedTableBuilder::InsertBlockInCache(const Slice& block_contents,

                                                  const CompressionType type,

    const BlockHandle* handle,

    FileWriterWithOffsetAndCachePrefix* writer_info) {

  Rep* r = rep_;

  Cache* block_cache_compressed = r->table_options.block_cache_compressed.get();

  if (type != kNoCompression && 
block_cache_compressed != nullptr1.03M
) {

    size_t size = block_contents.size();

    std::unique_ptr<char[]> ubuf(new char[size + 1]);

    memcpy(ubuf.get(), block_contents.data(), size);

    ubuf[size] = type;

    BlockContents results(std::move(ubuf), size, true, type, r->mem_tracker);

    Block* block = new Block(std::move(results));

    // make cache key by appending the file offset to the cache prefix id

    char* end = EncodeVarint64(

                  writer_info->compressed_cache_key_prefix.data +

                  writer_info->compressed_cache_key_prefix.size,

                  handle->offset());

    Slice key(writer_info->compressed_cache_key_prefix.data,

        static_cast<size_t> (end - writer_info->compressed_cache_key_prefix.data));

    // Insert into compressed block cache.

    RETURN_NOT_OK(block_cache_compressed->Insert(

        key, kDefaultQueryId, block, block->usable_size(), &DeleteCachedBlock));

    // Invalidate OS cache.

    auto status = writer_info->writer->InvalidateCache(

        static_cast<size_t>(writer_info->offset), size);

    if (!status.ok() && 
!status.IsNotSupported()4.42k
) {

      return status;

    }

  }

  return Status::OK();

}

Status BlockBasedTableBuilder::Finish() {

  Rep* r = rep_;

  Slice end_slice;

  if (!r->data_block_builder.empty()) {

    FlushDataBlock(end_slice);  // no more data block

  }

  if (r->filter_block_builder != nullptr) {

    FlushFilterBlock(nullptr);  // no more filter block

  }

  assert(!r->closed);

  r->closed = true;

  auto index_finish_result = r->data_index_builder->FlushNextBlock(

      &r->data_index_blocks, r->last_index_block_handle);

  RETURN_NOT_OK(index_finish_result);

  if (index_finish_result.get()) {

    ++r->props.num_data_index_blocks;

  }

  // Write meta blocks and metaindex block with the following order.

  //    1. [meta block: filter]

  //    2. [other meta blocks]

  //    3. [meta block: properties]

  //    4. [metaindex block]

  // write meta blocks

  MetaIndexBuilder meta_index_builder;

  for (const auto& item : r->data_index_blocks.meta_blocks) {

    BlockHandle block_handle;

    WriteBlock(item.second, &block_handle, r->metadata_writer.get());

    meta_index_builder.Add(item.first, block_handle);

  }

  if (ok()) {

    if (r->filter_block_builder != nullptr) {

      // Add mapping from "<filter_block_prefix>.Name" to location of either filter block or

      // filter index (for fixed-size bloom filter). We only need filter index for fixed-size bloom

      // filter, which is stored as separate blocks in SST file. Filters of other types are stored

      // as single block in SST file, so we just need an offset to that block instead of index.

      std::string key;

      switch (r->filter_type) {

        case FilterType::kFullFilter:

          key = block_based_table::kFullFilterBlockPrefix;

          break;

        case FilterType::kBlockBasedFilter:

          key = block_based_table::kFilterBlockPrefix;

          break;

        case FilterType::kFixedSizeFilter:

          key = block_based_table::kFixedSizeFilterBlockPrefix;

          break;

        case FilterType::kNoFilter:

          RLOG(InfoLogLevel::FATAL_LEVEL, r->ioptions.info_log,

              "r->filter_block_builder should be null for FilterType::kNoFilter");

          assert(false);

      }

      key.append(r->table_options.filter_policy->Name());

      if (r->filter_type == FilterType::kFixedSizeFilter) {

        // Flush the fixed-size bloom filter index and add its offset under the corresponding

        // key to meta index.

        IndexBuilder::IndexBlocks filter_index_blocks;

        RETURN_NOT_OK(r->filter_index_builder->Finish(&filter_index_blocks));

        BlockHandle filter_index_block_handle;

        WriteBlock(filter_index_blocks.index_block_contents, &filter_index_block_handle,

            r->metadata_writer.get());

        meta_index_builder.Add(key, filter_index_block_handle);

        r->props.filter_index_size = r->filter_index_builder->EstimatedSize() + kBlockTrailerSize;

      } else {

        meta_index_builder.Add(key, r->filter_pending_handle);

      }

    }

    // Write properties block.

    {

      PropertyBlockBuilder property_block_builder;

      r->props.filter_policy_name = r->table_options.filter_policy != nullptr ?

          
r->table_options.filter_policy->Name()11.6k
 : "";

      r->props.data_index_size =

          r->data_index_builder->EstimatedSize() + kBlockTrailerSize;

      // Add basic properties

      property_block_builder.AddTableProperty(r->props);

      // Add use collected properties

      NotifyCollectTableCollectorsOnFinish(r->table_properties_collectors,

          r->ioptions.info_log,

          &property_block_builder);

      BlockHandle properties_block_handle;

      WriteRawBlock(

          property_block_builder.Finish(),

          kNoCompression,

          &properties_block_handle,

          r->metadata_writer.get()

      );

      meta_index_builder.Add(kPropertiesBlock, properties_block_handle);

    }  // end of properties block writing

  }    // meta blocks

  BlockHandle meta_index_block_handle;

  // Write meta index and index block.

  if (ok()) {

    // Flush the meta index block.

    WriteRawBlock(

        meta_index_builder.Finish(), kNoCompression, &meta_index_block_handle,

        r->metadata_writer.get());

    // Flush index block if not already flushed.

    if (index_finish_result.get()) {

      WriteBlock(

          r->data_index_blocks.index_block_contents, &r->last_index_block_handle,

          r->metadata_writer.get());

    }

  }

  // Write footer

  if (ok()) {

    // No need to write out new footer if we're using default checksum.

    // We're writing legacy magic number because we want old versions of RocksDB

    // be able to read files generated with new release (just in case if

    // somebody wants to roll back after an upgrade)

    // TODO(icanadi) at some point in the future, when we're absolutely sure

    // nobody will roll back to RocksDB 2.x versions, retire the legacy magic

    // number and always write new table files with new magic number

    bool legacy = (r->table_options.format_version == 0);

    // this is guaranteed by BlockBasedTableBuilder's constructor

    assert(r->table_options.checksum == kCRC32c ||

        r->table_options.format_version != 0);

    Footer footer(legacy ? 
kLegacyBlockBasedTableMagicNumber0

            : kBlockBasedTableMagicNumber,

        r->table_options.format_version);

    footer.set_metaindex_handle(meta_index_block_handle);

    footer.set_index_handle(r->last_index_block_handle);

    footer.set_checksum(r->table_options.checksum);

    std::string footer_encoding;

    footer.AppendEncodedTo(&footer_encoding);

    r->status = r->metadata_writer->writer->Append(footer_encoding);

    if (r->status.ok()) {

      r->metadata_writer->offset += footer_encoding.size();

    }

  }

  return r->status;

}

void BlockBasedTableBuilder::Abandon() {

  Rep* r = rep_;

  assert(!r->closed);

  r->closed = true;

}

uint64_t BlockBasedTableBuilder::NumEntries() const {

  return rep_->props.num_entries;

}

uint64_t BlockBasedTableBuilder::TotalFileSize() const {

  return rep_->is_split_sst() ? 
rep_->metadata_writer->offset + rep_->data_writer->offset47.5M
 :

      
rep_->metadata_writer->offset140k
;

}

uint64_t BlockBasedTableBuilder::BaseFileSize() const {

  return rep_->metadata_writer->offset;

}

bool BlockBasedTableBuilder::NeedCompact() const {

  for (const auto& collector : rep_->table_properties_collectors) {

    if (collector->NeedCompact()) {

      return true;

    }

  }

  return false;

}

TableProperties BlockBasedTableBuilder::GetTableProperties() const {

  TableProperties ret = rep_->props;

  for (const auto& collector : rep_->table_properties_collectors) {

    for (const auto& prop : collector->GetReadableProperties()) {

      ret.readable_properties.insert(prop);

    }

    CHECK_OK(collector->Finish(&ret.user_collected_properties));

  }

  return ret;

}

void BlockBasedTableBuilder::TEST_skip_writing_key_value_encoding_format() {

  rep_->TEST_skip_writing_key_value_encoding_format_ = true;

}

}  // namespace rocksdb