<commit_before>#include <algorithm>
#include <fstream>
#include <iostream>
#include <map>
#include <set>
#include <sstream>
#include <string>
#include <vector>

#include "../script_datatypes.h"
#include "file_types.h"
#include "relation_strings_file_io.h"
#include "raw_file_db.h"

#include <string.h>
#include <stdio.h>
#include <time.h>

#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <unistd.h>

using namespace std;

typedef char int8;
typedef short int int16;
typedef int int32;
typedef long long int64;

typedef unsigned char uint8;
typedef unsigned short int uint16;
typedef unsigned int uint32;
typedef unsigned long long uint64;

//-----------------------------------------------------------------------------

const string RELATION_STRING_DATA("relation_strings.dat");
const string RELATION_STRING_IDX("relation_strings.idx");

const char* RELATION_TAG_TMPAPREFIX = "/tmp/relation_strings_";
const char* RELATION_TAG_TMPB = "/tmp/relation_tag_ids";

//-----------------------------------------------------------------------------

uint32 RelationCollection::next_relation_tag_id(0);

void flush_relation_tags(uint& current_run, map< RelationKeyValue, RelationCollection >& relation_tags, vector< Relation_ >& relations)
{
  //determine whether a certain tag exists only local
  sort(relations.begin(), relations.end(), Relation_Less_By_Id());
  for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
       it != relation_tags.end(); ++it)
  {
    uint32 position(*(it->second.relations.begin()));
    uint32 bitmask(0);
    for (vector< uint32 >::iterator it2(it->second.relations.begin()); it2 != it->second.relations.end(); ++it2)
    {
      bitmask |= (position ^ (*it2));
    }
    it->second.position = position;
    it->second.bitmask = bitmask;
  }
  
  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<current_run;
  int dest_fd = open64(temp.str().c_str(), O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64(temp.str().c_str(), O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "flush_relation_tags:1");
  
  if (current_run == 0)
  {
    int dest_id_fd = open64(RELATION_TAG_TMPB, O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
    if (dest_id_fd < 0)
      throw File_Error(errno, RELATION_TAG_TMPB, "flush_relation_tags:2");
    
    for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
         it != relation_tags.end(); ++it)
    {
      uint16 key_size(it->first.key.size());
      uint16 value_size(it->first.value.size());
      if (it->second.bitmask>>8)
        it->second.position = 0xffffffff;
      else
        it->second.position &= 0xffffff00;
      write(dest_fd, &(it->second.id), sizeof(uint32));
      write(dest_fd, &(it->second.position), sizeof(uint32));
      write(dest_fd, &(key_size), sizeof(uint16));
      write(dest_fd, &(value_size), sizeof(uint16));
      write(dest_fd, &(it->first.key[0]), key_size);
      write(dest_fd, &(it->first.value[0]), value_size);
      
      uint32 nc_size(it->second.relations.size());
      write(dest_id_fd, &(it->second.id), sizeof(uint32));
      write(dest_id_fd, &(nc_size), sizeof(uint32));
      for (vector< uint32 >::const_iterator it2(it->second.relations.begin());
	   it2 != it->second.relations.end(); ++it2)
	write(dest_id_fd, &(*it2), sizeof(uint32));
    }
    
    close(dest_id_fd);
  }
  else
  {
    uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
    uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
    char* key_rd_buf = (char*) malloc(64*1024);
    char* value_rd_buf = (char*) malloc(64*1024);
    
    temp.str("");
    temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
    int source_fd = open64(temp.str().c_str(), O_RDONLY);
    if (source_fd < 0)
      throw File_Error(errno, temp.str().c_str(), "flush_relation_tags:3");
    
    map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
    while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
    {
      read(source_fd, key_rd_buf, size_rd_buf[0]);
      key_rd_buf[size_rd_buf[0]] = 0;
      read(source_fd, value_rd_buf, size_rd_buf[1]);
      value_rd_buf[size_rd_buf[1]] = 0;
      while ((it != relation_tags.end()) &&
             (strcmp(key_rd_buf, it->first.key.c_str()) > 0))
      {
        uint16 key_size(it->first.key.size());
        uint16 value_size(it->first.value.size());
        if (it->second.bitmask>>8)
          it->second.position = 0xffffffff;
        else
          it->second.position &= 0xffffff00;
        write(dest_fd, &(it->second.id), sizeof(uint32));
        write(dest_fd, &(it->second.position), sizeof(uint32));
        write(dest_fd, &(key_size), sizeof(uint16));
        write(dest_fd, &(value_size), sizeof(uint16));
        write(dest_fd, &(it->first.key[0]), key_size);
        write(dest_fd, &(it->first.value[0]), value_size);
        ++it;
      }
      while ((it != relation_tags.end()) &&
             (strcmp(key_rd_buf, it->first.key.c_str()) == 0) &&
             (strcmp(value_rd_buf, it->first.value.c_str()) > 0))
      {
        uint16 key_size(it->first.key.size());
        uint16 value_size(it->first.value.size());
        if (it->second.bitmask>>8)
          it->second.position = 0xffffffff;
        else
          it->second.position &= 0xffffff00;
        write(dest_fd, &(it->second.id), sizeof(uint32));
        write(dest_fd, &(it->second.position), sizeof(uint32));
        write(dest_fd, &(key_size), sizeof(uint16));
        write(dest_fd, &(value_size), sizeof(uint16));
        write(dest_fd, &(it->first.key[0]), key_size);
        write(dest_fd, &(it->first.value[0]), value_size);
        ++it;
      }
      if ((it != relation_tags.end()) &&
          (strcmp(key_rd_buf, it->first.key.c_str()) == 0) &&
          (strcmp(value_rd_buf, it->first.value.c_str()) == 0))
      {
        it->second.merge(cnt_rd_buf[0], cnt_rd_buf[1]);
        if ((cnt_rd_buf[1] == 0xffffffff) || (it->second.bitmask>>8))
          cnt_rd_buf[1] = 0xffffffff;
        else
          cnt_rd_buf[1] &= 0xffffff00;
        ++it;
      }
      write(dest_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16));
      write(dest_fd, key_rd_buf, size_rd_buf[0]);
      write(dest_fd, value_rd_buf, size_rd_buf[1]);
    }
    while (it != relation_tags.end())
    {
      uint16 key_size(it->first.key.size());
      uint16 value_size(it->first.value.size());
      if (it->second.bitmask>>8)
	it->second.position = 0xffffffff;
      else
	it->second.position &= 0xffffff00;
      write(dest_fd, &(it->second.id), sizeof(uint32));
      write(dest_fd, &(it->second.position), sizeof(uint32));
      write(dest_fd, &(key_size), sizeof(uint16));
      write(dest_fd, &(value_size), sizeof(uint16));
      write(dest_fd, &(it->first.key[0]), key_size);
      write(dest_fd, &(it->first.value[0]), value_size);
      ++it;
    }
    
    free(cnt_rd_buf);
    free(key_rd_buf);
    free(value_rd_buf);
    
    close(source_fd);
    temp.str("");
    temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
    remove(temp.str().c_str());
    
    int dest_id_fd = open64(RELATION_TAG_TMPB, O_WRONLY|O_APPEND, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
    if (dest_id_fd < 0)
      throw File_Error(errno, RELATION_TAG_TMPB, "flush_relation_tags:4");
    
    for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
	 it != relation_tags.end(); ++it)
    {
      uint32 nc_size(it->second.relations.size());
      write(dest_id_fd, &(it->second.id), sizeof(uint32));
      write(dest_id_fd, &(nc_size), sizeof(uint32));
      for (vector< uint32 >::const_iterator it2(it->second.relations.begin());
	   it2 != it->second.relations.end(); ++it2)
	write(dest_id_fd, &(*it2), sizeof(uint32));	
    }
    
    close(dest_id_fd);
  }
  
  close(dest_fd);
  ++current_run;
}

