StringDictionary Class Reference

#include <StringDictionary.h>

Collaboration diagram for StringDictionary:

Classes
struct	compare_cache_value_t
struct	PayloadString
struct	StringIdxEntry

Public Member Functions
	StringDictionary (const std::string &folder, const bool isTemp, const bool recover, const bool materializeHashes=false, size_t initial_capacity=256)
	StringDictionary (const LeafHostInfo &host, const DictRef dict_ref)
	~StringDictionary () noexcept
int32_t	getOrAdd (const std::string &str) noexcept
template<class T >
void	getOrAddBulk (const std::vector< std::string > &string_vec, T *encoded_vec)
void	getOrAddBulkArray (const std::vector< std::vector< std::string >> &string_array_vec, std::vector< std::vector< int32_t >> &ids_array_vec)
int32_t	getIdOfString (const std::string &str) const
std::string	getString (int32_t string_id) const
std::pair< char *, size_t >	getStringBytes (int32_t string_id) const noexcept
size_t	storageEntryCount () const
std::vector< int32_t >	getLike (const std::string &pattern, const bool icase, const bool is_simple, const char escape, const size_t generation) const
std::vector< int32_t >	getCompare (const std::string &pattern, const std::string &comp_operator, const size_t generation)
std::vector< int32_t >	getRegexpLike (const std::string &pattern, const char escape, const size_t generation) const
std::shared_ptr< const std::vector< std::string > >	copyStrings () const
bool	checkpoint () noexcept

Static Public Member Functions
static void	populate_string_ids (std::vector< int32_t > &dest_ids, StringDictionary *dest_dict, const std::vector< int32_t > &source_ids, const StringDictionary *source_dict, const std::map< int32_t, std::string > transient_mapping={})
	Populates provided dest_ids vector with string ids corresponding to given source strings. More...
static void	populate_string_array_ids (std::vector< std::vector< int32_t >> &dest_array_ids, StringDictionary *dest_dict, const std::vector< std::vector< int32_t >> &source_array_ids, const StringDictionary *source_dict)

Static Public Attributes
static constexpr int32_t	INVALID_STR_ID = -1
static constexpr size_t	MAX_STRLEN = (1 << 15) - 1
static constexpr size_t	MAX_STRCOUNT = (1U << 31) - 1

Private Member Functions
void	processDictionaryFutures (std::vector< std::future< std::vector< std::pair< uint32_t, unsigned int >>>> &dictionary_futures)
bool	fillRateIsHigh () const noexcept
void	increaseCapacity () noexcept
int32_t	getOrAddImpl (const std::string &str) noexcept
template<class T >
void	getOrAddBulkRemote (const std::vector< std::string > &string_vec, T *encoded_vec)
int32_t	getUnlocked (const std::string &str) const noexcept
std::string	getStringUnlocked (int32_t string_id) const noexcept
std::string	getStringChecked (const int string_id) const noexcept
std::pair< char *, size_t >	getStringBytesChecked (const int string_id) const noexcept
uint32_t	computeBucket (const uint32_t hash, const std::string str, const std::vector< int32_t > &data, const bool unique) const noexcept
uint32_t	computeUniqueBucketWithHash (const uint32_t hash, const std::vector< int32_t > &data) const noexcept
void	appendToStorage (const std::string &str) noexcept
PayloadString	getStringFromStorage (const int string_id) const noexcept
void	addPayloadCapacity () noexcept
void	addOffsetCapacity () noexcept
size_t	addStorageCapacity (int fd) noexcept
void *	addMemoryCapacity (void *addr, size_t &mem_size) noexcept
void	invalidateInvertedIndex () noexcept
std::vector< int32_t >	getEquals (std::string pattern, std::string comp_operator, size_t generation)
void	buildSortedCache ()
void	insertInSortedCache (std::string str, int32_t str_id)
void	sortCache (std::vector< int32_t > &cache)
void	mergeSortedCache (std::vector< int32_t > &temp_sorted_cache)
compare_cache_value_t *	binary_search_cache (const std::string &pattern) const

Private Attributes
size_t	str_count_
std::vector< int32_t >	str_ids_
std::vector< uint32_t >	rk_hashes_
std::vector< int32_t >	sorted_cache
bool	isTemp_
bool	materialize_hashes_
std::string	offsets_path_
int	payload_fd_
int	offset_fd_
StringIdxEntry *	offset_map_
char *	payload_map_
size_t	offset_file_size_
size_t	payload_file_size_
size_t	payload_file_off_
mapd_shared_mutex	rw_mutex_
std::map< std::tuple< std::string, bool, bool, char >, std::vector< int32_t > >	like_cache_
std::map< std::pair< std::string, char >, std::vector< int32_t > >	regex_cache_
std::map< std::string, int32_t >	equal_cache_
DictionaryCache< std::string, compare_cache_value_t >	compare_cache_
std::shared_ptr< std::vector< std::string > >	strings_cache_
std::unique_ptr< StringDictionaryClient >	client_
std::unique_ptr< StringDictionaryClient >	client_no_timeout_

Static Private Attributes
static char *	CANARY_BUFFER {nullptr}

Detailed Description

Definition at line 41 of file StringDictionary.h.

Constructor & Destructor Documentation

StringDictionary() [1/2]

StringDictionary::StringDictionary	(	const std::string &	folder,
		const bool	isTemp,
		const bool	recover,
		const bool	materializeHashes = false,
		size_t	initial_capacity = 256
	)

Definition at line 106 of file StringDictionary.cpp.

References addOffsetCapacity(), addPayloadCapacity(), CHECK_EQ, anonymous_namespace{StringDictionary.cpp}::checked_mmap(), anonymous_namespace{StringDictionary.cpp}::checked_open(), anonymous_namespace{StringDictionary.cpp}::file_size(), getStringFromStorage(), INVALID_STR_ID, isTemp_, LOG, materialize_hashes_, offset_fd_, offset_file_size_, offset_map_, offsets_path_, payload_fd_, payload_file_size_, payload_map_, processDictionaryFutures(), anonymous_namespace{StringDictionary.cpp}::rk_hash(), rk_hashes_, anonymous_namespace{StringDictionary.cpp}::round_up_p2(), rw_mutex_, str_ids_, and logger::WARNING.

    : str_count_(0)
    , str_ids_(initial_capacity, INVALID_STR_ID)
    , rk_hashes_(initial_capacity)
    , isTemp_(isTemp)
    , materialize_hashes_(materializeHashes)
    , payload_fd_(-1)
    , offset_fd_(-1)
    , offset_map_(nullptr)
    , payload_map_(nullptr)
    , offset_file_size_(0)
    , payload_file_size_(0)
    , payload_file_off_(0)
    , strings_cache_(nullptr) {
  if (!isTemp && folder.empty()) {
    return;
  }

  // initial capacity must be a power of two for efficient bucket computation
  CHECK_EQ(size_t(0), (initial_capacity & (initial_capacity - 1)));
  if (!isTemp_) {
    boost::filesystem::path storage_path(folder);
    offsets_path_ = (storage_path / boost::filesystem::path("DictOffsets")).string();
    const auto payload_path =
        (storage_path / boost::filesystem::path("DictPayload")).string();
    payload_fd_ = checked_open(payload_path.c_str(), recover);
    offset_fd_ = checked_open(offsets_path_.c_str(), recover);
    payload_file_size_ = file_size(payload_fd_);
    offset_file_size_ = file_size(offset_fd_);
  }

  if (payload_file_size_ == 0) {
    addPayloadCapacity();
  }
  if (offset_file_size_ == 0) {
    addOffsetCapacity();
  }
  if (!isTemp_) {  // we never mmap or recover temp dictionaries
    payload_map_ = reinterpret_cast<char*>(checked_mmap(payload_fd_, payload_file_size_));
    offset_map_ =
        reinterpret_cast<StringIdxEntry*>(checked_mmap(offset_fd_, offset_file_size_));
    if (recover) {
      const size_t bytes = file_size(offset_fd_);
      if (bytes % sizeof(StringIdxEntry) != 0) {
        LOG(WARNING) << "Offsets " << offsets_path_ << " file is truncated";
      }
      const uint64_t str_count = bytes / sizeof(StringIdxEntry);
      // at this point we know the size of the StringDict we need to load
      // so lets reallocate the vector to the correct size
      const uint64_t max_entries = round_up_p2(str_count * 2 + 1);
      std::vector<int32_t> new_str_ids(max_entries, INVALID_STR_ID);
      str_ids_.swap(new_str_ids);
      if (materialize_hashes_) {
        std::vector<uint32_t> new_rk_hashes(max_entries / 2);
        rk_hashes_.swap(new_rk_hashes);
      }
      unsigned string_id = 0;
      mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);

      uint32_t thread_inits = 0;
      const auto thread_count = std::thread::hardware_concurrency();
      const uint32_t items_per_thread = std::max<uint32_t>(
          2000, std::min<uint32_t>(200000, (str_count / thread_count) + 1));
      std::vector<std::future<std::vector<std::pair<uint32_t, unsigned int>>>>
          dictionary_futures;
      for (string_id = 0; string_id < str_count; string_id += items_per_thread) {
        dictionary_futures.emplace_back(std::async(
            std::launch::async, [string_id, str_count, items_per_thread, this] {
              std::vector<std::pair<uint32_t, unsigned int>> hashVec;
              for (uint32_t curr_id = string_id;
                   curr_id < string_id + items_per_thread && curr_id < str_count;
                   curr_id++) {
                const auto recovered = getStringFromStorage(curr_id);
                if (recovered.canary) {
                  // hit the canary, recovery finished
                  break;
                } else {
                  std::string temp(recovered.c_str_ptr, recovered.size);
                  hashVec.emplace_back(std::make_pair(rk_hash(temp), temp.size()));
                }
              }
              return hashVec;
            }));
        thread_inits++;
        if (thread_inits % thread_count == 0) {
          processDictionaryFutures(dictionary_futures);
        }
      }
      // gather last few threads
      if (dictionary_futures.size() != 0) {
        processDictionaryFutures(dictionary_futures);
      }
    }
  }
}

