/**
 * Helper method to determine a {@link Join}'s collation assuming that it
 * uses a merge-join algorithm.
 *
 * <p>If the inputs are sorted on other keys <em>in addition to</em> the join
 * key, the result preserves those collations too.
 */
public static List<RelCollation> mergeJoin(
		RelMetadataQuery mq,
		RelNode left, RelNode right,
		ImmutableIntList leftKeys,
		ImmutableIntList rightKeys) {
	final com.google.common.collect.ImmutableList.Builder<RelCollation> builder =
			com.google.common.collect.ImmutableList.builder();

	final com.google.common.collect.ImmutableList<RelCollation> leftCollations = mq.collations(left);
	assert RelCollations.contains(leftCollations, leftKeys)
			: "cannot merge join: left input is not sorted on left keys";
	builder.addAll(leftCollations);

	final com.google.common.collect.ImmutableList<RelCollation> rightCollations = mq.collations(right);
	assert RelCollations.contains(rightCollations, rightKeys)
			: "cannot merge join: right input is not sorted on right keys";
	final int leftFieldCount = left.getRowType().getFieldCount();
	for (RelCollation collation : rightCollations) {
		builder.add(RelCollations.shift(collation, leftFieldCount));
	}
	return builder.build();
}
 

public static void validateCollation(RelOptCluster cluster, RelNode child, ImmutableBitSet groupSet) {
  if (groupSet.isEmpty()) {
    // If no groups, no collation is required
    return;
  }

  final RelCollation requiredCollation = RelCollations.of(
      StreamSupport.stream(groupSet.spliterator(), false).map(RelFieldCollation::new).collect(Collectors.toList()));

  final RelMetadataQuery mq = cluster.getMetadataQuery();
  final List<RelCollation> collations = mq.collations(child);

  for(RelCollation collation: collations) {
    if (collation.satisfies(requiredCollation)) {
      return;
    }
  }

  throw new AssertionError("child collations [" + collations + "] does not match expected collation [" + requiredCollation + "]");
}
 

/** Helper method to determine a {@link Join}'s collation assuming that it
 * uses a merge-join algorithm.
 *
 * <p>If the inputs are sorted on other keys <em>in addition to</em> the join
 * key, the result preserves those collations too. */
public static List<RelCollation> mergeJoin(RelMetadataQuery mq,
    RelNode left, RelNode right,
    ImmutableIntList leftKeys, ImmutableIntList rightKeys, JoinRelType joinType) {
  assert EnumerableMergeJoin.isMergeJoinSupported(joinType)
      : "EnumerableMergeJoin unsupported for join type " + joinType;

  final ImmutableList<RelCollation> leftCollations = mq.collations(left);
  if (!joinType.projectsRight()) {
    return leftCollations;
  }

  final ImmutableList.Builder<RelCollation> builder = ImmutableList.builder();
  builder.addAll(leftCollations);

  final ImmutableList<RelCollation> rightCollations = mq.collations(right);
  final int leftFieldCount = left.getRowType().getFieldCount();
  for (RelCollation collation : rightCollations) {
    builder.add(RelCollations.shift(collation, leftFieldCount));
  }
  return builder.build();
}
 

@Override
public Prel prepareForLateralUnnestPipeline(List<RelNode> children) {
  List<RelFieldCollation> relFieldCollations = Lists.newArrayList();
  relFieldCollations.add(new RelFieldCollation(0,
                        RelFieldCollation.Direction.ASCENDING, RelFieldCollation.NullDirection.FIRST));
  for (RelFieldCollation fieldCollation : this.collation.getFieldCollations()) {
    relFieldCollations.add(new RelFieldCollation(fieldCollation.getFieldIndex() + 1,
            fieldCollation.direction, fieldCollation.nullDirection));
  }

  RelCollation collationTrait = RelCollationImpl.of(relFieldCollations);
  RelTraitSet traits = RelTraitSet.createEmpty()
                                  .replace(this.getTraitSet().getTrait(DrillDistributionTraitDef.INSTANCE))
                                  .replace(collationTrait)
                                  .replace(DRILL_PHYSICAL);
  return transformTopNToSortAndLimit(children, traits, collationTrait);
}
 

