Skip to content

Commit 9e0a481

Browse files
janniklindemboehm7
janniklinde
authored and
mboehm7
committed
[SYSTEMDS-3924] Additional tests for OOC stream creation/collect
Closes #2340.
1 parent bc3216a commit 9e0a481

File tree

5 files changed

+315
-25
lines changed

5 files changed

+315
-25
lines changed

‎src/test/java/org/apache/sysds/test/functions/ooc/StreamCollectTest.java‎

Lines changed: 194 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,194 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.sysds.test.functions.ooc;
21+
22+
import org.apache.sysds.common.Types;
23+
import org.apache.sysds.runtime.controlprogram.LocalVariableMap;
24+
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
25+
import org.apache.sysds.runtime.instructions.Instruction;
26+
import org.apache.sysds.runtime.instructions.cp.ReorgCPInstruction;
27+
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
28+
import org.apache.sysds.runtime.instructions.ooc.ReblockOOCInstruction;
29+
import org.apache.sysds.runtime.instructions.ooc.TransposeOOCInstruction;
30+
import org.apache.sysds.runtime.io.MatrixWriter;
31+
import org.apache.sysds.runtime.io.MatrixWriterFactory;
32+
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
33+
import org.apache.sysds.runtime.matrix.data.MatrixValue;
34+
import org.apache.sysds.runtime.meta.MatrixCharacteristics;
35+
import org.apache.sysds.runtime.util.DataConverter;
36+
import org.apache.sysds.runtime.util.HDFSTool;
37+
import org.apache.sysds.test.AutomatedTestBase;
38+
import org.apache.sysds.test.TestConfiguration;
39+
import org.apache.sysds.test.TestUtils;
40+
import org.apache.sysds.utils.Statistics;
41+
import org.junit.Assert;
42+
import org.junit.Test;
43+
44+
import java.io.IOException;
45+
import java.util.ArrayList;
46+
import java.util.HashMap;
47+
import java.util.List;
48+
49+
public class StreamCollectTest extends AutomatedTestBase {
50+
private final static String TEST_NAME1 = "StreamCollect";
51+
private final static String TEST_DIR = "functions/ooc/";
52+
private final static String TEST_CLASS_DIR = TEST_DIR + StreamCollectTest.class.getSimpleName() + "/";
53+
private final static int rows = 2000;
54+
private final static int cols = 1000;
55+
private final static String INPUT_NAME = "input";
56+
private final static String OUTPUT_NAME = "res";
57+
private final static double eps = 1e-10;
58+
59+
@Override
60+
public void setUp() {
61+
TestUtils.clearAssertionInformation();
62+
TestConfiguration config = new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1);
63+
addTestConfiguration(TEST_NAME1, config);
64+
}
65+
66+
@Test
67+
public void runRawInstructionSequenceTest() {
68+
try {
69+
getAndLoadTestConfiguration(TEST_NAME1);
70+
MatrixBlock mb = MatrixBlock.randOperations(rows, cols, 1.0, -1, 1, "uniform", 7);
71+
MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY);
72+
writer.writeMatrixToHDFS(mb, input(INPUT_NAME), rows, cols, 1000, rows * cols);
73+
HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64,
74+
new MatrixCharacteristics(rows, cols, 1000, rows * cols), Types.FileFormat.BINARY);
75+
76+
ExecutionContext ec = new ExecutionContext(new LocalVariableMap());
77+
78+
VariableCPInstruction createIn = VariableCPInstruction.parseInstruction(
79+
"CP°createvar°pREADX°" + input(INPUT_NAME) + "°false°MATRIX°binary°" + rows + "°" + cols +
80+
"°1000°700147°copy");
81+
VariableCPInstruction createInRblk = VariableCPInstruction.parseInstruction(
82+
"CP°createvar°_mVar0°" + input("tmp0") + "°true°MATRIX°binary°" + rows + "°" + cols +
83+
"°1000°700147°copy");
84+
ReblockOOCInstruction rblkIn = ReblockOOCInstruction.parseInstruction(
85+
"OOC°rblk°pREADX·MATRIX·FP64°_mVar0·MATRIX·FP64°1000°true");
86+
VariableCPInstruction createOut = VariableCPInstruction.parseInstruction(
87+
"CP°createvar°_mVar1°" + input("tmp1") + "°true°MATRIX°binary°" + rows + "°" + cols +
88+
"°1000°700147°copy");
89+
TransposeOOCInstruction oocTranspose = TransposeOOCInstruction.parseInstruction(
90+
"OOC°r'°_mVar0·MATRIX·FP64°_mVar1·MATRIX·FP64");
91+
VariableCPInstruction createOut2 = VariableCPInstruction.parseInstruction(
92+
"CP°createvar°_mVar2°" + input("tmp2") + "°true°MATRIX°binary°" + rows + "°" + cols +
93+
"°1000°700147°copy");
94+
ReorgCPInstruction cpTranspose = ReorgCPInstruction.parseInstruction(
95+
"CP°r'°_mVar1·MATRIX·FP64°_mVar2·MATRIX·FP64°1");
96+
97+
createIn.processInstruction(ec);
98+
createInRblk.processInstruction(ec);
99+
rblkIn.processInstruction(ec);
100+
createOut.processInstruction(ec);
101+
oocTranspose.processInstruction(ec);
102+
createOut2.processInstruction(ec);
103+
cpTranspose.processInstruction(ec);
104+
}
105+
catch(Exception ex) {
106+
Assert.fail(ex.getMessage());
107+
}
108+
}
109+
110+
@Test
111+
public void runAppTest1() {
112+
runAppTest("_1", List.of(List.of(1000D, 1D), List.of(1000D, 1000D)));
113+
}
114+
115+
@Test
116+
public void runAppTest2() {
117+
runAppTest("_2", List.of(List.of(500D, 1D), List.of(500D, 500D)));
118+
}
119+
120+
@Test
121+
public void runAppTest3() {
122+
runAppTest("_3", List.of(List.of(1500D, 1D), List.of(1500D, 1500D)));
123+
}
124+
125+
private void runAppTest(String scriptSuffix, List<List<Double>> matrixDims) {
126+
Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE);
127+
128+
try {
129+
getAndLoadTestConfiguration(TEST_NAME1);
130+
String HOME = SCRIPT_DIR + TEST_DIR;
131+
fullDMLScriptName = HOME + TEST_NAME1 + scriptSuffix + ".dml";
132+
List<String> proArgs = new ArrayList<>();
133+
proArgs.add("-explain");
134+
proArgs.add("-stats");
135+
proArgs.add("-ooc");
136+
proArgs.add("-args");
137+
//programArgs = new String[]{"-explain", "-stats", "-ooc",
138+
// "-args", input(INPUT_NAME), output(OUTPUT_NAME)};
139+
140+
for(int i = 0; i < matrixDims.size(); i++) {
141+
List<Double> dims = matrixDims.get(i);
142+
int mrows = dims.get(0).intValue();
143+
int mcols = dims.get(1).intValue();
144+
145+
// 1. Generate the data as MatrixBlock object
146+
double[][] A_data = getRandomMatrix(mrows, mcols, 0, 10, 1, 10);
147+
148+
// 2. Convert the double arrays to MatrixBlock object
149+
MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data);
150+
151+
// 3. Create a binary matrix writer
152+
MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY);
153+
154+
// 4. Write matrix A to a binary SequenceFile
155+
writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME + "_" + i), mrows, mcols, 1000, A_mb.getNonZeros());
156+
HDFSTool.writeMetaDataFile(input(INPUT_NAME + "_" + i + ".mtd"), Types.ValueType.FP64,
157+
new MatrixCharacteristics(mrows, mcols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY);
158+
proArgs.add(input(INPUT_NAME + "_" + i));
159+
}
160+
161+
proArgs.add(output(OUTPUT_NAME));
162+
163+
programArgs = proArgs.toArray(String[]::new);
164+
165+
boolean exceptionExpected = false;
166+
runTest(true, exceptionExpected, null, -1);
167+
168+
// Validate that at least one OOC instruction was used
169+
Assert.assertTrue("OOC wasn't used", heavyHittersContainOOC());
170+
171+
proArgs.remove(2); // Remove ooc flag
172+
proArgs.set(proArgs.size() - 1, output(OUTPUT_NAME + "_target"));
173+
programArgs = proArgs.toArray(String[]::new);
174+
runTest(true, exceptionExpected, null, -1);
175+
176+
HashMap<MatrixValue.CellIndex, Double> result = readDMLMatrixFromOutputDir(OUTPUT_NAME);
177+
HashMap<MatrixValue.CellIndex, Double> target = readDMLMatrixFromOutputDir(OUTPUT_NAME + "_target");
178+
TestUtils.compareMatrices(result, target, eps, "Result", "Target");
179+
}
180+
catch(IOException e) {
181+
throw new RuntimeException(e);
182+
}
183+
finally {
184+
resetExecMode(platformOld);
185+
}
186+
}
187+
188+
private boolean heavyHittersContainOOC() {
189+
for(String opcode : Statistics.getCPHeavyHitterOpCodes())
190+
if(opcode.startsWith(Instruction.OOC_INST_PREFIX))
191+
return true;
192+
return false;
193+
}
194+
}

