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

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

//

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

#pragma once

#include <array>

#include <cstdint>

#include <string>

#include "file/file_prefetch_buffer.h"

#include "file/random_access_file_reader.h"

#include "memory/memory_allocator_impl.h"

#include "options/cf_options.h"

#include "port/malloc.h"

#include "port/port.h"  // noexcept

#include "rocksdb/slice.h"

#include "rocksdb/status.h"

#include "rocksdb/table.h"

#include "util/hash.h"

namespace ROCKSDB_NAMESPACE {

class RandomAccessFile;

struct ReadOptions;

bool ShouldReportDetailedTime(Env* env, Statistics* stats);

// the length of the magic number in bytes.

constexpr uint32_t kMagicNumberLengthByte = 8;

extern const uint64_t kBlockBasedTableMagicNumber;

extern const uint64_t kLegacyPlainTableMagicNumber;

extern const uint64_t kPlainTableMagicNumber;

extern const uint64_t kCuckooTableMagicNumber;

// BlockHandle is a pointer to the extent of a file that stores a data

// block or a meta block.

class BlockHandle {

 public:

  // Creates a block handle with special values indicating "uninitialized,"

  // distinct from the "null" block handle.

  BlockHandle();

  BlockHandle(uint64_t offset, uint64_t size);

  // The offset of the block in the file.

  uint64_t offset() const { return offset_; }

  void set_offset(uint64_t _offset) { offset_ = _offset; }

  // The size of the stored block, this size does not include the block trailer.

  uint64_t size() const { return size_; }

  void set_size(uint64_t _size) { size_ = _size; }

  void EncodeTo(std::string* dst) const;

  char* EncodeTo(char* dst) const;

  Status DecodeFrom(Slice* input);

  Status DecodeSizeFrom(uint64_t offset, Slice* input);

  // Return a string that contains the copy of handle.

  std::string ToString(bool hex = true) const;

  // if the block handle's offset and size are both "0", we will view it

  // as a null block handle that points to no where.

  bool IsNull() const { return offset_ == 0 && size_ == 0; }

  static const BlockHandle& NullBlockHandle() { return kNullBlockHandle; }

  // Maximum encoding length of a BlockHandle

  static constexpr uint32_t kMaxEncodedLength = 2 * kMaxVarint64Length;

  inline bool operator==(const BlockHandle& rhs) const {

    return offset_ == rhs.offset_ && size_ == rhs.size_;

  }

  inline bool operator!=(const BlockHandle& rhs) const {

    return !(*this == rhs);

  }

 private:

  uint64_t offset_;

  uint64_t size_;

  static const BlockHandle kNullBlockHandle;

};

struct EncodedBlockHandle {

  explicit EncodedBlockHandle(const BlockHandle& h) {

    auto end = h.EncodeTo(buffer.data());

    size = end - buffer.data();

  }

  Slice AsSlice() const { return Slice(buffer.data(), size); }

  std::array<char, BlockHandle::kMaxEncodedLength> buffer;

  size_t size;

};

// Value in block-based table file index.

//

// The index entry for block n is: y -> h, [x],

// where: y is some key between the last key of block n (inclusive) and the

// first key of block n+1 (exclusive); h is BlockHandle pointing to block n;

// x, if present, is the first key of block n (unshortened).

// This struct represents the "h, [x]" part.

struct IndexValue {

  BlockHandle handle;

  // Empty means unknown.

  Slice first_internal_key;

  IndexValue() = default;

  IndexValue(BlockHandle _handle, Slice _first_internal_key)

      : handle(_handle), first_internal_key(_first_internal_key) {}

  // have_first_key indicates whether the `first_internal_key` is used.

  // If previous_handle is not null, delta encoding is used;

  // in this case, the two handles must point to consecutive blocks:

  // handle.offset() ==

  //     previous_handle->offset() + previous_handle->size() + kBlockTrailerSize

  void EncodeTo(std::string* dst, bool have_first_key,

                const BlockHandle* previous_handle) const;

