<commit_before>#include "page.h"
#include "bloom_filter.h"
#include <fstream>
#include <cassert>
#include <algorithm>
#include <cstring>

using namespace dariadb::storage;


class PageCursor : public dariadb::storage::Cursor {
public:
	PageCursor(Page*page, const dariadb::IdArray&ids, dariadb::Time from, dariadb::Time to, dariadb::Flag flag) :
		link(page),
		_ids(ids),
		_from(from),
		_to(to),
		_flag(flag)
	{
		reset_pos();
	}

	~PageCursor() {
		if (link != nullptr) {
			link->dec_reader();
			link = nullptr;
		}
	}

	bool is_end()const override {
		return _is_end;
	}

	void readNext(Cursor::Callback*cbk)  override {
		//TODO refact this
		std::lock_guard<std::mutex> lg(_locker);
        if(read_poses.empty()){
			_is_end = true;
            return;
        }
        auto _index_it=this->link->index[read_poses.front()];
        read_poses.pop_front();
        for (; !_is_end; ) {
            if (_is_end) {
				Chunk_Ptr empty;
				cbk->call(empty);
				_is_end = true;
				break;
			}
		
            if ((_ids.size() != 0) && (std::find(_ids.begin(), _ids.end(), _index_it.info.first.id) == _ids.end())) {
				if (!read_poses.empty()) {
					_index_it = this->link->index[read_poses.front()];
					read_poses.pop_front();
				}
				else { break; }
				continue;
			}

            if (!dariadb::storage::bloom_check(_index_it.info.flag_bloom, _flag)) {
				if (!read_poses.empty()) {
					_index_it = this->link->index[read_poses.front()];
					read_poses.pop_front();
				}
				else { break; }
				continue;
			}

			if (check_index_rec(_index_it)) {
                auto ptr = new Chunk(_index_it.info, link->chunks + _index_it.offset, link->header->chunk_size);
				Chunk_Ptr c{ ptr };
				assert(c->last.time != 0);
				cbk->call(c);
				break;
			}
			else {//end of data;
				_is_end = true;
				Chunk_Ptr empty;
				cbk->call(empty);
				break;
			}
		}
		if (read_poses.empty()) {
			_is_end = true;
			return;
		}
	}
	
    bool check_index_rec(Page_ChunkIndex&it) const{
        return ((dariadb::utils::inInterval(_from, _to, it.info.minTime)) || (dariadb::utils::inInterval(_from, _to, it.info.maxTime)));
	}

	void reset_pos() override { //start read from begining;
		_is_end = false; 
        this->read_poses.clear();
		//TODO lock this->link; does'n need when call from ctor.
		auto sz = this->link->_itree.size();
		if (sz != 0) {
			auto it_to = this->link->_itree.end();// this->link->_itree.upper_bound(this->_to);
			auto it_from = this->link->_itree.begin();//this->link->_itree.lower_bound(this->_from);

            /*if(it_from!= this->link->_itree.begin()){
                if(it_from->first!=this->_from){
                    --it_from;
                }
            }*/
            for(auto it=it_from;it!=this->link->_itree.end();++it){
                this->read_poses.push_back(it->second);
                if(it==it_to){
                    break;
                }
            }
            if(read_poses.empty()){
                _is_end=true;
                return;
            }
        }else{
            _is_end=true;
            return;
        }
	}
protected:
	Page* link;
	bool _is_end;
	dariadb::IdArray _ids;
	dariadb::Time _from, _to;
	dariadb::Flag _flag;
	std::mutex _locker;
    std::list<uint32_t> read_poses;
};


Page::~Page() {
	region=nullptr;
	header=nullptr;
	index=nullptr;
	chunks=nullptr;
	mmap->close();
}

Page* Page::create(std::string file_name, uint64_t sz, uint32_t chunk_per_storage, uint32_t chunk_size, MODE mode) {
	auto res = new Page;
	auto mmap = utils::fs::MappedFile::touch(file_name, sz);
	auto region = mmap->data();
	std::fill(region, region + sz, 0);

	res->mmap = mmap;
	res->region = region;
	res->header = reinterpret_cast<PageHeader*>(region);
	res->index = reinterpret_cast<Page_ChunkIndex*>(region + sizeof(PageHeader));
	res->chunks = reinterpret_cast<uint8_t*>(region + sizeof(PageHeader) + sizeof(Page_ChunkIndex)*chunk_per_storage);

	res->header->chunk_per_storage = chunk_per_storage;
	res->header->chunk_size = chunk_size;
	res->header->maxTime = dariadb::Time(0);
	res->header->minTime = std::numeric_limits<dariadb::Time>::max();
    res->header->is_overwrite=false;
    res->header->mode=mode;

	for (uint32_t i = 0; i < res->header->chunk_per_storage; ++i) {
		assert(!irec->is_init);
		res->_free_poses.push_back(i);
	}
	return res;
}

size_t get_header_offset(){
    return 0;
}

size_t get_index_offset(){
    return sizeof(PageHeader);
}

