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.parser;
  
  import java.io.DataOutput;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.InputStreamReader;
  import java.util.BitSet;
  import java.util.List;
  
  import org.apache.asterix.builders.AbvsBuilderFactory;
  import org.apache.asterix.builders.IARecordBuilder;
  import org.apache.asterix.builders.IAsterixListBuilder;
  import org.apache.asterix.builders.ListBuilderFactory;
  import org.apache.asterix.builders.OrderedListBuilder;
  import org.apache.asterix.builders.RecordBuilderFactory;
  import org.apache.asterix.builders.UnorderedListBuilder;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.dataflow.data.nontagged.serde.APolygonSerializerDeserializer;
  import org.apache.asterix.external.api.IRawRecord;
  import org.apache.asterix.external.api.IRecordDataParser;
  import org.apache.asterix.external.api.IStreamDataParser;
  import org.apache.asterix.om.base.ABoolean;
  import org.apache.asterix.om.base.ANull;
  import org.apache.asterix.om.base.temporal.GregorianCalendarSystem;
  import org.apache.asterix.om.types.AOrderedListType;
  import org.apache.asterix.om.types.ARecordType;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.AUnionType;
  import org.apache.asterix.om.types.AUnorderedListType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
  import org.apache.asterix.om.types.hierachy.ITypeConvertComputer;
  import org.apache.asterix.om.util.container.IObjectPool;
  import org.apache.asterix.om.util.container.ListObjectPool;
  import org.apache.asterix.runtime.operators.file.adm.AdmLexer;
  import org.apache.asterix.runtime.operators.file.adm.AdmLexer.TokenImage;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.data.std.api.IMutableValueStorage;
  import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
  import org.apache.hyracks.util.LogRedactionUtil;
  
  /**
   * Parser for ADM formatted data.
   */
  public class ADMDataParser extends AbstractDataParser implements IStreamDataParser, IRecordDataParser<char[]> {
      private AdmLexer admLexer;
      private final ARecordType recordType;
  
      private final ArrayBackedValueStorage castBuffer = new ArrayBackedValueStorage();
  
      private final IObjectPool<IARecordBuilder, ATypeTag> recordBuilderPool =
              new ListObjectPool<IARecordBuilder, ATypeTag>(new RecordBuilderFactory());
      private final IObjectPool<IAsterixListBuilder, ATypeTag> listBuilderPool =
              new ListObjectPool<IAsterixListBuilder, ATypeTag>(new ListBuilderFactory());
      private final IObjectPool<IMutableValueStorage, ATypeTag> abvsBuilderPool =
              new ListObjectPool<IMutableValueStorage, ATypeTag>(new AbvsBuilderFactory());
  
      private final TokenImage tmpTokenImage = new TokenImage();
  
      private static final String mismatchErrorMessage = "Mismatch Type, expecting a value of type ";
      private static final String mismatchErrorMessage2 = " got a value of type ";
  
      public ADMDataParser(ARecordType recordType, boolean isStream) {
          this(null, recordType, isStream);
      }
  
      public ADMDataParser(String filename, ARecordType recordType, boolean isStream) {
          this.filename = filename;
          this.recordType = recordType;
          if (!isStream) {
              this.admLexer = new AdmLexer();
          }
      }
  
      @Override
      public boolean parse(DataOutput out) throws HyracksDataException {
          try {
              resetPools();
              return parseAdmInstance(recordType, out);
          } catch (ParseException e) {
              e.setLocation(filename, admLexer.getLine(), admLexer.getColumn());
              throw e;
          } catch (IOException e) {
              throw new ParseException(e, filename, admLexer.getLine(), admLexer.getColumn());
          }
      }
  
      @Override
      public boolean parse(IRawRecord<? extends char[]> record, DataOutput out) throws HyracksDataException {
          try {
              resetPools();
              admLexer.setBuffer(record.get());
              return parseAdmInstance(recordType, out);
          } catch (ParseException e) {
              e.setLocation(filename, admLexer.getLine(), admLexer.getColumn());
              throw e;
          } catch (IOException e) {
              throw new ParseException(e, filename, admLexer.getLine(), admLexer.getColumn());
          }
      }
  
      @Override
      public void setInputStream(InputStream in) throws IOException {
          admLexer = new AdmLexer(new java.io.InputStreamReader(in));
      }
  
      private boolean parseAdmInstance(IAType objectType, DataOutput out) throws IOException {
          int token = admLexer.next();
          if (token == AdmLexer.TOKEN_EOF) {
              return false;
          } else {
              admFromLexerStream(token, objectType, out);
              return true;
          }
      }
  
      private void admFromLexerStream(int token, IAType objectType, DataOutput out) throws IOException {
  
          switch (token) {
              case AdmLexer.TOKEN_NULL_LITERAL:
                  if (checkType(ATypeTag.NULL, objectType)) {
                      nullSerde.serialize(ANull.NULL, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_FIELD_NOT_NULL, "");
                  }
                  break;
              case AdmLexer.TOKEN_TRUE_LITERAL:
                  if (checkType(ATypeTag.BOOLEAN, objectType)) {
                      booleanSerde.serialize(ABoolean.TRUE, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_BOOLEAN_CONS:
                  parseConstructor(ATypeTag.BOOLEAN, objectType, out);
                  break;
              case AdmLexer.TOKEN_FALSE_LITERAL:
                  if (checkType(ATypeTag.BOOLEAN, objectType)) {
                      booleanSerde.serialize(ABoolean.FALSE, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_DOUBLE_LITERAL:
                  parseToNumericTarget(ATypeTag.DOUBLE, objectType, out);
                  break;
              case AdmLexer.TOKEN_DOUBLE_CONS:
                  parseConstructor(ATypeTag.DOUBLE, objectType, out);
                  break;
              case AdmLexer.TOKEN_FLOAT_LITERAL:
                  parseToNumericTarget(ATypeTag.FLOAT, objectType, out);
                  break;
              case AdmLexer.TOKEN_FLOAT_CONS:
                  parseConstructor(ATypeTag.FLOAT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT8_LITERAL:
                  parseAndCastNumeric(ATypeTag.TINYINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT8_CONS:
                  parseConstructor(ATypeTag.TINYINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT16_LITERAL:
                  parseAndCastNumeric(ATypeTag.SMALLINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT16_CONS:
                  parseConstructor(ATypeTag.SMALLINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT_LITERAL:
                  // For an INT value without any suffix, we return it as BIGINT type value since it is
                  // the default integer type.
                  parseAndCastNumeric(ATypeTag.BIGINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT32_LITERAL:
                  parseAndCastNumeric(ATypeTag.INTEGER, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT32_CONS:
                  parseConstructor(ATypeTag.INTEGER, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT64_LITERAL:
                  parseAndCastNumeric(ATypeTag.BIGINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_INT64_CONS:
                  parseConstructor(ATypeTag.BIGINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_STRING_LITERAL:
                  if (checkType(ATypeTag.STRING, objectType)) {
                      admLexer.getLastTokenImage(tmpTokenImage);
                      if (admLexer.containsEscapes()) {
                          replaceEscapes(tmpTokenImage);
                      }
                      int begin = tmpTokenImage.getBegin() + 1;
                      int len = tmpTokenImage.getLength() - 2;
                      parseString(tmpTokenImage.getBuffer(), begin, len, out);
                  } else if (checkType(ATypeTag.UUID, objectType)) {
                      // Dealing with UUID type that is represented by a string
                      admLexer.getLastTokenImage(tmpTokenImage);
                      aUUID.parseUUIDString(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin() + 1,
                              tmpTokenImage.getLength() - 2);
                      uuidSerde.serialize(aUUID, out);
                  } else if (checkType(ATypeTag.GEOMETRY, objectType)) {
                      // Parse the string as a WKT-encoded geometry
                      String tokenImage =
                              admLexer.getLastTokenImage().substring(1, admLexer.getLastTokenImage().length() - 1);
                      aGeomtry.parseWKT(tokenImage);
                      out.writeByte(ATypeTag.GEOMETRY.serialize());
                      geomSerde.serialize(aGeomtry, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_STRING_CONS:
                  parseConstructor(ATypeTag.STRING, objectType, out);
                  break;
              case AdmLexer.TOKEN_HEX_CONS:
              case AdmLexer.TOKEN_BASE64_CONS:
                  if (checkType(ATypeTag.BINARY, objectType)) {
                      if (admLexer.next() == AdmLexer.TOKEN_CONSTRUCTOR_OPEN) {
                          if (admLexer.next() == AdmLexer.TOKEN_STRING_LITERAL) {
                              parseToBinaryTarget(token, admLexer.getLastTokenImage(), out);
                              if (admLexer.next() == AdmLexer.TOKEN_CONSTRUCTOR_CLOSE) {
                                  break;
                              }
                          }
                      }
                  }
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
              case AdmLexer.TOKEN_DATE_CONS:
                  parseConstructor(ATypeTag.DATE, objectType, out);
                  break;
              case AdmLexer.TOKEN_TIME_CONS:
                  parseConstructor(ATypeTag.TIME, objectType, out);
                  break;
              case AdmLexer.TOKEN_DATETIME_CONS:
                  parseConstructor(ATypeTag.DATETIME, objectType, out);
                  break;
              case AdmLexer.TOKEN_INTERVAL_CONS:
                  if (checkType(ATypeTag.INTERVAL, objectType)) {
                      objectType = getComplexType(objectType, ATypeTag.INTERVAL);
                      parseInterval(ATypeTag.INTERVAL, objectType, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_DURATION_CONS:
                  parseConstructor(ATypeTag.DURATION, objectType, out);
                  break;
              case AdmLexer.TOKEN_YEAR_MONTH_DURATION_CONS:
                  parseConstructor(ATypeTag.YEARMONTHDURATION, objectType, out);
                  break;
              case AdmLexer.TOKEN_DAY_TIME_DURATION_CONS:
                  parseConstructor(ATypeTag.DAYTIMEDURATION, objectType, out);
                  break;
              case AdmLexer.TOKEN_POINT_CONS:
                  parseConstructor(ATypeTag.POINT, objectType, out);
                  break;
              case AdmLexer.TOKEN_POINT3D_CONS:
                  parseConstructor(ATypeTag.POINT3D, objectType, out);
                  break;
              case AdmLexer.TOKEN_CIRCLE_CONS:
                  parseConstructor(ATypeTag.CIRCLE, objectType, out);
                  break;
              case AdmLexer.TOKEN_RECTANGLE_CONS:
                  parseConstructor(ATypeTag.RECTANGLE, objectType, out);
                  break;
              case AdmLexer.TOKEN_LINE_CONS:
                  parseConstructor(ATypeTag.LINE, objectType, out);
                  break;
              case AdmLexer.TOKEN_POLYGON_CONS:
                  parseConstructor(ATypeTag.POLYGON, objectType, out);
                  break;
              case AdmLexer.TOKEN_START_UNORDERED_LIST:
                  if (checkType(ATypeTag.MULTISET, objectType)) {
                      objectType = getComplexType(objectType, ATypeTag.MULTISET);
                      parseUnorderedList((AUnorderedListType) objectType, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_START_ORDERED_LIST:
                  if (checkType(ATypeTag.ARRAY, objectType)) {
                      objectType = getComplexType(objectType, ATypeTag.ARRAY);
                      parseOrderedList((AOrderedListType) objectType, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_START_RECORD:
                  if (checkType(ATypeTag.OBJECT, objectType)) {
                      objectType = getComplexType(objectType, ATypeTag.OBJECT);
                      parseRecord((ARecordType) objectType, out);
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH, objectType.getTypeName());
                  }
                  break;
              case AdmLexer.TOKEN_UUID_CONS:
                  parseConstructor(ATypeTag.UUID, objectType, out);
                  break;
              case AdmLexer.TOKEN_EOF:
                  break;
              default:
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_UNEXPECTED_TOKEN_KIND,
                          AdmLexer.tokenKindToString(token));
          }
  
      }
  
      // TODO: This function should be optimized. Currently it has complexity of O(N*N)!
      private void replaceEscapes(TokenImage tokenImage) throws ParseException {
          char[] chars = tokenImage.getBuffer();
          int end = tokenImage.getBegin() + tokenImage.getLength();
          int readpos = tokenImage.getBegin();
          int writepos = tokenImage.getBegin();
          int movemarker = tokenImage.getBegin();
          while (readpos < end) {
              if (chars[readpos] == '\\') {
                  moveChars(chars, movemarker, readpos, readpos - writepos);
                  switch (chars[readpos + 1]) {
                      case '\\':
                      case '\"':
                      case '/':
                          chars[writepos] = chars[readpos + 1];
                          break;
                      case 'b':
                          chars[writepos] = '\b';
                          break;
                      case 'f':
                          chars[writepos] = '\f';
                          break;
                      case 'n':
                          chars[writepos] = '\n';
                          break;
                      case 'r':
                          chars[writepos] = '\r';
                          break;
                      case 't':
                          chars[writepos] = '\t';
                          break;
                      case 'u':
                          chars[writepos] = (char) Integer.parseInt(new String(chars, readpos + 2, 4), 16);
                          readpos += 4;
                          break;
                      default:
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_ILLEGAL_ESCAPE, chars[readpos + 1]);
                  }
                  ++readpos;
                  movemarker = readpos + 1;
              }
              ++writepos;
              ++readpos;
          }
          moveChars(chars, movemarker, end, readpos - writepos);
          tokenImage.reset(chars, tokenImage.getBegin(), tokenImage.getLength() - (readpos - writepos));
      }
  
      private static void moveChars(char[] chars, int start, int end, int offset) {
          if (offset == 0) {
              return;
          }
          for (int i = start; i < end; ++i) {
              chars[i - offset] = chars[i];
          }
      }
  
      private IAType getComplexType(IAType aObjectType, ATypeTag tag) {
          if (aObjectType == null) {
              return null;
          }
  
          if (aObjectType.getTypeTag() == tag) {
              return aObjectType;
          }
  
          if (aObjectType.getTypeTag() == ATypeTag.UNION) {
              AUnionType unionType = (AUnionType) aObjectType;
              IAType type = unionType.getActualType();
              if (type.getTypeTag() == tag) {
                  return type;
              }
          }
          return null; // wont get here
      }
  
      private ATypeTag getTargetTypeTag(ATypeTag expectedTypeTag, IAType aObjectType) throws HyracksDataException {
          if (aObjectType == null) {
              return expectedTypeTag;
          }
          if (aObjectType.getTypeTag() != ATypeTag.UNION) {
              ATypeTag typeTag = aObjectType.getTypeTag();
              if (ATypeHierarchy.canPromote(expectedTypeTag, typeTag)
                      || ATypeHierarchy.canDemote(expectedTypeTag, typeTag)) {
                  return typeTag;
              } else {
                  return null;
              }
          } else { // union
              List<IAType> unionList = ((AUnionType) aObjectType).getUnionList();
              for (IAType t : unionList) {
                  final ATypeTag typeTag = t.getTypeTag();
                  if (ATypeHierarchy.canPromote(expectedTypeTag, typeTag)
                          || ATypeHierarchy.canDemote(expectedTypeTag, typeTag)) {
                      return typeTag;
                  }
              }
          }
          return null;
      }
  
      private boolean checkType(ATypeTag expectedTypeTag, IAType aObjectType) throws IOException {
          return getTargetTypeTag(expectedTypeTag, aObjectType) != null;
      }
  
      private void parseRecord(ARecordType recType, DataOutput out) throws IOException {
          ArrayBackedValueStorage fieldValueBuffer = getTempBuffer();
          ArrayBackedValueStorage fieldNameBuffer = getTempBuffer();
          IARecordBuilder recBuilder = getRecordBuilder();
  
          BitSet nulls = null;
          if (recType != null) {
              // TODO: use BitSet Pool
              nulls = new BitSet(recType.getFieldNames().length);
              recBuilder.reset(recType);
          } else {
              recBuilder.reset(null);
          }
  
          recBuilder.init();
          int token;
          boolean inRecord = true;
          boolean expectingRecordField = false;
          boolean first = true;
  
          Boolean openRecordField = false;
          int fieldId = 0;
          IAType fieldType = null;
          do {
              token = admLexer.next();
              switch (token) {
                  case AdmLexer.TOKEN_END_RECORD:
                      if (expectingRecordField) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_RECORD_END_UNEXPECTED);
                      }
                      inRecord = false;
                      break;
                  case AdmLexer.TOKEN_STRING_LITERAL:
                      // we've read the name of the field
                      // now read the content
                      fieldNameBuffer.reset();
                      fieldValueBuffer.reset();
                      expectingRecordField = false;
  
                      if (recType != null) {
                          admLexer.getLastTokenImage(tmpTokenImage);
                          String fldName = new String(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin() + 1,
                                  tmpTokenImage.getLength() - 2);
                          fieldId = recBuilder.getFieldId(fldName);
                          if ((fieldId < 0) && !recType.isOpen()) {
                              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_EXTRA_FIELD_IN_CLOSED_RECORD,
                                      LogRedactionUtil.userData(fldName));
                          } else if ((fieldId < 0) && recType.isOpen()) {
                              parseString(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin() + 1,
                                      tmpTokenImage.getLength() - 2, fieldNameBuffer.getDataOutput());
                              openRecordField = true;
                              fieldType = null;
                          } else {
                              // a closed field
                              nulls.set(fieldId);
                              fieldType = recType.getFieldTypes()[fieldId];
                              openRecordField = false;
                          }
                      } else {
                          admLexer.getLastTokenImage(tmpTokenImage);
                          parseString(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin() + 1,
                                  tmpTokenImage.getLength() - 2, fieldNameBuffer.getDataOutput());
                          openRecordField = true;
                          fieldType = null;
                      }
  
                      token = admLexer.next();
                      if (token != AdmLexer.TOKEN_COLON) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_UNEXPECTED_TOKEN_WHEN_EXPECT_COMMA,
                                  AdmLexer.tokenKindToString(token));
                      }
  
                      token = admLexer.next();
                      this.admFromLexerStream(token, fieldType, fieldValueBuffer.getDataOutput());
                      if (openRecordField) {
                          recBuilder.addField(fieldNameBuffer, fieldValueBuffer);
                      } else {
                          recBuilder.addField(fieldId, fieldValueBuffer);
                      }
  
                      break;
                  case AdmLexer.TOKEN_COMMA:
                      if (first) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_FOUND_COMMA_WHEN, "before any");
                      }
                      if (expectingRecordField) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_FOUND_COMMA_WHEN, "expecting a");
                      }
                      expectingRecordField = true;
                      break;
                  default:
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_UNEXPECTED_TOKEN_KIND,
                              AdmLexer.tokenKindToString(token));
              }
              first = false;
          } while (inRecord);
  
          if (recType != null) {
              final int nullableFieldId = checkOptionalConstraints(recType, nulls);
              if (nullableFieldId != -1) {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_FIELD_NOT_NULL,
                          LogRedactionUtil.userData(recType.getFieldNames()[nullableFieldId]));
              }
          }
          recBuilder.write(out, true);
      }
  
      private int checkOptionalConstraints(ARecordType recType, BitSet nulls) {
          for (int i = 0; i < recType.getFieldTypes().length; i++) {
              if (nulls.get(i) == false) {
                  IAType type = recType.getFieldTypes()[i];
                  if ((type.getTypeTag() != ATypeTag.NULL) && (type.getTypeTag() != ATypeTag.UNION)) {
                      return i;
                  }
  
                  if (type.getTypeTag() != ATypeTag.UNION) {
                      continue;
                  }
                  // union
                  AUnionType unionType = (AUnionType) type;
                  if (!unionType.isUnknownableType()) {
                      return i;
                  }
              }
          }
          return -1;
      }
  
      private void parseInterval(ATypeTag typeTag, IAType objectType, DataOutput out) throws IOException {
          long start = 0, end = 0;
          byte tag = 0;
          int token = admLexer.next();
          if (token == AdmLexer.TOKEN_CONSTRUCTOR_OPEN) {
              ATypeTag intervalType;
  
              token = admLexer.next();
              switch (token) {
                  case AdmLexer.TOKEN_DATE_CONS:
                      intervalType = ATypeTag.DATE;
                      break;
                  case AdmLexer.TOKEN_TIME_CONS:
                      intervalType = ATypeTag.TIME;
                      break;
                  case AdmLexer.TOKEN_DATETIME_CONS:
                      intervalType = ATypeTag.DATETIME;
                      break;
                  default:
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_UNSUPPORTED_INTERVAL_TYPE,
                              AdmLexer.tokenKindToString(token));
              }
  
              // Interval
              start = parseIntervalArgument(intervalType);
              end = parseIntervalSecondArgument(token, intervalType);
              tag = intervalType.serialize();
          }
  
          // Closing interval.
          token = admLexer.next();
          if (token == AdmLexer.TOKEN_CONSTRUCTOR_CLOSE) {
              try {
                  aInterval.setValue(start, end, tag);
              } catch (HyracksDataException e) {
                  throw new ParseException(e);
              }
          } else {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_NOT_CLOSED);
          }
          intervalSerde.serialize(aInterval, out);
      }
  
      private long parseIntervalSecondArgument(int startToken, ATypeTag parseType) throws IOException {
          int token = admLexer.next();
          if (token == AdmLexer.TOKEN_COMMA) {
              token = admLexer.next();
              if (token == startToken) {
                  return parseIntervalArgument(parseType);
              } else {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_BEGIN_END_POINT_MISMATCH,
                          AdmLexer.tokenKindToString(startToken), AdmLexer.tokenKindToString(token));
              }
          } else {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_MISSING_COMMA);
          }
      }
  
      private long parseIntervalArgument(ATypeTag tag) throws IOException {
          int token = admLexer.next();
          if (token == AdmLexer.TOKEN_CONSTRUCTOR_OPEN) {
              token = admLexer.next();
              if (token == AdmLexer.TOKEN_STRING_LITERAL) {
                  long chrononTimeInMs = 0;
                  String arg = admLexer.getLastTokenImage();
                  switch (tag) {
                      case DATE:
                          chrononTimeInMs +=
                                  (parseDatePart(arg, 0, arg.length() - 1) / GregorianCalendarSystem.CHRONON_OF_DAY);
                          break;
                      case TIME:
                          chrononTimeInMs += parseTimePart(arg, 0, arg.length() - 1);
                          break;
                      case DATETIME:
                          int timeSeperatorOffsetInDatetimeString = arg.indexOf('T');
                          if (timeSeperatorOffsetInDatetimeString < 0) {
                              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_INVALID_DATETIME);
                          }
                          chrononTimeInMs += parseDatePart(arg, 0, timeSeperatorOffsetInDatetimeString - 1);
                          chrononTimeInMs +=
                                  parseTimePart(arg, timeSeperatorOffsetInDatetimeString + 1, arg.length() - 1);
                          break;
                      default:
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_UNSUPPORTED_TYPE);
                  }
                  token = admLexer.next();
                  if (token == AdmLexer.TOKEN_CONSTRUCTOR_CLOSE) {
                      return chrononTimeInMs;
                  }
              }
          }
          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_INTERVAL_INTERVAL_ARGUMENT_ERROR);
      }
  
      private void parseOrderedList(AOrderedListType oltype, DataOutput out) throws IOException {
          ArrayBackedValueStorage itemBuffer = getTempBuffer();
          OrderedListBuilder orderedListBuilder = (OrderedListBuilder) getOrderedListBuilder();
  
          IAType itemType = null;
          if (oltype != null) {
              itemType = oltype.getItemType();
          }
          orderedListBuilder.reset(oltype);
  
          int token;
          boolean inList = true;
          boolean expectingListItem = false;
          boolean first = true;
          do {
              token = admLexer.next();
              if (token == AdmLexer.TOKEN_END_ORDERED_LIST) {
                  if (expectingListItem) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_END_COLLECTION);
                  }
                  inList = false;
              } else if (token == AdmLexer.TOKEN_COMMA) {
                  if (first) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_COMMA_BEFORE_LIST);
                  }
                  if (expectingListItem) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_COMMA_EXPECTING_ITEM);
                  }
                  expectingListItem = true;
              } else {
                  expectingListItem = false;
                  itemBuffer.reset();
  
                  admFromLexerStream(token, itemType, itemBuffer.getDataOutput());
                  orderedListBuilder.addItem(itemBuffer);
              }
              first = false;
          } while (inList);
          orderedListBuilder.write(out, true);
      }
  
      private void parseUnorderedList(AUnorderedListType uoltype, DataOutput out) throws IOException {
          ArrayBackedValueStorage itemBuffer = getTempBuffer();
          UnorderedListBuilder unorderedListBuilder = (UnorderedListBuilder) getUnorderedListBuilder();
  
          IAType itemType = null;
  
          if (uoltype != null) {
              itemType = uoltype.getItemType();
          }
          unorderedListBuilder.reset(uoltype);
  
          int token;
          boolean inList = true;
          boolean expectingListItem = false;
          boolean first = true;
          do {
              token = admLexer.next();
              if (token == AdmLexer.TOKEN_END_RECORD) {
                  if (admLexer.next() == AdmLexer.TOKEN_END_RECORD) {
                      if (expectingListItem) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_END_COLLECTION);
                      } else {
                          inList = false;
                      }
                  } else {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_END_RECOD);
                  }
              } else if (token == AdmLexer.TOKEN_COMMA) {
                  if (first) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_COMMA_BEFORE_LIST);
                  }
                  if (expectingListItem) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_LIST_FOUND_COMMA_EXPECTING_ITEM);
                  }
                  expectingListItem = true;
              } else {
                  expectingListItem = false;
                  itemBuffer.reset();
                  admFromLexerStream(token, itemType, itemBuffer.getDataOutput());
                  unorderedListBuilder.addItem(itemBuffer);
              }
              first = false;
          } while (inList);
          unorderedListBuilder.write(out, true);
      }
  
      private IARecordBuilder getRecordBuilder() {
          return recordBuilderPool.allocate(ATypeTag.OBJECT);
      }
  
      private IAsterixListBuilder getOrderedListBuilder() {
          return listBuilderPool.allocate(ATypeTag.ARRAY);
      }
  
      private IAsterixListBuilder getUnorderedListBuilder() {
          return listBuilderPool.allocate(ATypeTag.MULTISET);
      }
  
      private ArrayBackedValueStorage getTempBuffer() {
          return (ArrayBackedValueStorage) abvsBuilderPool.allocate(ATypeTag.BINARY);
      }
  
      private void parseToBinaryTarget(int lexerToken, String tokenImage, DataOutput out)
              throws ParseException, HyracksDataException {
          switch (lexerToken) {
              case AdmLexer.TOKEN_HEX_CONS:
                  parseHexBinaryString(tokenImage.toCharArray(), 1, tokenImage.length() - 2, out);
                  break;
              case AdmLexer.TOKEN_BASE64_CONS:
                  parseBase64BinaryString(tokenImage.toCharArray(), 1, tokenImage.length() - 2, out);
                  break;
          }
      }
  
      private void parseToNumericTarget(ATypeTag typeTag, IAType objectType, DataOutput out) throws IOException {
          ATypeTag targetTypeTag = getTargetTypeTag(typeTag, objectType);
          boolean parsed = false;
          if (targetTypeTag != null) {
              admLexer.getLastTokenImage(tmpTokenImage);
              parsed = parseValue(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin(), tmpTokenImage.getLength(),
                      targetTypeTag, out);
          }
          if (!parsed) {
              throw new ParseException(mismatchErrorMessage + objectType.getTypeName() + mismatchErrorMessage2 + typeTag);
          }
      }
  
      private void parseAndCastNumeric(ATypeTag typeTag, IAType objectType, DataOutput out) throws IOException {
          ATypeTag targetTypeTag = getTargetTypeTag(typeTag, objectType);
          DataOutput dataOutput = out;
          if (targetTypeTag != typeTag) {
              castBuffer.reset();
              dataOutput = castBuffer.getDataOutput();
          }
          boolean parsed = false;
          if (targetTypeTag != null) {
              admLexer.getLastTokenImage(tmpTokenImage);
              parsed = parseValue(tmpTokenImage.getBuffer(), tmpTokenImage.getBegin(), tmpTokenImage.getLength(), typeTag,
                      dataOutput);
          }
          if (!parsed) {
              throw new ParseException(mismatchErrorMessage + objectType.getTypeName() + mismatchErrorMessage2 + typeTag);
          }
  
          // If two type tags are not the same, either we try to promote or demote source type to the
          // target type
          if (targetTypeTag != typeTag) {
              if (ATypeHierarchy.canPromote(typeTag, targetTypeTag)) {
                  // can promote typeTag to targetTypeTag
                  ITypeConvertComputer promoteComputer = ATypeHierarchy.getTypePromoteComputer(typeTag, targetTypeTag);
                  if (promoteComputer == null) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_CAST_ERROR, typeTag, targetTypeTag);
                  }
                  // do the promotion; note that the type tag field should be skipped
                  promoteComputer.convertType(castBuffer.getByteArray(), castBuffer.getStartOffset() + 1,
                          castBuffer.getLength() - 1, out);
              } else if (ATypeHierarchy.canDemote(typeTag, targetTypeTag)) {
                  // can demote source type to the target type
                  ITypeConvertComputer demoteComputer =
                          ATypeHierarchy.getTypeDemoteComputer(typeTag, targetTypeTag, true);
                  if (demoteComputer == null) {
                      throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_CAST_ERROR, typeTag, targetTypeTag);
                  }
                  // do the demotion; note that the type tag field should be skipped
                  demoteComputer.convertType(castBuffer.getByteArray(), castBuffer.getStartOffset() + 1,
                          castBuffer.getLength() - 1, out);
              }
          }
      }
  
      private void parseConstructor(ATypeTag typeTag, IAType objectType, DataOutput out) throws IOException {
          ATypeTag targetTypeTag = getTargetTypeTag(typeTag, objectType);
          if (targetTypeTag != null) {
              DataOutput dataOutput = out;
              if (targetTypeTag != typeTag) {
                  castBuffer.reset();
                  dataOutput = castBuffer.getDataOutput();
              }
              int token = admLexer.next();
              if (token == AdmLexer.TOKEN_CONSTRUCTOR_OPEN) {
                  token = admLexer.next();
                  if (token == AdmLexer.TOKEN_STRING_LITERAL) {
                      admLexer.getLastTokenImage(tmpTokenImage);
                      int begin = tmpTokenImage.getBegin() + 1;
                      int len = tmpTokenImage.getLength() - 2;
                      // unquoted value
                      if (!parseValue(tmpTokenImage.getBuffer(), begin, len, typeTag, dataOutput)) {
                          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_CONSTRUCTOR_MISSING_DESERIALIZER,
                                  AdmLexer.tokenKindToString(token));
                      }
                      token = admLexer.next();
                      if (token == AdmLexer.TOKEN_CONSTRUCTOR_CLOSE) {
                          if (targetTypeTag != typeTag) {
                              ITypeConvertComputer promoteComputer =
                                      ATypeHierarchy.getTypePromoteComputer(typeTag, targetTypeTag);
                              // the availability if the promote computer should be consistent with
                              // the availability of a target type
                              assert promoteComputer != null;
                              // do the promotion; note that the type tag field should be skipped
                              promoteComputer.convertType(castBuffer.getByteArray(), castBuffer.getStartOffset() + 1,
                                      castBuffer.getLength() - 1, out);
                          }
                          return;
                      }
                  }
              }
          }
  
          throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_TYPE_MISMATCH,
                  objectType.getTypeName() + " got " + typeTag);
      }
  
      private boolean parseValue(char[] buffer, int begin, int len, ATypeTag typeTag, DataOutput out)
              throws HyracksDataException {
          switch (typeTag) {
              case BOOLEAN:
                  parseBoolean(buffer, begin, len, out);
                  return true;
              case TINYINT:
                  parseInt8(buffer, begin, len, out);
                  return true;
              case SMALLINT:
                  parseInt16(buffer, begin, len, out);
                  return true;
              case INTEGER:
                  parseInt32(buffer, begin, len, out);
                  return true;
              case BIGINT:
                  parseInt64(buffer, begin, len, out);
                  return true;
              case FLOAT:
                  if (matches("INF", buffer, begin, len)) {
                      aFloat.setValue(Float.POSITIVE_INFINITY);
                  } else if (matches("-INF", buffer, begin, len)) {
                      aFloat.setValue(Float.NEGATIVE_INFINITY);
                  } else {
                      aFloat.setValue(parseFloat(buffer, begin, len));
                  }
                  floatSerde.serialize(aFloat, out);
                  return true;
              case DOUBLE:
                  if (matches("INF", buffer, begin, len)) {
                      aDouble.setValue(Double.POSITIVE_INFINITY);
                  } else if (matches("-INF", buffer, begin, len)) {
                      aDouble.setValue(Double.NEGATIVE_INFINITY);
                  } else {
                      aDouble.setValue(parseDouble(buffer, begin, len));
                  }
                  doubleSerde.serialize(aDouble, out);
                  return true;
              case STRING:
                  parseString(buffer, begin, len, out);
                  return true;
              case TIME:
                  parseTime(buffer, begin, len, out);
                  return true;
              case DATE:
                  parseDate(buffer, begin, len, out);
                  return true;
              case DATETIME:
                  parseDateTime(buffer, begin, len, out);
                  return true;
              case DURATION:
                  parseDuration(buffer, begin, len, out);
                  return true;
              case DAYTIMEDURATION:
                  parseDateTimeDuration(buffer, begin, len, out);
                  return true;
              case YEARMONTHDURATION:
                  parseYearMonthDuration(buffer, begin, len, out);
                  return true;
              case POINT:
                  parsePoint(buffer, begin, len, out);
                  return true;
              case POINT3D:
                  parse3DPoint(buffer, begin, len, out);
                  return true;
              case CIRCLE:
                  parseCircle(buffer, begin, len, out);
                  return true;
              case RECTANGLE:
                  parseRectangle(buffer, begin, len, out);
                  return true;
              case LINE:
                  parseLine(buffer, begin, len, out);
                  return true;
              case POLYGON:
                  //TODO: optimize
                  APolygonSerializerDeserializer.parse(new String(buffer, begin, len), out);
                  return true;
              case UUID:
                  aUUID.parseUUIDString(buffer, begin, len);
                  uuidSerde.serialize(aUUID, out);
                  return true;
              default:
                  return false;
          }
      }
  
      private boolean matches(String value, char[] buffer, int begin, int len) {
          if (len != value.length()) {
              return false;
          }
          for (int i = 0; i < len; i++) {
              if (value.charAt(i) != buffer[i + begin]) {
                  return false;
              }
          }
          return true;
      }
  
      private void parseBoolean(char[] buffer, int begin, int len, DataOutput out) throws HyracksDataException {
          if (matches("true", buffer, begin, len)) {
              booleanSerde.serialize(ABoolean.TRUE, out);
          } else if (matches("false", buffer, begin, len)) {
              booleanSerde.serialize(ABoolean.FALSE, out);
          } else {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE, new String(buffer, begin, len),
                      "boolean");
          }
      }
  
      private void parseInt8(char[] buffer, int begin, int len, DataOutput out) throws HyracksDataException {
          boolean positive = true;
          byte value = 0;
          int offset = begin;
  
          if (buffer[offset] == '+') {
              offset++;
          } else if (buffer[offset] == '-') {
              offset++;
              positive = false;
          }
          for (; offset < begin + len; offset++) {
              if ((buffer[offset] >= '0') && (buffer[offset] <= '9')) {
                  value = (byte) (((value * 10) + buffer[offset]) - '0');
              } else if (buffer[offset] == 'i' && buffer[offset + 1] == '8' && offset + 2 == begin + len) {
                  break;
              } else {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE,
                          new String(buffer, begin, len), "int8");
              }
          }
          if (value < 0) {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE, new String(buffer, begin, len),
                      "int8");
          }
          if ((value > 0) && !positive) {
              value *= -1;
          }
          aInt8.setValue(value);
          int8Serde.serialize(aInt8, out);
      }
  
      private void parseInt16(char[] buffer, int begin, int len, DataOutput out) throws HyracksDataException {
          boolean positive = true;
          short value = 0;
          int offset = begin;
  
          if (buffer[offset] == '+') {
              offset++;
          } else if (buffer[offset] == '-') {
              offset++;
              positive = false;
          }
          for (; offset < begin + len; offset++) {
              if (buffer[offset] >= '0' && buffer[offset] <= '9') {
                  value = (short) ((value * 10) + buffer[offset] - '0');
              } else if (buffer[offset] == 'i' && buffer[offset + 1] == '1' && buffer[offset + 2] == '6'
                      && offset + 3 == begin + len) {
                  break;
              } else {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE,
                          new String(buffer, begin, len), "int16");
              }
          }
          if (value < 0) {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE, new String(buffer, begin, len),
                      "int16");
          }
          if ((value > 0) && !positive) {
              value *= -1;
          }
          aInt16.setValue(value);
          int16Serde.serialize(aInt16, out);
      }
  
      private void parseInt32(char[] buffer, int begin, int len, DataOutput out) throws HyracksDataException {
          boolean positive = true;
          int value = 0;
          int offset = begin;
  
          if (buffer[offset] == '+') {
              offset++;
          } else if (buffer[offset] == '-') {
              offset++;
              positive = false;
          }
          for (; offset < begin + len; offset++) {
              if (buffer[offset] >= '0' && buffer[offset] <= '9') {
                  value = (value * 10) + buffer[offset] - '0';
              } else if (buffer[offset] == 'i' && buffer[offset + 1] == '3' && buffer[offset + 2] == '2'
                      && offset + 3 == begin + len) {
                  break;
              } else {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE,
                          new String(buffer, begin, len), "int32");
              }
          }
          if (value < 0) {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE, new String(buffer, begin, len),
                      "int32");
          }
          if ((value > 0) && !positive) {
              value *= -1;
          }
  
          aInt32.setValue(value);
          int32Serde.serialize(aInt32, out);
      }
  
      private void parseInt64(char[] buffer, int begin, int len, DataOutput out) throws HyracksDataException {
          boolean positive = true;
          long value = 0;
          int offset = begin;
  
          if (buffer[offset] == '+') {
              offset++;
          } else if (buffer[offset] == '-') {
              offset++;
              positive = false;
          }
          for (; offset < begin + len; offset++) {
              if (buffer[offset] >= '0' && buffer[offset] <= '9') {
                  value = (value * 10) + buffer[offset] - '0';
              } else if (buffer[offset] == 'i' && buffer[offset + 1] == '6' && buffer[offset + 2] == '4'
                      && offset + 3 == begin + len) {
                  break;
              } else {
                  throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE,
                          new String(buffer, begin, len), "int64");
              }
          }
          if (value < 0) {
              throw new ParseException(ErrorCode.PARSER_ADM_DATA_PARSER_WRONG_INSTANCE, new String(buffer, begin, len),
                      "int64");
          }
          if ((value > 0) && !positive) {
              value *= -1;
          }
  
          aInt64.setValue(value);
          int64Serde.serialize(aInt64, out);
      }
  
      /**
       * Resets the pools before parsing a top-level record.
       * In this way the elements in those pools can be re-used.
       */
      private void resetPools() {
          listBuilderPool.reset();
          recordBuilderPool.reset();
          abvsBuilderPool.reset();
      }
  
      @Override
      public boolean reset(InputStream in) throws IOException {
          admLexer.reInit(new InputStreamReader(in));
          return true;
      }
  }
  