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.dataflow.data.nontagged.comparators;
  
  import static org.apache.asterix.om.util.container.ObjectFactories.RECORD_FACTORY;
  import static org.apache.asterix.om.util.container.ObjectFactories.VALUE_FACTORY;
  
  import java.io.IOException;
  
  import org.apache.asterix.dataflow.data.common.ListAccessorUtil;
  import org.apache.asterix.dataflow.data.common.TaggedValueReference;
  import org.apache.asterix.dataflow.data.nontagged.serde.ADateSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.ADateTimeSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.ADayTimeDurationSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.ATimeSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.AYearMonthDurationSerializerDeserializer;
  import org.apache.asterix.dataflow.data.nontagged.serde.SerializerDeserializerUtil;
  import org.apache.asterix.om.pointables.nonvisitor.RecordField;
  import org.apache.asterix.om.pointables.nonvisitor.SortedRecord;
  import org.apache.asterix.om.typecomputer.impl.TypeComputeUtils;
  import org.apache.asterix.om.types.ARecordType;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.om.types.AbstractCollectionType;
  import org.apache.asterix.om.types.IAType;
  import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
  import org.apache.asterix.om.types.hierachy.ATypeHierarchy.Domain;
  import org.apache.asterix.om.util.container.ListObjectPool;
  import org.apache.hyracks.api.dataflow.value.IBinaryComparator;
  import org.apache.hyracks.api.exceptions.HyracksDataException;
  import org.apache.hyracks.data.std.accessors.RawBinaryComparatorFactory;
  import org.apache.hyracks.data.std.primitive.BooleanPointable;
  import org.apache.hyracks.data.std.primitive.ByteArrayPointable;
  import org.apache.hyracks.data.std.primitive.UTF8StringPointable;
  
  /**
   * This comparator is an ordering comparator. It deals with MISSING, NULL, and incompatible types different than the
   * logical comparison.
   */
  abstract class AbstractAGenericBinaryComparator implements IBinaryComparator {
  
      protected final IAType leftType;
      protected final IAType rightType;
      private final ListObjectPool<SortedRecord, ARecordType> recordPool = new ListObjectPool<>(RECORD_FACTORY);
      private final ListObjectPool<TaggedValueReference, Void> taggedValueAllocator = new ListObjectPool<>(VALUE_FACTORY);
  
      AbstractAGenericBinaryComparator(IAType leftType, IAType rightType) {
          // factory should have already made sure to get the actual type (and no null types)
          this.leftType = leftType;
          this.rightType = rightType;
      }
  
      protected final int compare(IAType leftType, TaggedValueReference leftValue, IAType rightType,
              TaggedValueReference rightValue) throws HyracksDataException {
          ATypeTag tag1 = leftValue.getTag();
          ATypeTag tag2 = rightValue.getTag();
          if (tag1 == null || tag2 == null) {
              throw new IllegalStateException("Could not recognize the type of data.");
          }
          if (tag1 == ATypeTag.MISSING) {
              return tag2 == ATypeTag.MISSING ? 0 : -1;
          } else if (tag2 == ATypeTag.MISSING) {
              return 1;
          }
          if (tag1 == ATypeTag.NULL) {
              return tag2 == ATypeTag.NULL ? 0 : -1;
          } else if (tag2 == ATypeTag.NULL) {
              return 1;
          }
          byte[] b1 = leftValue.getByteArray();
          int s1 = leftValue.getStartOffset();
          int l1 = leftValue.getLength();
          byte[] b2 = rightValue.getByteArray();
          int s2 = rightValue.getStartOffset();
          int l2 = rightValue.getLength();
  
          if (ATypeHierarchy.isCompatible(tag1, tag2) && ATypeHierarchy.getTypeDomain(tag1) == Domain.NUMERIC) {
              return ComparatorUtil.compareNumbers(tag1, b1, s1, tag2, b2, s2);
          }
          // currently only numbers are compatible. if two tags are not compatible, we compare the tags to generate order
          if (tag1 != tag2) {
              return Byte.compare(tag1.serialize(), tag2.serialize());
          }
  
          switch (tag1) {
              case STRING:
                  return UTF8StringPointable.compare(b1, s1, l1, b2, s2, l2);
              case UUID:
                  return AUUIDPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case BOOLEAN:
                  return BooleanPointable.compare(b1, s1, l1, b2, s2, l2);
              case TIME:
                  return Integer.compare(ATimeSerializerDeserializer.getChronon(b1, s1),
                          ATimeSerializerDeserializer.getChronon(b2, s2));
              case DATE:
                  return Integer.compare(ADateSerializerDeserializer.getChronon(b1, s1),
                          ADateSerializerDeserializer.getChronon(b2, s2));
              case YEARMONTHDURATION:
                  return Integer.compare(AYearMonthDurationSerializerDeserializer.getYearMonth(b1, s1),
                          AYearMonthDurationSerializerDeserializer.getYearMonth(b2, s2));
              case DATETIME:
                  return Long.compare(ADateTimeSerializerDeserializer.getChronon(b1, s1),
                          ADateTimeSerializerDeserializer.getChronon(b2, s2));
              case DAYTIMEDURATION:
                  return Long.compare(ADayTimeDurationSerializerDeserializer.getDayTime(b1, s1),
                          ADayTimeDurationSerializerDeserializer.getDayTime(b2, s2));
              case RECTANGLE:
                  return ARectanglePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case CIRCLE:
                  return ACirclePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case POINT:
                  return APointPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case POINT3D:
                  return APoint3DPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case LINE:
                  return ALinePartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case POLYGON:
                  return APolygonPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case GEOMETRY:
                  return AGeometryPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case DURATION:
                  return ADurationPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
              case INTERVAL:
                  return compareInterval(b1, s1, l1, b2, s2, l2);
              case BINARY:
                  return ByteArrayPointable.compare(b1, s1, l1, b2, s2, l2);
              case ARRAY:
                  return compareArrays(leftType, leftValue, rightType, rightValue);
              case OBJECT:
                  return compareRecords(leftType, b1, s1, rightType, b2, s2);
              default:
                  return RawBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
          }
      }
  
      protected int compareInterval(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
          return AIntervalAscPartialBinaryComparatorFactory.compare(b1, s1, l1, b2, s2, l2);
      }
  
      private int compareArrays(IAType leftArrayType, TaggedValueReference leftArray, IAType rightArrayType,
              TaggedValueReference rightArray) throws HyracksDataException {
          int leftNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(leftArray);
          int rightNumItems = SerializerDeserializerUtil.getNumberOfItemsNonTagged(rightArray);
          IAType leftItemType =
                  ((AbstractCollectionType) TypeComputeUtils.getActualTypeOrOpen(leftArrayType, ATypeTag.ARRAY))
                          .getItemType();
          IAType rightItemType =
                  ((AbstractCollectionType) TypeComputeUtils.getActualTypeOrOpen(rightArrayType, ATypeTag.ARRAY))
                          .getItemType();
          ATypeTag leftArrayItemTag = leftItemType.getTypeTag();
          ATypeTag rightArrayItemTag = rightItemType.getTypeTag();
          TaggedValueReference leftItem = taggedValueAllocator.allocate(null);
          TaggedValueReference rightItem = taggedValueAllocator.allocate(null);
          boolean leftItemHasTag = leftArrayItemTag == ATypeTag.ANY;
          boolean rightItemHasTag = rightArrayItemTag == ATypeTag.ANY;
          int result;
          try {
              for (int i = 0; i < leftNumItems && i < rightNumItems; i++) {
                  ListAccessorUtil.getItemFromList(leftArray, i, leftItem, leftArrayItemTag, leftItemHasTag);
                  ListAccessorUtil.getItemFromList(rightArray, i, rightItem, rightArrayItemTag, rightItemHasTag);
                  result = compare(leftItemType, leftItem, rightItemType, rightItem);
                  if (result != 0) {
                      return result;
                  }
              }
              return Integer.compare(leftNumItems, rightNumItems);
          } catch (IOException e) {
              throw HyracksDataException.create(e);
          } finally {
              taggedValueAllocator.free(rightItem);
              taggedValueAllocator.free(leftItem);
          }
      }
  
      private int compareRecords(IAType leftType, byte[] b1, int s1, IAType rightType, byte[] b2, int s2)
              throws HyracksDataException {
          ARecordType leftRecordType = (ARecordType) TypeComputeUtils.getActualTypeOrOpen(leftType, ATypeTag.OBJECT);
          ARecordType rightRecordType = (ARecordType) TypeComputeUtils.getActualTypeOrOpen(rightType, ATypeTag.OBJECT);
          SortedRecord leftRecord = recordPool.allocate(leftRecordType);
          SortedRecord rightRecord = recordPool.allocate(rightRecordType);
          TaggedValueReference leftFieldValue = taggedValueAllocator.allocate(null);
          TaggedValueReference rightFieldValue = taggedValueAllocator.allocate(null);
          try {
              leftRecord.resetNonTagged(b1, s1);
              rightRecord.resetNonTagged(b2, s2);
              int result;
              while (!leftRecord.isEmpty() && !rightRecord.isEmpty()) {
                  RecordField leftField = leftRecord.poll();
                  RecordField rightField = rightRecord.poll();
                  // compare the names first
                  result = RecordField.FIELD_NAME_COMP.compare(leftField, rightField);
                  if (result != 0) {
                      return result;
                  }
                  // then compare the values if the names are equal
                  leftRecord.getFieldValue(leftField, leftFieldValue);
                  rightRecord.getFieldValue(rightField, rightFieldValue);
                  result = compare(leftRecord.getFieldType(leftField), leftFieldValue,
                          rightRecord.getFieldType(rightField), rightFieldValue);
                  if (result != 0) {
                      return result;
                  }
              }
              return Integer.compare(leftRecord.size(), rightRecord.size());
          } catch (IOException e) {
              throw HyracksDataException.create(e);
          } finally {
              recordPool.free(rightRecord);
              recordPool.free(leftRecord);
              taggedValueAllocator.free(rightFieldValue);
              taggedValueAllocator.free(leftFieldValue);
          }
      }
  }
  