size_t get_chunks_offset(uint32_t chunk_per_storage){<commit_msg>build.<commit_after>#include "page.h"
#include "bloom_filter.h"
#include <fstream>
#include <cassert>
#include <algorithm>
#include <cstring>

using namespace dariadb::storage;


class PageCursor : public dariadb::storage::Cursor {
public:
	PageCursor(Page*page, const dariadb::IdArray&ids, dariadb::Time from, dariadb::Time to, dariadb::Flag flag) :
		link(page),
		_ids(ids),
		_from(from),
		_to(to),
		_flag(flag)
	{
		reset_pos();
	}

	~PageCursor() {
		if (link != nullptr) {
			link->dec_reader();
			link = nullptr;
		}
	}

	bool is_end()const override {
		return _is_end;
	}

	void readNext(Cursor::Callback*cbk)  override {
		//TODO refact this
		std::lock_guard<std::mutex> lg(_locker);
        if(read_poses.empty()){
			_is_end = true;
            return;
        }
        auto _index_it=this->link->index[read_poses.front()];
        read_poses.pop_front();
        for (; !_is_end; ) {
            if (_is_end) {
				Chunk_Ptr empty;
				cbk->call(empty);
				_is_end = true;
				break;
			}
		
            if ((_ids.size() != 0) && (std::find(_ids.begin(), _ids.end(), _index_it.info.first.id) == _ids.end())) {
				if (!read_poses.empty()) {
					_index_it = this->link->index[read_poses.front()];
					read_poses.pop_front();
				}
				else { break; }
				continue;
			}

            if (!dariadb::storage::bloom_check(_index_it.info.flag_bloom, _flag)) {
				if (!read_poses.empty()) {
					_index_it = this->link->index[read_poses.front()];
					read_poses.pop_front();
				}
				else { break; }
				continue;
			}

			if (check_index_rec(_index_it)) {
                auto ptr = new Chunk(_index_it.info, link->chunks + _index_it.offset, link->header->chunk_size);
				Chunk_Ptr c{ ptr };
				assert(c->last.time != 0);
				cbk->call(c);
				break;
			}
			else {//end of data;
				_is_end = true;
				Chunk_Ptr empty;
				cbk->call(empty);
				break;
			}
		}
		if (read_poses.empty()) {
			_is_end = true;
			return;
		}
	}
	
    bool check_index_rec(Page_ChunkIndex&it) const{
        return ((dariadb::utils::inInterval(_from, _to, it.info.minTime)) || (dariadb::utils::inInterval(_from, _to, it.info.maxTime)));
	}

	void reset_pos() override { //start read from begining;
		_is_end = false; 
        this->read_poses.clear();
		//TODO lock this->link; does'n need when call from ctor.
		auto sz = this->link->_itree.size();
		if (sz != 0) {
			auto it_to = this->link->_itree.end();// this->link->_itree.upper_bound(this->_to);
			auto it_from = this->link->_itree.begin();//this->link->_itree.lower_bound(this->_from);

            /*if(it_from!= this->link->_itree.begin()){
                if(it_from->first!=this->_from){
                    --it_from;
                }
            }*/
            for(auto it=it_from;it!=this->link->_itree.end();++it){
                this->read_poses.push_back(it->second);
                if(it==it_to){
                    break;
                }
            }
            if(read_poses.empty()){
                _is_end=true;
                return;
            }
        }else{
            _is_end=true;
            return;
        }
	}
protected:
	Page* link;
	bool _is_end;
	dariadb::IdArray _ids;
	dariadb::Time _from, _to;
	dariadb::Flag _flag;
	std::mutex _locker;
    std::list<uint32_t> read_poses;
};


Page::~Page() {
	region=nullptr;
	header=nullptr;
	index=nullptr;
	chunks=nullptr;
	mmap->close();
}

Page* Page::create(std::string file_name, uint64_t sz, uint32_t chunk_per_storage, uint32_t chunk_size, MODE mode) {
	auto res = new Page;
	auto mmap = utils::fs::MappedFile::touch(file_name, sz);
	auto region = mmap->data();
	std::fill(region, region + sz, 0);

	res->mmap = mmap;
	res->region = region;
	res->header = reinterpret_cast<PageHeader*>(region);
	res->index = reinterpret_cast<Page_ChunkIndex*>(region + sizeof(PageHeader));
	res->chunks = reinterpret_cast<uint8_t*>(region + sizeof(PageHeader) + sizeof(Page_ChunkIndex)*chunk_per_storage);

	res->header->chunk_per_storage = chunk_per_storage;
	res->header->chunk_size = chunk_size;
	res->header->maxTime = dariadb::Time(0);
	res->header->minTime = std::numeric_limits<dariadb::Time>::max();
    res->header->is_overwrite=false;
    res->header->mode=mode;

	for (uint32_t i = 0; i < res->header->chunk_per_storage; ++i) {
		res->_free_poses.push_back(i);
	}
	return res;
}

size_t get_header_offset(){
    return 0;
}

size_t get_index_offset(){
    return sizeof(PageHeader);
}

size_t get_chunks_offset(uint32_t chunk_per_storage){
{'fraction_non_alphanumeric': 0.16205709208041763, 'fraction_numerical': 0.004442963456625569, 'mean_word_length': 2.9422066549912436, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 140, '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': '5496025', 'n_tokens_mistral': 3607, 'n_tokens_neox': 3291, 'n_words': 556}