void relation_tag_statistics(uint& current_run, vector< uint32 >& split_idx)
{
  cerr<<'s';
  
  if (current_run == 0)
  {
    cerr<<"No relation tags.\n";
    return;
  }
  
  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
  int source_fd = open64(temp.str().c_str(), O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "relation_tag_statistics:1");
  
  uint32 global_count(0);
  uint32* spatial_count = (uint32*) calloc(16*1024*1024, sizeof(uint32));
  uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
  uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
  
  while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
  {
    lseek64(source_fd, size_rd_buf[0] + size_rd_buf[1], SEEK_CUR);
    ++(spatial_count[cnt_rd_buf[1]>>8]);
    ++global_count;
  }
  global_count -= spatial_count[0x00ffffff];
  
  split_idx.clear();
  uint32 split_count(0);
  for (unsigned int i(0); i < 16*1024*1024; ++i)
  {
    split_count += spatial_count[i];
    if (split_count >= global_count / RELATION_TAG_SPATIAL_PARTS)
    {
      split_idx.push_back(i<<8);
      split_count -= global_count / RELATION_TAG_SPATIAL_PARTS;
    }
  }
  while (split_idx.size() < RELATION_TAG_SPATIAL_PARTS)
    split_idx.push_back(0xffffff<<8);
  split_idx[RELATION_TAG_SPATIAL_PARTS-1] = 0xffffff<<8;
  
  free(cnt_rd_buf);
  free(spatial_count);
  
  close(source_fd);

  cerr<<'s';
}

void relation_tag_split_and_index(uint& current_run, vector< uint32 >& split_idx, uint32*& block_of_id)
{
  cerr<<'p';

  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
  int source_fd = open64(temp.str().c_str(), O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "relation_tag_split_and_index:1");
  
  int dest_fd = open64((DATADIR + db_subdir + RELATION_STRING_DATA).c_str(),
                       O_WRONLY|O_CREAT|O_TRUNC,
		       S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64((DATADIR + db_subdir + RELATION_STRING_DATA).c_str(),
                   O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_DATA,
                     "relation_tag_split_and_index:2");
  
  int dest_idx_fd =
    open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(),
           O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_idx_fd);
  
  dest_idx_fd = open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(),
                       O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_IDX,
                     "relation_tag_split_and_index:3");
  
  write(dest_idx_fd, &(RelationCollection::next_relation_tag_id), sizeof(uint32));
  for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS; ++i)
    write(dest_idx_fd, &(split_idx[i]), sizeof(uint32));
  
  uint8* cur_block_count = (uint8*) calloc(RELATION_TAG_SPATIAL_PARTS+1, sizeof(uint8));
  block_of_id = (uint32*) calloc((RelationCollection::next_relation_tag_id+1), sizeof(uint32));
  
  uint32* write_pos = (uint32*) malloc((RELATION_TAG_SPATIAL_PARTS+1)*sizeof(uint32));
  for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
    write_pos[i] = sizeof(uint32);
  char* write_blocks = (char*) malloc((RELATION_TAG_SPATIAL_PARTS+1) * RELATION_STRING_BLOCK_SIZE);
  uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
  uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
  
  while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
  {
    uint16 block(0);
    if (cnt_rd_buf[1] != 0xffffffff)
    {
      while (split_idx[block] <= cnt_rd_buf[1])
	++block;
      ++block;
    }
    
    if (write_pos[block] + 2*sizeof(uint32) + 2*sizeof(uint16) + size_rd_buf[0] + size_rd_buf[1]
	>= RELATION_STRING_BLOCK_SIZE*4/5)
    {
      *((uint32*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE])) = write_pos[block] - sizeof(uint32);
      write(dest_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE]), RELATION_STRING_BLOCK_SIZE);
      write(dest_idx_fd, &(block), sizeof(uint16));
      write(dest_idx_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE
	  + 3*sizeof(uint32)]), *(uint16*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE
	      + 3*sizeof(uint32)]) + *(uint16*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE
		  + 3*sizeof(uint32) + sizeof(uint16)]) + 2*sizeof(uint16));
    
      write_pos[block] = sizeof(uint32);

      memcpy(&(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]), cnt_rd_buf,
	       2*sizeof(uint32) + 2*sizeof(uint16));
      write_pos[block] += 2*sizeof(uint32) + 2*sizeof(uint16);
      read(source_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]),
	   size_rd_buf[0] + size_rd_buf[1]);
      write_pos[block] += size_rd_buf[0] + size_rd_buf[1];
      
      block_of_id[cnt_rd_buf[0]] = (++cur_block_count[block]) | ((cnt_rd_buf[1]) & (0xffffff00));
    }
    else
    {
      memcpy(&(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]), cnt_rd_buf,
	       2*sizeof(uint32) + 2*sizeof(uint16));
      write_pos[block] += 2*sizeof(uint32) + 2*sizeof(uint16);
      read(source_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]),
	   size_rd_buf[0] + size_rd_buf[1]);
      write_pos[block] += size_rd_buf[0] + size_rd_buf[1];
    
      block_of_id[cnt_rd_buf[0]] = (cur_block_count[block]) | ((cnt_rd_buf[1]) & (0xffffff00));
    }
  }
  
  for (unsigned int i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
  {
    *((uint32*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE])) = write_pos[i] - sizeof(uint32);
    write(dest_fd, &(write_blocks[i * RELATION_STRING_BLOCK_SIZE]), RELATION_STRING_BLOCK_SIZE);
    write(dest_idx_fd, &(i), sizeof(uint16));
    write(dest_idx_fd, &(write_blocks[i * RELATION_STRING_BLOCK_SIZE
	+ 3*sizeof(uint32)]), *(uint16*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE
	    + 3*sizeof(uint32)]) + *(uint16*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE
		+ 3*sizeof(uint32) + sizeof(uint16)]) + 2*sizeof(uint16));
  }
  
  free(cur_block_count);
  free(write_pos);
  free(write_blocks);
  free(cnt_rd_buf);
  
  close(source_fd);
  close(dest_fd);

  cerr<<'p';
}

struct tag_id_local_less : public binary_function< uint32*, uint32*, bool >
{
  tag_id_local_less(uint32* block_of_id) : block_of_id_(block_of_id) {}

  bool operator() (uint32* const& a, uint32* const& b)
  {
    if ((block_of_id_[*a] & 0xffffff00) < (block_of_id_[*b] & 0xffffff00))
      return true;
    else if ((block_of_id_[*a] & 0xffffff00) > (block_of_id_[*b] & 0xffffff00))
      return false;
    return (*a < *b);
  }
private:
  uint32* block_of_id_;
};