Here is the call graph for this function:

StringDictionary() [2/2]

StringDictionary::StringDictionary	(	const LeafHostInfo &	host,
		const DictRef	dict_ref
	)

Definition at line 225 of file StringDictionary.cpp.

    : strings_cache_(nullptr)
    , client_(new StringDictionaryClient(host, dict_ref, true))
    , client_no_timeout_(new StringDictionaryClient(host, dict_ref, false)) {}

~StringDictionary()

StringDictionary::~StringDictionary ( )

noexcept

Definition at line 230 of file StringDictionary.cpp.

References CHECK, CHECK_GE, anonymous_namespace{StringDictionary.cpp}::checked_munmap(), client_, File_Namespace::close(), isTemp_, offset_fd_, offset_file_size_, offset_map_, payload_fd_, payload_file_size_, and payload_map_.

                                             {
  if (client_) {
    return;
  }
  if (payload_map_) {
    if (!isTemp_) {
      CHECK(offset_map_);
      checked_munmap(payload_map_, payload_file_size_);
      checked_munmap(offset_map_, offset_file_size_);
      CHECK_GE(payload_fd_, 0);
      close(payload_fd_);
      CHECK_GE(offset_fd_, 0);
      close(offset_fd_);
    } else {
      CHECK(offset_map_);
      free(payload_map_);
      free(offset_map_);
    }
  }
}

Here is the call graph for this function:

Member Function Documentation

addMemoryCapacity()

void * StringDictionary::addMemoryCapacity ( void *  addr, size_t &  mem_size  )

privatenoexcept

Definition at line 1039 of file StringDictionary.cpp.

References CANARY_BUFFER, CHECK, and anonymous_namespace{StringDictionary.cpp}::SYSTEM_PAGE_SIZE.

Referenced by addOffsetCapacity(), and addPayloadCapacity().

                                                                               {
  static const ssize_t CANARY_BUFF_SIZE = 1024 * SYSTEM_PAGE_SIZE;
  if (!CANARY_BUFFER) {
    CANARY_BUFFER = reinterpret_cast<char*>(malloc(CANARY_BUFF_SIZE));
    CHECK(CANARY_BUFFER);
    memset(CANARY_BUFFER, 0xff, CANARY_BUFF_SIZE);
  }
  void* new_addr = realloc(addr, mem_size + CANARY_BUFF_SIZE);
  CHECK(new_addr);
  void* write_addr = reinterpret_cast<void*>(static_cast<char*>(new_addr) + mem_size);
  CHECK(memcpy(write_addr, CANARY_BUFFER, CANARY_BUFF_SIZE));
  mem_size += CANARY_BUFF_SIZE;
  return new_addr;
}

Here is the caller graph for this function:

addOffsetCapacity()

void StringDictionary::addOffsetCapacity ( )

privatenoexcept

Definition at line 1018 of file StringDictionary.cpp.

References addMemoryCapacity(), addStorageCapacity(), isTemp_, offset_fd_, offset_file_size_, and offset_map_.

