dict writable

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/compress/io/DictWritable.java

@@ -24,14 +24,19 @@
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import org.apache.hadoop.io.Writable;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.DictionaryFactory;
 import org.apache.sysds.runtime.compress.colgroup.dictionary.IDictionary;
 
 public class DictWritable implements Writable, Serializable {
 	private static final long serialVersionUID = 731937201435558L;
+	
 	public List<IDictionary> dicts;
 
 	public DictWritable() {
@@ -44,26 +49,73 @@ protected DictWritable(List<IDictionary> dicts) {
 
 	@Override
 	public void write(DataOutput out) throws IOException {
+		// the dicts can contain duplicates.
+		// to avoid writing duplicates we run though once to detect them
+		Set<IDictionary> ud = new HashSet<>();
+		for(IDictionary d: dicts){
+			if(ud.contains(d)){
+				writeWithDuplicates(out);
+				return;
+			}
+			ud.add(d);
+		}
+
 		out.writeInt(dicts.size());
 		for(int i = 0; i < dicts.size(); i++)
 			dicts.get(i).write(out);
 	}
 
+	private void writeWithDuplicates(DataOutput out) throws IOException {
+		// indicate that we use duplicate detection
+		out.writeInt(dicts.size() * -1); 
+		Map<IDictionary, Integer> m = new HashMap<>();
+		
+		for(int i = 0; i < dicts.size(); i++){
+			int id = m.getOrDefault(dicts.get(i), m.size() );
+			out.writeInt(id);
+
+			if(!m.containsKey(dicts.get(i))){
+				m.put(dicts.get(i), m.size());
+				dicts.get(i).write(out);
+			}
+
+		}
+	}
+
 	@Override
 	public void readFields(DataInput in) throws IOException {
 		int s = in.readInt();
+		if( s < 0){
+			readFieldsWithDuplicates(Math.abs(s), in);
+		}
+		else{
+			dicts = new ArrayList<>(s);
+			for(int i = 0; i < s; i++)
+				dicts.add(DictionaryFactory.read(in));
+		}
+	}
+
+	private void readFieldsWithDuplicates(int s, DataInput in) throws IOException {
+
 		dicts = new ArrayList<>(s);
-		for(int i = 0; i < s; i++)
-			dicts.add(DictionaryFactory.read(in));
+		for(int i = 0; i < s; i++){
+			int id = in.readInt();
+			if(id < i)
+				dicts.set(i, dicts.get(id));
+			else
+				dicts.add(DictionaryFactory.read(in));
+		}
 	}
 
+
 	@Override
 	public String toString() {
 		StringBuilder sb = new StringBuilder();
 		sb.append("Written dictionaries:\n");
 		for(IDictionary d : dicts) {
 			sb.append(d);
 			sb.append("\n");
+
 		}
 		return sb.toString();
 	}

src/main/java/org/apache/sysds/runtime/compress/io/WriterCompressed.java

@@ -57,6 +57,7 @@
 import org.apache.sysds.runtime.meta.MatrixCharacteristics;
 import org.apache.sysds.runtime.util.CommonThreadPool;
 import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
+import org.apache.sysds.runtime.util.HDFSTool;
 
 public final class WriterCompressed extends MatrixWriter {
 
@@ -146,7 +147,7 @@ private void write(MatrixBlock src, final String fname, final int blen) throws I
 		}
 
 		fs = IOUtilFunctions.getFileSystem(new Path(fname), job);
-
+	
 		int k = OptimizerUtils.getParallelBinaryWriteParallelism();
 
 		k =  Math.min(k, (int)(src.getInMemorySize() /  InfrastructureAnalyzer.getBlockSize(fs)));
@@ -213,8 +214,6 @@ private void writeMultiBlockCompressedSingleThread(MatrixBlock mb, final int rle
 		throws IOException {
 		try {
 			final CompressedMatrixBlock cmb = (CompressedMatrixBlock) mb;
-
-			setupWrite();
 			final Path path = new Path(fname);
 			Writer w = generateWriter(job, path, fs);
 			for(int bc = 0; bc * blen < clen; bc++) {// column blocks
@@ -244,7 +243,6 @@ private void writeMultiBlockCompressedSingleThread(MatrixBlock mb, final int rle
 	private void writeMultiBlockCompressedParallel(MatrixBlock b, final int rlen, final int clen, final int blen, int k)
 		throws IOException {
 
-		setupWrite();
 		final ExecutorService pool = CommonThreadPool.get(k);
 		try {
 			final ArrayList<Callable<Object>> tasks = new ArrayList<>();
@@ -265,7 +263,8 @@ private void writeMultiBlockCompressedParallel(MatrixBlock b, final int rlen, fi
 			final int colBlocks = (int) Math.ceil((double) clen / blen );
 			final int nBlocks = (int) Math.ceil((double) rlen / blen);
 			final int blocksPerThread = Math.max(1, nBlocks * colBlocks / k );
-
+			HDFSTool.deleteFileIfExistOnHDFS(new Path(fname + ".dict"), job);
+			
 			int i = 0;
 			for(int bc = 0; bc * blen < clen; bc++) {// column blocks
 				final int sC = bc * blen;
@@ -307,13 +306,6 @@ private void writeMultiBlockCompressedParallel(MatrixBlock b, final int rlen, fi
 		}
 	}
 
-	private void setupWrite() throws IOException {
-		// final Path path = new Path(fname);
-		// final JobConf job = ConfigurationManager.getCachedJobConf();
-		// HDFSTool.deleteFileIfExistOnHDFS(path, job);
-		// HDFSTool.createDirIfNotExistOnHDFS(path, DMLConfig.DEFAULT_SHARED_DIR_PERMISSION);
-	}
-
 	private Path getPath(int id) {
 		return new Path(fname, IOUtilFunctions.getPartFileName(id));
 	}
@@ -397,6 +389,7 @@ protected DictWriteTask(String fname, List<IDictionary> dicts, int id) {
 		public Object call() throws Exception {
 
 			Path p = new Path(fname + ".dict", IOUtilFunctions.getPartFileName(id));
+			HDFSTool.deleteFileIfExistOnHDFS(p, job);
 			try(Writer w = SequenceFile.createWriter(job, Writer.file(p), //
 				Writer.bufferSize(4096), //
 				Writer.keyClass(DictWritable.K.class), //