[MINOR] Fix edgecase of Compressed OffsetSlice

This commit fixes an edgecase of offset slices where
a single element is sliced.

Closes #2163

Signed-off-by: Sebastian Baunsgaard <[email protected]>

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/compress/colgroup/offset/AOffset.java

@@ -569,6 +569,8 @@ public OffsetSliceInfo slice(int l, int u) {
 			else
 				return new OffsetSliceInfo(0, s, moveIndex(l));
 		}
+		else if (u < first)
+			return EMPTY_SLICE;
 
 		final AIterator it = getIteratorSkipCache(l);
 
@@ -587,6 +589,7 @@ private OffsetSliceInfo genericSlice(int l, int u, AIterator it) {
 		final int low = it.getDataIndex();
 		final int lowOff = it.getOffsetsIndex();
 		final int lowValue = it.value();
+
 		// set c
 		int high = low;
 		int highOff = lowOff;
@@ -607,7 +610,7 @@ private OffsetSliceInfo genericSlice(int l, int u, AIterator it) {
 
 	private final OffsetSliceInfo constructSliceReturn(int l, int u, int low, int high, int lowOff, int highOff,
 		int lowValue, int highValue) {
-		if(low == high)
+		if(low == high) // Implicit lowValue == highValue
 			return new OffsetSliceInfo(low, high + 1, new OffsetSingle(lowValue - l));
 		else if(low + 1 == high)
 			return new OffsetSliceInfo(low, high + 1, new OffsetTwo(lowValue - l, highValue - l));

src/main/java/org/apache/sysds/runtime/compress/colgroup/offset/OffsetTwo.java

@@ -97,7 +97,7 @@ public static OffsetTwo readFields(DataInput in) throws IOException {
 	@Override
 	public OffsetSliceInfo slice(int l, int u) {
 		if(l <= first) {
-			if(u < first)
+			if(u <= first)
 				return EMPTY_SLICE;
 			else if(u > last)
 				return new OffsetSliceInfo(0, 2, moveIndex(l));

src/test/java/org/apache/sysds/test/component/compress/offset/OffsetTests.java

@@ -58,7 +58,7 @@
 
 @RunWith(value = Parameterized.class)
 public class OffsetTests {
-	static{
+	static {
 		CompressedMatrixBlock.debug = true;
 	}
 
@@ -77,6 +77,10 @@ public static Collection<Object[]> data() {
 		for(OFF_TYPE t : OFF_TYPE.values()) {
 			tests.add(new Object[] {new int[] {}, t});
 			tests.add(new Object[] {new int[] {1, 2}, t});
+			tests.add(new Object[] {new int[] {1, 3, 5, 7}, t});
+			tests.add(new Object[] {new int[] {2, 3, 6, 7}, t});
+			tests.add(new Object[] {new int[] {2, 5, 10, 12}, t});
+			tests.add(new Object[] {new int[] {2, 6, 8, 14}, t});
 			tests.add(new Object[] {new int[] {2, 142}, t});
 			tests.add(new Object[] {new int[] {142, 421}, t});
 			tests.add(new Object[] {new int[] {1, 1023}, t});
@@ -562,6 +566,46 @@ public void sliceFirst100() {
 		slice(0, 100);
 	}
 
+	@Test
+	public void slice1_2() {
+		slice(1, 2);
+	}
+
+	@Test
+	public void slice3_4() {
+		slice(3, 4);
+	}
+
+	@Test
+	public void sliceRange() {
+		for(int i = 0; i < 100; i++) {
+			// not allowed to slice smaller or equal range.
+			// slice(i, i - 1);
+			// slice(i, i);
+			slice(i, i + 1);
+			slice(i, i + 2);
+			slice(i, i + 3);
+		}
+	}
+
+	@Test
+	public void sliceRange256() {
+		slice(250, 260);
+		slice(254, 260);
+		slice(254, 257);
+		slice(255, 257);
+		slice(255, 256);
+	}
+
+	@Test
+	public void sliceRange2x256() {
+		slice(250 * 2, 260 * 2);
+		slice(254 * 2, 260 * 2);
+		slice(254 * 2, 257 * 2);
+		slice(255 * 2, 257 * 2);
+		slice(255 * 2, 256 * 2);
+	}
+
 	@Test
 	public void sliceToLast() {
 		if(data.length > 1)
@@ -579,11 +623,6 @@ public void slice100to10000() {
 		slice(100, 10000);
 	}
 
-	@Test
-	public void verify() {
-		o.verify(o.getSize());
-	}
-
 	@Test
 	public void slice1to4() {
 		slice(1, 4);
@@ -594,7 +633,7 @@ public void slice() {
 		if(data.length > 1) {
 			int n = data[data.length - 1];
 			for(int i = 0; i < n && i < 100; i++) {
-				for(int j = i; j < n + 1 && j < 100; j++) {
+				for(int j = i+1; j < n + 1 && j < 100; j++) {
 					slice(i, j, false);
 				}
 			}
@@ -624,12 +663,28 @@ private void slice(int l, int u, boolean str) {
 				int i = 0;
 				while(i < data.length && data[i] < l)
 					i++;
-
-				while(data[i] < u) {
-					assertEquals(data[i] - l, it.value());
-					if(a.offsetSlice.getOffsetToLast() > it.value())
-						it.next();
-					i++;
+					
+				if(! (a.offsetSlice instanceof OffsetEmpty)){
+					int t = 0;
+					final int lasstSliceOffset = a.offsetSlice.getOffsetToLast();
+					while(data[i] < u) {
+						assertEquals(data[i] - l, it.value());
+						if(lasstSliceOffset > it.value())
+							it.next();
+						i++;
+						t++;
+					}
+
+					int sliceSize = a.offsetSlice.getSize();
+					if(sliceSize != t) {
+						fail("Slice size is not equal to elements that should have been sliced:\n" + sliceSize + " vs " + t
+							+ "\n" + a + "\nrange: " + l + " " + u + "\ninput " + o);
+					}
+				}
+				else {
+					if( i < data.length){
+						assertTrue(data[i] >= u);
+					}
 				}
 			}
 		}
@@ -640,6 +695,14 @@ private void slice(int l, int u, boolean str) {
 
 	}
 
+
+	@Test
+	public void verify() {
+		o.verify(o.getSize());
+	}
+
+
+
 	@Test
 	public void append() {
 		if(data.length > 0) {