struct tag_id_global_less : public binary_function< uint32*, uint32*, bool >
{
  bool operator() (uint32* const& a, uint32* const& b)
  {
    return (*a < *b);
  }
};

void relation_tag_create_id_relation_idx(uint32* block_of_id)
{
  uint32 rd_buf_pos(0);
  Tag_Id_Relation_Local_Writer env_local(block_of_id);
  Tag_Id_Relation_Global_Writer env_global;
  
  int source_fd = open64(RELATION_TAG_TMPB, O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, RELATION_TAG_TMPB, "relation_tag_create_id_relation_idx:1");
  
  uint32* tag_rd_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
  uint32* tag_alt_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
  uint32 max_pos(0);
  
  while ((max_pos =
	 read(source_fd, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32))))
  {
    vector< uint32* > tag_id_local, tag_id_global;
    uint32 alt_buf_pos(0);
    max_pos += rd_buf_pos*sizeof(uint32);
    rd_buf_pos = 0;
  
    while ((rd_buf_pos + 1 < max_pos / sizeof(uint32)) &&
	    (rd_buf_pos + tag_rd_buf[rd_buf_pos+1] + 1 < max_pos / sizeof(uint32)))
    {
      uint32 next_pos(tag_rd_buf[rd_buf_pos+1] + 2);
      
      if ((~(block_of_id[tag_rd_buf[rd_buf_pos]])) & 0xffffff00)
      {
	tag_id_local.push_back(&(tag_rd_buf[rd_buf_pos]));
      
	while (tag_rd_buf[rd_buf_pos+1] > 255)
	{
	  if (tag_rd_buf[rd_buf_pos+1] > 510)
	  {
	    tag_id_local.push_back(&(tag_alt_buf[alt_buf_pos]));
	    tag_id_local.push_back(&(tag_rd_buf[rd_buf_pos + 510]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]), 255*sizeof(uint32));
	    alt_buf_pos += 255;
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  
	    next_pos -= 510;
	    rd_buf_pos += 510;
	  
	    tag_rd_buf[rd_buf_pos] = tag_rd_buf[rd_buf_pos - 510];
	    tag_rd_buf[rd_buf_pos+1] = next_pos - 2;
	  }
	  else
	  {
	    tag_id_local.push_back(&(tag_alt_buf[alt_buf_pos]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos+1] - 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]),
		     (tag_rd_buf[rd_buf_pos+1] - 255)*sizeof(uint32));
	    alt_buf_pos += (tag_rd_buf[rd_buf_pos+1] - 255);
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  }
	}
      }
      else
      {
	tag_id_global.push_back(&(tag_rd_buf[rd_buf_pos]));
      
	while (tag_rd_buf[rd_buf_pos+1] > 255)
	{
	  if (tag_rd_buf[rd_buf_pos+1] > 510)
	  {
	    tag_id_global.push_back(&(tag_alt_buf[alt_buf_pos]));
	    tag_id_global.push_back(&(tag_rd_buf[rd_buf_pos + 510]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]), 255*sizeof(uint32));
	    alt_buf_pos += 255;
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  
	    next_pos -= 510;
	    rd_buf_pos += 510;
	  
	    tag_rd_buf[rd_buf_pos] = tag_rd_buf[rd_buf_pos - 510];
	    tag_rd_buf[rd_buf_pos+1] = next_pos - 2;
	  }
	  else
	  {
	    tag_id_global.push_back(&(tag_alt_buf[alt_buf_pos]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos+1] - 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]),
		     (tag_rd_buf[rd_buf_pos+1] - 255)*sizeof(uint32));
	    alt_buf_pos += (tag_rd_buf[rd_buf_pos+1] - 255);
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  }
	}
      }
      
      rd_buf_pos += next_pos;
    }
    
    sort(tag_id_local.begin(), tag_id_local.end(), tag_id_local_less(block_of_id));
    sort(tag_id_global.begin(), tag_id_global.end(), tag_id_global_less());
    
    flush_data< Tag_Id_Relation_Local_Writer >
	(env_local, tag_id_local.begin(), tag_id_local.end());
    flush_data< Tag_Id_Relation_Global_Writer >
        (env_global, tag_id_global.begin(), tag_id_global.end());
    
    cerr<<'.';
    
    memmove(tag_rd_buf, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32));
    rd_buf_pos = RELATION_TAG_SORT_BUFFER_SIZE / sizeof(uint32) - rd_buf_pos;
  }
  
  make_block_index< Tag_Id_Relation_Local_Writer >(env_local);
  make_block_index< Tag_Id_Relation_Global_Writer >(env_global);
  
  free(tag_rd_buf);
  free(tag_alt_buf);
  
  close(source_fd);
}

struct relation_idx_less : public binary_function< uint32, uint32, bool >
{
  relation_idx_less(uint32* ll_idx__) : ll_idx_(ll_idx__) {}
  
  bool operator() (const uint32& a, const uint32& b)
  {
    return (ll_idx_[a] < ll_idx_[b]);
  }
  
  private:
    uint32* ll_idx_;
};

void relation_tag_create_relation_id_idx(uint32* block_of_id, uint32 max_relation_id)
{
  const uint32 max_relations_ram = 32*1024*1024;
  
  int source_fd = open64(RELATION_TAG_TMPB, O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, RELATION_TAG_TMPB, "relation_tag_create_id_relation_idx:2");
  
  uint8* blocklet_of_id = (uint8*) malloc((RelationCollection::next_relation_tag_id+1)*sizeof(uint8));
  for (uint32 i(0); i < (RelationCollection::next_relation_tag_id+1); ++i)
  {
    if ((block_of_id[i] & 0xffffff00) == 0xffffff00)
      blocklet_of_id[i] = (block_of_id[i] & 0xff);
    else
      blocklet_of_id[i] = 0xff;
  }
  free(block_of_id);
  
  uint32 count(max_relations_ram);
  uint32* ll_idx_ = (uint32*) malloc(sizeof(uint32)*count);
  
  Tag_Relation_Id_Writer env(ll_idx_, blocklet_of_id);
  env.offset = 1;
  
  while (env.offset < max_relation_id)
  {
    env.ids_of_relation.clear();
    env.ids_of_relation.resize(count);
  
    cerr<<'n';
    Relation_Id_Relation_Dump dump(env.offset, count, ll_idx_);
    select_all< Relation_Id_Relation_Dump >(dump);
    cerr<<'n';
    lseek64(source_fd, 0, SEEK_SET);
    
    uint32* tag_rd_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
    
    uint32 rd_buf_pos(0), max_pos(0);
    while ((max_pos =
	    read(source_fd, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32))))
    {
      max_pos += rd_buf_pos*sizeof(uint32);
      rd_buf_pos = 0;
  
      while ((rd_buf_pos + 1 < max_pos / sizeof(uint32)) &&
	      (rd_buf_pos + tag_rd_buf[rd_buf_pos+1] + 1 < max_pos / sizeof(uint32)))
      {
	if (blocklet_of_id[tag_rd_buf[rd_buf_pos]] != 0xff)
	{
	  for (uint32 i(0); i < tag_rd_buf[rd_buf_pos+1]; ++i)
	  {
	    if ((tag_rd_buf[rd_buf_pos+2+i] >= env.offset) &&
			(tag_rd_buf[rd_buf_pos+2+i] - env.offset < count))
	    {
	      if (env.ids_of_relation[tag_rd_buf[rd_buf_pos+2+i] - env.offset].size() < 65535)
		env.ids_of_relation[tag_rd_buf[rd_buf_pos+2+i] - env.offset].push_back(tag_rd_buf[rd_buf_pos]);
	      else
	      {
		cerr<<"Relation "<<dec<<tag_rd_buf[rd_buf_pos+2+i]<<" has more than 2^16 tags.\n";
		exit(0);
	      }
	    }
	  }
	}
	rd_buf_pos += tag_rd_buf[rd_buf_pos+1] + 2;
      }
      cerr<<'t';
    
      memmove(tag_rd_buf, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32));
      rd_buf_pos = RELATION_TAG_SORT_BUFFER_SIZE / sizeof(uint32) - rd_buf_pos;
    }
    env.read_order.clear();
    for (uint32 i(0); i < count; ++i)
    {
      if (env.ids_of_relation[i].size() > 0)
	env.read_order.push_back(i);
    }
    sort(env.read_order.begin(), env.read_order.end(), relation_idx_less(ll_idx_));
    
    free(tag_rd_buf);
    
    flush_data< Tag_Relation_Id_Writer >(env, env.begin(), env.end());
    
    env.offset += count;
  }
  
  make_block_index< Tag_Relation_Id_Writer >(env);
    
  free(ll_idx_);
  free(blocklet_of_id);
  
  close(source_fd);
}