Referenced by appendToStorage(), and StringDictionary().

                                                  {
  if (!isTemp_) {
    offset_file_size_ += addStorageCapacity(offset_fd_);
  } else {
    offset_map_ =
        static_cast<StringIdxEntry*>(addMemoryCapacity(offset_map_, offset_file_size_));
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

addPayloadCapacity()

void StringDictionary::addPayloadCapacity ( )

privatenoexcept

Definition at line 1009 of file StringDictionary.cpp.

References addMemoryCapacity(), addStorageCapacity(), isTemp_, payload_fd_, payload_file_size_, and payload_map_.

Referenced by appendToStorage(), and StringDictionary().

                                                   {
  if (!isTemp_) {
    payload_file_size_ += addStorageCapacity(payload_fd_);
  } else {
    payload_map_ =
        static_cast<char*>(addMemoryCapacity(payload_map_, payload_file_size_));
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

addStorageCapacity()

size_t StringDictionary::addStorageCapacity ( int  fd )

privatenoexcept

Definition at line 1027 of file StringDictionary.cpp.

References CANARY_BUFFER, CHECK, CHECK_NE, anonymous_namespace{StringDictionary.cpp}::SYSTEM_PAGE_SIZE, and File_Namespace::write().

Referenced by addOffsetCapacity(), and addPayloadCapacity().

                                                           {
  static const ssize_t CANARY_BUFF_SIZE = 1024 * SYSTEM_PAGE_SIZE;
  if (!CANARY_BUFFER) {
    CANARY_BUFFER = static_cast<char*>(malloc(CANARY_BUFF_SIZE));
    CHECK(CANARY_BUFFER);
    memset(CANARY_BUFFER, 0xff, CANARY_BUFF_SIZE);
  }
  CHECK_NE(lseek(fd, 0, SEEK_END), -1);
  CHECK(write(fd, CANARY_BUFFER, CANARY_BUFF_SIZE) == CANARY_BUFF_SIZE);
  return CANARY_BUFF_SIZE;
}

Here is the call graph for this function:

Here is the caller graph for this function:

appendToStorage()

void StringDictionary::appendToStorage ( const std::string &  str )

privatenoexcept

Definition at line 958 of file StringDictionary.cpp.

References addOffsetCapacity(), addPayloadCapacity(), CHECK, CHECK_GE, anonymous_namespace{StringDictionary.cpp}::checked_mmap(), anonymous_namespace{StringDictionary.cpp}::checked_munmap(), isTemp_, offset_fd_, offset_file_size_, offset_map_, payload_fd_, payload_file_off_, payload_file_size_, payload_map_, StringDictionary::StringIdxEntry::size, and str_count_.

Referenced by getOrAddBulk(), and getOrAddImpl().

                                                                    {
  if (!isTemp_) {
    CHECK_GE(payload_fd_, 0);
    CHECK_GE(offset_fd_, 0);
  }
  // write the payload
  if (payload_file_off_ + str.size() > payload_file_size_) {
    if (!isTemp_) {
      checked_munmap(payload_map_, payload_file_size_);
      addPayloadCapacity();
      CHECK(payload_file_off_ + str.size() <= payload_file_size_);
      payload_map_ =
          reinterpret_cast<char*>(checked_mmap(payload_fd_, payload_file_size_));
    } else {
      addPayloadCapacity();
    }
  }
  memcpy(payload_map_ + payload_file_off_, str.c_str(), str.size());
  // write the offset and length
  size_t offset_file_off = str_count_ * sizeof(StringIdxEntry);
  StringIdxEntry str_meta{static_cast<uint64_t>(payload_file_off_), str.size()};
  payload_file_off_ += str.size();
  if (offset_file_off + sizeof(str_meta) >= offset_file_size_) {
    if (!isTemp_) {
      checked_munmap(offset_map_, offset_file_size_);
      addOffsetCapacity();
      CHECK(offset_file_off + sizeof(str_meta) <= offset_file_size_);
      offset_map_ =
          reinterpret_cast<StringIdxEntry*>(checked_mmap(offset_fd_, offset_file_size_));
    } else {
      addOffsetCapacity();
    }
  }
  memcpy(offset_map_ + str_count_, &str_meta, sizeof(str_meta));
}

Here is the call graph for this function:

Here is the caller graph for this function:

binary_search_cache()

compare_cache_value_t* StringDictionary::binary_search_cache ( const std::string &  pattern ) const

private

buildSortedCache()

void StringDictionary::buildSortedCache ( )

private

Definition at line 1086 of file StringDictionary.cpp.

References mergeSortedCache(), sortCache(), sorted_cache, and str_count_.

Referenced by getCompare().

                                        {
  // This method is not thread-safe.
  const auto cur_cache_size = sorted_cache.size();
  std::vector<int32_t> temp_sorted_cache;
  for (size_t i = cur_cache_size; i < str_count_; i++) {
    temp_sorted_cache.push_back(i);
  }
  sortCache(temp_sorted_cache);
  mergeSortedCache(temp_sorted_cache);
}

Here is the call graph for this function:

Here is the caller graph for this function:

checkpoint()

bool StringDictionary::checkpoint ( )

noexcept

Definition at line 1069 of file StringDictionary.cpp.

References CHECK, client_, isTemp_, offset_fd_, offset_file_size_, offset_map_, payload_fd_, payload_file_size_, and payload_map_.

                                           {
  if (client_) {
    try {
      return client_->checkpoint();
    } catch (...) {
      return false;
    }
  }
  CHECK(!isTemp_);
  bool ret = true;
  ret = ret && (msync((void*)offset_map_, offset_file_size_, MS_SYNC) == 0);
  ret = ret && (msync((void*)payload_map_, payload_file_size_, MS_SYNC) == 0);
  ret = ret && (fsync(offset_fd_) == 0);
  ret = ret && (fsync(payload_fd_) == 0);
  return ret;
}

computeBucket()

uint32_t StringDictionary::computeBucket ( const uint32_t  hash, const std::string  str, const std::vector< int32_t > &  data, const bool  unique  ) const

privatenoexcept

Definition at line 901 of file StringDictionary.cpp.

References getStringFromStorage(), INVALID_STR_ID, materialize_hashes_, and rk_hashes_.

Referenced by getOrAddBulk(), getOrAddImpl(), and getUnlocked().

                                                                           {
  auto bucket = hash & (data.size() - 1);
  while (true) {
    if (data[bucket] ==
        INVALID_STR_ID) {  // In this case it means the slot is available for use
      break;
    }
    // if records are unique I don't need to do this test as I know it will not be the
    // same
    if (!unique) {
      if (materialize_hashes_) {
        if (hash == rk_hashes_[data[bucket]]) {
          // can't be the same string if hash is different
          const auto old_str = getStringFromStorage(data[bucket]);
          if (str.size() == old_str.size &&
              !memcmp(str.c_str(), old_str.c_str_ptr, str.size())) {
            // found the string
            break;
          }
        }
      } else {
        const auto old_str = getStringFromStorage(data[bucket]);
        if (str.size() == old_str.size &&
            !memcmp(str.c_str(), old_str.c_str_ptr, str.size())) {
          // found the string
          break;
        }
      }
    }
    // wrap around
    if (++bucket == data.size()) {
      bucket = 0;
    }
  }
  return bucket;
}

Here is the call graph for this function:

Here is the caller graph for this function:

computeUniqueBucketWithHash()

uint32_t StringDictionary::computeUniqueBucketWithHash ( const uint32_t  hash, const std::vector< int32_t > &  data  ) const

privatenoexcept

Definition at line 941 of file StringDictionary.cpp.

References INVALID_STR_ID.

Referenced by increaseCapacity(), and processDictionaryFutures().

                                                   {
  auto bucket = hash & (data.size() - 1);
  while (true) {
    if (data[bucket] ==
        INVALID_STR_ID) {  // In this case it means the slot is available for use
      break;
    }
    // wrap around
    if (++bucket == data.size()) {
      bucket = 0;
    }
  }
  return bucket;
}

Here is the caller graph for this function:

copyStrings()

std::shared_ptr< const std::vector< std::string > > StringDictionary::copyStrings

(

)

const

Definition at line 768 of file StringDictionary.cpp.

References CHECK_EQ, CHECK_GT, CHECK_LE, client_, cpu_threads(), getStringUnlocked(), rw_mutex_, str_count_, and strings_cache_.

                                                                              {
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);
  if (client_) {
    // TODO(miyu): support remote string dictionary
    throw std::runtime_error(
        "copying dictionaries from remote server is not supported yet.");
  }

  if (strings_cache_) {
    return strings_cache_;
  }

  strings_cache_ = std::make_shared<std::vector<std::string>>();
  strings_cache_->reserve(str_count_);
  const bool multithreaded = str_count_ > 10000;
  const auto worker_count =
      multithreaded ? static_cast<size_t>(cpu_threads()) : size_t(1);
  CHECK_GT(worker_count, 0UL);
  std::vector<std::vector<std::string>> worker_results(worker_count);
  auto copy = [this](std::vector<std::string>& str_list,
                     const size_t start_id,
                     const size_t end_id) {
    CHECK_LE(start_id, end_id);
    str_list.reserve(end_id - start_id);
    for (size_t string_id = start_id; string_id < end_id; ++string_id) {
      str_list.push_back(getStringUnlocked(string_id));
    }
  };
  if (multithreaded) {
    std::vector<std::future<void>> workers;
    const auto stride = (str_count_ + (worker_count - 1)) / worker_count;
    for (size_t worker_idx = 0, start = 0, end = std::min(start + stride, str_count_);
         worker_idx < worker_count && start < str_count_;
         ++worker_idx, start += stride, end = std::min(start + stride, str_count_)) {
      workers.push_back(std::async(
          std::launch::async, copy, std::ref(worker_results[worker_idx]), start, end));
    }
    for (auto& worker : workers) {
      worker.get();
    }
  } else {
    CHECK_EQ(worker_results.size(), size_t(1));
    copy(worker_results[0], 0, str_count_);
  }

  for (const auto& worker_result : worker_results) {
    strings_cache_->insert(
        strings_cache_->end(), worker_result.begin(), worker_result.end());
  }
  return strings_cache_;
}

Here is the call graph for this function:

fillRateIsHigh()

bool StringDictionary::fillRateIsHigh ( ) const

privatenoexcept

Definition at line 820 of file StringDictionary.cpp.

References str_count_, and str_ids_.

Referenced by getOrAddBulk(), and getOrAddImpl().

                                                     {
  return str_ids_.size() < str_count_ * 2;
}

Here is the caller graph for this function:

getCompare()

std::vector< int32_t > StringDictionary::getCompare	(	const std::string &	pattern,
		const std::string &	comp_operator,
		const size_t	generation
	)

Definition at line 560 of file StringDictionary.cpp.

References buildSortedCache(), client_, compare_cache_, getEquals(), getStringFromStorage(), rw_mutex_, sorted_cache, str_count_, string_eq(), and string_lt().

                                                                           {
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);
  if (client_) {
    return client_->get_compare(pattern, comp_operator, generation);
  }
  std::vector<int32_t> ret;
  if (str_count_ == 0) {
    return ret;
  }
  if (sorted_cache.size() < str_count_) {
    if (comp_operator == "=" || comp_operator == "<>") {
      return getEquals(pattern, comp_operator, generation);
    }

    buildSortedCache();
  }
  auto cache_index = compare_cache_.get(pattern);

  if (!cache_index) {
    cache_index = std::make_shared<StringDictionary::compare_cache_value_t>();
    const auto cache_itr = std::lower_bound(
        sorted_cache.begin(),
        sorted_cache.end(),
        pattern,
        [this](decltype(sorted_cache)::value_type const& a, decltype(pattern)& b) {
          auto a_str = this->getStringFromStorage(a);
          return string_lt(a_str.c_str_ptr, a_str.size, b.c_str(), b.size());
        });

    if (cache_itr == sorted_cache.end()) {
      cache_index->index = sorted_cache.size() - 1;
      cache_index->diff = 1;
    } else {
      const auto cache_str = getStringFromStorage(*cache_itr);
      if (!string_eq(
              cache_str.c_str_ptr, cache_str.size, pattern.c_str(), pattern.size())) {
        cache_index->index = cache_itr - sorted_cache.begin() - 1;
        cache_index->diff = 1;
      } else {
        cache_index->index = cache_itr - sorted_cache.begin();
        cache_index->diff = 0;
      }
    }

    compare_cache_.put(pattern, cache_index);
  }

  // since we have a cache in form of vector of ints which is sorted according to
  // corresponding strings in the dictionary all we need is the index of the element
  // which equal to the pattern that we are trying to match or the index of “biggest”
  // element smaller than the pattern, to perform all the comparison operators over
  // string. The search function guarantees we have such index so now it is just the
  // matter to include all the elements in the result vector.

  // For < operator if the index that we have points to the element which is equal to
  // the pattern that we are searching for we simply get all the elements less than the
  // index. If the element pointed by the index is not equal to the pattern we are
  // comparing with we also need to include that index in result vector, except when the
  // index points to 0 and the pattern is lesser than the smallest value in the string
  // dictionary.

  if (comp_operator == "<") {
    size_t idx = cache_index->index;
    if (cache_index->diff) {
      idx = cache_index->index + 1;
      if (cache_index->index == 0 && cache_index->diff > 0) {
        idx = cache_index->index;
      }
    }
    for (size_t i = 0; i < idx; i++) {
      ret.push_back(sorted_cache[i]);
    }

    // For <= operator if the index that we have points to the element which is equal to
    // the pattern that we are searching for we want to include the element pointed by
    // the index in the result set. If the element pointed by the index is not equal to
    // the pattern we are comparing with we just want to include all the ids with index
    // less than the index that is cached, except when pattern that we are searching for
    // is smaller than the smallest string in the dictionary.

  } else if (comp_operator == "<=") {
    size_t idx = cache_index->index + 1;
    if (cache_index == 0 && cache_index->diff > 0) {
      idx = cache_index->index;
    }
    for (size_t i = 0; i < idx; i++) {
      ret.push_back(sorted_cache[i]);
    }

    // For > operator we want to get all the elements with index greater than the index
    // that we have except, when the pattern we are searching for is lesser than the
    // smallest string in the dictionary we also want to include the id of the index
    // that we have.

  } else if (comp_operator == ">") {
    size_t idx = cache_index->index + 1;
    if (cache_index->index == 0 && cache_index->diff > 0) {
      idx = cache_index->index;
    }
    for (size_t i = idx; i < sorted_cache.size(); i++) {
      ret.push_back(sorted_cache[i]);
    }

    // For >= operator when the indexed element that we have points to element which is
    // equal to the pattern we are searching for we want to include that in the result
    // vector. If the index that we have does not point to the string which is equal to
    // the pattern we are searching we don’t want to include that id into the result
    // vector except when the index is 0.

  } else if (comp_operator == ">=") {
    size_t idx = cache_index->index;
    if (cache_index->diff) {
      idx = cache_index->index + 1;
      if (cache_index->index == 0 && cache_index->diff > 0) {
        idx = cache_index->index;
      }
    }
    for (size_t i = idx; i < sorted_cache.size(); i++) {
      ret.push_back(sorted_cache[i]);
    }
  } else if (comp_operator == "=") {
    if (!cache_index->diff) {
      ret.push_back(sorted_cache[cache_index->index]);
    }

    // For <> operator it is simple matter of not including id of string which is equal
    // to pattern we are searching for.
  } else if (comp_operator == "<>") {
    if (!cache_index->diff) {
      size_t idx = cache_index->index;
      for (size_t i = 0; i < idx; i++) {
        ret.push_back(sorted_cache[i]);
      }
      ++idx;
      for (size_t i = idx; i < sorted_cache.size(); i++) {
        ret.push_back(sorted_cache[i]);
      }
    } else {
      for (size_t i = 0; i < sorted_cache.size(); i++) {
        ret.insert(ret.begin(), sorted_cache.begin(), sorted_cache.end());
      }
    }

  } else {
    std::runtime_error("Unsupported string comparison operator");
  }
  return ret;
}

Here is the call graph for this function:

getEquals()

std::vector< int32_t > StringDictionary::getEquals ( std::string  pattern, std::string  comp_operator, size_t  generation  )

private

Definition at line 500 of file StringDictionary.cpp.

References CHECK, CHECK_GT, CHECK_LE, cpu_threads(), equal_cache_, getStringUnlocked(), MAX_STRLEN, run_benchmark_import::result, and str_count_.

Referenced by getCompare().

                                                                    {
  std::vector<int32_t> result;
  auto eq_id_itr = equal_cache_.find(pattern);
  int32_t eq_id = MAX_STRLEN + 1;
  int32_t cur_size = str_count_;
  if (eq_id_itr != equal_cache_.end()) {
    auto eq_id = eq_id_itr->second;
    if (comp_operator == "=") {
      result.push_back(eq_id);
    } else {
      for (int32_t idx = 0; idx <= cur_size; idx++) {
        if (idx == eq_id) {
          continue;
        }
        result.push_back(idx);
      }
    }
  } else {
    std::vector<std::thread> workers;
    int worker_count = cpu_threads();
    CHECK_GT(worker_count, 0);
    std::vector<std::vector<int32_t>> worker_results(worker_count);
    CHECK_LE(generation, str_count_);
    for (int worker_idx = 0; worker_idx < worker_count; ++worker_idx) {
      workers.emplace_back(
          [&worker_results, &pattern, generation, worker_idx, worker_count, this]() {
            for (size_t string_id = worker_idx; string_id < generation;
                 string_id += worker_count) {
              const auto str = getStringUnlocked(string_id);
              if (str == pattern) {
                worker_results[worker_idx].push_back(string_id);
              }
            }
          });
    }
    for (auto& worker : workers) {
      worker.join();
    }
    for (const auto& worker_result : worker_results) {
      result.insert(result.end(), worker_result.begin(), worker_result.end());
    }
    if (result.size() > 0) {
      const auto it_ok = equal_cache_.insert(std::make_pair(pattern, result[0]));
      CHECK(it_ok.second);
      eq_id = result[0];
    }
    if (comp_operator == "<>") {
      for (int32_t idx = 0; idx <= cur_size; idx++) {
        if (idx == eq_id) {
          continue;
        }
        result.push_back(idx);
      }
    }
  }
  return result;
}

Here is the call graph for this function:

Here is the caller graph for this function:

getIdOfString()

int32_t StringDictionary::getIdOfString

(

const std::string &

str

)

const

Definition at line 375 of file StringDictionary.cpp.

References client_, getUnlocked(), and rw_mutex_.

                                                                  {
  mapd_shared_lock<mapd_shared_mutex> read_lock(rw_mutex_);
  if (client_) {
    return client_->get(str);
  }
  return getUnlocked(str);
}

Here is the call graph for this function:

getLike()

std::vector< int32_t > StringDictionary::getLike	(	const std::string &	pattern,
		const bool	icase,
		const bool	is_simple,
		const char	escape,
		const size_t	generation
	)		const

Definition at line 447 of file StringDictionary.cpp.

References CHECK, CHECK_GT, CHECK_LE, client_, cpu_threads(), getStringUnlocked(), anonymous_namespace{StringDictionary.cpp}::is_like(), like_cache_, run_benchmark_import::result, rw_mutex_, and str_count_.

                                                                              {
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);
  if (client_) {
    return client_->get_like(pattern, icase, is_simple, escape, generation);
  }
  const auto cache_key = std::make_tuple(pattern, icase, is_simple, escape);
  const auto it = like_cache_.find(cache_key);
  if (it != like_cache_.end()) {
    return it->second;
  }
  std::vector<int32_t> result;
  std::vector<std::thread> workers;
  int worker_count = cpu_threads();
  CHECK_GT(worker_count, 0);
  std::vector<std::vector<int32_t>> worker_results(worker_count);
  CHECK_LE(generation, str_count_);
  for (int worker_idx = 0; worker_idx < worker_count; ++worker_idx) {
    workers.emplace_back([&worker_results,
                          &pattern,
                          generation,
                          icase,
                          is_simple,
                          escape,
                          worker_idx,
                          worker_count,
                          this]() {
      for (size_t string_id = worker_idx; string_id < generation;
           string_id += worker_count) {
        const auto str = getStringUnlocked(string_id);
        if (is_like(str, pattern, icase, is_simple, escape)) {
          worker_results[worker_idx].push_back(string_id);
        }
      }
    });
  }
  for (auto& worker : workers) {
    worker.join();
  }
  for (const auto& worker_result : worker_results) {
    result.insert(result.end(), worker_result.begin(), worker_result.end());
  }
  // place result into cache for reuse if similar query
  const auto it_ok = like_cache_.insert(std::make_pair(cache_key, result));

  CHECK(it_ok.second);

  return result;
}

Here is the call graph for this function:

getOrAdd()

int32_t StringDictionary::getOrAdd ( const std::string &  str )

noexcept

Definition at line 251 of file StringDictionary.cpp.

References CHECK_EQ, client_, and getOrAddImpl().

                                                                {
  if (client_) {
    std::vector<int32_t> string_ids;
    client_->get_or_add_bulk(string_ids, {str});
    CHECK_EQ(size_t(1), string_ids.size());
    return string_ids.front();
  }
  return getOrAddImpl(str);
}

Here is the call graph for this function:

getOrAddBulk()

template<class T >

template void StringDictionary::getOrAddBulk	(	const std::vector< std::string > &	string_vec,
		T *	encoded_vec
	)

Definition at line 285 of file StringDictionary.cpp.

References appendToStorage(), CHECK, CHECK_LT, client_no_timeout_, computeBucket(), fillRateIsHigh(), getOrAddBulkRemote(), increaseCapacity(), INVALID_STR_ID, invalidateInvertedIndex(), materialize_hashes_, MAX_STRCOUNT, MAX_STRLEN, offsets_path_, anonymous_namespace{StringDictionary.cpp}::rk_hash(), rk_hashes_, rw_mutex_, str_count_, and str_ids_.

Referenced by getOrAddBulkArray(), and populate_string_ids().

                                                    {
  if (client_no_timeout_) {
    getOrAddBulkRemote(string_vec, encoded_vec);
    return;
  }
  size_t out_idx{0};
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);

  for (const auto& str : string_vec) {
    if (str.empty()) {
      encoded_vec[out_idx++] = inline_int_null_value<T>();
      continue;
    }
    CHECK(str.size() <= MAX_STRLEN);
    uint32_t bucket;
    const uint32_t hash = rk_hash(str);
    bucket = computeBucket(hash, str, str_ids_, false);
    if (str_ids_[bucket] != INVALID_STR_ID) {
      encoded_vec[out_idx++] = str_ids_[bucket];
      continue;
    }
    // need to add record to dictionary
    // check there is room
    if (str_count_ == static_cast<size_t>(max_valid_int_value<T>())) {
      log_encoding_error<T>(str);
      encoded_vec[out_idx++] = inline_int_null_value<T>();
      continue;
    }
    if (str_ids_[bucket] == INVALID_STR_ID) {
      CHECK_LT(str_count_, MAX_STRCOUNT)
          << "Maximum number (" << str_count_
          << ") of Dictionary encoded Strings reached for this column, offset path "
             "for column is  "
          << offsets_path_;
      if (fillRateIsHigh()) {
        // resize when more than 50% is full
        increaseCapacity();
        bucket = computeBucket(hash, str, str_ids_, false);
      }
      appendToStorage(str);

      str_ids_[bucket] = static_cast<int32_t>(str_count_);
      if (materialize_hashes_) {
        rk_hashes_[str_count_] = hash;
      }
      ++str_count_;
    }
    encoded_vec[out_idx++] = str_ids_[bucket];
  }
  invalidateInvertedIndex();
}

