<commit_before>#include <sys/types.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <cstdint>
#include <cstring>
#include <fstream>
#include <stdexcept>
#include <iostream>
#include <unordered_map>

#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filter/zlib.hpp>

#include <boost/algorithm/string.hpp>

#include <boost/lexical_cast.hpp>

#include <filter/filter.h>
#include <filter/equality_expression.h>

#include "schemareader.h"
#include "limiter.h"

#include "node/node.h"
#include "node/record.h"
#include "node/union.h"
#include "node/string.h"
#include "node/custom.h"
#include "node/int.h"
#include "node/long.h"
#include "node/array.h"
#include "node/boolean.h"
#include "node/enum.h"
#include "node/float.h"
#include "node/double.h"
#include "node/map.h"
#include "node/null.h"

#include "deflatedbuffer.h"
#include "filehandler.h"
#include "predicate/predicate.h"
#include "stringbuffer.h"
#include "zigzag.hpp"
#include "reader.h"

namespace {
const std::string AVRO_MAGICK = "Obj\001"; // 4 bytes
}

namespace avro {

class Reader::Private {
public:
    std::string filename; // TODO: move to FileHandle

    std::unique_ptr<StringBuffer> input;

    DeflatedBuffer deflate_buffer;
    Limiter &limit;

    FileHandle file;
    std::shared_ptr<filter::Filter> filter;
    std::unordered_map<const Node *, std::shared_ptr<predicate::Predicate>> filterItems;

    Private(const std::string& filename, Limiter &limit) : filename(filename), limit(limit), file(filename) {
    }
};

Reader::Reader(const std::string& filename, Limiter &limit) :
    d(new Private(filename, limit)) {

    d->input = d->file.mmapFile();
}

Reader::~Reader() {
    delete d;
}


header Reader::readHeader() {

    
    std::string magick;
    magick.resize(AVRO_MAGICK.size());
    d->input->read(&magick[0], AVRO_MAGICK.size());
    
    if (magick != AVRO_MAGICK) {
        throw std::runtime_error("Bad avro file (wrong magick)");
    }

    header header;


    int64_t recordsNumber = readZigZagLong(*d->input);
    
    for(uint i = 0; i < recordsNumber; ++i) {

        size_t len = readZigZagLong(*d->input);
        const std::string &key = d->input->getStdString(len);
        const std::string &value = d->input->getStdString(readZigZagLong(*d->input));

        header.metadata[key] = value;

    }

    SchemaReader schemaReader(header.metadata["avro.schema"]);
    header.schema = schemaReader.parse();
    header.nodesNumber = schemaReader.nodesNumber();
    //dumpShema(schemaRoot);
    
    char c = d->input->getChar();

    assert(c == 0); // TODO: what is it?
    d->input->read(&header.sync[0], sizeof header.sync);


    return header;

}
constexpr static const char *indents[] = {
        "",
        "\t",
        "\t\t",
        "\t\t\t",
        "\t\t\t\t",
        "\t\t\t\t\t",
        "\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t\t\t",
    };

class DocumentDumper {
public:
    void String(const StringBuffer &s, const node::String &n) {
        std::cout << indents[level] << n.getItemName() << ": \"" << s << '"' << std::endl;
    }

    void MapName(const StringBuffer &name) {
        std::cout << indents[level] << name;
    }

    void MapValue(const StringBuffer &s, const node::String &n) {
        std::cout << ": \"" << s << "\"" << std::endl;
    }

    void MapValue(int i, const node::Int &n) {
        std::cout << ": " << i << std::endl;
    }

    void Int(int i, const node::Int &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << i << std::endl;
    }

    void Long(long l, const node::Long &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << l << std::endl;
    }

    void Float(float f, const node::Float &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << f << std::endl;
    }

    void Double(double d, const node::Double &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << d << std::endl;
    }

    void Boolean(bool b, const node::Boolean &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << b << std::endl;
    }

    void Null(const node::Null &n) {
        std::cout << indents[level] << n.getItemName() << ": null" << std::endl;
    }

    void RecordBegin(const node::Record &r) {
        std::cout << "{\n";
        level++;
    }

    void RecordEnd(const node::Record &r) {
        --level;
        std::cout << indents[level] << "}\n";
    }

    void ArrayBegin(const node::Array &a) {
        std::cout << "[\n";
        level++;

    }

    void ArrayEnd(const node::Array &a) {
        level--;
        std::cout << indents[level] << "]\n";
    }

