TestHistogram zipfan dataset as well the interfaces of the sort based…

asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/AbstractSortMergeJoinPOperator.java

@@ -0,0 +1,81 @@
+/*
+ * 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.algebra.operators.physical;
+
+import java.util.List;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+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.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator.JoinKind;
+import org.apache.hyracks.algebricks.core.algebra.operators.physical.AbstractJoinPOperator;
+import org.apache.hyracks.algebricks.core.algebra.properties.IPhysicalPropertiesVector;
+import org.apache.hyracks.algebricks.core.algebra.properties.PhysicalRequirements;
+
+public abstract class AbstractSortMergeJoinPOperator extends AbstractJoinPOperator {
+    // Current for Single column Band, Theta join, will be extended onto multiple columns as well as Metric and Skyline join.
+
+    protected List<LogicalVariable> keysLeftTopPartition;
+    protected List<LogicalVariable> keysRightTopPartition;
+    protected Pair<ILogicalExpression, ILogicalExpression> limitRange;
+    protected ILogicalExpression partitionGranularity;
+
+    public AbstractSortMergeJoinPOperator(JoinKind kind, JoinPartitioningType partitioningType,
+            List<LogicalVariable> sideLeft, List<LogicalVariable> sideRight,
+            Pair<ILogicalExpression, ILogicalExpression> range, ILogicalExpression gran) {
+        super(kind, partitioningType);
+        this.keysLeftTopPartition = sideLeft;
+        this.keysRightTopPartition = sideRight;
+        this.limitRange = range;
+        this.partitionGranularity = gran;
+    }
+
+    public List<LogicalVariable> getKeysLeftTopPartition() {
+        return keysLeftTopPartition;
+    }
+
+    public List<LogicalVariable> getKeysRightTopPartition() {
+        return keysRightTopPartition;
+    }
+
+    public Pair<ILogicalExpression, ILogicalExpression> getLimitRange() {
+        return limitRange;
+    }
+
+    public ILogicalExpression getGranularity() {
+        return partitionGranularity;
+    }
+
+    @Override
+    public PhysicalRequirements getRequiredPropertiesForChildren(ILogicalOperator op,
+            IPhysicalPropertiesVector reqdByParent, IOptimizationContext context) {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+    @Override
+    public void computeDeliveredProperties(ILogicalOperator op, IOptimizationContext context)
+            throws AlgebricksException {
+        // TODO Auto-generated method stub
+
+    }
+}

asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/BandSortMergeJoinPOperator.java

@@ -0,0 +1,82 @@
+/*
+ * 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.algebra.operators.physical;
+
+import java.util.List;
+import java.util.logging.Logger;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder;
+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.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator.JoinKind;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
+import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
+
+public class BandSortMergeJoinPOperator extends AbstractSortMergeJoinPOperator {
+    private final int memSizeInFrames;
+    private final int maxInputBuildSizeInFrames;
+    private final int slidingWindowSizeInFrames;
+    private final int aveRecordsPerFrame;
+    private final double fudgeFactor;
+
+    private static final Logger LOGGER = Logger.getLogger(BandSortMergeJoinPOperator.class.getName());
+
+    public BandSortMergeJoinPOperator(JoinKind kind, JoinPartitioningType partitioningType,
+            List<LogicalVariable> sideLeft, List<LogicalVariable> sideRight,
+            Pair<ILogicalExpression, ILogicalExpression> range, ILogicalExpression gran, int memSizeInFrames,
+            int maxInputBuildSizeInFrames, int slidingWindowSizeInFrames, int aveRecordsPerFrame, double fudgeFactor) {
+        super(kind, partitioningType, sideLeft, sideRight, range, gran);
+        this.memSizeInFrames = memSizeInFrames;
+        this.maxInputBuildSizeInFrames = maxInputBuildSizeInFrames;
+        this.slidingWindowSizeInFrames = slidingWindowSizeInFrames;
+        this.aveRecordsPerFrame = aveRecordsPerFrame;
+        this.fudgeFactor = fudgeFactor;
+
+        LOGGER.fine("BandSortMergeJoinPOperator constructed with: JoinKind: " + kind + " JoinPartitioningType="
+                + partitioningType + " List<LogicalVariable>=" + sideLeft + " List<LogicalVariable=" + sideRight
+                + " Pair<ILogicalExpression,ILogicalExpression>=" + range + " ILogicalExpression=" + gran
+                + " memSizeInFrames=" + memSizeInFrames + " maxInputBuildSizeInFrames=" + maxInputBuildSizeInFrames
+                + " slidingWindowSizeInFrames=" + slidingWindowSizeInFrames + " aveRecordsPerFrame="
+                + aveRecordsPerFrame + " fudgeFactor=" + fudgeFactor);
+    }
+
+    @Override
+    public PhysicalOperatorTag getOperatorTag() {
+        // TODO Auto-generated method stub
+        return PhysicalOperatorTag.BAND_SORTMERGE_JOIN;
+    }
+
+    @Override
+    public boolean isMicroOperator() {
+        // TODO Auto-generated method stub
+        return false;
+    }
+
+    @Override
+    public void contributeRuntimeOperator(IHyracksJobBuilder builder, JobGenContext context, ILogicalOperator op,
+            IOperatorSchema propagatedSchema, IOperatorSchema[] inputSchemas, IOperatorSchema outerPlanSchema)
+            throws AlgebricksException {
+        // TODO Auto-generated method stub
+
+    }
+}

asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/base/AnalysisUtil.java

@@ -19,8 +19,10 @@
 package org.apache.asterix.optimizer.base;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractDataSourceOperator;
 import org.apache.asterix.metadata.declared.AqlSourceId;
 import org.apache.asterix.om.functions.AsterixBuiltinFunctions;
 import org.apache.asterix.optimizer.rules.am.AccessMethodUtils;
@@ -31,12 +33,20 @@
 import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
 import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+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.AbstractFunctionCallExpression.FunctionKind;
+import org.apache.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.SortMergeJoinExpressionAnnotation.SortMergeJoinType;
+import org.apache.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
 import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractLogicalExpression;
+import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions;
+import org.apache.hyracks.algebricks.core.algebra.functions.AlgebricksBuiltinFunctions.ComparisonKind;
 import org.apache.hyracks.algebricks.core.algebra.functions.FunctionIdentifier;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractDataSourceOperator;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractBinaryJoinOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
 import org.apache.hyracks.algebricks.core.algebra.operators.logical.UnnestMapOperator;
+import org.apache.hyracks.algebricks.core.config.AlgebricksConfig;
 
 public class AnalysisUtil {
     /*
@@ -84,6 +94,140 @@ public static boolean isRunnableFieldAccessFunction(FunctionIdentifier fid) {
         return fieldAccessFunctions.contains(fid);
     }
 
+    private static SortMergeJoinType getSortMergeJoinable(ILogicalExpression e, Collection<LogicalVariable> inLeftAll,
+            Collection<LogicalVariable> inRightAll, List<LogicalVariable> outLeftFields,
+            List<LogicalVariable> outRightFields, List<Pair<ILogicalExpression, ILogicalExpression>> outBandRanges) {
+        AbstractFunctionCallExpression fexp = (AbstractFunctionCallExpression) e;
+        FunctionIdentifier fi = fexp.getFunctionIdentifier();
+        if (fi.equals(AlgebricksBuiltinFunctions.AND)) {
+            SortMergeJoinType retType = SortMergeJoinType.NESTLOOP;
+            for (Mutable<ILogicalExpression> a : fexp.getArguments()) {
+                SortMergeJoinType childType = getSortMergeJoinable(a.getValue(), inLeftAll, inRightAll, outLeftFields,
+                        outRightFields, outBandRanges);
+                if (SortMergeJoinType.BAND == childType)
+                    retType = SortMergeJoinType.BAND;
+                else if (retType != SortMergeJoinType.BAND && SortMergeJoinType.THETA == childType)
+                    retType = SortMergeJoinType.THETA;
+                // else if ...
+                // For Metric and Skyline join type in the future.
+            }
+            return retType;
+        } else {
+            ComparisonKind ck = AlgebricksBuiltinFunctions.getComparisonType(fi);
+            if (null == ck || ck == ComparisonKind.EQ) {
+                AlgebricksConfig.ALGEBRICKS_LOGGER
+                        .info("// SortMerge joinable encounter equal or fj condition -- Condition for" + e + ": " + ck);
+                return null;
+            }
+            ILogicalExpression opLeft = fexp.getArguments().get(0).getValue();
+            ILogicalExpression opRight = fexp.getArguments().get(1).getValue();
+            if (opLeft.getExpressionTag() == LogicalExpressionTag.FUNCTION_CALL
+                    && opRight.getExpressionTag() == LogicalExpressionTag.CONSTANT) {
+                ScalarFunctionCallExpression sfe = (ScalarFunctionCallExpression) opLeft;
+                if (FunctionKind.SCALAR != sfe.getKind()
+                        && AsterixBuiltinFunctions.NUMERIC_SUBTRACT != sfe.getFunctionIdentifier())
+                    return null;
+                for (int j = 0; j < 2; j++) {
+                    LogicalVariable varLeft = ((VariableReferenceExpression) (sfe.getArguments().get(j).getValue()))
+                            .getVariableReference();
+                    LogicalVariable varRight = ((VariableReferenceExpression) (sfe.getArguments().get((j + 1) % 2)
+                            .getValue())).getVariableReference();
+                    // We did not provide the merge of the partial ConstantExpression.
+                    if (inLeftAll.contains(varLeft) && inRightAll.contains(varRight)) {
+                        for (int i = 0; i < outLeftFields.size(); i++) {
+                            if (varLeft.equals(outLeftFields.get(i)) && varRight.equals(outRightFields.get(i))) {
+                                return updateAndGetRanges(outLeftFields, outRightFields, outBandRanges, i, ck, opRight);
+                            }
+                        }
+                        outLeftFields.add(varLeft);
+                        outRightFields.add(varRight);
+                        outBandRanges.add(new Pair<ILogicalExpression, ILogicalExpression>(null, null));
+                        return updateAndGetRanges(outLeftFields, outRightFields, outBandRanges,
+                                outBandRanges.size() - 1, ck, opRight);
+                    }
+                }
+            }
+        }
+        return SortMergeJoinType.NESTLOOP;
+    }
+
+    private static SortMergeJoinType updateAndGetRanges(List<LogicalVariable> outLeftFields,
+            List<LogicalVariable> outRightFields, List<Pair<ILogicalExpression, ILogicalExpression>> bandRanges,
+            int index, ComparisonKind ck, ILogicalExpression value) {
+        switch (ck) {
+            case GT:
+            case GE: {
+                // Add the ConstantExpression merge here in future.
+                if (bandRanges.size() < index || null == bandRanges.get(index)) {
+                    AlgebricksConfig.ALGEBRICKS_LOGGER.info("// Band condition left insert exception -- Condition for"
+                            + value + ": " + ck);
+                }
+                bandRanges.get(index).first = value;
+                break;
+            }
+            case LT:
+            case LE: {
+                if (bandRanges.size() < index || null == bandRanges.get(index)) {
+                    AlgebricksConfig.ALGEBRICKS_LOGGER.info("// Band condition right insert exception -- Condition for"
+                            + value + ": " + ck);
+                }
+                bandRanges.get(index).second = value;
+                break;
+            }
+            default:
+                break;
+        }
+
+        if (isBandRange(outLeftFields, outRightFields, bandRanges))
+            return SortMergeJoinType.BAND;
+        else if (isThetaRange(outLeftFields, outRightFields, bandRanges))
+            return SortMergeJoinType.THETA;
+        // Further for Metric and Skyline join.
+        else
+            return SortMergeJoinType.NESTLOOP;
+    }
+
+    private static boolean isBandRange(List<LogicalVariable> leftVars, List<LogicalVariable> rightVars,
+            List<Pair<ILogicalExpression, ILogicalExpression>> bandRanges) {
+        if (leftVars.size() != rightVars.size() || leftVars.size() != bandRanges.size())
+            return false;
+        for (int i = 0; i < bandRanges.size(); i++) {
+            if (bandRanges.get(i).first != null && bandRanges.get(i).second != null)
+                return true;
+        }
+        return false;
+    }
+
+    private static boolean isThetaRange(List<LogicalVariable> leftVars, List<LogicalVariable> rightVars,
+            List<Pair<ILogicalExpression, ILogicalExpression>> bandRanges) {
+        if (leftVars.size() != rightVars.size() || leftVars.size() != bandRanges.size())
+            return false;
+        for (int i = 0; i < bandRanges.size(); i++) {
+            if (bandRanges.get(i).first != null || bandRanges.get(i).second != null)
+                return true;
+        }
+        return false;
+    }
+
+    // Currently, we support the int and float/double and will make it general in the future.
+    public static SortMergeJoinType getSortMergeJoinable(ILogicalOperator op, Collection<LogicalVariable> inLeftAll,
+            Collection<LogicalVariable> inRightAll, List<LogicalVariable> outLeftFields,
+            List<LogicalVariable> outRightFields, List<Pair<ILogicalExpression, ILogicalExpression>> outBandRanges) {
+        // Three SortMergeJoinable operations: band, theta, metric and skyline. Currently just for band.
+        ILogicalExpression e = ((AbstractBinaryJoinOperator) op).getCondition().getValue();
+        switch (e.getExpressionTag()) {
+            case FUNCTION_CALL: {
+                // outBandRanges post process and cut off the band from the Select operator
+                if (SortMergeJoinType.BAND == getSortMergeJoinable(e, inLeftAll, inRightAll, outLeftFields,
+                        outRightFields, outBandRanges))
+                    return SortMergeJoinType.BAND;
+            }
+            default:
+                break;
+        }
+        return SortMergeJoinType.NESTLOOP;
+    }
+
     public static boolean isDataSetCall(ILogicalExpression e) {
         if (((AbstractLogicalExpression) e).getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
             return false;