Here is the call graph for this function:

Here is the caller graph for this function:

getOrAddBulkArray()

void StringDictionary::getOrAddBulkArray	(	const std::vector< std::vector< std::string >> &	string_array_vec,
		std::vector< std::vector< int32_t >> &	ids_array_vec
	)

Definition at line 272 of file StringDictionary.cpp.

References getOrAddBulk().

                                                  {
  ids_array_vec.resize(string_array_vec.size());
  for (size_t i = 0; i < string_array_vec.size(); i++) {
    auto& strings = string_array_vec[i];
    auto& ids = ids_array_vec[i];
    ids.resize(strings.size());
    getOrAddBulk(strings, &ids[0]);
  }
}

Here is the call graph for this function:

getOrAddBulkRemote()

template<class T >

template void StringDictionary::getOrAddBulkRemote ( const std::vector< std::string > &  string_vec, T *  encoded_vec  )

private

Definition at line 345 of file StringDictionary.cpp.

References CHECK, and client_no_timeout_.

Referenced by getOrAddBulk().

                                                          {
  CHECK(client_no_timeout_);
  std::vector<int32_t> string_ids;
  client_no_timeout_->get_or_add_bulk(string_ids, string_vec);
  size_t out_idx{0};
  for (size_t i = 0; i < string_ids.size(); ++i) {
    const auto string_id = string_ids[i];
    const bool invalid = string_id > max_valid_int_value<T>();
    if (invalid || string_id == inline_int_null_value<int32_t>()) {
      if (invalid) {
        log_encoding_error<T>(string_vec[i]);
      }
      encoded_vec[out_idx++] = inline_int_null_value<T>();
      continue;
    }
    encoded_vec[out_idx++] = string_id;
  }
}