@Override
public ResultSet create(ColumnMetaData.AvaticaType elementType,
                        Iterable<Object> iterable) {
  final List<ColumnMetaData> columnMetaDataList;
  if (elementType instanceof ColumnMetaData.StructType) {
    columnMetaDataList = ((ColumnMetaData.StructType) elementType).columns;
  } else {
    columnMetaDataList =
        ImmutableList.of(ColumnMetaData.dummy(elementType, false));
  }
  final CalcitePrepare.CalciteSignature signature =
      (CalcitePrepare.CalciteSignature) this.signature;
  final CalcitePrepare.CalciteSignature<Object> newSignature =
      new CalcitePrepare.CalciteSignature<>(signature.sql,
          signature.parameters, signature.internalParameters,
          signature.rowType, columnMetaDataList, Meta.CursorFactory.ARRAY,
          signature.rootSchema, ImmutableList.<RelCollation>of(), -1, null);
  ResultSetMetaData subResultSetMetaData =
      new AvaticaResultSetMetaData(statement, null, newSignature);
  final QuarkResultSet resultSet =
      new QuarkResultSet(statement, signature, subResultSetMetaData,
          localCalendar.getTimeZone(), new Meta.Frame(0, true, iterable));
  final Cursor cursor = resultSet.createCursor(elementType, iterable);
  return resultSet.execute2(cursor, columnMetaDataList);
}
 

public void rewriteRel(Sort rel) {
    RelCollation oldCollation = rel.getCollation();
    final RelNode oldChild = rel.getInput();
    final RelNode newChild = getNewForOldRel(oldChild);
    final Mappings.TargetMapping mapping = getNewForOldInputMapping(oldChild);

    // validate
    for (RelFieldCollation field : oldCollation.getFieldCollations()) {
        int oldInput = field.getFieldIndex();
        RelDataType sortFieldType = oldChild.getRowType().getFieldList().get(oldInput).getType();
        if (sortFieldType.isStruct()) {
            // TODO jvs 10-Feb-2005
            throw Util.needToImplement("sorting on structured types");
        }
    }
    RelCollation newCollation = RexUtil.apply(mapping, oldCollation);
    Sort newRel = LogicalSort.create(newChild, newCollation, rel.offset, rel.fetch);
    setNewForOldRel(rel, newRel);
}
 

@Override
public RelNode convert(final RelNode rel) {

    //  KYLIN-3281
    //  OLAPProjectRule can't normal working with projectRel[input=sortRel]
    final LogicalProject project = (LogicalProject) rel;
    final RelNode convertChild = convert(project.getInput(),
            project.getInput().getTraitSet().replace(OLAPRel.CONVENTION));
    final RelOptCluster cluster = convertChild.getCluster();
    final RelTraitSet traitSet = cluster.traitSet().replace(OLAPRel.CONVENTION)
            .replaceIfs(RelCollationTraitDef.INSTANCE, new Supplier<List<RelCollation>>() {
                public List<RelCollation> get() {
                    //  CALCITE-88
                    return RelMdCollation.project(cluster.getMetadataQuery(), convertChild, project.getProjects());
                }
            });
    return new OLAPProjectRel(convertChild.getCluster(), traitSet, convertChild, project.getProjects(),
            project.getRowType());
}
 

/**
 * Returns the collation of {@link EnumerableNestedLoopJoin}
 * based on its inputs and the join type.
 */
public static List<RelCollation> enumerableNestedLoopJoin(
		RelMetadataQuery mq,
		RelNode left,
		RelNode right,
		JoinRelType joinType) {
	return enumerableJoin0(mq, left, right, joinType);
}
 

@Override
public SqlMonotonicity getMonotonicity(String columnName) {
    for (RelCollation collation : table.getStatistic().getCollations()) {
        final RelFieldCollation fieldCollation = collation.getFieldCollations().get(0);
        final int fieldIndex = fieldCollation.getFieldIndex();
        if (fieldIndex < rowType.getFieldCount() && rowType.getFieldNames().get(fieldIndex).equals(columnName)) {
            return fieldCollation.direction.monotonicity();
        }
    }
    return SqlMonotonicity.NOT_MONOTONIC;
}
 

/**
 * Creates a LogicalMatch.
 */
