| /* |
| * 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.lineage; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.LinkedHashMap; |
| import java.util.Map; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.log4j.Level; |
| import org.apache.log4j.Logger; |
| import org.apache.sysds.api.DMLScript; |
| import org.apache.sysds.common.Types.AggOp; |
| import org.apache.sysds.common.Types.Direction; |
| import org.apache.sysds.common.Types.OpOp2; |
| import org.apache.sysds.common.Types.OpOpN; |
| import org.apache.sysds.common.Types.ParamBuiltinOp; |
| import org.apache.sysds.common.Types.ValueType; |
| import org.apache.sysds.hops.AggBinaryOp; |
| import org.apache.sysds.hops.AggUnaryOp; |
| import org.apache.sysds.hops.BinaryOp; |
| import org.apache.sysds.hops.DataOp; |
| import org.apache.sysds.hops.Hop; |
| import org.apache.sysds.hops.IndexingOp; |
| import org.apache.sysds.hops.LiteralOp; |
| import org.apache.sysds.hops.NaryOp; |
| import org.apache.sysds.hops.ParameterizedBuiltinOp; |
| import org.apache.sysds.hops.ReorgOp; |
| import org.apache.sysds.hops.recompile.Recompiler; |
| import org.apache.sysds.hops.rewrite.HopRewriteUtils; |
| import org.apache.sysds.parser.Statement; |
| import org.apache.sysds.runtime.DMLRuntimeException; |
| import org.apache.sysds.runtime.controlprogram.BasicProgramBlock; |
| import org.apache.sysds.runtime.controlprogram.Program; |
| import org.apache.sysds.runtime.controlprogram.caching.MatrixObject; |
| import org.apache.sysds.runtime.controlprogram.context.ExecutionContext; |
| import org.apache.sysds.runtime.controlprogram.context.ExecutionContextFactory; |
| import org.apache.sysds.runtime.instructions.Instruction; |
| import org.apache.sysds.runtime.instructions.InstructionParser; |
| import org.apache.sysds.runtime.instructions.InstructionUtils; |
| import org.apache.sysds.runtime.instructions.cp.ComputationCPInstruction; |
| import org.apache.sysds.runtime.instructions.cp.DataGenCPInstruction; |
| import org.apache.sysds.runtime.instructions.cp.ParameterizedBuiltinCPInstruction; |
| import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction; |
| import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType; |
| import org.apache.sysds.runtime.lineage.LineageItem.LineageItemType; |
| import org.apache.sysds.runtime.matrix.data.MatrixBlock; |
| import org.apache.sysds.runtime.meta.MetaData; |
| import org.apache.sysds.utils.Explain; |
| import org.apache.sysds.utils.Explain.ExplainType; |
| |
| public class LineageRewriteReuse |
| { |
| private static final String LR_VAR = "__lrwrt"; |
| private static BasicProgramBlock _lrPB = null; |
| private static ExecutionContext _lrEC = null; |
| private static boolean _disableReuse = true; |
| private static final Log LOG = LogFactory.getLog(LineageRewriteReuse.class.getName()); |
| |
| public static boolean executeRewrites (Instruction curr, ExecutionContext ec) |
| { |
| ExecutionContext lrwec = getExecutionContext(); |
| ExplainType et = DMLScript.EXPLAIN; |
| // Disable explain not to print unnecessary logs |
| // TODO extend recompiler to allow use without explain output |
| DMLScript.EXPLAIN = ExplainType.NONE; |
| |
| //check applicability and apply rewrite |
| //tsmm(cbind(X, ones)) -> rbind(t(colSums(cbind(X, ones))[, 1:ncol-1]), colSums(cbind(X, ones))) |
| ArrayList<Instruction> newInst = rewriteTsmmCbindOnes(curr, ec, lrwec); |
| //tsmm(cbind(X, deltaX)) -> rbind(cbind(tsmm(X), t(X) %*% deltaX), cbind(t(deltaX) %*%X, tsmm(deltaX))) |
| newInst = (newInst == null) ? rewriteTsmmCbind(curr, ec, lrwec) : newInst; |
| //tsmm(cbind(cbind(X, deltaX), ones)) -> TODO |
| newInst = (newInst == null) ? rewriteTsmm2Cbind(curr, ec, lrwec) : newInst; |
| //tsmm(cbind(cbind(X, deltaX), ones)) -> TODO |
| newInst = (newInst == null) ? rewriteTsmm2CbindSameLeft(curr, ec, lrwec) : newInst; |
| //tsmm(rbind(X, deltaX)) -> tsmm(X) + tsmm(deltaX) |
| newInst = (newInst == null) ? rewriteTsmmRbind(curr, ec, lrwec) : newInst; |
| //rbind(X,deltaX) %*% Y -> rbind(X %*% Y, deltaX %*% Y) |
| newInst = (newInst == null) ? rewriteMatMulRbindLeft(curr, ec, lrwec) : newInst; |
| //X %*% cbind(Y,ones)) -> cbind(X %*% Y, rowSums(X)) |
| newInst = (newInst == null) ? rewriteMatMulCbindRightOnes(curr, ec, lrwec) : newInst; |
| //X %*% cbind(Y,deltaY)) -> cbind(X %*% Y, X %*% deltaY) |
| newInst = (newInst == null) ? rewriteMatMulCbindRight(curr, ec, lrwec) : newInst; |
| //rbind(X, deltaX) * rbind(Y, deltaY) -> rbind(X * Y, deltaX * deltaY) |
| newInst = (newInst == null) ? rewriteElementMulRbind(curr, ec, lrwec) : newInst; |
| //cbind(X, deltaX) * cbind(Y, deltaY) -> cbind(X * Y, deltaX * deltaY) |
| newInst = (newInst == null) ? rewriteElementMulCbind(curr, ec, lrwec) : newInst; |
| //aggregate(target=cbind(X, deltaX,...) = cbind(aggregate(target=X,...), aggregate(target=deltaX,...)) for same agg function |
| newInst = (newInst == null) ? rewriteAggregateCbind(curr, ec, lrwec) : newInst; |
| //A %*% B[,1:k] = (A %*% B)[,1:k]; |
| newInst = (newInst == null) ? rewriteIndexingMatMul(curr, ec, lrwec) : newInst; |
| //PCA --> lmDS pipeline |
| newInst = (newInst == null) ? rewritePcaTsmm(curr, ec, lrwec) : newInst; |
| |
| if (newInst == null) |
| return false; |
| |
| //execute instructions & write the o/p to symbol table |
| long t0 = System.nanoTime(); |
| executeInst(newInst, lrwec); |
| long t1 = System.nanoTime(); |
| ec.setVariable(((ComputationCPInstruction)curr).output.getName(), lrwec.getVariable(LR_VAR)); |
| |
| //put the result into the cache |
| LineageCache.putMatrix(curr, ec, t1-t0); |
| DMLScript.EXPLAIN = et; //TODO can't change this here |
| |
| //cleanup execution context |
| lrwec.getVariables().removeAll(); |
| |
| return true; |
| } |
| |
| /*--------------------------------REWRITE METHODS------------------------------*/ |
| |
| private static ArrayList<Instruction> rewriteTsmmCbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if(!isTsmmCbind(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the cached tsmm result |
| MatrixObject cachedEntry = toMatrixObject(inCache.get("lastMatrix")); |
| lrwec.setVariable("cachedEntry", cachedEntry); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| // Create rightIndex op to find the last matrix |
| // TODO: For now assumption is that a single column is being appended in a loop |
| // Need to go down the lineage to find number of columns are being appended. |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| |
| // Use X from cache, or create rightIndex |
| Hop oldMatrix = inCache.containsKey("X") ? |
| setupTReadCachedInput("X", inCache, lrwec) : |
| HopRewriteUtils.createIndexingOp(newMatrix, 1L, mo.getNumRows(), 1L, mo.getNumColumns()-1); |
| |
| // Use deltaX from cache, or create rightIndex |
| Hop lastCol = inCache.containsKey("deltaX") ? |
| setupTReadCachedInput("deltaX", inCache, lrwec) : |
| HopRewriteUtils.createIndexingOp(newMatrix, 1L, mo.getNumRows(), mo.getNumColumns(), mo.getNumColumns()); |
| |
| Hop lrwHop = HopRewriteUtils.createPartialTsmmCbind(oldMatrix, lastCol, lastRes); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteTsmmCbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "X", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteTsmmCbindOnes (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // This is a specialization of rewriteTsmmCbind. This qualifies if |
| // the appended matrix is a column matrix of 1s (deltaX = 1s). |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if(!isTsmmCbindOnes(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the cached tsmm result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| // Create a transient read op over current input |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("newMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("newMatrix", mo); |
| // rowTwo = colSums(newMatrix) |
| AggUnaryOp rowTwo = HopRewriteUtils.createAggUnaryOp(newMatrix, AggOp.SUM, Direction.Col); |
| // topRight = t(rowTwo[, 1:ncols-1]) |
| IndexingOp tmp = HopRewriteUtils.createIndexingOp(rowTwo, new LiteralOp(1), new LiteralOp(1), |
| new LiteralOp(1), new LiteralOp(mo.getNumColumns()-1)); |
| ReorgOp topRight = HopRewriteUtils.createTranspose(tmp); |
| // rowOne = cbind(lastRes, topRight) |
| BinaryOp rowOne = HopRewriteUtils.createBinary(lastRes, topRight, OpOp2.CBIND); |
| // rbind(rowOne, rowTwo) |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(rowOne, rowTwo, OpOp2.RBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteTsmmCbindOnes APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteTsmmRbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isTsmmRbind(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last tsmm result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| Hop lastRow; |
| // Use deltaX from cache, or create rightIndex |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastRow = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else |
| lastRow = HopRewriteUtils.createIndexingOp(newMatrix, new LiteralOp(mo.getNumRows()), |
| new LiteralOp(mo.getNumRows()), new LiteralOp(1), new LiteralOp(mo.getNumColumns())); |
| // tsmm(X + lastRow) = tsmm(X) + tsmm(lastRow) |
| ReorgOp tlastRow = HopRewriteUtils.createTranspose(lastRow); |
| AggBinaryOp tsmm_lr = HopRewriteUtils.createMatrixMultiply(tlastRow, lastRow); |
| BinaryOp lrwHop = HopRewriteUtils.createBinary(lastRes, tsmm_lr, OpOp2.PLUS); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteTsmmRbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteTsmm2Cbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isTsmm2Cbind(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last tsmm result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| MatrixObject newmo = toMatrixObject(cachedEntry); |
| lrwec.setVariable("cachedEntry", newmo); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| |
| // pull out the newly added column(2nd last) from the input matrix |
| Hop lastCol; |
| // Use deltaX from cache, or create rightIndex |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastCol = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else |
| lastCol = HopRewriteUtils.createIndexingOp(newMatrix, new LiteralOp(1), new LiteralOp(mo.getNumRows()), |
| new LiteralOp(mo.getNumColumns()-1), new LiteralOp(mo.getNumColumns()-1)); |
| // apply t(lastCol) on i/p matrix to get the result vectors. |
| ReorgOp tlastCol = HopRewriteUtils.createTranspose(lastCol); |
| AggBinaryOp newCol = HopRewriteUtils.createMatrixMultiply(tlastCol, newMatrix); |
| ReorgOp tnewCol = HopRewriteUtils.createTranspose(newCol); |
| |
| // push the result row & column inside the cashed block as 2nd last row and col respectively. |
| IndexingOp topLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-1), |
| new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-1)); |
| IndexingOp topRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-1), |
| new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp bottomLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-1)); |
| IndexingOp bottomRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp topCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(1), new LiteralOp(mo.getNumColumns()-2), |
| new LiteralOp(1), new LiteralOp(1)); |
| IndexingOp bottomCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(mo.getNumColumns()), |
| new LiteralOp(mo.getNumColumns()), new LiteralOp(1), new LiteralOp(1)); |
| NaryOp rowOne = HopRewriteUtils.createNary(OpOpN.CBIND, topLeft, topCol, topRight); |
| NaryOp rowTwo = HopRewriteUtils.createNary(OpOpN.CBIND, bottomLeft, bottomCol, bottomRight); |
| NaryOp lrwHop = HopRewriteUtils.createNary(OpOpN.RBIND, rowOne, newCol, rowTwo); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteTsmm2Cbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteTsmm2CbindSameLeft (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| /* The difference between rewriteTsmm2Cbind and this rewrite is that the former applies |
| * when columns are increasingly appended where the later applies when different columns |
| * are appended to a single base matrix. |
| */ |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isTsmm2CbindSameLeft(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last tsmm result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| MatrixObject newmo = toMatrixObject(cachedEntry); |
| lrwec.setVariable("cachedEntry", newmo); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| |
| // Create a transient read op over the input to this tsmm |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| |
| // pull out the newly added column(2nd last) from the input matrix |
| Hop lastCol; |
| // Use deltaX from cache, or create rightIndex |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastCol = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else |
| lastCol = HopRewriteUtils.createIndexingOp(newMatrix, new LiteralOp(1), new LiteralOp(mo.getNumRows()), |
| new LiteralOp(mo.getNumColumns()-1), new LiteralOp(mo.getNumColumns()-1)); |
| |
| // apply t(lastCol) on i/p matrix to get the result vectors. |
| ReorgOp tlastCol = HopRewriteUtils.createTranspose(lastCol); |
| AggBinaryOp newCol = HopRewriteUtils.createMatrixMultiply(tlastCol, newMatrix); |
| ReorgOp tnewCol = HopRewriteUtils.createTranspose(newCol); |
| |
| // Replace the 2nd last row and column of the last tsmm resutl with the result vector. |
| IndexingOp topLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-2), |
| new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-2)); |
| IndexingOp topRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-2), |
| new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp bottomLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-2)); |
| IndexingOp bottomRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp topCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(1), new LiteralOp(mo.getNumColumns()-2), |
| new LiteralOp(1), new LiteralOp(1)); |
| IndexingOp bottomCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(mo.getNumColumns()), |
| new LiteralOp(mo.getNumColumns()), new LiteralOp(1), new LiteralOp(1)); |
| NaryOp rowOne = HopRewriteUtils.createNary(OpOpN.CBIND, topLeft, topCol, topRight); |
| NaryOp rowTwo = HopRewriteUtils.createNary(OpOpN.CBIND, bottomLeft, bottomCol, bottomRight); |
| NaryOp lrwHop = HopRewriteUtils.createNary(OpOpN.RBIND, rowOne, newCol, rowTwo); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteTsmm2CbindSameLeft APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteMatMulRbindLeft (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isMatMulRbindLeft(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last ba+* result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("leftMatrix", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("leftMatrix", moL); |
| MatrixObject moR = ec.getMatrixObject(((ComputationCPInstruction)curr).input2); |
| lrwec.setVariable("rightMatrix", moR); |
| DataOp rightMatrix = HopRewriteUtils.createTransientRead("rightMatrix", moR); |
| Hop lastRow; |
| // Use deltaX from cache, or create rightIndex |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastRow = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| lastRow = HopRewriteUtils.createIndexingOp(leftMatrix, new LiteralOp(moL.getNumRows()), |
| new LiteralOp(moL.getNumRows()), new LiteralOp(1), new LiteralOp(moL.getNumColumns())); |
| // ba+*(X+lastRow, Y) = rbind(ba+*(X, Y), ba+*(lastRow, Y)) |
| AggBinaryOp rowTwo = HopRewriteUtils.createMatrixMultiply(lastRow, rightMatrix); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, rowTwo, OpOp2.RBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteMetMulRbindLeft APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteMatMulCbindRight (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isMatMulCbindRight(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last ba+* result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("leftMatrix", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("leftMatrix", moL); |
| MatrixObject moR = ec.getMatrixObject(((ComputationCPInstruction)curr).input2); |
| lrwec.setVariable("rightMatrix", moR); |
| DataOp rightMatrix = HopRewriteUtils.createTransientRead("rightMatrix", moR); |
| Hop lastCol; |
| // Use deltaY from cache, or create rightIndex |
| if (inCache.containsKey("deltaY")) { |
| MatrixBlock cachedRI = inCache.get("deltaY"); |
| lrwec.setVariable("deltaY", toMatrixObject(cachedRI)); |
| lastCol = HopRewriteUtils.createTransientRead("deltaY", cachedRI); |
| } |
| lastCol = HopRewriteUtils.createIndexingOp(rightMatrix, new LiteralOp(1), new LiteralOp(moR.getNumRows()), |
| new LiteralOp(moR.getNumColumns()), new LiteralOp(moR.getNumColumns())); |
| // ba+*(X, cbind(Y, lastCol)) = cbind(ba+*(X, Y), ba+*(X, lastCol)) |
| AggBinaryOp colTwo = HopRewriteUtils.createMatrixMultiply(leftMatrix, lastCol); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, colTwo, OpOp2.CBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteMatMulCbindRight APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaY"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteMatMulCbindRightOnes (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // This is a specialization of rewriteMatMulCbindRight. This qualifies |
| // if the right matrix is appended with a matrix of 1s (deltaY == 1s). |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isMatMulCbindRightOnes(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last ba+* result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("leftMatrix", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("leftMatrix", moL); |
| // ba+*(X, cbind(Y, ones)) = cbind(ba+*(X, Y), rowSums(X)) |
| AggUnaryOp colTwo = HopRewriteUtils.createAggUnaryOp(leftMatrix, AggOp.SUM, Direction.Row); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, colTwo, OpOp2.CBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteMatMulCbindRightOnes APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteElementMulRbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isElementMulRbind(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last * result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| Hop lastRowL, lastRowR; |
| // Use deltaX and deltaY from cache, or create rightIndices |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastRowL = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else { |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("leftMatrix", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("leftMatrix", moL); |
| lastRowL = HopRewriteUtils.createIndexingOp(leftMatrix, new LiteralOp(moL.getNumRows()), |
| new LiteralOp(moL.getNumRows()), new LiteralOp(1), new LiteralOp(moL.getNumColumns())); |
| } |
| if (inCache.containsKey("deltaY")) { |
| MatrixBlock cachedRI = inCache.get("deltaY"); |
| lrwec.setVariable("deltaY", toMatrixObject(cachedRI)); |
| lastRowR = HopRewriteUtils.createTransientRead("deltaY", cachedRI); |
| } |
| else { |
| MatrixObject moR = ec.getMatrixObject(((ComputationCPInstruction)curr).input2); |
| lrwec.setVariable("rightMatrix", moR); |
| DataOp rightMatrix = HopRewriteUtils.createTransientRead("rightMatrix", moR); |
| lastRowR = HopRewriteUtils.createIndexingOp(rightMatrix, new LiteralOp(moR.getNumRows()), |
| new LiteralOp(moR.getNumRows()), new LiteralOp(1), new LiteralOp(moR.getNumColumns())); |
| } |
| // *(X+lastRowL, Y+lastRowR) = rbind(*(X, Y), *(lastRowL, lastRowR)) |
| BinaryOp rowTwo = HopRewriteUtils.createBinary(lastRowL, lastRowR, OpOp2.MULT); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, rowTwo, OpOp2.RBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteElementMulRbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX", "deltaY"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteElementMulCbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isElementMulCbind(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last * result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| Hop lastColL, lastColR; |
| // Use deltaX and deltaY from cache, or create rightIndices |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastColL = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else { |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("leftMatrix", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("leftMatrix", moL); |
| lastColL = HopRewriteUtils.createIndexingOp(leftMatrix, new LiteralOp(1), |
| new LiteralOp(moL.getNumRows()), new LiteralOp(moL.getNumColumns()), new LiteralOp(moL.getNumColumns())); |
| } |
| if (inCache.containsKey("deltaY")) { |
| MatrixBlock cachedRI = inCache.get("deltaY"); |
| lrwec.setVariable("deltaY", toMatrixObject(cachedRI)); |
| lastColR = HopRewriteUtils.createTransientRead("deltaY", cachedRI); |
| } |
| else { |
| MatrixObject moR = ec.getMatrixObject(((ComputationCPInstruction)curr).input2); |
| lrwec.setVariable("rightMatrix", moR); |
| DataOp rightMatrix = HopRewriteUtils.createTransientRead("rightMatrix", moR); |
| lastColR = HopRewriteUtils.createIndexingOp(rightMatrix, new LiteralOp(1), |
| new LiteralOp(moR.getNumRows()), new LiteralOp(moR.getNumColumns()), new LiteralOp(moR.getNumColumns())); |
| } |
| // *(X+lastRowL, Y+lastRowR) = cbind(*(X, Y), *(lastColL, lastColR)) |
| BinaryOp rowTwo = HopRewriteUtils.createBinary(lastColL, lastColR, OpOp2.MULT); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, rowTwo, OpOp2.CBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteElementMulCbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX", "deltaY"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteAggregateCbind (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isAggCbind (curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last * result |
| MatrixBlock cachedEntry = inCache.get("lastMatrix"); |
| lrwec.setVariable("cachedEntry", toMatrixObject(cachedEntry)); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", cachedEntry); |
| //TODO: support for block of rows |
| HashMap<String, String> params = ((ParameterizedBuiltinCPInstruction)curr).getParameterMap(); |
| MatrixObject mo = ec.getMatrixObject(params.get(Statement.GAGG_TARGET)); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| MatrixObject moG = ec.getMatrixObject(params.get(Statement.GAGG_GROUPS)); |
| lrwec.setVariable("groups", moG); |
| DataOp groups = HopRewriteUtils.createTransientRead("groups", moG); |
| String fn = params.get(Statement.GAGG_FN); |
| int ngroups = (params.get(Statement.GAGG_NUM_GROUPS) != null) ? |
| (int)Double.parseDouble(params.get(Statement.GAGG_NUM_GROUPS)) : -1; |
| Hop lastCol; |
| // Use deltaX from cache, or create rightIndex |
| if (inCache.containsKey("deltaX")) { |
| MatrixBlock cachedRI = inCache.get("deltaX"); |
| lrwec.setVariable("deltaX", toMatrixObject(cachedRI)); |
| lastCol = HopRewriteUtils.createTransientRead("deltaX", cachedRI); |
| } |
| else |
| lastCol = HopRewriteUtils.createIndexingOp(newMatrix, new LiteralOp(1), new LiteralOp(mo.getNumRows()), |
| new LiteralOp(mo.getNumColumns()), new LiteralOp(mo.getNumColumns())); |
| // aggregate(target=X+lastCol,...) = cbind(aggregate(target=X,...), aggregate(target=lastCol,...)) |
| LinkedHashMap<String, Hop> args = new LinkedHashMap<>(); |
| args.put("target", lastCol); |
| args.put("groups", groups); |
| args.put("fn", new LiteralOp(fn)); |
| if (ngroups != -1) |
| args.put("ngroups", new LiteralOp(ngroups)); |
| ParameterizedBuiltinOp rowTwo = HopRewriteUtils.createParameterizedBuiltinOp(newMatrix, args, ParamBuiltinOp.GROUPEDAGG); |
| BinaryOp lrwHop= HopRewriteUtils.createBinary(lastRes, rowTwo, OpOp2.CBIND); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteElementMulCbind APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "deltaX"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewriteIndexingMatMul (Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| /* This rewrite replaces the indexed matrix with its source as an |
| * input to matrix multiplication, with the hope that in future |
| * iterations all the outputs can be sliced out from the cached |
| * result (e.g. PCA in a loop) |
| * Note: this particular rewrite needs to cache the compensation plan |
| * execution results (unlike other rewrites) to be effective. |
| * TODO: Generalize for all cases and move to compiler |
| */ |
| // Check the applicability of this rewrite. |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isIndexingMatMul (curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the input to rightIndex |
| MatrixBlock indexSource = inCache.get("indexSource"); |
| lrwec.setVariable("indexSource", toMatrixObject(indexSource)); |
| DataOp input2Index = HopRewriteUtils.createTransientRead("indexSource", indexSource); |
| // Create or read the matrix multiplication |
| Hop matMultRes; |
| MatrixObject moL = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| if (inCache.containsKey("BigMatMult")) { |
| MatrixObject BigMatMult = toMatrixObject(inCache.get("BigMatMult")); |
| lrwec.setVariable("BigMatMult", BigMatMult); |
| matMultRes = HopRewriteUtils.createTransientRead("BigMatMult", BigMatMult); |
| } |
| else { |
| lrwec.setVariable("left", moL); |
| DataOp leftMatrix = HopRewriteUtils.createTransientRead("left", moL); |
| matMultRes = HopRewriteUtils.createMatrixMultiply(leftMatrix, input2Index); |
| // Perform the multiplication once and cache for future iterations. |
| } |
| // Gather the indexing parameters. |
| MatrixObject moR = ec.getMatrixObject(((ComputationCPInstruction)curr).input2); |
| IndexingOp lrwHop = HopRewriteUtils.createIndexingOp(matMultRes, new LiteralOp(1), |
| new LiteralOp(moL.getNumRows()), new LiteralOp(1), new LiteralOp(moR.getNumColumns())); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewriteIndexingMatMul APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| // Keep reuse enabled |
| _disableReuse = false; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "indexSource", "BigMatMult"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| private static ArrayList<Instruction> rewritePcaTsmm(Instruction curr, ExecutionContext ec, ExecutionContext lrwec) |
| { |
| Map<String, MatrixBlock> inCache = new HashMap<>(); |
| if (!isPcaTsmm(curr, ec, inCache)) |
| return null; |
| |
| // Create a transient read op over the last tsmm result |
| MatrixObject newmo = toMatrixObject(inCache.get("lastMatrix")); |
| lrwec.setVariable("cachedEntry", newmo); |
| DataOp lastRes = HopRewriteUtils.createTransientRead("cachedEntry", newmo); |
| |
| // Create a transient read op over this tsmm's input |
| MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction)curr).input1); |
| lrwec.setVariable("oldMatrix", mo); |
| DataOp newMatrix = HopRewriteUtils.createTransientRead("oldMatrix", mo); |
| |
| // Index out the added column from the projected matrix |
| MatrixObject projmo = toMatrixObject(inCache.get("projected")); |
| lrwec.setVariable("projected", projmo); |
| DataOp projRes = HopRewriteUtils.createTransientRead("projected", projmo); |
| IndexingOp lastCol = HopRewriteUtils.createIndexingOp(projRes, new LiteralOp(1), new LiteralOp(projmo.getNumRows()), |
| new LiteralOp(projmo.getNumColumns()), new LiteralOp(projmo.getNumColumns())); |
| |
| // Apply t(lastCol) on i/p matrix to get the result vectors. |
| ReorgOp tlastCol = HopRewriteUtils.createTranspose(lastCol); |
| AggBinaryOp newCol = HopRewriteUtils.createMatrixMultiply(tlastCol, newMatrix); |
| ReorgOp tnewCol = HopRewriteUtils.createTranspose(newCol); |
| |
| // Push the result row & column inside the cashed block as 2nd last row and col respectively. |
| IndexingOp topLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-1), |
| new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-1)); |
| IndexingOp topRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(1), new LiteralOp(newmo.getNumRows()-1), |
| new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp bottomLeft = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(1), new LiteralOp(newmo.getNumColumns()-1)); |
| IndexingOp bottomRight = HopRewriteUtils.createIndexingOp(lastRes, new LiteralOp(newmo.getNumRows()), |
| new LiteralOp(newmo.getNumRows()), new LiteralOp(newmo.getNumColumns()), new LiteralOp(newmo.getNumColumns())); |
| IndexingOp topCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(1), new LiteralOp(mo.getNumColumns()-2), |
| new LiteralOp(1), new LiteralOp(1)); |
| IndexingOp bottomCol = HopRewriteUtils.createIndexingOp(tnewCol, new LiteralOp(mo.getNumColumns()), |
| new LiteralOp(mo.getNumColumns()), new LiteralOp(1), new LiteralOp(1)); |
| NaryOp rowOne = HopRewriteUtils.createNary(OpOpN.CBIND, topLeft, topCol, topRight); |
| NaryOp rowTwo = HopRewriteUtils.createNary(OpOpN.CBIND, bottomLeft, bottomCol, bottomRight); |
| NaryOp lrwHop = HopRewriteUtils.createNary(OpOpN.RBIND, rowOne, newCol, rowTwo); |
| DataOp lrwWrite = HopRewriteUtils.createTransientWrite(LR_VAR, lrwHop); |
| |
| // Generate runtime instructions |
| if (LOG.isDebugEnabled()) |
| LOG.debug("LINEAGE REWRITE rewritePcaTsmm APPLIED"); |
| ArrayList<Instruction> inst = genInst(lrwWrite, lrwec); |
| _disableReuse = true; |
| |
| // cleanup buffer pool |
| addRmvarInstructions(inst, lrwec, "cachedEntry", "projected"); |
| |
| if (DMLScript.STATISTICS) |
| LineageCacheStatistics.incrementPRewrites(); |
| return inst; |
| } |
| |
| /*------------------------REWRITE APPLICABILITY CHECKS-------------------------*/ |
| |
| private static boolean isTsmmCbind(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) { |
| return false; |
| } |
| |
| // If the input to tsmm came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem source = item.getInputs()[0]; |
| if (source.getOpcode().equalsIgnoreCase("cbind")) { |
| // create tsmm lineage on top of the input of last append |
| LineageItem input1 = source.getInputs()[0]; |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {input1}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the old matrix in cache |
| if( LineageCache.probe(input1) ) |
| inCache.put("X", LineageCache.getMatrix(input1)); |
| // look for the appended column in cache |
| if (LineageCache.probe(source.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(source.getInputs()[1])); |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isTsmmCbindOnes(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) { |
| return false; |
| } |
| |
| // If the input to tsmm came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem source = item.getInputs()[0]; |
| if (source.getOpcode().equalsIgnoreCase("cbind")) { |
| // check if the appended column is a matrix of 1s |
| LineageItem input2 = source.getInputs()[1]; |
| if (input2.getType() != LineageItemType.Creation) |
| return false; |
| Instruction ins = InstructionParser.parseSingleInstruction(input2.getData()); |
| if (!((DataGenCPInstruction)ins).isOnesCol()) |
| return false; |
| // create tsmm lineage on top of the input of last append |
| LineageItem input1 = source.getInputs()[0]; |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {input1}); |
| if( LineageCache.probe(input1) ) |
| inCache.put("X", LineageCache.getMatrix(input1)); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| } |
| } |
| // return true only if the last tsmm result is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isTsmmRbind(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the input to tsmm came from rbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem source = item.getInputs()[0]; |
| if (source.getOpcode().equalsIgnoreCase("rbind")) { |
| // create tsmm lineage on top of the input of last append |
| LineageItem input1 = source.getInputs()[0]; |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {input1}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended column in cache |
| if (source.getInputs().length>1 && LineageCache.probe(source.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(source.getInputs()[1])); |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isTsmm2Cbind (Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| //TODO: support nary cbind |
| // If the input to tsmm came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| // look for two consecutive cbinds |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem source = item.getInputs()[0]; |
| if (source.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem input = source.getInputs()[0]; |
| if (input.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem L2appin1 = input.getInputs()[0]; |
| LineageItem tmp = new LineageItem("cbind", new LineageItem[] {L2appin1, source.getInputs()[1]}); |
| LineageItem toProbe = new LineageItem(curr.getOpcode(), new LineageItem[] {tmp}); |
| if (LineageCache.probe(toProbe)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(toProbe)); |
| // look for the appended column in cache |
| if (LineageCache.probe(input.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(input.getInputs()[1])); |
| } |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isTsmm2CbindSameLeft (Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| //TODO: support nary cbind |
| // If the input to tsmm came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| // look for two consecutive cbinds |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem source = item.getInputs()[0]; |
| if (source.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem input = source.getInputs()[0]; |
| if (input.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem L2appin1 = input.getInputs()[0]; |
| if (!L2appin1.getOpcode().equalsIgnoreCase("rightIndex")) |
| return false; |
| LineageItem RI = input.getInputs()[1]; |
| if (LineageCache.probe(RI)) |
| inCache.put("deltaX", LineageCache.getMatrix(RI)); |
| LineageItem cu = RI.getInputs()[4]; |
| LineageItem old_cu = reduceColByOne(cu); |
| LineageItem old_RI = new LineageItem("rightIndex", new LineageItem[] {RI.getInputs()[0], |
| RI.getInputs()[1], RI.getInputs()[2], old_cu, old_cu}); |
| LineageItem old_cbind = new LineageItem("cbind", new LineageItem[] {L2appin1, old_RI}); |
| LineageItem tmp = new LineageItem("cbind", new LineageItem[] {old_cbind, source.getInputs()[1]}); |
| LineageItem toProbe = new LineageItem(curr.getOpcode(), new LineageItem[] {tmp}); |
| if (LineageCache.probe(toProbe)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(toProbe)); |
| } |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isMatMulRbindLeft(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the left input to ba+* came from rbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("ba+*")) { |
| LineageItem left= item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (left.getOpcode().equalsIgnoreCase("rbind")){ |
| LineageItem leftSource = left.getInputs()[0]; //left inpur of rbind = X |
| // create ba+* lineage on top of the input of last append |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {leftSource, right}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended column in cache |
| if (LineageCache.probe(left.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(left.getInputs()[1])); |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isMatMulCbindRight(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the right input to ba+* came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("ba+*")) { |
| LineageItem left = item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (right.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem rightSource = right.getInputs()[0]; //left inpur of rbind = X |
| // create ba+* lineage on top of the input of last append |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {left, rightSource}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended column in cache |
| if (LineageCache.probe(right.getInputs()[1])) |
| inCache.put("deltaY", LineageCache.getMatrix(right.getInputs()[1])); |
| } |
| } |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isMatMulCbindRightOnes(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the right input to ba+* came from cbind of a matrix and ones. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("ba+*")) { |
| LineageItem left = item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (right.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem rightSource1 = right.getInputs()[0]; //left input of cbind is X |
| LineageItem rightSource2 = right.getInputs()[1]; |
| // check if the right input to cbind is a matrix of 1s. |
| if (rightSource2.getType() != LineageItemType.Creation) |
| return false; |
| Instruction ins = InstructionParser.parseSingleInstruction(rightSource2.getData()); |
| if (!((DataGenCPInstruction)ins).isOnesCol()) |
| return false; |
| // create ba+* lineage on top of the input of last append |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {left, rightSource1}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| } |
| } |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isElementMulRbind(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the inputs to * came from rbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("*")) { |
| LineageItem left= item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (left.getOpcode().equalsIgnoreCase("rbind") && right.getOpcode().equalsIgnoreCase("rbind")){ |
| LineageItem leftSource = left.getInputs()[0]; //left inpur of rbind = X |
| LineageItem rightSource = right.getInputs()[0]; //right inpur of rbind = Y |
| // create * lineage on top of the input of last append |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {leftSource, rightSource}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended rows in cache |
| if (LineageCache.probe(left.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(left.getInputs()[1])); |
| if (LineageCache.probe(right.getInputs()[1])) |
| inCache.put("deltaY", LineageCache.getMatrix(right.getInputs()[1])); |
| } |
| } |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isElementMulCbind(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) |
| return false; |
| |
| // If the inputs to * came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("*")) { |
| LineageItem left= item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (left.getOpcode().equalsIgnoreCase("cbind") && right.getOpcode().equalsIgnoreCase("cbind")){ |
| LineageItem leftSource = left.getInputs()[0]; //left inpur of cbind = X |
| LineageItem rightSource = right.getInputs()[0]; //right inpur of cbind = Y |
| // create * lineage on top of the input of last append |
| LineageItem tmp = new LineageItem(curr.getOpcode(), new LineageItem[] {leftSource, rightSource}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended columns in cache |
| if (LineageCache.probe(left.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(left.getInputs()[1])); |
| if (LineageCache.probe(right.getInputs()[1])) |
| inCache.put("deltaY", LineageCache.getMatrix(right.getInputs()[1])); |
| } |
| } |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isAggCbind (Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) |
| { |
| if (!LineageCacheConfig.isReusable(curr, ec)) { |
| return false; |
| } |
| |
| // If the input to groupedagg came from cbind, look for both the inputs in cache. |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("groupedagg")) { |
| LineageItem target = item.getInputs()[0]; |
| LineageItem groups = item.getInputs()[1]; |
| LineageItem weights = item.getInputs()[2]; |
| LineageItem fn = item.getInputs()[3]; |
| LineageItem ngroups = item.getInputs()[4]; |
| if (target.getOpcode().equalsIgnoreCase("cbind")) { |
| // create groupedagg lineage on top of the input of last append |
| LineageItem input1 = target.getInputs()[0]; |
| LineageItem tmp = new LineageItem(curr.getOpcode(), |
| new LineageItem[] {input1, groups, weights, fn, ngroups}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(tmp)); |
| // look for the appended column in cache |
| if (LineageCache.probe(target.getInputs()[1])) |
| inCache.put("deltaX", LineageCache.getMatrix(target.getInputs()[1])); |
| } |
| } |
| // return true only if the last tsmm is found |
| return inCache.containsKey("lastMatrix") ? true : false; |
| } |
| |
| private static boolean isIndexingMatMul(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) { |
| if (!LineageCacheConfig.isReusable(curr, ec)) { |
| return false; |
| } |
| /* rightIndex -> ba+* is to generic. |
| * Use ba+* -> rightIndex -> ba+* to avoid false positives. |
| * TODO: generalized but robust applicability function |
| */ |
| |
| // Check if the right input of ba+* came from rightindex |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("ba+*")) { |
| LineageItem left = item.getInputs()[0]; |
| LineageItem right = item.getInputs()[1]; |
| if (right.getOpcode().equalsIgnoreCase("rightIndex")) { |
| LineageItem indexSource = right.getInputs()[0]; |
| if (LineageCache.probe(indexSource) && indexSource.getOpcode().equalsIgnoreCase("ba+*")) |
| inCache.put("indexSource", LineageCache.getMatrix(indexSource)); |
| LineageItem tmp = new LineageItem(item.getOpcode(), new LineageItem[] {left, indexSource}); |
| if (LineageCache.probe(tmp)) |
| inCache.put("BigMatMult", LineageCache.getMatrix(tmp)); |
| } |
| } |
| // return true only if the input to rightIndex is found |
| return inCache.containsKey("indexSource") ? true : false; |
| } |
| |
| private static boolean isPcaTsmm(Instruction curr, ExecutionContext ec, Map<String, MatrixBlock> inCache) { |
| if (!LineageCacheConfig.isReusable(curr, ec)) { |
| return false; |
| } |
| |
| LineageItem item = ((ComputationCPInstruction) curr).getLineageItem(ec).getValue(); |
| if (curr.getOpcode().equalsIgnoreCase("tsmm")) { |
| LineageItem src1 = item.getInputs()[0]; |
| if (src1.getOpcode().equalsIgnoreCase("cbind")) { |
| LineageItem src21 = src1.getInputs()[0]; |
| LineageItem src22 = src1.getInputs()[1]; //ones |
| if (src21.getOpcode().equalsIgnoreCase("ba+*")) { |
| if (LineageCache.probe(src21)) |
| inCache.put("projected", LineageCache.getMatrix(src21)); |
| |
| LineageItem src31 = src21.getInputs()[1]; |
| LineageItem src32 = src21.getInputs()[0]; |
| if (src31.getOpcode().equalsIgnoreCase("rightIndex")) { |
| LineageItem cu = src31.getInputs()[4]; |
| //TODO: delta with more than one column |
| LineageItem old_cu = reduceColByOne(cu); |
| LineageItem old_RI = new LineageItem("rightIndex", new LineageItem[] {src31.getInputs()[0], |
| src31.getInputs()[1], src31.getInputs()[2], src31.getInputs()[3], old_cu}); |
| LineageItem old_ba = new LineageItem("ba+*", new LineageItem[] {src32, old_RI}); |
| LineageItem old_cbind = new LineageItem("cbind", new LineageItem[] {old_ba, src22}); |
| LineageItem old_tsmm = new LineageItem("tsmm", new LineageItem[] {old_cbind}); |
| if (LineageCache.probe(old_tsmm)) |
| inCache.put("lastMatrix", LineageCache.getMatrix(old_tsmm)); |
| } |
| } |
| } |
| } |
| return inCache.containsKey("projected") && inCache.containsKey("lastMatrix"); |
| } |
| |
| /*----------------------INSTRUCTIONS GENERATION & EXECUTION-----------------------*/ |
| |
| private static ArrayList<Instruction> genInst(Hop hops, ExecutionContext ec) { |
| ArrayList<Instruction> newInst = Recompiler.recompileHopsDag(hops, ec.getVariables(), null, true, true, 0); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("COMPENSATION PLAN: "); |
| LOG.debug("EXPLAIN LINEAGE REWRITE (HOP) \n" + Explain.explain(hops,1)); |
| LOG.debug("EXPLAIN LINEAGE REWRITE (INSTRUCTION) \n" + Explain.explain(newInst,1)); |
| } |
| return newInst; |
| } |
| |
| private static DataOp setupTReadCachedInput(String name, Map<String, MatrixBlock> inCache, ExecutionContext ec) { |
| MatrixBlock cachedRI = inCache.get(name); |
| ec.setVariable(name, toMatrixObject(cachedRI)); |
| return HopRewriteUtils.createTransientRead(name, cachedRI); |
| } |
| |
| private static void executeInst (ArrayList<Instruction> newInst, ExecutionContext lrwec) |
| { |
| // Disable explain not to print unnecessary logs |
| // TODO extend recompiler to allow use without explain output |
| DMLScript.EXPLAIN = ExplainType.NONE; |
| |
| try { |
| //execute instructions |
| BasicProgramBlock pb = getProgramBlock(); |
| pb.setInstructions(newInst); |
| ReuseCacheType oldReuseOption = DMLScript.LINEAGE_REUSE; |
| if (_disableReuse) |
| LineageCacheConfig.shutdownReuse(); |
| pb.execute(lrwec); |
| if (_disableReuse) |
| LineageCacheConfig.restartReuse(oldReuseOption); |
| } |
| catch (Exception e) { |
| throw new DMLRuntimeException("Error executing lineage rewrites" , e); |
| } |
| } |
| |
| /*-------------------------------UTILITY METHODS----------------------------------*/ |
| |
| private static MatrixObject toMatrixObject(MatrixBlock mb) { |
| MetaData md = new MetaData(mb.getDataCharacteristics()); |
| MatrixObject mo = new MatrixObject(ValueType.FP64, null, md); |
| mo.acquireModify(mb); |
| mo.release(); |
| return mo; |
| } |
| |
| private static void addRmvarInstructions(ArrayList<Instruction> inst, ExecutionContext ec, String... varnames) { |
| //note: we can't directly call rmvar because the instructions are not executed yet |
| ArrayList<String> tmp = new ArrayList<>(); |
| for(String varname : varnames) |
| if(ec.containsVariable(varname)) |
| tmp.add(varname); |
| inst.add(VariableCPInstruction.prepareRemoveInstruction(tmp.toArray(new String[0]))); |
| } |
| |
| private static LineageItem reduceColByOne(LineageItem cu) { |
| String old_data = null; |
| try { |
| String data = cu.getData(); //xx·SCALAR·INT64·true |
| String[] parts = data.split(Instruction.VALUETYPE_PREFIX); |
| float cuNum = Float.valueOf(parts[0]); |
| parts[0] = String.valueOf((int)cuNum-1); |
| old_data = InstructionUtils.concatOperandParts(parts); |
| } |
| catch (Exception e) { |
| throw new DMLRuntimeException("Error reading 'cu' from RightIndex instruction" , e); |
| } |
| return(new LineageItem(old_data)); |
| } |
| |
| private static ExecutionContext getExecutionContext() { |
| if( _lrEC == null ) |
| _lrEC = ExecutionContextFactory.createContext(); |
| return _lrEC; |
| } |
| |
| private static BasicProgramBlock getProgramBlock() { |
| if( _lrPB == null ) |
| _lrPB = new BasicProgramBlock(new Program()); |
| return _lrPB; |
| } |
| } |
