Batch operation

.changeset/young-geese-teach.md

@@ -0,0 +1,5 @@
+---
+"barnard59-base": patch
+---
+
+added batch operation

packages/base/batch.js

@@ -0,0 +1,26 @@
+import { Transform } from 'readable-stream'
+
+function batch(size = 0) {
+  let array = []
+  const maxSize = Number(size)
+  return new Transform({
+    readableObjectMode: true,
+    writableObjectMode: true,
+    readableHighWaterMark: maxSize,
+    writableHighWaterMark: maxSize,
+    transform(item, _encoding, callback) {
+      array.push(item)
+      if (array.length === maxSize) {
+        this.push(array)
+        array = []
+      }
+      callback()
+    },
+    flush(callback) {
+      if (array.length > 0) this.push(array)
+      callback()
+    },
+  })
+}
+
+export default batch

packages/base/manifest.ttl

@@ -31,6 +31,13 @@
     code:link <node:barnard59-base/filter.js#default>
   ].
 
+<batch> a p:Operation, p:WritableObjectMode, p:ReadableObjectMode;
+  rdfs:label "Batch";
+  rdfs:comment "Groups incoming items into arrays of the given size.";
+  code:implementedBy [ a code:EcmaScriptModule;
+    code:link <node:barnard59-base/batch.js#default>
+  ].
+
 <flatten> a p:Operation, p:WritableObjectMode, p:ReadableObjectMode;
   rdfs:label "Flatten";
   rdfs:comment "Separates incoming arrays into their elements and emits each element as a single chunk.";

packages/base/test/batch.test.js

@@ -0,0 +1,59 @@
+import { deepStrictEqual, strictEqual } from 'assert'
+import { array } from 'get-stream'
+import { isReadable, isWritable } from 'isstream'
+import { Readable } from 'readable-stream'
+import batch from '../batch.js'
+
+describe('batch', () => {
+  it('should be a function', () => {
+    strictEqual(typeof batch, 'function')
+  })
+
+  it('should return a duplex stream', async () => {
+    const result = batch()
+
+    strictEqual(isWritable(result), true)
+    strictEqual(isReadable(result), true)
+  })
+
+  it('should do nothing if there are no input chunks', async () => {
+    const input = Readable({
+      objectMode: true,
+      read: () => {
+        input.push(null)
+      },
+    })
+
+    const result = await array(input.pipe(batch()))
+
+    deepStrictEqual(result, [])
+  })
+
+  it('should split input in batches', async () => {
+    const expected = [['a', 'b'], ['c', 'd'], ['e']]
+    const input = Readable({
+      objectMode: true,
+      read: () => {
+        ['a', 'b', 'c', 'd', 'e', null].forEach(item => input.push(item))
+      },
+    })
+
+    const result = await array(input.pipe(batch(2)))
+
+    deepStrictEqual(result, expected)
+  })
+
+  it('should emit a single batch', async () => {
+    const expected = [['a', 'b', 'c', 'd', 'e']]
+    const input = Readable({
+      objectMode: true,
+      read: () => {
+        ['a', 'b', 'c', 'd', 'e', null].forEach(item => input.push(item))
+      },
+    })
+
+    const result = await array(input.pipe(batch(0)))
+
+    deepStrictEqual(result, expected)
+  })
+})

packages/cube/pipeline/check-observations.ttl

@@ -45,7 +45,8 @@ _:sortChunkSize a p:Variable ;
         [ n3:parse () ]
         _:sortBySubject
         [ splitDataset:bySubject () ]
-        _:batch
+        [ base:batch ("batchSize"^^p:VariableName) ]
+        [ base:flatten () ]
         [ shacl:report
           [code:name "shape" ; code:value _:getConstraint ] ,
           [code:name "maxErrors" ; code:value "maxViolations"^^p:VariableName ]
@@ -56,16 +57,6 @@ _:sortChunkSize a p:Variable ;
     ]
 .
 
-_:batch a p:Step ;
-  code:implementedBy
-    [
-      a code:EcmaScriptModule ;
-      code:link <file:../lib/batch.js#batch> ;
-    ] ;
-  code:arguments ("batchSize"^^p:VariableName)
-.
-
-
 _:getConstraint a p:Pipeline , p:ReadableObjectMode ;
   p:variables [ p:variable _:constraint ] ;
   p:steps