BufferedColumnarFile.h

Go to the documentation of this file.

#ifndef CASACORE_BUFFERED_COLUMNAR_FILE_H_
#define CASACORE_BUFFERED_COLUMNAR_FILE_H_
 
#include <cassert>
#include <complex>
#include <cstdint>
#include <string>
#include <vector>
 
#include "BitPacking.h"
#include "RowBasedFile.h"
 
namespace casacore {
 
template <uint64_t BufferSize = 100 * 1024>
class VarBufferedColumnarFile : private RowBasedFile {
 public:
  using RowBasedFile::AddRows;
  using RowBasedFile::DeleteRow;
  using RowBasedFile::Filename;
  using RowBasedFile::IsOpen;
  using RowBasedFile::NRows;
  using RowBasedFile::ReadHeader;
  using RowBasedFile::Stride;
  using RowBasedFile::WriteHeader;
 
  VarBufferedColumnarFile() noexcept = default;
 
  VarBufferedColumnarFile(const VarBufferedColumnarFile& rhs) = delete;
  VarBufferedColumnarFile(VarBufferedColumnarFile&& rhs) noexcept
      : packed_buffer_(std::move(rhs.packed_buffer_)),
        block_changed_(rhs.block_changed_),
        active_block_(rhs.active_block_),
        rows_per_block_(rhs.rows_per_block_),
        block_buffer_(std::move(rhs.block_buffer_)) {
    rhs.block_changed_ = false;
    rhs.active_block_ = 0;
    rhs.rows_per_block_ = 0;
  }
 
  ~VarBufferedColumnarFile() noexcept {
    if (IsOpen()) {
      if (block_changed_) {
        const uint64_t start_row = active_block_ * rows_per_block_;
        const size_t n_rows_to_write =
            std::min(rows_per_block_, std::max(NRows(), start_row) - start_row);
        Seek(start_row * Stride() + DataLocation(), SEEK_SET);
        WriteData(block_buffer_.data(), n_rows_to_write * Stride());
        block_changed_ = false;
      }
    }
  }
 
  VarBufferedColumnarFile& operator=(VarBufferedColumnarFile&& rhs) {
    Close();
    RowBasedFile::operator=(std::move(rhs));
    std::swap(packed_buffer_, rhs.packed_buffer_);
    std::swap(block_changed_, rhs.block_changed_);
    std::swap(active_block_, rhs.active_block_);
    std::swap(rows_per_block_, rhs.rows_per_block_);
    std::swap(block_buffer_, rhs.block_buffer_);
    return *this;
  }
 
  void Close() {
    if (IsOpen()) {
      if (block_changed_) {
        try {
          WriteActiveBlock();
        } catch (...) {
          // Even if writing fails, the file should still be closed to
          // prevent the file to remain open after destruction.
          RowBasedFile::Close();
          throw;
        }
      }
      RowBasedFile::Close();
    }
  }
 
  static VarBufferedColumnarFile CreateNew(const std::string& filename,
                                           uint64_t header_size,
                                           uint64_t stride) {
    return VarBufferedColumnarFile(filename, header_size, stride);
  }
 
  static VarBufferedColumnarFile OpenExisting(const std::string& filename,
                                              size_t header_size) {
    return VarBufferedColumnarFile(filename, header_size);
  }
 
  void Read(uint64_t row, uint64_t column_offset, float* data, uint64_t n) {
    ReadImplementation(row, column_offset, data, n);
  }
 
  void Read(uint64_t row, uint64_t column_offset, double* data, uint64_t n) {
    ReadImplementation(row, column_offset, data, n);
  }
 
  void Read(uint64_t row, uint64_t column_offset, int32_t* data, uint64_t n) {
    ReadImplementation(row, column_offset, data, n);
  }
 
  void Read(uint64_t row, uint64_t column_offset, std::complex<float>* data,
            uint64_t n) {
    ReadImplementation(row, column_offset, data, n);
  }
 
  void Read(uint64_t row, uint64_t column_offset, bool* data, uint64_t n) {
    const size_t byte_size = (n + 7) / 8;
    assert(column_offset + byte_size <= Stride());
    ActivateBlock(row);
    if (row >= NRows()) {
      std::fill_n(data, n, false);
    } else {
      Seek(row * Stride() + column_offset + DataLocation(), SEEK_SET);
      ReadData(packed_buffer_.data(), byte_size);
      UnpackBoolArray(data, packed_buffer_.data(), n);
    }
  }
  void Write(uint64_t row, uint64_t column_offset, const float* data,
             uint64_t n) {
    WriteImplementation(row, column_offset, data, n);
  }
 
  void Write(uint64_t row, uint64_t column_offset, const double* data,
             uint64_t n) {
    WriteImplementation(row, column_offset, data, n);
  }
 
  void Write(uint64_t row, uint64_t column_offset, const int32_t* data,
             uint64_t n) {
    WriteImplementation(row, column_offset, data, n);
  }
 
  void Write(uint64_t row, uint64_t column_offset,
             const std::complex<float>* data, uint64_t n) {
    WriteImplementation(row, column_offset, data, n);
  }
 