‎src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java‎

Lines changed: 29 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -44,9 +44,9 @@ public class TransposeTest extends AutomatedTestBase {
4444
private static final String INPUT_NAME = "X";
4545
private static final String OUTPUT_NAME = "res";
4646

47-
private final static int rows = 1000;
48-
private final static int cols_wide = 1000;
49-
//private final static int cols_skinny = 500;
47+
private final static int rows = 1500;
48+
private final static int cols_wide = 2500;
49+
private final static int cols_skinny = 500;
5050

5151
private final static double sparsity1 = 0.7;
5252
private final static double sparsity2 = 0.1;
@@ -63,25 +63,32 @@ public void testTranspose1() {
6363
runTransposeTest(cols_wide, false);
6464
}
6565

66-
// @Test
67-
// public void testTranspose2() {
68-
// runTransposeTest(cols_skinny, false);
69-
// }
66+
@Test
67+
public void testTranspose2() {
68+
runTransposeTest(cols_skinny, false);
69+
}
70+
71+
@Test
72+
public void testTransposeSparse1() {
73+
runTransposeTest(cols_wide, true);
74+
}
75+
76+
@Test
77+
public void testTransposeSparse2() {
78+
runTransposeTest(cols_skinny, true);
79+
}
7080

71-
private void runTransposeTest(int cols, boolean sparse )
72-
{
81+
private void runTransposeTest(int cols, boolean sparse) {
7382
Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE);
7483

75-
try
76-
{
84+
try {
7785
getAndLoadTestConfiguration(TEST_NAME1);
7886
String HOME = SCRIPT_DIR + TEST_DIR;
7987
fullDMLScriptName = HOME + TEST_NAME1 + ".dml";
80-
programArgs = new String[]{"-explain", "-stats", "-ooc",
81-
"-args", input(INPUT_NAME), output(OUTPUT_NAME)};
88+
programArgs = new String[] {"-explain", "-stats", "-ooc", "-args", input(INPUT_NAME), output(OUTPUT_NAME)};
8289

8390
// 1. Generate the data as MatrixBlock object
84-
double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse?sparsity2:sparsity1, 10);
91+
double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse ? sparsity2 : sparsity1, 10);
8592

8693
// 2. Convert the double arrays to MatrixBlock object
8794
MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data);
@@ -92,12 +99,12 @@ private void runTransposeTest(int cols, boolean sparse )
9299
// 4. Write matrix A to a binary SequenceFile
93100
writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME), rows, cols, 1000, A_mb.getNonZeros());
94101
HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64,
95-
new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY);
102+
new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY);
96103

