From cf13f721e78565c7075058181f978f2bc58c9512 Mon Sep 17 00:00:00 2001 From: Janardhan Pulivarthi Date: Wed, 27 Aug 2025 19:31:49 +0530 Subject: [PATCH 1/2] [SYSTEMDS-3911] OOC Transpose operation --- .../instructions/OOCInstructionParser.java | 3 + .../instructions/ooc/OOCInstruction.java | 2 +- .../ooc/TransposeOOCInstruction.java | 90 ++++++++++++ .../test/functions/ooc/TransposeTest.java | 133 ++++++++++++++++++ src/test/scripts/functions/ooc/Transpose.dml | 27 ++++ 5 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java create mode 100644 src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java create mode 100644 src/test/scripts/functions/ooc/Transpose.dml diff --git a/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java b/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java index 9b1165b819b..73b5ca02618 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java @@ -29,6 +29,7 @@ import org.apache.sysds.runtime.instructions.ooc.ReblockOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.UnaryOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.MatrixVectorBinaryOOCInstruction; +import org.apache.sysds.runtime.instructions.ooc.TransposeOOCInstruction; public class OOCInstructionParser extends InstructionParser { protected static final Log LOG = LogFactory.getLog(OOCInstructionParser.class.getName()); @@ -60,6 +61,8 @@ public static OOCInstruction parseSingleInstruction(InstructionType ooctype, Str case AggregateBinary: case MAPMM: return MatrixVectorBinaryOOCInstruction.parseInstruction(str); + case Reorg: + return TransposeOOCInstruction.parseInstruction(str); default: throw new DMLRuntimeException("Invalid OOC Instruction Type: " + ooctype); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java index d3c2dfcbd77..0495dcfde51 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java @@ -30,7 +30,7 @@ public abstract class OOCInstruction extends Instruction { protected static final Log LOG = LogFactory.getLog(OOCInstruction.class.getName()); public enum OOCType { - Reblock, AggregateUnary, Binary, Unary, MAPMM, AggregateBinary + Reblock, AggregateUnary, Binary, Unary, MAPMM, Reorg, AggregateBinary } protected final OOCInstruction.OOCType _ooctype; diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java new file mode 100644 index 00000000000..3fe9e2439de --- /dev/null +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java @@ -0,0 +1,90 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.runtime.instructions.ooc; + +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.controlprogram.caching.MatrixObject; +import org.apache.sysds.runtime.controlprogram.context.ExecutionContext; +import org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue; +import org.apache.sysds.runtime.functionobjects.SwapIndex; +import org.apache.sysds.runtime.instructions.InstructionUtils; +import org.apache.sysds.runtime.instructions.cp.CPOperand; +import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.matrix.data.MatrixIndexes; +import org.apache.sysds.runtime.matrix.operators.ReorgOperator; +import org.apache.sysds.runtime.matrix.operators.UnaryOperator; +import org.apache.sysds.runtime.util.CommonThreadPool; + +import java.util.concurrent.ExecutorService; + +public class TransposeOOCInstruction extends ComputationOOCInstruction { + + protected TransposeOOCInstruction(OOCType type, ReorgOperator op, CPOperand in1, CPOperand out, String opcode, String istr) { + super(type, op, in1, out, opcode, istr); + + } + + public static TransposeOOCInstruction parseInstruction(String str) { + String[] parts = InstructionUtils.getInstructionPartsWithValueType(str); + InstructionUtils.checkNumFields(parts, 2); + String opcode = parts[0]; + CPOperand in1 = new CPOperand(parts[1]); + CPOperand out = new CPOperand(parts[2]); + + ReorgOperator reorg = new ReorgOperator(SwapIndex.getSwapIndexFnObject()); + return new TransposeOOCInstruction(OOCType.Reorg, reorg, in1, out, opcode, str); + } + + public void processInstruction( ExecutionContext ec ) { + + // Create thread and process the transpose operation + MatrixObject min = ec.getMatrixObject(input1); + LocalTaskQueue qIn = min.getStreamHandle(); + LocalTaskQueue qOut = new LocalTaskQueue<>(); + ec.getMatrixObject(output).setStreamHandle(qOut); + + + ExecutorService pool = CommonThreadPool.get(); + try { + pool.submit(() -> { + IndexedMatrixValue tmp = null; + try { + while ((tmp = qIn.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) { + MatrixBlock inBlock = (MatrixBlock)tmp.getValue(); + long oldRowIdx = tmp.getIndexes().getRowIndex(); + long oldColIdx = tmp.getIndexes().getColumnIndex(); + + MatrixBlock outBlock = inBlock.reorgOperations((ReorgOperator) _optr, new MatrixBlock(), -1, -1, -1); + qOut.enqueueTask(new IndexedMatrixValue(new MatrixIndexes(oldColIdx, oldRowIdx), outBlock)); + } + qOut.closeInput(); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + }); + } catch (Exception ex) { + throw new DMLRuntimeException(ex); + } finally { + pool.shutdown(); + } + } +} diff --git a/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java b/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java new file mode 100644 index 00000000000..4ea1f888ee6 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.test.functions.ooc; + +import org.apache.sysds.common.Opcodes; +import org.apache.sysds.common.Types; +import org.apache.sysds.runtime.instructions.Instruction; +import org.apache.sysds.runtime.io.MatrixWriter; +import org.apache.sysds.runtime.io.MatrixWriterFactory; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; +import org.apache.sysds.runtime.util.DataConverter; +import org.apache.sysds.runtime.util.HDFSTool; +import org.apache.sysds.test.AutomatedTestBase; +import org.apache.sysds.test.TestConfiguration; +import org.apache.sysds.test.TestUtils; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; + +public class TransposeTest extends AutomatedTestBase { + private final static String TEST_NAME1 = "Transpose"; + private final static String TEST_DIR = "functions/ooc/"; + private final static String TEST_CLASS_DIR = TEST_DIR + TransposeTest.class.getSimpleName() + "/"; + private final static double eps = 1e-10; + private static final String INPUT_NAME = "X"; + private static final String OUTPUT_NAME = "res"; + + private final static int rows = 1000; + private final static int cols_wide = 1000; + private final static int cols_skinny = 500; + + private final static double sparsity1 = 0.7; + private final static double sparsity2 = 0.1; + + @Override + public void setUp() { + TestUtils.clearAssertionInformation(); + TestConfiguration config = new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1); + addTestConfiguration(TEST_NAME1, config); + } + + @Test + public void testTranspose1() { + runTransposeTest(cols_wide, false); + } + +// @Test +// public void testTranspose2() { +// runTransposeTest(cols_skinny, false); +// } + + private void runTransposeTest(int cols, boolean sparse ) + { + Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE); + + try + { + getAndLoadTestConfiguration(TEST_NAME1); + String HOME = SCRIPT_DIR + TEST_DIR; + fullDMLScriptName = HOME + TEST_NAME1 + ".dml"; + programArgs = new String[]{"-explain", "-stats", "-ooc", + "-args", input(INPUT_NAME), output(OUTPUT_NAME)}; + + // 1. Generate the data as MatrixBlock object + double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse?sparsity2:sparsity1, 10); + + // 2. Convert the double arrays to MatrixBlock object + MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data); + + // 3. Create a binary matrix writer + MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY); + + // 4. Write matrix A to a binary SequenceFile + writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME), rows, cols, 1000, A_mb.getNonZeros()); + HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64, + new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY); + + boolean exceptionExpected = false; + runTest(true, exceptionExpected, null, -1); + + double[][] C1 = readMatrix(output(OUTPUT_NAME), Types.FileFormat.BINARY, rows, cols, 1000, 1000); + double result = 0.0; + for(int i = 0; i < rows; i++) { // verify the results with Java + double expected = 0.0; + for(int j = 0; j < cols; j++) { + expected = A_mb.get(i, j); + result = C1[j][i]; + Assert.assertEquals(expected, result, eps); + } + + } + + String prefix = Instruction.OOC_INST_PREFIX; + Assert.assertTrue("OOC wasn't used for RBLK", + heavyHittersContainsString(prefix + Opcodes.RBLK)); + Assert.assertTrue("OOC wasn't used for TRANSPOSE", + heavyHittersContainsString(prefix + Opcodes.TRANSPOSE)); + } + catch (IOException e) { + throw new RuntimeException(e); + } + finally { + resetExecMode(platformOld); + } + } + + private static double[][] readMatrix(String fname, Types.FileFormat fmt, long rows, long cols, int brows, int bcols ) + throws IOException + { + MatrixBlock mb = DataConverter.readMatrixFromHDFS(fname, fmt, rows, cols, brows, bcols); + double[][] C = DataConverter.convertToDoubleMatrix(mb); + return C; + } +} diff --git a/src/test/scripts/functions/ooc/Transpose.dml b/src/test/scripts/functions/ooc/Transpose.dml new file mode 100644 index 00000000000..9b38939a2e1 --- /dev/null +++ b/src/test/scripts/functions/ooc/Transpose.dml @@ -0,0 +1,27 @@ +#------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +#------------------------------------------------------------- + +# Read the input matrix as a stream +X = read($1); + +res = t(X); + +write(res, $2, format="binary"); From 0967536c05c8afbb3a3a346eb24f1c3aef8391cf Mon Sep 17 00:00:00 2001 From: Janardhan Pulivarthi Date: Thu, 28 Aug 2025 08:05:18 +0530 Subject: [PATCH 2/2] fix formatting --- .../ooc/TransposeOOCInstruction.java | 84 ++++----- .../test/functions/ooc/TransposeTest.java | 174 +++++++++--------- 2 files changed, 129 insertions(+), 129 deletions(-) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java index 3fe9e2439de..fffd7ee7ed5 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TransposeOOCInstruction.java @@ -37,54 +37,54 @@ public class TransposeOOCInstruction extends ComputationOOCInstruction { - protected TransposeOOCInstruction(OOCType type, ReorgOperator op, CPOperand in1, CPOperand out, String opcode, String istr) { - super(type, op, in1, out, opcode, istr); + protected TransposeOOCInstruction(OOCType type, ReorgOperator op, CPOperand in1, CPOperand out, String opcode, String istr) { + super(type, op, in1, out, opcode, istr); - } + } - public static TransposeOOCInstruction parseInstruction(String str) { - String[] parts = InstructionUtils.getInstructionPartsWithValueType(str); - InstructionUtils.checkNumFields(parts, 2); - String opcode = parts[0]; - CPOperand in1 = new CPOperand(parts[1]); - CPOperand out = new CPOperand(parts[2]); + public static TransposeOOCInstruction parseInstruction(String str) { + String[] parts = InstructionUtils.getInstructionPartsWithValueType(str); + InstructionUtils.checkNumFields(parts, 2); + String opcode = parts[0]; + CPOperand in1 = new CPOperand(parts[1]); + CPOperand out = new CPOperand(parts[2]); - ReorgOperator reorg = new ReorgOperator(SwapIndex.getSwapIndexFnObject()); - return new TransposeOOCInstruction(OOCType.Reorg, reorg, in1, out, opcode, str); - } + ReorgOperator reorg = new ReorgOperator(SwapIndex.getSwapIndexFnObject()); + return new TransposeOOCInstruction(OOCType.Reorg, reorg, in1, out, opcode, str); + } - public void processInstruction( ExecutionContext ec ) { + public void processInstruction( ExecutionContext ec ) { - // Create thread and process the transpose operation - MatrixObject min = ec.getMatrixObject(input1); - LocalTaskQueue qIn = min.getStreamHandle(); - LocalTaskQueue qOut = new LocalTaskQueue<>(); - ec.getMatrixObject(output).setStreamHandle(qOut); + // Create thread and process the transpose operation + MatrixObject min = ec.getMatrixObject(input1); + LocalTaskQueue qIn = min.getStreamHandle(); + LocalTaskQueue qOut = new LocalTaskQueue<>(); + ec.getMatrixObject(output).setStreamHandle(qOut); - ExecutorService pool = CommonThreadPool.get(); - try { - pool.submit(() -> { - IndexedMatrixValue tmp = null; - try { - while ((tmp = qIn.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) { - MatrixBlock inBlock = (MatrixBlock)tmp.getValue(); - long oldRowIdx = tmp.getIndexes().getRowIndex(); - long oldColIdx = tmp.getIndexes().getColumnIndex(); + ExecutorService pool = CommonThreadPool.get(); + try { + pool.submit(() -> { + IndexedMatrixValue tmp = null; + try { + while ((tmp = qIn.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) { + MatrixBlock inBlock = (MatrixBlock)tmp.getValue(); + long oldRowIdx = tmp.getIndexes().getRowIndex(); + long oldColIdx = tmp.getIndexes().getColumnIndex(); - MatrixBlock outBlock = inBlock.reorgOperations((ReorgOperator) _optr, new MatrixBlock(), -1, -1, -1); - qOut.enqueueTask(new IndexedMatrixValue(new MatrixIndexes(oldColIdx, oldRowIdx), outBlock)); - } - qOut.closeInput(); - } - catch(Exception ex) { - throw new DMLRuntimeException(ex); - } - }); - } catch (Exception ex) { - throw new DMLRuntimeException(ex); - } finally { - pool.shutdown(); - } - } + MatrixBlock outBlock = inBlock.reorgOperations((ReorgOperator) _optr, new MatrixBlock(), -1, -1, -1); + qOut.enqueueTask(new IndexedMatrixValue(new MatrixIndexes(oldColIdx, oldRowIdx), outBlock)); + } + qOut.closeInput(); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + }); + } catch (Exception ex) { + throw new DMLRuntimeException(ex); + } finally { + pool.shutdown(); + } + } } diff --git a/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java b/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java index 4ea1f888ee6..0dc04043d49 100644 --- a/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java +++ b/src/test/java/org/apache/sysds/test/functions/ooc/TransposeTest.java @@ -37,97 +37,97 @@ import java.io.IOException; public class TransposeTest extends AutomatedTestBase { - private final static String TEST_NAME1 = "Transpose"; - private final static String TEST_DIR = "functions/ooc/"; - private final static String TEST_CLASS_DIR = TEST_DIR + TransposeTest.class.getSimpleName() + "/"; - private final static double eps = 1e-10; - private static final String INPUT_NAME = "X"; - private static final String OUTPUT_NAME = "res"; - - private final static int rows = 1000; - private final static int cols_wide = 1000; - private final static int cols_skinny = 500; - - private final static double sparsity1 = 0.7; - private final static double sparsity2 = 0.1; - - @Override - public void setUp() { - TestUtils.clearAssertionInformation(); - TestConfiguration config = new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1); - addTestConfiguration(TEST_NAME1, config); - } - - @Test - public void testTranspose1() { - runTransposeTest(cols_wide, false); - } + private final static String TEST_NAME1 = "Transpose"; + private final static String TEST_DIR = "functions/ooc/"; + private final static String TEST_CLASS_DIR = TEST_DIR + TransposeTest.class.getSimpleName() + "/"; + private final static double eps = 1e-10; + private static final String INPUT_NAME = "X"; + private static final String OUTPUT_NAME = "res"; + + private final static int rows = 1000; + private final static int cols_wide = 1000; + private final static int cols_skinny = 500; + + private final static double sparsity1 = 0.7; + private final static double sparsity2 = 0.1; + + @Override + public void setUp() { + TestUtils.clearAssertionInformation(); + TestConfiguration config = new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1); + addTestConfiguration(TEST_NAME1, config); + } + + @Test + public void testTranspose1() { + runTransposeTest(cols_wide, false); + } // @Test // public void testTranspose2() { // runTransposeTest(cols_skinny, false); // } - private void runTransposeTest(int cols, boolean sparse ) - { - Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE); - - try - { - getAndLoadTestConfiguration(TEST_NAME1); - String HOME = SCRIPT_DIR + TEST_DIR; - fullDMLScriptName = HOME + TEST_NAME1 + ".dml"; - programArgs = new String[]{"-explain", "-stats", "-ooc", - "-args", input(INPUT_NAME), output(OUTPUT_NAME)}; - - // 1. Generate the data as MatrixBlock object - double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse?sparsity2:sparsity1, 10); - - // 2. Convert the double arrays to MatrixBlock object - MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data); - - // 3. Create a binary matrix writer - MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY); - - // 4. Write matrix A to a binary SequenceFile - writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME), rows, cols, 1000, A_mb.getNonZeros()); - HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64, - new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY); - - boolean exceptionExpected = false; - runTest(true, exceptionExpected, null, -1); - - double[][] C1 = readMatrix(output(OUTPUT_NAME), Types.FileFormat.BINARY, rows, cols, 1000, 1000); - double result = 0.0; - for(int i = 0; i < rows; i++) { // verify the results with Java - double expected = 0.0; - for(int j = 0; j < cols; j++) { - expected = A_mb.get(i, j); - result = C1[j][i]; - Assert.assertEquals(expected, result, eps); - } - - } - - String prefix = Instruction.OOC_INST_PREFIX; - Assert.assertTrue("OOC wasn't used for RBLK", - heavyHittersContainsString(prefix + Opcodes.RBLK)); - Assert.assertTrue("OOC wasn't used for TRANSPOSE", - heavyHittersContainsString(prefix + Opcodes.TRANSPOSE)); - } - catch (IOException e) { - throw new RuntimeException(e); - } - finally { - resetExecMode(platformOld); - } - } - - private static double[][] readMatrix(String fname, Types.FileFormat fmt, long rows, long cols, int brows, int bcols ) - throws IOException - { - MatrixBlock mb = DataConverter.readMatrixFromHDFS(fname, fmt, rows, cols, brows, bcols); - double[][] C = DataConverter.convertToDoubleMatrix(mb); - return C; - } + private void runTransposeTest(int cols, boolean sparse ) + { + Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE); + + try + { + getAndLoadTestConfiguration(TEST_NAME1); + String HOME = SCRIPT_DIR + TEST_DIR; + fullDMLScriptName = HOME + TEST_NAME1 + ".dml"; + programArgs = new String[]{"-explain", "-stats", "-ooc", + "-args", input(INPUT_NAME), output(OUTPUT_NAME)}; + + // 1. Generate the data as MatrixBlock object + double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse?sparsity2:sparsity1, 10); + + // 2. Convert the double arrays to MatrixBlock object + MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data); + + // 3. Create a binary matrix writer + MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY); + + // 4. Write matrix A to a binary SequenceFile + writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME), rows, cols, 1000, A_mb.getNonZeros()); + HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64, + new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY); + + boolean exceptionExpected = false; + runTest(true, exceptionExpected, null, -1); + + double[][] C1 = readMatrix(output(OUTPUT_NAME), Types.FileFormat.BINARY, rows, cols, 1000, 1000); + double result = 0.0; + for(int i = 0; i < rows; i++) { // verify the results with Java + double expected = 0.0; + for(int j = 0; j < cols; j++) { + expected = A_mb.get(i, j); + result = C1[j][i]; + Assert.assertEquals(expected, result, eps); + } + + } + + String prefix = Instruction.OOC_INST_PREFIX; + Assert.assertTrue("OOC wasn't used for RBLK", + heavyHittersContainsString(prefix + Opcodes.RBLK)); + Assert.assertTrue("OOC wasn't used for TRANSPOSE", + heavyHittersContainsString(prefix + Opcodes.TRANSPOSE)); + } + catch (IOException e) { + throw new RuntimeException(e); + } + finally { + resetExecMode(platformOld); + } + } + + private static double[][] readMatrix(String fname, Types.FileFormat fmt, long rows, long cols, int brows, int bcols ) + throws IOException + { + MatrixBlock mb = DataConverter.readMatrixFromHDFS(fname, fmt, rows, cols, brows, bcols); + double[][] C = DataConverter.convertToDoubleMatrix(mb); + return C; + } }