delimited-text-parser.inline.h

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.


#ifndef IMPALA_EXEC_DELIMITED_TEXT_PARSER_INLINE_H
#define IMPALA_EXEC_DELIMITED_TEXT_PARSER_INLINE_H

#include "delimited-text-parser.h"
#include "util/cpu-info.h"
#include "util/sse-util.h"

namespace impala {

inline void ProcessEscapeMask(uint16_t escape_mask, bool* last_char_is_escape,
                              uint16_t* delim_mask) {
  // Escape characters can escape escape characters.
  bool first_char_is_escape = *last_char_is_escape;
  bool escape_next = first_char_is_escape;
  for (int i = 0; i < SSEUtil::CHARS_PER_128_BIT_REGISTER; ++i) {
    if (escape_next) {
      escape_mask &= ~SSEUtil::SSE_BITMASK[i];
    }
    escape_next = escape_mask & SSEUtil::SSE_BITMASK[i];
  }

  // Remember last character for the next iteration
  *last_char_is_escape = escape_mask &
    SSEUtil::SSE_BITMASK[SSEUtil::CHARS_PER_128_BIT_REGISTER - 1];

  // Shift escape mask up one so they match at the same bit index as the tuple and
  // field mask (instead of being the character before) and set the correct first bit
  escape_mask = escape_mask << 1 | (first_char_is_escape ? 1 : 0);

  // If escape_mask[n] is true, then tuple/field_mask[n] is escaped
  *delim_mask &= ~escape_mask;
}

template <bool process_escapes>
inline void DelimitedTextParser::AddColumn(int len, char** next_column_start,
    int* num_fields, FieldLocation* field_locations) {
  if (ReturnCurrentColumn()) {
    // Found a column that needs to be parsed, write the start/len to 'field_locations'
    field_locations[*num_fields].start = *next_column_start;
    if (process_escapes && current_column_has_escape_) {
      field_locations[*num_fields].len = -len;
    } else {
      field_locations[*num_fields].len = len;
    }
    ++(*num_fields);
  }
  if (process_escapes) current_column_has_escape_ = false;
  *next_column_start += len + 1;
  ++column_idx_;
}

template <bool process_escapes>
void inline DelimitedTextParser:: FillColumns(int len, char** last_column,
    int* num_fields, FieldLocation* field_locations) {
  // Fill in any columns missing from the end of the tuple.
  char* dummy = NULL;
  if (last_column == NULL) last_column = &dummy;
  while (column_idx_ < num_cols_) {
    AddColumn<process_escapes>(len, last_column, num_fields, field_locations);
    // The rest of the columns will be null.
    last_column = &dummy;
    len = 0;
  }
}

//
template <bool process_escapes>
inline void DelimitedTextParser::ParseSse(int max_tuples,
    int64_t* remaining_len, char** byte_buffer_ptr,
    char** row_end_locations,
    FieldLocation* field_locations,
    int* num_tuples, int* num_fields, char** next_column_start) {
  DCHECK(CpuInfo::IsSupported(CpuInfo::SSE4_2));

  // To parse using SSE, we:
  //  1. Load into different sse registers the different characters we need to search for
  //        tuple breaks, field breaks, escape characters
  //  2. Load 16 characters at a time into the sse register
  //  3. Use the SSE instruction to do strchr on those 16 chars, the result is a bitmask
  //  4. Compute the bitmask for tuple breaks, field breaks and escape characters.
  //  5. If there are escape characters, fix up the matching masked bits in the
  //        field/tuple mask
  //  6. Go through the mask bit by bit and write the parsed data.

  // xmm registers:
  //  - xmm_buffer: the register holding the current (16 chars) we're working on from the
  //        file
  //  - xmm_delim_search_: the delim search register.  Contains field delimiter,
  //        collection_item delim_char and tuple delimiter
  //  - xmm_escape_search_: the escape search register. Only contains escape char
  //  - xmm_delim_mask: the result of doing strchr for the delimiters
  //  - xmm_escape_mask: the result of doing strchr for the escape char
  __m128i xmm_buffer, xmm_delim_mask, xmm_escape_mask;

  while (LIKELY(*remaining_len >= SSEUtil::CHARS_PER_128_BIT_REGISTER)) {
    // Load the next 16 bytes into the xmm register
    xmm_buffer = _mm_loadu_si128(reinterpret_cast<__m128i*>(*byte_buffer_ptr));

    // Do the strchr for tuple and field breaks
    // The strchr sse instruction returns the result in the lower bits of the sse
    // register.  Since we only process 16 characters at a time, only the lower 16 bits
    // can contain non-zero values.
    // _mm_extract_epi16 will extract 16 bits out of the xmm register.  The second
    // parameter specifies which 16 bits to extract (0 for the lowest 16 bits).
    xmm_delim_mask = SSE4_cmpestrm(xmm_delim_search_, num_delims_, xmm_buffer,
        SSEUtil::CHARS_PER_128_BIT_REGISTER, SSEUtil::STRCHR_MODE);
    uint16_t delim_mask = _mm_extract_epi16(xmm_delim_mask, 0);

    uint16_t escape_mask = 0;
    // If the table does not use escape characters, skip processing for it.
    if (process_escapes) {
      DCHECK(escape_char_ != '\0');
      xmm_escape_mask = SSE4_cmpestrm(xmm_escape_search_, 1,
          xmm_buffer, SSEUtil::CHARS_PER_128_BIT_REGISTER, SSEUtil::STRCHR_MODE);
      escape_mask = _mm_extract_epi16(xmm_escape_mask, 0);
      ProcessEscapeMask(escape_mask, &last_char_is_escape_, &delim_mask);
    }

    char* last_char = *byte_buffer_ptr + 15;
    bool last_char_is_unescaped_delim = delim_mask >> 15;
    unfinished_tuple_ = !(last_char_is_unescaped_delim &&
        (*last_char == tuple_delim_ || (tuple_delim_ == '\n' && *last_char == '\r')));

    int last_col_idx = 0;
    // Process all non-zero bits in the delim_mask from lsb->msb.  If a bit
    // is set, the character in that spot is either a field or tuple delimiter.
    while (delim_mask != 0) {
      // ffs is a libc function that returns the index of the first set bit (1-indexed)
      int n = ffs(delim_mask) - 1;
      DCHECK_GE(n, 0);
      DCHECK_LT(n, 16);
      // clear current bit
      delim_mask &= ~(SSEUtil::SSE_BITMASK[n]);

      if (process_escapes) {
        // Determine if there was an escape character between [last_col_idx, n]
        bool escaped = (escape_mask & low_mask_[last_col_idx] & high_mask_[n]) != 0;
        current_column_has_escape_ |= escaped;
        last_col_idx = n;
      }

      char* delim_ptr = *byte_buffer_ptr + n;

      if (*delim_ptr == field_delim_ || *delim_ptr == collection_item_delim_) {
        AddColumn<process_escapes>(delim_ptr - *next_column_start,
            next_column_start, num_fields, field_locations);
        continue;
      }

      if (*delim_ptr == tuple_delim_ || (tuple_delim_ == '\n' && *delim_ptr == '\r')) {
        if (UNLIKELY(
                last_row_delim_offset_ == *remaining_len - n && *delim_ptr == '\n')) {
          // If the row ended in \r\n then move the next start past the \n
          ++*next_column_start;
          last_row_delim_offset_ = -1;
          continue;
        }
        AddColumn<process_escapes>(delim_ptr - *next_column_start,
            next_column_start, num_fields, field_locations);
        FillColumns<false>(0, NULL, num_fields, field_locations);
        column_idx_ = num_partition_keys_;
        row_end_locations[*num_tuples] = delim_ptr;
        ++(*num_tuples);
        // Remember where we saw the last \r.
        last_row_delim_offset_ = *delim_ptr == '\r' ? *remaining_len - n - 1 : -1;
        if (UNLIKELY(*num_tuples == max_tuples)) {
          (*byte_buffer_ptr) += (n + 1);
          if (process_escapes) last_char_is_escape_ = false;
          *remaining_len -= (n + 1);
          // If the last character we processed was \r then set the offset to 0
          // so that we will use it at the beginning of the next batch.
          if (last_row_delim_offset_ == *remaining_len) last_row_delim_offset_ = 0;
          return;
        }
      }
    }

    if (process_escapes) {
      // Determine if there was an escape character between (last_col_idx, 15)
      bool unprocessed_escape = escape_mask & low_mask_[last_col_idx] & high_mask_[15];
      current_column_has_escape_ |= unprocessed_escape;
    }

    *remaining_len -= SSEUtil::CHARS_PER_128_BIT_REGISTER;
    *byte_buffer_ptr += SSEUtil::CHARS_PER_128_BIT_REGISTER;
  }
}

template <bool process_escapes>
inline void DelimitedTextParser::ParseSingleTuple(int64_t remaining_len, char* buffer,
    FieldLocation* field_locations, int* num_fields) {
  char* next_column_start = buffer;
  __m128i xmm_buffer, xmm_delim_mask, xmm_escape_mask;

  column_idx_ = num_partition_keys_;
  current_column_has_escape_ = false;
  if (LIKELY(CpuInfo::IsSupported(CpuInfo::SSE4_2))) {
    while (LIKELY(remaining_len >= SSEUtil::CHARS_PER_128_BIT_REGISTER)) {
      // Load the next 16 bytes into the xmm register
      xmm_buffer = _mm_loadu_si128(reinterpret_cast<__m128i*>(buffer));

      xmm_delim_mask = SSE4_cmpestrm(xmm_delim_search_, num_delims_, xmm_buffer,
          SSEUtil::CHARS_PER_128_BIT_REGISTER, SSEUtil::STRCHR_MODE);
      uint16_t delim_mask = _mm_extract_epi16(xmm_delim_mask, 0);

      uint16_t escape_mask = 0;
      // If the table does not use escape characters, skip processing for it.
      if (process_escapes) {
        DCHECK(escape_char_ != '\0');
        xmm_escape_mask = SSE4_cmpestrm(xmm_escape_search_, 1, xmm_buffer,
            SSEUtil::CHARS_PER_128_BIT_REGISTER, SSEUtil::STRCHR_MODE);
        escape_mask = _mm_extract_epi16(xmm_escape_mask, 0);
        ProcessEscapeMask(escape_mask, &last_char_is_escape_, &delim_mask);
      }

      int last_col_idx = 0;
      // Process all non-zero bits in the delim_mask from lsb->msb.  If a bit
      // is set, the character in that spot is a field.
      while (delim_mask != 0) {
        // ffs is a libc function that returns the index of the first set bit (1-indexed)
        int n = ffs(delim_mask) - 1;
        DCHECK_GE(n, 0);
        DCHECK_LT(n, 16);

        if (process_escapes) {
          // Determine if there was an escape character between [last_col_idx, n]
          bool escaped = (escape_mask & low_mask_[last_col_idx] & high_mask_[n]) != 0;
          current_column_has_escape_ |= escaped;
          last_col_idx = n;
        }

        // clear current bit
        delim_mask &= ~(SSEUtil::SSE_BITMASK[n]);

        AddColumn<process_escapes>(buffer + n - next_column_start,
            &next_column_start, num_fields, field_locations);
      }

      if (process_escapes) {
        // Determine if there was an escape character between (last_col_idx, 15)
        bool unprocessed_escape = escape_mask & low_mask_[last_col_idx] & high_mask_[15];
        current_column_has_escape_ |= unprocessed_escape;
      }

      remaining_len -= SSEUtil::CHARS_PER_128_BIT_REGISTER;
      buffer += SSEUtil::CHARS_PER_128_BIT_REGISTER;
    }
  }

  while (remaining_len > 0) {
    if (*buffer == escape_char_) {
      current_column_has_escape_ = true;
      last_char_is_escape_ = !last_char_is_escape_;
    } else {
      last_char_is_escape_ = false;
    }

    if (!last_char_is_escape_ &&
          (*buffer == field_delim_ || *buffer == collection_item_delim_)) {
      AddColumn<process_escapes>(buffer - next_column_start,
          &next_column_start, num_fields, field_locations);
    }

    --remaining_len;
    ++buffer;
  }

  // Last column does not have a delimiter after it.  Add that column and also
  // pad with empty cols if the input is ragged.
  FillColumns<process_escapes>(buffer - next_column_start,
        &next_column_start, num_fields, field_locations);
}

}

#endif