97104
boolean exceptionExpected = false;
98105
runTest(true, exceptionExpected, null, -1);
99106

100-
double[][] C1 = readMatrix(output(OUTPUT_NAME), Types.FileFormat.BINARY, rows, cols, 1000, 1000);
107+
double[][] C1 = readMatrix(output(OUTPUT_NAME), Types.FileFormat.BINARY, cols, rows, 1000);
101108
double result = 0.0;
102109
for(int i = 0; i < rows; i++) { // verify the results with Java
103110
double expected = 0.0;
@@ -110,23 +117,20 @@ private void runTransposeTest(int cols, boolean sparse )
110117
}
111118

112119
String prefix = Instruction.OOC_INST_PREFIX;
113-
Assert.assertTrue("OOC wasn't used for RBLK",
114-
heavyHittersContainsString(prefix + Opcodes.RBLK));
115-
Assert.assertTrue("OOC wasn't used for TRANSPOSE",
116-
heavyHittersContainsString(prefix + Opcodes.TRANSPOSE));
120+
Assert.assertTrue("OOC wasn't used for RBLK", heavyHittersContainsString(prefix + Opcodes.RBLK));
121+
Assert.assertTrue("OOC wasn't used for TRANSPOSE", heavyHittersContainsString(prefix + Opcodes.TRANSPOSE));
117122
}
118-
catch (IOException e) {
123+
catch(IOException e) {
119124
throw new RuntimeException(e);
120125
}
121126
finally {
122127
resetExecMode(platformOld);
123128
}
124129
}
125130

