decoded_block.cpp

#include "decoded_block.hpp"
#include "cache.hpp"
#include "drive.hpp"
#include "optimized_rows.hpp"
#include "logic_error.hpp"
#include "constants.hpp"
#include "fast_hash.hpp"

namespace AppenDB {
  /*
    Each entry is in this form:
      32-bit: time
      32-bit: row hash
      32-bit: length
      20-bit: row start
      4-bit: upper index match bloom = BITOR of (1 << ((indexed_field >> 3) & 3))
      8-bit: lower index match bloom = BITOR of (1 << ((indexed_field) & 7))
  */

  /*
    Bloom filter:
      size (in bits): round_up_to_next_power_of_2((# of rows) * 8) bits
      hash: hash64(time + (hash32(row)) << 32)
      first bit index: (hash >> 8) & (size - 1)
      second bit index: (hash >> (8 + log2(size))) & (size - 1)
  */

  /*
    Condition filter:
      store up to 64 conditions
      keep them up-to-date as rows are added
      TRUEs, FALSEs, NOTs, ANDs, XORs and ORs are not stored in bitmap
      1 represents condition match, 0 represents condition not match
      can be used for:
        (1 bit) - nullness of field
        (1 bit) - matching a specific field = value
        (n bits) - matching a specififc field to a range of possible values (or other)
        undefined - not specifically set one way or another

     Using condition filters:
       step 1: mask out bits that don't matter
       step 2: xor bits to flip them
       ...
  */

  static void add_to_bloom(char* bloom_filter, size_t const mask, size_t bloom_filter_power_of_2_size_in_bits, uint64_t hash){
    size_t const hash1 = (hash >> 8) & mask;
    size_t const hash2 = (hash >> (8 + bloom_filter_power_of_2_size_in_bits)) & mask;

    bloom_filter[hash1 >> 3] |= 1 << (hash1 & 7);
    bloom_filter[hash2 >> 3] |= 1 << (hash2 & 7);
  }

  static unsigned int get_power_of_two_size(unsigned int size){
    return ((sizeof(unsigned int) * 8) - __builtin_clz(size));
  }

  static uint64_t get_row_hash(uint32_t time, uint32_t h32){
    return hash64(static_cast<uint64_t>(time) + (static_cast<uint64_t>(h32) << 32));
  }

  DecodedBlock::~DecodedBlock(){
    free_memory(_decoded_rows, row_index_size());
    if(has_bloom()) free_memory(_bloom_filter, bloom_filter_size());
  }

  uint32_t DecodedBlock::get_misc_bloom_bits(size_t i){
    return (0x100000 << ((i >> 3) & 3)) | (0x1000000 << (i & 7));
  }

  namespace {
    struct RowAccumulator {
      // store the data we're going to save in the buffer
      uint32_t buffer[4];

      // for setting the index mask on the row
      uint32_t field_to_index[256];

      uint32_t shard;
      uint32_t shard_mask;

      template<typename RowDefinitionType> RowAccumulator(uint32_t s, uint32_t sm, const RowDefinitionType &definition) : shard(s), shard_mask(sm) {
        // set field_to_index
        for(typename RowDefinitionType::FieldIndexType i = 0; i != definition.num_fields(); ++i){
          typename RowDefinitionType::Field field_meta = definition[i];
          if(field_meta.is_indexed()) field_to_index[i] = (0x100000 << ((i >> 3) & 3)) | (0x1000000 << (i & 7));
          else field_to_index[i] = 0;
        }
      }

      void start(){ buffer[3] = 0; }

      void finish_and_set_hash_size_and_start(uint32_t hash32, size_t size, size_t start_offset){
        buffer[1] = hash32;
        buffer[2] = size;
        buffer[3] |= static_cast<uint32_t>(start_offset);
      }

      template<typename SourceType> void set(size_t index, const SourceType &s){
        // save time
        if(index == 0) buffer[0] = Data::Standard::field_as<uint32_t>(s);

        uint32_t field_index_mask = field_to_index[index];
        if(!field_index_mask) return;
        if(s.is_null()) return;
        if((hash32(s) & shard_mask) == shard) buffer[3] |= field_index_mask;
      }

