/home/palakars/workspace-inprogress/code-in-progress/aitools3/aitools3-aq-invertedindex-cpp/src/core/IndexBuilder.hpp

Go to the documentation of this file.

/* Copyright (C) 2010 webis.de
 * All rights reserved.
 */
#ifndef AITOOLS_INVERTEDINDEX_INDEX_BUILDER_HPP
#define AITOOLS_INVERTEDINDEX_INDEX_BUILDER_HPP

#include "System.hpp"
#include "Record.hpp"
#include "Builder.hpp"
#include "Logging.hpp"
#include "Checksum.hpp"
#include "Converter.hpp"
#include "RecordReader.hpp"
#include "StorageBuilder.hpp"
#include "PostlistSorter.hpp"
#include "PostlistBuilder.hpp"
#include "InvertedFileReader.hpp"
#include <boost/algorithm/string.hpp>
#include <iomanip>
#include <cstdio>

namespace aitools {
namespace invertedindex {

template<typename Value>
class IndexBuilder : public Builder {

private: // nested types

        typedef std::vector<FILE*>                   FileVector;
        typedef Configuration::Sorting               Sorting;
        typedef std::tr1::unordered_map<std::string,
                PostlistBuilder::SharedPointer>      BuilderMap;

private: // class data

        static const std::string tmp_dir;
        static const std::string tmp_file;

public: // class data

        static const std::string header_file;

public: // c'tor / d'tor

        IndexBuilder(RecordReader<Value>* reader = new InvertedFileReader<Value>);

        ~IndexBuilder();

public: // member

        void build(const Configuration& config) throw (std::exception);

private: // class member

        static void build_postlists(const bfs::path& path, BuilderMap& builders)
        throw (std::exception);

        static const Configuration& check_config(const Configuration& config)
        throw (std::exception);

        static void store_header(const Header& header, const bfs::path& path)
        throw (std::exception);

        static void print_header(const Header& header);

private: // member

        void index_random_records_() throw (std::exception);

        void index_sorted_records_() throw (std::exception);

        bfs::path partition_records_() throw (std::exception);

private: // data