  Status DecodeFrom(Slice* input, bool have_first_key,

                    const BlockHandle* previous_handle);

  std::string ToString(bool hex, bool have_first_key) const;

};

// Given a file's base_context_checksum and an offset of a block within that

// file, choose a 32-bit value that is as unique as possible. This value will

// be added to the standard checksum to get a checksum "with context," or can

// be subtracted to "remove" context. Returns zero (no modifier) if feature is

// disabled with base_context_checksum == 0.

inline uint32_t ChecksumModifierForContext(uint32_t base_context_checksum,

                                           uint64_t offset) {

  // To disable on base_context_checksum == 0, we could write

  // `if (base_context_checksum == 0) return 0;` but benchmarking shows

  // measurable performance penalty vs. this: compute the modifier

  // unconditionally and use an "all or nothing" bit mask to enable

  // or disable.

  uint32_t all_or_nothing = uint32_t{0} - (base_context_checksum != 0);

  // Desired properties:

  // (call this function f(b, o) where b = base and o = offset)

  // 1. Fast

  // 2. f(b1, o) == f(b2, o) iff b1 == b2

  //    (Perfectly preserve base entropy)

  // 3. f(b, o1) == f(b, o2) only if o1 == o2 or |o1-o2| >= 4 billion

  //    (Guaranteed uniqueness for nearby offsets)

  // 3. f(b, o + j * 2**32) == f(b, o + k * 2**32) only if j == k

  //    (Upper bits matter, and *aligned* misplacement fails check)

  // 4. f(b1, o) == f(b2, o + x) then preferably not

  //    f(b1, o + y) == f(b2, o + x + y)

  //    (Avoid linearly correlated matches)

  // 5. f(b, o) == 0 depends on both b and o

  //    (No predictable overlap with non-context checksums)

  uint32_t modifier =

      base_context_checksum ^ (Lower32of64(offset) + Upper32of64(offset));

  return modifier & all_or_nothing;

}

constexpr uint32_t kLatestBbtFormatVersion = 7;

// Minimum format version supported for reading SST files in block-based format.

//

// When phasing out old format versions, first increase the write minimum,

// then later (>= 6 mo) increase the read minimum when removing the

// implementation for both read and write.

constexpr uint32_t kMinSupportedBbtFormatVersionForRead = 2;

// Minimum format version supported for writing new SST files in block-based

// format. This should be >= kMinSupportedFormatVersionForRead.

//

// When phasing out old format versions, first increase the write minimum,

// then later (>= 6 mo) increase the read minimum when removing the

// implementation for both read and write.

constexpr uint32_t kMinSupportedBbtFormatVersionForWrite = 2;

static_assert(kMinSupportedBbtFormatVersionForWrite >=

              kMinSupportedBbtFormatVersionForRead);

inline bool IsSupportedFormatVersionForRead(uint64_t magic, uint32_t version) {

  if (magic == kBlockBasedTableMagicNumber) {

    return version >= kMinSupportedBbtFormatVersionForRead &&

           version <= kLatestBbtFormatVersion;

  } else if (magic == kPlainTableMagicNumber) {

    return version == 0;

  } else if (magic == kCuckooTableMagicNumber) {

    return version == 1;

  } else {

    return false;

  }

}

inline bool IsSupportedFormatVersionForWrite(uint64_t magic, uint32_t version) {

  if (magic == kBlockBasedTableMagicNumber) {

    return version >= kMinSupportedBbtFormatVersionForWrite &&

           version <= kLatestBbtFormatVersion;

  } else if (magic == kPlainTableMagicNumber) {

    return version == 0;

  } else if (magic == kCuckooTableMagicNumber) {

    return version == 1;

  } else {

    return false;

  }

}

// Same as having a unique id in footer.

inline bool FormatVersionUsesContextChecksum(uint32_t version) {

  return version >= 6;

}

inline bool FormatVersionUsesIndexHandleInFooter(uint32_t version) {

  return version < 6;

}

inline bool FormatVersionUsesCompressionManagerName(uint32_t version) {

  return version >= 7;

}

// Footer encapsulates the fixed information stored at the tail end of every

// SST file. In general, it should only include things that cannot go

// elsewhere under the metaindex block. For example, checksum_type is

// required for verifying metaindex block checksum (when applicable), but

// index block handle can easily go in metaindex block. See also FooterBuilder

// below.

class Footer {