  void Write(uint64_t row, uint64_t column_offset,
             const std::complex<double>* data, uint64_t n) {
    WriteImplementation(row, column_offset, data, n);
  }
 
  void Write(uint64_t row, uint64_t column_offset, const bool* data,
             uint64_t n) {
    const size_t byte_size = (n + 7) / 8;
    assert(column_offset + byte_size <= Stride());
    ActivateBlock(row);
    PackBoolArray(packed_buffer_.data(), data, n);
    Seek(row * Stride() + column_offset + DataLocation(), SEEK_SET);
    WriteData(packed_buffer_.data(), byte_size);
    SetNRows(std::max(row + 1, NRows()));
  }
 
  void SetStride(uint64_t new_stride) {
    RowBasedFile::SetStride(new_stride);
    packed_buffer_.resize(new_stride);
    active_block_ = std::numeric_limits<uint64_t>::max();
    rows_per_block_ =
        new_stride == 0 ? 0 : std::max<size_t>(1, BufferSize / new_stride);
    block_buffer_.resize(rows_per_block_ * new_stride);
    block_changed_ = false;
  }
 
 private:
  // Create or overwrite a new columnar file on disk
  VarBufferedColumnarFile(const std::string& filename, uint64_t header_size,
                          uint64_t stride)
      : RowBasedFile(filename, header_size, stride),
        packed_buffer_(stride),
        rows_per_block_(stride == 0 ? 0
                                    : std::max<size_t>(1, BufferSize / stride)),
        block_buffer_(rows_per_block_ * stride) {}
 
  // Open an existing columnar file
  VarBufferedColumnarFile(const std::string& filename, size_t header_size)
      : RowBasedFile(filename, header_size) {
    if (Stride() != 0) {
      packed_buffer_.resize(Stride());
      rows_per_block_ = std::max<size_t>(1, BufferSize / Stride());
      block_buffer_.resize(rows_per_block_ * Stride());
    }
    active_block_ = std::numeric_limits<uint64_t>::max();
  }
 
  void ActivateBlock(uint64_t row) {
    const uint64_t block = row / rows_per_block_;
    if (active_block_ != block) {
      if (block_changed_) {
        WriteActiveBlock();
      }
 
      const uint64_t start_row = block * rows_per_block_;
      const size_t n_rows_to_read =
          std::min(rows_per_block_, std::max(NRows(), start_row) - start_row);
      Seek(start_row * Stride() + DataLocation(), SEEK_SET);
      ReadData(block_buffer_.data(), n_rows_to_read * Stride());
      // Fill the remainder of block_buffer_ with zeroes. Doing it here makes
      // the code robust and avoids the need for inserting zeroes when adding
      // rows out-of-order, e.g., when adding row 5 while NRows() is 2."
      std::fill(block_buffer_.begin() + n_rows_to_read * Stride(),
                block_buffer_.end(), 0);
 
      active_block_ = block;
    }
  }
 
  template <typename ValueType>
  void ReadImplementation(uint64_t row, uint64_t column_offset, ValueType* data,
                          uint64_t n) {
    assert(column_offset + n * sizeof(ValueType) <= Stride());
    if (row >= NRows()) {
      std::fill_n(data, n, ValueType());
    } else {
      ActivateBlock(row);
      const uint64_t block_row = active_block_ * rows_per_block_;
      const unsigned char* position =
          block_buffer_.data() + (row - block_row) * Stride() + column_offset;
      std::copy_n(position, n * sizeof(ValueType),
                  reinterpret_cast<unsigned char*>(data));
    }
  }
 
  template <typename ValueType>
  void WriteImplementation(uint64_t row, uint64_t column_offset,
                           const ValueType* data, uint64_t n) {
    assert(column_offset + n * sizeof(ValueType) <= Stride());
    ActivateBlock(row);
    const uint64_t block_row = active_block_ * rows_per_block_;
    unsigned char* position =
        block_buffer_.data() + (row - block_row) * Stride() + column_offset;
    std::copy_n(reinterpret_cast<const unsigned char*>(data),
                n * sizeof(ValueType), position);
    SetNRows(std::max(row + 1, NRows()));
    block_changed_ = true;
  }
 
  void WriteActiveBlock() {
    const uint64_t start_row = active_block_ * rows_per_block_;
    const size_t n_rows_to_write =
        std::min(rows_per_block_, std::max(NRows(), start_row) - start_row);
    Seek(start_row * Stride() + DataLocation(), SEEK_SET);
    WriteData(block_buffer_.data(), n_rows_to_write * Stride());
    block_changed_ = false;
  }
 
  std::vector<unsigned char> packed_buffer_;
 
  bool block_changed_ = false;
  uint64_t active_block_ = 0;
  uint64_t rows_per_block_ = 0;
  std::vector<unsigned char> block_buffer_;
};
 
using BufferedColumnarFile = VarBufferedColumnarFile<100 * 1024>;
 
}  // namespace casacore
 
#endif