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 "buffer/Buffer.hh"
#include "ValidSchema.hh"
#include "Specific.hh"
#include "Stream.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();

    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_; }

    DataFileReaderBase(const char* filename);

    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();
};

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);
    }

    DataFileReader(const char* filename) :
        base_(new DataFileReaderBase(filename)) {
        base_->init();
    }

    DataFileReader(std::unique_ptr<InputStream> inputStream) :
        base_(new DataFileReaderBase(std::move(inputStream))) {
        base_->init();
    }

    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