content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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.
   */
  package org.apache.asterix.external.input.record.reader.stream;
  
  import static org.apache.asterix.external.util.ExternalDataConstants.BYTE_ORDER_MARK;
  import static org.apache.asterix.external.util.ExternalDataConstants.CLOSING_BRACKET;
  import static org.apache.asterix.external.util.ExternalDataConstants.COMMA;
  import static org.apache.asterix.external.util.ExternalDataConstants.CR;
  import static org.apache.asterix.external.util.ExternalDataConstants.KEY_RECORD_END;
  import static org.apache.asterix.external.util.ExternalDataConstants.LF;
  import static org.apache.asterix.external.util.ExternalDataConstants.OPEN_BRACKET;
  import static org.apache.asterix.external.util.ExternalDataConstants.REC_ENDED_AT_EOF;
  import static org.apache.asterix.external.util.ExternalDataConstants.SPACE;
  import static org.apache.asterix.external.util.ExternalDataConstants.TAB;
  
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.function.LongSupplier;
  
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.common.exceptions.RuntimeDataException;
  import org.apache.asterix.external.api.AsterixInputStream;
  import org.apache.asterix.external.util.ExternalDataConstants;
  import org.apache.asterix.external.util.ExternalDataUtils;
  import org.apache.hyracks.api.context.IHyracksTaskContext;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.api.exceptions.IWarningCollector;
  import org.apache.hyracks.util.ParseUtil;
  
  public class SemiStructuredRecordReader extends StreamRecordReader {
  
      private enum State {
          TOP_LEVEL, // valid chars at this state: '{' or '[' to start a new record or array of records
          ARRAY, // valid chars at this state: '{' or ']' to start the first nested record or close the array
          NESTED_OBJECT, // valid chars at this state: ',' or ']' to close the array or expect another nested record
          AFTER_COMMA // valid chars at this state: '{' to start a new nested record
      }
  
      private IWarningCollector warnings;
      private int depth;
      private boolean prevCharEscape;
      private boolean inString;
      private char recordStart;
      private char recordEnd;
      private boolean hasStarted;
      private boolean hasFinished;
      private boolean isLastCharCR;
      private State state = State.TOP_LEVEL;
      private long beginLineNumber = 1;
      private long lineNumber = 1;
  
      private static final List<String> recordReaderFormats = Collections.unmodifiableList(
              Arrays.asList(ExternalDataConstants.FORMAT_ADM, ExternalDataConstants.FORMAT_JSON_LOWER_CASE,
                      ExternalDataConstants.FORMAT_JSON_UPPER_CASE, ExternalDataConstants.FORMAT_SEMISTRUCTURED));
      private static final String REQUIRED_CONFIGS = "";
  
      @Override
      public void configure(IHyracksTaskContext ctx, AsterixInputStream stream, Map<String, String> config)
              throws HyracksDataException {
          super.configure(stream, config);
          stream.setNotificationHandler(this);
          warnings = ctx.getWarningCollector();
          // set record opening char
          recordStart = ExternalDataUtils.validateGetRecordStart(config);
          // set record ending char
          recordEnd = ExternalDataUtils.validateGetRecordEnd(config);
          if (recordStart == recordEnd) {
              throw new RuntimeDataException(ErrorCode.INVALID_REQ_PARAM_VAL, KEY_RECORD_END, recordEnd);
          }
      }
  
      @Override
      public void notifyNewSource() {
          if (hasStarted && warnings.shouldWarn()) {
              ParseUtil.warn(warnings, getPreviousStreamName(), lineNumber, 0, REC_ENDED_AT_EOF);
          }
          beginLineNumber = 1;
          lineNumber = 1;
          state = State.TOP_LEVEL;
          resetForNewRecord();
      }
  
      @Override
      public LongSupplier getLineNumber() {
          return this::getBeginLineNumber;
      }
  
      private long getBeginLineNumber() {
          return beginLineNumber;
      }
  
      @Override
      public boolean hasNext() throws IOException {
          if (done) {
              return false;
          }
          resetForNewRecord();
          beginLineNumber = lineNumber;
          do {
              int startPosn = bufferPosn; // starting from where we left off the last time
              if (bufferPosn >= bufferLength) {
                  startPosn = bufferPosn = 0;
                  bufferLength = reader.read(inputBuffer);
                  if (bufferLength < 0) {
                      if (hasStarted && warnings.shouldWarn()) {
                          ParseUtil.warn(warnings, getDataSourceName().get(), lineNumber, 0, REC_ENDED_AT_EOF);
                      }
                      close();
                      return false; // EOF
                  }
              }
              if (!hasStarted) {
                  for (; bufferPosn < bufferLength; ++bufferPosn) { // search for record begin
                      char c = inputBuffer[bufferPosn];
                      if (c == LF || isLastCharCR) {
                          lineNumber++;
                      }
                      isLastCharCR = c == CR;
                      if (c == SPACE || c == TAB || c == LF || c == CR || c == BYTE_ORDER_MARK) {
                          continue;
                      }
                      if (c == recordStart && state != State.NESTED_OBJECT) {
                          // '{' is allowed at the top level, after '[' and after ','
                          if (state == State.ARRAY || state == State.AFTER_COMMA) {
                              state = State.NESTED_OBJECT;
                          }
                          beginLineNumber = lineNumber;
                          startPosn = bufferPosn;
                          hasStarted = true;
                          depth = 1;
                          ++bufferPosn;
                          break;
                      } else if (c == OPEN_BRACKET && state == State.TOP_LEVEL) {
                          // '[' is allowed at the top level only
                          state = State.ARRAY;
                      } else if (c == CLOSING_BRACKET && (state == State.ARRAY || state == State.NESTED_OBJECT)) {
                          // ']' is allowed after '[' and after capturing a record in an array
                          state = State.TOP_LEVEL;
                      } else if (c == COMMA && state == State.NESTED_OBJECT) {
                          // ',' is allowed after capturing a record in an array
                          state = State.AFTER_COMMA;
                      } else {
                          // corrupted file. clear the buffer and stop reading
                          reader.reset();
                          bufferPosn = bufferLength = 0;
                          throw new RuntimeDataException(ErrorCode.RECORD_READER_MALFORMED_INPUT_STREAM);
                      }
                  }
              }
              if (hasStarted) {
                  for (; bufferPosn < bufferLength; ++bufferPosn) {
                      char c = inputBuffer[bufferPosn];
                      if (c == LF || isLastCharCR) {
                          lineNumber++;
                      }
                      if (inString) {
                          // we are in a string, we only care about the string end
                          if (c == ExternalDataConstants.QUOTE && !prevCharEscape) {
                              inString = false;
                          }
                          prevCharEscape = c == ExternalDataConstants.ESCAPE && !prevCharEscape;
                      } else {
                          if (c == ExternalDataConstants.QUOTE) {
                              inString = true;
                          } else if (c == recordStart) {
                              depth += 1;
                          } else if (c == recordEnd) {
                              depth -= 1;
                              if (depth == 0) {
                                  hasFinished = true;
                                  ++bufferPosn; // at next invocation proceed from following byte
                                  break;
                              }
                          }
                      }
                      isLastCharCR = c == CR;
                  }
  
                  int appendLength = bufferPosn - startPosn;
                  if (appendLength > 0) {
                      try {
                          record.append(inputBuffer, startPosn, appendLength);
                      } catch (RuntimeDataException e) {
                          reader.reset();
                          bufferPosn = bufferLength = 0;
                          throw e;
                      }
                  }
              }
          } while (!hasFinished);
          record.endRecord();
          return true;
      }
  
      @Override
      public List<String> getRecordReaderFormats() {
          return recordReaderFormats;
      }
  
      @Override
      public String getRequiredConfigs() {
          return REQUIRED_CONFIGS;
      }
  
      private void resetForNewRecord() {
          record.reset();
          hasStarted = false;
          hasFinished = false;
          prevCharEscape = false;
          isLastCharCR = false;
          inString = false;
          depth = 0;
      }
  }
  