    void CustomBegin(const node::Custom &c) {
        std::cout << indents[level] << c.getItemName();
    }

    void Enum(const node::Enum &e, int index) {
        const auto &value = e[index];
        std::cout << ": \"" << value  << '"' << std::endl;
    }

    void MapBegin(const node::Map &m) {
        std::cout << "{\n";
        level++;
    }

    void MapEnd(const node::Map &m) {
        level--;
        std::cout << indents[level] << "}\n";
    }

    void EndDocument() {
    }

private:
    int level = 0;

};

class Dumper {
public:
    virtual ~Dumper() {}
    virtual void dump(std::ostream &os) = 0;
};

template<typename T>
class TDumper : public Dumper {
public:

    TDumper(const T &t) : Dumper(), t(t) {
    }
    void dump(std::ostream &os) {
        os << t;
    }
private:
    T t;
};

class TsvDumper {
public:
    TsvDumper(const TsvExpression &wd) : whatDump(wd) {
    	toDump.resize(whatDump.pos);
    	// std::cout << "TsvDumper() " << whatDump.what.size() << std::endl;
    }

    template<typename T, typename NodeType>
    void addIfNecessary(const T &t, const NodeType &n) {
        // std::cout << "sz " << whatDump.what.size() << std::endl;
    	auto p = whatDump.what.find(n.getNumber());
    	if (p != whatDump.what.end()) {
            // std::cout << "adding " << n.getItemName() << " num=" << n.getNumber() << " pos = " << p->second << std::endl;

    		toDump[p->second].reset(new TDumper<T>(t));
    	} else {
            // std::cout << "NOT adding " << n.getItemName() << " num=" << n.getNumber() << " pos = " << p->second << " addIfNecessary() " << whatDump.what.size() << std::endl;
    	}
    }

    void String(const StringBuffer &s, const node::String &n) {
    	addIfNecessary(s, n);
    }

    void MapName(const StringBuffer &name) {
        // std::cout << indents[level] << s;
    }

    void MapValue(const StringBuffer &s, const node::String &n) {
        //std::cout << ": \"" << s  << "\"" << std::endl;
    }
    void MapValue(int i, const node::Int &n) {
        //std::cout << ": \"" << s  << "\"" << std::endl;
    }

    void Int(int i, const node::Int &n) {
    	addIfNecessary(i, n);
    }

    void Long(long l, const node::Long &n) {
    	addIfNecessary(l, n);
    }

    void Float(float f, const node::Float &n) {
    	addIfNecessary(f, n);
    }

    void Double(double d, const node::Double &n) {
    	addIfNecessary(d, n);
    }

    void Boolean(bool b, const node::Boolean &n) {
    	addIfNecessary(b, n);
    }

    void Null(const node::Null &n) {
    	addIfNecessary(std::string("null"), n);
    }

    void Union(int index, const node::Union &n) {

    }

    void RecordBegin(const node::Record &r) {
    }

    void RecordEnd(const node::Record &r) {
    }

    void ArrayBegin(const node::Array &a) {
    }

    void ArrayEnd(const node::Array &a) {
    }

    void CustomBegin(const node::Custom &c) {
    }

    void Enum(const node::Enum &e, int index) {
    	addIfNecessary(e[index], e);
    }

    void MapBegin(const node::Map &m) {
    }

    void MapEnd(const node::Map &m) {
    }