 public:

  // Create empty. Populate using DecodeFrom.

  Footer() {}

  void Reset() {

    table_magic_number_ = kNullTableMagicNumber;

    format_version_ = kInvalidFormatVersion;

    base_context_checksum_ = 0;

    metaindex_handle_ = BlockHandle::NullBlockHandle();

    index_handle_ = BlockHandle::NullBlockHandle();

    checksum_type_ = kInvalidChecksumType;

    block_trailer_size_ = 0;

  }

  // Deserialize a footer (populate fields) from `input` and check for various

  // corruptions. `input_offset` is the offset within the target file of

  // `input` buffer, which is needed for verifying format_version >= 6 footer.

  // If enforce_table_magic_number != 0, will return corruption if table magic

  // number is not equal to enforce_table_magic_number.

  Status DecodeFrom(Slice input, uint64_t input_offset,

                    uint64_t enforce_table_magic_number = 0);

  // Table magic number identifies file as RocksDB SST file and which kind of

  // SST format is use.

  uint64_t table_magic_number() const { return table_magic_number_; }

  // A version (footer and more) within a kind of SST. (It would add more

  // unnecessary complexity to separate footer versions and

  // BBTO::format_version.)

  uint32_t format_version() const { return format_version_; }

  // See ChecksumModifierForContext()

  uint32_t base_context_checksum() const { return base_context_checksum_; }

  // Block handle for metaindex block.

  const BlockHandle& metaindex_handle() const { return metaindex_handle_; }

  // Block handle for (top-level) index block.

  // TODO? remove from this struct and only read on decode for legacy cases

  const BlockHandle& index_handle() const { return index_handle_; }

  // Checksum type used in the file, including footer for format version >= 6.

  ChecksumType checksum_type() const {

    return static_cast<ChecksumType>(checksum_type_);

  }

  // Block trailer size used by file with this footer (e.g. 5 for block-based

  // table and 0 for plain table). This is inferred from magic number so

  // not in the serialized form.

  inline size_t GetBlockTrailerSize() const { return block_trailer_size_; }

  // Convert this object to a human readable form

  std::string ToString() const;

  // Encoded lengths of Footers. Bytes for serialized Footer will always be

  // >= kMinEncodedLength and <= kMaxEncodedLength.

  //

  // Footer version 0 (legacy) will always occupy exactly this many bytes.

  // It consists of two block handles, padding, and a magic number.

  static constexpr uint32_t kVersion0EncodedLength =

      2 * BlockHandle::kMaxEncodedLength + kMagicNumberLengthByte;

  static constexpr uint32_t kMinEncodedLength = kVersion0EncodedLength;

  // Footer of versions 1 and higher will always occupy exactly this many

  // bytes. It originally consisted of the checksum type, two block handles,

  // padding (to maximum handle encoding size), a format version number, and a

  // magic number.

  static constexpr uint32_t kNewVersionsEncodedLength =

      1 + 2 * BlockHandle::kMaxEncodedLength + 4 + kMagicNumberLengthByte;

  static constexpr uint32_t kMaxEncodedLength = kNewVersionsEncodedLength;

  static constexpr uint64_t kNullTableMagicNumber = 0;

  static constexpr uint32_t kInvalidFormatVersion = 0xffffffffU;

 private:

  static constexpr int kInvalidChecksumType =

      (1 << (sizeof(ChecksumType) * 8)) | kNoChecksum;

  uint64_t table_magic_number_ = kNullTableMagicNumber;

  uint32_t format_version_ = kInvalidFormatVersion;

  uint32_t base_context_checksum_ = 0;

  BlockHandle metaindex_handle_;

  BlockHandle index_handle_;