void relation_tag_id_statistics()
{
  int dest_fd = open64((DATADIR + db_subdir + RELATION_TAG_ID_STATS).c_str(),
                       O_WRONLY|O_CREAT|O_TRUNC,
		       S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64((DATADIR + db_subdir + RELATION_TAG_ID_STATS).c_str(),
                   O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_TAG_ID_STATS,
                     "relation_tag_id_statistics:1");
  
  vector< uint32 > id_count(RelationCollection::next_relation_tag_id);
  Relation_Tag_Id_Count_Local_Reader local_stats(id_count);
  select_all< Relation_Tag_Id_Count_Local_Reader >(local_stats);
  Relation_Tag_Id_Count_Global_Reader global_stats(id_count);
  select_all< Relation_Tag_Id_Count_Global_Reader >(global_stats);
  
  for (vector< uint32 >::const_iterator it(id_count.begin()); it != id_count.end(); ++it)
    write(dest_fd, &(*it), sizeof(uint32));

  close(dest_fd);
}

//-----------------------------------------------------------------------------

struct Relation_String_Cache
{
  static const vector< uint32 >& get_spatial_boundaries()
  {
    if (spatial_boundaries.empty())
      init();
    return spatial_boundaries;
  }
  
  static const vector< vector< pair< string, string > > >& get_kv_to_id_idx()
  {
    if (kv_to_id_idx.empty())
      init();
    return kv_to_id_idx;
  }
  
  static const vector< vector< uint16 > >& get_kv_to_id_block_idx()
  {
    if (kv_to_id_block_idx.empty())
      init();
    return kv_to_id_block_idx;
  }
  
  static uint32 get_next_relation_tag_id()
  {
    if (next_relation_tag_id == 0xffffffff)
      init();
    return next_relation_tag_id;
  }
  
  static void set_next_relation_tag_id(uint32 i)
  {
    next_relation_tag_id = i;
  }
  
  static void reset()
  {
    spatial_boundaries.clear();
    kv_to_id_idx.clear();
    kv_to_id_block_idx.clear();
    next_relation_tag_id = 0xffffffff;
  }
  
  private:
    static void init()
    {
      spatial_boundaries.clear();
    
      int string_idx_fd =
        open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(), O_RDONLY);
      if (string_idx_fd < 0)
	throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_IDX,
	                 "Relation_String_Cache.init():1");
  
      uint32* string_spat_idx_buf = (uint32*) malloc(RELATION_TAG_SPATIAL_PARTS*sizeof(uint32));
      read(string_idx_fd, &next_relation_tag_id, sizeof(uint32));
      read(string_idx_fd, string_spat_idx_buf, RELATION_TAG_SPATIAL_PARTS*sizeof(uint32));
      for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS; ++i)
	spatial_boundaries.push_back(string_spat_idx_buf[i]);
      free(string_spat_idx_buf);
    
      vector< vector< pair< pair< string, string >, uint16 > > >
	  kv_to_all(RELATION_TAG_SPATIAL_PARTS+1);
      
      uint16* kv_to_id_idx_buf_1 = (uint16*) malloc(3*sizeof(uint16));
      char* kv_to_id_idx_buf_2 = (char*) malloc(2*64*1024);
      uint32 block_id(0);
      while (read(string_idx_fd, kv_to_id_idx_buf_1, 3*sizeof(uint16)))
      {
	read(string_idx_fd, kv_to_id_idx_buf_2, kv_to_id_idx_buf_1[1] + kv_to_id_idx_buf_1[2]);
	kv_to_all[kv_to_id_idx_buf_1[0]].push_back(make_pair< pair< string, string >, uint16 >
	    (make_pair< string, string >
	     (string(kv_to_id_idx_buf_2, kv_to_id_idx_buf_1[1]),
	      string(&(kv_to_id_idx_buf_2[kv_to_id_idx_buf_1[1]]), kv_to_id_idx_buf_1[2])),
		       block_id++));
      }
      free(kv_to_id_idx_buf_2);
      free(kv_to_id_idx_buf_1);
  
      close(string_idx_fd);
    
      kv_to_id_idx.clear();
      kv_to_id_idx.resize(RELATION_TAG_SPATIAL_PARTS+1);
      kv_to_id_block_idx.clear();
      kv_to_id_block_idx.resize(RELATION_TAG_SPATIAL_PARTS+1);
      for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
      {
	sort(kv_to_all[i].begin(), kv_to_all[i].end());
	for (vector< pair< pair< string, string >, uint16 > >::const_iterator
		    it(kv_to_all[i].begin()); it != kv_to_all[i].end(); ++it)
	{
	  kv_to_id_idx[i].push_back(it->first);
	  cout<<i<<'\t'<<it->second<<'\t'<<it->first.first<<'\t'<<it->first.second<<'\n';
	  kv_to_id_block_idx[i].push_back(it->second);
	}
      }
    }
  <commit_msg>*** empty log message ***<commit_after>#include <algorithm>
#include <fstream>
#include <iostream>
#include <map>
#include <set>
#include <sstream>
#include <string>
#include <vector>

#include "../script_datatypes.h"
#include "file_types.h"
#include "relation_strings_file_io.h"
#include "raw_file_db.h"

#include <string.h>
#include <stdio.h>
#include <time.h>

#include <errno.h>
#include <fcntl.h>
#include <string.h>
#include <unistd.h>

using namespace std;

typedef char int8;
typedef short int int16;
typedef int int32;
typedef long long int64;

typedef unsigned char uint8;
typedef unsigned short int uint16;
typedef unsigned int uint32;
typedef unsigned long long uint64;

//-----------------------------------------------------------------------------

const string RELATION_STRING_DATA("relation_strings.dat");
const string RELATION_STRING_IDX("relation_strings.idx");

const char* RELATION_TAG_TMPAPREFIX = "/tmp/relation_strings_";
const char* RELATION_TAG_TMPB = "/tmp/relation_tag_ids";

