text: //  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.
  //
  // 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 "table/block_based_table_builder.h"
  
  #include <assert.h>
  #include <inttypes.h>
  #include <stdio.h>
  
  #include <map>
  #include <memory>
  #include <string>
  #include <utility>
  
  #include "db/dbformat.h"
  #include "table/block.h"
  #include "table/block_based_table_factory.h"
  #include "table/block_based_table_reader.h"
  #include "table/block_builder.h"
  #include "table/format.h"
  #include "table/index_builder.h"
  #include "table/meta_blocks.h"
  #include "table/table_builder.h"
  #include "util/coding.h"
  #include "util/compression.h"
  #include "util/crc32c.h"
  #include "util/stop_watch.h"
  #include "util/string_util.h"
  #include "vidardb/cache.h"
  #include "vidardb/comparator.h"
  #include "vidardb/env.h"
  #include "vidardb/flush_block_policy.h"
  #include "vidardb/table.h"
  
  namespace vidardb {
  
  
  // Without anonymous namespace here, we fail the warning -Wmissing-prototypes
  namespace {
  
  // Create a index builder based on its type.
  IndexBuilder* CreateIndexBuilder(const Comparator* comparator,
                                   int index_block_restart_interval) {
    return new ShortenedIndexBuilder(comparator, index_block_restart_interval);
  }
  
  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);
  }
  
  Slice CompressBlock(const Slice& raw,
                      const CompressionOptions& compression_options,
                      CompressionType* type, const Slice& compression_dict,
                      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.data(), 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),
                raw.data(), raw.size(), compressed_output, compression_dict) &&
            GoodCompressionRatio(compressed_output->size(), raw.size())) {
          return *compressed_output;
        }
        break;  // fall back to no compression.
      case kBZip2Compression:
        if (BZip2_Compress(
                compression_options,
                GetCompressFormatForVersion(kBZip2Compression),
                raw.data(), raw.size(), compressed_output) &&
            GoodCompressionRatio(compressed_output->size(), raw.size())) {
          return *compressed_output;
        }
        break;  // fall back to no compression.
      default: {}  // 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 vidardb.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;
  
  struct BlockBasedTableBuilder::Rep {
    const ImmutableCFOptions ioptions;
    const BlockBasedTableOptions table_options;
    const InternalKeyComparator& internal_comparator;
    WritableFileWriter* file;
    uint64_t offset = 0;
    Status status;
    BlockBuilder data_block;
  
    std::unique_ptr<IndexBuilder> index_builder;
  
    std::string last_key;
    const CompressionType compression_type;
    const CompressionOptions compression_opts;
    // Data for presetting the compression library's dictionary, or nullptr.
    const std::string* compression_dict;
    TableProperties props;
  
    bool closed = false;  // Either Finish() or Abandon() has been called.
  
    BlockHandle pending_handle;  // Handle to add to index block
  
    std::string compressed_output;
    std::unique_ptr<FlushBlockPolicy> flush_block_policy;
    uint32_t column_family_id;
    const std::string& column_family_name;
  
    std::vector<std::unique_ptr<IntTblPropCollector>> table_properties_collectors;
  
    Rep(const ImmutableCFOptions& _ioptions,
        const BlockBasedTableOptions& table_opt,
        const InternalKeyComparator& icomparator,
        const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
            int_tbl_prop_collector_factories,
        uint32_t _column_family_id, WritableFileWriter* f,
        const CompressionType _compression_type,
        const CompressionOptions& _compression_opts,
        const std::string* _compression_dict,
        const std::string& _column_family_name)
        : ioptions(_ioptions),
          table_options(table_opt),
          internal_comparator(icomparator),
          file(f),
          data_block(table_options.block_restart_interval),
          index_builder(
              CreateIndexBuilder(&internal_comparator,
                                 table_options.index_block_restart_interval)),
          compression_type(_compression_type),
          compression_opts(_compression_opts),
          compression_dict(_compression_dict),
          flush_block_policy(
              table_options.flush_block_policy_factory->NewFlushBlockPolicy(
                  table_options, data_block)),
          column_family_id(_column_family_id),
          column_family_name(_column_family_name) {
      for (auto& collector_factories : *int_tbl_prop_collector_factories) {
        table_properties_collectors.emplace_back(
            collector_factories->CreateIntTblPropCollector(column_family_id));
      }
    }
  };
  
  BlockBasedTableBuilder::BlockBasedTableBuilder(
      const ImmutableCFOptions& ioptions,
      const BlockBasedTableOptions& table_options,
      const InternalKeyComparator& internal_comparator,
      const std::vector<std::unique_ptr<IntTblPropCollectorFactory>>*
          int_tbl_prop_collector_factories,
      uint32_t column_family_id, WritableFileWriter* file,
      const CompressionType compression_type,
      const CompressionOptions& compression_opts,
      const std::string* compression_dict,
      const std::string& column_family_name) {
    rep_ = new Rep(ioptions, table_options, internal_comparator,
                   int_tbl_prop_collector_factories, column_family_id, file,
                   compression_type, compression_opts, compression_dict,
                   column_family_name);
  }
  
  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* r = rep_;
    assert(!r->closed);
    if (!ok()) return;
    if (r->props.num_entries > 0) {
      assert(r->internal_comparator.Compare(key, Slice(r->last_key)) > 0);
    }
  
    auto should_flush = r->flush_block_policy->Update(key, value);
    if (should_flush) {
      assert(!r->data_block.empty());
      Flush();
  
      // 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.
      if (ok()) {
        r->index_builder->AddIndexEntry(&r->last_key, &key, r->pending_handle);
      }
    }
  
    r->last_key.assign(key.data(), key.size());
    r->data_block.Add(key, value);
    r->props.num_entries++;
    r->props.raw_key_size += key.size();
    r->props.raw_value_size += value.size();
    r->index_builder->OnKeyAdded(key);
    NotifyCollectTableCollectorsOnAdd(key, value, r->offset,
                                      r->table_properties_collectors,
                                      r->ioptions.info_log);
  }
  
  void BlockBasedTableBuilder::Flush() {
    Rep* r = rep_;
    assert(!r->closed);
    if (!ok()) return;
    if (r->data_block.empty()) return;
    WriteBlock(&r->data_block, &r->pending_handle, true /* is_data_block */);
    if (ok()) {
      r->status = r->file->Flush();
    }
    r->props.data_size = r->offset;
    ++r->props.num_data_blocks;
  }
  
  void BlockBasedTableBuilder::WriteBlock(BlockBuilder* block,
                                          BlockHandle* handle,
                                          bool is_data_block) {
    WriteBlock(block->Finish(), handle, is_data_block);
    block->Reset();
  }
  
  void BlockBasedTableBuilder::WriteBlock(const Slice& raw_block_contents,
                                          BlockHandle* handle,
                                          bool is_data_block) {
    // 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) {
      Slice compression_dict;
      if (is_data_block && r->compression_dict && r->compression_dict->size()) {
        compression_dict = *r->compression_dict;
      }
      block_contents = CompressBlock(raw_block_contents, r->compression_opts,
                                     &type, compression_dict,
                                     &r->compressed_output);
    } else {
      RecordTick(r->ioptions.statistics, NUMBER_BLOCK_NOT_COMPRESSED);
      type = kNoCompression;
      block_contents = raw_block_contents;
    }
    WriteRawBlock(block_contents, type, handle);
    r->compressed_output.clear();
  }
  
  void BlockBasedTableBuilder::WriteRawBlock(const Slice& block_contents,
                                             CompressionType type,
                                             BlockHandle* handle) {
    Rep* r = rep_;
    StopWatch sw(r->ioptions.env, r->ioptions.statistics, WRITE_RAW_BLOCK_MICROS);
    handle->set_offset(r->offset);
    handle->set_size(block_contents.size());
    r->status = r->file->Append(block_contents);
    if (r->status.ok()) {
      char trailer[kBlockTrailerSize];
      trailer[0] = type;
      char* trailer_without_type = trailer + 1;
  
      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));
  
      r->status = r->file->Append(Slice(trailer, kBlockTrailerSize));
      if (r->status.ok()) {
        r->offset += block_contents.size() + kBlockTrailerSize;
      }
    }
  }
  
  Status BlockBasedTableBuilder::status() const {
    return rep_->status;
  }
  
  Status BlockBasedTableBuilder::Finish() {
    Rep* r = rep_;
    bool empty_data_block = r->data_block.empty();
    Flush();
    assert(!r->closed);
    r->closed = true;
  
    BlockHandle metaindex_block_handle, index_block_handle,
        compression_dict_block_handle;
  
    // To make sure properties block is able to keep the accurate size of index
    // block, we will finish writing all index entries here and flush them
    // to storage after metaindex block is written.
    if (ok() && !empty_data_block) {
      r->index_builder->AddIndexEntry(
          &r->last_key, nullptr /* no next data block */, r->pending_handle);
    }
  
    IndexBuilder::IndexBlocks index_blocks;
    auto s = r->index_builder->Finish(&index_blocks);
    if (!s.ok()) {
      return s;
    }
  
    // Write meta blocks and metaindex block with the following order.
    //    1. [properties]
    //    2. [compression_dict]
    //    3. [meta_index_builder]
    //    4. [index_blocks]
    MetaIndexBuilder meta_index_builder;
  
    if (ok()) {
      // Write properties and compression dictionary blocks.
      {
        PropertyBlockBuilder property_block_builder;
        r->props.column_family_id = r->column_family_id;
        r->props.column_family_name = r->column_family_name;
        r->props.index_size =
            r->index_builder->EstimatedSize() + kBlockTrailerSize;
        r->props.comparator_name = r->ioptions.comparator != nullptr
                                       ? r->ioptions.comparator->Name()
                                       : "nullptr";
        r->props.compression_name = CompressionTypeToString(r->compression_type);
  
        std::string property_collectors_names = "[";
        property_collectors_names = "[";
        for (size_t i = 0;
             i < r->ioptions.table_properties_collector_factories.size(); ++i) {
          if (i != 0) {
            property_collectors_names += ",";
          }
          property_collectors_names +=
              r->ioptions.table_properties_collector_factories[i]->Name();
        }
        property_collectors_names += "]";
        r->props.property_collectors_names = property_collectors_names;
  
        // Add basic properties
        property_block_builder.AddTableProperty(r->props);
  
        // Add user 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);
        meta_index_builder.Add(kPropertiesBlock, properties_block_handle);
  
        // Write compression dictionary block
        if (r->compression_dict && r->compression_dict->size()) {
          WriteRawBlock(*r->compression_dict, kNoCompression,
                        &compression_dict_block_handle);
          meta_index_builder.Add(kCompressionDictBlock,
                                 compression_dict_block_handle);
        }
      }  // end of properties/compression dictionary block writing
    }    // meta blocks
  
    // Write index block
    if (ok()) {
      // flush the meta index block
      WriteRawBlock(meta_index_builder.Finish(), kNoCompression,
                    &metaindex_block_handle);
      WriteBlock(index_blocks.index_block_contents, &index_block_handle, false);
    }
  
    // Write footer
    if (ok()) {
      // No need to write out new footer if we're using default checksum.
      Footer footer(kBlockBasedTableMagicNumber);
      footer.set_metaindex_handle(metaindex_block_handle);
      footer.set_index_handle(index_block_handle);
      std::string footer_encoding;
      footer.EncodeTo(&footer_encoding);
      r->status = r->file->Append(footer_encoding);
      if (r->status.ok()) {
        r->offset += footer_encoding.size();
      }
    }
  
    return r->status;
  }
  
  void BlockBasedTableBuilder::Abandon() {
    assert(!rep_->closed);
    rep_->closed = true;
  }
  
  uint64_t BlockBasedTableBuilder::NumEntries() const {
    return rep_->props.num_entries;
  }
  
  uint64_t BlockBasedTableBuilder::FileSize() const {
    return rep_->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);
      }
      collector->Finish(&ret.user_collected_properties);
    }
    return ret;
  }
  
  }  // namespace vidardb
  