db: Support robot scans

db/ztests/robot.yaml

@@ -0,0 +1,13 @@
+script: |
+  super db init -q
+  super db create -q test1
+  super db create -q test2
+  echo {x:1} | super db load -q -use test1 -
+  echo {x:2} | super db load -q -use test2 -
+  super db -s -c 'values "test1", "test2" | from f"{this}"'
+
+outputs:
+  - name: stdout
+    data: |
+      {x:1}
+      {x:2}

runtime/exec/environment.go

@@ -13,6 +13,8 @@ import (
 	"github.com/brimdata/super/dbid"
 	"github.com/brimdata/super/order"
 	"github.com/brimdata/super/pkg/storage"
+	"github.com/brimdata/super/runtime"
+	"github.com/brimdata/super/runtime/sam/op/meta"
 	"github.com/brimdata/super/sbuf"
 	"github.com/brimdata/super/sio/anyio"
 	"github.com/brimdata/super/vector"
@@ -120,6 +122,23 @@ func (e *Environment) OpenHTTP(ctx context.Context, sctx *super.Context, url, fo
 	return file, nil
 }
 
+func (e *Environment) OpenPool(ctx context.Context, sctx *super.Context, id ksuid.KSUID, pushdown sbuf.Pushdown) (vio.Puller, error) {
+	pool, err := e.db.OpenPool(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	branch, err := e.CommitObject(ctx, id, "main")
+	if err != nil {
+		return nil, err
+	}
+	l, err := meta.NewSortedLister(ctx, sctx, pool, branch, nil)
+	if err != nil {
+		return nil, err
+	}
+	scanner := meta.NewSequenceScanner(runtime.NewContext(ctx, sctx), l, pool, pushdown, nil, nil)
+	return sbuf.NewDematerializer(sctx, scanner), nil
+}
+
 func newConcurrentPuller(path string, puller vio.Puller) ConcurrentPuller {
 	cp, ok := puller.(ConcurrentPuller)
 	if !ok {

runtime/vam/op/robot.go

@@ -1,8 +1,6 @@
 package op
 
 import (
-	"fmt"
-
 	"github.com/brimdata/super/runtime"
 	"github.com/brimdata/super/runtime/exec"
 	"github.com/brimdata/super/runtime/vam/expr"
@@ -134,9 +132,12 @@ func (o *Robot) nextVec() (vector.Any, error) {
 }
 
 func (o *Robot) open(path string) (vio.Puller, error) {
-	// This check for attached database will be removed when we add support for pools here.
 	if o.env.IsAttached() {
-		return nil, fmt.Errorf("%s: cannot open in a database environment", path)
+		// XXX we should support committish values
+		id, err := o.env.PoolID(o.rctx, path)
+		if err == nil {
+			return o.env.OpenPool(o.rctx, o.rctx.Sctx, id, o.pushdown)
+		}
 	}
 	return o.env.Open(o.rctx.Context, o.rctx.Sctx, path, o.format, o.pushdown, 1)
 }