//-----------------------------------------------------------------------------

uint32 RelationCollection::next_relation_tag_id(0);

void flush_relation_tags(uint& current_run, map< RelationKeyValue, RelationCollection >& relation_tags, vector< Relation_ >& relations)
{
  //determine whether a certain tag exists only local
  sort(relations.begin(), relations.end(), Relation_Less_By_Id());
  for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
       it != relation_tags.end(); ++it)
  {
    uint32 position(*(it->second.relations.begin()));
    uint32 bitmask(0);
    for (vector< uint32 >::iterator it2(it->second.relations.begin()); it2 != it->second.relations.end(); ++it2)
    {
      bitmask |= (position ^ (*it2));
    }
    it->second.position = position;
    it->second.bitmask = bitmask;
  }
  
  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<current_run;
  int dest_fd = open64(temp.str().c_str(), O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64(temp.str().c_str(), O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "flush_relation_tags:1");
  
  if (current_run == 0)
  {
    int dest_id_fd = open64(RELATION_TAG_TMPB, O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
    if (dest_id_fd < 0)
      throw File_Error(errno, RELATION_TAG_TMPB, "flush_relation_tags:2");
    
    for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
         it != relation_tags.end(); ++it)
    {
      uint16 key_size(it->first.key.size());
      uint16 value_size(it->first.value.size());
      if (it->second.bitmask>>8)
        it->second.position = 0xffffffff;
      else
        it->second.position &= 0xffffff00;
      write(dest_fd, &(it->second.id), sizeof(uint32));
      write(dest_fd, &(it->second.position), sizeof(uint32));
      write(dest_fd, &(key_size), sizeof(uint16));
      write(dest_fd, &(value_size), sizeof(uint16));
      write(dest_fd, &(it->first.key[0]), key_size);
      write(dest_fd, &(it->first.value[0]), value_size);
      
      uint32 nc_size(it->second.relations.size());
      write(dest_id_fd, &(it->second.id), sizeof(uint32));
      write(dest_id_fd, &(nc_size), sizeof(uint32));
      for (vector< uint32 >::const_iterator it2(it->second.relations.begin());
	   it2 != it->second.relations.end(); ++it2)
	write(dest_id_fd, &(*it2), sizeof(uint32));
    }
    
    close(dest_id_fd);
  }
  else
  {
    uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
    uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
    char* key_rd_buf = (char*) malloc(64*1024);
    char* value_rd_buf = (char*) malloc(64*1024);
    
    temp.str("");
    temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
    int source_fd = open64(temp.str().c_str(), O_RDONLY);
    if (source_fd < 0)
      throw File_Error(errno, temp.str().c_str(), "flush_relation_tags:3");
    
    map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
    while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
    {
      read(source_fd, key_rd_buf, size_rd_buf[0]);
      key_rd_buf[size_rd_buf[0]] = 0;
      read(source_fd, value_rd_buf, size_rd_buf[1]);
      value_rd_buf[size_rd_buf[1]] = 0;
      while ((it != relation_tags.end()) &&
             (strcmp(key_rd_buf, it->first.key.c_str()) > 0))
      {
        uint16 key_size(it->first.key.size());
        uint16 value_size(it->first.value.size());
        if (it->second.bitmask>>8)
          it->second.position = 0xffffffff;
        else
          it->second.position &= 0xffffff00;
        write(dest_fd, &(it->second.id), sizeof(uint32));
        write(dest_fd, &(it->second.position), sizeof(uint32));
        write(dest_fd, &(key_size), sizeof(uint16));
        write(dest_fd, &(value_size), sizeof(uint16));
        write(dest_fd, &(it->first.key[0]), key_size);
        write(dest_fd, &(it->first.value[0]), value_size);
        ++it;
      }
      while ((it != relation_tags.end()) &&
             (strcmp(key_rd_buf, it->first.key.c_str()) == 0) &&
             (strcmp(value_rd_buf, it->first.value.c_str()) > 0))
      {
        uint16 key_size(it->first.key.size());
        uint16 value_size(it->first.value.size());
        if (it->second.bitmask>>8)
          it->second.position = 0xffffffff;
        else
          it->second.position &= 0xffffff00;
        write(dest_fd, &(it->second.id), sizeof(uint32));
        write(dest_fd, &(it->second.position), sizeof(uint32));
        write(dest_fd, &(key_size), sizeof(uint16));
        write(dest_fd, &(value_size), sizeof(uint16));
        write(dest_fd, &(it->first.key[0]), key_size);
        write(dest_fd, &(it->first.value[0]), value_size);
        ++it;
      }
      if ((it != relation_tags.end()) &&
          (strcmp(key_rd_buf, it->first.key.c_str()) == 0) &&
          (strcmp(value_rd_buf, it->first.value.c_str()) == 0))
      {
        it->second.merge(cnt_rd_buf[0], cnt_rd_buf[1]);
        if ((cnt_rd_buf[1] == 0xffffffff) || (it->second.bitmask>>8))
          cnt_rd_buf[1] = 0xffffffff;
        else
          cnt_rd_buf[1] &= 0xffffff00;
        ++it;
      }
      write(dest_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16));
      write(dest_fd, key_rd_buf, size_rd_buf[0]);
      write(dest_fd, value_rd_buf, size_rd_buf[1]);
    }
    while (it != relation_tags.end())
    {
      uint16 key_size(it->first.key.size());
      uint16 value_size(it->first.value.size());
      if (it->second.bitmask>>8)
	it->second.position = 0xffffffff;
      else
	it->second.position &= 0xffffff00;
      write(dest_fd, &(it->second.id), sizeof(uint32));
      write(dest_fd, &(it->second.position), sizeof(uint32));
      write(dest_fd, &(key_size), sizeof(uint16));
      write(dest_fd, &(value_size), sizeof(uint16));
      write(dest_fd, &(it->first.key[0]), key_size);
      write(dest_fd, &(it->first.value[0]), value_size);
      ++it;
    }
    
    free(cnt_rd_buf);
    free(key_rd_buf);
    free(value_rd_buf);
    
    close(source_fd);
    temp.str("");
    temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
    remove(temp.str().c_str());
    
    int dest_id_fd = open64(RELATION_TAG_TMPB, O_WRONLY|O_APPEND, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
    if (dest_id_fd < 0)
      throw File_Error(errno, RELATION_TAG_TMPB, "flush_relation_tags:4");
    
    for (map< RelationKeyValue, RelationCollection >::iterator it(relation_tags.begin());
	 it != relation_tags.end(); ++it)
    {
      uint32 nc_size(it->second.relations.size());
      write(dest_id_fd, &(it->second.id), sizeof(uint32));
      write(dest_id_fd, &(nc_size), sizeof(uint32));
      for (vector< uint32 >::const_iterator it2(it->second.relations.begin());
	   it2 != it->second.relations.end(); ++it2)
	write(dest_id_fd, &(*it2), sizeof(uint32));	
    }
    
    close(dest_id_fd);
  }
  
  close(dest_fd);
  ++current_run;
}