      template<size_t index, typename SourceType> void set(const SourceType &s){ set(index, s); }
    };
  }

  DecodedBlock::DecodedBlock(Cache::Cacheable::CachePool &pool, uint32_t block_index) : Cache::Cacheable(pool) {
    Drive drive = Drive::get_any_drive();
    _base = drive.block_data_start(block_index);
    size_t const block_start = _base - drive.disk();
    size_t const initial_block_size = drive.block_data_size(block_index);
    drive.data_range_add_lock(block_start, initial_block_size);

    // make sure that we aren't going to try and read data that is not currently in memory
    drive.async_range_ready_notify(block_start, initial_block_size, [this, initial_block_size, block_index](){
      Drive drive = Drive::get_any_drive();
      const char* const block_end = drive.block_data_end(block_index);

      // use a shared, reusable buffer for accumulating data that will be free before a function exits
      UseTempBuffer temp_buffer;
      Entry* temp_decoded_rows = reinterpret_cast<Entry*>(temp_buffer.buffer());

      Table definition = drive.get_table(drive.block_type(block_index));
      uint16_t shard_filter = drive.block_shard_filter(block_index);
      unsigned int num_shard_bits = ((sizeof(unsigned int) * 8) - __builtin_clz(shard_filter)) - 1;
      uint32_t shard_mask = (1 << num_shard_bits) - 1;
      uint32_t shard = shard_filter & shard_mask;
      RowAccumulator row(shard, shard_mask, definition);
      row.start();
      const char *row_begin;
      size_t row_length;
      _num_rows = 0;

      size_t const MAX_OUT_OF_ORDER = 64;
      size_t num_out_of_order = 0;
      uint32_t last_time = 0;
      struct OutOfOrderEntry {
        uint32_t time;
        uint32_t row_index;
      } out_of_order[MAX_OUT_OF_ORDER];
      uint32_t out_of_order_correction_indexes[MAX_OUT_OF_ORDER];

      const char *iterator = _base;
      bool const full_sort_required = OptimizedRows::any_row(definition, row, iterator, block_end, [this, temp_decoded_rows, &row, &row_begin, &row_length, &last_time, &num_out_of_order, &definition, &iterator, &out_of_order, &out_of_order_correction_indexes, block_end](){
        uint64_t h32 = hash32(row_begin, row_length);
        row.finish_and_set_hash_size_and_start(h32, row_length, row_begin - _base);
        ::std::memcpy(temp_decoded_rows + _num_rows, row.buffer, 16);

        uint32_t const time = row.buffer[0];
        if(last_time > time){
          if(num_out_of_order == MAX_OUT_OF_ORDER){
            // we're going to need to sort all these rows, so we will stop looking to see if we have a finite number to sort
            ++_num_rows;
            assert(UseTempBuffer::MAX_SIZE >= (_num_rows * 16));
            row.start();

            OptimizedRows::any_row(definition, row, iterator, block_end, [this, temp_decoded_rows, &row, &row_begin, &row_length](){
              uint64_t h32 = hash32(row_begin, row_length);
              row.finish_and_set_hash_size_and_start(h32, row_length, row_begin - _base);
              ::std::memcpy(temp_decoded_rows + _num_rows, row.buffer, 16);
              ++_num_rows;
              assert(UseTempBuffer::MAX_SIZE >= (_num_rows * 16));
              row.start();
              return false;
            }, row_begin, row_length);
            return true;
          }
          else {
            out_of_order_correction_indexes[num_out_of_order] = _num_rows;
            OutOfOrderEntry& oooe = out_of_order[num_out_of_order++];
            oooe.time = time;
            oooe.row_index = _num_rows;
          }
        }
        else last_time = time;

        ++_num_rows;
        assert(UseTempBuffer::MAX_SIZE >= (_num_rows * 16));
        row.start();
        return false;
      }, row_begin, row_length);

      size_t used_memory = row_index_size();
      _decoded_rows = reinterpret_cast<Entry*>(allocate_memory(used_memory));

      if(!full_sort_required){

        // if we have a few rows out of order, we'll try to merge them in
        if(num_out_of_order){
          emit_performance_warning("Rows stored slightly out of order");
          ::std::sort(out_of_order, out_of_order + num_out_of_order, [](const OutOfOrderEntry &a, const OutOfOrderEntry &b){ return a.time < b.time; });
          size_t read_pointer = 0;
          size_t write_pointer = 0;
          size_t out_of_order_index = 0;
          size_t out_of_order_correction_indexes_index = 0;

          // the main loop for when we're actively merging out of order rows in
          while(out_of_order_index != num_out_of_order){
            // when we run into a row that is in the out of order buffer, skip it since we have it in the other buffer
            if(out_of_order_correction_indexes[out_of_order_correction_indexes_index] == read_pointer){
              ++out_of_order_correction_indexes_index;
              ++read_pointer;
              continue;
            }

            // when we have a row to be inserted from the out of order buffer, insert it
            if(out_of_order[out_of_order_index].time <= temp_decoded_rows[read_pointer].time){
              _decoded_rows[write_pointer++] = temp_decoded_rows[out_of_order[out_of_order_index++].row_index];
              continue;
            }

            // in other cases, just read the next item
            _decoded_rows[write_pointer++] = temp_decoded_rows[read_pointer++];
          }

          // when we no longer have any out of order rows, switch to this slightly faster loop
          while(out_of_order_correction_indexes_index != num_out_of_order){

            // skip any rows that were out of order
            if(out_of_order_correction_indexes[out_of_order_correction_indexes_index] == read_pointer){
              ++out_of_order_correction_indexes_index;
              ++read_pointer;
              continue;
            }

            _decoded_rows[write_pointer++] = temp_decoded_rows[read_pointer++];
          }

          // finish out the rest when there are no more rows to insert and no more rows to skip
          while(read_pointer != _num_rows){
            _decoded_rows[write_pointer++] = temp_decoded_rows[read_pointer++];
          }

          assert(read_pointer == write_pointer);
        }
        else {
          // everything was in order, yay!
          ::std::memcpy(_decoded_rows, temp_decoded_rows, _num_rows * 16);
        }
      }
      else {
        emit_performance_warning("Rows stored majorly out of order");
        ::std::sort(temp_decoded_rows, temp_decoded_rows + _num_rows, [](const Entry &a, const Entry &b){ return a.time < b.time; });
        ::std::memcpy(_decoded_rows, temp_decoded_rows, _num_rows * 16);
      }

      // generate bloom filter
      regenerate_bloom_filter();

      // remove the drive lock
      drive.data_range_remove_lock(_base - drive.disk(), initial_block_size);

      this->set_ready<DecodedBlock>();
    });
  }

