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.formats.nontagged;
  
  import java.io.Serializable;
  
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.common.exceptions.RuntimeDataException;
  import org.apache.asterix.dataflow.data.nontagged.comparators.ACirclePartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.ADurationPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AGenericAscBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AGenericDescBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AGeometryPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AIntervalAscPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AIntervalDescPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.ALinePartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.APoint3DPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.APointPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.APolygonPartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.ARectanglePartialBinaryComparatorFactory;
  import org.apache.asterix.dataflow.data.nontagged.comparators.AUUIDPartialBinaryComparatorFactory;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.BuiltinType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
  import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.BooleanBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.ByteArrayBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.ByteBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.DoubleBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.FloatBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.IntegerBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.LongBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.RawBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.ShortBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.UTF8StringBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.UTF8StringLowercaseBinaryComparatorFactory;
  import org.apache.hyracks.data.std.accessors.UTF8StringLowercaseTokenBinaryComparatorFactory;
  
  public class BinaryComparatorFactoryProvider implements IBinaryComparatorFactoryProvider, Serializable {
  
      private static final long serialVersionUID = 1L;
      public static final BinaryComparatorFactoryProvider INSTANCE = new BinaryComparatorFactoryProvider();
      public static final IBinaryComparatorFactory BYTE_POINTABLE_INSTANCE = ByteBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory SHORT_POINTABLE_INSTANCE = ShortBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory INTEGER_POINTABLE_INSTANCE = IntegerBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory LONG_POINTABLE_INSTANCE = LongBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory FLOAT_POINTABLE_INSTANCE = FloatBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory DOUBLE_POINTABLE_INSTANCE = DoubleBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory UTF8STRING_POINTABLE_INSTANCE =
              UTF8StringBinaryComparatorFactory.INSTANCE;
      // Equivalent to UTF8STRING_POINTABLE_INSTANCE but all characters are considered lower case to implement
      // case-insensitive comparisons.
      public static final IBinaryComparatorFactory UTF8STRING_LOWERCASE_POINTABLE_INSTANCE =
              UTF8StringLowercaseBinaryComparatorFactory.INSTANCE;
      // Equivalent to UTF8STRING_LOWERCASE_POINTABLE_INSTANCE but the length information is kept separately,
      // rather than keeping them in the beginning of a string. It is especially useful for the string tokens
      public static final IBinaryComparatorFactory UTF8STRING_LOWERCASE_TOKEN_POINTABLE_INSTANCE =
              UTF8StringLowercaseTokenBinaryComparatorFactory.INSTANCE;
      public static final IBinaryComparatorFactory BINARY_POINTABLE_INSTANCE = ByteArrayBinaryComparatorFactory.INSTANCE;
  
      private BinaryComparatorFactoryProvider() {
      }
  
      // This method add the option of ignoring the case in string comparisons.
      // TODO: We should incorporate this option more nicely, but I'd have to change algebricks.
      @Override
      public IBinaryComparatorFactory getBinaryComparatorFactory(Object type, boolean ascending, boolean ignoreCase) {
          return getBinaryComparatorFactory(type, type, ascending, ignoreCase);
      }
  
      @Override
      public IBinaryComparatorFactory getBinaryComparatorFactory(Object type, boolean ascending) {
          return getBinaryComparatorFactory(type, type, ascending);
      }
  
      @Override
      public IBinaryComparatorFactory getBinaryComparatorFactory(Object leftType, Object rightType, boolean ascending,
              boolean ignoreCase) {
          IAType left = (IAType) leftType;
          IAType right = (IAType) rightType;
          // TODO(ali): what if someone passed ignoreCase=true and type ANY (at runtime it could be a string)?
          if (left.getTypeTag() == ATypeTag.STRING && right.getTypeTag() == ATypeTag.STRING && ignoreCase) {
              return addOffset(UTF8STRING_LOWERCASE_POINTABLE_INSTANCE, ascending);
          }
          if (isShortWithoutTag(left, right)) {
              return SHORT_POINTABLE_INSTANCE;
          }
          return createGenericBinaryComparatorFactory(left, right, ascending);
      }
  
      @Override
      public IBinaryComparatorFactory getBinaryComparatorFactory(Object leftType, Object rightType, boolean ascending) {
          if (isShortWithoutTag((IAType) leftType, (IAType) rightType)) {
              return SHORT_POINTABLE_INSTANCE;
          }
          // During a comparison, since proper type promotion among several numeric types are required,
          // we will use AGenericAscBinaryComparatorFactory, instead of using a specific comparator
          return createGenericBinaryComparatorFactory((IAType) leftType, (IAType) rightType, ascending);
      }
  
      public IBinaryComparatorFactory getBinaryComparatorFactory(ATypeTag type, boolean ascending)
              throws RuntimeDataException {
          switch (type) {
              case ANY:
                  return createGenericBinaryComparatorFactory(BuiltinType.ANY, BuiltinType.ANY, ascending);
              case UNION:
                  throw new RuntimeDataException(ErrorCode.TYPE_UNSUPPORTED, "Comparator", type);
              case NULL:
              case MISSING:
                  return new AnyBinaryComparatorFactory();
              case BOOLEAN:
                  return addOffset(BooleanBinaryComparatorFactory.INSTANCE, ascending);
              case TINYINT:
                  return addOffset(BYTE_POINTABLE_INSTANCE, ascending);
              case SMALLINT:
                  return addOffset(SHORT_POINTABLE_INSTANCE, ascending);
              case DATE:
              case TIME:
              case YEARMONTHDURATION:
              case INTEGER:
                  return addOffset(INTEGER_POINTABLE_INSTANCE, ascending);
              case DATETIME:
              case DAYTIMEDURATION:
              case BIGINT:
                  return addOffset(LONG_POINTABLE_INSTANCE, ascending);
              case FLOAT:
                  return addOffset(FLOAT_POINTABLE_INSTANCE, ascending);
              case DOUBLE:
                  return addOffset(DOUBLE_POINTABLE_INSTANCE, ascending);
              case STRING:
                  return addOffset(UTF8STRING_POINTABLE_INSTANCE, ascending);
              case RECTANGLE:
                  return addOffset(ARectanglePartialBinaryComparatorFactory.INSTANCE, ascending);
              case CIRCLE:
                  return addOffset(ACirclePartialBinaryComparatorFactory.INSTANCE, ascending);
              case POINT:
                  return addOffset(APointPartialBinaryComparatorFactory.INSTANCE, ascending);
              case POINT3D:
                  return addOffset(APoint3DPartialBinaryComparatorFactory.INSTANCE, ascending);
              case LINE:
                  return addOffset(ALinePartialBinaryComparatorFactory.INSTANCE, ascending);
              case POLYGON:
                  return addOffset(APolygonPartialBinaryComparatorFactory.INSTANCE, ascending);
              case GEOMETRY:
                  return addOffset(AGeometryPartialBinaryComparatorFactory.INSTANCE, ascending);
              case DURATION:
                  return addOffset(ADurationPartialBinaryComparatorFactory.INSTANCE, ascending);
              case INTERVAL:
                  return addOffset(intervalBinaryComparatorFactory(ascending), ascending);
              case UUID:
                  return addOffset(AUUIDPartialBinaryComparatorFactory.INSTANCE, ascending);
              case BINARY:
                  return addOffset(BINARY_POINTABLE_INSTANCE, ascending);
              default:
                  return addOffset(RawBinaryComparatorFactory.INSTANCE, ascending);
          }
      }
  
      private IBinaryComparatorFactory addOffset(final IBinaryComparatorFactory inst, final boolean ascending) {
          return new OrderedBinaryComparatorFactory(inst, ascending);
      }
  
      private IBinaryComparatorFactory createGenericBinaryComparatorFactory(IAType leftType, IAType rightType,
              boolean ascending) {
          if (ascending) {
              return new AGenericAscBinaryComparatorFactory(leftType, rightType);
          } else {
              return new AGenericDescBinaryComparatorFactory(leftType, rightType);
          }
      }
  
      private IBinaryComparatorFactory intervalBinaryComparatorFactory(boolean ascending) {
          // Intervals have separate binary comparator factories, since asc is primarily based on start point
          // and desc is similarly based on end point.
          if (ascending) {
              return AIntervalAscPartialBinaryComparatorFactory.INSTANCE;
          } else {
              return AIntervalDescPartialBinaryComparatorFactory.INSTANCE;
          }
      }
  
      private static boolean isShortWithoutTag(IAType left, IAType right) {
          ATypeTag leftTag = left.getTypeTag();
          ATypeTag rightTag = right.getTypeTag();
          if (leftTag != ATypeTag.SHORTWITHOUTTYPEINFO && rightTag != ATypeTag.SHORTWITHOUTTYPEINFO) {
              return false;
          }
          if (leftTag != rightTag) {
              // this should not happen (i.e. comparing untagged (short without tag) vs some tagged)
              throw new IllegalStateException();
          }
          return true;
      }
  }
  