void relation_tag_statistics(uint& current_run, vector< uint32 >& split_idx)
{
  cerr<<'s';
  
  if (current_run == 0)
  {
    cerr<<"No relation tags.\n";
    return;
  }
  
  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
  int source_fd = open64(temp.str().c_str(), O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "relation_tag_statistics:1");
  
  uint32 global_count(0);
  uint32* spatial_count = (uint32*) calloc(16*1024*1024, sizeof(uint32));
  uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
  uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
  
  while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
  {
    lseek64(source_fd, size_rd_buf[0] + size_rd_buf[1], SEEK_CUR);
    ++(spatial_count[cnt_rd_buf[1]>>8]);
    ++global_count;
  }
  global_count -= spatial_count[0x00ffffff];
  
  split_idx.clear();
  uint32 split_count(0);
  for (unsigned int i(0); i < 16*1024*1024; ++i)
  {
    split_count += spatial_count[i];
    if (split_count >= global_count / RELATION_TAG_SPATIAL_PARTS)
    {
      split_idx.push_back(i<<8);
      split_count -= global_count / RELATION_TAG_SPATIAL_PARTS;
    }
  }
  while (split_idx.size() < RELATION_TAG_SPATIAL_PARTS)
    split_idx.push_back(0xffffff<<8);
  split_idx[RELATION_TAG_SPATIAL_PARTS-1] = 0xffffff<<8;
  
  free(cnt_rd_buf);
  free(spatial_count);
  
  close(source_fd);

  cerr<<'s';
}

void relation_tag_split_and_index(uint& current_run, vector< uint32 >& split_idx, uint32*& block_of_id)
{
  cerr<<'p';

  ostringstream temp;
  temp<<RELATION_TAG_TMPAPREFIX<<(current_run-1);
  int source_fd = open64(temp.str().c_str(), O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, temp.str().c_str(), "relation_tag_split_and_index:1");
  
  int dest_fd = open64((DATADIR + db_subdir + RELATION_STRING_DATA).c_str(),
                       O_WRONLY|O_CREAT|O_TRUNC,
		       S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64((DATADIR + db_subdir + RELATION_STRING_DATA).c_str(),
                   O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_DATA,
                     "relation_tag_split_and_index:2");
  
  int dest_idx_fd =
    open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(),
           O_WRONLY|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_idx_fd);
  
  dest_idx_fd = open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(),
                       O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_IDX,
                     "relation_tag_split_and_index:3");
  
  write(dest_idx_fd, &(RelationCollection::next_relation_tag_id), sizeof(uint32));
  for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS; ++i)
    write(dest_idx_fd, &(split_idx[i]), sizeof(uint32));
  
  uint8* cur_block_count = (uint8*) calloc(RELATION_TAG_SPATIAL_PARTS+1, sizeof(uint8));
  block_of_id = (uint32*) calloc((RelationCollection::next_relation_tag_id+1), sizeof(uint32));
  
  uint32* write_pos = (uint32*) malloc((RELATION_TAG_SPATIAL_PARTS+1)*sizeof(uint32));
  for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
    write_pos[i] = sizeof(uint32);
  char* write_blocks = (char*) malloc((RELATION_TAG_SPATIAL_PARTS+1) * RELATION_STRING_BLOCK_SIZE);
  uint32* cnt_rd_buf = (uint32*) malloc(2*sizeof(uint32) + 2*sizeof(uint16));
  uint16* size_rd_buf = (uint16*) &(cnt_rd_buf[2]);
  
  while (read(source_fd, cnt_rd_buf, 2*sizeof(uint32) + 2*sizeof(uint16)))
  {
    uint16 block(0);
    if (cnt_rd_buf[1] != 0xffffffff)
    {
      while (split_idx[block] <= cnt_rd_buf[1])
	++block;
      ++block;
    }
    
    if (write_pos[block] + 2*sizeof(uint32) + 2*sizeof(uint16) + size_rd_buf[0] + size_rd_buf[1]
	>= RELATION_STRING_BLOCK_SIZE*4/5)
    {
      *((uint32*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE])) = write_pos[block] - sizeof(uint32);
      write(dest_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE]), RELATION_STRING_BLOCK_SIZE);
      write(dest_idx_fd, &(block), sizeof(uint16));
      write(dest_idx_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE
	  + 3*sizeof(uint32)]), *(uint16*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE
	      + 3*sizeof(uint32)]) + *(uint16*)&(write_blocks[block * RELATION_STRING_BLOCK_SIZE
		  + 3*sizeof(uint32) + sizeof(uint16)]) + 2*sizeof(uint16));
    
      write_pos[block] = sizeof(uint32);

      memcpy(&(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]), cnt_rd_buf,
	       2*sizeof(uint32) + 2*sizeof(uint16));
      write_pos[block] += 2*sizeof(uint32) + 2*sizeof(uint16);
      read(source_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]),
	   size_rd_buf[0] + size_rd_buf[1]);
      write_pos[block] += size_rd_buf[0] + size_rd_buf[1];
      
      block_of_id[cnt_rd_buf[0]] = (++cur_block_count[block]) | ((cnt_rd_buf[1]) & (0xffffff00));
    }
    else
    {
      memcpy(&(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]), cnt_rd_buf,
	       2*sizeof(uint32) + 2*sizeof(uint16));
      write_pos[block] += 2*sizeof(uint32) + 2*sizeof(uint16);
      read(source_fd, &(write_blocks[block * RELATION_STRING_BLOCK_SIZE + write_pos[block]]),
	   size_rd_buf[0] + size_rd_buf[1]);
      write_pos[block] += size_rd_buf[0] + size_rd_buf[1];
    
      block_of_id[cnt_rd_buf[0]] = (cur_block_count[block]) | ((cnt_rd_buf[1]) & (0xffffff00));
    }
  }
  
  for (unsigned int i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
  {
    *((uint32*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE])) = write_pos[i] - sizeof(uint32);
    write(dest_fd, &(write_blocks[i * RELATION_STRING_BLOCK_SIZE]), RELATION_STRING_BLOCK_SIZE);
    write(dest_idx_fd, &(i), sizeof(uint16));
    write(dest_idx_fd, &(write_blocks[i * RELATION_STRING_BLOCK_SIZE
	+ 3*sizeof(uint32)]), *(uint16*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE
	    + 3*sizeof(uint32)]) + *(uint16*)&(write_blocks[i * RELATION_STRING_BLOCK_SIZE
		+ 3*sizeof(uint32) + sizeof(uint16)]) + 2*sizeof(uint16));
  }
  
  free(cur_block_count);
  free(write_pos);
  free(write_blocks);
  free(cnt_rd_buf);
  
  close(source_fd);
  close(dest_fd);

  cerr<<'p';
}

struct tag_id_local_less : public binary_function< uint32*, uint32*, bool >
{
  tag_id_local_less(uint32* block_of_id) : block_of_id_(block_of_id) {}

  bool operator() (uint32* const& a, uint32* const& b)
  {
    if ((block_of_id_[*a] & 0xffffff00) < (block_of_id_[*b] & 0xffffff00))
      return true;
    else if ((block_of_id_[*a] & 0xffffff00) > (block_of_id_[*b] & 0xffffff00))
      return false;
    return (*a < *b);
  }
private:
  uint32* block_of_id_;
};

struct tag_id_global_less : public binary_function< uint32*, uint32*, bool >
{
  bool operator() (uint32* const& a, uint32* const& b)
  {
    return (*a < *b);
  }
};