Here is the caller graph for this function:

getOrAddImpl()

int32_t StringDictionary::getOrAddImpl ( const std::string &  str )

privatenoexcept

Definition at line 847 of file StringDictionary.cpp.

References appendToStorage(), CHECK, CHECK_LT, computeBucket(), fillRateIsHigh(), increaseCapacity(), INVALID_STR_ID, invalidateInvertedIndex(), materialize_hashes_, MAX_STRCOUNT, MAX_STRLEN, offsets_path_, anonymous_namespace{StringDictionary.cpp}::rk_hash(), rk_hashes_, rw_mutex_, str_count_, and str_ids_.

Referenced by getOrAdd().

                                                                    {
  // @TODO(wei) treat empty string as NULL for now
  if (str.size() == 0) {
    return inline_int_null_value<int32_t>();
  }
  CHECK(str.size() <= MAX_STRLEN);
  uint32_t bucket;
  const uint32_t hash = rk_hash(str);
  {
    mapd_shared_lock<mapd_shared_mutex> read_lock(rw_mutex_);
    bucket = computeBucket(hash, str, str_ids_, false);
    if (str_ids_[bucket] != INVALID_STR_ID) {
      return str_ids_[bucket];
    }
  }
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);
  // need to recalculate the bucket in case it changed before
  // we got the lock
  bucket = computeBucket(hash, str, str_ids_, false);
  if (str_ids_[bucket] == INVALID_STR_ID) {
    CHECK_LT(str_count_, MAX_STRCOUNT)
        << "Maximum number (" << str_count_
        << ") of Dictionary encoded Strings reached for this column, offset path "
           "for column is  "
        << offsets_path_;
    if (fillRateIsHigh()) {
      // resize when more than 50% is full
      increaseCapacity();
      bucket = computeBucket(hash, str, str_ids_, false);
    }
    appendToStorage(str);
    str_ids_[bucket] = static_cast<int32_t>(str_count_);
    if (materialize_hashes_) {
      rk_hashes_[str_count_] = hash;
    }
    ++str_count_;
    invalidateInvertedIndex();
  }
  return str_ids_[bucket];
}