  void DecodedBlock::regenerate_bloom_filter(){
    size_t bloom_used_memory = bloom_filter_size();
    _bloom_filter = reinterpret_cast<char*>(allocate_memory(bloom_used_memory));
    ::std::memset(_bloom_filter, 0, bloom_used_memory);
    size_t const num_bits = bloom_filter_bits();
    size_t const bloom_filter_power_of_2_size_in_bits = get_power_of_two_size(num_bits);
    size_t const mask = num_bits - 1;

    for(size_t i = 0; i != _num_rows; ++i){
      uint64_t const h64 = get_row_hash(_decoded_rows[i].time, _decoded_rows[i].hash);
      add_to_bloom(_bloom_filter, mask, bloom_filter_power_of_2_size_in_bits, h64);
    }
  }

  bool DecodedBlock::has_row_in_bloom(uint32_t time, uint32_t h32) const {
    assert(is_ready());
    uint64_t const hash = get_row_hash(time, h32);
    size_t const num_bits = bloom_filter_bits();
    size_t const bloom_filter_power_of_2_size_in_bits = get_power_of_two_size(num_bits);
    size_t const mask = num_bits - 1;

    size_t const hash1 = (hash >> 8) & mask;
    size_t const hash2 = (hash >> (8 + bloom_filter_power_of_2_size_in_bits)) & mask;

    return (
      (_bloom_filter[hash1 >> 3] & (1 << (hash1 & 7)))
        &&
      (_bloom_filter[hash2 >> 3] & (1 << (hash2 & 7)))
    );
  }