void relation_tag_create_id_relation_idx(uint32* block_of_id)
{
  uint32 rd_buf_pos(0);
  Tag_Id_Relation_Local_Writer env_local(block_of_id);
  Tag_Id_Relation_Global_Writer env_global;
  
  int source_fd = open64(RELATION_TAG_TMPB, O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, RELATION_TAG_TMPB, "relation_tag_create_id_relation_idx:1");
  
  uint32* tag_rd_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
  uint32* tag_alt_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
  uint32 max_pos(0);
  
  while ((max_pos =
	 read(source_fd, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32))))
  {
    vector< uint32* > tag_id_local, tag_id_global;
    uint32 alt_buf_pos(0);
    max_pos += rd_buf_pos*sizeof(uint32);
    rd_buf_pos = 0;
  
    while ((rd_buf_pos + 1 < max_pos / sizeof(uint32)) &&
	    (rd_buf_pos + tag_rd_buf[rd_buf_pos+1] + 1 < max_pos / sizeof(uint32)))
    {
      uint32 next_pos(tag_rd_buf[rd_buf_pos+1] + 2);
      
      if ((~(block_of_id[tag_rd_buf[rd_buf_pos]])) & 0xffffff00)
      {
	tag_id_local.push_back(&(tag_rd_buf[rd_buf_pos]));
      
	while (tag_rd_buf[rd_buf_pos+1] > 255)
	{
	  if (tag_rd_buf[rd_buf_pos+1] > 510)
	  {
	    tag_id_local.push_back(&(tag_alt_buf[alt_buf_pos]));
	    tag_id_local.push_back(&(tag_rd_buf[rd_buf_pos + 510]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]), 255*sizeof(uint32));
	    alt_buf_pos += 255;
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  
	    next_pos -= 510;
	    rd_buf_pos += 510;
	  
	    tag_rd_buf[rd_buf_pos] = tag_rd_buf[rd_buf_pos - 510];
	    tag_rd_buf[rd_buf_pos+1] = next_pos - 2;
	  }
	  else
	  {
	    tag_id_local.push_back(&(tag_alt_buf[alt_buf_pos]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos+1] - 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]),
		     (tag_rd_buf[rd_buf_pos+1] - 255)*sizeof(uint32));
	    alt_buf_pos += (tag_rd_buf[rd_buf_pos+1] - 255);
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  }
	}
      }
      else
      {
	tag_id_global.push_back(&(tag_rd_buf[rd_buf_pos]));
      
	while (tag_rd_buf[rd_buf_pos+1] > 255)
	{
	  if (tag_rd_buf[rd_buf_pos+1] > 510)
	  {
	    tag_id_global.push_back(&(tag_alt_buf[alt_buf_pos]));
	    tag_id_global.push_back(&(tag_rd_buf[rd_buf_pos + 510]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]), 255*sizeof(uint32));
	    alt_buf_pos += 255;
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  
	    next_pos -= 510;
	    rd_buf_pos += 510;
	  
	    tag_rd_buf[rd_buf_pos] = tag_rd_buf[rd_buf_pos - 510];
	    tag_rd_buf[rd_buf_pos+1] = next_pos - 2;
	  }
	  else
	  {
	    tag_id_global.push_back(&(tag_alt_buf[alt_buf_pos]));
	  
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos];
	    tag_alt_buf[alt_buf_pos++] = tag_rd_buf[rd_buf_pos+1] - 255;
	    memcpy(&(tag_alt_buf[alt_buf_pos]), &(tag_rd_buf[rd_buf_pos+257]),
		     (tag_rd_buf[rd_buf_pos+1] - 255)*sizeof(uint32));
	    alt_buf_pos += (tag_rd_buf[rd_buf_pos+1] - 255);
	    tag_rd_buf[rd_buf_pos+1] = 255;
	  }
	}
      }
      
      rd_buf_pos += next_pos;
    }
    
    sort(tag_id_local.begin(), tag_id_local.end(), tag_id_local_less(block_of_id));
    sort(tag_id_global.begin(), tag_id_global.end(), tag_id_global_less());
    
    flush_data< Tag_Id_Relation_Local_Writer >
	(env_local, tag_id_local.begin(), tag_id_local.end());
    flush_data< Tag_Id_Relation_Global_Writer >
        (env_global, tag_id_global.begin(), tag_id_global.end());
    
    cerr<<'.';
    
    memmove(tag_rd_buf, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32));
    rd_buf_pos = RELATION_TAG_SORT_BUFFER_SIZE / sizeof(uint32) - rd_buf_pos;
  }
  
  make_block_index< Tag_Id_Relation_Local_Writer >(env_local);
  make_block_index< Tag_Id_Relation_Global_Writer >(env_global);
  
  free(tag_rd_buf);
  free(tag_alt_buf);
  
  close(source_fd);
}

struct relation_idx_less : public binary_function< uint32, uint32, bool >
{
  relation_idx_less(uint32* ll_idx__) : ll_idx_(ll_idx__) {}
  
  bool operator() (const uint32& a, const uint32& b)
  {
    return (ll_idx_[a] < ll_idx_[b]);
  }
  
  private:
    uint32* ll_idx_;
};

void relation_tag_create_relation_id_idx(uint32* block_of_id, uint32 max_relation_id)
{
  const uint32 max_relations_ram = 32*1024*1024;
  
  int source_fd = open64(RELATION_TAG_TMPB, O_RDONLY);
  if (source_fd < 0)
    throw File_Error(errno, RELATION_TAG_TMPB, "relation_tag_create_id_relation_idx:2");
  
  uint8* blocklet_of_id = (uint8*) malloc((RelationCollection::next_relation_tag_id+1)*sizeof(uint8));
  for (uint32 i(0); i < (RelationCollection::next_relation_tag_id+1); ++i)
  {
    if ((block_of_id[i] & 0xffffff00) == 0xffffff00)
      blocklet_of_id[i] = (block_of_id[i] & 0xff);
    else
      blocklet_of_id[i] = 0xff;
  }
  free(block_of_id);
  
  uint32 count(max_relations_ram);
  uint32* ll_idx_ = (uint32*) malloc(sizeof(uint32)*count);
  
  Tag_Relation_Id_Writer env(ll_idx_, blocklet_of_id);
  env.offset = 1;
  
  while (env.offset < max_relation_id)
  {
    env.ids_of_relation.clear();
    env.ids_of_relation.resize(count);
  
    cerr<<'n';
    Relation_Id_Relation_Dump dump(env.offset, count, ll_idx_);
    select_all< Relation_Id_Relation_Dump >(dump);
    cerr<<'n';
    lseek64(source_fd, 0, SEEK_SET);
    
    uint32* tag_rd_buf = (uint32*) malloc(RELATION_TAG_SORT_BUFFER_SIZE);
    
    uint32 rd_buf_pos(0), max_pos(0);
    while ((max_pos =
	    read(source_fd, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32))))
    {
      max_pos += rd_buf_pos*sizeof(uint32);
      rd_buf_pos = 0;
  
      while ((rd_buf_pos + 1 < max_pos / sizeof(uint32)) &&
	      (rd_buf_pos + tag_rd_buf[rd_buf_pos+1] + 1 < max_pos / sizeof(uint32)))
      {
	if (blocklet_of_id[tag_rd_buf[rd_buf_pos]] != 0xff)
	{
	  for (uint32 i(0); i < tag_rd_buf[rd_buf_pos+1]; ++i)
	  {
	    if ((tag_rd_buf[rd_buf_pos+2+i] >= env.offset) &&
			(tag_rd_buf[rd_buf_pos+2+i] - env.offset < count))
	    {
	      if (env.ids_of_relation[tag_rd_buf[rd_buf_pos+2+i] - env.offset].size() < 65535)
		env.ids_of_relation[tag_rd_buf[rd_buf_pos+2+i] - env.offset].push_back(tag_rd_buf[rd_buf_pos]);
	      else
	      {
		cerr<<"Relation "<<dec<<tag_rd_buf[rd_buf_pos+2+i]<<" has more than 2^16 tags.\n";
		exit(0);
	      }
	    }
	  }
	}
	rd_buf_pos += tag_rd_buf[rd_buf_pos+1] + 2;
      }
      cerr<<'t';
    
      memmove(tag_rd_buf, &(tag_rd_buf[rd_buf_pos]), RELATION_TAG_SORT_BUFFER_SIZE - rd_buf_pos*sizeof(uint32));
      rd_buf_pos = RELATION_TAG_SORT_BUFFER_SIZE / sizeof(uint32) - rd_buf_pos;
    }
    env.read_order.clear();
    for (uint32 i(0); i < count; ++i)
    {
      if (env.ids_of_relation[i].size() > 0)
	env.read_order.push_back(i);
    }
    sort(env.read_order.begin(), env.read_order.end(), relation_idx_less(ll_idx_));
    
    free(tag_rd_buf);
    
    flush_data< Tag_Relation_Id_Writer >(env, env.begin(), env.end());
    
    env.offset += count;
  }
  
  make_block_index< Tag_Relation_Id_Writer >(env);
    
  free(ll_idx_);
  free(blocklet_of_id);
  
  close(source_fd);
}

