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.optimizer.rules.util;
  
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import org.apache.asterix.algebra.operators.physical.IntervalMergeJoinPOperator;
  import org.apache.asterix.common.annotations.RangeAnnotation;
  import org.apache.asterix.common.exceptions.CompilationException;
  import org.apache.asterix.common.exceptions.ErrorCode;
  import org.apache.asterix.lang.common.util.FunctionUtil;
  import org.apache.asterix.om.functions.BuiltinFunctions;
  import org.apache.asterix.om.types.ATypeTag;
  import org.apache.asterix.runtime.operators.joins.interval.utils.AfterIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.BeforeIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.CoveredByIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.CoversIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.IIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.OverlappedByIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.OverlappingIntervalJoinUtilFactory;
  import org.apache.asterix.runtime.operators.joins.interval.utils.OverlapsIntervalJoinUtilFactory;
  import org.apache.commons.lang3.mutable.Mutable;
  import org.apache.commons.lang3.mutable.MutableObject;
  import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
  import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
  import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
  import org.apache.hyracks.algebricks.core.algebra.base.IOptimizationContext;
  import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
  import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
  import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
  import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
  import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
  import org.apache.hyracks.algebricks.core.algebra.functions.IFunctionInfo;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.physical.AbstractJoinPOperator;
  import org.apache.hyracks.algebricks.core.algebra.operators.physical.AssignPOperator;
  import org.apache.hyracks.algebricks.core.algebra.properties.IPartitioningProperty.PartitioningType;
  import org.apache.hyracks.algebricks.core.algebra.properties.IntervalColumn;
  import org.apache.hyracks.api.exceptions.IWarningCollector;
  import org.apache.hyracks.api.exceptions.Warning;
  import org.apache.hyracks.dataflow.common.data.partition.range.RangeMap;
  
  public class IntervalJoinUtils {
  
      private static final Map<FunctionIdentifier, FunctionIdentifier> INTERVAL_JOIN_CONDITIONS = new HashMap<>();
  
      static {
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_AFTER, BuiltinFunctions.INTERVAL_BEFORE);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_BEFORE, BuiltinFunctions.INTERVAL_AFTER);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_COVERED_BY, BuiltinFunctions.INTERVAL_COVERS);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_COVERS, BuiltinFunctions.INTERVAL_COVERED_BY);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_OVERLAPPED_BY, BuiltinFunctions.INTERVAL_OVERLAPS);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_OVERLAPPING, BuiltinFunctions.INTERVAL_OVERLAPPING);
          INTERVAL_JOIN_CONDITIONS.put(BuiltinFunctions.INTERVAL_OVERLAPS, BuiltinFunctions.INTERVAL_OVERLAPPED_BY);
      }
  
      protected static boolean tryIntervalJoinAssignment(AbstractBinaryJoinOperator op, IOptimizationContext context,
              ILogicalExpression joinCondition, int left, int right) throws AlgebricksException {
          List<LogicalVariable> sideLeft = new ArrayList<>(1);
          List<LogicalVariable> sideRight = new ArrayList<>(1);
          List<LogicalVariable> varsLeft = op.getInputs().get(left).getValue().getSchema();
          List<LogicalVariable> varsRight = op.getInputs().get(right).getValue().getSchema();
  
          AbstractFunctionCallExpression funcExpr = (AbstractFunctionCallExpression) joinCondition;
          FunctionIdentifier fi = IntervalJoinUtils.isIntervalJoinCondition(funcExpr, varsLeft, varsRight, sideLeft,
                  sideRight, left, right);
          if (fi == null) {
              return false;
          }
  
          // Existing workflow for interval merge join
          RangeAnnotation rangeAnnotation = IntervalJoinUtils.findRangeAnnotation(funcExpr);
          if (rangeAnnotation == null) {
              return false;
          }
  
          //Check RangeMap type
          RangeMap rangeMap = rangeAnnotation.getRangeMap();
          if (rangeMap.getTag(0, 0) != ATypeTag.DATETIME.serialize() && rangeMap.getTag(0, 0) != ATypeTag.DATE.serialize()
                  && rangeMap.getTag(0, 0) != ATypeTag.TIME.serialize()) {
              IWarningCollector warningCollector = context.getWarningCollector();
              if (warningCollector.shouldWarn()) {
                  warningCollector.warn(Warning.of(op.getSourceLocation(),
                          org.apache.hyracks.api.exceptions.ErrorCode.INAPPLICABLE_HINT,
                          "Date, DateTime, and Time are only range hints types supported for interval joins"));
              }
              return false;
          }
          IntervalPartitions intervalPartitions =
                  IntervalJoinUtils.createIntervalPartitions(op, fi, sideLeft, sideRight, rangeMap, context, left, right);
          IntervalJoinUtils.setSortMergeIntervalJoinOp(op, fi, sideLeft, sideRight, context, intervalPartitions);
          return true;
      }
  
      protected static RangeAnnotation findRangeAnnotation(AbstractFunctionCallExpression fexp) {
          return fexp.getAnnotation(RangeAnnotation.class);
      }
  
      protected static void setSortMergeIntervalJoinOp(AbstractBinaryJoinOperator op, FunctionIdentifier fi,
              List<LogicalVariable> sideLeft, List<LogicalVariable> sideRight, IOptimizationContext context,
              IntervalPartitions intervalPartitions) throws CompilationException {
          IIntervalJoinUtilFactory mjcf = createIntervalJoinCheckerFactory(fi, intervalPartitions.getRangeMap());
          op.setPhysicalOperator(new IntervalMergeJoinPOperator(op.getJoinKind(),
                  AbstractJoinPOperator.JoinPartitioningType.BROADCAST, sideLeft, sideRight,
                  context.getPhysicalOptimizationConfig().getMaxFramesForJoin(), mjcf, intervalPartitions));
      }
  
      /**
       * Certain Relations not yet supported as seen below. Will default to regular join.
       * Inserts partition sort key.
       */
      protected static IntervalPartitions createIntervalPartitions(AbstractBinaryJoinOperator op, FunctionIdentifier fi,
              List<LogicalVariable> sideLeft, List<LogicalVariable> sideRight, RangeMap rangeMap,
              IOptimizationContext context, int left, int right) throws AlgebricksException {
  
          List<LogicalVariable> leftPartitionVar = new ArrayList<>(2);
          leftPartitionVar.add(context.newVar());
          leftPartitionVar.add(context.newVar());
          List<LogicalVariable> rightPartitionVar = new ArrayList<>(2);
          rightPartitionVar.add(context.newVar());
          rightPartitionVar.add(context.newVar());
  
          insertPartitionSortKey(op, left, leftPartitionVar, sideLeft.get(0), context);
          insertPartitionSortKey(op, right, rightPartitionVar, sideRight.get(0), context);
  
          List<IntervalColumn> leftIC = Collections.singletonList(new IntervalColumn(leftPartitionVar.get(0),
                  leftPartitionVar.get(1), OrderOperator.IOrder.OrderKind.ASC));
          List<IntervalColumn> rightIC = Collections.singletonList(new IntervalColumn(rightPartitionVar.get(0),
                  rightPartitionVar.get(1), OrderOperator.IOrder.OrderKind.ASC));
  
          //Set Partitioning Types
          PartitioningType leftPartitioningType = PartitioningType.ORDERED_PARTITIONED;
          PartitioningType rightPartitioningType = PartitioningType.ORDERED_PARTITIONED;
          if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPPED_BY)) {
              rightPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPS)) {
              leftPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPPING)) {
              leftPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
              rightPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_COVERS)) {
              leftPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_COVERED_BY)) {
              rightPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_INTERSECT;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_BEFORE)) {
              leftPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_FOLLOWING;
          } else if (fi.equals(BuiltinFunctions.INTERVAL_AFTER)) {
              rightPartitioningType = PartitioningType.PARTIAL_BROADCAST_ORDERED_FOLLOWING;
          } else {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE, fi.getName());
          }
          return new IntervalPartitions(rangeMap, leftIC, rightIC, leftPartitioningType, rightPartitioningType);
      }
  
      protected static FunctionIdentifier isIntervalJoinCondition(ILogicalExpression e,
              Collection<LogicalVariable> inLeftAll, Collection<LogicalVariable> inRightAll,
              Collection<LogicalVariable> outLeftFields, Collection<LogicalVariable> outRightFields, int left,
              int right) {
          FunctionIdentifier fiReturn;
          boolean switchArguments = false;
          if (e.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
              return null;
          }
          AbstractFunctionCallExpression fexp = (AbstractFunctionCallExpression) e;
          FunctionIdentifier fi = fexp.getFunctionIdentifier();
          if (isIntervalFunction(fi)) {
              fiReturn = fi;
          } else {
              return null;
          }
          ILogicalExpression opLeft = fexp.getArguments().get(left).getValue();
          ILogicalExpression opRight = fexp.getArguments().get(right).getValue();
          if (opLeft.getExpressionTag() != LogicalExpressionTag.VARIABLE
                  || opRight.getExpressionTag() != LogicalExpressionTag.VARIABLE) {
              return null;
          }
          LogicalVariable var1 = ((VariableReferenceExpression) opLeft).getVariableReference();
          if (inLeftAll.contains(var1) && !outLeftFields.contains(var1)) {
              outLeftFields.add(var1);
          } else if (inRightAll.contains(var1) && !outRightFields.contains(var1)) {
              outRightFields.add(var1);
              fiReturn = getInverseIntervalFunction(fi);
              switchArguments = true;
          } else {
              return null;
          }
          LogicalVariable var2 = ((VariableReferenceExpression) opRight).getVariableReference();
          if (inLeftAll.contains(var2) && !outLeftFields.contains(var2) && switchArguments) {
              outLeftFields.add(var2);
          } else if (inRightAll.contains(var2) && !outRightFields.contains(var2) && !switchArguments) {
              outRightFields.add(var2);
          } else {
              return null;
          }
          return fiReturn;
      }
  
      /**
       * Certain Relations not yet supported as seen below. Will default to regular join.
       */
      private static IIntervalJoinUtilFactory createIntervalJoinCheckerFactory(FunctionIdentifier fi, RangeMap rangeMap)
              throws CompilationException {
          IIntervalJoinUtilFactory mjcf;
          if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPPED_BY)) {
              mjcf = new OverlappedByIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPS)) {
              mjcf = new OverlapsIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_COVERS)) {
              mjcf = new CoversIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_COVERED_BY)) {
              mjcf = new CoveredByIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_BEFORE)) {
              mjcf = new BeforeIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_AFTER)) {
              mjcf = new AfterIntervalJoinUtilFactory();
          } else if (fi.equals(BuiltinFunctions.INTERVAL_OVERLAPPING)) {
              mjcf = new OverlappingIntervalJoinUtilFactory(rangeMap);
          } else {
              throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE, fi.getName());
          }
          return mjcf;
      }
  
      private static boolean isIntervalFunction(FunctionIdentifier fi) {
          return INTERVAL_JOIN_CONDITIONS.containsKey(fi);
      }
  
      private static FunctionIdentifier getInverseIntervalFunction(FunctionIdentifier fi) {
          return INTERVAL_JOIN_CONDITIONS.get(fi);
      }
  
      private static void insertPartitionSortKey(AbstractBinaryJoinOperator op, int branch,
              List<LogicalVariable> partitionVars, LogicalVariable intervalVar, IOptimizationContext context)
              throws AlgebricksException {
          List<Mutable<ILogicalExpression>> assignExps = new ArrayList<>();
          // Start partition
          VariableReferenceExpression intervalVarRef1 = new VariableReferenceExpression(intervalVar);
          intervalVarRef1.setSourceLocation(op.getSourceLocation());
          IFunctionInfo startFi = FunctionUtil.getFunctionInfo(BuiltinFunctions.ACCESSOR_TEMPORAL_INTERVAL_START);
          ScalarFunctionCallExpression startPartitionExp =
                  new ScalarFunctionCallExpression(startFi, new MutableObject<>(intervalVarRef1));
          startPartitionExp.setSourceLocation(op.getSourceLocation());
          assignExps.add(new MutableObject<>(startPartitionExp));
          // End partition
          VariableReferenceExpression intervalVarRef2 = new VariableReferenceExpression(intervalVar);
          intervalVarRef2.setSourceLocation(op.getSourceLocation());
          IFunctionInfo endFi = FunctionUtil.getFunctionInfo(BuiltinFunctions.ACCESSOR_TEMPORAL_INTERVAL_END);
          ScalarFunctionCallExpression endPartitionExp =
                  new ScalarFunctionCallExpression(endFi, new MutableObject<>(intervalVarRef2));
          endPartitionExp.setSourceLocation(op.getSourceLocation());
          assignExps.add(new MutableObject<>(endPartitionExp));
  
          AssignOperator ao = new AssignOperator(partitionVars, assignExps);
          ao.setSourceLocation(op.getSourceLocation());
          ao.setExecutionMode(op.getExecutionMode());
          AssignPOperator apo = new AssignPOperator();
          ao.setPhysicalOperator(apo);
          Mutable<ILogicalOperator> aoRef = new MutableObject<>(ao);
          ao.getInputs().add(op.getInputs().get(branch));
          op.getInputs().set(branch, aoRef);
  
          context.computeAndSetTypeEnvironmentForOperator(ao);
          ao.recomputeSchema();
      }
  }
  