Here is the call graph for this function:

Here is the caller graph for this function:

getRegexpLike()

std::vector< int32_t > StringDictionary::getRegexpLike	(	const std::string &	pattern,
		const char	escape,
		const size_t	generation
	)		const

Definition at line 721 of file StringDictionary.cpp.

References CHECK, CHECK_GT, CHECK_LE, client_, cpu_threads(), getStringUnlocked(), anonymous_namespace{StringDictionary.cpp}::is_regexp_like(), regex_cache_, run_benchmark_import::result, rw_mutex_, and str_count_.

                                                                                    {
  mapd_lock_guard<mapd_shared_mutex> write_lock(rw_mutex_);
  if (client_) {
    return client_->get_regexp_like(pattern, escape, generation);
  }
  const auto cache_key = std::make_pair(pattern, escape);
  const auto it = regex_cache_.find(cache_key);
  if (it != regex_cache_.end()) {
    return it->second;
  }
  std::vector<int32_t> result;
  std::vector<std::thread> workers;
  int worker_count = cpu_threads();
  CHECK_GT(worker_count, 0);
  std::vector<std::vector<int32_t>> worker_results(worker_count);
  CHECK_LE(generation, str_count_);
  for (int worker_idx = 0; worker_idx < worker_count; ++worker_idx) {
    workers.emplace_back([&worker_results,
                          &pattern,
                          generation,
                          escape,
                          worker_idx,
                          worker_count,
                          this]() {
      for (size_t string_id = worker_idx; string_id < generation;
           string_id += worker_count) {
        const auto str = getStringUnlocked(string_id);
        if (is_regexp_like(str, pattern, escape)) {
          worker_results[worker_idx].push_back(string_id);
        }
      }
    });
  }
  for (auto& worker : workers) {
    worker.join();
  }
  for (const auto& worker_result : worker_results) {
    result.insert(result.end(), worker_result.begin(), worker_result.end());
  }
  const auto it_ok = regex_cache_.insert(std::make_pair(cache_key, result));
  CHECK(it_ok.second);

  return result;
}

Here is the call graph for this function:

getString()

std::string StringDictionary::getString

(

int32_t

string_id

)

const

Definition at line 389 of file StringDictionary.cpp.

References client_, getStringUnlocked(), and rw_mutex_.

Referenced by StringValueConverter::convertToColumnarFormatFromDict(), and populate_string_ids().

                                                             {
  mapd_shared_lock<mapd_shared_mutex> read_lock(rw_mutex_);
  if (client_) {
    std::string ret;
    client_->get_string(ret, string_id);
    return ret;
  }
  return getStringUnlocked(string_id);
}

Here is the call graph for this function:

Here is the caller graph for this function:

getStringBytes()

std::pair< char *, size_t > StringDictionary::getStringBytes ( int32_t  string_id ) const

noexcept

Definition at line 404 of file StringDictionary.cpp.

References CHECK, CHECK_LE, CHECK_LT, client_, getStringBytesChecked(), rw_mutex_, and str_count_.

             {
  mapd_shared_lock<mapd_shared_mutex> read_lock(rw_mutex_);
  CHECK(!client_);
  CHECK_LE(0, string_id);
  CHECK_LT(string_id, static_cast<int32_t>(str_count_));
  return getStringBytesChecked(string_id);
}

Here is the call graph for this function:

getStringBytesChecked()

std::pair< char *, size_t > StringDictionary::getStringBytesChecked ( const int  string_id ) const

privatenoexcept

Definition at line 894 of file StringDictionary.cpp.

References CHECK, and getStringFromStorage().

Referenced by getStringBytes().

                                        {
  const auto str_canary = getStringFromStorage(string_id);
  CHECK(!str_canary.canary);
  return std::make_pair(str_canary.c_str_ptr, str_canary.size);
}

Here is the call graph for this function:

Here is the caller graph for this function:

getStringChecked()

std::string StringDictionary::getStringChecked ( const int  string_id ) const

privatenoexcept

Definition at line 888 of file StringDictionary.cpp.

References CHECK, and getStringFromStorage().

Referenced by getStringUnlocked(), and increaseCapacity().

                                                                               {
  const auto str_canary = getStringFromStorage(string_id);
  CHECK(!str_canary.canary);
  return std::string(str_canary.c_str_ptr, str_canary.size);
}

Here is the call graph for this function:

Here is the caller graph for this function:

getStringFromStorage()

StringDictionary::PayloadString StringDictionary::getStringFromStorage ( const int  string_id ) const

privatenoexcept

Definition at line 994 of file StringDictionary.cpp.

References CHECK_GE, isTemp_, StringDictionary::StringIdxEntry::off, offset_fd_, offset_map_, payload_fd_, payload_map_, and StringDictionary::StringIdxEntry::size.

Referenced by computeBucket(), getCompare(), getStringBytesChecked(), getStringChecked(), mergeSortedCache(), sortCache(), and StringDictionary().

                                        {
  if (!isTemp_) {
    CHECK_GE(payload_fd_, 0);
    CHECK_GE(offset_fd_, 0);
  }
  CHECK_GE(string_id, 0);
  const StringIdxEntry* str_meta = offset_map_ + string_id;
  if (str_meta->size == 0xffff) {
    // hit the canary
    return {nullptr, 0, true};
  }
  return {payload_map_ + str_meta->off, str_meta->size, false};
}

Here is the caller graph for this function:

getStringUnlocked()

std::string StringDictionary::getStringUnlocked ( int32_t  string_id ) const

privatenoexcept

Definition at line 399 of file StringDictionary.cpp.

References CHECK_LT, getStringChecked(), and str_count_.

Referenced by copyStrings(), getEquals(), getLike(), getRegexpLike(), and getString().

                                                                              {
  CHECK_LT(string_id, static_cast<int32_t>(str_count_));
  return getStringChecked(string_id);
}

Here is the call graph for this function:

Here is the caller graph for this function:

getUnlocked()

int32_t StringDictionary::getUnlocked ( const std::string &  str ) const

privatenoexcept

Definition at line 383 of file StringDictionary.cpp.

References computeBucket(), anonymous_namespace{StringDictionary.cpp}::rk_hash(), and str_ids_.

Referenced by getIdOfString().

                                                                         {
  const uint32_t hash = rk_hash(str);
  auto str_id = str_ids_[computeBucket(hash, str, str_ids_, false)];
  return str_id;
}

Here is the call graph for this function:

Here is the caller graph for this function:

increaseCapacity()

void StringDictionary::increaseCapacity ( )

privatenoexcept

Definition at line 824 of file StringDictionary.cpp.

References computeUniqueBucketWithHash(), getStringChecked(), INVALID_STR_ID, materialize_hashes_, anonymous_namespace{StringDictionary.cpp}::rk_hash(), rk_hashes_, str_count_, and str_ids_.

Referenced by getOrAddBulk(), and getOrAddImpl().

                                                 {
  std::vector<int32_t> new_str_ids(str_ids_.size() * 2, INVALID_STR_ID);

  if (materialize_hashes_) {
    for (size_t i = 0; i < str_ids_.size(); ++i) {
      if (str_ids_[i] != INVALID_STR_ID) {
        const uint32_t hash = rk_hashes_[str_ids_[i]];
        uint32_t bucket = computeUniqueBucketWithHash(hash, new_str_ids);
        new_str_ids[bucket] = str_ids_[i];
      }
    }
    rk_hashes_.resize(rk_hashes_.size() * 2);
  } else {
    for (size_t i = 0; i < str_count_; ++i) {
      const auto str = getStringChecked(i);
      const uint32_t hash = rk_hash(str);
      uint32_t bucket = computeUniqueBucketWithHash(hash, new_str_ids);
      new_str_ids[bucket] = i;
    }
  }
  str_ids_.swap(new_str_ids);
}