void relation_tag_id_statistics()
{
  int dest_fd = open64((DATADIR + db_subdir + RELATION_TAG_ID_STATS).c_str(),
                       O_WRONLY|O_CREAT|O_TRUNC,
		       S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  close(dest_fd);
  
  dest_fd = open64((DATADIR + db_subdir + RELATION_TAG_ID_STATS).c_str(),
                   O_WRONLY|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
  if (dest_fd < 0)
    throw File_Error(errno, DATADIR + db_subdir + RELATION_TAG_ID_STATS,
                     "relation_tag_id_statistics:1");
  
  vector< uint32 > id_count(RelationCollection::next_relation_tag_id);
  Relation_Tag_Id_Count_Local_Reader local_stats(id_count);
  select_all< Relation_Tag_Id_Count_Local_Reader >(local_stats);
  Relation_Tag_Id_Count_Global_Reader global_stats(id_count);
  select_all< Relation_Tag_Id_Count_Global_Reader >(global_stats);
  
  for (vector< uint32 >::const_iterator it(id_count.begin()); it != id_count.end(); ++it)
    write(dest_fd, &(*it), sizeof(uint32));

  close(dest_fd);
}

//-----------------------------------------------------------------------------

struct Relation_String_Cache
{
  static const vector< uint32 >& get_spatial_boundaries()
  {
    if (spatial_boundaries.empty())
      init();
    return spatial_boundaries;
  }
  
  static const vector< vector< pair< string, string > > >& get_kv_to_id_idx()
  {
    if (kv_to_id_idx.empty())
      init();
    return kv_to_id_idx;
  }
  
  static const vector< vector< uint16 > >& get_kv_to_id_block_idx()
  {
    if (kv_to_id_block_idx.empty())
      init();
    return kv_to_id_block_idx;
  }
  
  static uint32 get_next_relation_tag_id()
  {
    if (next_relation_tag_id == 0xffffffff)
      init();
    return next_relation_tag_id;
  }
  
  static void set_next_relation_tag_id(uint32 i)
  {
    next_relation_tag_id = i;
  }
  
  static void reset()
  {
    spatial_boundaries.clear();
    kv_to_id_idx.clear();
    kv_to_id_block_idx.clear();
    next_relation_tag_id = 0xffffffff;
  }
  
  private:
    static void init()
    {
      spatial_boundaries.clear();
    
      int string_idx_fd =
        open64((DATADIR + db_subdir + RELATION_STRING_IDX).c_str(), O_RDONLY);
      if (string_idx_fd < 0)
	throw File_Error(errno, DATADIR + db_subdir + RELATION_STRING_IDX,
	                 "Relation_String_Cache.init():1");
  
      uint32* string_spat_idx_buf = (uint32*) malloc(RELATION_TAG_SPATIAL_PARTS*sizeof(uint32));
      read(string_idx_fd, &next_relation_tag_id, sizeof(uint32));
      read(string_idx_fd, string_spat_idx_buf, RELATION_TAG_SPATIAL_PARTS*sizeof(uint32));
      for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS; ++i)
	spatial_boundaries.push_back(string_spat_idx_buf[i]);
      free(string_spat_idx_buf);
    
      vector< vector< pair< pair< string, string >, uint16 > > >
	  kv_to_all(RELATION_TAG_SPATIAL_PARTS+1);
      
      uint16* kv_to_id_idx_buf_1 = (uint16*) malloc(3*sizeof(uint16));
      char* kv_to_id_idx_buf_2 = (char*) malloc(2*64*1024);
      uint32 block_id(0);
      while (read(string_idx_fd, kv_to_id_idx_buf_1, 3*sizeof(uint16)))
      {
	read(string_idx_fd, kv_to_id_idx_buf_2, kv_to_id_idx_buf_1[1] + kv_to_id_idx_buf_1[2]);
	kv_to_all[kv_to_id_idx_buf_1[0]].push_back(make_pair< pair< string, string >, uint16 >
	    (make_pair< string, string >
	     (string(kv_to_id_idx_buf_2, kv_to_id_idx_buf_1[1]),
	      string(&(kv_to_id_idx_buf_2[kv_to_id_idx_buf_1[1]]), kv_to_id_idx_buf_1[2])),
		       block_id++));
      }
      free(kv_to_id_idx_buf_2);
      free(kv_to_id_idx_buf_1);
  
      close(string_idx_fd);
    
      kv_to_id_idx.clear();
      kv_to_id_idx.resize(RELATION_TAG_SPATIAL_PARTS+1);
      kv_to_id_block_idx.clear();
      kv_to_id_block_idx.resize(RELATION_TAG_SPATIAL_PARTS+1);
      for (uint32 i(0); i < RELATION_TAG_SPATIAL_PARTS+1; ++i)
      {
	sort(kv_to_all[i].begin(), kv_to_all[i].end());
	for (vector< pair< pair< string, string >, uint16 > >::const_iterator
		    it(kv_to_all[i].begin()); it != kv_to_all[i].end(); ++it)
	{
	  kv_to_id_idx[i].push_back(it->first);
	  kv_to_id_block_idx[i].push_back(it->second);
	}
      }
    }
  
{'fraction_non_alphanumeric': 0.15715549865381356, 'fraction_numerical': 0.03306158567582324, 'mean_word_length': 3.858580314672521, 'pattern_counts': {'":': 0, '<': 305, '<?xml version=': 0, '>': 282, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 18, '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': '5981788', 'n_tokens_mistral': 22579, 'n_tokens_neox': 21251, 'n_words': 2921}