  bool DecodedBlock::has_row_in_decoded_rows(uint32_t time, uint64_t r64, const char *data, size_t length) const {
    assert(is_ready());
    uint32_t const hash = hash64_to_32(r64);
    Entry *entries_end = _decoded_rows + _num_rows;

    const Entry* lb = ::std::lower_bound(_decoded_rows, entries_end, time, [](const Entry &entry, const uint32_t &time){ return entry.time < time; });
    for(;;){
      if(lb == entries_end) return false;
      if(lb->time != time) return false;
      if((lb->hash == hash) && (lb->length == length) && !::std::memcmp(_base + lb->row_start_offset(), data, length)) return true;
      ++lb;
    }
  }

  // ensure that we keep the cache up-to-date
  void DecodedBlock::update_cache_for_new_row(uint16_t cache_index, uint32_t block_index, uint32_t time, uint64_t hash_64, uint32_t length, uint32_t misc){
    uint32_t const hash = hash64_to_32(hash_64);
    DecodedBlock* decoded_block = Cache::CachePool::get(cache_index).try_fetch_cacheable<DecodedBlock>(block_index);
    if(decoded_block){
      if(__builtin_popcount(decoded_block->_num_rows) == 1){
        assert(decoded_block->row_index_capacity() == decoded_block->num_rows());

        // increase our capacity
        Entry* new_decoded_rows = reinterpret_cast<Entry*>(decoded_block->allocate_memory(decoded_block->row_index_size() * 2));
        decoded_block->free_memory(decoded_block->_bloom_filter, decoded_block->bloom_filter_size());
        decoded_block->_bloom_filter = nullptr;

        const Entry *source = decoded_block->_decoded_rows;
        Entry *destination = new_decoded_rows;

        // fast path: we can just append our row
        if(source[decoded_block->_num_rows - 1].time <= time){
          ::std::memcpy(destination, source, decoded_block->row_index_size());
          Entry &e = destination[decoded_block->_num_rows];
          e.time = time; e.hash = hash; e.length = length; e.misc = misc;
        }

        // slow path: we must insert the row in the middle as we copy
        else {
          for(;;){
            if(source->time > time) break;
            *destination++ = *source++;
          }

          Entry &e = *destination++;
          e.time = time; e.hash = hash; e.length = length; e.misc = misc;

          ::std::memcpy(destination, source, decoded_block->row_index_size() - (sizeof(Entry) * (source - decoded_block->_decoded_rows)));
        }

        decoded_block->free_memory(decoded_block->_decoded_rows, decoded_block->row_index_size());
        decoded_block->_decoded_rows = new_decoded_rows;
        ++decoded_block->_num_rows;
        decoded_block->regenerate_bloom_filter();
      }

      else {
        assert(decoded_block->row_index_capacity() > decoded_block->num_rows());

        // fast path: we can just append our row
        if((decoded_block->_decoded_rows)[decoded_block->_num_rows - 1].time <= time){
          Entry &e = (decoded_block->_decoded_rows)[decoded_block->_num_rows];
          e.time = time; e.hash = hash; e.length = length; e.misc = misc;
        }

        // slow path: we must insert the row in the middle as we copy
        else {
          Entry* const end_of_existing_entries = decoded_block->_decoded_rows + decoded_block->_num_rows;
          Entry* insert_point = ::std::upper_bound(decoded_block->_decoded_rows, end_of_existing_entries, time, [](uint32_t time, const Entry &entry){ return time < entry.time; });
          ::std::move_backward(insert_point, end_of_existing_entries, end_of_existing_entries + 1);
          
          Entry &e = *insert_point;
          e.time = time; e.hash = hash; e.length = length; e.misc = misc;
        }

        ++decoded_block->_num_rows;

        // add row to the bloom filter
        size_t const num_bits = decoded_block->bloom_filter_bits();
        size_t const bloom_filter_power_of_2_size_in_bits = get_power_of_two_size(num_bits);
        size_t const mask = num_bits - 1;
        add_to_bloom(decoded_block->_bloom_filter, mask, bloom_filter_power_of_2_size_in_bits, get_row_hash(time, hash));
      }
    }
  }
}