Here is the call graph for this function:

Here is the caller graph for this function:

insertInSortedCache()

void StringDictionary::insertInSortedCache ( std::string  str, int32_t  str_id  )

private

invalidateInvertedIndex()

void StringDictionary::invalidateInvertedIndex ( )

privatenoexcept

Definition at line 1054 of file StringDictionary.cpp.

References CANARY_BUFFER, compare_cache_, equal_cache_, like_cache_, and regex_cache_.

Referenced by getOrAddBulk(), and getOrAddImpl().

                                                        {
  if (!like_cache_.empty()) {
    decltype(like_cache_)().swap(like_cache_);
  }
  if (!regex_cache_.empty()) {
    decltype(regex_cache_)().swap(regex_cache_);
  }
  if (!equal_cache_.empty()) {
    decltype(equal_cache_)().swap(equal_cache_);
  }
  compare_cache_.invalidateInvertedIndex();
}

Here is the caller graph for this function:

mergeSortedCache()

void StringDictionary::mergeSortedCache ( std::vector< int32_t > &  temp_sorted_cache )

private

Definition at line 1110 of file StringDictionary.cpp.

References getStringFromStorage(), sorted_cache, and string_lt().

Referenced by buildSortedCache().

                                                                             {
  // this method is not thread safe
  std::vector<int32_t> updated_cache(temp_sorted_cache.size() + sorted_cache.size());
  size_t t_idx = 0, s_idx = 0, idx = 0;
  for (; t_idx < temp_sorted_cache.size() && s_idx < sorted_cache.size(); idx++) {
    auto t_string = getStringFromStorage(temp_sorted_cache[t_idx]);
    auto s_string = getStringFromStorage(sorted_cache[s_idx]);
    const auto insert_from_temp_cache =
        string_lt(t_string.c_str_ptr, t_string.size, s_string.c_str_ptr, s_string.size);
    if (insert_from_temp_cache) {
      updated_cache[idx] = temp_sorted_cache[t_idx++];
    } else {
      updated_cache[idx] = sorted_cache[s_idx++];
    }
  }
  while (t_idx < temp_sorted_cache.size()) {
    updated_cache[idx++] = temp_sorted_cache[t_idx++];
  }
  while (s_idx < sorted_cache.size()) {
    updated_cache[idx++] = sorted_cache[s_idx++];
  }
  sorted_cache.swap(updated_cache);
}

Here is the call graph for this function:

Here is the caller graph for this function:

populate_string_array_ids()

void StringDictionary::populate_string_array_ids ( std::vector< std::vector< int32_t >> &  dest_array_ids, StringDictionary *  dest_dict, const std::vector< std::vector< int32_t >> &  source_array_ids, const StringDictionary *  source_dict  )

static

Definition at line 1161 of file StringDictionary.cpp.

References populate_string_ids().

