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;
  
  import static org.apache.asterix.om.types.EnumDeserializer.ATYPETAGDESERIALIZER;
  
  import org.apache.asterix.builders.OrderedListBuilder;
  import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
  import org.apache.asterix.om.base.AMutableDouble;
  import org.apache.asterix.om.base.AMutableInt64;
  import org.apache.asterix.om.typecomputer.impl.ArrayRangeTypeComputer;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
  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.ISerializerDeserializer;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.api.exceptions.SourceLocation;
  import org.apache.hyracks.data.std.api.IPointable;
  import org.apache.hyracks.data.std.primitive.TaggedValuePointable;
  import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
  import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
  
  /**
   * <pre>
   * array_range(start_num, end_num, step_num?) returns a new ordered list, list of long items or double items
   * depending on the supplied arguments. One floating-point arg will make it a list of double items. step_num is optional
   * where the default is 1. It returns an empty list for arguments like:
   * array_range(2, 20, -2), array_range(10, 3, 4) and array_range(1,6,0) where it cannot determine a proper sequence.
   *
   * It throws an error at compile time if the number of arguments < 2 or > 3
   *
   * It returns in order:
   * 1. missing, if any argument is missing.
   * 2. null, if any argument is null or they are not numeric or they are NaN +-INF.
   * 3. otherwise, a new list.
   *
   * </pre>
  */
  
  public class ArrayRangeEvaluator extends AbstractScalarEval {
      private final OrderedListBuilder listBuilder;
      private final ArrayBackedValueStorage storage;
      private final IScalarEvaluator startNumEval;
      private final TaggedValuePointable start;
      private final IScalarEvaluator endNumEval;
      private final TaggedValuePointable end;
      private final AMutableDouble aDouble;
      private final AMutableInt64 aLong;
      private IScalarEvaluator stepNumEval;
      private TaggedValuePointable step;
  
      ArrayRangeEvaluator(IScalarEvaluatorFactory[] args, IEvaluatorContext ctx, SourceLocation sourceLocation,
              FunctionIdentifier functionIdentifier) throws HyracksDataException {
          super(sourceLocation, functionIdentifier);
          storage = new ArrayBackedValueStorage();
          start = new TaggedValuePointable();
          end = new TaggedValuePointable();
          startNumEval = args[0].createScalarEvaluator(ctx);
          endNumEval = args[1].createScalarEvaluator(ctx);
          listBuilder = new OrderedListBuilder();
          aDouble = new AMutableDouble(0);
          aLong = new AMutableInt64(0);
          if (args.length == 3) {
              stepNumEval = args[2].createScalarEvaluator(ctx);
              step = new TaggedValuePointable();
          }
      }
  
      @Override
      public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
          startNumEval.evaluate(tuple, start);
          endNumEval.evaluate(tuple, end);
          if (stepNumEval != null) {
              stepNumEval.evaluate(tuple, step);
          }
  
          if (PointableHelper.checkAndSetMissingOrNull(result, start, end, step)) {
              return;
          }
  
          String n = funID.getName();
          ATypeTag startTag = ATYPETAGDESERIALIZER.deserialize(start.getTag());
          ATypeTag endTag = ATYPETAGDESERIALIZER.deserialize(end.getTag());
          ATypeTag stepTag = ATypeTag.INTEGER;
          double stepNum = 1;
          if (stepNumEval != null) {
              stepNumEval.evaluate(tuple, step);
              stepTag = ATYPETAGDESERIALIZER.deserialize(step.getTag());
              if (!ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, stepTag)) {
                  PointableHelper.setNull(result);
                  return;
              }
              stepNum = ATypeHierarchy.getDoubleValue(n, 2, step.getByteArray(), step.getStartOffset());
          }
  
          if (!ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, startTag) || Double.isNaN(stepNum)
                  || !ATypeHierarchy.isCompatible(ATypeTag.DOUBLE, endTag) || Double.isInfinite(stepNum)) {
              PointableHelper.setNull(result);
              return;
          }
  
          ISerializerDeserializer serde;
          if (ATypeHierarchy.canPromote(startTag, ATypeTag.BIGINT) && ATypeHierarchy.canPromote(endTag, ATypeTag.BIGINT)
                  && ATypeHierarchy.canPromote(stepTag, ATypeTag.BIGINT)) {
              // all 3 numbers are whole numbers
              serde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT64);
              long startNum = ATypeHierarchy.getLongValue(n, 0, start.getByteArray(), start.getStartOffset());
              long endNum = ATypeHierarchy.getLongValue(n, 1, end.getByteArray(), end.getStartOffset());
              listBuilder.reset(ArrayRangeTypeComputer.LONG_LIST);
              while ((startNum < endNum && stepNum > 0) || (startNum > endNum && stepNum < 0)) {
                  aLong.setValue(startNum);
                  storage.reset();
                  serde.serialize(aLong, storage.getDataOutput());
                  listBuilder.addItem(storage);
                  startNum += stepNum;
              }
          } else {
              // one number is a floating-point number
              serde = SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ADOUBLE);
              double startNum = ATypeHierarchy.getDoubleValue(n, 0, start.getByteArray(), start.getStartOffset());
              double endNum = ATypeHierarchy.getDoubleValue(n, 1, end.getByteArray(), end.getStartOffset());
              if (Double.isNaN(startNum) || Double.isInfinite(startNum) || Double.isNaN(endNum)
                      || Double.isInfinite(endNum)) {
                  PointableHelper.setNull(result);
                  return;
              }
              listBuilder.reset(ArrayRangeTypeComputer.DOUBLE_LIST);
              while ((startNum < endNum && stepNum > 0) || (startNum > endNum && stepNum < 0)) {
                  aDouble.setValue(startNum);
                  storage.reset();
                  serde.serialize(aDouble, storage.getDataOutput());
                  listBuilder.addItem(storage);
                  startNum += stepNum;
              }
          }
  
          storage.reset();
          listBuilder.write(storage.getDataOutput(), true);
          result.set(storage);
      }
  }
  