From 7775070e1f33a4b80c7ef6e952a138bcced1f784 Mon Sep 17 00:00:00 2001 From: MegaByteTron Date: Sun, 21 Jun 2026 21:50:40 +0200 Subject: [PATCH 1/4] [SYSTEMDS-3922] Fix federated quantile VALUEPICK averaging for even-length input The federated QuantilePickFEDInstruction only enabled the even-length averaging branch for MEDIAN, while the CP reference path (QuantilePickCPInstruction) applies the same averaging convention for VALUEPICK by passing matBlock.getLength()%2==0 to MatrixBlock.pickValue. As a result, quantile(A, p) on a federated matrix with an even row count diverged from the CP reference (which compareResults(1e-9) rejects), causing the federatedQuantile{1,2,3}{CP,SP} tests to fail. They were previously marked @Ignore to hide the failure. Extend the average flag to also cover VALUEPICK so the federated path mirrors the CP averaging contract. The existing guard average = average && (... rows ...) % 2 == 0 keeps averaging disabled for odd-length input, matching CP. IQM is intentionally excluded because it has its own interpolation path. Un-ignore the six previously-disabled tests. --- .../instructions/fed/QuantilePickFEDInstruction.java | 2 +- .../federated/primitives/part2/FederatedQuantileTest.java | 7 ------- 2 files changed, 1 insertion(+), 8 deletions(-) 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..c5753bcac9c 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) : diff --git a/src/test/java/org/apache/sysds/test/functions/federated/primitives/part2/FederatedQuantileTest.java b/src/test/java/org/apache/sysds/test/functions/federated/primitives/part2/FederatedQuantileTest.java index 6f13fce81b2..57ca28055a4 100644 --- a/src/test/java/org/apache/sysds/test/functions/federated/primitives/part2/FederatedQuantileTest.java +++ b/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; @@ -71,19 +70,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 +100,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); } From ad13a21590cc114408222e4511d7df4cff4ce62d Mon Sep 17 00:00:00 2001 From: MegaByteTron Date: Sun, 12 Jul 2026 23:18:07 +0200 Subject: [PATCH 2/4] [SYSTEMDS-3898] Fix quantile pick averaging for even-length SP execution The Spark path in QuantilePickSPInstruction.getWeightedQuantileSummary picked the value at ceil(p * n) without averaging with its right neighbor, diverging from the CP fix in 3ae6a77 that established even-length averaging as the SystemDS quantile convention. As a result the same DML program returned different values under CP vs SP for any matrix with an even row count (most visibly at p = 0.5). Mirror the CP condition: when the input length is even and the picked key is not the last row, add the next value and divide by two. This brings the unweighted SP path into agreement with CP for every quantile, not just the median. Also mirror the robustness guard from ebdabf1: the key < mc.getRows() check prevents an out-of-bounds lookup for boundary quantiles (p == 1.0 on even n). Un-ignore testQuantileBugSP, which is the SP analog of testQuantileBugCP and now passes on the same [1,5,7,10] median input. The weighted branch (mc.getCols() == 2) still uses the same non-averaged key lookup and remains flagged by the existing TODO for a later follow-up, matching the scope note in 3ae6a77. --- .../instructions/spark/QuantilePickSPInstruction.java | 10 ++++++++-- .../test/functions/binary/matrix/QuantileTest.java | 8 ++++---- 2 files changed, 12 insertions(+), 6 deletions(-) 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..3c4f89a3153 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 @@ -216,8 +216,14 @@ private static double[] getWeightedQuantileSummary(JavaPairRDD Date: Mon, 13 Jul 2026 16:37:30 +0200 Subject: [PATCH 3/4] [SYSTEMDS-3898] Fix IQM regression from SP quantile averaging change The prior fix added even-length averaging inside getWeightedQuantileSummary unconditionally, which broke IQM because computeIQMCorrection uses the raw values at the Q25/Q75 positions as boundary values (wt[3], wt[4]) for the fractional-cell correction, not as picked values. Gate the averaging behind a new `average` flag: VALUEPICK and MEDIAN pass true (they consume wt[3] as the picked value), IQM passes false. Restores testIQM1_SP without regressing testQuantileBugSP or the FED tests. --- .../spark/QuantilePickSPInstruction.java | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) 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 3c4f89a3153..e510f01d156 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 @@ -115,12 +115,12 @@ public void processInstruction(ExecutionContext ec) { if( input2.isScalar() ) { ScalarObject quantile = ec.getScalarInput(input2); double[] wt = getWeightedQuantileSummary(in, mc, - new double[]{quantile.getDoubleValue()}); + 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()))); + .convertToDoubleVector(ec.getMatrixInput(input2.getName())), true); ec.releaseMatrixInput(input2.getName()); int qlen = wt.length/3; MatrixBlock out = new MatrixBlock(qlen,1,false); @@ -130,15 +130,16 @@ 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}); + //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 @@ -165,10 +166,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) + 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 + if( mc.getCols()==2 ) //weighted { //sort blocks (values sorted but blocks and partitions are not) w = w.sortByKey(); @@ -220,7 +221,7 @@ private static double[] getWeightedQuantileSummary(JavaPairRDD Date: Wed, 15 Jul 2026 16:23:00 +0200 Subject: [PATCH 4/4] [SYSTEMDS-3922] Enable column-federated VALUEPICK averaging and fix code style Pass the average flag through the ValuePick UDF (used by the column- federated path) so single-shard VALUEPICK picks now average adjacent order statistics for even-length inputs, matching CP semantics. Enables the previously commented-out {1000, 1, false} case in FederatedQuantileTest. Also simplifies the processRowQPick initialization so the parity guard in the subsequent 'Average for median' step stays a straight AND, and applies the Eclipse code-style fixes flagged by the Java Format Check workflow on lines touched by this branch. --- .../fed/QuantilePickFEDInstruction.java | 4 +- .../spark/QuantilePickSPInstruction.java | 47 +++++++++---------- .../part2/FederatedQuantileTest.java | 4 +- 3 files changed, 26 insertions(+), 29 deletions(-) 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 c5753bcac9c..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 @@ -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 e510f01d156..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()}, true); + 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())), true); + 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); @@ -132,19 +131,19 @@ public void processInstruction(ExecutionContext ec) { } case MEDIAN: { - double[] wt = getWeightedQuantileSummary(in, mc, new double[]{0.5}, true); + double[] wt = getWeightedQuantileSummary(in, mc, new double[] {0.5}, true); ec.setScalarOutput(output.getName(), new DoubleObject(wt[3])); break; } case IQM: { - //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())); + // 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]); @@ -166,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, boolean average) - { - 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(); @@ -214,16 +213,16 @@ 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