Loading...
Searching...
No Matches
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
BitPacking.h
UnpackBoolArray
void UnpackBoolArray(bool *output, const unsigned char *packed_input, size_t n)
Definition BitPacking.h:29
PackBoolArray
void PackBoolArray(unsigned char *packed_buffer, const bool *input, size_t n)
Definition BitPacking.h:6
RowBasedFile.h
casacore::RowBasedFile::WriteData
void WriteData(const unsigned char *data, uint64_t size)
Definition RowBasedFile.h:166
casacore::RowBasedFile::Seek
void Seek(off_t pos, int seek_direction)
Definition RowBasedFile.h:152
casacore::RowBasedFile::DataLocation
uint64_t DataLocation() const
Offset of the first row in the file.
Definition RowBasedFile.h:177
casacore::RowBasedFile::SetStride
void SetStride(uint64_t new_stride)
Set the number of bytes per row for this file.
Definition RowBasedFile.h:219
casacore::RowBasedFile::ReadHeader
void ReadHeader(unsigned char *data)
Read an optional extra header to the file.
Definition RowBasedFile.h:196
casacore::RowBasedFile::Filename
const std::string & Filename() const
Definition RowBasedFile.h:178
casacore::RowBasedFile::RowBasedFile
RowBasedFile()=default
casacore::RowBasedFile::operator=
RowBasedFile & operator=(RowBasedFile &&rhs)
Definition RowBasedFile.h:109
casacore::RowBasedFile::WriteHeader
void WriteHeader(const unsigned char *data)
Write an optional extra header to the file.
Definition RowBasedFile.h:189
casacore::RowBasedFile::AddRows
void AddRows(uint64_t n_rows)
Adds a given number of rows to the back of the file.
Definition RowBasedFile.h:232
casacore::RowBasedFile::NRows
uint64_t NRows() const
Total number of rows stored in this file.
Definition RowBasedFile.h:203
casacore::RowBasedFile::ReadData
void ReadData(unsigned char *data, uint64_t size)
Definition RowBasedFile.h:159
casacore::RowBasedFile::DeleteRow
void DeleteRow()
Deletes the last row.
Definition RowBasedFile.h:237
casacore::RowBasedFile::Stride
uint64_t Stride() const
Total number of bytes in one row.
Definition RowBasedFile.h:214
casacore::RowBasedFile::SetNRows
void SetNRows(uint64_t new_n_rows)
Definition RowBasedFile.h:205
casacore::RowBasedFile::Close
void Close()
Close the file.
Definition RowBasedFile.h:126
casacore::RowBasedFile::IsOpen
bool IsOpen() const
Definition RowBasedFile.h:172
casacore::VarBufferedColumnarFile
Class that provides binary table I/O.
Definition BufferedColumnarFile.h:33
casacore::VarBufferedColumnarFile::CreateNew
static VarBufferedColumnarFile CreateNew(const std::string &filename, uint64_t header_size, uint64_t stride)
Create a new file on disk.
Definition BufferedColumnarFile.h:108
casacore::VarBufferedColumnarFile::OpenExisting
static VarBufferedColumnarFile OpenExisting(const std::string &filename, size_t header_size)
Open an existing file from disk.
Definition BufferedColumnarFile.h:120
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const std::complex< double > *data, uint64_t n)
Write an array of complex doubles.
Definition BufferedColumnarFile.h:212
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const bool *data, uint64_t n)
Write an array of bools.
Definition BufferedColumnarFile.h:221
casacore::VarBufferedColumnarFile::ReadImplementation
void ReadImplementation(uint64_t row, uint64_t column_offset, ValueType *data, uint64_t n)
Definition BufferedColumnarFile.h:290
casacore::VarBufferedColumnarFile::ActivateBlock
void ActivateBlock(uint64_t row)
Definition BufferedColumnarFile.h:267
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const double *data, uint64_t n)
Write an array of doubles.
Definition BufferedColumnarFile.h:188
casacore::VarBufferedColumnarFile::Read
void Read(uint64_t row, uint64_t column_offset, int32_t *data, uint64_t n)
Read array of int32_t.
Definition BufferedColumnarFile.h:148
casacore::VarBufferedColumnarFile< 100 *1024 >::rows_per_block_
uint64_t rows_per_block_
Definition BufferedColumnarFile.h:336
casacore::VarBufferedColumnarFile< 100 *1024 >::active_block_
uint64_t active_block_
Definition BufferedColumnarFile.h:335
casacore::VarBufferedColumnarFile::Read
void Read(uint64_t row, uint64_t column_offset, std::complex< float > *data, uint64_t n)
Read array of complex floats.
Definition BufferedColumnarFile.h:155
casacore::VarBufferedColumnarFile::SetStride
void SetStride(uint64_t new_stride)
Set the number of bytes per row for this file.
Definition BufferedColumnarFile.h:236
casacore::VarBufferedColumnarFile::VarBufferedColumnarFile
VarBufferedColumnarFile() noexcept=default
casacore::VarBufferedColumnarFile::VarBufferedColumnarFile
VarBufferedColumnarFile(const std::string &filename, size_t header_size)
Open an existing columnar file.
Definition BufferedColumnarFile.h:257
casacore::VarBufferedColumnarFile::Read
void Read(uint64_t row, uint64_t column_offset, double *data, uint64_t n)
Read array of doubles.
Definition BufferedColumnarFile.h:141
casacore::VarBufferedColumnarFile::WriteImplementation
void WriteImplementation(uint64_t row, uint64_t column_offset, const ValueType *data, uint64_t n)
Definition BufferedColumnarFile.h:306
casacore::VarBufferedColumnarFile::WriteActiveBlock
void WriteActiveBlock()
Definition BufferedColumnarFile.h:319
casacore::VarBufferedColumnarFile::VarBufferedColumnarFile
VarBufferedColumnarFile(const std::string &filename, uint64_t header_size, uint64_t stride)
Create or overwrite a new columnar file on disk.
Definition BufferedColumnarFile.h:248
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const int32_t *data, uint64_t n)
Write an array of int32_t.
Definition BufferedColumnarFile.h:196
casacore::VarBufferedColumnarFile< 100 *1024 >::block_buffer_
std::vector< unsigned char > block_buffer_
Definition BufferedColumnarFile.h:337
casacore::VarBufferedColumnarFile::operator=
VarBufferedColumnarFile & operator=(VarBufferedColumnarFile &&rhs)
Definition BufferedColumnarFile.h:71
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const std::complex< float > *data, uint64_t n)
Write an array of complex floats.
Definition BufferedColumnarFile.h:204
casacore::VarBufferedColumnarFile::Read
void Read(uint64_t row, uint64_t column_offset, bool *data, uint64_t n)
Read an array of bools.
Definition BufferedColumnarFile.h:164
casacore::VarBufferedColumnarFile::NRows
uint64_t NRows() const
Total number of rows stored in this file.
Definition RowBasedFile.h:203
casacore::VarBufferedColumnarFile::Read
void Read(uint64_t row, uint64_t column_offset, float *data, uint64_t n)
Read one cell containing an array of floats.
Definition BufferedColumnarFile.h:134
casacore::VarBufferedColumnarFile::Write
void Write(uint64_t row, uint64_t column_offset, const float *data, uint64_t n)
Write one cell containing an array of floats.
Definition BufferedColumnarFile.h:180
casacore::VarBufferedColumnarFile::~VarBufferedColumnarFile
~VarBufferedColumnarFile() noexcept
Definition BufferedColumnarFile.h:58
casacore::VarBufferedColumnarFile< 100 *1024 >::packed_buffer_
std::vector< unsigned char > packed_buffer_
Definition BufferedColumnarFile.h:332
casacore::VarBufferedColumnarFile::Stride
uint64_t Stride() const
Total number of bytes in one row.
Definition RowBasedFile.h:214
casacore::VarBufferedColumnarFile< 100 *1024 >::block_changed_
bool block_changed_
Definition BufferedColumnarFile.h:334
casacore::VarBufferedColumnarFile::IsOpen
bool IsOpen() const
Definition RowBasedFile.h:172
casacore::VarBufferedColumnarFile::Close
void Close()
Close the file.
Definition BufferedColumnarFile.h:87
casacore
this file contains all the compiler specific defines
Definition mainpage.dox:28
casacore::max
LatticeExprNode max(const LatticeExprNode &left, const LatticeExprNode &right)
casacore::move
void move(TYPE *target, int npixels) const
casacore::BufferedColumnarFile
VarBufferedColumnarFile< 100 *1024 > BufferedColumnarFile
Definition BufferedColumnarFile.h:340
std
Define real & complex conjugation for non-complex types and put comparisons into std namespace.
Definition Complex.h:350