  int checksum_type_ = kInvalidChecksumType;

  uint8_t block_trailer_size_ = 0;

};

// Builder for Footer

class FooterBuilder {

 public:

  // Run builder in inputs. This is a single step with lots of parameters for

  // efficiency (based on perf testing).

  // * table_magic_number identifies file as RocksDB SST file and which kind of

  // SST format is use.

  // * format_version is a version for the footer and can also apply to other

  // aspects of the SST file (see BlockBasedTableOptions::format_version).

  // NOTE: To save complexity in the caller, when format_version == 0 and

  // there is a corresponding legacy magic number to the one specified, the

  // legacy magic number will be written for forward compatibility.

  // * footer_offset is the file offset where the footer will be written

  // (for future use).

  // * checksum_type is for formats using block checksums.

  // * index_handle is optional for some SST kinds and (for caller convenience)

  // ignored when format_version >= 6. (Must be added to metaindex in that

  // case.)

  // * unique_id must be specified if format_vesion >= 6 and SST uses block

  // checksums with context. Otherwise, auto-generated if format_vesion >= 6.

  Status Build(uint64_t table_magic_number, uint32_t format_version,

               uint64_t footer_offset, ChecksumType checksum_type,

               const BlockHandle& metaindex_handle,

               const BlockHandle& index_handle = BlockHandle::NullBlockHandle(),

               uint32_t base_context_checksum = 0);

  // After Builder, get a Slice for the serialized Footer, backed by this

  // FooterBuilder.

  const Slice& GetSlice() const {

    assert(slice_.size());

    return slice_;

  }

 private:

  Slice slice_;

  std::array<char, Footer::kMaxEncodedLength> data_;

};

// Set to true to allow unit testing of writing unsupported block-based table

// format versions (to test read side)

bool& TEST_AllowUnsupportedFormatVersion();

// Read the footer from file

// If enforce_table_magic_number != 0, ReadFooterFromFile() will return

// corruption if table_magic number is not equal to enforce_table_magic_number

Status ReadFooterFromFile(const IOOptions& opts, RandomAccessFileReader* file,

                          FileSystem& fs, FilePrefetchBuffer* prefetch_buffer,

                          uint64_t file_size, Footer* footer,

                          uint64_t enforce_table_magic_number = 0,

                          Statistics* stats = nullptr);

// Computes a checksum using the given ChecksumType. Sometimes we need to

// include one more input byte logically at the end but not part of the main

// data buffer. If data_size >= 1, then

// ComputeBuiltinChecksum(type, data, size)

// ==

// ComputeBuiltinChecksumWithLastByte(type, data, size - 1, data[size - 1])

uint32_t ComputeBuiltinChecksum(ChecksumType type, const char* data,

                                size_t size);

uint32_t ComputeBuiltinChecksumWithLastByte(ChecksumType type, const char* data,

                                            size_t size, char last_byte);

// Represents the contents of a block read from an SST file. Depending on how

// it's created, it may or may not own the actual block bytes. As an example,

// BlockContents objects representing data read from mmapped files only point

// into the mmapped region. Depending on context, it might be a serialized

// (potentially compressed) block, including a trailer beyond `size`, or an

// uncompressed block.

//

// Please try to use this terminology when dealing with blocks:

// * "Serialized block" - bytes that go into storage. For block-based table

// (usually the case) this includes the block trailer. Here the `size` does

// not include the trailer, but other places in code might include the trailer

// in the size.

// * "Maybe compressed block" - like a serialized block, but without the

// trailer (or no promise of including a trailer). Must be accompanied by a

// CompressionType in some other variable or field.

// * "Uncompressed block" - "payload" bytes that are either stored with no

// compression, used as input to compression function, or result of

// decompression function.

// * "Parsed block" - an in-memory form of a block in block cache, as it is

// used by the table reader. Different C++ types are used depending on the

// block type (see block_cache.h). Only trivially parsable block types

// use BlockContents as the parsed form.

//

struct BlockContents {

  // Points to block payload (without trailer)