public static LogicalMatch create(RelOptCluster cluster,
    RelTraitSet traitSet, RelNode input, RelDataType rowType,
    RexNode pattern, boolean strictStart, boolean strictEnd,
    Map<String, RexNode> patternDefinitions, Map<String, RexNode> measures,
    RexNode after, Map<String, ? extends SortedSet<String>> subsets,
    boolean allRows, ImmutableBitSet partitionKeys, RelCollation orderKeys,
    RexNode interval) {
  return new LogicalMatch(cluster, traitSet, input, rowType, pattern,
      strictStart, strictEnd, patternDefinitions, measures, after, subsets,
      allRows, partitionKeys, orderKeys, interval);
}
 

@Override public Match copy(RelNode input, RelDataType rowType,
    RexNode pattern, boolean strictStart, boolean strictEnd,
    Map<String, RexNode> patternDefinitions, Map<String, RexNode> measures,
    RexNode after, Map<String, ? extends SortedSet<String>> subsets,
    boolean allRows, List<RexNode> partitionKeys, RelCollation orderKeys,
    RexNode interval) {
  final RelTraitSet traitSet = getCluster().traitSetOf(Convention.NONE);
  return new LogicalMatch(getCluster(), traitSet,
      input,
      rowType,
      pattern, strictStart, strictEnd, patternDefinitions, measures,
      after, subsets, allRows, partitionKeys, orderKeys,
      interval);
}
 

@Override
public SortPrel copy(
    RelTraitSet traitSet,
    RelNode newInput,
    RelCollation newCollation,
    RexNode offset,
    RexNode fetch) {
  return SortPrel.create(getCluster(), traitSet, newInput, newCollation);
}
 

@Override
protected void createBroadcastPlan(
    final RelOptRuleCall call,
    final JoinRel join,
    final RexNode joinCondition,
    final RelNode incomingLeft,
    final RelNode incomingRight,
    final RelCollation collationLeft,
    final RelCollation collationRight) throws InvalidRelException {
  throw new UnsupportedOperationException("Not used.");
}
 

/**
 * Copy constructor.
 */
private MockModifiableViewRelOptTable(MockModifiableViewTable modifiableViewTable,
    MockCatalogReader catalogReader, boolean stream, double rowCount,
    List<Map.Entry<String, RelDataType>> columnList, List<Integer> keyList,
    RelDataType rowType, List<RelCollation> collationList, List<String> names,
    Set<String> monotonicColumnSet, StructKind kind, ColumnResolver resolver,
    InitializerExpressionFactory initializerFactory) {
  super(catalogReader, stream, false, rowCount, columnList, keyList,
      rowType, collationList, names,
      monotonicColumnSet, kind, resolver, initializerFactory);
  this.modifiableViewTable = modifiableViewTable;
}
 

@Override public RelNode passThrough(final RelTraitSet required) {
  RelCollation collation = required.getCollation();
  if (collation == null || collation.isDefault()) {
    return null;
  }

  // A Values with 0 or 1 rows can be ordered by any collation.
  if (tuples.size() > 1) {
    Ordering<List<RexLiteral>> ordering = null;
    // Generate ordering comparator according to the required collations.
    for (RelFieldCollation fc : collation.getFieldCollations()) {
      Ordering<List<RexLiteral>> comparator = RelMdCollation.comparator(fc);
      if (ordering == null) {
        ordering = comparator;
      } else {
        ordering = ordering.compound(comparator);
      }
    }
    // Check whether the tuples are sorted by required collations.
    if (!ordering.isOrdered(tuples)) {
      return null;
    }
  }

  // The tuples order satisfies the collation, we just create a new
  // relnode with required collation info.
  return copy(traitSet.replace(collation), ImmutableList.of());
}
 

public CassandraSort(RelOptCluster cluster, RelTraitSet traitSet,
    RelNode child, RelCollation collation) {
  super(cluster, traitSet, child, collation, null, null);

  assert getConvention() == CassandraRel.CONVENTION;
  assert getConvention() == child.getConvention();
}
 

/** Creates a binding of the appropriate type. */
public static RexCallBinding create(RelDataTypeFactory typeFactory,
    RexCall call,
    List<RelCollation> inputCollations) {
  switch (call.getKind()) {
  case CAST:
    return new RexCastCallBinding(typeFactory, call.getOperator(),
        call.getOperands(), call.getType(), inputCollations);
  }
  return new RexCallBinding(typeFactory, call.getOperator(),
      call.getOperands(), inputCollations);
}
 