126-
private static double[][] readMatrix(String fname, Types.FileFormat fmt, long rows, long cols, int brows, int bcols )
127-
throws IOException
128-
{
129-
MatrixBlock mb = DataConverter.readMatrixFromHDFS(fname, fmt, rows, cols, brows, bcols);
131+
private static double[][] readMatrix(String fname, Types.FileFormat fmt, long rows, long cols, int blen)
132+
throws IOException {
133+
MatrixBlock mb = DataConverter.readMatrixFromHDFS(fname, fmt, rows, cols, blen);
130134
double[][] C = DataConverter.convertToDoubleMatrix(mb);
131135
return C;
132136
}

‎src/test/scripts/functions/ooc/StreamCollect_1.dml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
#-------------------------------------------------------------
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one
4+
# or more contributor license agreements. See the NOTICE file
5+
# distributed with this work for additional information
6+
# regarding copyright ownership. The ASF licenses this file
7+
# to you under the Apache License, Version 2.0 (the
8+
# "License"); you may not use this file except in compliance
9+
# with the License. You may obtain a copy of the License at
10+
#
11+
# http://www.apache.org/licenses/LICENSE-2.0
12+
#
13+
# Unless required by applicable law or agreed to in writing,
14+
# software distributed under the License is distributed on an
15+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
# KIND, either express or implied. See the License for the
17+
# specific language governing permissions and limitations
18+
# under the License.
19+
#
20+
#-------------------------------------------------------------
21+
22+
# Read the input matrix as a stream
23+
b = read($1);
24+
X = read($2);
25+
26+
OOC = rowSums(t(X)) %*% t(rowSums(t(X)));
27+
OOC = OOC / 1.5;
28+
29+
res = solve(OOC, b);
30+
31+
write(res, $3, format="text");

‎src/test/scripts/functions/ooc/StreamCollect_2.dml‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
#-------------------------------------------------------------
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one
4+
# or more contributor license agreements. See the NOTICE file
5+
# distributed with this work for additional information
6+
# regarding copyright ownership. The ASF licenses this file
7+
# to you under the Apache License, Version 2.0 (the
8+
# "License"); you may not use this file except in compliance
9+
# with the License. You may obtain a copy of the License at
10+
#
11+
# http://www.apache.org/licenses/LICENSE-2.0
12+
#
13+
# Unless required by applicable law or agreed to in writing,
14+
# software distributed under the License is distributed on an
15+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
# KIND, either express or implied. See the License for the
17+
# specific language governing permissions and limitations
18+
# under the License.
19+
#
20+
#-------------------------------------------------------------
21+
22+
# Read the input matrix as a stream
23+
b = read($1);
24+
X = read($2);
25+
26+
OOC = X %*% b;
27+
OOC = OOC %*% t(OOC);
28+
29+
res = solve(OOC, b);
30+
31+
write(res, $3, format="text");

‎src/test/scripts/functions/ooc/StreamCollect_3.dml‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
#-------------------------------------------------------------
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one
4+
# or more contributor license agreements. See the NOTICE file
5+
# distributed with this work for additional information
6+
# regarding copyright ownership. The ASF licenses this file
7+
# to you under the Apache License, Version 2.0 (the
8+
# "License"); you may not use this file except in compliance
9+
# with the License. You may obtain a copy of the License at
10+
#
11+
# http://www.apache.org/licenses/LICENSE-2.0
12+
#
13+
# Unless required by applicable law or agreed to in writing,
14+
# software distributed under the License is distributed on an
15+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
# KIND, either express or implied. See the License for the
17+
# specific language governing permissions and limitations
18+
# under the License.
19+
#
20+
#-------------------------------------------------------------
21+
22+
# Read the input matrix as a stream
23+
b = read($1);
24+
X = read($2);
25+
26+
OOC = ceil(X);
27+
28+
res = solve(OOC, b);
29+
30+
write(res, $3, format="text");

0 commit comments

Comments
 (0)