Fix / Range selector (#620)

* Fix negative offset in range selector

* Do not double pump in streaming encoder base class

Author: martin-traverse

tracdap-libs/tracdap-lib-data/src/main/java/org/finos/tracdap/common/codec/StreamingEncoder.java

@@ -46,11 +46,6 @@ public boolean isReady() {
 
     @Override
     public void pump() {
-
-        if (context == null)
-            return;
-
-        if (context.readyToUnload() && consumerReady())
-            onBatch();
+        // No-op, streaming encoder pushes batches through directly
     }
 }

tracdap-libs/tracdap-lib-data/src/main/java/org/finos/tracdap/common/data/pipeline/RangeSelector.java

@@ -23,6 +23,8 @@
 import org.finos.tracdap.common.data.ArrowVsrContext;
 import org.finos.tracdap.common.data.DataPipeline;
 import org.finos.tracdap.common.exception.EUnexpected;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 
@@ -35,6 +37,7 @@ public class RangeSelector
         DataPipeline.DataConsumer<DataPipeline.ArrowApi>,
         DataPipeline.DataProducer<DataPipeline.ArrowApi> {
 
+    private static final Logger log = LoggerFactory.getLogger(RangeSelector.class);
     private final long offset;
     private final long limit;
     private long currentRow;
@@ -123,24 +126,25 @@ public void onBatch() {
 
             sliceRoot.setRowCount(batchSize);
             sliceRoot.setLoaded();
-            consumer().onBatch();
         }
         else if (batchEndRow >= offset && (batchStartRow < offset + limit || limit == 0)) {
 
-            var sliceStart = (int) (offset - batchStartRow);
+            var sliceStart = (int) Math.max(offset - batchStartRow, 0);
             var sliceEnd = (int) Math.min(offset + limit - batchStartRow, batchSize);
             var sliceLength = sliceEnd - sliceStart;
 
             sliceTransfers.forEach(slice -> slice.splitAndTransfer(sliceStart, sliceLength));
 
             sliceRoot.setRowCount(sliceLength);
             sliceRoot.setLoaded();
-            consumer().onBatch();
         }
 
         // Always consume the incoming data
         incomingRoot.setUnloaded();
         currentRow += batchSize;
+
+        if (sliceRoot.readyToUnload())
+            consumer().onBatch();
     }
 
     @Override