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.util.Iterator;
  
  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.common.exceptions.ErrorCode;
  import org.apache.asterix.common.exceptions.RuntimeDataException;
  import org.apache.asterix.external.api.IRawRecord;
  import org.apache.asterix.external.api.IRecordDataParser;
  import org.apache.asterix.om.base.ANull;
  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.BuiltinType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.om.util.container.IObjectPool;
  import org.apache.asterix.om.util.container.ListObjectPool;
  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;
  import org.apache.hyracks.util.string.UTF8StringWriter;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.ObjectMapper;
  
  /**
   * This class was introduced to parse Twitter data. As Tweets are JSON formatted records, we can use the JSON parser to
   * to parse them instead of having this dedicated parser. In the future, we could either deprecate this class, or add
   * some Tweet specific parsing processes into this class.
   */
  
  public class TweetParser extends AbstractDataParser implements IRecordDataParser<char[]> {
      private final IObjectPool<IARecordBuilder, ATypeTag> recordBuilderPool =
              new ListObjectPool<>(new RecordBuilderFactory());
      private final IObjectPool<IAsterixListBuilder, ATypeTag> listBuilderPool =
              new ListObjectPool<>(new ListBuilderFactory());
      private final IObjectPool<IMutableValueStorage, ATypeTag> abvsBuilderPool =
              new ListObjectPool<>(new AbvsBuilderFactory());
      private ARecordType recordType;
      private UTF8StringWriter utf8Writer = new UTF8StringWriter();
  
      public TweetParser(ARecordType recordType) {
          this.recordType = recordType;
      }
  
      private void parseArrayList(JsonNode jArray, DataOutput output) throws IOException {
          ArrayBackedValueStorage itemBuffer = getTempBuffer();
          OrderedListBuilder arrayBuilder = (OrderedListBuilder) getArrayBuilder();
  
          arrayBuilder.reset(null);
          for (int iter1 = 0; iter1 < jArray.size(); iter1++) {
              itemBuffer.reset();
              if (writeField(jArray.get(iter1), null, itemBuffer.getDataOutput())) {
                  arrayBuilder.addItem(itemBuffer);
              }
          }
          arrayBuilder.write(output, true);
      }
  
      private boolean writeField(JsonNode fieldObj, IAType originalFieldType, DataOutput out) throws IOException {
          boolean writeResult = true;
          IAType fieldType = originalFieldType;
          if (originalFieldType instanceof AUnionType) {
              fieldType = ((AUnionType) originalFieldType).getActualType();
          }
          if (fieldType != null) {
              switch (fieldType.getTypeTag()) {
                  case STRING:
                      out.write(BuiltinType.ASTRING.getTypeTag().serialize());
                      utf8Writer.writeUTF8(fieldObj.asText(), out);
                      break;
                  case BIGINT:
                      aInt64.setValue(fieldObj.asLong());
                      int64Serde.serialize(aInt64, out);
                      break;
                  case INTEGER:
                      out.write(BuiltinType.AINT32.getTypeTag().serialize());
                      out.writeInt(fieldObj.asInt());
                      break;
                  case DOUBLE:
                      out.write(BuiltinType.ADOUBLE.getTypeTag().serialize());
                      out.writeDouble(fieldObj.asDouble());
                      break;
                  case BOOLEAN:
                      out.write(BuiltinType.ABOOLEAN.getTypeTag().serialize());
                      out.writeBoolean(fieldObj.asBoolean());
                      break;
                  case OBJECT:
                      writeRecord(fieldObj, out, (ARecordType) fieldType);
                      break;
                  case ARRAY:
                      parseArrayList(fieldObj, out);
                      break;
                  default:
                      writeResult = false;
              }
          } else {
              if (fieldObj.isNull()) {
                  nullSerde.serialize(ANull.NULL, out);
              } else if (fieldObj.isInt()) {
                  out.write(BuiltinType.AINT32.getTypeTag().serialize());
                  out.writeInt(fieldObj.asInt());
              } else if (fieldObj.isBoolean()) {
                  out.write(BuiltinType.ABOOLEAN.getTypeTag().serialize());
                  out.writeBoolean(fieldObj.asBoolean());
              } else if (fieldObj.isDouble()) {
                  out.write(BuiltinType.ADOUBLE.getTypeTag().serialize());
                  out.writeDouble(fieldObj.asDouble());
              } else if (fieldObj.isLong()) {
                  out.write(BuiltinType.AINT64.getTypeTag().serialize());
                  out.writeLong(fieldObj.asLong());
              } else if (fieldObj.isTextual()) {
                  out.write(BuiltinType.ASTRING.getTypeTag().serialize());
                  utf8Writer.writeUTF8(fieldObj.asText(), out);
              } else if (fieldObj.isArray()) {
                  if ((fieldObj).size() != 0) {
                      parseArrayList(fieldObj, out);
                  } else {
                      writeResult = false;
                  }
              } else if (fieldObj.isObject()) {
                  if ((fieldObj).size() != 0) {
                      writeRecord(fieldObj, out, null);
                  } else {
                      writeResult = false;
                  }
              }
          }
          return writeResult;
      }
  
      private int checkAttrNameIdx(String[] nameList, String name) {
          int idx = 0;
          if (nameList != null) {
              for (String nln : nameList) {
                  if (name.equals(nln)) {
                      return idx;
                  }
                  idx++;
              }
          }
          return -1;
      }
  
      public void writeRecord(JsonNode obj, DataOutput out, ARecordType curRecType) throws IOException {
          IAType[] curTypes = null;
          String[] curFNames = null;
          int fieldN;
          int attrIdx;
          int expectedFieldsCount = 0;
  
          ArrayBackedValueStorage fieldValueBuffer = getTempBuffer();
          ArrayBackedValueStorage fieldNameBuffer = getTempBuffer();
          IARecordBuilder recBuilder = getRecordBuilder();
  
          if (curRecType != null) {
              curTypes = curRecType.getFieldTypes();
              curFNames = curRecType.getFieldNames();
              for (IAType curType : curTypes) {
                  if (!(curType instanceof AUnionType)) {
                      expectedFieldsCount++;
                  }
              }
          }
  
          recBuilder.reset(curRecType);
          recBuilder.init();
  
          if (curRecType != null && !curRecType.isOpen()) {
              // closed record type
              fieldN = curFNames.length;
              for (int iter1 = 0; iter1 < fieldN; iter1++) {
                  fieldValueBuffer.reset();
                  DataOutput fieldOutput = fieldValueBuffer.getDataOutput();
                  if (obj.get(curFNames[iter1]).isNull() && !(curTypes[iter1] instanceof AUnionType)) {
                      if (curRecType.isClosedField(curFNames[iter1])) {
                          throw new RuntimeDataException(ErrorCode.PARSER_EXT_DATA_PARSER_CLOSED_FIELD_NULL,
                                  LogRedactionUtil.userData(curFNames[iter1]));
                      } else {
                          continue;
                      }
                  } else {
                      if (writeField(obj.get(curFNames[iter1]), curTypes[iter1], fieldOutput)) {
                          recBuilder.addField(iter1, fieldValueBuffer);
                      }
                  }
              }
          } else {
              //open record type
              int closedFieldCount = 0;
              IAType curFieldType = null;
              String attrName;
              Iterator<String> iter = obj.fieldNames();
              while (iter.hasNext()) {
                  attrName = iter.next();
                  if (obj.get(attrName) == null || obj.get(attrName).isNull() || obj.size() == 0) {
                      continue;
                  }
                  attrIdx = checkAttrNameIdx(curFNames, attrName);
                  if (curRecType != null) {
                      curFieldType = curRecType.getFieldType(attrName);
                  }
                  fieldValueBuffer.reset();
                  fieldNameBuffer.reset();
                  DataOutput fieldOutput = fieldValueBuffer.getDataOutput();
                  if (writeField(obj.get(attrName), curFieldType, fieldOutput)) {
                      if (attrIdx == -1) {
                          aString.setValue(attrName);
                          stringSerde.serialize(aString, fieldNameBuffer.getDataOutput());
                          recBuilder.addField(fieldNameBuffer, fieldValueBuffer);
                      } else {
                          recBuilder.addField(attrIdx, fieldValueBuffer);
                          closedFieldCount++;
                      }
                  }
              }
              if (curRecType != null && closedFieldCount < expectedFieldsCount) {
                  throw new HyracksDataException("Non-null field is null");
              }
          }
          recBuilder.write(out, true);
      }
  
      private IARecordBuilder getRecordBuilder() {
          return recordBuilderPool.allocate(ATypeTag.OBJECT);
      }
  
      private IAsterixListBuilder getArrayBuilder() {
          return listBuilderPool.allocate(ATypeTag.ARRAY);
      }
  
      private ArrayBackedValueStorage getTempBuffer() {
          return (ArrayBackedValueStorage) abvsBuilderPool.allocate(ATypeTag.BINARY);
      }
  
      @Override
      public boolean parse(IRawRecord<? extends char[]> record, DataOutput out) throws HyracksDataException {
          try {
              //TODO get rid of this temporary json
              resetPools();
              ObjectMapper om = new ObjectMapper();
              writeRecord(om.readTree(record.getBytes()), out, recordType);
              return true;
          } catch (IOException e) {
              throw HyracksDataException.create(e);
          }
      }
  
      private void resetPools() {
          listBuilderPool.reset();
          recordBuilderPool.reset();
          abvsBuilderPool.reset();
      }
  }
  