Fix race in parallel composite decode breaking dummycode+recode

The multi-threaded DecoderComposite.decode submitted one task per
decoder per row block, running all decoders concurrently. This broke
the ordering dependency between decoders: recode-on-output reads the
category indexes written by the dummycode decoder, so when the recode
task raced ahead it read unwritten cells and produced null or the raw
index instead of the original value.

Parallelize over row blocks instead, running all decoders in order
within each block via the sequential block decode. Also short-circuit
to the single-threaded path when k <= 1.

Fixes order-dependent failures in TransformFrameEncodeDecodeTest and
TransformFrameEncodeColmapTest (dummycode single-node/hybrid) that
surfaced once transformdecode started using the parallel decode path.

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/transform/decode/DecoderComposite.java

@@ -62,17 +62,20 @@ public FrameBlock decode(MatrixBlock in, FrameBlock out) {
 
 	@Override
 	public FrameBlock decode(final MatrixBlock in, final FrameBlock out, final int k) {
+		if(k <= 1)
+			return decode(in, out);
 		final ExecutorService pool = CommonThreadPool.get(k);
 		out.ensureAllocatedColumns(in.getNumRows());
 		try {
 			final List<Future<?>> tasks = new ArrayList<>();
 			int blz = Math.max(in.getNumRows() / k, 1000);
-			for(Decoder decoder : _decoders){
-				for(int i = 0; i < in.getNumRows(); i += blz){
-					final int start = i;
-					final int end = Math.min(in.getNumRows(), i + blz);
-					tasks.add(pool.submit(() -> decoder.decode(in, out, start, end)));
-				}
+			// Parallelize over row blocks (not over decoders): all decoders must
+			// run in order within a block, e.g. recode-on-output depends on the
+			// category indexes produced by the preceding dummycode decoder.
+			for(int i = 0; i < in.getNumRows(); i += blz){
+				final int start = i;
+				final int end = Math.min(in.getNumRows(), i + blz);
+				tasks.add(pool.submit(() -> decode(in, out, start, end)));
 			}
 			for(Future<?> f : tasks)
 				f.get();