package org.qcri.rheem.basic.operators;

import org.apache.commons.lang3.Validate;
import org.qcri.rheem.core.api.Configuration;
import org.qcri.rheem.core.function.FunctionDescriptor;
import org.qcri.rheem.core.function.ReduceDescriptor;
import org.qcri.rheem.core.function.TransformationDescriptor;
import org.qcri.rheem.core.optimizer.cardinality.CardinalityEstimator;
import org.qcri.rheem.core.optimizer.cardinality.DefaultCardinalityEstimator;
import org.qcri.rheem.core.plan.rheemplan.UnaryToUnaryOperator;
import org.qcri.rheem.core.types.DataSetType;

import java.util.Optional;

/**
 * This operator groups the elements of a data set and aggregates the groups.
 */
public class ReduceByOperator<Type, Key> extends UnaryToUnaryOperator<Type, Type> {

    protected final TransformationDescriptor<Type, Key> keyDescriptor;

    protected final ReduceDescriptor<Type> reduceDescriptor;

    /**
     * Creates a new instance.
     */
    public ReduceByOperator(FunctionDescriptor.SerializableFunction<Type, Key> keyFunction,
                            FunctionDescriptor.SerializableBinaryOperator<Type> reduceDescriptor,
                            Class<Key> keyClass,
                            Class<Type> typeClass) {
        this(new TransformationDescriptor<>(keyFunction, typeClass, keyClass),
                new ReduceDescriptor<>(reduceDescriptor, typeClass));
    }

    /**
     * Creates a new instance.
     *
     * @param keyDescriptor    describes how to extract the key from data units
     * @param reduceDescriptor describes the reduction to be performed on the elements
     */
    public ReduceByOperator(TransformationDescriptor<Type, Key> keyDescriptor,
                            ReduceDescriptor<Type> reduceDescriptor) {
        this(keyDescriptor, reduceDescriptor, DataSetType.createDefault(keyDescriptor.getInputType()));
    }

    /**
     * Creates a new instance.
     *
     * @param keyDescriptor    describes how to extract the key from data units
     * @param reduceDescriptor describes the reduction to be performed on the elements
     * @param type             type of the reduce elements (i.e., type of {@link #getInput()} and {@link #getOutput()})
     */
    public ReduceByOperator(TransformationDescriptor<Type, Key> keyDescriptor,
                            ReduceDescriptor<Type> reduceDescriptor,
                            DataSetType<Type> type) {
        super(type, type, true);
        this.keyDescriptor = keyDescriptor;
        this.reduceDescriptor = reduceDescriptor;
    }

    /**
     * Copies an instance (exclusive of broadcasts).
     *
     * @param that that should be copied
     */
    public ReduceByOperator(ReduceByOperator<Type, Key> that) {
        super(that);
        this.keyDescriptor = that.getKeyDescriptor();
        this.reduceDescriptor = that.getReduceDescriptor();
    }

    public DataSetType<Type> getType() {
        return this.getInputType();
    }

    public TransformationDescriptor<Type, Key> getKeyDescriptor() {
        return this.keyDescriptor;
    }

    public ReduceDescriptor<Type> getReduceDescriptor() {
        return this.reduceDescriptor;
    }

    public String getSelectKeyString(){
        if (this.getReduceDescriptor().getUdfSelectivity() != null){
            return this.getReduceDescriptor().getUdfSelectivityKeyString();
        } else {
            return "";
        }
    }


    @Override
    public Optional<CardinalityEstimator> createCardinalityEstimator(
            final int outputIndex,
            final Configuration configuration) {
        Validate.inclusiveBetween(0, this.getNumOutputs() - 1, outputIndex);
        // TODO: Come up with a decent way to estimate the "distinctness" of reduction keys.
        return Optional.of(new DefaultCardinalityEstimator(
                0.5d,
                1,
                this.isSupportingBroadcastInputs(),
                inputCards -> (long) (inputCards[0] * 0.1)));
    }
}

@@ -170,32 +170,89 @@
 tionDescriptor;%0A
+import org.qcri.rheem.core.function.PredicateDescriptor;%0A
 import org.qcri.
@@ -343,32 +343,226 @@
 tionDescriptor;%0A
+import org.qcri.rheem.core.optimizer.OptimizationContext;%0Aimport org.qcri.rheem.core.optimizer.ProbabilisticDoubleInterval;%0Aimport org.qcri.rheem.core.optimizer.cardinality.CardinalityEstimate;%0A
 import org.qcri.
@@ -3689,16 +3689,58 @@
 ptional%3C
+org.qcri.rheem.core.optimizer.cardinality.
 Cardinal
@@ -3773,32 +3773,32 @@
 alityEstimator(%0A
-
             fina
@@ -4066,23 +4066,171 @@
 .of(new 
-Default
+ReduceByOperator.CardinalityEstimator(configuration));%0A    %7D%0A%0A%0A    private class CardinalityEstimator implements org.qcri.rheem.core.optimizer.cardinality.
 Cardinal
@@ -4233,33 +4233,35 @@
 inalityEstimator
-(
+ %7B%0A
 %0A               
@@ -4253,24 +4253,28 @@
         
+/**%0A
         
 0.5d,%0A  
@@ -4269,13 +4269,67 @@
     
-0.5d,
+ * The expected selectivity to be applied in this instance.
 %0A   
@@ -4338,68 +4338,511 @@
     
+*/%0A
        
-1,%0A                this.isSupportingBroadcast
+ private final ProbabilisticDoubleInterval selectivity;%0A%0A        public CardinalityEstimator(Configuration configuration) %7B%0A            this.selectivity = configuration.getUdfSelectivityProvider().provideFor(ReduceByOperator.this.reduceDescriptor);%0A        %7D%0A%0A        @Override%0A        public CardinalityEstimate estimate(OptimizationContext optimizationContext, CardinalityEstimate... inputEstimates) %7B%0A            Validate.isTrue(inputEstimates.length == ReduceByOperator.this.getNum
 Inputs()
 ,%0A  
@@ -4837,17 +4837,18 @@
 Inputs()
-,
+);
 %0A       
@@ -4856,57 +4856,449 @@
     
-    inputCards -%3E (long) (inputCards%5B0%5D * 0.1)));
+final CardinalityEstimate inputEstimate = inputEstimates%5B0%5D;%0A%0A            return new CardinalityEstimate(%0A                    (long) (inputEstimate.getLowerEstimate() * this.selectivity.getLowerEstimate()),%0A                    (long) (inputEstimate.getUpperEstimate() * this.selectivity.getUpperEstimate()),%0A                    inputEstimate.getCorrectnessProbability() * this.selectivity.getCorrectnessProbability()%0A            );%0A        %7D
 %0A   

make reduceBy operator use selectivity repository