/*
 * 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 org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.sysds.runtime.instructions.spark.RandSPInstruction;
import org.apache.sysds.runtime.io.IOUtilFunctions;
import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.apache.sysds.runtime.lineage.LineageItem.LineageItemType;
import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.AggOp;
import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.common.Types.Direction;
import org.apache.sysds.common.Types.OpOp1;
import org.apache.sysds.common.Types.OpOp2;
import org.apache.sysds.common.Types.OpOp3;
import org.apache.sysds.common.Types.OpOpDG;
import org.apache.sysds.common.Types.OpOpData;
import org.apache.sysds.common.Types.OpOpN;
import org.apache.sysds.common.Types.ParamBuiltinOp;
import org.apache.sysds.common.Types.ReOrgOp;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.conf.ConfigurationManager;
import org.apache.sysds.hops.AggBinaryOp;
import org.apache.sysds.hops.AggUnaryOp;
import org.apache.sysds.hops.BinaryOp;
import org.apache.sysds.hops.DataGenOp;
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.ReorgOp;
import org.apache.sysds.hops.TernaryOp;
import org.apache.sysds.hops.UnaryOp;
import org.apache.sysds.hops.codegen.SpoofFusedOp;
import org.apache.sysds.hops.rewrite.HopRewriteUtils;
import org.apache.sysds.lops.Lop;
import org.apache.sysds.lops.PartialAggregate;
import org.apache.sysds.lops.UnaryCP;
import org.apache.sysds.lops.compile.Dag;
import org.apache.sysds.parser.DataExpression;
import org.apache.sysds.parser.DataIdentifier;
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.CacheableData;
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.CPOperand;
import org.apache.sysds.runtime.instructions.cp.Data;
import org.apache.sysds.runtime.instructions.cp.DataGenCPInstruction;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
import org.apache.sysds.runtime.instructions.cp.ScalarObjectFactory;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
import org.apache.sysds.runtime.instructions.spark.SPInstruction.SPType;
import org.apache.sysds.runtime.instructions.cp.CPInstruction.CPType;
import org.apache.sysds.runtime.util.HDFSTool;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.Stack;
import java.util.stream.Collectors;

public class LineageItemUtils {
	
	private static final String LVARPREFIX = "lvar";
	private static final String LPLACEHOLDER = "IN#";
	
	public static LineageItemType getType(String str) {
		if (str.length() == 1) {
			switch (str) {
				case "C":
					return LineageItemType.Creation;
				case "L":
					return LineageItemType.Literal;
				case "I":
					return LineageItemType.Instruction;
				case "D":
					return LineageItemType.Dedup;
				default:
					throw new DMLRuntimeException("Unknown LineageItemType given!");
			}
		} else
			throw new DMLRuntimeException("Unknown LineageItemType given!");
	}
	
	private static String getString(LineageItemType lit) {
		switch (lit) {
			case Creation:
				return "C";
			case Literal:
				return "L";
			case Instruction:
				return "I";
			case Dedup:
				return "D";
			default:
				throw new DMLRuntimeException("Unknown LineageItemType given!");
		}
	}
	
	private static String getString(LineageItem li) {
		return getString(li.getType());
	}
	
	public static String explainSingleLineageItem(LineageItem li) {
		StringBuilder sb = new StringBuilder();
		sb.append("(").append(li.getId()).append(") ");
		sb.append("(").append(getString(li)).append(") ");
		
		if (li.isLeaf()) {
			sb.append(li.getData()).append(" ");
		} else {
			if (li.getType() == LineageItemType.Dedup)
				sb.append(li.getOpcode()).append(li.getData()).append(" ");
			else
				sb.append(li.getOpcode()).append(" ");
			
			String ids = Arrays.stream(li.getInputs())
				.map(i -> String.format("(%d)", i.getId()))
				.collect(Collectors.joining(" "));
			sb.append(ids);
		}
		return sb.toString().trim();
	}
	
	public static Data computeByLineage(LineageItem root) {
		long rootId = root.getOpcode().equals("write") ?
			root.getInputs()[0].getId() : root.getId();
		String varname = LVARPREFIX + rootId;
		
		//recursively construct hops 
		root.resetVisitStatusNR();
		Map<Long, Hop> operands = new HashMap<>();
		rConstructHops(root, operands);
		Hop out = HopRewriteUtils.createTransientWrite(
			varname, operands.get(rootId));
		
		//generate instructions for temporary hops
		ExecutionContext ec = ExecutionContextFactory.createContext();
		BasicProgramBlock pb = new BasicProgramBlock(new Program());
		Dag<Lop> dag = new Dag<>();
		Lop lops = out.constructLops();
		lops.addToDag(dag);
		pb.setInstructions(dag.getJobs(null,
			ConfigurationManager.getDMLConfig()));
		
		// reset cache due to cleaned data objects
		LineageCache.resetCache();
		//execute instructions and get result
		pb.execute(ec);
		return ec.getVariable(varname);
	}
	
	public static LineageItem[] getLineage(ExecutionContext ec, CPOperand... operands) {
		return Arrays.stream(operands).filter(c -> c!=null)
			.map(c -> ec.getLineage().getOrCreate(c)).toArray(LineageItem[]::new);
	}
	
	private static void rConstructHops(LineageItem item, Map<Long, Hop> operands) {
		if (item.isVisited())
			return;
		
		//recursively process children (ordering by data dependencies)
		if (!item.isLeaf())
			for (LineageItem c : item.getInputs())
				rConstructHops(c, operands);
		
		//process current lineage item
		//NOTE: we generate instructions from hops (but without rewrites) to automatically
		//handle execution types, rmvar instructions, and rewiring of inputs/outputs
		switch (item.getType()) {
			case Creation: {
				Instruction inst = InstructionParser.parseSingleInstruction(item.getData());
				
				if (inst instanceof DataGenCPInstruction) {
					DataGenCPInstruction rand = (DataGenCPInstruction) inst;
					HashMap<String, Hop> params = new HashMap<>();
					if( rand.getOpcode().equals("rand") ) {
						if( rand.output.getDataType() == DataType.TENSOR)
							params.put(DataExpression.RAND_DIMS, new LiteralOp(rand.getDims()));
						else {
							params.put(DataExpression.RAND_ROWS, new LiteralOp(rand.getRows()));
							params.put(DataExpression.RAND_COLS, new LiteralOp(rand.getCols()));
						}
						params.put(DataExpression.RAND_MIN, new LiteralOp(rand.getMinValue()));
						params.put(DataExpression.RAND_MAX, new LiteralOp(rand.getMaxValue()));
						params.put(DataExpression.RAND_PDF, new LiteralOp(rand.getPdf()));
						params.put(DataExpression.RAND_LAMBDA, new LiteralOp(rand.getPdfParams()));
						params.put(DataExpression.RAND_SPARSITY, new LiteralOp(rand.getSparsity()));
						params.put(DataExpression.RAND_SEED, new LiteralOp(rand.getSeed()));
					}
					else if( rand.getOpcode().equals("seq") ) {
						params.put(Statement.SEQ_FROM, new LiteralOp(rand.getFrom()));
						params.put(Statement.SEQ_TO, new LiteralOp(rand.getTo()));
						params.put(Statement.SEQ_INCR, new LiteralOp(rand.getIncr()));
					}
					Hop datagen = new DataGenOp(OpOpDG.valueOf(rand.getOpcode().toUpperCase()),
						new DataIdentifier("tmp"), params);
					datagen.setBlocksize(rand.getBlocksize());
					operands.put(item.getId(), datagen);
				} else if (inst instanceof VariableCPInstruction
						&& ((VariableCPInstruction) inst).isCreateVariable()) {
					String parts[] = InstructionUtils.getInstructionPartsWithValueType(inst.toString());
					DataType dt = DataType.valueOf(parts[4]);
					ValueType vt = dt == DataType.MATRIX ? ValueType.FP64 : ValueType.STRING;
					HashMap<String, Hop> params = new HashMap<>();
					params.put(DataExpression.IO_FILENAME, new LiteralOp(parts[2]));
					params.put(DataExpression.READROWPARAM, new LiteralOp(Long.parseLong(parts[6])));
					params.put(DataExpression.READCOLPARAM, new LiteralOp(Long.parseLong(parts[7])));
					params.put(DataExpression.READNNZPARAM, new LiteralOp(Long.parseLong(parts[8])));
					params.put(DataExpression.FORMAT_TYPE, new LiteralOp(parts[5]));
					DataOp pread = new DataOp(parts[1].substring(5), dt, vt, OpOpData.PERSISTENTREAD, params);
					pread.setFileName(parts[2]);
					operands.put(item.getId(), pread);
				}
				else if  (inst instanceof RandSPInstruction) {
					RandSPInstruction rand = (RandSPInstruction) inst;
					HashMap<String, Hop> params = new HashMap<>();
					if (rand.output.getDataType() == DataType.TENSOR)
						params.put(DataExpression.RAND_DIMS, new LiteralOp(rand.getDims()));
					else {
						params.put(DataExpression.RAND_ROWS, new LiteralOp(rand.getRows()));
						params.put(DataExpression.RAND_COLS, new LiteralOp(rand.getCols()));
					}
					params.put(DataExpression.RAND_MIN, new LiteralOp(rand.getMinValue()));
					params.put(DataExpression.RAND_MAX, new LiteralOp(rand.getMaxValue()));
					params.put(DataExpression.RAND_PDF, new LiteralOp(rand.getPdf()));
					params.put(DataExpression.RAND_LAMBDA, new LiteralOp(rand.getPdfParams()));
					params.put(DataExpression.RAND_SPARSITY, new LiteralOp(rand.getSparsity()));
					params.put(DataExpression.RAND_SEED, new LiteralOp(rand.getSeed()));
					Hop datagen = new DataGenOp(OpOpDG.RAND, new DataIdentifier("tmp"), params);
					datagen.setBlocksize(rand.getBlocksize());
					operands.put(item.getId(), datagen);
				}
				break;
			}
			case Instruction: {
				CPType ctype = InstructionUtils.getCPTypeByOpcode(item.getOpcode());
				SPType stype = InstructionUtils.getSPTypeByOpcode(item.getOpcode());
				
				if (ctype != null) {
					switch (ctype) {
						case AggregateUnary: {
							Hop input = operands.get(item.getInputs()[0].getId());
							Hop aggunary = InstructionUtils.isUnaryMetadata(item.getOpcode()) ?
								HopRewriteUtils.createUnary(input, OpOp1.valueOfByOpcode(item.getOpcode())) :
								HopRewriteUtils.createAggUnaryOp(input, item.getOpcode());
							operands.put(item.getId(), aggunary);
							break;
						}
						case AggregateBinary: {
							Hop input1 = operands.get(item.getInputs()[0].getId());
							Hop input2 = operands.get(item.getInputs()[1].getId());
							Hop aggbinary = HopRewriteUtils.createMatrixMultiply(input1, input2);
							operands.put(item.getId(), aggbinary);
							break;
						}
						case AggregateTernary: {
							Hop input1 = operands.get(item.getInputs()[0].getId());
							Hop input2 = operands.get(item.getInputs()[1].getId());
							Hop input3 = operands.get(item.getInputs()[2].getId());
							Hop aggternary = HopRewriteUtils.createSum(
								HopRewriteUtils.createBinary(
								HopRewriteUtils.createBinary(input1, input2, OpOp2.MULT),
								input3, OpOp2.MULT));
							operands.put(item.getId(), aggternary);
							break;
						}
						case Unary:
						case Builtin: {
							Hop input = operands.get(item.getInputs()[0].getId());
							Hop unary = HopRewriteUtils.createUnary(input, item.getOpcode());
							operands.put(item.getId(), unary);
							break;
						}
						case Reorg: {
							operands.put(item.getId(), HopRewriteUtils.createReorg(
								operands.get(item.getInputs()[0].getId()), item.getOpcode()));
							break;
						}
						case Reshape: {
							ArrayList<Hop> inputs = new ArrayList<>();
							for(int i=0; i<5; i++)
								inputs.add(operands.get(item.getInputs()[i].getId()));
							operands.put(item.getId(), HopRewriteUtils.createReorg(inputs, ReOrgOp.RESHAPE));
							break;
						}
						case Binary: {
							//handle special cases of binary operations 
							String opcode = ("^2".equals(item.getOpcode()) 
								|| "*2".equals(item.getOpcode())) ? 
								item.getOpcode().substring(0, 1) : item.getOpcode();
							Hop input1 = operands.get(item.getInputs()[0].getId());
							Hop input2 = operands.get(item.getInputs()[1].getId());
							Hop binary = HopRewriteUtils.createBinary(input1, input2, opcode);
							operands.put(item.getId(), binary);
							break;
						}
						case Ternary: {
							operands.put(item.getId(), HopRewriteUtils.createTernary(
								operands.get(item.getInputs()[0].getId()), 
								operands.get(item.getInputs()[1].getId()), 
								operands.get(item.getInputs()[2].getId()), item.getOpcode()));
							break;
						}
						case Ctable: { //e.g., ctable 
							if( item.getInputs().length==3 )
								operands.put(item.getId(), HopRewriteUtils.createTernary(
									operands.get(item.getInputs()[0].getId()),
									operands.get(item.getInputs()[1].getId()),
									operands.get(item.getInputs()[2].getId()), OpOp3.CTABLE));
							else if( item.getInputs().length==5 )
								operands.put(item.getId(), HopRewriteUtils.createTernary(
									operands.get(item.getInputs()[0].getId()),
									operands.get(item.getInputs()[1].getId()),
									operands.get(item.getInputs()[2].getId()),
									operands.get(item.getInputs()[3].getId()),
									operands.get(item.getInputs()[4].getId()), OpOp3.CTABLE));
							break;
						}
						case BuiltinNary: {
							String opcode = item.getOpcode().equals("n+") ? "plus" : item.getOpcode();
							operands.put(item.getId(), HopRewriteUtils.createNary(
								OpOpN.valueOf(opcode.toUpperCase()), createNaryInputs(item, operands)));
							break;
						}
						case ParameterizedBuiltin: {
							operands.put(item.getId(), constructParameterizedBuiltinOp(item, operands));
							break;
						}
						case MatrixIndexing: {
							operands.put(item.getId(), constructIndexingOp(item, operands));
							break;
						}
						case MMTSJ: {
							//TODO handling of tsmm type left and right -> placement transpose
							Hop input = operands.get(item.getInputs()[0].getId());
							Hop aggunary = HopRewriteUtils.createMatrixMultiply(
								HopRewriteUtils.createTranspose(input), input);
							operands.put(item.getId(), aggunary);
							break;
						}
						case Variable: {
							if( item.getOpcode().startsWith("cast") )
								operands.put(item.getId(), HopRewriteUtils.createUnary(
									operands.get(item.getInputs()[0].getId()),
									OpOp1.valueOfByOpcode(item.getOpcode())));
							else //cpvar, write
								operands.put(item.getId(), operands.get(item.getInputs()[0].getId()));
							break;
						}
						default:
							throw new DMLRuntimeException("Unsupported instruction "
								+ "type: " + ctype.name() + " (" + item.getOpcode() + ").");
					}
				}
				else if( stype != null ) {
					switch(stype) {
						case Reblock: {
							Hop input = operands.get(item.getInputs()[0].getId());
							input.setBlocksize(ConfigurationManager.getBlocksize());
							input.setRequiresReblock(true);
							operands.put(item.getId(), input);
							break;
						}
						case Checkpoint: {
							Hop input = operands.get(item.getInputs()[0].getId());
							operands.put(item.getId(), input);
							break;
						}
						case MatrixIndexing: {
							operands.put(item.getId(), constructIndexingOp(item, operands));
							break;
						}
						case GAppend: {
							operands.put(item.getId(), HopRewriteUtils.createBinary(
								operands.get(item.getInputs()[0].getId()),
								operands.get(item.getInputs()[1].getId()), OpOp2.CBIND));
							break;
						}
						default:
							throw new DMLRuntimeException("Unsupported instruction "
								+ "type: " + stype.name() + " (" + item.getOpcode() + ").");
					}
				}
				else
					throw new DMLRuntimeException("Unsupported instruction: " + item.getOpcode());
				break;
			}
			case Literal: {
				CPOperand op = new CPOperand(item.getData());
				operands.put(item.getId(), ScalarObjectFactory
					.createLiteralOp(op.getValueType(), op.getName()));
				break;
			}
			case Dedup: {
				throw new NotImplementedException();
			}
		}
		
		item.setVisited();
	}

	public static void constructLineageFromHops(Hop[] roots, String claName, Hop[] inputs, HashMap<Long, Hop> spoofmap) {
		//probe existence and only generate lineage if non-existing
		//(a fused operator might be used in multiple places of a program)
		if( LineageCodegenItem.getCodegenLTrace(claName) == null ) {
			//recursively construct lineage for fused operator
			Map<Long, LineageItem> operands = new HashMap<>();
			//reset visit status once for entire sub DAG
			for( Hop root : roots )
				root.resetVisitStatus();

			//construct lineage dags for roots (potentially overlapping)
			for( Hop root : roots )
				rConstructLineageFromHops(root, inputs, operands, spoofmap);
			
			//create single lineage item (single root or multiagg)
			LineageItem out = operands.get(roots[0].getHopID());
			if( roots.length > 1 ) { //multi-agg
				LineageItem[] outputs = Arrays.stream(roots)
					.map(h -> new LineageItem("", UnaryCP.CAST_AS_MATRIX_OPCODE,
						new LineageItem[]{operands.get(h.getHopID())}))
					.toArray(LineageItem[]::new);
				out = new LineageItem("", "cbind", outputs);
			}
			
			//cache to avoid reconstruction
			LineageCodegenItem.setCodegenLTrace(claName, out);
			
			for (Hop root : roots)
				root.resetVisitStatus();
		}
	}

	public static void rConstructLineageFromHops(Hop root, Hop[] inputs, Map<Long, LineageItem> operands, HashMap<Long, Hop> spoofmap) {
		if (root.isVisited())
			return;
		
		boolean spoof = root instanceof SpoofFusedOp && ArrayUtils.contains(inputs, spoofmap.get(root.getHopID())); 
		if (ArrayUtils.contains(inputs, root) || spoof) {
			Hop tmp = spoof ? spoofmap.get(root.getHopID()) : root;
			int pos = ArrayUtils.indexOf(inputs, tmp);
			LineageItem li = new LineageItem(LPLACEHOLDER+pos,
				"Create"+String.valueOf(root.getHopID()));
			operands.put(tmp.getHopID(), li);
			return;
		}

		for (int i = 0; i < root.getInput().size(); i++) 
			rConstructLineageFromHops(root.getInput().get(i), inputs, operands, spoofmap);
	
		LineageItem li = null;
		LineageItem[] LIinputs = root.getInput().stream()
			.map(h->ArrayUtils.contains(inputs, spoofmap.get(h.getHopID())) ? spoofmap.get(h.getHopID()) : h)
			.map(h->operands.get(h.getHopID()))
			.toArray(LineageItem[]::new);

		String name = Dag.getNextUniqueVarname(root.getDataType());
		
		if (root instanceof ReorgOp)
			li = new LineageItem(name, "r'", LIinputs);
		else if (root instanceof UnaryOp) {
			String opcode = ((UnaryOp) root).getOp().toString();
			li = new LineageItem(name, opcode, LIinputs);
		}
		else if (root instanceof AggBinaryOp)
			li = new LineageItem(name, "ba+*", LIinputs);
		else if (root instanceof BinaryOp)
			li = new LineageItem(name, ((BinaryOp)root).getOp().toString(), LIinputs);
		else if (root instanceof TernaryOp) {
			String opcode = ((TernaryOp) root).getOp().toString();
			li = new LineageItem(name, opcode, LIinputs);
		}
		else if (root instanceof AggUnaryOp) {
			AggOp op = ((AggUnaryOp) root).getOp();
			Direction dir = ((AggUnaryOp) root).getDirection();
			String opcode = PartialAggregate.getOpcode(op, dir);
			li = new LineageItem(name, opcode, LIinputs);
		}
		else if (root instanceof IndexingOp)
			li = new LineageItem(name, "rightIndex", LIinputs);
		else if (root instanceof SpoofFusedOp)
			li = LineageCodegenItem.getCodegenLTrace(((SpoofFusedOp) root).getClassName());
		
		else if (root instanceof LiteralOp) {  //TODO: remove redundancy
			StringBuilder sb = new StringBuilder(root.getName());
			sb.append(Instruction.VALUETYPE_PREFIX);
			sb.append(root.getDataType().toString());
			sb.append(Instruction.VALUETYPE_PREFIX);
			sb.append(root.getValueType().toString());
			sb.append(Instruction.VALUETYPE_PREFIX);
			sb.append(true); //isLiteral = true
			li = new LineageItem(sb.toString());
		}
		else
			throw new DMLRuntimeException("Unsupported hop: "+root.getOpString());

		//TODO: include all the other hops
		operands.put(root.getHopID(), li);
		root.setVisited();
	}
	
	private static Hop constructIndexingOp(LineageItem item, Map<Long, Hop> operands) {
		Hop input = operands.get(item.getInputs()[0].getId());
		if( "rightIndex".equals(item.getOpcode()) )
			return HopRewriteUtils.createIndexingOp(input,
				operands.get(item.getInputs()[1].getId()), //rl
				operands.get(item.getInputs()[2].getId()), //ru
				operands.get(item.getInputs()[3].getId()), //cl
				operands.get(item.getInputs()[4].getId())); //cu
		else if( "leftIndex".equals(item.getOpcode()) 
				|| "mapLeftIndex".equals(item.getOpcode()) )
			return HopRewriteUtils.createLeftIndexingOp(input,
				operands.get(item.getInputs()[1].getId()), //rhs
				operands.get(item.getInputs()[2].getId()), //rl
				operands.get(item.getInputs()[3].getId()), //ru
				operands.get(item.getInputs()[4].getId()), //cl
				operands.get(item.getInputs()[5].getId())); //cu
		throw new DMLRuntimeException("Unsupported opcode: "+item.getOpcode());
	}
	
	private static Hop constructParameterizedBuiltinOp(LineageItem item, Map<Long, Hop> operands) {
		String opcode = item.getOpcode();
		Hop target = operands.get(item.getInputs()[0].getId());
		LinkedHashMap<String,Hop> args = new LinkedHashMap<>();
		if( opcode.equals("groupedagg") ) {
			args.put("target", target);
			args.put(Statement.GAGG_GROUPS, operands.get(item.getInputs()[1].getId()));
			args.put(Statement.GAGG_WEIGHTS, operands.get(item.getInputs()[2].getId()));
			args.put(Statement.GAGG_FN, operands.get(item.getInputs()[3].getId()));
			args.put(Statement.GAGG_NUM_GROUPS, operands.get(item.getInputs()[4].getId()));
		}
		else if (opcode.equalsIgnoreCase("rmempty")) {
			args.put("target", target);
			args.put("margin", operands.get(item.getInputs()[1].getId()));
			args.put("select", operands.get(item.getInputs()[2].getId()));
		}
		else if(opcode.equalsIgnoreCase("replace")) {
			args.put("target", target);
			args.put("pattern", operands.get(item.getInputs()[1].getId()));
			args.put("replacement", operands.get(item.getInputs()[2].getId()));
		}
		else if(opcode.equalsIgnoreCase("rexpand")) {
			args.put("target", target);
			args.put("max", operands.get(item.getInputs()[1].getId()));
			args.put("dir", operands.get(item.getInputs()[2].getId()));
			args.put("cast", operands.get(item.getInputs()[3].getId()));
			args.put("ignore", operands.get(item.getInputs()[4].getId()));
		}
		
		return HopRewriteUtils.createParameterizedBuiltinOp(
			target, args, ParamBuiltinOp.valueOf(opcode.toUpperCase()));
	}
	
	public static LineageItem rDecompress(LineageItem item) {
		if (item.getType() == LineageItemType.Dedup) {
			LineageItem dedupInput = rDecompress(item.getInputs()[0]);
			ArrayList<LineageItem> inputs = new ArrayList<>();
			
			for (LineageItem li : item.getInputs()[1].getInputs())
				inputs.add(rDecompress(li));
			
			LineageItem li = new LineageItem(item.getInputs()[1].getData(),
				item.getInputs()[1].getOpcode(), inputs.toArray(new LineageItem[0]));
			
			li.resetVisitStatusNR();
			rSetDedupInputOntoOutput(item.getData(), li, dedupInput);
			li.resetVisitStatusNR();
			return li;
		}
		else {
			ArrayList<LineageItem> inputs = new ArrayList<>();
			if (item.getInputs() != null) {
				for (LineageItem li : item.getInputs())
					inputs.add(rDecompress(li));
			}
			return new LineageItem(
				item.getData(), item.getOpcode(), inputs.toArray(new LineageItem[0]));
		}
	}
	
	public static void writeTraceToHDFS(String trace, String fname) {
		try {
			HDFSTool.writeStringToHDFS(trace, fname);
			FileSystem fs = IOUtilFunctions.getFileSystem(fname);
			if (fs instanceof LocalFileSystem) {
				Path path = new Path(fname);
				IOUtilFunctions.deleteCrcFilesFromLocalFileSystem(fs, path);
			}
			
		} catch (IOException e) {
			throw new DMLRuntimeException(e);
		}
	}
	
	private static void rSetDedupInputOntoOutput(String name, LineageItem item, LineageItem dedupInput) {
		if (item.isVisited())
			return;
		
		if (item.getInputs() != null)
			for (int i = 0; i < item.getInputs().length; i++) {
				LineageItem li = item.getInputs()[i];
				//replace CPOperand literals (placeholders)
				//TODO should use the same placeholder meta data as codegen
				if( li.getType() == LineageItemType.Literal ) {
					CPOperand tmp = new CPOperand(li.getData());
					if( !tmp.isLiteral() && tmp.getName().equals(name) )
						item.getInputs()[i] = dedupInput;
				}
				if (li.getType() == LineageItemType.Creation) {
					item.getInputs()[i] = dedupInput;
				}
				
				rSetDedupInputOntoOutput(name, li, dedupInput);
			}
		
		item.setVisited();
	}
	
	public static LineageItem replace(LineageItem root, LineageItem liOld, LineageItem liNew) {
		if( liNew == null )
			throw new DMLRuntimeException("Invalid null lineage item for "+liOld.getId());
		root.resetVisitStatusNR();
		rReplaceNR(root, liOld, liNew);
		root.resetVisitStatusNR();
		return root;
	}
	
	/**
	 * Non-recursive equivalent of {@link #rReplace(LineageItem, LineageItem, LineageItem)} 
	 * for robustness with regard to stack overflow errors.
	 * 
	 * @param current Current lineage item
	 * @param liOld Old lineage item
	 * @param liNew New Lineage item.
	 */
	public static void rReplaceNR(LineageItem current, LineageItem liOld, LineageItem liNew) {
		Stack<LineageItem> q = new Stack<>();
		q.push(current);
		while( !q.empty() ) {
			LineageItem tmp = q.pop();
			if( tmp.isVisited() || tmp.getInputs() == null )
				continue;
			//process children until old item found, then replace
			for(int i=0; i<tmp.getInputs().length; i++) {
				LineageItem ctmp = tmp.getInputs()[i];
				if (liOld.getId() == ctmp.getId() && liOld.equals(ctmp))
					tmp.setInput(i, liNew);
				else
					q.push(ctmp);
			}
			tmp.setVisited(true);
		}
	}
	
	@Deprecated
	@SuppressWarnings("unused")
	private static void rReplace(LineageItem current, LineageItem liOld, LineageItem liNew) {
		if( current.isVisited() || current.getInputs() == null )
			return;
		if( liNew == null )
			throw new DMLRuntimeException("Invalid null lineage item for "+liOld.getId());
		//process children until old item found, then replace
		for(int i=0; i<current.getInputs().length; i++) {
			LineageItem tmp = current.getInputs()[i];
			if (liOld.equals(tmp))
				current.setInput(i, liNew);
			else
				rReplace(tmp, liOld, liNew);
		}
		current.setVisited();
	}
	
	public static void replaceDagLeaves(ExecutionContext ec, LineageItem root, CPOperand[] newLeaves) {
		//find and replace the placeholder leaves
		root.resetVisitStatusNR();
		rReplaceDagLeaves(root, LineageItemUtils.getLineage(ec, newLeaves));
		root.resetVisitStatusNR();
	}
	
	public static void rReplaceDagLeaves(LineageItem root, LineageItem[] newleaves) {
		if (root.isVisited() || root.isLeaf())
			return;
		
		for (int i=0; i<root.getInputs().length; i++) {
			LineageItem li = root.getInputs()[i];
			if (li.isLeaf() && li.getType() != LineageItemType.Literal
				&& li.getData().startsWith(LPLACEHOLDER))
				//order-preserving replacement. IN#<xxx> represents relative position xxx
				root.setInput(i, newleaves[Integer.parseInt(li.getData().substring(3))]);
			else
				rReplaceDagLeaves(li, newleaves);
		}

		root.setVisited();
	}

	public static void rGetDagLeaves(HashSet<LineageItem> leaves, LineageItem root) {
		if (root.isVisited())
			return;
		
		if (root.isLeaf())
			leaves.add(root);
		else {
			for (LineageItem li : root.getInputs())
				rGetDagLeaves(leaves, li);
		}
		root.setVisited();
	}
	
	public static void checkCycles(LineageItem current) {
		current.resetVisitStatusNR();
		rCheckCycles(current, new HashSet<Long>(), true);
		current.resetVisitStatusNR();
	}
	
	public static void rCheckCycles(LineageItem current, Set<Long> probe, boolean useObjIdent) {
		if( current.isVisited() )
			return;
		long id = useObjIdent ? System.identityHashCode(current) : current.getId();
		if( probe.contains(id) )
			throw new DMLRuntimeException("Cycle detected for "+current.toString());
		probe.add(id);
		if( current.getInputs() != null )
			for(LineageItem li : current.getInputs())
				rCheckCycles(li, probe, useObjIdent);
		current.setVisited();
	}
	
	private static Hop[] createNaryInputs(LineageItem item, Map<Long, Hop> operands) {
		int len = item.getInputs().length;
		Hop[] ret = new Hop[len];
		for( int i=0; i<len; i++ )
			ret[i] = operands.get(item.getInputs()[i].getId());
		return ret;
	}
	
	public static boolean containsRandDataGen(HashSet<LineageItem> entries, LineageItem root) {
		if (entries.contains(root) || root.isVisited())
			return false;
		boolean isRand = isNonDeterministic(root);
		if (!root.isLeaf() && !isRand) 
			for (LineageItem input : root.getInputs())
				isRand |= containsRandDataGen(entries, input);
		root.setVisited();
		return isRand;
	}
	
	private static boolean isNonDeterministic(LineageItem li) {
		if (li.getType() != LineageItemType.Creation)
			return false;

		boolean isND = false;
		DataGenCPInstruction cprand = null;
		RandSPInstruction sprand = null;
		Instruction ins = InstructionParser.parseSingleInstruction(li.getData());

		if (ins instanceof DataGenCPInstruction)
			cprand = (DataGenCPInstruction)ins;
		else if (ins instanceof RandSPInstruction)
			sprand = (RandSPInstruction)ins;
		else 
			return false;

		switch(li.getOpcode().toUpperCase())
		{
			case "RAND":
				if (cprand != null)
					if ((cprand.getMinValue() != cprand.getMaxValue()) || (cprand.getSparsity() != 1))
						isND = true;
				if (sprand!= null)
					if ((sprand.getMinValue() != sprand.getMaxValue()) || (sprand.getSparsity() != 1))
						isND = true;
					//NOTE:It is hard to detect in runtime if rand was called with unspecified seed
					//as -1 is already replaced by computed seed. Solution is to unmark for caching in 
					//compile time. That way we can differentiate between given and unspecified seed.
				break;
			case "SAMPLE":
				isND = true;
				break;
			default:
				isND = false;
				break;
		}
		//TODO: add 'read' in this list
		return isND;
	}
	
	public static LineageItem[] getLineageItemInputstoSB(ArrayList<String> inputs, ExecutionContext ec) {
		if (ReuseCacheType.isNone() && !DMLScript.LINEAGE_DEDUP)
			return null;
		
		ArrayList<CPOperand> CPOpInputs = inputs.size() > 0 ? new ArrayList<>() : null;
		for (int i=0; i<inputs.size(); i++) {
			Data value = ec.getVariable(inputs.get(i));
			if (value != null) {
				CPOpInputs.add(new CPOperand(value instanceof ScalarObject ? value.toString() : inputs.get(i),
					value.getValueType(), value.getDataType()));
			}
		}
		return(CPOpInputs != null ? LineageItemUtils.getLineage(ec, 
			CPOpInputs.toArray(new CPOperand[CPOpInputs.size()])) : null);
	}
	
	public static void addAllDataLineage(ExecutionContext ec) {
		for( Entry<String, Data> e : ec.getVariables().entrySet() ) {
			if( e.getValue() instanceof CacheableData<?> ) {
				CacheableData<?> cdata = (CacheableData<?>) e.getValue();
				//only createvar instruction with pREAD prefix added to lineage
				String fromVar = org.apache.sysds.lops.Data.PREAD_PREFIX+e.getKey();
				ec.traceLineage(VariableCPInstruction.prepCreatevarInstruction(
					fromVar, "CacheableData::"+cdata.getUniqueID(), false, "binary"));
				//move from pREADx to x
				ec.traceLineage(VariableCPInstruction.prepMoveInstruction(fromVar, e.getKey()));
			}
		}
	}
}