improve the logic to handle groups split by opposite-signed index

Author: aalkin

Framework/Core/include/Framework/Kernels.h

@@ -81,22 +81,29 @@ auto sliceByColumn(
   for (auto i = 0; i < column->num_chunks(); ++i) {
     T prev = 0;
     T cur = 0;
-    T nprev = -1;
+    T lastNeg = -1;
+    T lastPos = 0;
+
     auto array = static_cast<arrow::NumericArray<typename detail::ConversionTraits<T>::ArrowType>>(column->chunk(i)->data());
     for (auto e = 0; e < array.length(); ++e) {
-      if (cur >= 0) {
-        prev = cur;
+      prev = cur;
+      if (prev >= 0) {
+        lastPos = prev;
       } else {
-        nprev = cur;
+        lastNeg = prev;
       }
       cur = array.Value(e);
       if (cur >= 0) {
-        if (prev > cur) {
-          throw runtime_error_f("Table %s index %s is not sorted: next value %d < previous value %d!", target, key, cur, prev);
+        if (lastPos > cur) {
+          throw runtime_error_f("Table %s index %s is not sorted: next value %d < previous value %d!", target, key, cur, lastPos);
+        } else if (lastPos == cur && prev < 0) {
+          throw runtime_error_f("Table %s index %s has a group with index %d that is split by %d", target, key, cur, prev);
         }
       } else {
-        if (nprev < cur) {
-          throw runtime_error_f("Table %s index %s is not sorted: next negative value %d > previous negative value %d!", target, key, cur, nprev);
+        if (lastNeg < cur) {
+          throw runtime_error_f("Table %s index %s is not sorted: next negative value %d > previous negative value %d!", target, key, cur, lastNeg);
+        } else if (lastNeg == cur && prev >= 0) {
+          throw runtime_error_f("Table %s index %s has a group with index %d that is split by %d", target, key, cur, prev);
         }
       }
     }