[SYSTEMDS-3922] Fix federated quantile VALUEPICK averaging for even-length input

src/main/java/org/apache/sysds/runtime/instructions/fed/QuantilePickFEDInstruction.java

@@ -234,7 +234,7 @@ public <T> MatrixBlock getEquiHeightBins(ExecutionContext ec, int colID, double[
 	public <T> 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});
 			}

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<MatrixIndexes,MatrixBlock> 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<MatrixIndexes, MatrixBlock> 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<MatrixIndexes,MatrixBlock> w, DataCharacteristics mc, double[] quantiles)
-	{
-		double[] ret = new double[3*quantiles.length + 1];
-		if( mc.getCols()==2 ) //weighted 
+	private static double[] getWeightedQuantileSummary(JavaPairRDD<MatrixIndexes, MatrixBlock> 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<MatrixIndexes,Mat
 		}
 		else {
 			ret[0] = mc.getRows();
-			for( int i=0; i<quantiles.length; i++ ){
-				ret[i+1] = quantiles[i] * mc.getRows();
-				ret[i+quantiles.length+1] = Math.ceil(ret[i+1])-ret[i+1];
-				ret[i+2*quantiles.length+1] = lookupKey(w, 
-					(long)Math.ceil(ret[i+1]), mc.getBlocksize());
+			for(int i = 0; i < quantiles.length; i++) {
+				ret[i + 1] = quantiles[i] * mc.getRows();
+				ret[i + quantiles.length + 1] = Math.ceil(ret[i + 1]) - ret[i + 1];
+				long key = (long) Math.ceil(ret[i + 1]);
+				ret[i + 2 * quantiles.length + 1] = lookupKey(w, key, mc.getBlocksize());
+
+				// average w/ next value for even-length arrays (mirrors CP QuantilePickCPInstruction)
+				if(average && mc.getRows() % 2 == 0 && key < mc.getRows()) {
+					ret[i + 2 * quantiles.length + 1] += lookupKey(w, key + 1, mc.getBlocksize());
+					ret[i + 2 * quantiles.length + 1] /= 2;
+				}
 			}
 		}
 

src/test/java/org/apache/sysds/test/functions/binary/matrix/QuantileTest.java

@@ -165,10 +165,10 @@ public void testQuantileBugCP() {
 	}
 
 // TODO reimplement distributed value pick logic
-//	@Test
-//	public void testQuantileBugSP() {
-//		runQuantileTest(TEST_NAME4, 0.5, false, ExecType.SPARK);
-//	}
+	@Test
+	public void testQuantileBugSP() {
+		runQuantileTest(TEST_NAME4, 0.5, false, ExecType.SPARK);
+	}
 
 	private void runQuantileTest( String TEST_NAME, double p, boolean sparse, ExecType et)
 	{

src/test/java/org/apache/sysds/test/functions/federated/primitives/part2/FederatedQuantileTest.java

@@ -30,7 +30,6 @@
 import org.apache.sysds.test.TestConfiguration;
 import org.apache.sysds.test.TestUtils;
 import org.junit.Assert;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -56,9 +55,7 @@ public class FederatedQuantileTest extends AutomatedTestBase {
 
 	@Parameterized.Parameters
 	public static Collection<Object[]> 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);
 	}