Referenced by DictionaryValueConverter< TARGET_TYPE >::processArrayBuffer().

                                         {
  dest_array_ids.resize(source_array_ids.size());

  std::atomic<size_t> row_idx{0};
  auto processor = [&row_idx, &dest_array_ids, dest_dict, &source_array_ids, source_dict](
                       int thread_id) {
    for (;;) {
      auto row = row_idx.fetch_add(1);

      if (row >= dest_array_ids.size()) {
        return;
      }
      const auto& source_ids = source_array_ids[row];
      auto& dest_ids = dest_array_ids[row];
      populate_string_ids(dest_ids, dest_dict, source_ids, source_dict);
    }
  };

  const int num_worker_threads = std::thread::hardware_concurrency();

  if (source_array_ids.size() / num_worker_threads > 10) {
    std::vector<std::future<void>> worker_threads;
    for (int i = 0; i < num_worker_threads; ++i) {
      worker_threads.push_back(std::async(std::launch::async, processor, i));
    }

    for (auto& child : worker_threads) {
      child.wait();
    }
    for (auto& child : worker_threads) {
      child.get();
    }
  } else {
    processor(0);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

populate_string_ids()

void StringDictionary::populate_string_ids ( std::vector< int32_t > &  dest_ids, StringDictionary *  dest_dict, const std::vector< int32_t > &  source_ids, const StringDictionary *  source_dict, const std::map< int32_t, std::string >  transient_mapping = {}  )

static

Populates provided dest_ids vector with string ids corresponding to given source strings.

Given a vector of source string ids and corresponding source dictionary, this method populates a vector of destination string ids by either returning the string id of matching strings in the destination dictionary or creating new entries in the dictionary. Source string ids can also be transient if they were created by a function (e.g LOWER/UPPER functions). A map of transient string ids to string values is provided in order to handle this use case.

Parameters

dest_ids	- vector of destination string ids to be populated
dest_dict	- destination dictionary
source_ids	- vector of source string ids for which destination ids are needed
source_dict	- source dictionary
transient_mapping	- map of transient source string ids to string values

Definition at line 1134 of file StringDictionary.cpp.

References getOrAddBulk(), and getString().

Referenced by populate_string_array_ids(), and DictionaryValueConverter< TARGET_TYPE >::processBuffer().

                                                        {
  std::vector<std::string> strings;

  for (const int32_t source_id : source_ids) {
    if (source_id == std::numeric_limits<int32_t>::min()) {
      strings.emplace_back("");
    } else if (source_id < 0) {
      if (auto string_itr = transient_mapping.find(source_id);
          string_itr != transient_mapping.end()) {
        strings.emplace_back(string_itr->second);
      } else {
        throw std::runtime_error("Unexpected negative source ID");
      }
    } else {
      strings.push_back(source_dict->getString(source_id));
    }
  }

  dest_ids.resize(strings.size());
  dest_dict->getOrAddBulk(strings, &dest_ids[0]);
}

Here is the call graph for this function:

Here is the caller graph for this function:

processDictionaryFutures()

void StringDictionary::processDictionaryFutures ( std::vector< std::future< std::vector< std::pair< uint32_t, unsigned int >>>> &  dictionary_futures )

private

Definition at line 206 of file StringDictionary.cpp.

References computeUniqueBucketWithHash(), materialize_hashes_, payload_file_off_, rk_hashes_, str_count_, and str_ids_.

Referenced by StringDictionary().

                            {
  for (auto& dictionary_future : dictionary_futures) {
    dictionary_future.wait();
    auto hashVec = dictionary_future.get();
    for (auto& hash : hashVec) {
      uint32_t bucket = computeUniqueBucketWithHash(hash.first, str_ids_);
      payload_file_off_ += hash.second;
      str_ids_[bucket] = static_cast<int32_t>(str_count_);
      if (materialize_hashes_) {
        rk_hashes_[str_count_] = hash.first;
      }
      ++str_count_;
    }
  }
  dictionary_futures.clear();
}

Here is the call graph for this function:

Here is the caller graph for this function:

sortCache()

void StringDictionary::sortCache ( std::vector< int32_t > &  cache )

private

Definition at line 1097 of file StringDictionary.cpp.

References getStringFromStorage(), and string_lt().

Referenced by buildSortedCache().

                                                          {
  // This method is not thread-safe.

  // this boost sort is creating some problems when we use UTF-8 encoded strings.
  // TODO (vraj): investigate What is wrong with boost sort and try to mitigate it.

  std::sort(cache.begin(), cache.end(), [this](int32_t a, int32_t b) {
    auto a_str = this->getStringFromStorage(a);
    auto b_str = this->getStringFromStorage(b);
    return string_lt(a_str.c_str_ptr, a_str.size, b_str.c_str_ptr, b_str.size);
  });
}

Here is the call graph for this function:

Here is the caller graph for this function:

storageEntryCount()

size_t StringDictionary::storageEntryCount

(

)

const

Definition at line 413 of file StringDictionary.cpp.

References client_, rw_mutex_, and str_count_.

                                                 {
  mapd_shared_lock<mapd_shared_mutex> read_lock(rw_mutex_);
  if (client_) {
    return client_->storage_entry_count();
  }
  return str_count_;
}

Member Data Documentation

CANARY_BUFFER

char * StringDictionary::CANARY_BUFFER {nullptr}

staticprivate

Definition at line 192 of file StringDictionary.h.

Referenced by addMemoryCapacity(), addStorageCapacity(), and invalidateInvertedIndex().

client_

std::unique_ptr<StringDictionaryClient> StringDictionary::client_

private

Definition at line 189 of file StringDictionary.h.

Referenced by checkpoint(), copyStrings(), getCompare(), getIdOfString(), getLike(), getOrAdd(), getRegexpLike(), getString(), getStringBytes(), storageEntryCount(), and ~StringDictionary().

client_no_timeout_

std::unique_ptr<StringDictionaryClient> StringDictionary::client_no_timeout_

private

Definition at line 190 of file StringDictionary.h.

Referenced by getOrAddBulk(), and getOrAddBulkRemote().

compare_cache_

DictionaryCache<std::string, compare_cache_value_t> StringDictionary::compare_cache_

mutableprivate

Definition at line 187 of file StringDictionary.h.

Referenced by getCompare(), and invalidateInvertedIndex().

equal_cache_

std::map<std::string, int32_t> StringDictionary::equal_cache_

mutableprivate

Definition at line 186 of file StringDictionary.h.

Referenced by getEquals(), and invalidateInvertedIndex().

INVALID_STR_ID

constexpr int32_t StringDictionary::INVALID_STR_ID = -1

static

Definition at line 109 of file StringDictionary.h.

Referenced by computeBucket(), computeUniqueBucketWithHash(), count_matches_impl(), count_matches_sharded(), CodeGenerator::createInValuesBitmap(), anonymous_namespace{RelAlgTranslator.cpp}::fill_dictionary_encoded_in_vals(), fill_hash_join_buff_impl(), fill_hash_join_buff_sharded_impl(), fill_row_ids_impl(), fill_row_ids_sharded_impl(), StringDictionaryProxy::getIdOfString(), StringDictionaryProxy::getIdOfStringNoGeneration(), getOrAddBulk(), getOrAddImpl(), StringDictionaryProxy::getOrAddTransient(), StringDictionaryProxy::getString(), increaseCapacity(), GenericKeyHandler::operator()(), anonymous_namespace{StringDictionary.cpp}::rk_hash(), StringDictionary(), anonymous_namespace{HashJoinRuntime.cpp}::translate_str_id_to_outer_dict(), and truncate_to_generation().

isTemp_

bool StringDictionary::isTemp_

private

Definition at line 172 of file StringDictionary.h.

Referenced by addOffsetCapacity(), addPayloadCapacity(), appendToStorage(), checkpoint(), getStringFromStorage(), StringDictionary(), and ~StringDictionary().

like_cache_

std::map<std::tuple<std::string, bool, bool, char>, std::vector<int32_t> > StringDictionary::like_cache_

mutableprivate

Definition at line 184 of file StringDictionary.h.

Referenced by getLike(), and invalidateInvertedIndex().

materialize_hashes_

bool StringDictionary::materialize_hashes_

private

Definition at line 173 of file StringDictionary.h.

Referenced by computeBucket(), getOrAddBulk(), getOrAddImpl(), increaseCapacity(), processDictionaryFutures(), and StringDictionary().

MAX_STRCOUNT

constexpr size_t StringDictionary::MAX_STRCOUNT = (1U << 31) - 1

static

Definition at line 111 of file StringDictionary.h.

Referenced by getOrAddBulk(), getOrAddImpl(), and anonymous_namespace{StringDictionary.cpp}::rk_hash().

MAX_STRLEN

constexpr size_t StringDictionary::MAX_STRLEN = (1 << 15) - 1

static

Definition at line 110 of file StringDictionary.h.

Referenced by Importer_NS::TypedImportBuffer::add_value(), Importer_NS::TypedImportBuffer::addDictEncodedString(), Importer_NS::TypedImportBuffer::addDictEncodedStringArray(), getEquals(), getOrAddBulk(), getOrAddImpl(), Importer_NS::DelimitedParserUtils::parseStringArray(), and anonymous_namespace{StringDictionary.cpp}::rk_hash().

offset_fd_

int StringDictionary::offset_fd_

private

Definition at line 176 of file StringDictionary.h.

Referenced by addOffsetCapacity(), appendToStorage(), checkpoint(), getStringFromStorage(), StringDictionary(), and ~StringDictionary().

offset_file_size_

size_t StringDictionary::offset_file_size_

private

Definition at line 179 of file StringDictionary.h.

Referenced by addOffsetCapacity(), appendToStorage(), checkpoint(), StringDictionary(), and ~StringDictionary().

offset_map_

StringIdxEntry* StringDictionary::offset_map_

private

Definition at line 177 of file StringDictionary.h.

Referenced by addOffsetCapacity(), appendToStorage(), checkpoint(), getStringFromStorage(), StringDictionary(), and ~StringDictionary().

offsets_path_

std::string StringDictionary::offsets_path_

private

Definition at line 174 of file StringDictionary.h.

Referenced by getOrAddBulk(), getOrAddImpl(), and StringDictionary().

payload_fd_

int StringDictionary::payload_fd_

private

Definition at line 175 of file StringDictionary.h.

Referenced by addPayloadCapacity(), appendToStorage(), checkpoint(), getStringFromStorage(), StringDictionary(), and ~StringDictionary().

payload_file_off_

size_t StringDictionary::payload_file_off_

private

Definition at line 181 of file StringDictionary.h.

Referenced by appendToStorage(), and processDictionaryFutures().

payload_file_size_

size_t StringDictionary::payload_file_size_

private

Definition at line 180 of file StringDictionary.h.

Referenced by addPayloadCapacity(), appendToStorage(), checkpoint(), StringDictionary(), and ~StringDictionary().

payload_map_

char* StringDictionary::payload_map_

private

Definition at line 178 of file StringDictionary.h.

Referenced by addPayloadCapacity(), appendToStorage(), checkpoint(), getStringFromStorage(), StringDictionary(), and ~StringDictionary().

regex_cache_

std::map<std::pair<std::string, char>, std::vector<int32_t> > StringDictionary::regex_cache_

mutableprivate

Definition at line 185 of file StringDictionary.h.

Referenced by getRegexpLike(), and invalidateInvertedIndex().

rk_hashes_

std::vector<uint32_t> StringDictionary::rk_hashes_

private

Definition at line 170 of file StringDictionary.h.

Referenced by computeBucket(), getOrAddBulk(), getOrAddImpl(), increaseCapacity(), processDictionaryFutures(), and StringDictionary().

rw_mutex_

mapd_shared_mutex StringDictionary::rw_mutex_

mutableprivate

Definition at line 182 of file StringDictionary.h.

Referenced by copyStrings(), getCompare(), getIdOfString(), getLike(), getOrAddBulk(), getOrAddImpl(), getRegexpLike(), getString(), getStringBytes(), storageEntryCount(), and StringDictionary().

sorted_cache

std::vector<int32_t> StringDictionary::sorted_cache

private

Definition at line 171 of file StringDictionary.h.

Referenced by buildSortedCache(), getCompare(), and mergeSortedCache().

str_count_

size_t StringDictionary::str_count_

private

Definition at line 168 of file StringDictionary.h.

Referenced by appendToStorage(), buildSortedCache(), copyStrings(), fillRateIsHigh(), getCompare(), getEquals(), getLike(), getOrAddBulk(), getOrAddImpl(), getRegexpLike(), getStringBytes(), getStringUnlocked(), increaseCapacity(), processDictionaryFutures(), and storageEntryCount().

str_ids_

std::vector<int32_t> StringDictionary::str_ids_

private

Definition at line 169 of file StringDictionary.h.

Referenced by fillRateIsHigh(), getOrAddBulk(), getOrAddImpl(), getUnlocked(), increaseCapacity(), processDictionaryFutures(), and StringDictionary().

strings_cache_

std::shared_ptr<std::vector<std::string> > StringDictionary::strings_cache_

mutableprivate

Definition at line 188 of file StringDictionary.h.

Referenced by copyStrings().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/StringDictionary/StringDictionary.h
• /home/jenkins-slave/workspace/core-os-doxygen/StringDictionary/StringDictionary.cpp