tablesinktostore.h

Go to the documentation of this file.

/**************************************************************
 **
 ** @file data/store/tablesinktostore.h
 **
 ** Copyright (C) 2012  Xpace, LLC.  All rights reserved
 **
 ** www.xpace.net
 **
 **************************************************************/

#include "data/store/table.h"
#include "data/store/store.h"
#include "data/store/classtree.h"

namespace Xpace
{
  // ================================ CONVERT TABLE -> STORE ==

  class XPACE_EXPORT TableSinkToStore : public Table::Sink
  {
  public:
    static const char* DOC_ID_TAG;     ///< tag of the field (column) that contains the document identifier
    static const char* LOC_TAG;        ///< tag of fields (columns) that contain location in the final store

    TableSinkToStore
      (const Configuration& config);

    virtual uint add
      (const String& name,
       DerivedDataType type,
       uint columnNum,
       bool* added = 0) override
    {
      assert(0);
      return ~0;
    }

    virtual uint add
      (const Configuration& config,
       uint col,
       bool* added) override;

    virtual int64 start
      (uint64) override
    {
      if (!targ)
      {
        store = std::unique_ptr<Store>(new Store(class_tree.toString(), SA_bulkWrite));
        targ = store->getSink();
      }
      return 0;
    }

    virtual bool set
      (uint column,
       int64 value) override
    {
      return id_set(column, value);
    }

    virtual bool set
      (uint column,
       const DecimalFloat& value) override
    {
      return no_id_set(column, value);
    }

    virtual bool set
      (uint column,
       const String8& value) override
    {
      return id_set(column, value);
    }

    virtual bool set
      (uint column,
       const String16& value) override
    {
      // TODO: 16-bit IDs
      return no_id_set(column, value);
    }

    virtual bool set
      (uint column,
       const BytesRef& value) override
    {
      return no_id_set(column, value);
    }

    virtual void close
      () override
    {
      if (targ)
        targ->close();
    }

    class NoDocID : public Exception
    {
    public:
      NoDocID
        ()
      {
      }
    };

    class BadCode : public Exception
    {
    public:
      BadCode
        (const String& code) :
          Exception("Bad code \"%1\".")
      {
        addParam(code);
      }
    };      

  private:
    std::unique_ptr<Store> store;
    Store::Sink* targ;
    uint64 targ_doc;

    struct ct_node
    {
      bool isLeaf;
    };
    ClassTree<ct_node> class_tree;

    struct code_node
    {
      uint child;
      std::map<String8, code_node> codes;
    };
    code_node root_cn, *cur_cn;
    BytePool code_pool;

    uint columns;

    String doc_id_name;
    std::vector<String> loc_names;

    uint doc_id_column;
    std::vector<uint> loc_columns;

    std::vector<std::vector<uint>> data_columns;

    class equal_doc_id
    {
    public:
      equal_doc_id
        ();
      bool operator()
        (int64 n);
      bool operator()
        (const DecimalFloat& n);
      bool operator()
        (const String8& str);
      bool operator()
        (const String16& str);

    private:
      bool first;

      int64 int_val;
      DecimalFloat float_val;
      String8 string8_val;
      String16 string16_val;
      std::vector<byte> bytesref_val;

      std::vector<utf8_t> string8_buf;
      std::vector<utf16_t> string16_buf;
    };
    equal_doc_id eq_doc_id;

    std::vector<bool> is_id;
    void id_column
      (uint col);

    template <typename T>
    bool set_targ
      (uint column,
       T t)   
    {
      if (!cur_data_column)
      {
        cur_data_column = &data_columns[first_data_column];
        targ->move(0);
      }
      uint cur_sibling(cur_data_column->front());
      cur_data_column = &data_columns[column];
      assert(cur_data_column->front() >= cur_sibling);
      if ((cur_data_column->front() != cur_sibling) &&
          ((!targ->move(-1) || !targ->move(cur_data_column->front()))))
        return false;

      return targ->set((*cur_data_column)[1], t);
    }

    template <typename T>
    bool no_id_set
      (uint column,
       T t)
    {
      assert(!is_id[column]);
      return set_targ(column, t);
    }

    template <typename T>
    bool check_doc_id
      (T t)
    {
      cur_data_column = 0;
      cur_cn = &root_cn;
      return (eq_doc_id(t)) ? targ->move(Store::Location()) : !!targ->start(targ_doc);
    }

    bool id_set
      (uint column,
       int64 t)
    {
      if (column == doc_id_column) 
        return check_doc_id(t);
      return set_targ(column, t);
    }

    bool id_set
      (uint column,
       const String8& t)
    {
      if ((column < is_id.size()) && is_id[column]) 
      {
        if (column == doc_id_column) 
          return check_doc_id(t);
        // move based on code
        if ((column == doc_id_column + 1) && (cur_cn != &root_cn))
          throw NoDocID();
        auto new_cn(cur_cn->codes.find(t));
        if (new_cn == cur_cn->codes.end())
          throw BadCode(String(t.data, t.length));
        cur_cn = &(new_cn->second);
        return targ->move(cur_cn->child);
      }
      return set_targ(column, t);
    }    

    // status
    String vector_config;
    size_t first_data_column;
    const std::vector<uint>* cur_data_column;
  };
}