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.comparisons;
  
  import static org.apache.asterix.om.types.ATypeTag.VALUE_TYPE_MAPPING;
  
  import java.io.DataOutput;
  
  import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator;
  import org.apache.asterix.dataflow.data.common.ILogicalBinaryComparator.Result;
  import org.apache.asterix.dataflow.data.common.TaggedValueReference;
  import org.apache.asterix.dataflow.data.nontagged.comparators.ComparatorUtil;
  import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AInt16SerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AInt8SerializerDeserializer;
  import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
  import org.apache.asterix.om.base.ADouble;
  import org.apache.asterix.om.base.AFloat;
  import org.apache.asterix.om.base.AInt16;
  import org.apache.asterix.om.base.AInt32;
  import org.apache.asterix.om.base.AInt64;
  import org.apache.asterix.om.base.AInt8;
  import org.apache.asterix.om.base.AMissing;
  import org.apache.asterix.om.base.ANull;
  import org.apache.asterix.om.base.IAObject;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.EnumDeserializer;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.runtime.evaluators.functions.PointableHelper;
  import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
  import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
  import org.apache.hyracks.algebricks.runtime.evaluators.ConstantEvalFactory;
  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;
  
  public abstract class AbstractComparisonEvaluator implements IScalarEvaluator {
  
      @SuppressWarnings("unchecked")
      protected final ISerializerDeserializer<AMissing> missingSerde =
              SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AMISSING);
      @SuppressWarnings("unchecked")
      protected final ISerializerDeserializer<ANull> nullSerde =
              SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.ANULL);
      protected final ArrayBackedValueStorage resultStorage = new ArrayBackedValueStorage();
      protected final DataOutput out = resultStorage.getDataOutput();
      protected final TaggedValuePointable argLeft = TaggedValuePointable.FACTORY.createPointable();
      protected final TaggedValuePointable argRight = TaggedValuePointable.FACTORY.createPointable();
      protected final SourceLocation sourceLoc;
      protected final IEvaluatorContext ctx;
      private final TaggedValueReference leftVal = new TaggedValueReference();
      private final TaggedValueReference rightVal = new TaggedValueReference();
      private final IScalarEvaluator evalLeft;
      private final IScalarEvaluator evalRight;
      private final ILogicalBinaryComparator logicalComparator;
      private final IAObject leftConstant;
      private final IAObject rightConstant;
      private final boolean checkUnknown;
  
      AbstractComparisonEvaluator(IScalarEvaluatorFactory evalLeftFactory, IAType leftType,
              IScalarEvaluatorFactory evalRightFactory, IAType rightType, IEvaluatorContext ctx, SourceLocation sourceLoc,
              boolean isEquality, boolean checkUnknown) throws HyracksDataException {
          this.ctx = ctx;
          this.checkUnknown = checkUnknown;
          this.evalLeft = evalLeftFactory.createScalarEvaluator(ctx);
          this.evalRight = evalRightFactory.createScalarEvaluator(ctx);
          this.sourceLoc = sourceLoc;
          logicalComparator = ComparatorUtil.createLogicalComparator(leftType, rightType, isEquality);
          leftConstant = getValueOfConstantEval(evalLeftFactory);
          rightConstant = getValueOfConstantEval(evalRightFactory);
      }
  
      private static IAObject getValueOfConstantEval(IScalarEvaluatorFactory factory) {
          return factory instanceof ConstantEvalFactory ? getConstant(((ConstantEvalFactory) factory).getValue()) : null;
      }
  
      @Override
      public void evaluate(IFrameTupleReference tuple, IPointable result) throws HyracksDataException {
          // Evaluates input args.
          evalLeft.evaluate(tuple, argLeft);
          evalRight.evaluate(tuple, argRight);
  
          if (checkUnknown && PointableHelper.checkAndSetMissingOrNull(result, argLeft, argRight)) {
              return;
          }
          leftVal.set(argLeft.getByteArray(), argLeft.getStartOffset() + 1, argLeft.getLength() - 1,
                  VALUE_TYPE_MAPPING[argLeft.getTag()]);
          rightVal.set(argRight.getByteArray(), argRight.getStartOffset() + 1, argRight.getLength() - 1,
                  VALUE_TYPE_MAPPING[argRight.getTag()]);
          evaluateImpl(result);
      }
  
      protected abstract void evaluateImpl(IPointable result) throws HyracksDataException;
  
      final Result compare() throws HyracksDataException {
          if (leftConstant != null) {
              return rightConstant != null ? logicalComparator.compare(leftConstant, rightConstant)
                      : logicalComparator.compare(leftConstant, rightVal);
          } else {
              return rightConstant != null ? logicalComparator.compare(leftVal, rightConstant)
                      : logicalComparator.compare(leftVal, rightVal);
          }
      }
  
      final void writeMissing(IPointable result) throws HyracksDataException {
          resultStorage.reset();
          missingSerde.serialize(AMissing.MISSING, out);
          result.set(resultStorage);
      }
  
      final void writeNull(IPointable result) throws HyracksDataException {
          resultStorage.reset();
          nullSerde.serialize(ANull.NULL, out);
          result.set(resultStorage);
      }
  
      private static IAObject getConstant(byte[] bytes) {
          int start = 0;
          ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(bytes[start]);
          if (typeTag == null) {
              return null;
          }
          start++;
          switch (typeTag) {
              case TINYINT:
                  return new AInt8(AInt8SerializerDeserializer.getByte(bytes, start));
              case SMALLINT:
                  return new AInt16(AInt16SerializerDeserializer.getShort(bytes, start));
              case INTEGER:
                  return new AInt32(AInt32SerializerDeserializer.getInt(bytes, start));
              case BIGINT:
                  return new AInt64(AInt64SerializerDeserializer.getLong(bytes, start));
              case FLOAT:
                  return new AFloat(AFloatSerializerDeserializer.getFloat(bytes, start));
              case DOUBLE:
                  return new ADouble(ADoubleSerializerDeserializer.getDouble(bytes, start));
              default:
                  return null;
          }
      }
  }
  