        Header               header_;
        Configuration        config_;
        RecordReader<Value>* reader_;

};

template<typename Value>
const std::string
IndexBuilder<Value>::tmp_dir("tmp");

template<typename Value>
const std::string
IndexBuilder<Value>::tmp_file("tmp.");

template<typename Value>
const std::string
IndexBuilder<Value>::header_file("header");

template<typename Value>
IndexBuilder<Value>::IndexBuilder(RecordReader<Value>* reader)
:       reader_(reader)
{
        assert(reader);
}

template<typename Value>
IndexBuilder<Value>::~IndexBuilder()
{
        delete reader_;
}

// class member

template<typename Value>
void
IndexBuilder<Value>::build_postlists
(const bfs::path& path, BuilderMap& builders) throw (std::exception)
{
        GenericRecord record;
        BuilderMap::const_iterator it;
        BuilderMap::const_iterator end(builders.end());
        FILE* bucket(System::fopen(path, "rb"));
        while (record.read(bucket))
        {
                if ((it = builders.find(record.key())) == end) // key is unknown
                {
                        PostlistBuilder::SharedPointer builder;
                        builder.reset(new PostlistBuilder);
                        builder->set_checksum(Checksum::hash16(record.key()));
                        it = builders.insert(std::make_pair(record.key(), builder)).first;
                }
                it->second->append(record.value());
        }
        System::fclose(bucket);
}

template<typename Value>
const Configuration&
IndexBuilder<Value>::check_config(const Configuration& config)
throw (std::exception)
{
        if (!bfs::exists(config.input_directory()))
        {
                std::string msg("Does not exist");
                Exception::throw_invalid_argument(msg, config.input_directory());
        }
        if (!bfs::exists(config.index_directory()))
        {
                std::string msg("Does not exist");
                Exception::throw_invalid_argument(msg, config.index_directory());
        }
        if (bfs::is_empty(config.input_directory()))
        {
                std::string msg("Is empty");
                Exception::throw_invalid_argument(msg, config.input_directory());
        }
        if (!bfs::is_empty(config.index_directory()))
        {
                std::string msg("Is not empty");
                Exception::throw_invalid_argument(msg, config.index_directory());
        }
        return config;
}

template<typename Value>
void
IndexBuilder<Value>::store_header(const Header& header, const bfs::path& path)
throw (std::exception)
{
        bfs::ofstream ofs(path, std::ios::binary);
        if (!ofs)
        {
                Exception::throw_invalid_argument("Cannot create", path);
        }
        else if (!header.SerializeToOstream(&ofs))
        {
                Exception::throw_runtime_error("Cannot serialize header");
        }
        ofs.close();
}

template<typename Value>
void
IndexBuilder<Value>::print_header(const Header& header)
{
        double payload((double)header.payload() / 1024 / 1024);
        double runtime((double)(header.indexing_duration()) / 60);
        std::clog << "\nNumber of keys    = " << header.key_count()
                          << "\nNumber of values  = " << header.value_count()
                          << std::fixed << std::setprecision(2)
                          << "\nData payload size = " << payload << " Megabyte"
                          << "\nIndexing duration = " << runtime << " Minutes\n"
                          << std::endl;
}

// member

template<typename Value>
void
IndexBuilder<Value>::build(const Configuration& config) throw (std::exception)
{
        config_ = check_config(config);
        header_.set_value_class_name(Value::classname);
        header_.set_major_version(INDEX_VERSION_MAJOR);
        header_.set_minor_version(INDEX_VERSION_MINOR);
        header_.set_indexing_duration(Logging::time());
        header_.set_postlist_sorting(config.postlist_sorting());
        if (config.input_format() == Configuration::REAL_INVERTED)
        {
                index_sorted_records_();
        }
        else
        {
                index_random_records_();
        }
        Logging::log("Indexing succeeded.");
        header_.set_indexing_duration(Logging::time()-header_.indexing_duration());
        store_header(header_, bfs::path(config.index_directory()) / header_file);
        print_header(header_);
}

template<typename Value>
void
IndexBuilder<Value>::index_random_records_() throw (std::exception)
{
        /* Partition Step: Distribute all records to a number of bucket files.
         */
        const bfs::path bucket_dir(partition_records_());

        /* Shuffle Step: Iterate bucket files, build postlists from records,
         * sort theses postlists and store them on external hard drive.
         */
        int64_t key_cnt(0);
        int64_t val_cnt(0);
        int64_t payload(0);
        Quantile quantile;
        BuilderMap builders;
        StorageBuilder storage;
        Postlist<Value> postlist;
        bfs::directory_iterator dend;
        storage.open(config_.index_directory());
        BuilderMap::const_iterator bend(builders.end());
        Sorting sort_mode(config_.postlist_sorting());
        for (bfs::directory_iterator dit(bucket_dir); dit != dend; ++dit)
        {
                Logging::log("Gather Step " + dit->path().string());

                build_postlists(dit->path(), builders);
                for (BuilderMap::iterator bit(builders.begin()); bit != bend; ++bit)
                {
                        postlist = bit->second->build();
                        PostlistSorter::sort(postlist, quantile, sort_mode);
                        storage.put(bit->first, postlist.iterator());
                        if (sort_mode != Configuration::DISABLED)
                        {
                                storage.put(bit->first, quantile);
                        }
                        key_cnt++;
                        val_cnt += postlist.length();
                        payload += postlist.payload();
                }
                bfs::remove(dit->path());
                builders.clear();
        }
        header_.set_value_count(val_cnt);
        header_.set_key_count(key_cnt);
        header_.set_payload(payload);
        bfs::remove(bucket_dir);
        storage.close();
}

template<typename Value>
void
IndexBuilder<Value>::index_sorted_records_() throw (std::exception)
{
        int64_t key_cnt(0);
        int64_t val_cnt(0);
        int64_t payload(0);
        std::string key;
        Quantile quantile;
        Record<Value> record;
        GenericRecord grecord;
        StorageBuilder storage;
        PostlistBuilder builder;
        Postlist<Value> postlist;
        bfs::directory_iterator end;
        storage.open(config_.index_directory());
        Sorting sort_mode(config_.postlist_sorting());
        for (bfs::directory_iterator it(config_.input_directory()); it != end; ++it)
        {
                Logging::log("Gather Step " + it->path().string());

                reader_->open(it->path());
                while (reader_->next(record))
                {
                        record.to_generic(grecord);
                        if (grecord.key().size() > GenericRecord::max_key_size)
                        {
                                Exception::throw_out_of_range("Too long key", record);
                        }
                        if (grecord.value().size() > GenericRecord::max_val_size)
                        {
                                Exception::throw_out_of_range("Too long value", record);
                        }
                        if (key.empty())
                        {
                                key = grecord.key();
                        }
                        else if (grecord.key() != key)
                        {
                                postlist = builder.build();
                                PostlistSorter::sort(postlist, quantile, sort_mode);
                                storage.put(key, postlist.iterator());
                                if (sort_mode != Configuration::DISABLED)
                                {
                                        storage.put(key, quantile);
                                }
                                key_cnt++;
                                val_cnt += postlist.length();
                                payload += postlist.payload();
                                builder.set_checksum(Checksum::hash16(grecord.key()));
                                key = grecord.key();
                        }
                        builder.append(grecord.value());
                }
                reader_->close();
        }
        postlist = builder.build();
        PostlistSorter::sort(postlist, quantile, sort_mode);
        storage.put(key, postlist.iterator()); // put last
        if (sort_mode != Configuration::DISABLED)
        {
                storage.put(key, quantile);
        }
        key_cnt++;
        val_cnt += postlist.length();
        payload += postlist.payload();
        header_.set_value_count(val_cnt);
        header_.set_key_count(key_cnt);
        header_.set_payload(payload);
        storage.close();
}

template<typename Value>
bfs::path
IndexBuilder<Value>::partition_records_() throw (std::exception)
{
        // create tmp files as buckets
        const bfs::path bucket_dir(bfs::path(config_.index_directory()) / tmp_dir);
        const uint64_t payload(System::directory_size(config_.input_directory()));
        if (config_.available_memory() < 1000) config_.set_available_memory(1000);
        unsigned bucket_count(payload / (700000 * config_.available_memory()));
        bucket_count = Checksum::next_prime(bucket_count);
        std::vector<FILE*> bucket_files(bucket_count);
        bfs::create_directory(bucket_dir);
        std::string bucket_file_name;
        for (unsigned i(0); i != bucket_count; ++i)
        {
                bucket_file_name = tmp_file + Converter::ui32_to_str(i);
                bucket_files[i] = System::fopen(bucket_dir / bucket_file_name, "wb");
        }

        // read records and partition them
        uint32_t checksum;
        Record<Value> record;
        GenericRecord gen_record;
        const bfs::directory_iterator end;
        for (bfs::directory_iterator it(config_.input_directory()); it != end; ++it)
        {
                Logging::log("Scatter Step " + it->path().string());

                reader_->open(it->path());
                while (reader_->next(record))
                {
                        record.to_generic(gen_record);
                        if (gen_record.key().size() > GenericRecord::max_key_size)
                        {
                                Exception::throw_out_of_range("Too long key", record);
                        }
                        if (gen_record.value().size() > GenericRecord::max_val_size)
                        {
                                Exception::throw_out_of_range("Too long value", record);
                        }
                        checksum = Checksum::hash32(gen_record.key());
                        gen_record.write(bucket_files[checksum % bucket_count]);
                }
                reader_->close();
        }

        // close bucket files
        for (unsigned i(0); i != bucket_count; ++i)
        {
                System::fclose(bucket_files[i]);
        }
        return bucket_dir;
}

} // namespace invertedindex
} // namespace aitools

#endif // AITOOLS_INVERTEDINDEX_INDEX_BUILDER_HPP

---