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.om.typecomputer.impl;
  
  import java.util.Arrays;
  import java.util.List;
  
  import org.apache.asterix.common.exceptions.CompilationException;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.om.exceptions.TypeMismatchException;
  import org.apache.asterix.om.typecomputer.base.IResultTypeComputer;
  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.types.TypeHelper;
  import org.apache.asterix.om.utils.ConstantExpressionUtil;
  import org.apache.commons.lang3.mutable.Mutable;
  import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
  import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
  import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
  import org.apache.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
  import org.apache.hyracks.api.exceptions.SourceLocation;
  
  public class PutAutogeneratedKeyTypeComputer implements IResultTypeComputer {
  
      public static final PutAutogeneratedKeyTypeComputer INSTANCE = new PutAutogeneratedKeyTypeComputer();
  
      private PutAutogeneratedKeyTypeComputer() {
      }
  
      @Override
      public IAType computeType(ILogicalExpression expression, IVariableTypeEnvironment env,
              IMetadataProvider<?, ?> metadataProvider) throws AlgebricksException {
          AbstractFunctionCallExpression f = (AbstractFunctionCallExpression) expression;
          FunctionIdentifier funcId = f.getFunctionIdentifier();
          List<Mutable<ILogicalExpression>> funArgs = f.getArguments();
          IAType inRecArg = (IAType) env.getType(funArgs.get(0).getValue());
          boolean unknownable = TypeHelper.canBeUnknown(inRecArg);
          ARecordType inRecType = TypeComputeUtils.extractRecordType(inRecArg);
          if (inRecType == null) {
              throw new TypeMismatchException(f.getSourceLocation(), funcId, 0, inRecArg.getTypeTag(), ATypeTag.OBJECT);
          }
          String[] keyNameParts = getKeyNameParts(funArgs);
          return computeOutRecType(inRecType, keyNameParts, unknownable, f.getSourceLocation());
      }
  
      private static IAType computeOutRecType(ARecordType inRecType, String[] keyNameParts, boolean unknownable,
              SourceLocation sourceLocation) throws AlgebricksException {
          String[] inFieldNames = inRecType.getFieldNames();
          String[] resultFieldNames = Arrays.copyOf(inFieldNames, inFieldNames.length);
          IAType[] resultFieldTypes = new IAType[resultFieldNames.length];
          int k = 0;
          return computeResultType(resultFieldNames, resultFieldTypes, inRecType, keyNameParts, k, unknownable,
                  sourceLocation);
      }
  
      private static IAType computeResultType(String[] resultFieldNames, IAType[] resultFieldTypes, ARecordType inRecType,
              String[] keyNameParts, int k, boolean unknownable, SourceLocation sourceLocation)
              throws CompilationException {
          boolean keyFound = false;
          for (int i = 0; i < resultFieldNames.length; i++) {
              String fName = resultFieldNames[i];
              IAType fType = inRecType.getFieldType(fName);
              if (keyNameParts[k].equals(fName)) {
                  keyFound = true;
                  resultFieldTypes[i] = computeAutoType(fType, keyNameParts, k, sourceLocation);
              } else {
                  if (fType.getTypeTag() == ATypeTag.OBJECT) {
                      ARecordType nestedType = (ARecordType) fType;
                      // deep copy prevents altering of input types
                      resultFieldTypes[i] = nestedType.deepCopy(nestedType);
                  } else {
                      resultFieldTypes[i] = fType;
                  }
              }
          }
          boolean isOpen = inRecType.isOpen();
          IAType resultType;
          if (!keyFound) {
              resultType = addAutoKeyType(resultFieldNames, resultFieldTypes, isOpen, keyNameParts, k);
          } else {
              resultType = new ARecordType("", resultFieldNames, resultFieldTypes, isOpen);
          }
          if (unknownable) {
              resultType = AUnionType.createUnknownableType(resultType);
          }
          return resultType;
      }
  
      private static IAType addAutoKeyType(String[] resultFieldNames, IAType[] resultFieldTypes, boolean isOpen,
              String[] keyNameParts, int k) {
          IAType computedNestedType = BuiltinType.AUUID;
          int lastPart = keyNameParts.length - 1;
          for (int i = lastPart; i > k; i--) {
              computedNestedType =
                      new ARecordType("", new String[] { keyNameParts[i] }, new IAType[] { computedNestedType }, isOpen);
          }
          String[] finalResultFieldNames = Arrays.copyOf(resultFieldNames, resultFieldNames.length + 1);
          IAType[] finalResultFieldTypes = Arrays.copyOf(resultFieldTypes, resultFieldTypes.length + 1);
          finalResultFieldNames[finalResultFieldNames.length - 1] = keyNameParts[k];
          finalResultFieldTypes[finalResultFieldTypes.length - 1] = computedNestedType;
          return new ARecordType("", finalResultFieldNames, finalResultFieldTypes, isOpen);
      }
  
      private static IAType computeAutoType(IAType keyPartType, String[] keyNameParts, int k,
              SourceLocation sourceLocation) throws CompilationException {
          IAType fType = TypeComputeUtils.getActualType(keyPartType);
          if (k == keyNameParts.length - 1) {
              // reached the final key name part, check it's UUID
              if (fType.getTypeTag() != ATypeTag.UUID) {
                  throw new CompilationException(ErrorCode.CASTING_FIELD, sourceLocation, fType.getTypeTag(),
                          ATypeTag.UUID);
              }
              return fType;
          }
          // keyPartType should be a record because there is more nesting until reaching the final key part
          if (fType.getTypeTag() != ATypeTag.OBJECT) {
              throw new CompilationException(ErrorCode.CASTING_FIELD, sourceLocation, fType.getTypeTag(),
                      ATypeTag.OBJECT);
          }
          boolean unknownable = TypeHelper.canBeUnknown(keyPartType);
          ARecordType recType = (ARecordType) fType;
          String[] inFieldNames = recType.getFieldNames();
          String[] resultFieldNames = Arrays.copyOf(inFieldNames, inFieldNames.length);
          IAType[] resultFieldTypes = new IAType[resultFieldNames.length];
          return computeResultType(resultFieldNames, resultFieldTypes, recType, keyNameParts, k + 1, unknownable,
                  sourceLocation);
      }
  
      private static String[] getKeyNameParts(List<Mutable<ILogicalExpression>> funArgs) {
          String[] keyName = new String[funArgs.size() - 1];
          for (int i = 1, k = 0; i < funArgs.size(); i++, k++) {
              keyName[k] = ConstantExpressionUtil.getStringConstant(funArgs.get(i).getValue());
          }
          return keyName;
      }
  }
  