RexCastCallBinding(
    RelDataTypeFactory typeFactory,
    SqlOperator sqlOperator, List<? extends RexNode> operands,
    RelDataType type,
    List<RelCollation> inputCollations) {
  super(typeFactory, sqlOperator, operands, inputCollations);
  this.type = type;
}
 

/** Creates an AggregateCall. */
public static AggregateCall create(SqlAggFunction aggFunction,
    boolean distinct, boolean approximate, boolean ignoreNulls,
    List<Integer> argList, int filterArg, RelCollation collation,
    RelDataType type, String name) {
  final boolean distinct2 = distinct
      && (aggFunction.getDistinctOptionality() != Optionality.IGNORED);
  return new AggregateCall(aggFunction, distinct2, approximate, ignoreNulls,
      argList, filterArg, collation, type, name);
}
 

@Deprecated // to be removed before 2.0
protected Calc(
    RelOptCluster cluster,
    RelTraitSet traits,
    RelNode child,
    RexProgram program,
    List<RelCollation> collationList) {
  this(cluster, traits, ImmutableList.of(), child, program);
  Util.discard(collationList);
}
 

/**
 * Creates a MutableMatch.
 *
 */
public static MutableMatch of(RelDataType rowType,
    MutableRel input, RexNode pattern, boolean strictStart, boolean strictEnd,
    Map<String, RexNode> patternDefinitions, Map<String, RexNode> measures,
    RexNode after, Map<String, ? extends SortedSet<String>> subsets,
    boolean allRows, ImmutableBitSet partitionKeys, RelCollation orderKeys,
    RexNode interval) {
  return new MutableMatch(rowType, input, pattern, strictStart, strictEnd,
      patternDefinitions, measures, after, subsets, allRows, partitionKeys,
      orderKeys, interval);
}
 

/** Check if it is possible to exploit native CQL sorting for a given collation.
 *
 * @return True if it is possible to achieve this sort in Cassandra
 */
private boolean collationsCompatible(RelCollation sortCollation,
    RelCollation implicitCollation) {
  List<RelFieldCollation> sortFieldCollations = sortCollation.getFieldCollations();
  List<RelFieldCollation> implicitFieldCollations = implicitCollation.getFieldCollations();

  if (sortFieldCollations.size() > implicitFieldCollations.size()) {
    return false;
  }
  if (sortFieldCollations.size() == 0) {
    return true;
  }

  // Check if we need to reverse the order of the implicit collation
  boolean reversed = reverseDirection(sortFieldCollations.get(0).getDirection())
      == implicitFieldCollations.get(0).getDirection();

  for (int i = 0; i < sortFieldCollations.size(); i++) {
    RelFieldCollation sorted = sortFieldCollations.get(i);
    RelFieldCollation implied = implicitFieldCollations.get(i);

    // Check that the fields being sorted match
    if (sorted.getFieldIndex() != implied.getFieldIndex()) {
      return false;
    }

    // Either all fields must be sorted in the same direction
    // or the opposite direction based on whether we decided
    // if the sort direction should be reversed above
    RelFieldCollation.Direction sortDirection = sorted.getDirection();
    RelFieldCollation.Direction implicitDirection = implied.getDirection();
    if ((!reversed && sortDirection != implicitDirection)
        || (reversed && reverseDirection(sortDirection) != implicitDirection)) {
      return false;
    }
  }

  return true;
}
 

public RexCallBinding(
    RelDataTypeFactory typeFactory,
    SqlOperator sqlOperator,
    List<? extends RexNode> operands,
    List<RelCollation> inputCollations) {
  super(typeFactory, sqlOperator);
  this.operands = ImmutableList.copyOf(operands);
  this.inputCollations = ImmutableList.copyOf(inputCollations);
}
 

/**
 * Creates an EnumerableMatch.
 *
 * <p>Use {@link #create} unless you know what you're doing.
 */
public EnumerableMatch(RelOptCluster cluster, RelTraitSet traitSet,
    RelNode input, RelDataType rowType, RexNode pattern,
    boolean strictStart, boolean strictEnd,
    Map<String, RexNode> patternDefinitions, Map<String, RexNode> measures,
    RexNode after, Map<String, ? extends SortedSet<String>> subsets,
    boolean allRows, ImmutableBitSet partitionKeys, RelCollation orderKeys,
    RexNode interval) {
  super(cluster, traitSet, input, rowType, pattern, strictStart, strictEnd,
      patternDefinitions, measures, after, subsets, allRows, partitionKeys,
      orderKeys, interval);
}
 

