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.runtime.evaluators.functions.records;
  
  import java.io.DataOutput;
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  
  import org.apache.asterix.builders.RecordBuilder;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.common.exceptions.RuntimeDataException;
  import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
  import org.apache.asterix.om.base.AGeneratedUUID;
  import org.apache.asterix.om.base.AUUID;
  import org.apache.asterix.om.pointables.ARecordVisitablePointable;
  import org.apache.asterix.om.pointables.PointableAllocator;
  import org.apache.asterix.om.pointables.base.IVisitablePointable;
  import org.apache.asterix.om.types.ARecordType;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.runtime.RuntimeRecordTypeInfo;
  import org.apache.asterix.runtime.evaluators.functions.AbstractScalarEval;
  import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
  import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
  import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
  import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
  import org.apache.hyracks.api.context.IEvaluatorContext;
  import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
  import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.api.exceptions.SourceLocation;
  import org.apache.hyracks.data.std.accessors.UTF8StringBinaryComparatorFactory;
  import org.apache.hyracks.data.std.api.IPointable;
  import org.apache.hyracks.data.std.api.IValueReference;
  import org.apache.hyracks.data.std.primitive.VoidPointable;
  import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
  import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
  
  public class PutAutogeneratedKeyEvaluator extends AbstractScalarEval {
  
      private final IPointable inRecPointable = new VoidPointable();
      private final List<RecordBuilder> rbStack = new ArrayList<>();
      private final ArrayBackedValueStorage tempStorage = new ArrayBackedValueStorage();
      private final IBinaryComparator stringBinaryComparator =
              UTF8StringBinaryComparatorFactory.INSTANCE.createBinaryComparator();
      private final RuntimeRecordTypeInfo runtimeRecordTypeInfo = new RuntimeRecordTypeInfo();
      private final ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
      private final DataOutput out = resultStorage.getDataOutput();
      private final ArrayBackedValueStorage uuidStorage = new ArrayBackedValueStorage();
      private final DataOutput uuidOut = uuidStorage.getDataOutput();
      private final AGeneratedUUID uuid = new AGeneratedUUID();
      @SuppressWarnings("unchecked")
      private final ISerializerDeserializer<AUUID> uuidSerDe =
              SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AUUID);
  
      private final ARecordType outRecType;
      private final IVisitablePointable inRecVisitable;
      private final IPointable[] keyNameParts;
      private final IScalarEvaluator inRecEval;
      private final IScalarEvaluator[] keyNamePartsEvals;
  
      PutAutogeneratedKeyEvaluator(IEvaluatorContext ctx, IScalarEvaluatorFactory[] args, ARecordType outRecType,
              ARecordType inRecType, SourceLocation sourceLocation, FunctionIdentifier identifier)
              throws HyracksDataException {
          super(sourceLocation, identifier);
          this.outRecType = outRecType;
          inRecEval = args[0].createScalarEvaluator(ctx);
          keyNamePartsEvals = new IScalarEvaluator[args.length - 1];
          keyNameParts = new VoidPointable[args.length - 1];
          for (int i = 1, e = 0; i < args.length; i++, e++) {
              keyNamePartsEvals[e] = args[i].createScalarEvaluator(ctx);
              keyNameParts[e] = new VoidPointable();
          }
          PointableAllocator pa = new PointableAllocator();
          inRecVisitable = pa.allocateRecordValue(inRecType);
      }
  
      @Override
      public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
          inRecEval.evaluate(tuple, inRecPointable);
          if (PointableHelper.checkAndSetMissingOrNull(result, inRecPointable)) {
              // TODO: can probably fail instead of producing NULL/MISSING
              return;
          }
          ATypeTag inputTypeTag = PointableHelper.getTypeTag(inRecPointable);
          if (inputTypeTag != ATypeTag.OBJECT) {
              throw new RuntimeDataException(ErrorCode.CASTING_FIELD, srcLoc, inputTypeTag, ATypeTag.OBJECT);
          }
          for (int i = 0; i < keyNamePartsEvals.length; i++) {
              keyNamePartsEvals[i].evaluate(tuple, keyNameParts[i]);
              if (PointableHelper.checkAndSetMissingOrNull(result, keyNameParts[i])) {
                  // TODO: can probably fail instead of producing NULL/MISSING
                  return;
              }
          }
          resultStorage.reset();
          processTuple(tuple);
          result.set(resultStorage);
      }
  
      private void processTuple(IFrameTupleReference tuple) throws HyracksDataException {
          inRecVisitable.set(inRecPointable);
          try {
              mergeAutoKeyToRecord(outRecType, (ARecordVisitablePointable) inRecVisitable, 0, tuple, keyNameParts[0]);
              rbStack.get(0).write(out, true);
          } catch (IOException e) {
              throw HyracksDataException.create(e);
          }
      }
  
      private void mergeAutoKeyToRecord(ARecordType outType, ARecordVisitablePointable record, int nestedLevel,
              IFrameTupleReference tuple, IValueReference keyNamePart) throws IOException {
          if (rbStack.size() < (nestedLevel + 1)) {
              rbStack.add(new RecordBuilder());
          }
          rbStack.get(nestedLevel).reset(outType);
          rbStack.get(nestedLevel).init();
          boolean foundKeyNamePart = false;
          if (record != null) {
              for (int i = 0; i < record.getFieldNames().size(); i++) {
                  IVisitablePointable fieldName = record.getFieldNames().get(i);
                  IVisitablePointable fieldVal = record.getFieldValues().get(i);
                  if (PointableHelper.isEqual(fieldName, keyNamePart, stringBinaryComparator)) {
                      foundKeyNamePart = true;
                      addKeyPart(outType, nestedLevel, tuple, fieldName, fieldVal);
                  } else {
                      addField(outType, fieldName, fieldVal, nestedLevel);
                  }
              }
          }
          if (!foundKeyNamePart) {
              // the input record does not have the key, generate a new key with all the field(s) nesting needed
              if (lastKeyPartName(nestedLevel, keyNamePartsEvals.length)) {
                  addAutogeneratedKey(outType, keyNamePart, nestedLevel);
              } else {
                  mergeAutoKeyToRecordField(outType, keyNamePart, null, nestedLevel, tuple);
              }
          }
      }
  
      private void addKeyPart(ARecordType outType, int nestedLevel, IFrameTupleReference tuple,
              IVisitablePointable keyPartName, IVisitablePointable keyPartVal) throws IOException {
          ATypeTag typeTag = PointableHelper.getTypeTag(keyPartVal);
          if (lastKeyPartName(nestedLevel, keyNamePartsEvals.length)) {
              switch (typeTag) {
                  case UUID:
                      addField(outType, keyPartName, keyPartVal, nestedLevel);
                      break;
                  case MISSING:
                      addAutogeneratedKey(outType, keyPartName, nestedLevel);
                      break;
                  default:
                      throw new RuntimeDataException(ErrorCode.CASTING_FIELD, srcLoc, typeTag, ATypeTag.UUID);
              }
          } else {
              if (typeTag != ATypeTag.OBJECT) {
                  throw new RuntimeDataException(ErrorCode.CASTING_FIELD, srcLoc, typeTag, ATypeTag.OBJECT);
              }
              mergeAutoKeyToRecordField(outType, keyPartName, (ARecordVisitablePointable) keyPartVal, nestedLevel, tuple);
          }
      }
  
      private void mergeAutoKeyToRecordField(ARecordType combinedType, IValueReference recFieldName,
              ARecordVisitablePointable recFieldVal, int nestedLevel, IFrameTupleReference tuple) throws IOException {
          runtimeRecordTypeInfo.reset(combinedType);
          int pos = runtimeRecordTypeInfo.getFieldIndex(recFieldName.getByteArray(), recFieldName.getStartOffset() + 1,
                  recFieldName.getLength() - 1);
          mergeAutoKeyToRecord((ARecordType) combinedType.getFieldTypes()[pos], recFieldVal, nestedLevel + 1, tuple,
                  keyNameParts[nestedLevel + 1]);
          tempStorage.reset();
          rbStack.get(nestedLevel + 1).write(tempStorage.getDataOutput(), true);
          rbStack.get(nestedLevel).addField(pos, tempStorage);
      }
  
      private void addAutogeneratedKey(ARecordType outType, IValueReference keyNamePart, int nestedLevel)
              throws IOException {
          uuidStorage.reset();
          uuid.nextUUID();
          uuidSerDe.serialize(uuid, uuidOut);
          addField(outType, keyNamePart, uuidStorage, nestedLevel);
      }
  
      private void addField(ARecordType outType, IValueReference fieldName, IValueReference fieldValue, int nestedLevel)
              throws IOException {
          runtimeRecordTypeInfo.reset(outType);
          int pos = runtimeRecordTypeInfo.getFieldIndex(fieldName.getByteArray(), fieldName.getStartOffset() + 1,
                  fieldName.getLength() - 1);
          if (pos >= 0) {
              rbStack.get(nestedLevel).addField(pos, fieldValue);
          } else {
              rbStack.get(nestedLevel).addField(fieldName, fieldValue);
          }
      }
  
      private static boolean lastKeyPartName(int nestedLevel, int keyNameParts) {
          return nestedLevel == keyNameParts - 1;
      }
  }
  