diff --git a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java index e70f1dffa63..152cdb95531 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java @@ -234,7 +234,7 @@ public MatrixBlock getEquiHeightBins(ExecutionContext ec, int colID, double[ public void processRowQPick(ExecutionContext ec) { MatrixObject in = ec.getMatrixObject(input1); FederationMap fedMap = in.getFedMapping(); - boolean average = _type == OperationTypes.MEDIAN; + boolean average = _type == OperationTypes.MEDIAN || _type == OperationTypes.VALUEPICK; double[] quantiles = input2 != null ? (input2.isMatrix() ? ec.getMatrixInput(input2).getDenseBlockValues() : input2.isScalar() ? new double[] {ec.getScalarInput(input2).getDoubleValue()} : null) : @@ -755,10 +755,10 @@ public FederatedResponse execute(ExecutionContext ec, Data... data) { MatrixBlock picked; if (_quantiles.getLength() == 1) { return new FederatedResponse(FederatedResponse.ResponseType.SUCCESS, - new Object[] {mb.pickValue(_quantiles.get(0, 0))}); + new Object[] {mb.pickValue(_quantiles.get(0, 0), mb.getLength() % 2 == 0)}); } else { - picked = mb.pickValues(_quantiles, new MatrixBlock()); + picked = mb.pickValues(_quantiles, new MatrixBlock(), mb.getLength() % 2 == 0); return new FederatedResponse(FederatedResponse.ResponseType.SUCCESS, new Object[] {picked}); } diff --git a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java index a30fbc456cb..72270f55e4e 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/spark/QuantilePickSPInstruction.java @@ -114,13 +114,12 @@ public void processInstruction(ExecutionContext ec) { case VALUEPICK: { if( input2.isScalar() ) { ScalarObject quantile = ec.getScalarInput(input2); - double[] wt = getWeightedQuantileSummary(in, mc, - new double[]{quantile.getDoubleValue()}); + double[] wt = getWeightedQuantileSummary(in, mc, new double[] {quantile.getDoubleValue()}, true); ec.setScalarOutput(output.getName(), new DoubleObject(wt[3])); } else { - double[] wt = getWeightedQuantileSummary(in, mc, DataConverter - .convertToDoubleVector(ec.getMatrixInput(input2.getName()))); + double[] wt = getWeightedQuantileSummary(in, mc, + DataConverter.convertToDoubleVector(ec.getMatrixInput(input2.getName())), true); ec.releaseMatrixInput(input2.getName()); int qlen = wt.length/3; MatrixBlock out = new MatrixBlock(qlen,1,false); @@ -130,20 +129,21 @@ public void processInstruction(ExecutionContext ec) { } break; } - + case MEDIAN: { - double[] wt = getWeightedQuantileSummary(in, mc, new double[]{0.5}); + double[] wt = getWeightedQuantileSummary(in, mc, new double[] {0.5}, true); ec.setScalarOutput(output.getName(), new DoubleObject(wt[3])); break; } - + case IQM: { - double[] wt = getWeightedQuantileSummary(in, mc, new double[]{0.25,0.75}); - long key25 = (long)Math.ceil(wt[1]); - long key75 = (long)Math.ceil(wt[2]); - JavaPairRDD out = in - .filter(new FilterFunction(key25+1,key75,mc.getBlocksize())) - .mapToPair(new ExtractAndSumFunction(key25+1, key75, mc.getBlocksize())); + // IQM uses wt[3]/wt[4] as raw boundary values for computeIQMCorrection, so no averaging + double[] wt = getWeightedQuantileSummary(in, mc, new double[] {0.25, 0.75}, false); + long key25 = (long) Math.ceil(wt[1]); + long key75 = (long) Math.ceil(wt[2]); + JavaPairRDD out = in + .filter(new FilterFunction(key25 + 1, key75, mc.getBlocksize())) + .mapToPair(new ExtractAndSumFunction(key25 + 1, key75, mc.getBlocksize())); double sum = RDDAggregateUtils.sumStable(out).get(0, 0); double val = MatrixBlock.computeIQMCorrection( sum, wt[0], wt[3], wt[5], wt[4], wt[6]); @@ -165,10 +165,10 @@ public void processInstruction(ExecutionContext ec) { * @param quantiles one or more quantiles between 0 and 1. * @return a summary of weighted quantiles */ - private static double[] getWeightedQuantileSummary(JavaPairRDD w, DataCharacteristics mc, double[] quantiles) - { - double[] ret = new double[3*quantiles.length + 1]; - if( mc.getCols()==2 ) //weighted + private static double[] getWeightedQuantileSummary(JavaPairRDD w, + DataCharacteristics mc, double[] quantiles, boolean average) { + double[] ret = new double[3 * quantiles.length + 1]; + if(mc.getCols() == 2) // weighted { //sort blocks (values sorted but blocks and partitions are not) w = w.sortByKey(); @@ -213,11 +213,17 @@ private static double[] getWeightedQuantileSummary(JavaPairRDD data() { - return Arrays.asList(new Object[][] { - // {1000, 1, false}, - {128, 1, true}}); + return Arrays.asList(new Object[][] {{1000, 1, false}, {128, 1, true}}); } @Override @@ -71,19 +68,16 @@ public void setUp() { } @Test - @Ignore public void federatedQuantile1CP() { federatedQuartile(Types.ExecMode.SINGLE_NODE, TEST_NAME1, 0.25); } @Test - @Ignore public void federatedQuantile2CP() { federatedQuartile(Types.ExecMode.SINGLE_NODE, TEST_NAME1, 0.5); } @Test - @Ignore public void federatedQuantile3CP() { federatedQuartile(Types.ExecMode.SINGLE_NODE, TEST_NAME1, 0.75); } @@ -104,19 +98,16 @@ public void federatedQuantilesCP() { } @Test - @Ignore public void federatedQuantile1SP() { federatedQuartile(Types.ExecMode.SPARK, TEST_NAME1, 0.25); } @Test - @Ignore public void federatedQuantile2SP() { federatedQuartile(Types.ExecMode.SPARK, TEST_NAME1, 0.5); } @Test - @Ignore public void federatedQuantile3SP() { federatedQuartile(Types.ExecMode.SPARK, TEST_NAME1, 0.75); }