  Slice data;

  CacheAllocationPtr allocation;

#ifndef NDEBUG

  // Whether there is a known trailer after what is pointed to by `data`.

  // See BlockBasedTable::GetCompressionType.

  bool has_trailer = false;

#endif  // NDEBUG

  BlockContents() {}

  // Does not take ownership of the underlying data bytes.

  BlockContents(const Slice& _data) : data(_data) {}

  // Takes ownership of the underlying data bytes.

  BlockContents(CacheAllocationPtr&& _data, size_t _size)

      : data(_data.get(), _size), allocation(std::move(_data)) {}

  // Takes ownership of the underlying data bytes.

  BlockContents(std::unique_ptr<char[]>&& _data, size_t _size)

      : data(_data.get(), _size) {

    allocation.reset(_data.release());

  }

  // Returns whether the object has ownership of the underlying data bytes.

  bool own_bytes() const { return allocation.get() != nullptr; }

  // The additional memory space taken by the block data.

  size_t usable_size() const {

    // FIXME: doesn't account for possible block trailer

    if (allocation.get() != nullptr) {

      auto allocator = allocation.get_deleter().allocator;

      if (allocator) {

        return allocator->UsableSize(allocation.get(), data.size());

      }

#ifdef ROCKSDB_MALLOC_USABLE_SIZE

      return malloc_usable_size(allocation.get());

#else

      return data.size();

#endif  // ROCKSDB_MALLOC_USABLE_SIZE

    } else {

      return 0;  // no extra memory is occupied by the data

    }

  }

  size_t ApproximateMemoryUsage() const {

    return usable_size() + sizeof(*this);

  }

  BlockContents(BlockContents&& other) noexcept { *this = std::move(other); }

  BlockContents& operator=(BlockContents&& other) {

    data = std::move(other.data);

    allocation = std::move(other.allocation);

#ifndef NDEBUG

    has_trailer = other.has_trailer;

#endif  // NDEBUG

    return *this;

  }

};

// The `data` points to serialized block contents read in from file, which

// must be compressed and include a trailer beyond `size`. A new buffer is

// allocated with the given allocator (or default) and the uncompressed

// contents are returned in `out_contents`. Statistics updated.

Status DecompressSerializedBlock(const char* data, size_t size,

                                 CompressionType type,

                                 Decompressor& decompressor,

                                 BlockContents* out_contents,

                                 const ImmutableOptions& ioptions,

                                 MemoryAllocator* allocator = nullptr);

Status DecompressSerializedBlock(Decompressor::Args& args,

                                 Decompressor& decompressor,

                                 BlockContents* out_contents,

                                 const ImmutableOptions& ioptions,

                                 MemoryAllocator* allocator = nullptr);

// This is a variant of DecompressSerializedBlock that does not expect a

// block trailer beyond `size`. (CompressionType is passed in.)

Status DecompressBlockData(

    const char* data, size_t size, CompressionType type,

    Decompressor& decompressor, BlockContents* out_contents,

    const ImmutableOptions& ioptions, MemoryAllocator* allocator = nullptr,

    Decompressor::ManagedWorkingArea* working_area = nullptr);

Status DecompressBlockData(Decompressor::Args& args, Decompressor& decompressor,

                           BlockContents* out_contents,

                           const ImmutableOptions& ioptions,

                           MemoryAllocator* allocator = nullptr);

// Replace db_host_id contents with the real hostname if necessary

Status ReifyDbHostIdProperty(Env* env, std::string* db_host_id);

// Implementation details follow.  Clients should ignore,

// TODO(andrewkr): we should prefer one way of representing a null/uninitialized

// BlockHandle. Currently we use zeros for null and use negation-of-zeros for

// uninitialized.

inline BlockHandle::BlockHandle() : BlockHandle(~uint64_t{0}, ~uint64_t{0}) {}

inline BlockHandle::BlockHandle(uint64_t _offset, uint64_t _size)

    : offset_(_offset), size_(_size) {}

}  // namespace ROCKSDB_NAMESPACE