DataFile.hh

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef avro_DataFile_hh__
#define avro_DataFile_hh__
 
#include "Config.hh"
#include "Encoder.hh"
#include "Specific.hh"
#include "Stream.hh"
#include "ValidSchema.hh"
#include "buffer/Buffer.hh"
 
#include <map>
#include <string>
#include <vector>
 
#include "array"
#include "boost/utility.hpp"
#include <boost/iostreams/filtering_stream.hpp>
 
namespace avro {
 
enum Codec {
    NULL_CODEC,
    DEFLATE_CODEC,
 
#ifdef SNAPPY_CODEC_AVAILABLE
    SNAPPY_CODEC
#endif
 
};
 
const int SyncSize = 16;
typedef std::array<uint8_t, SyncSize> DataFileSync;
 
class AVRO_DECL DataFileWriterBase : boost::noncopyable {
    const std::string filename_;
    const ValidSchema schema_;
    const EncoderPtr encoderPtr_;
    const size_t syncInterval_;
    Codec codec_;
 
    std::unique_ptr<OutputStream> stream_;
    std::unique_ptr<OutputStream> buffer_;
    const DataFileSync sync_;
    int64_t objectCount_;
 
    typedef std::map<std::string, std::vector<uint8_t>> Metadata;
 
    Metadata metadata_;
    int64_t lastSync_;
 
    static std::unique_ptr<OutputStream> makeStream(const char *filename);
    static DataFileSync makeSync();
 
    void writeHeader();
    void setMetadata(const std::string &key, const std::string &value);
 
    void sync();
 
    void init(const ValidSchema &schema, size_t syncInterval, const Codec &codec);
 
public:
    Encoder &encoder() const { return *encoderPtr_; }
 
    void syncIfNeeded();
 
    uint64_t getCurrentBlockStart() const;
 
    void incr() {
        ++objectCount_;
    }
    DataFileWriterBase(const char *filename, const ValidSchema &schema,
                       size_t syncInterval, Codec codec = NULL_CODEC);
    DataFileWriterBase(std::unique_ptr<OutputStream> outputStream,
                       const ValidSchema &schema, size_t syncInterval, Codec codec);
 
    ~DataFileWriterBase();
    void close();
 
    const ValidSchema &schema() const { return schema_; }
 
    void flush();
};
 
template<typename T>
class DataFileWriter : boost::noncopyable {
    std::unique_ptr<DataFileWriterBase> base_;
 
public:
    DataFileWriter(const char *filename, const ValidSchema &schema,
                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC) : base_(new DataFileWriterBase(filename, schema, syncInterval, codec)) {}
 
    DataFileWriter(std::unique_ptr<OutputStream> outputStream, const ValidSchema &schema,
                   size_t syncInterval = 16 * 1024, Codec codec = NULL_CODEC) : base_(new DataFileWriterBase(std::move(outputStream), schema, syncInterval, codec)) {}
 
    void write(const T &datum) {
        base_->syncIfNeeded();
        avro::encode(base_->encoder(), datum);
        base_->incr();
    }
 
    uint64_t getCurrentBlockStart() { return base_->getCurrentBlockStart(); }
 
    void close() { base_->close(); }
 
    const ValidSchema &schema() const { return base_->schema(); }
 
    void flush() { base_->flush(); }
};
 
class AVRO_DECL DataFileReaderBase : boost::noncopyable {
    const std::string filename_;
    const std::unique_ptr<InputStream> stream_;
    const DecoderPtr decoder_;
    int64_t objectCount_;
    bool eof_;
    Codec codec_;
    int64_t blockStart_{};
    int64_t blockEnd_{};
 
    ValidSchema readerSchema_;
    ValidSchema dataSchema_;
    DecoderPtr dataDecoder_;
    std::unique_ptr<InputStream> dataStream_;
    typedef std::map<std::string, std::vector<uint8_t>> Metadata;
 
    Metadata metadata_;
    DataFileSync sync_{};
 
    // for compressed buffer
    std::unique_ptr<boost::iostreams::filtering_istream> os_;
    std::vector<char> compressed_;
    std::string uncompressed;
    void readHeader();
 
    void readDataBlock();
    void doSeek(int64_t position);
 
public:
    Decoder &decoder() { return *dataDecoder_; }
 
    bool hasMore();
 
    void decr() { --objectCount_; }
 
    explicit DataFileReaderBase(const char *filename);
 
    explicit DataFileReaderBase(std::unique_ptr<InputStream> inputStream);
 
    void init();
 
    void init(const ValidSchema &readerSchema);
 
    const ValidSchema &readerSchema() { return readerSchema_; }
 
    const ValidSchema &dataSchema() { return dataSchema_; }
 
    void close();
 
    void seek(int64_t position);
 
    void sync(int64_t position);
 
    bool pastSync(int64_t position);
 
    int64_t previousSync() const;
};
 
template<typename T>
class DataFileReader : boost::noncopyable {
    std::unique_ptr<DataFileReaderBase> base_;
 
public:
    DataFileReader(const char *filename, const ValidSchema &readerSchema) : base_(new DataFileReaderBase(filename)) {
        base_->init(readerSchema);
    }
 
    DataFileReader(std::unique_ptr<InputStream> inputStream, const ValidSchema &readerSchema) : base_(new DataFileReaderBase(std::move(inputStream))) {
        base_->init(readerSchema);
    }
 
    explicit DataFileReader(const char *filename) : base_(new DataFileReaderBase(filename)) {
        base_->init();
    }
 
    explicit DataFileReader(std::unique_ptr<InputStream> inputStream) : base_(new DataFileReaderBase(std::move(inputStream))) {
        base_->init();
    }
 
    explicit DataFileReader(std::unique_ptr<DataFileReaderBase> base) : base_(std::move(base)) {
        base_->init();
    }
 
    DataFileReader(std::unique_ptr<DataFileReaderBase> base,
                   const ValidSchema &readerSchema) : base_(std::move(base)) {
        base_->init(readerSchema);
    }
 
    bool read(T &datum) {
        if (base_->hasMore()) {
            base_->decr();
            avro::decode(base_->decoder(), datum);
            return true;
        }
        return false;
    }
 
    const ValidSchema &readerSchema() { return base_->readerSchema(); }
 
    const ValidSchema &dataSchema() { return base_->dataSchema(); }
 
    void close() { return base_->close(); }
 
    void seek(int64_t position) { base_->seek(position); }
 
    void sync(int64_t position) { base_->sync(position); }
 
    bool pastSync(int64_t position) { return base_->pastSync(position); }
 
    int64_t previousSync() { return base_->previousSync(); }
};
 
} // namespace avro
#endif