    void EndDocument() {
    	auto p = toDump.begin();
    	(*p)->dump(std::cout);
    	// std::cout << *p;
    	++p;
    	while(p != toDump.end()) {
    		std::cout << "\t";
    		(*p)->dump(std::cout);
    		++p;
    	}
    	std::cout << std::endl;
    }

private:
    std::vector<std::unique_ptr<Dumper>> toDump;
    const TsvExpression &whatDump;
};



void Reader::setFilter(std::shared_ptr<filter::Filter> filter, const header &header) {
    d->filter = filter;

    for(auto &filterPredicate : filter->getPredicates()) {
        const Node * filterNode = schemaNodeByPath(filterPredicate->identifier, header);

        if (filterNode->is<node::Union>()) {
            for( auto &p : filterNode->as<node::Union>().getChildren()) {
                if (p->is<node::String>()) {
                    filterNode = p.get();
                    break;
                } else if (p->is<node::Int>() || p->is<node::Long>()) {
                    filterNode = p.get();
                    break;
                }
            }
        }
        d->filterItems[filterNode] = std::make_shared<predicate::Predicate>(filterPredicate);
    }
}

void Reader::readBlock(const header &header, const FilterExpression *filter, const TsvExpression &wd) {

    int64_t objectCountInBlock = readZigZagLong(*d->input);
    int64_t blockBytesNum = readZigZagLong(*d->input);

    if (header.metadata.at("avro.codec") == "deflate") { // TODO: check it once

        d->deflate_buffer.assignData(d->input->getAndSkip(blockBytesNum), blockBytesNum);

        for(int i = 0; i < objectCountInBlock; ++i) {
            // TODO: rewrite it using hierarcy of filters/decoders.
            // TODO: implement counter as a filter  
            if (!filter && false) {
                // dumpDocument(d->deflate_buffer, header.schema, 0);
                if (wd.pos > 0) {
                    auto dumper = TsvDumper(wd);
                    // std::cout << "readBlock " << wd.what.size() << std::endl;
                    dumpDocument(d->deflate_buffer, header.schema, dumper);
                    dumper.EndDocument();
                } else {
                    auto dumper = DocumentDumper();
                    dumpDocument(d->deflate_buffer, header.schema, dumper);
                    dumper.EndDocument();
                }
                d->limit.documentFinished();
            } else {
                try {
                    d->deflate_buffer.startDocument();
                    decodeDocument(d->deflate_buffer, header.schema, filter);
                    if (d->filter->expressionPassed()) {
                        d->deflate_buffer.resetToDocument();
                        if (wd.pos > 0) {
                            auto dumper = TsvDumper(wd);
                            dumpDocument(d->deflate_buffer, header.schema, dumper);
                            dumper.EndDocument();
                        } else {
                            auto dumper = DocumentDumper();
                            dumpDocument(d->deflate_buffer, header.schema, dumper);
                            dumper.EndDocument();
                        }
                        d->limit.documentFinished();
                    }<commit_msg>Check if there's no filter set<commit_after>#include <sys/types.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <cstdint>
#include <cstring>
#include <fstream>
#include <stdexcept>
#include <iostream>
#include <unordered_map>

#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filter/zlib.hpp>

#include <boost/algorithm/string.hpp>

#include <boost/lexical_cast.hpp>

#include <filter/filter.h>
#include <filter/equality_expression.h>

#include "schemareader.h"
#include "limiter.h"

#include "node/node.h"
#include "node/record.h"
#include "node/union.h"
#include "node/string.h"
#include "node/custom.h"
#include "node/int.h"
#include "node/long.h"
#include "node/array.h"
#include "node/boolean.h"
#include "node/enum.h"
#include "node/float.h"
#include "node/double.h"
#include "node/map.h"
#include "node/null.h"

#include "deflatedbuffer.h"
#include "filehandler.h"
#include "predicate/predicate.h"
#include "stringbuffer.h"
#include "zigzag.hpp"
#include "reader.h"

namespace {
const std::string AVRO_MAGICK = "Obj\001"; // 4 bytes
}

namespace avro {

class Reader::Private {
public:
    std::string filename; // TODO: move to FileHandle

    std::unique_ptr<StringBuffer> input;

    DeflatedBuffer deflate_buffer;
    Limiter &limit;

    FileHandle file;
    std::shared_ptr<filter::Filter> filter;
    std::unordered_map<const Node *, std::shared_ptr<predicate::Predicate>> filterItems;

    Private(const std::string& filename, Limiter &limit) : filename(filename), limit(limit), file(filename) {
    }
};

Reader::Reader(const std::string& filename, Limiter &limit) :
    d(new Private(filename, limit)) {

    d->input = d->file.mmapFile();
}

Reader::~Reader() {
    delete d;
}


header Reader::readHeader() {

    
    std::string magick;
    magick.resize(AVRO_MAGICK.size());
    d->input->read(&magick[0], AVRO_MAGICK.size());
    
    if (magick != AVRO_MAGICK) {
        throw std::runtime_error("Bad avro file (wrong magick)");
    }

    header header;


    int64_t recordsNumber = readZigZagLong(*d->input);
    
    for(uint i = 0; i < recordsNumber; ++i) {

        size_t len = readZigZagLong(*d->input);
        const std::string &key = d->input->getStdString(len);
        const std::string &value = d->input->getStdString(readZigZagLong(*d->input));

        header.metadata[key] = value;

    }

    SchemaReader schemaReader(header.metadata["avro.schema"]);
    header.schema = schemaReader.parse();
    header.nodesNumber = schemaReader.nodesNumber();
    //dumpShema(schemaRoot);
    
    char c = d->input->getChar();

    assert(c == 0); // TODO: what is it?
    d->input->read(&header.sync[0], sizeof header.sync);


    return header;

}
constexpr static const char *indents[] = {
        "",
        "\t",
        "\t\t",
        "\t\t\t",
        "\t\t\t\t",
        "\t\t\t\t\t",
        "\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t\t",
        "\t\t\t\t\t\t\t\t\t\t\t\t",
    };

class DocumentDumper {
public:
    void String(const StringBuffer &s, const node::String &n) {
        std::cout << indents[level] << n.getItemName() << ": \"" << s << '"' << std::endl;
    }

    void MapName(const StringBuffer &name) {
        std::cout << indents[level] << name;
    }

    void MapValue(const StringBuffer &s, const node::String &n) {
        std::cout << ": \"" << s << "\"" << std::endl;
    }

    void MapValue(int i, const node::Int &n) {
        std::cout << ": " << i << std::endl;
    }

    void Int(int i, const node::Int &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << i << std::endl;
    }

    void Long(long l, const node::Long &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << l << std::endl;
    }

    void Float(float f, const node::Float &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << f << std::endl;
    }

    void Double(double d, const node::Double &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << d << std::endl;
    }

    void Boolean(bool b, const node::Boolean &n) {
        std::cout << indents[level] << n.getItemName() << ':' << ' ' << b << std::endl;
    }

    void Null(const node::Null &n) {
        std::cout << indents[level] << n.getItemName() << ": null" << std::endl;
    }

    void RecordBegin(const node::Record &r) {
        std::cout << "{\n";
        level++;
    }

    void RecordEnd(const node::Record &r) {
        --level;
        std::cout << indents[level] << "}\n";
    }

    void ArrayBegin(const node::Array &a) {
        std::cout << "[\n";
        level++;

    }

    void ArrayEnd(const node::Array &a) {
        level--;
        std::cout << indents[level] << "]\n";
    }

    void CustomBegin(const node::Custom &c) {
        std::cout << indents[level] << c.getItemName();
    }

    void Enum(const node::Enum &e, int index) {
        const auto &value = e[index];
        std::cout << ": \"" << value  << '"' << std::endl;
    }

    void MapBegin(const node::Map &m) {
        std::cout << "{\n";
        level++;
    }

    void MapEnd(const node::Map &m) {
        level--;
        std::cout << indents[level] << "}\n";
    }

    void EndDocument() {
    }

private:
    int level = 0;

};

class Dumper {
public:
    virtual ~Dumper() {}
    virtual void dump(std::ostream &os) = 0;
};

template<typename T>
class TDumper : public Dumper {
public:

    TDumper(const T &t) : Dumper(), t(t) {
    }
    void dump(std::ostream &os) {
        os << t;
    }
private:
    T t;
};

class TsvDumper {
public:
    TsvDumper(const TsvExpression &wd) : whatDump(wd) {
    	toDump.resize(whatDump.pos);
    	// std::cout << "TsvDumper() " << whatDump.what.size() << std::endl;
    }

    template<typename T, typename NodeType>
    void addIfNecessary(const T &t, const NodeType &n) {
        // std::cout << "sz " << whatDump.what.size() << std::endl;
    	auto p = whatDump.what.find(n.getNumber());
    	if (p != whatDump.what.end()) {
            // std::cout << "adding " << n.getItemName() << " num=" << n.getNumber() << " pos = " << p->second << std::endl;

    		toDump[p->second].reset(new TDumper<T>(t));
    	} else {
            // std::cout << "NOT adding " << n.getItemName() << " num=" << n.getNumber() << " pos = " << p->second << " addIfNecessary() " << whatDump.what.size() << std::endl;
    	}
    }

    void String(const StringBuffer &s, const node::String &n) {
    	addIfNecessary(s, n);
    }

    void MapName(const StringBuffer &name) {
        // std::cout << indents[level] << s;
    }

    void MapValue(const StringBuffer &s, const node::String &n) {
        //std::cout << ": \"" << s  << "\"" << std::endl;
    }
    void MapValue(int i, const node::Int &n) {
        //std::cout << ": \"" << s  << "\"" << std::endl;
    }

    void Int(int i, const node::Int &n) {
    	addIfNecessary(i, n);
    }

    void Long(long l, const node::Long &n) {
    	addIfNecessary(l, n);
    }

    void Float(float f, const node::Float &n) {
    	addIfNecessary(f, n);
    }

    void Double(double d, const node::Double &n) {
    	addIfNecessary(d, n);
    }

    void Boolean(bool b, const node::Boolean &n) {
    	addIfNecessary(b, n);
    }

    void Null(const node::Null &n) {
    	addIfNecessary(std::string("null"), n);
    }

    void Union(int index, const node::Union &n) {

    }

    void RecordBegin(const node::Record &r) {
    }

    void RecordEnd(const node::Record &r) {
    }

    void ArrayBegin(const node::Array &a) {
    }

    void ArrayEnd(const node::Array &a) {
    }

    void CustomBegin(const node::Custom &c) {
    }

    void Enum(const node::Enum &e, int index) {
    	addIfNecessary(e[index], e);
    }

    void MapBegin(const node::Map &m) {
    }

    void MapEnd(const node::Map &m) {
    }

    void EndDocument() {
    	auto p = toDump.begin();
    	(*p)->dump(std::cout);
    	// std::cout << *p;
    	++p;
    	while(p != toDump.end()) {
    		std::cout << "\t";
    		(*p)->dump(std::cout);
    		++p;
    	}
    	std::cout << std::endl;
    }

private:
    std::vector<std::unique_ptr<Dumper>> toDump;
    const TsvExpression &whatDump;
};



void Reader::setFilter(std::shared_ptr<filter::Filter> filter, const header &header) {
    d->filter = filter;

    for(auto &filterPredicate : filter->getPredicates()) {
        const Node * filterNode = schemaNodeByPath(filterPredicate->identifier, header);

        if (filterNode->is<node::Union>()) {
            for( auto &p : filterNode->as<node::Union>().getChildren()) {
                if (p->is<node::String>()) {
                    filterNode = p.get();
                    break;
                } else if (p->is<node::Int>() || p->is<node::Long>()) {
                    filterNode = p.get();
                    break;
                }
            }
        }
        d->filterItems[filterNode] = std::make_shared<predicate::Predicate>(filterPredicate);
    }
}

void Reader::readBlock(const header &header, const FilterExpression *filter, const TsvExpression &wd) {

    int64_t objectCountInBlock = readZigZagLong(*d->input);
    int64_t blockBytesNum = readZigZagLong(*d->input);

    if (header.metadata.at("avro.codec") == "deflate") { // TODO: check it once

        d->deflate_buffer.assignData(d->input->getAndSkip(blockBytesNum), blockBytesNum);

        for(int i = 0; i < objectCountInBlock; ++i) {
            // TODO: rewrite it using hierarcy of filters/decoders.
            // TODO: implement counter as a filter  
            if (!filter && false) {
                // dumpDocument(d->deflate_buffer, header.schema, 0);
                if (wd.pos > 0) {
                    auto dumper = TsvDumper(wd);
                    // std::cout << "readBlock " << wd.what.size() << std::endl;
                    dumpDocument(d->deflate_buffer, header.schema, dumper);
                    dumper.EndDocument();
                } else {
                    auto dumper = DocumentDumper();
                    dumpDocument(d->deflate_buffer, header.schema, dumper);
                    dumper.EndDocument();
                }
                d->limit.documentFinished();
            } else {
                try {
                    d->deflate_buffer.startDocument();
                    decodeDocument(d->deflate_buffer, header.schema, filter);
                    if (!d->filter || d->filter->expressionPassed()) {
                        d->deflate_buffer.resetToDocument();
                        if (wd.pos > 0) {
                            auto dumper = TsvDumper(wd);
                            dumpDocument(d->deflate_buffer, header.schema, dumper);
                            dumper.EndDocument();
                        } else {
                            auto dumper = DocumentDumper();
                            dumpDocument(d->deflate_buffer, header.schema, dumper);
                            dumper.EndDocument();
                        }
                        d->limit.documentFinished();
                    }
{'fraction_non_alphanumeric': 0.1772276783712274, 'fraction_numerical': 0.001793802412664245, 'mean_word_length': 2.2342277012327774, 'pattern_counts': {'":': 14, '<': 483, '<?xml version=': 0, '>': 164, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3583158', 'n_tokens_mistral': 7756, 'n_tokens_neox': 7521, 'n_words': 1508}