@Override
public void onMatch(RelOptRuleCall call) {
  PlannerSettings settings = PrelUtil.getPlannerSettings(call.getPlanner());
  final DrillJoinRel join = call.rel(0);
  final RelNode left = join.getLeft();
  final RelNode right = join.getRight();

  if (!checkPreconditions(join, left, right, settings)) {
    return;
  }

  boolean hashSingleKey = PrelUtil.getPlannerSettings(call.getPlanner()).isHashSingleKey();

  try {
    RelCollation collationLeft = getCollation(join.getLeftKeys());
    RelCollation collationRight = getCollation(join.getRightKeys());

    if(isDist){
      createDistBothPlan(call, join, PhysicalJoinType.MERGE_JOIN, left, right, collationLeft, collationRight, hashSingleKey);
    }else{
      if (checkBroadcastConditions(call.getPlanner(), join, left, right)) {
        createBroadcastPlan(call, join, join.getCondition(), PhysicalJoinType.MERGE_JOIN,
            left, right, collationLeft, collationRight);
      }
    }

  } catch (InvalidRelException e) {
    tracer.warn(e.toString());
  }
}
 

/**
 * Creates a SortExchange.
 *
 * @param cluster   Cluster this relational expression belongs to
 * @param traitSet  Trait set
 * @param input     Input relational expression
 * @param distribution Distribution specification
 */
protected SortExchange(RelOptCluster cluster, RelTraitSet traitSet,
    RelNode input, RelDistribution distribution, RelCollation collation) {
  super(cluster, traitSet, input, distribution);
  this.collation = Objects.requireNonNull(collation);

  assert traitSet.containsIfApplicable(collation)
      : "traits=" + traitSet + ", collation=" + collation;
}
 

/**
 * Returns the collation of {@link EnumerableHashJoin} based on its inputs and the join type.
 */
public static List<RelCollation> enumerableHashJoin(RelMetadataQuery mq,
    RelNode left, RelNode right, JoinRelType joinType) {
  if (joinType == JoinRelType.SEMI) {
    return enumerableSemiJoin(mq, left, right);
  } else {
    return enumerableJoin0(mq, left, right, joinType);
  }
}
 

/**
 * Trims the fields of an input relational expression.
 *
 * @param rel        Relational expression
 * @param input      Input relational expression, whose fields to trim
 * @param fieldsUsed Bitmap of fields needed by the consumer
 * @return New relational expression and its field mapping
 */
protected TrimResult trimChild(RelNode rel, RelNode input, final ImmutableBitSet fieldsUsed,
        Set<RelDataTypeField> extraFields) {
    final ImmutableBitSet.Builder fieldsUsedBuilder = fieldsUsed.rebuild();

    // Fields that define the collation cannot be discarded.
    final RelMetadataQuery mq = rel.getCluster().getMetadataQuery();
    final ImmutableList<RelCollation> collations = mq.collations(input);
    for (RelCollation collation : collations) {
        for (RelFieldCollation fieldCollation : collation.getFieldCollations()) {
            fieldsUsedBuilder.set(fieldCollation.getFieldIndex());
        }
    }

    // Correlating variables are a means for other relational expressions to use
    // fields.
    for (final CorrelationId correlation : rel.getVariablesSet()) {
        rel.accept(new CorrelationReferenceFinder() {
            @Override
            protected RexNode handle(RexFieldAccess fieldAccess) {
                final RexCorrelVariable v = (RexCorrelVariable) fieldAccess.getReferenceExpr();
                if (v.getCorrelationId().equals(correlation)) {
                    fieldsUsedBuilder.set(fieldAccess.getField().getIndex());
                }
                return fieldAccess;
            }
        });
    }

    return dispatchTrimFields(input, fieldsUsedBuilder.build(), extraFields);
}
 

public OrderedMuxExchangePrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, RelCollation collation) {
  super(cluster, traits, child);
  this.fieldCollation = collation;
}
 

/**
 * Returns the collation of {@link EnumerableNestedLoopJoin}
 * based on its inputs and the join type.
 */
public static List<RelCollation> enumerableNestedLoopJoin(RelMetadataQuery mq,
    RelNode left, RelNode right, JoinRelType joinType) {
  return enumerableJoin0(mq, left, right, joinType);
}