Repository: brimsec/zq
Pull request #2059 (2021-02-03 10:17:03)

Fix fill() logic


Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 52c6078de..dd2797687 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -44,7 +44,7 @@ jobs:
       - uses: actions/checkout@v1
       - run: make install
       - run: make zng-output-check
-  test-postgres:
+  test-services:
     runs-on: ubuntu-latest
     services:
       postgres:
@@ -59,9 +59,18 @@ jobs:
           --health-timeout 5s
         ports:
           - 5432:5432
+      redis:
+        image: redis:6
+        ports:
+          - "6379:6379"
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     steps:
       - uses: actions/setup-go@v2
         with:
           go-version: '1.15'
       - uses: actions/checkout@v1
-      - run: make test-postgres
+      - run: make test-services
diff --git a/Makefile b/Makefile
index 9a341c223..0e56b13fa 100644
--- a/Makefile
+++ b/Makefile
@@ -86,16 +86,19 @@ test-heavy: build $(SAMPLEDATA)
 test-pcapingest: bin/$(ZEEKPATH)
 	@ZEEK="$(CURDIR)/bin/$(ZEEKPATH)/zeekrunner" go test -v -run=PcapPost -tags=pcapingest ./ppl/zqd
 
-.PHONY: test-postgres
-test-postgres: build 
+.PHONY: test-services
+test-services: build
 	@ZTEST_PATH="$(CURDIR)/dist:$(CURDIR)/bin" \
-		ZTEST_TAG=postgres \
+		ZTEST_TAG=services \
 		go test -v -run TestZq/ppl/zqd/db/postgresdb/ztests .
+	@ZTEST_PATH="$(CURDIR)/dist:$(CURDIR)/bin" \
+		ZTEST_TAG=services \
+		go test -v -run TestZq/ppl/zqd/ztests/redis .
 
-.PHONY: test-postgres-docker
-test-postgres-docker:
+.PHONY: test-services-docker
+test-services-docker:
 	@docker-compose -f $(CURDIR)/ppl/zqd/scripts/dkc-services.yaml up -d
-	$(MAKE) test-postgres; \
+	$(MAKE) test-services; \
 		status=$$?; \
 		docker-compose -f $(CURDIR)/ppl/zqd/scripts/dkc-services.yaml down || exit; \
 		exit $$status
diff --git a/cli/inputflags/flags.go b/cli/inputflags/flags.go
index da10d22d7..7325751fb 100644
--- a/cli/inputflags/flags.go
+++ b/cli/inputflags/flags.go
@@ -10,6 +10,7 @@ import (
 	"regexp"
 
 	"github.com/brimsec/zq/cli/auto"
+	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio"
 	"github.com/brimsec/zq/zio/azngio"
@@ -87,7 +88,7 @@ func (f *Flags) Open(zctx *resolver.Context, paths []string, stopOnErr bool) ([]
 	var warned bool
 	for _, path := range paths {
 		if path == "-" {
-			path = detector.StdinPath
+			path = iosrc.Stdin
 		}
 		file, err := detector.OpenFile(zctx, path, f.ReaderOpts)
 		if err != nil {
diff --git a/cmd/microindex/convert/command.go b/cmd/microindex/convert/command.go
index 0ade9e57b..d942770ec 100644
--- a/cmd/microindex/convert/command.go
+++ b/cmd/microindex/convert/command.go
@@ -8,6 +8,7 @@ import (
 	"github.com/brimsec/zq/cmd/microindex/root"
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/microindex"
+	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/detector"
 	"github.com/brimsec/zq/zng/resolver"
@@ -70,7 +71,7 @@ func (c *Command) Run(args []string) error {
 	}
 	path := args[0]
 	if path == "-" {
-		path = detector.StdinPath
+		path = iosrc.Stdin
 	}
 	zctx := resolver.NewContext()
 	file, err := detector.OpenFile(zctx, path, c.inputFlags.Options())
diff --git a/cmd/microindex/create/command.go b/cmd/microindex/create/command.go
index 854f0d8ca..4ec71fb4e 100644
--- a/cmd/microindex/create/command.go
+++ b/cmd/microindex/create/command.go
@@ -9,6 +9,7 @@ import (
 	"github.com/brimsec/zq/expr"
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/microindex"
+	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/detector"
 	"github.com/brimsec/zq/zng/resolver"
@@ -72,7 +73,7 @@ func (c *Command) Run(args []string) error {
 	}
 	path := args[0]
 	if path == "-" {
-		path = detector.StdinPath
+		path = iosrc.Stdin
 	}
 	zctx := resolver.NewContext()
 	file, err := detector.OpenFile(zctx, path, c.inputFlags.Options())
diff --git a/cmd/zapi/cmd/index/find.go b/cmd/zapi/cmd/index/find.go
index 940bd14d0..0dc25fec5 100644
--- a/cmd/zapi/cmd/index/find.go
+++ b/cmd/zapi/cmd/index/find.go
@@ -7,7 +7,7 @@ import (
 	"github.com/brimsec/zq/cli/outputflags"
 	"github.com/brimsec/zq/cmd/zapi/cmd"
 	"github.com/brimsec/zq/emitter"
-	"github.com/brimsec/zq/ppl/archive"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/mccanne/charm"
 )
@@ -62,7 +62,7 @@ type FindCmd struct {
 func NewFind(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 	c := &FindCmd{Command: parent.(*Command).Command}
 	f.StringVar(&c.indexFile, "x", "", "name of microindex for custom index searches")
-	f.StringVar(&c.pathField, "l", archive.DefaultAddPathField, "zng field name for path name of log file")
+	f.StringVar(&c.pathField, "l", lake.DefaultAddPathField, "zng field name for path name of log file")
 	f.BoolVar(&c.relativePaths, "relative", false, "display paths relative to root")
 	f.BoolVar(&c.zng, "z", false, "write results as zng stream rather than list of files")
 
diff --git a/cmd/zapi/cmd/space.go b/cmd/zapi/cmd/space.go
index 5756ec22d..835f82c4f 100644
--- a/cmd/zapi/cmd/space.go
+++ b/cmd/zapi/cmd/space.go
@@ -12,7 +12,7 @@ import (
 	"github.com/brimsec/zq/api/client"
 	"github.com/brimsec/zq/pkg/glob"
 	"github.com/brimsec/zq/pkg/units"
-	"github.com/brimsec/zq/ppl/archive"
+	"github.com/brimsec/zq/ppl/lake"
 )
 
 var (
@@ -27,7 +27,7 @@ type SpaceCreateFlags struct {
 }
 
 func (f *SpaceCreateFlags) SetFlags(fs *flag.FlagSet) {
-	f.thresh = archive.DefaultLogSizeThreshold
+	f.thresh = lake.DefaultLogSizeThreshold
 	fs.Var(&f.kind, "k", "kind of storage for this space")
 	fs.StringVar(&f.datapath, "d", "", "specific directory for storage data")
 	fs.Var(&f.thresh, "thresh", "target size of chopped files, as '10MB', '4GiB', etc.")
diff --git a/compiler/proc.go b/compiler/proc.go
index 0ee4d46a9..552b9e283 100644
--- a/compiler/proc.go
+++ b/compiler/proc.go
@@ -147,13 +147,6 @@ func compileProc(custom Hook, node ast.Proc, pctx *proc.Context, parent proc.Int
 		if err != nil {
 			return nil, err
 		}
-		// XXX fix this limitation (see issue #1753)
-		for _, c := range clauses {
-			if len(c.LHS) > 1 {
-				name := c.LHS.String()
-				return nil, fmt.Errorf("%s: put currently supports only top-level field assignemnts", name)
-			}
-		}
 		put, err := put.New(pctx, parent, clauses)
 		if err != nil {
 			return nil, err
diff --git a/expr/shaper.go b/expr/shaper.go
index 18481b09b..0deeb39df 100644
--- a/expr/shaper.go
+++ b/expr/shaper.go
@@ -45,7 +45,7 @@ func (s *op) append(step op) {
 
 // create the op needed to build a record of type out from a
 // record of type in. The two types must be compatible, meaning that
-// the input type must be an unordered sub-tree of the input type
+// the input type must be an unordered subset of the input type
 // (where 'unordered' means that if the output type has record fields
 // [a b] and the input type has fields [b a] that is ok). It is also
 // ok for leaf primitive types to be different; if they are a casting
@@ -320,52 +320,26 @@ func (c *Shaper) orderRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord,
 // fillRecordType applies a fill (as specified by the record type 'spec')
 // to a record type and returns the resulting record type.
 func (c *Shaper) fillRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
-	cols := make([]zng.Column, 0)
-	names := make([]string, 0)
-	// Compute ordered list of fields. This takes all the input
-	// fields and adds the filled fields at the end. For
-	// standalone uses of 'fill()' (as opposed to shape() which
-	// includes order)', this might be surprising and it would
-	// likely be preferable to have the filled fields be woven
-	// into the input records. This can come later as needed.
-	for _, col := range input.Columns {
-		names = append(names, col.Name)
-	}
-	for _, specCol := range spec.Columns {
-		if !input.HasField(specCol.Name) {
-			names = append(names, specCol.Name)
-		}
-	}
-	for _, name := range names {
-		ind, _ := spec.ColumnOfField(name)
-		specCol := spec.Columns[ind]
-		if !input.HasField(name) {
-			// Field not in input: fill.
-			cols = append(cols, specCol)
-			continue
-		}
-		ind, _ = input.ColumnOfField(name)
-		inCol := input.Columns[ind]
-		_, inIsRec := inCol.Type.(*zng.TypeRecord)
-		_, specIsRec := specCol.Type.(*zng.TypeRecord)
 
-		switch {
-		case inIsRec && specIsRec:
-			col, err := c.fillRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord))
-			if err != nil {
-				return nil, err
+	cols := make([]zng.Column, len(input.Columns), len(input.Columns)+len(spec.Columns))
+	copy(cols, input.Columns)
+	for _, specCol := range spec.Columns {
+		if i, ok := input.ColumnOfField(specCol.Name); ok {
+			// Field is present both in input and spec: recurse if
+			// both records, or select appropriate type if not.
+			if specRecType, ok := specCol.Type.(*zng.TypeRecord); ok {
+				if inRecType, ok := input.Columns[i].Type.(*zng.TypeRecord); ok {
+					filled, err := c.fillRecordType(inRecType, specRecType)
+					if err != nil {
+						return nil, err
+					}
+					cols[i] = zng.Column{specCol.Name, filled}
+				} else {
+					cols[i] = specCol
+				}
 			}
-			cols = append(cols, zng.Column{inCol.Name, col})
-
-		case !inIsRec && !specIsRec:
-			cols = append(cols, inCol)
-
-		case inIsRec && !specIsRec:
-			cols = append(cols, inCol)
-
-		case !inIsRec && specIsRec:
+		} else {
 			cols = append(cols, specCol)
-
 		}
 	}
 	return c.zctx.LookupTypeRecord(cols)
diff --git a/field/field.go b/field/field.go
index 3e985a971..9cc48bf7d 100644
--- a/field/field.go
+++ b/field/field.go
@@ -16,6 +16,9 @@ func NewRoot() Static {
 }
 
 func (f Static) String() string {
+	if len(f) == 0 {
+		return "."
+	}
 	return strings.Join(f, ".")
 }
 
@@ -35,6 +38,18 @@ func (f Static) Equal(to Static) bool {
 	return true
 }
 
+func (f Static) IsRoot() bool {
+	return len(f) == 0
+}
+
+func (f Static) HasStrictPrefix(prefix Static) bool {
+	return len(f) > len(prefix) && prefix.Equal(f[:len(prefix)])
+}
+
+func (f Static) HasPrefix(prefix Static) bool {
+	return len(f) >= len(prefix) && prefix.Equal(f[:len(prefix)])
+}
+
 func Dotted(s string) Static {
 	return strings.Split(s, ".")
 }
diff --git a/go.mod b/go.mod
index 4c70e45ba..ed7916991 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/go-pg/pg/v10 v10.7.3
+	github.com/go-redis/redis/v8 v8.4.11
 	github.com/go-resty/resty/v2 v2.2.0
 	github.com/golang-migrate/migrate/v4 v4.14.1
 	github.com/golang/mock v1.4.4
diff --git a/go.sum b/go.sum
index 702ac3b81..558a9e9cf 100644
--- a/go.sum
+++ b/go.sum
@@ -96,6 +96,8 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumC
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc h1:8WFBn63wegobsYAX0YjD+8suexZDga5CctH4CCTx2+8=
 github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc/go.mod h1:c9O8+fpSOX1DM8cPNSkX/qsBWdkD4yd2dpciOWQjpBw=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dhui/dktest v0.3.3 h1:DBuH/9GFaWbDRa42qsut/hbQu+srAQ0rPWnUoiGX7CA=
 github.com/dhui/dktest v0.3.3/go.mod h1:EML9sP4sqJELHn4jV7B0TY8oF6077nk83/tz7M56jcQ=
 github.com/docker/distribution v2.7.1+incompatible h1:a5mlkVzth6W5A4fOsS3D2EO5BUmsJpcB+cRlLU7cSug=
@@ -126,6 +128,8 @@ github.com/go-pg/pg/v10 v10.7.3 h1:oL/Hz5MJie/9epmwxlZjfReO+2wlLPOK6BeGb9I+XHk=
 github.com/go-pg/pg/v10 v10.7.3/go.mod h1:UsDYtA+ihbBNX1OeIvDejxkL4RXzL3wsZYoEv5NUEqM=
 github.com/go-pg/zerochecker v0.2.0 h1:pp7f72c3DobMWOb2ErtZsnrPaSvHd2W4o9//8HtF4mU=
 github.com/go-pg/zerochecker v0.2.0/go.mod h1:NJZ4wKL0NmTtz0GKCoJ8kym6Xn/EQzXRl2OnAe7MmDo=
+github.com/go-redis/redis/v8 v8.4.11 h1:t2lToev01VTrqYQcv+QFbxtGgcf64K+VUMgf9Ap6A/E=
+github.com/go-redis/redis/v8 v8.4.11/go.mod h1:d5yY/TlkQyYBSBHnXUmnf1OrHbyQere5JV4dLKwvXmo=
 github.com/go-resty/resty/v2 v2.2.0 h1:vgZ1cdblp8Aw4jZj3ZsKh6yKAlMg3CHMrqFSFFd+jgY=
 github.com/go-resty/resty/v2 v2.2.0/go.mod h1:nYW/8rxqQCmI3bPz9Fsmjbr2FBjGuR2Mzt6kDh3zZ7w=
 github.com/go-sql-driver/mysql v1.4.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
@@ -182,8 +186,9 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.3 h1:x95R7cp+rSeeqAMI2knLtQ0DKlaBhv2NrtrOvafPHRo=
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.4 h1:L8R9j+yAqZuZjsqh/z+F1NCffTKKLShY6zXTItVIZ8M=
+github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-github v17.0.0+incompatible/go.mod h1:zLgOLi98H3fifZn+44m+umXrS52loVEgC2AApnigrVQ=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -328,8 +333,9 @@ github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9k
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.9.0/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
-github.com/onsi/gomega v1.10.3 h1:gph6h/qe9GSUw1NhH1gp+qb+h8rXD8Cy60Z32Qw3ELA=
 github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
+github.com/onsi/gomega v1.10.4 h1:NiTx7EEvBzu9sFOD1zORteLSt3o8gnlvZZwSE9TnY9U=
+github.com/onsi/gomega v1.10.4/go.mod h1:g/HbgYopi++010VEqkFgJHKC09uJiW9UkXvMUuKHUCQ=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.0.1 h1:JMemWkRwHx4Zj+fVxWoMCFm/8sYGGrUVojFA6h/TRcI=
@@ -429,8 +435,9 @@ go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
-go.opentelemetry.io/otel v0.14.0 h1:YFBEfjCk9MTjaytCNSUkp9Q8lF7QJezA06T71FbQxLQ=
 go.opentelemetry.io/otel v0.14.0/go.mod h1:vH5xEuwy7Rts0GNtsCW3HYQoZDY+OmBJ6t1bFGGlxgw=
+go.opentelemetry.io/otel v0.16.0 h1:uIWEbdeb4vpKPGITLsRVUS44L5oDbDUCZxn8lkxhmgw=
+go.opentelemetry.io/otel v0.16.0/go.mod h1:e4GKElweB8W2gWUqbghw0B8t5MCTccc9212eNHnOHwA=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk=
@@ -529,8 +536,9 @@ golang.org/x/net v0.0.0-20200813134508-3edf25e44fcc/go.mod h1:/O7V0waA8r7cgGh81R
 golang.org/x/net v0.0.0-20200904194848-62affa334b73/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201029221708-28c70e62bb1d/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201110031124-69a78807bb2b h1:uwuIcX0g4Yl1NC5XAz37xsr2lTtcqevgzYNVt49waME=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb h1:eBmm0M9fYhWpKZLjQUUKka/LtIxf46G4fxeEz5KJr9U=
+golang.org/x/net v0.0.0-20201202161906-c7110b5ffcbb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/oauth2 v0.0.0-20180227000427-d7d64896b5ff/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181106182150-f42d05182288/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
diff --git a/pkg/iosrc/uri.go b/pkg/iosrc/uri.go
index e174f0330..8c4e5ec56 100644
--- a/pkg/iosrc/uri.go
+++ b/pkg/iosrc/uri.go
@@ -9,9 +9,9 @@ import (
 type URI url.URL
 
 const (
-	stdin  = "stdio:///stdin"
-	stdout = "stdio:///stdout"
-	stderr = "stdio:///stderr"
+	Stdin  = "stdio:///stdin"
+	Stdout = "stdio:///stdout"
+	Stderr = "stdio:///stderr"
 )
 
 // ParseURI parses the path using `url.Parse`. If the provided uri does not
diff --git a/ppl/archive/immcache/immcache.go b/ppl/archive/immcache/immcache.go
deleted file mode 100644
index a30d72437..000000000
--- a/ppl/archive/immcache/immcache.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// Package immcache contains facilities for caching immutable files for an
-// archive.
-package immcache
-
-import (
-	"context"
-	"flag"
-
-	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/prometheus/client_golang/prometheus"
-)
-
-type ImmutableCache interface {
-	ReadFile(context.Context, iosrc.URI) ([]byte, error)
-}
-
-type Config struct {
-	// LocalCacheSize specifies the number of immutable files to keep in a
-	// local lru cache used to speed up searches. Values less than or equal to 0
-	// (default), disables local caching of immutable files.
-	LocalCacheSize int
-}
-
-func (c *Config) SetFlags(fs *flag.FlagSet) {
-	fs.IntVar(&c.LocalCacheSize, "immcache.localsize", 0, "enables local caching of up to N small files (disabled if 0)")
-}
-
-func New(conf Config, reg prometheus.Registerer) (ImmutableCache, error) {
-	if conf.LocalCacheSize > 0 {
-		return NewLocalCache(conf.LocalCacheSize, reg)
-	}
-	return nil, nil
-}
diff --git a/ppl/cmd/zar/compact/command.go b/ppl/cmd/zar/compact/command.go
index d731471d1..2a37efd5a 100644
--- a/ppl/cmd/zar/compact/command.go
+++ b/ppl/cmd/zar/compact/command.go
@@ -5,8 +5,8 @@ import (
 	"flag"
 	"os"
 
-	"github.com/brimsec/zq/ppl/archive"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/mccanne/charm"
 )
 
@@ -39,16 +39,16 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 }
 
 func (c *Command) Run(args []string) error {
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
 	ctx := context.TODO()
-	if err := archive.Compact(ctx, ark, nil); err != nil {
+	if err := lake.Compact(ctx, lk, nil); err != nil {
 		return err
 	}
 	if c.purge {
-		return archive.Purge(ctx, ark, nil)
+		return lake.Purge(ctx, lk, nil)
 	}
 	return nil
 }
diff --git a/ppl/cmd/zar/find/command.go b/ppl/cmd/zar/find/command.go
index 21b63b79d..caa6a3e40 100644
--- a/ppl/cmd/zar/find/command.go
+++ b/ppl/cmd/zar/find/command.go
@@ -8,9 +8,9 @@ import (
 
 	"github.com/brimsec/zq/cli/outputflags"
 	"github.com/brimsec/zq/emitter"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/index"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng"
 	"github.com/brimsec/zq/zng/resolver"
@@ -75,7 +75,7 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 	f.StringVar(&c.root, "R", os.Getenv("ZAR_ROOT"), "root location of zar archive to walk")
 	f.BoolVar(&c.skipMissing, "Q", false, "skip errors caused by missing index files ")
 	f.StringVar(&c.indexFile, "x", "", "name of microindex for custom index searches")
-	f.StringVar(&c.pathField, "l", archive.DefaultAddPathField, "zng field name for path name of log file")
+	f.StringVar(&c.pathField, "l", lake.DefaultAddPathField, "zng field name for path name of log file")
 	f.BoolVar(&c.relativePaths, "relative", false, "display paths relative to root")
 	f.BoolVar(&c.zng, "z", false, "write results as zng stream rather than list of files")
 
@@ -91,7 +91,7 @@ func (c *Command) Run(args []string) error {
 		return err
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
@@ -101,12 +101,12 @@ func (c *Command) Run(args []string) error {
 		return err
 	}
 
-	var findOptions []archive.FindOption
+	var findOptions []lake.FindOption
 	if c.pathField != "" {
-		findOptions = append(findOptions, archive.AddPath(c.pathField, !c.relativePaths))
+		findOptions = append(findOptions, lake.AddPath(c.pathField, !c.relativePaths))
 	}
 	if c.skipMissing {
-		findOptions = append(findOptions, archive.SkipMissing())
+		findOptions = append(findOptions, lake.SkipMissing())
 	}
 
 	ctx, cancel := context.WithCancel(context.TODO())
@@ -127,7 +127,7 @@ func (c *Command) Run(args []string) error {
 	hits := make(chan *zng.Record)
 	var searchErr error
 	go func() {
-		searchErr = archive.Find(ctx, resolver.NewContext(), ark, query, hits, findOptions...)
+		searchErr = lake.Find(ctx, resolver.NewContext(), lk, query, hits, findOptions...)
 		close(hits)
 	}()
 	for hit := range hits {
diff --git a/ppl/cmd/zar/import/command.go b/ppl/cmd/zar/import/command.go
index a17d7bd67..8daa582ea 100644
--- a/ppl/cmd/zar/import/command.go
+++ b/ppl/cmd/zar/import/command.go
@@ -10,8 +10,8 @@ import (
 	"github.com/brimsec/zq/pkg/rlimit"
 	"github.com/brimsec/zq/pkg/signalctx"
 	"github.com/brimsec/zq/pkg/units"
-	"github.com/brimsec/zq/ppl/archive"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/mccanne/charm"
@@ -55,11 +55,11 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 	f.BoolVar(&c.asc, "asc", false, "store archive data in ascending order")
 	f.StringVar(&c.root, "R", os.Getenv("ZAR_ROOT"), "root location of zar archive to walk")
 	f.StringVar(&c.dataPath, "data", "", "location for storing data files (defaults to root)")
-	c.thresh = archive.DefaultLogSizeThreshold
+	c.thresh = lake.DefaultLogSizeThreshold
 	f.Var(&c.thresh, "s", "target size of chunk files, as '10MB' or '4GiB', etc.")
-	c.importBufSize = units.Bytes(archive.ImportBufSize)
+	c.importBufSize = units.Bytes(lake.ImportBufSize)
 	f.Var(&c.importBufSize, "bufsize", "maximum size of data read into memory before flushing to disk, as '99MB', '4GiB', etc.")
-	f.IntVar(&c.importStreamRecordMax, "streammax", archive.ImportStreamRecordsMax, "limit for number of records in each ZNG stream (0 for no limit)")
+	f.IntVar(&c.importStreamRecordMax, "streammax", lake.ImportStreamRecordsMax, "limit for number of records in each ZNG stream (0 for no limit)")
 	f.BoolVar(&c.empty, "empty", false, "create an archive without initial data")
 	c.inputFlags.SetFlags(f)
 	c.procFlags.SetFlags(f)
@@ -76,11 +76,11 @@ func (c *Command) Run(args []string) error {
 	} else if !c.empty && len(args) == 0 {
 		return errors.New("zar import: at least one input file must be specified (- for stdin)")
 	}
-	archive.ImportBufSize = int64(c.importBufSize)
-	archive.ImportStreamRecordsMax = c.importStreamRecordMax
+	lake.ImportBufSize = int64(c.importBufSize)
+	lake.ImportStreamRecordsMax = c.importStreamRecordMax
 
 	thresh := int64(c.thresh)
-	co := &archive.CreateOptions{
+	co := &lake.CreateOptions{
 		DataPath:         c.dataPath,
 		SortAscending:    c.asc,
 		LogSizeThreshold: &thresh,
@@ -90,7 +90,7 @@ func (c *Command) Run(args []string) error {
 		return err
 	}
 
-	ark, err := archive.CreateOrOpenArchive(c.root, co, nil)
+	lk, err := lake.CreateOrOpenLake(c.root, co, nil)
 	if err != nil {
 		return err
 	}
@@ -109,9 +109,9 @@ func (c *Command) Run(args []string) error {
 		return err
 	}
 	defer zbuf.CloseReaders(readers)
-	reader, err := zbuf.MergeReadersByTsAsReader(ctx, readers, ark.DataOrder)
+	reader, err := zbuf.MergeReadersByTsAsReader(ctx, readers, lk.DataOrder)
 	if err != nil {
 		return err
 	}
-	return archive.Import(ctx, ark, zctx, reader)
+	return lake.Import(ctx, lk, zctx, reader)
 }
diff --git a/ppl/cmd/zar/index/create.go b/ppl/cmd/zar/index/create.go
index 6b282540a..1484da0fc 100644
--- a/ppl/cmd/zar/index/create.go
+++ b/ppl/cmd/zar/index/create.go
@@ -11,9 +11,9 @@ import (
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/pkg/rlimit"
 	"github.com/brimsec/zq/pkg/signalctx"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/index"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/mccanne/charm"
 )
 
@@ -86,7 +86,7 @@ func (c *CreateCommand) Run(args []string) error {
 		return err
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
@@ -94,7 +94,7 @@ func (c *CreateCommand) Run(args []string) error {
 	ctx, cancel := signalctx.New(os.Interrupt)
 	defer cancel()
 
-	defs, err := c.addRules(ctx, ark, args)
+	defs, err := c.addRules(ctx, lk, args)
 	if err != nil {
 		return err
 	}
@@ -104,7 +104,7 @@ func (c *CreateCommand) Run(args []string) error {
 	}
 
 	if c.ensure {
-		defs, err = ark.ReadDefinitions(ctx)
+		defs, err = lk.ReadDefinitions(ctx)
 		if err != nil {
 			return err
 		}
@@ -115,13 +115,13 @@ func (c *CreateCommand) Run(args []string) error {
 		go c.displayProgress()
 	}
 
-	if err := archive.WriteIndices(ctx, ark, c.progress, defs...); err != nil {
+	if err := lake.WriteIndices(ctx, lk, c.progress, defs...); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (c *CreateCommand) addRules(ctx context.Context, ark *archive.Archive, args []string) ([]*index.Definition, error) {
+func (c *CreateCommand) addRules(ctx context.Context, lk *lake.Lake, args []string) ([]*index.Definition, error) {
 	if len(args) == 0 && c.zql == "" {
 		return nil, nil
 	}
@@ -150,7 +150,7 @@ func (c *CreateCommand) addRules(ctx context.Context, ark *archive.Archive, args
 		rules = append(rules, rule)
 	}
 
-	return archive.AddRules(ctx, ark, rules)
+	return lake.AddRules(ctx, lk, rules)
 }
 
 func (c *CreateCommand) displayProgress() {
diff --git a/ppl/cmd/zar/index/drop.go b/ppl/cmd/zar/index/drop.go
index 0057de009..a7afafed0 100644
--- a/ppl/cmd/zar/index/drop.go
+++ b/ppl/cmd/zar/index/drop.go
@@ -7,9 +7,9 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/index"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/mccanne/charm"
 	"github.com/segmentio/ksuid"
 )
@@ -55,12 +55,12 @@ func (c *DropCommand) run(args []string) error {
 		return errors.New("no index definition specified")
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
 
-	alldefs, err := ark.ReadDefinitions(context.TODO())
+	alldefs, err := lk.ReadDefinitions(context.TODO())
 	if err != nil {
 		return err
 	}
@@ -83,7 +83,7 @@ func (c *DropCommand) run(args []string) error {
 		return errors.New("no definitions deleted")
 	}
 
-	if err := archive.RemoveDefinitions(context.TODO(), ark, defs...); err != nil {
+	if err := lake.RemoveDefinitions(context.TODO(), lk, defs...); err != nil {
 		return err
 	}
 
@@ -92,7 +92,7 @@ func (c *DropCommand) run(args []string) error {
 			c.progress = make(chan string)
 			go c.displayProgress()
 		}
-		if err := archive.RemoveIndices(context.TODO(), ark, c.progress, defs...); err != nil {
+		if err := lake.RemoveIndices(context.TODO(), lk, c.progress, defs...); err != nil {
 			return err
 		}
 	}
diff --git a/ppl/cmd/zar/index/ls.go b/ppl/cmd/zar/index/ls.go
index 3f4455341..7c00aefbd 100644
--- a/ppl/cmd/zar/index/ls.go
+++ b/ppl/cmd/zar/index/ls.go
@@ -9,9 +9,9 @@ import (
 	"strings"
 
 	"github.com/brimsec/zq/cli/outputflags"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/index"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/mccanne/charm"
 	"github.com/segmentio/ksuid"
@@ -66,12 +66,12 @@ func (c *LsCommand) Run(args []string) error {
 		return errors.New("a directory must be specified with -R or ZAR_ROOT")
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
 
-	defs, err := ark.ReadDefinitions(context.TODO())
+	defs, err := lk.ReadDefinitions(context.TODO())
 	if err != nil {
 		return err
 	}
@@ -82,9 +82,9 @@ func (c *LsCommand) Run(args []string) error {
 	}
 	defer w.Close()
 
-	var stats map[ksuid.KSUID]archive.IndexInfo
+	var stats map[ksuid.KSUID]lake.IndexInfo
 	if c.stats {
-		if stats, err = c.getStats(ark, defs); err != nil {
+		if stats, err = c.getStats(lk, defs); err != nil {
 			return err
 		}
 	}
@@ -135,12 +135,12 @@ func (c *LsCommand) Run(args []string) error {
 	return nil
 }
 
-func (c *LsCommand) getStats(ark *archive.Archive, defs []*index.Definition) (map[ksuid.KSUID]archive.IndexInfo, error) {
-	stats, err := archive.IndexStat(context.TODO(), ark, defs)
+func (c *LsCommand) getStats(lk *lake.Lake, defs []*index.Definition) (map[ksuid.KSUID]lake.IndexInfo, error) {
+	stats, err := lake.IndexStat(context.TODO(), lk, defs)
 	if err != nil {
 		return nil, err
 	}
-	m := make(map[ksuid.KSUID]archive.IndexInfo)
+	m := make(map[ksuid.KSUID]lake.IndexInfo)
 	for _, stat := range stats {
 		m[stat.DefinitionID] = stat
 	}
diff --git a/ppl/cmd/zar/ls/command.go b/ppl/cmd/zar/ls/command.go
index 845ded516..dbc93c4d3 100644
--- a/ppl/cmd/zar/ls/command.go
+++ b/ppl/cmd/zar/ls/command.go
@@ -9,10 +9,10 @@ import (
 	"strings"
 
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/mccanne/charm"
 )
 
@@ -33,7 +33,7 @@ func init() {
 
 type Command struct {
 	*root.Command
-	ark           *archive.Archive
+	lk            *lake.Lake
 	root          string
 	lflag         bool
 	indexDesc     bool
@@ -60,12 +60,12 @@ func (c *Command) Run(args []string) error {
 	}
 
 	var err error
-	c.ark, err = archive.OpenArchive(c.root, nil)
+	c.lk, err = lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
 
-	defs, err := c.ark.ReadDefinitions(context.TODO())
+	defs, err := c.lk.ReadDefinitions(context.TODO())
 	if err != nil {
 		return err
 	}
@@ -77,19 +77,19 @@ func (c *Command) Run(args []string) error {
 		pattern = args[0]
 	}
 	if c.spanInfos {
-		return archive.SpanWalk(context.TODO(), c.ark, nano.MaxSpan, func(si archive.SpanInfo) error {
+		return lake.SpanWalk(context.TODO(), c.lk, nano.MaxSpan, func(si lake.SpanInfo) error {
 			c.printSpanInfo(si, pattern)
 			return nil
 		})
 	}
-	return archive.Walk(context.TODO(), c.ark, func(chunk chunk.Chunk) error {
+	return lake.Walk(context.TODO(), c.lk, func(chunk chunk.Chunk) error {
 		c.printChunk(0, chunk, pattern)
 		return nil
 	})
 }
 
-func (c *Command) printSpanInfo(si archive.SpanInfo, pattern string) {
-	fmt.Println(si.Range(c.ark.DataOrder) + ":")
+func (c *Command) printSpanInfo(si lake.SpanInfo, pattern string) {
+	fmt.Println(si.Range(c.lk.DataOrder) + ":")
 	for _, chunk := range si.Chunks {
 		c.printChunk(1, chunk, pattern)
 	}
@@ -147,7 +147,7 @@ func (c *Command) chunkString(chunk chunk.Chunk) string {
 		return chunk.Range()
 	}
 	if c.relativePaths {
-		return strings.TrimSuffix(c.ark.DataPath.RelPath(chunk.ZarDir()), "/")
+		return strings.TrimSuffix(c.lk.DataPath.RelPath(chunk.ZarDir()), "/")
 	}
 	return chunk.ZarDir().String()
 }
diff --git a/ppl/cmd/zar/map/command.go b/ppl/cmd/zar/map/command.go
index 3680ecc38..e93002ac0 100644
--- a/ppl/cmd/zar/map/command.go
+++ b/ppl/cmd/zar/map/command.go
@@ -14,9 +14,9 @@ import (
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/rlimit"
 	"github.com/brimsec/zq/pkg/signalctx"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/chunk"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio"
 	"github.com/brimsec/zq/zio/detector"
@@ -93,14 +93,14 @@ func (c *Command) Run(args []string) error {
 	ctx, cancel := signalctx.New(os.Interrupt)
 	defer cancel()
 
-	ark, err := archive.OpenArchiveWithContext(ctx, c.root, nil)
+	lk, err := lake.OpenLakeWithContext(ctx, c.root, nil)
 	if err != nil {
 		return err
 	}
 
 	// XXX this is parallelizable except for writing to stdout when
 	// concatenating results
-	return archive.Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	return lake.Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		zardir := chunk.ZarDir()
 		var paths []string
 		for _, input := range inputs {
@@ -124,7 +124,7 @@ func (c *Command) Run(args []string) error {
 		}
 		err = driver.Run(ctx, d, query, zctx, reader, driver.Config{
 			ReaderSortKey:     "ts",
-			ReaderSortReverse: ark.DataOrder == zbuf.OrderDesc,
+			ReaderSortReverse: lk.DataOrder == zbuf.OrderDesc,
 		})
 		if closeErr := writer.Close(); closeErr != nil && err == nil {
 			err = closeErr
diff --git a/ppl/cmd/zar/rm/command.go b/ppl/cmd/zar/rm/command.go
index 71f9c0af1..cdf0eff76 100644
--- a/ppl/cmd/zar/rm/command.go
+++ b/ppl/cmd/zar/rm/command.go
@@ -10,9 +10,9 @@ import (
 	"strings"
 
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/chunk"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	"github.com/brimsec/zq/zqe"
 	"github.com/mccanne/charm"
 )
@@ -55,36 +55,36 @@ func (c *Command) Run(args []string) error {
 		return errors.New("zar rm: no archive root specified with -R or ZAR_ROOT")
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
 
-	return archive.Walk(context.TODO(), ark, func(chunk chunk.Chunk) error {
-		return c.remove(ark, chunk, args)
+	return lake.Walk(context.TODO(), lk, func(chunk chunk.Chunk) error {
+		return c.remove(lk, chunk, args)
 	})
 }
 
-func (c *Command) remove(ark *archive.Archive, chunk chunk.Chunk, names []string) error {
+func (c *Command) remove(lk *lake.Lake, chunk chunk.Chunk, names []string) error {
 	for _, name := range names {
 		path := chunk.ZarDir().AppendPath(name)
 		if err := iosrc.Remove(context.TODO(), path); err != nil {
 			if zqe.IsNotFound(err) {
-				fmt.Printf("%s: not found\n", c.printable(ark, chunk, name))
+				fmt.Printf("%s: not found\n", c.printable(lk, chunk, name))
 				continue
 			}
 			return err
 		}
-		fmt.Printf("%s: removed\n", c.printable(ark, chunk, name))
+		fmt.Printf("%s: removed\n", c.printable(lk, chunk, name))
 	}
 	return nil
 }
-func (c *Command) printable(ark *archive.Archive, chunk chunk.Chunk, objPath string) string {
+func (c *Command) printable(lk *lake.Lake, chunk chunk.Chunk, objPath string) string {
 	if c.showRanges {
 		return path.Join(chunk.Range(), objPath)
 	}
 	if c.relativePaths {
-		return strings.TrimSuffix(ark.DataPath.RelPath(chunk.ZarDir().AppendPath(objPath)), "/")
+		return strings.TrimSuffix(lk.DataPath.RelPath(chunk.ZarDir().AppendPath(objPath)), "/")
 	}
 	return chunk.ZarDir().AppendPath(objPath).String()
 }
diff --git a/ppl/cmd/zar/rmdirs/command.go b/ppl/cmd/zar/rmdirs/command.go
index 812534451..d4e1d1f07 100644
--- a/ppl/cmd/zar/rmdirs/command.go
+++ b/ppl/cmd/zar/rmdirs/command.go
@@ -5,8 +5,8 @@ import (
 	"flag"
 	"os"
 
-	"github.com/brimsec/zq/ppl/archive"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/mccanne/charm"
 )
 
@@ -39,9 +39,9 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 }
 
 func (c *Command) Run(args []string) error {
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
-	return archive.RmDirs(context.TODO(), ark)
+	return lake.RmDirs(context.TODO(), lk)
 }
diff --git a/ppl/cmd/zar/stat/command.go b/ppl/cmd/zar/stat/command.go
index 43abca97c..82fff371f 100644
--- a/ppl/cmd/zar/stat/command.go
+++ b/ppl/cmd/zar/stat/command.go
@@ -7,8 +7,8 @@ import (
 	"os"
 
 	"github.com/brimsec/zq/emitter"
-	"github.com/brimsec/zq/ppl/archive"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio"
 	"github.com/brimsec/zq/zng/resolver"
@@ -48,7 +48,7 @@ func (c *Command) Run(args []string) (err error) {
 		return errors.New("zar stat: too many arguments")
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
@@ -64,7 +64,7 @@ func (c *Command) Run(args []string) (err error) {
 		}
 	}()
 
-	rc, err := archive.Stat(context.Background(), resolver.NewContext(), ark)
+	rc, err := lake.Stat(context.Background(), resolver.NewContext(), lk)
 	if err != nil {
 		return err
 	}
diff --git a/ppl/cmd/zar/zq/command.go b/ppl/cmd/zar/zq/command.go
index 23fe77fc1..329d2f13e 100644
--- a/ppl/cmd/zar/zq/command.go
+++ b/ppl/cmd/zar/zq/command.go
@@ -13,8 +13,8 @@ import (
 	"github.com/brimsec/zq/driver"
 	"github.com/brimsec/zq/pkg/rlimit"
 	"github.com/brimsec/zq/pkg/signalctx"
-	"github.com/brimsec/zq/ppl/archive"
 	"github.com/brimsec/zq/ppl/cmd/zar/root"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/mccanne/charm"
 )
@@ -75,11 +75,11 @@ func (c *Command) Run(args []string) error {
 		return err
 	}
 
-	ark, err := archive.OpenArchive(c.root, nil)
+	lk, err := lake.OpenLake(c.root, nil)
 	if err != nil {
 		return err
 	}
-	msrc := archive.NewMultiSource(ark, args[1:])
+	msrc := lake.NewMultiSource(lk, args[1:])
 
 	ctx, cancel := signalctx.New(os.Interrupt)
 	defer cancel()
@@ -104,7 +104,7 @@ func (c *Command) Run(args []string) error {
 	return err
 }
 
-func (c *Command) printStats(msrc archive.MultiSource) {
+func (c *Command) printStats(msrc lake.MultiSource) {
 	if c.stats {
 		stats := msrc.Stats()
 		w := tabwriter.NewWriter(os.Stderr, 0, 0, 1, ' ', 0)
diff --git a/ppl/cmd/zqd/listen/command.go b/ppl/cmd/zqd/listen/command.go
index 6e3e3a8f0..3287b9deb 100644
--- a/ppl/cmd/zqd/listen/command.go
+++ b/ppl/cmd/zqd/listen/command.go
@@ -72,6 +72,7 @@ func New(parent charm.Command, f *flag.FlagSet) (charm.Command, error) {
 	c.conf.Worker.SetFlags(f)
 	c.conf.DB.SetFlags(f)
 	c.conf.ImmutableCache.SetFlags(f)
+	c.conf.Redis.SetFlags(f)
 	c.conf.Version = cli.Version
 	f.IntVar(&c.brimfd, "brimfd", -1, "pipe read fd passed by brim to signal brim closure")
 	f.StringVar(&c.configfile, "config", "", "path to zqd config file")
diff --git a/ppl/archive/chunk/chunk.go b/ppl/lake/chunk/chunk.go
similarity index 100%
rename from ppl/archive/chunk/chunk.go
rename to ppl/lake/chunk/chunk.go
diff --git a/ppl/archive/chunk/metadata.go b/ppl/lake/chunk/metadata.go
similarity index 100%
rename from ppl/archive/chunk/metadata.go
rename to ppl/lake/chunk/metadata.go
diff --git a/ppl/archive/chunk/reader.go b/ppl/lake/chunk/reader.go
similarity index 96%
rename from ppl/archive/chunk/reader.go
rename to ppl/lake/chunk/reader.go
index 38fcc3776..e06ceceed 100644
--- a/ppl/archive/chunk/reader.go
+++ b/ppl/lake/chunk/reader.go
@@ -7,7 +7,7 @@ import (
 
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/seekindex"
+	"github.com/brimsec/zq/ppl/lake/seekindex"
 	"github.com/brimsec/zq/zqe"
 )
 
diff --git a/ppl/archive/chunk/writer.go b/ppl/lake/chunk/writer.go
similarity index 98%
rename from ppl/archive/chunk/writer.go
rename to ppl/lake/chunk/writer.go
index 75b363f10..1cc04f7c8 100644
--- a/ppl/archive/chunk/writer.go
+++ b/ppl/lake/chunk/writer.go
@@ -8,8 +8,8 @@ import (
 	"github.com/brimsec/zq/pkg/bufwriter"
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/index"
-	"github.com/brimsec/zq/ppl/archive/seekindex"
+	"github.com/brimsec/zq/ppl/lake/index"
+	"github.com/brimsec/zq/ppl/lake/seekindex"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/zngio"
 	"github.com/brimsec/zq/zng"
diff --git a/ppl/archive/finder.go b/ppl/lake/finder.go
similarity index 81%
rename from ppl/archive/finder.go
rename to ppl/lake/finder.go
index 6f3cb3c10..11f8bac2d 100644
--- a/ppl/archive/finder.go
+++ b/ppl/lake/finder.go
@@ -1,10 +1,10 @@
-package archive
+package lake
 
 import (
 	"context"
 
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng"
 	"github.com/brimsec/zq/zng/resolver"
@@ -14,7 +14,7 @@ import (
 type findOptions struct {
 	skipMissing bool
 	zctx        *resolver.Context
-	addPath     func(ark *Archive, chunk chunk.Chunk, rec *zng.Record) (*zng.Record, error)
+	addPath     func(lk *Lake, chunk chunk.Chunk, rec *zng.Record) (*zng.Record, error)
 }
 
 type FindOption func(*findOptions) error
@@ -45,16 +45,16 @@ func AddPath(pathField string, absolutePath bool) FindOption {
 			{"first", zng.TypeTime},
 			{"last", zng.TypeTime},
 		}
-		opt.addPath = func(ark *Archive, chunk chunk.Chunk, rec *zng.Record) (*zng.Record, error) {
+		opt.addPath = func(lk *Lake, chunk chunk.Chunk, rec *zng.Record) (*zng.Record, error) {
 			var path string
 			if absolutePath {
-				if ark.DataPath.Scheme == "file" {
+				if lk.DataPath.Scheme == "file" {
 					path = chunk.Path().Filepath()
 				} else {
 					path = chunk.Path().String()
 				}
 			} else {
-				path = ark.Root.RelPath(chunk.Path())
+				path = lk.Root.RelPath(chunk.Path())
 			}
 			return opt.zctx.AddColumns(rec, cols, []zng.Value{
 				{cols[0].Type, zng.EncodeString(path)},
@@ -82,17 +82,17 @@ func AddPath(pathField string, absolutePath bool) FindOption {
 // be more efficient at large scale to allow multipe patterns that
 // are effectively OR-ed together so that there is locality of
 // access to the microindex files.
-func Find(ctx context.Context, zctx *resolver.Context, ark *Archive, query index.Query, hits chan<- *zng.Record, opts ...FindOption) error {
+func Find(ctx context.Context, zctx *resolver.Context, lk *Lake, query index.Query, hits chan<- *zng.Record, opts ...FindOption) error {
 	opt := findOptions{
 		zctx:    zctx,
-		addPath: func(_ *Archive, _ chunk.Chunk, rec *zng.Record) (*zng.Record, error) { return rec, nil },
+		addPath: func(_ *Lake, _ chunk.Chunk, rec *zng.Record) (*zng.Record, error) { return rec, nil },
 	}
 	for _, o := range opts {
 		if err := o(&opt); err != nil {
 			return err
 		}
 	}
-	defs, err := ark.ReadDefinitions(ctx)
+	defs, err := lk.ReadDefinitions(ctx)
 	if err != nil {
 		return err
 	}
@@ -100,7 +100,7 @@ func Find(ctx context.Context, zctx *resolver.Context, ark *Archive, query index
 	if !ok {
 		return zqe.ErrInvalid("no matching index rule found")
 	}
-	return Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	return Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		dir := chunk.ZarDir()
 		rec, err := index.Find(ctx, zctx, dir, matched.DefID, matched.Values...)
 		if rec == nil || (err != nil && zqe.IsNotFound(err) && opt.skipMissing) {
@@ -111,7 +111,7 @@ func Find(ctx context.Context, zctx *resolver.Context, ark *Archive, query index
 			return nil
 		}
 
-		if rec, err = opt.addPath(ark, chunk, rec); err != nil {
+		if rec, err = opt.addPath(lk, chunk, rec); err != nil {
 			return err
 		}
 
@@ -148,7 +148,7 @@ func (f *findReadCloser) Close() error {
 	return nil
 }
 
-func FindReadCloser(ctx context.Context, zctx *resolver.Context, ark *Archive, query index.Query, opts ...FindOption) (zbuf.ReadCloser, error) {
+func FindReadCloser(ctx context.Context, zctx *resolver.Context, lk *Lake, query index.Query, opts ...FindOption) (zbuf.ReadCloser, error) {
 	ctx, cancel := context.WithCancel(ctx)
 	f := &findReadCloser{
 		ctx:    ctx,
@@ -156,7 +156,7 @@ func FindReadCloser(ctx context.Context, zctx *resolver.Context, ark *Archive, q
 		hits:   make(chan *zng.Record),
 	}
 	go func() {
-		f.err = Find(ctx, zctx, ark, query, f.hits, opts...)
+		f.err = Find(ctx, zctx, lk, query, f.hits, opts...)
 		close(f.hits)
 	}()
 	return f, nil
diff --git a/ppl/lake/immcache/immcache.go b/ppl/lake/immcache/immcache.go
new file mode 100644
index 000000000..8d4a32013
--- /dev/null
+++ b/ppl/lake/immcache/immcache.go
@@ -0,0 +1,97 @@
+// Package immcache contains facilities for caching immutable files for a lake.
+package immcache
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/brimsec/zq/pkg/iosrc"
+	"github.com/go-redis/redis/v8"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type Kind string
+
+const (
+	KindNone  Kind = "none"
+	KindLocal Kind = "local"
+	KindRedis Kind = "redis"
+)
+
+func (k *Kind) Set(s string) error {
+	switch s {
+	case "none", "":
+		*k = KindNone
+	case "local":
+		*k = KindLocal
+	case "redis":
+		*k = KindRedis
+	default:
+		return fmt.Errorf("unknown immutable cache kind: %q", s)
+	}
+	return nil
+}
+
+func (k Kind) String() string {
+	return string(k)
+}
+
+type ImmutableCache interface {
+	ReadFile(context.Context, iosrc.URI) ([]byte, error)
+}
+
+type Config struct {
+	Kind Kind
+	// LocalCacheSize specifies the number of immutable files to keep in a
+	// local lru cache used to speed up searches. Values less than or equal to 0
+	// (default), disables local caching of immutable files.
+	LocalCacheSize int
+	// RedisKeyExpiration is the expiration value used when creating keys.
+	// A value of zero (meaning no expiration) should only be used when
+	// Redis is configured with a key eviction policy.
+	RedisKeyExpiration time.Duration
+}
+
+func (c *Config) SetFlags(fs *flag.FlagSet) {
+	fs.Var(&c.Kind, "immcache.kind", "kind of immutable cache")
+	fs.IntVar(&c.LocalCacheSize, "immcache.local.size", 128, "number of small files to keep in local cache")
+	fs.DurationVar(&c.RedisKeyExpiration, "immcache.redis.keyexpiry", time.Hour*24, "expiration duration of immutable keys")
+}
+
+func New(conf Config, rclient *redis.Client, reg prometheus.Registerer) (ImmutableCache, error) {
+	switch conf.Kind {
+	case KindLocal:
+		return NewLocalCache(conf.LocalCacheSize, reg)
+	case KindRedis:
+		return NewRedisCache(rclient, conf, reg), nil
+	}
+	return nil, nil
+}
+
+type metrics struct {
+	hits   *prometheus.CounterVec
+	misses *prometheus.CounterVec
+}
+
+func newMetrics(reg prometheus.Registerer) metrics {
+	factory := promauto.With(reg)
+	return metrics{
+		hits: factory.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "archive_cache_hits_total",
+				Help: "Number of hits for a cache lookup.",
+			},
+			[]string{"kind"},
+		),
+		misses: factory.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: "archive_cache_misses_total",
+				Help: "Number of misses for a cache lookup.",
+			},
+			[]string{"kind"},
+		),
+	}
+}
diff --git a/ppl/archive/immcache/local.go b/ppl/lake/immcache/local.go
similarity index 59%
rename from ppl/archive/immcache/local.go
rename to ppl/lake/immcache/local.go
index 223cb35cb..aedeb8b3c 100644
--- a/ppl/archive/immcache/local.go
+++ b/ppl/lake/immcache/local.go
@@ -5,16 +5,14 @@ import (
 	"path"
 
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive/chunk"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
 type LocalCache struct {
-	lru    *lru.ARCCache
-	hits   *prometheus.CounterVec
-	misses *prometheus.CounterVec
+	metrics
+	lru *lru.ARCCache
 }
 
 func NewLocalCache(size int, registerer prometheus.Registerer) (*LocalCache, error) {
@@ -25,23 +23,9 @@ func NewLocalCache(size int, registerer prometheus.Registerer) (*LocalCache, err
 	if registerer == nil {
 		registerer = prometheus.NewRegistry()
 	}
-	factory := promauto.With(registerer)
 	return &LocalCache{
-		lru: lru,
-		hits: factory.NewCounterVec(
-			prometheus.CounterOpts{
-				Name: "archive_cache_hits_total",
-				Help: "Number of hits for a cache lookup.",
-			},
-			[]string{"kind"},
-		),
-		misses: factory.NewCounterVec(
-			prometheus.CounterOpts{
-				Name: "archive_cache_misses_total",
-				Help: "Number of misses for a cache lookup.",
-			},
-			[]string{"kind"},
-		),
+		metrics: newMetrics(registerer),
+		lru:     lru,
 	}, nil
 }
 
diff --git a/ppl/lake/immcache/redis.go b/ppl/lake/immcache/redis.go
new file mode 100644
index 000000000..7dfe23eb7
--- /dev/null
+++ b/ppl/lake/immcache/redis.go
@@ -0,0 +1,46 @@
+package immcache
+
+import (
+	"context"
+	"errors"
+	"path"
+	"time"
+
+	"github.com/brimsec/zq/pkg/iosrc"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/go-redis/redis/v8"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type RedisCache struct {
+	metrics
+	client *redis.Client
+	expiry time.Duration
+}
+
+func NewRedisCache(client *redis.Client, conf Config, reg prometheus.Registerer) *RedisCache {
+	return &RedisCache{
+		metrics: newMetrics(reg),
+		expiry:  time.Duration(conf.RedisKeyExpiration),
+		client:  client,
+	}
+}
+
+func (c *RedisCache) ReadFile(ctx context.Context, u iosrc.URI) ([]byte, error) {
+	kind, _, _ := chunk.FileMatch(path.Base(u.Path))
+	res := c.client.Get(ctx, u.String())
+	if err := res.Err(); err == nil {
+		c.hits.WithLabelValues(kind.Description()).Inc()
+		return res.Bytes()
+	} else if !errors.Is(err, redis.Nil) {
+		return nil, err
+	}
+
+	b, err := iosrc.ReadFile(ctx, u)
+	if err != nil {
+		return nil, err
+	}
+
+	c.misses.WithLabelValues(kind.Description()).Inc()
+	return b, c.client.Set(ctx, u.String(), b, c.expiry).Err()
+}
diff --git a/ppl/archive/import.go b/ppl/lake/import.go
similarity index 89%
rename from ppl/archive/import.go
rename to ppl/lake/import.go
index aae46df5c..d5e4fba7f 100644
--- a/ppl/archive/import.go
+++ b/ppl/lake/import.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -10,8 +10,8 @@ import (
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/proc/sort"
 	"github.com/brimsec/zq/proc/spill"
 	"github.com/brimsec/zq/zbuf"
@@ -34,8 +34,8 @@ var (
 
 const importDefaultStaleDuration = time.Second * 5
 
-func Import(ctx context.Context, ark *Archive, zctx *resolver.Context, r zbuf.Reader) error {
-	w, err := NewWriter(ctx, ark)
+func Import(ctx context.Context, lk *Lake, zctx *resolver.Context, r zbuf.Reader) error {
+	w, err := NewWriter(ctx, lk)
 	if err != nil {
 		return err
 	}
@@ -55,7 +55,7 @@ func Import(ctx context.Context, ark *Archive, zctx *resolver.Context, r zbuf.Re
 // written to temporary files. At some point this should have a maxTempFileSize
 // to ensure the Writer does not exceed the size of a provisioned tmpfs.
 type Writer struct {
-	ark           *Archive
+	lk            *Lake
 	cancel        context.CancelFunc
 	ctx           context.Context
 	defs          index.Definitions
@@ -71,15 +71,15 @@ type Writer struct {
 
 // NewWriter creates a zbuf.Writer compliant writer for writing data to an
 // archive.
-func NewWriter(ctx context.Context, ark *Archive) (*Writer, error) {
-	defs, err := ark.ReadDefinitions(ctx)
+func NewWriter(ctx context.Context, lk *Lake) (*Writer, error) {
+	defs, err := lk.ReadDefinitions(ctx)
 	if err != nil {
 		return nil, err
 	}
 	ctx, cancel := context.WithCancel(ctx)
 	g, ctx := errgroup.WithContext(ctx)
 	return &Writer{
-		ark:           ark,
+		lk:            lk,
 		cancel:        cancel,
 		ctx:           ctx,
 		errgroup:      g,
@@ -231,10 +231,10 @@ func (s *ImportStats) Copy() ImportStats {
 
 // tsDirWriter accumulates records for one tsDir.
 // When the expected size of writing the records is greater than
-// ark.LogSizeThreshold, they are written to a chunk file in
+// lk.LogSizeThreshold, they are written to a chunk file in
 // the archive.
 type tsDirWriter struct {
-	ark     *Archive
+	lk      *Lake
 	bufSize int64
 	ctx     context.Context
 	modts   nano.Ts
@@ -246,12 +246,12 @@ type tsDirWriter struct {
 
 func newTsDirWriter(w *Writer, tsDir tsDir) (*tsDirWriter, error) {
 	d := &tsDirWriter{
-		ark:    w.ark,
+		lk:     w.lk,
 		ctx:    w.ctx,
 		tsDir:  tsDir,
 		writer: w,
 	}
-	if err := iosrc.MkdirAll(tsDir.path(w.ark), 0755); err != nil {
+	if err := iosrc.MkdirAll(tsDir.path(w.lk), 0755); err != nil {
 		return nil, err
 	}
 	return d, nil
@@ -281,7 +281,7 @@ func (dw *tsDirWriter) Write(rec *zng.Record) error {
 	dw.records = append(dw.records, rec)
 	dw.addBufSize(int64(len(rec.Raw)))
 	dw.touch()
-	if dw.chunkSizeEstimate() > dw.ark.LogSizeThreshold {
+	if dw.chunkSizeEstimate() > dw.lk.LogSizeThreshold {
 		if err := dw.flush(); err != nil {
 			return err
 		}
@@ -299,7 +299,7 @@ func (dw *tsDirWriter) spill() error {
 	}
 	if dw.spiller == nil {
 		var err error
-		dw.spiller, err = spill.NewMergeSort(importCompareFn(dw.ark))
+		dw.spiller, err = spill.NewMergeSort(importCompareFn(dw.lk))
 		if err != nil {
 			return err
 		}
@@ -327,11 +327,11 @@ func (dw *tsDirWriter) flush() error {
 		if len(dw.records) == 0 {
 			return nil
 		}
-		expr.SortStable(dw.records, importCompareFn(dw.ark))
+		expr.SortStable(dw.records, importCompareFn(dw.lk))
 		r = zbuf.Array(dw.records).NewReader()
 	}
-	w, err := chunk.NewWriter(dw.ctx, dw.tsDir.path(dw.ark), chunk.WriterOpts{
-		Order:       dw.ark.DataOrder,
+	w, err := chunk.NewWriter(dw.ctx, dw.tsDir.path(dw.lk), chunk.WriterOpts{
+		Order:       dw.lk.DataOrder,
 		Definitions: dw.writer.defs,
 		Zng: zngio.WriterOpts{
 			StreamRecordsMax: ImportStreamRecordsMax,
@@ -358,6 +358,6 @@ func (dw *tsDirWriter) flush() error {
 	return nil
 }
 
-func importCompareFn(ark *Archive) expr.CompareFn {
-	return zbuf.NewCompareFn(field.New("ts"), ark.DataOrder == zbuf.OrderDesc)
+func importCompareFn(lk *Lake) expr.CompareFn {
+	return zbuf.NewCompareFn(field.New("ts"), lk.DataOrder == zbuf.OrderDesc)
 }
diff --git a/ppl/archive/index.go b/ppl/lake/index.go
similarity index 63%
rename from ppl/archive/index.go
rename to ppl/lake/index.go
index 1a2a6bafc..7783c3795 100644
--- a/ppl/archive/index.go
+++ b/ppl/lake/index.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -6,8 +6,8 @@ import (
 	"runtime"
 
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zio/zngio"
 	"github.com/brimsec/zq/zng/resolver"
 	"golang.org/x/sync/errgroup"
@@ -15,33 +15,33 @@ import (
 )
 
 // EnsureIndices walks through the entirety of an Achive's chunks ensuring that
-// all chunk indices are up-to-date with an Archive's IndexDefs. If the progress
+// all chunk indices are up-to-date with an Lake's IndexDefs. If the progress
 // channel is not nil, the paths of indices affected will be sent over it.
-func EnsureIndices(ctx context.Context, ark *Archive, progress chan<- string) error {
-	defs, err := ark.ReadDefinitions(ctx)
+func EnsureIndices(ctx context.Context, lk *Lake, progress chan<- string) error {
+	defs, err := lk.ReadDefinitions(ctx)
 	if err != nil {
 		return err
 	}
-	return WriteIndices(ctx, ark, progress, defs...)
+	return WriteIndices(ctx, lk, progress, defs...)
 }
 
-func AddRules(ctx context.Context, ark *Archive, rules []index.Rule) ([]*index.Definition, error) {
-	if err := iosrc.MkdirAll(ark.DefinitionsDir(), 0700); err != nil {
+func AddRules(ctx context.Context, lk *Lake, rules []index.Rule) ([]*index.Definition, error) {
+	if err := iosrc.MkdirAll(lk.DefinitionsDir(), 0700); err != nil {
 		return nil, err
 	}
-	return index.WriteRules(ctx, ark.DefinitionsDir(), rules)
+	return index.WriteRules(ctx, lk.DefinitionsDir(), rules)
 }
 
-func ApplyRules(ctx context.Context, ark *Archive, progress chan<- string, rules ...index.Rule) error {
-	defs, err := AddRules(ctx, ark, rules)
+func ApplyRules(ctx context.Context, lk *Lake, progress chan<- string, rules ...index.Rule) error {
+	defs, err := AddRules(ctx, lk, rules)
 	if err != nil {
 		return err
 	}
-	return WriteIndices(ctx, ark, progress, defs...)
+	return WriteIndices(ctx, lk, progress, defs...)
 }
 
-func RemoveDefinitions(ctx context.Context, ark *Archive, defs ...*index.Definition) error {
-	dir := ark.DefinitionsDir()
+func RemoveDefinitions(ctx context.Context, lk *Lake, defs ...*index.Definition) error {
+	dir := lk.DefinitionsDir()
 	for _, def := range defs {
 		if err := index.RemoveDefinition(ctx, dir, def.ID); err != nil {
 			return err
@@ -50,11 +50,11 @@ func RemoveDefinitions(ctx context.Context, ark *Archive, defs ...*index.Definit
 	return nil
 }
 
-func WriteIndices(ctx context.Context, ark *Archive, updates chan<- string, defs ...*index.Definition) error {
+func WriteIndices(ctx context.Context, lk *Lake, updates chan<- string, defs ...*index.Definition) error {
 	prog := progress(updates)
 	sem := semaphore.NewWeighted(int64(runtime.GOMAXPROCS(0)))
 	g, ctx := errgroup.WithContext(ctx)
-	err := Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	err := Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		if err := sem.Acquire(ctx, 1); err != nil {
 			return err
 		}
@@ -75,9 +75,9 @@ func WriteIndices(ctx context.Context, ark *Archive, updates chan<- string, defs
 	return err
 }
 
-func RemoveIndices(ctx context.Context, ark *Archive, updates chan<- string, defs ...*index.Definition) error {
+func RemoveIndices(ctx context.Context, lk *Lake, updates chan<- string, defs ...*index.Definition) error {
 	prog := progress(updates)
-	return Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	return Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		indices, err := index.RemoveIndices(ctx, chunk.ZarDir(), defs)
 		if err != nil {
 			return err
diff --git a/ppl/archive/multisource.go b/ppl/lake/multisource.go
similarity index 83%
rename from ppl/archive/multisource.go
rename to ppl/lake/multisource.go
index c201b99d4..4a4c601cb 100644
--- a/ppl/archive/multisource.go
+++ b/ppl/lake/multisource.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -13,7 +13,7 @@ import (
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio"
 	"github.com/brimsec/zq/zio/detector"
@@ -46,7 +46,7 @@ type pullerCloser struct {
 	io.Closer
 }
 
-func newSpanScanner(ctx context.Context, ark *Archive, zctx *resolver.Context, sf driver.SourceFilter, si SpanInfo) (sc *pullerCloser, stats ChunkStats, err error) {
+func newSpanScanner(ctx context.Context, lk *Lake, zctx *resolver.Context, sf driver.SourceFilter, si SpanInfo) (sc *pullerCloser, stats ChunkStats, err error) {
 	closers := make(multiCloser, 0, len(si.Chunks))
 	pullers := make([]zbuf.Puller, 0, len(si.Chunks))
 	scanners := make([]zbuf.Scanner, 0, len(si.Chunks))
@@ -69,40 +69,40 @@ func newSpanScanner(ctx context.Context, ark *Archive, zctx *resolver.Context, s
 		pullers = append(pullers, scanner)
 	}
 	return &pullerCloser{
-		Puller:     zbuf.MergeByTs(ctx, pullers, ark.DataOrder),
+		Puller:     zbuf.MergeByTs(ctx, pullers, lk.DataOrder),
 		MultiStats: scanners,
 		Closer:     closers,
 	}, stats, nil
 }
 
-// NewMultiSource returns a driver.MultiSource for an Archive. If no alternative
+// NewMultiSource returns a driver.MultiSource for an Lake. If no alternative
 // paths are specified, the MultiSource will send a source for each span in the
 // driver.SourceFilter span, and report the same ordering as the archive.
 //
 // Otherwise, the sources come from localizing the given alternative paths to
 // each chunk in the archive, recognizing "_" as the chunk file itself, with no
 // defined ordering.
-func NewMultiSource(ark *Archive, altPaths []string) MultiSource {
+func NewMultiSource(lk *Lake, altPaths []string) MultiSource {
 	if len(altPaths) == 1 && altPaths[0] == "_" {
 		altPaths = nil
 	}
 	if len(altPaths) != 0 {
 		return &chunkMultiSource{
-			ark:      ark,
+			lk:       lk,
 			altPaths: altPaths,
 			stats:    &ChunkStats{},
 		}
 	}
-	return &spanMultiSource{ark, &ChunkStats{}}
+	return &spanMultiSource{lk, &ChunkStats{}}
 }
 
 type spanMultiSource struct {
-	ark   *Archive
+	lk    *Lake
 	stats *ChunkStats
 }
 
 func (m *spanMultiSource) OrderInfo() (field.Static, bool) {
-	return field.New("ts"), m.ark.DataOrder == zbuf.OrderDesc
+	return field.New("ts"), m.lk.DataOrder == zbuf.OrderDesc
 }
 
 func (m *spanMultiSource) SendSources(ctx context.Context, span nano.Span, srcChan chan driver.Source) error {
@@ -112,8 +112,8 @@ func (m *spanMultiSource) SendSources(ctx context.Context, span nano.Span, srcCh
 	sinfosChan := make(chan []SpanInfo, tsDirPreFetch)
 	g, ctx := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		err := tsDirVisit(ctx, m.ark, span, func(_ tsDir, chunks []chunk.Chunk) error {
-			sinfos := mergeChunksToSpans(chunks, m.ark.DataOrder, span)
+		err := tsDirVisit(ctx, m.lk, span, func(_ tsDir, chunks []chunk.Chunk) error {
+			sinfos := mergeChunksToSpans(chunks, m.lk.DataOrder, span)
 			select {
 			case sinfosChan <- sinfos:
 				return nil
@@ -128,7 +128,7 @@ func (m *spanMultiSource) SendSources(ctx context.Context, span nano.Span, srcCh
 		for sinfos := range sinfosChan {
 			for _, si := range sinfos {
 				select {
-				case srcChan <- &spanSource{ark: m.ark, spanInfo: si, stats: m.stats}:
+				case srcChan <- &spanSource{lk: m.lk, spanInfo: si, stats: m.stats}:
 				case <-ctx.Done():
 					return ctx.Err()
 				}
@@ -151,22 +151,22 @@ func (m *spanMultiSource) SourceFromRequest(ctx context.Context, req *api.Worker
 		if !ok {
 			return nil, zqe.E(zqe.Invalid, "invalid chunk path: %v", p)
 		}
-		uri := tsdir.path(m.ark)
+		uri := tsdir.path(m.lk)
 		mdPath := chunk.MetadataPath(uri, id)
 
-		b, err := m.ark.immfiles.ReadFile(ctx, mdPath)
+		b, err := m.lk.immfiles.ReadFile(ctx, mdPath)
 		if err != nil {
 			return nil, err
 		}
 
-		md, err := chunk.UnmarshalMetadata(b, m.ark.DataOrder)
+		md, err := chunk.UnmarshalMetadata(b, m.lk.DataOrder)
 		if err != nil {
 			return nil, zqe.E("failed to read chunk metadata from %s: %w", mdPath.String(), err)
 		}
 		si.Chunks = append(si.Chunks, md.Chunk(uri, id))
 	}
 	return &spanSource{
-		ark:      m.ark,
+		lk:       m.lk,
 		spanInfo: si,
 		stats:    m.stats,
 	}, nil
@@ -177,31 +177,31 @@ func (m *spanMultiSource) Stats() ChunkStats {
 }
 
 type spanSource struct {
-	ark      *Archive
+	lk       *Lake
 	spanInfo SpanInfo
 	stats    *ChunkStats
 }
 
 func (s *spanSource) Open(ctx context.Context, zctx *resolver.Context, sf driver.SourceFilter) (driver.ScannerCloser, error) {
-	scn, stats, err := newSpanScanner(ctx, s.ark, zctx, sf, s.spanInfo)
+	scn, stats, err := newSpanScanner(ctx, s.lk, zctx, sf, s.spanInfo)
 	s.stats.Accumulate(stats)
 	return scn, err
 }
 
 func (s *spanSource) ToRequest(req *api.WorkerChunkRequest) error {
 	req.Span = s.spanInfo.Span
-	req.DataPath = s.ark.DataPath.String()
+	req.DataPath = s.lk.DataPath.String()
 	for _, c := range s.spanInfo.Chunks {
-		req.ChunkPaths = append(req.ChunkPaths, s.ark.Root.RelPath(c.Path()))
+		req.ChunkPaths = append(req.ChunkPaths, s.lk.Root.RelPath(c.Path()))
 	}
 	return nil
 }
 
-// A chunkMultiSource uses the archive.Walk call to provide a driver.Source
+// A chunkMultiSource uses the lake.Walk call to provide a driver.Source
 // for each chunk in the archive, possibly combining its data with files named
 // by altPaths located in the chunk's zar directory.
 type chunkMultiSource struct {
-	ark      *Archive
+	lk       *Lake
 	altPaths []string
 	stats    *ChunkStats
 }
@@ -211,7 +211,7 @@ func (cms *chunkMultiSource) OrderInfo() (field.Static, bool) {
 }
 
 func (cms *chunkMultiSource) SendSources(ctx context.Context, span nano.Span, srcChan chan driver.Source) error {
-	return Walk(ctx, cms.ark, func(chunk chunk.Chunk) error {
+	return Walk(ctx, cms.lk, func(chunk chunk.Chunk) error {
 		select {
 		case srcChan <- &chunkSource{cms: cms, chunk: chunk, stats: cms.stats}:
 			return nil
diff --git a/ppl/archive/overlap.go b/ppl/lake/overlap.go
similarity index 89%
rename from ppl/archive/overlap.go
rename to ppl/lake/overlap.go
index b0dafece9..de30e4ed6 100644
--- a/ppl/archive/overlap.go
+++ b/ppl/lake/overlap.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -6,8 +6,8 @@ import (
 
 	"github.com/brimsec/zq/driver"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/zngio"
 	"github.com/brimsec/zq/zng"
@@ -251,7 +251,7 @@ func removeMaskedChunks(spans []SpanInfo, trackMasked bool) []chunk.Chunk {
 }
 
 type compactWriter struct {
-	ark     *Archive
+	lk      *Lake
 	ctx     context.Context
 	created []chunk.Chunk
 	defs    index.Definitions
@@ -263,14 +263,14 @@ type compactWriter struct {
 func (cw *compactWriter) Write(rec *zng.Record) error {
 	if cw.w != nil {
 		pos, firstTs, lastTs := cw.w.Position()
-		if pos > cw.ark.LogSizeThreshold && lastTs != rec.Ts() {
+		if pos > cw.lk.LogSizeThreshold && lastTs != rec.Ts() {
 			// If we need to create a new chunk writer, we must ensure that the
 			// span for the current chunk leaves no gap between its last timestamp
 			// and the first timestamp of our next chunk, to ensure these chunks
 			// are seen as covering the entire span of the source chunks. Hence
 			// the use of 'chunkLastTs', and the lastTs check above to ensure we are
 			// not in a run of records with the same timestamp.
-			chunkLastTs := prevTs(rec.Ts(), cw.ark.DataOrder)
+			chunkLastTs := prevTs(rec.Ts(), cw.lk.DataOrder)
 			if err := cw.w.CloseWithTs(cw.ctx, firstTs, chunkLastTs); err != nil {
 				return err
 			}
@@ -280,8 +280,8 @@ func (cw *compactWriter) Write(rec *zng.Record) error {
 	}
 	if cw.w == nil {
 		var err error
-		cw.w, err = chunk.NewWriter(cw.ctx, cw.tsd.path(cw.ark), chunk.WriterOpts{
-			Order:       cw.ark.DataOrder,
+		cw.w, err = chunk.NewWriter(cw.ctx, cw.tsd.path(cw.lk), chunk.WriterOpts{
+			Order:       cw.lk.DataOrder,
 			Masks:       cw.masks,
 			Definitions: cw.defs,
 			Zng: zngio.WriterOpts{
@@ -321,11 +321,11 @@ func (cw *compactWriter) close(lastTs nano.Ts) error {
 	return nil
 }
 
-func compactOverlaps(ctx context.Context, ark *Archive, s SpanInfo) ([]chunk.Chunk, error) {
+func compactOverlaps(ctx context.Context, lk *Lake, s SpanInfo) ([]chunk.Chunk, error) {
 	if len(s.Chunks) == 1 {
 		return nil, nil
 	}
-	ss, _, err := newSpanScanner(ctx, ark, resolver.NewContext(), driver.SourceFilter{Span: nano.MaxSpan}, s)
+	ss, _, err := newSpanScanner(ctx, lk, resolver.NewContext(), driver.SourceFilter{Span: nano.MaxSpan}, s)
 	if err != nil {
 		return nil, err
 	}
@@ -334,18 +334,18 @@ func compactOverlaps(ctx context.Context, ark *Archive, s SpanInfo) ([]chunk.Chu
 	for _, c := range s.Chunks {
 		masks = append(masks, c.Id)
 	}
-	defs, err := ark.ReadDefinitions(ctx)
+	defs, err := lk.ReadDefinitions(ctx)
 	if err != nil {
 		return nil, err
 	}
 	mw := &compactWriter{
-		ark:   ark,
+		lk:    lk,
 		ctx:   ctx,
 		defs:  defs,
 		masks: masks,
 		tsd:   newTsDir(s.Span.Ts),
 	}
-	_, slast := spanToFirstLast(ark.DataOrder, s.Span)
+	_, slast := spanToFirstLast(lk.DataOrder, s.Span)
 	if err := zbuf.CopyWithContext(ctx, mw, zbuf.PullerReader(ss)); err != nil {
 		mw.abort()
 		return nil, err
@@ -379,17 +379,17 @@ outer:
 	return append(res, mergeSpanInfos(run, order))
 }
 
-func Compact(ctx context.Context, ark *Archive, logger *zap.Logger) error {
+func Compact(ctx context.Context, lk *Lake, logger *zap.Logger) error {
 	if logger == nil {
 		logger = zap.NewNop()
 	}
 
-	return tsDirVisit(ctx, ark, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
-		spans := alignChunksToSpans(chunks, ark.DataOrder, nano.MaxSpan)
+	return tsDirVisit(ctx, lk, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
+		spans := alignChunksToSpans(chunks, lk.DataOrder, nano.MaxSpan)
 		removeMaskedChunks(spans, false)
-		spans = mergeCommonChunkSpans(spans, ark.DataOrder)
+		spans = mergeCommonChunkSpans(spans, lk.DataOrder)
 		for _, s := range spans {
-			newchunks, err := compactOverlaps(ctx, ark, s)
+			newchunks, err := compactOverlaps(ctx, lk, s)
 			if err != nil {
 				return err
 			}
@@ -409,13 +409,13 @@ func Compact(ctx context.Context, ark *Archive, logger *zap.Logger) error {
 	})
 }
 
-func Purge(ctx context.Context, ark *Archive, logger *zap.Logger) error {
+func Purge(ctx context.Context, lk *Lake, logger *zap.Logger) error {
 	if logger == nil {
 		logger = zap.NewNop()
 	}
 
-	return tsDirVisit(ctx, ark, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
-		spans := alignChunksToSpans(chunks, ark.DataOrder, nano.MaxSpan)
+	return tsDirVisit(ctx, lk, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
+		spans := alignChunksToSpans(chunks, lk.DataOrder, nano.MaxSpan)
 		maskedChunks := removeMaskedChunks(spans, true)
 		if len(maskedChunks) == 0 {
 			return nil
diff --git a/ppl/archive/schema.go b/ppl/lake/schema.go
similarity index 68%
rename from ppl/archive/schema.go
rename to ppl/lake/schema.go
index e8e6c7879..82a55854c 100644
--- a/ppl/archive/schema.go
+++ b/ppl/lake/schema.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -8,8 +8,8 @@ import (
 	"time"
 
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive/immcache"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/immcache"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zqe"
 	"github.com/segmentio/ksuid"
@@ -89,7 +89,7 @@ func (c *CreateOptions) toMetadata() *Metadata {
 	return m
 }
 
-type Archive struct {
+type Lake struct {
 	Root             iosrc.URI
 	DataPath         iosrc.URI
 	DataOrder        zbuf.Order
@@ -101,25 +101,25 @@ type Archive struct {
 	}
 }
 
-func (ark *Archive) metaWrite() error {
+func (lk *Lake) metaWrite() error {
 	m := &Metadata{
 		Version:          0,
-		LogSizeThreshold: ark.LogSizeThreshold,
-		DataOrder:        ark.DataOrder,
-		DataPath:         ark.DataPath.String(),
+		LogSizeThreshold: lk.LogSizeThreshold,
+		DataOrder:        lk.DataOrder,
+		DataPath:         lk.DataPath.String(),
 	}
-	return m.Write(ark.mdURI())
+	return m.Write(lk.mdURI())
 }
 
-func (ark *Archive) mdURI() iosrc.URI {
-	return ark.Root.AppendPath(metadataFilename)
+func (lk *Lake) mdURI() iosrc.URI {
+	return lk.Root.AppendPath(metadataFilename)
 }
 
-func (ark *Archive) filterAllowed(id ksuid.KSUID) bool {
-	if len(ark.LogFilter) == 0 {
+func (lk *Lake) filterAllowed(id ksuid.KSUID) bool {
+	if len(lk.LogFilter) == 0 {
 		return true
 	}
-	for _, fid := range ark.LogFilter {
+	for _, fid := range lk.LogFilter {
 		if fid == id {
 			return true
 		}
@@ -127,14 +127,14 @@ func (ark *Archive) filterAllowed(id ksuid.KSUID) bool {
 	return false
 }
 
-func (ark *Archive) DefinitionsDir() iosrc.URI {
-	return ark.Root.AppendPath(indexdefsDir)
+func (lk *Lake) DefinitionsDir() iosrc.URI {
+	return lk.Root.AppendPath(indexdefsDir)
 }
 
-func (ark *Archive) ReadDefinitions(ctx context.Context) (index.Definitions, error) {
-	defs, err := index.ReadDefinitions(ctx, ark.DefinitionsDir())
+func (lk *Lake) ReadDefinitions(ctx context.Context) (index.Definitions, error) {
+	defs, err := index.ReadDefinitions(ctx, lk.DefinitionsDir())
 	if zqe.IsNotFound(err) {
-		err = iosrc.MkdirAll(ark.DefinitionsDir(), 0700)
+		err = iosrc.MkdirAll(lk.DefinitionsDir(), 0700)
 	}
 	return defs, err
 }
@@ -143,19 +143,19 @@ type OpenOptions struct {
 	ImmutableCache immcache.ImmutableCache
 }
 
-func OpenArchive(rpath string, oo *OpenOptions) (*Archive, error) {
-	return OpenArchiveWithContext(context.Background(), rpath, oo)
+func OpenLake(rpath string, oo *OpenOptions) (*Lake, error) {
+	return OpenLakeWithContext(context.Background(), rpath, oo)
 }
 
-func OpenArchiveWithContext(ctx context.Context, rpath string, oo *OpenOptions) (*Archive, error) {
+func OpenLakeWithContext(ctx context.Context, rpath string, oo *OpenOptions) (*Lake, error) {
 	root, err := iosrc.ParseURI(rpath)
 	if err != nil {
 		return nil, err
 	}
-	return openArchive(ctx, root, oo)
+	return openLake(ctx, root, oo)
 }
 
-func openArchive(ctx context.Context, root iosrc.URI, oo *OpenOptions) (*Archive, error) {
+func openLake(ctx context.Context, root iosrc.URI, oo *OpenOptions) (*Lake, error) {
 	m, err := MetadataRead(ctx, root.AppendPath(metadataFilename))
 	if err != nil {
 		return nil, err
@@ -168,7 +168,7 @@ func openArchive(ctx context.Context, root iosrc.URI, oo *OpenOptions) (*Archive
 		return nil, err
 	}
 
-	ark := &Archive{
+	lk := &Lake{
 		DataOrder:        m.DataOrder,
 		DataPath:         dpuri,
 		LogSizeThreshold: m.LogSizeThreshold,
@@ -177,17 +177,17 @@ func openArchive(ctx context.Context, root iosrc.URI, oo *OpenOptions) (*Archive
 	}
 
 	if oo != nil && oo.ImmutableCache != nil {
-		ark.immfiles = oo.ImmutableCache
+		lk.immfiles = oo.ImmutableCache
 	}
 
-	return ark, nil
+	return lk, nil
 }
 
-func CreateOrOpenArchive(rpath string, co *CreateOptions, oo *OpenOptions) (*Archive, error) {
-	return CreateOrOpenArchiveWithContext(context.Background(), rpath, co, oo)
+func CreateOrOpenLake(rpath string, co *CreateOptions, oo *OpenOptions) (*Lake, error) {
+	return CreateOrOpenLakeWithContext(context.Background(), rpath, co, oo)
 }
 
-func CreateOrOpenArchiveWithContext(ctx context.Context, rpath string, co *CreateOptions, oo *OpenOptions) (*Archive, error) {
+func CreateOrOpenLakeWithContext(ctx context.Context, rpath string, co *CreateOptions, oo *OpenOptions) (*Lake, error) {
 	root, err := iosrc.ParseURI(rpath)
 	if err != nil {
 		return nil, err
@@ -210,5 +210,5 @@ func CreateOrOpenArchiveWithContext(ctx context.Context, rpath string, co *Creat
 		}
 	}
 
-	return openArchive(ctx, root, oo)
+	return openLake(ctx, root, oo)
 }
diff --git a/ppl/archive/seekindex/builder.go b/ppl/lake/seekindex/builder.go
similarity index 100%
rename from ppl/archive/seekindex/builder.go
rename to ppl/lake/seekindex/builder.go
diff --git a/ppl/archive/seekindex/range.go b/ppl/lake/seekindex/range.go
similarity index 100%
rename from ppl/archive/seekindex/range.go
rename to ppl/lake/seekindex/range.go
diff --git a/ppl/archive/seekindex/seekindex.go b/ppl/lake/seekindex/seekindex.go
similarity index 100%
rename from ppl/archive/seekindex/seekindex.go
rename to ppl/lake/seekindex/seekindex.go
diff --git a/ppl/archive/seekindex/seekindex_test.go b/ppl/lake/seekindex/seekindex_test.go
similarity index 100%
rename from ppl/archive/seekindex/seekindex_test.go
rename to ppl/lake/seekindex/seekindex_test.go
diff --git a/ppl/archive/stat.go b/ppl/lake/stat.go
similarity index 85%
rename from ppl/archive/stat.go
rename to ppl/lake/stat.go
index ff671b057..f0b01fae7 100644
--- a/ppl/archive/stat.go
+++ b/ppl/lake/stat.go
@@ -1,12 +1,12 @@
-package archive
+package lake
 
 import (
 	"context"
 
 	"github.com/brimsec/zq/microindex"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng"
 	"github.com/brimsec/zq/zng/resolver"
@@ -15,7 +15,7 @@ import (
 
 // statReadCloser implements zbuf.ReadCloser.
 type statReadCloser struct {
-	ark      *Archive
+	lk       *Lake
 	ctx      context.Context
 	cancel   context.CancelFunc
 	defnames map[ksuid.KSUID]string
@@ -53,7 +53,7 @@ type chunkStat struct {
 func (s *statReadCloser) chunkRecord(chunk chunk.Chunk) error {
 	stat := chunkStat{
 		Type:        "chunk",
-		LogID:       s.ark.Root.RelPath(chunk.Path()),
+		LogID:       s.lk.Root.RelPath(chunk.Path()),
 		First:       chunk.First,
 		Last:        chunk.Last,
 		Size:        uint64(chunk.Size),
@@ -104,7 +104,7 @@ func (s *statReadCloser) indexRecords(chunk chunk.Chunk) error {
 		}
 		stat := indexStat{
 			Type:       "index",
-			LogID:      s.ark.Root.RelPath(chunk.Path()),
+			LogID:      s.lk.Root.RelPath(chunk.Path()),
 			First:      chunk.First,
 			Last:       chunk.Last,
 			Definition: defDesc{ID: id.String(), Description: defname},
@@ -134,7 +134,7 @@ func (s *statReadCloser) send(rec *zng.Record) error {
 func (s *statReadCloser) run() {
 	defer close(s.recs)
 
-	s.err = Walk(s.ctx, s.ark, func(chunk chunk.Chunk) error {
+	s.err = Walk(s.ctx, s.lk, func(chunk chunk.Chunk) error {
 		if err := s.chunkRecord(chunk); err != nil {
 			return err
 		}
@@ -145,9 +145,9 @@ func (s *statReadCloser) run() {
 	})
 }
 
-func RecordCount(ctx context.Context, ark *Archive) (uint64, error) {
+func RecordCount(ctx context.Context, lk *Lake) (uint64, error) {
 	var count uint64
-	err := Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	err := Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		count += chunk.RecordCount
 		return nil
 	})
@@ -160,13 +160,13 @@ type IndexInfo struct {
 	ChunkCount   uint64
 }
 
-func IndexStat(ctx context.Context, ark *Archive, defs []*index.Definition) ([]IndexInfo, error) {
+func IndexStat(ctx context.Context, lk *Lake, defs []*index.Definition) ([]IndexInfo, error) {
 	m := make(map[ksuid.KSUID]IndexInfo)
 	for _, def := range defs {
 		m[def.ID] = IndexInfo{DefinitionID: def.ID}
 	}
 	var chunkCount uint64
-	err := Walk(ctx, ark, func(chunk chunk.Chunk) error {
+	err := Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		chunkCount++
 		ids, err := index.ListDefinitionIDs(ctx, chunk.ZarDir())
 		if err != nil {
@@ -191,8 +191,8 @@ func IndexStat(ctx context.Context, ark *Archive, defs []*index.Definition) ([]I
 	return stats, nil
 }
 
-func Stat(ctx context.Context, zctx *resolver.Context, ark *Archive) (zbuf.ReadCloser, error) {
-	defs, err := ark.ReadDefinitions(ctx)
+func Stat(ctx context.Context, zctx *resolver.Context, lk *Lake) (zbuf.ReadCloser, error) {
+	defs, err := lk.ReadDefinitions(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -205,7 +205,7 @@ func Stat(ctx context.Context, zctx *resolver.Context, ark *Archive) (zbuf.ReadC
 	mzctx := resolver.NewMarshaler()
 	mzctx.Context = zctx
 	s := &statReadCloser{
-		ark:      ark,
+		lk:       lk,
 		ctx:      ctx,
 		cancel:   cancel,
 		defnames: defnames,
diff --git a/ppl/archive/staticsource.go b/ppl/lake/staticsource.go
similarity index 78%
rename from ppl/archive/staticsource.go
rename to ppl/lake/staticsource.go
index 711d0bc20..8a52c055e 100644
--- a/ppl/archive/staticsource.go
+++ b/ppl/lake/staticsource.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -17,15 +17,15 @@ type staticSource struct {
 	src driver.Source
 }
 
-func NewStaticSource(ark *Archive, src driver.Source) driver.MultiSource {
+func NewStaticSource(lk *Lake, src driver.Source) driver.MultiSource {
 	return &staticSource{
-		spanMultiSource: &spanMultiSource{ark: ark},
+		spanMultiSource: &spanMultiSource{lk: lk},
 		src:             src,
 	}
 }
 
 func (s *staticSource) OrderInfo() (field.Static, bool) {
-	return field.New("ts"), s.ark.DataOrder == zbuf.OrderDesc
+	return field.New("ts"), s.lk.DataOrder == zbuf.OrderDesc
 }
 
 func (s *staticSource) SendSources(ctx context.Context, span nano.Span, srcChan chan driver.Source) error {
diff --git a/ppl/archive/testdata/babble-mergelargestchunk1.tzng b/ppl/lake/testdata/babble-mergelargestchunk1.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-mergelargestchunk1.tzng
rename to ppl/lake/testdata/babble-mergelargestchunk1.tzng
diff --git a/ppl/archive/testdata/babble-mergelargestchunk2.tzng b/ppl/lake/testdata/babble-mergelargestchunk2.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-mergelargestchunk2.tzng
rename to ppl/lake/testdata/babble-mergelargestchunk2.tzng
diff --git a/ppl/archive/testdata/babble-norm-shuf-1.tzng b/ppl/lake/testdata/babble-norm-shuf-1.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm-shuf-1.tzng
rename to ppl/lake/testdata/babble-norm-shuf-1.tzng
diff --git a/ppl/archive/testdata/babble-norm-shuf-2.tzng b/ppl/lake/testdata/babble-norm-shuf-2.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm-shuf-2.tzng
rename to ppl/lake/testdata/babble-norm-shuf-2.tzng
diff --git a/ppl/archive/testdata/babble-norm-shuf-3.tzng b/ppl/lake/testdata/babble-norm-shuf-3.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm-shuf-3.tzng
rename to ppl/lake/testdata/babble-norm-shuf-3.tzng
diff --git a/ppl/archive/testdata/babble-norm-shuf-4.tzng b/ppl/lake/testdata/babble-norm-shuf-4.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm-shuf-4.tzng
rename to ppl/lake/testdata/babble-norm-shuf-4.tzng
diff --git a/ppl/archive/testdata/babble-norm-shuf-5.tzng b/ppl/lake/testdata/babble-norm-shuf-5.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm-shuf-5.tzng
rename to ppl/lake/testdata/babble-norm-shuf-5.tzng
diff --git a/ppl/archive/testdata/babble-norm.tzng b/ppl/lake/testdata/babble-norm.tzng
similarity index 100%
rename from ppl/archive/testdata/babble-norm.tzng
rename to ppl/lake/testdata/babble-norm.tzng
diff --git a/ppl/archive/testdata/group-by-ts.zng b/ppl/lake/testdata/group-by-ts.zng
similarity index 100%
rename from ppl/archive/testdata/group-by-ts.zng
rename to ppl/lake/testdata/group-by-ts.zng
diff --git a/ppl/archive/walk.go b/ppl/lake/walk.go
similarity index 78%
rename from ppl/archive/walk.go
rename to ppl/lake/walk.go
index 0b474ee65..8b50f57ff 100644
--- a/ppl/archive/walk.go
+++ b/ppl/lake/walk.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -8,7 +8,7 @@ import (
 
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zqe"
 	"github.com/segmentio/ksuid"
@@ -38,8 +38,8 @@ func parseTsDirName(name string) (tsDir, bool) {
 	return newTsDir(nano.TimeToTs(t)), true
 }
 
-func (t tsDir) path(ark *Archive) iosrc.URI {
-	return ark.DataPath.AppendPath(dataDirname, t.name())
+func (t tsDir) path(lk *Lake) iosrc.URI {
+	return lk.DataPath.AppendPath(dataDirname, t.name())
 }
 
 func (t tsDir) name() string {
@@ -51,8 +51,8 @@ type tsDirVisitor func(tsd tsDir, unsortedChunks []chunk.Chunk) error
 // tsDirVisit calls visitor for each tsDir whose span overlaps with the
 // given span. tsDirs are visited in the archive's order, but the
 // chunks passed to visitor are not sorted.
-func tsDirVisit(ctx context.Context, ark *Archive, filterSpan nano.Span, visitor tsDirVisitor) error {
-	zdDir := ark.DataPath.AppendPath(dataDirname)
+func tsDirVisit(ctx context.Context, lk *Lake, filterSpan nano.Span, visitor tsDirVisitor) error {
+	zdDir := lk.DataPath.AppendPath(dataDirname)
 	dirents, err := iosrc.ReadDir(ctx, zdDir)
 	if err != nil {
 		return err
@@ -69,7 +69,7 @@ func tsDirVisit(ctx context.Context, ark *Archive, filterSpan nano.Span, visitor
 		tsdirs = append(tsdirs, tsd)
 	}
 	sort.Slice(tsdirs, func(i, j int) bool {
-		if ark.DataOrder == zbuf.OrderAsc {
+		if lk.DataOrder == zbuf.OrderAsc {
 			return tsdirs[i].Ts < tsdirs[j].Ts
 		}
 		return tsdirs[j].Ts < tsdirs[i].Ts
@@ -79,7 +79,7 @@ func tsDirVisit(ctx context.Context, ark *Archive, filterSpan nano.Span, visitor
 		if err != nil {
 			return err
 		}
-		chunks, err := tsDirEntriesToChunks(ctx, ark, filterSpan, d, dirents)
+		chunks, err := tsDirEntriesToChunks(ctx, lk, filterSpan, d, dirents)
 		if err != nil {
 			return err
 		}
@@ -90,7 +90,7 @@ func tsDirVisit(ctx context.Context, ark *Archive, filterSpan nano.Span, visitor
 	return nil
 }
 
-func tsDirEntriesToChunks(ctx context.Context, ark *Archive, filterSpan nano.Span, tsDir tsDir, entries []iosrc.Info) ([]chunk.Chunk, error) {
+func tsDirEntriesToChunks(ctx context.Context, lk *Lake, filterSpan nano.Span, tsDir tsDir, entries []iosrc.Info) ([]chunk.Chunk, error) {
 	type seen struct {
 		data bool
 		meta bool
@@ -98,7 +98,7 @@ func tsDirEntriesToChunks(ctx context.Context, ark *Archive, filterSpan nano.Spa
 	m := make(map[ksuid.KSUID]seen)
 	for _, e := range entries {
 		if kind, id, ok := chunk.FileMatch(e.Name()); ok {
-			if !ark.filterAllowed(id) {
+			if !lk.filterAllowed(id) {
 				continue
 			}
 			s := m[id]
@@ -116,14 +116,14 @@ func tsDirEntriesToChunks(ctx context.Context, ark *Archive, filterSpan nano.Spa
 		if !seen.meta || !seen.data {
 			continue
 		}
-		dir := tsDir.path(ark)
+		dir := tsDir.path(lk)
 		mdPath := chunk.MetadataPath(dir, id)
-		b, err := ark.immfiles.ReadFile(ctx, mdPath)
+		b, err := lk.immfiles.ReadFile(ctx, mdPath)
 		if err != nil {
 			return nil, err
 		}
 
-		md, err := chunk.UnmarshalMetadata(b, ark.DataOrder)
+		md, err := chunk.UnmarshalMetadata(b, lk.DataOrder)
 		if err != nil {
 			if zqe.IsNotFound(err) {
 				continue
@@ -142,9 +142,9 @@ func tsDirEntriesToChunks(ctx context.Context, ark *Archive, filterSpan nano.Spa
 type Visitor func(chunk chunk.Chunk) error
 
 // Walk calls visitor for every data chunk in the archive.
-func Walk(ctx context.Context, ark *Archive, v Visitor) error {
-	return tsDirVisit(ctx, ark, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
-		chunk.Sort(ark.DataOrder, chunks)
+func Walk(ctx context.Context, lk *Lake, v Visitor) error {
+	return tsDirVisit(ctx, lk, nano.MaxSpan, func(_ tsDir, chunks []chunk.Chunk) error {
+		chunk.Sort(lk.DataOrder, chunks)
 		for _, c := range chunks {
 			if err := iosrc.MkdirAll(c.ZarDir(), 0700); err != nil {
 				return err
@@ -159,8 +159,8 @@ func Walk(ctx context.Context, ark *Archive, v Visitor) error {
 
 // RmDirs descends a directory hierarchy looking for zar dirs and remove
 // each such directory and all of its contents.
-func RmDirs(ctx context.Context, ark *Archive) error {
-	return Walk(ctx, ark, func(chunk chunk.Chunk) error {
+func RmDirs(ctx context.Context, lk *Lake) error {
+	return Walk(ctx, lk, func(chunk chunk.Chunk) error {
 		return iosrc.RemoveAll(ctx, chunk.ZarDir())
 	})
 }
@@ -221,9 +221,9 @@ func (si *SpanInfo) RemoveMasked() []chunk.Chunk {
 }
 
 // SpanWalk calls visitor with each SpanInfo within the filter span.
-func SpanWalk(ctx context.Context, ark *Archive, filter nano.Span, visitor func(si SpanInfo) error) error {
-	return tsDirVisit(ctx, ark, filter, func(_ tsDir, chunks []chunk.Chunk) error {
-		sinfos := mergeChunksToSpans(chunks, ark.DataOrder, filter)
+func SpanWalk(ctx context.Context, lk *Lake, filter nano.Span, visitor func(si SpanInfo) error) error {
+	return tsDirVisit(ctx, lk, filter, func(_ tsDir, chunks []chunk.Chunk) error {
+		sinfos := mergeChunksToSpans(chunks, lk.DataOrder, filter)
 		for _, s := range sinfos {
 			for _, c := range s.Chunks {
 				if err := iosrc.MkdirAll(c.ZarDir(), 0700); err != nil {
diff --git a/ppl/archive/ztests/basic.yaml b/ppl/lake/ztests/basic.yaml
similarity index 100%
rename from ppl/archive/ztests/basic.yaml
rename to ppl/lake/ztests/basic.yaml
diff --git a/ppl/archive/ztests/compact.yaml b/ppl/lake/ztests/compact.yaml
similarity index 100%
rename from ppl/archive/ztests/compact.yaml
rename to ppl/lake/ztests/compact.yaml
diff --git a/ppl/archive/ztests/consecutive-ts.yaml b/ppl/lake/ztests/consecutive-ts.yaml
similarity index 100%
rename from ppl/archive/ztests/consecutive-ts.yaml
rename to ppl/lake/ztests/consecutive-ts.yaml
diff --git a/ppl/archive/ztests/custom-index-multitype-error.yaml b/ppl/lake/ztests/custom-index-multitype-error.yaml
similarity index 100%
rename from ppl/archive/ztests/custom-index-multitype-error.yaml
rename to ppl/lake/ztests/custom-index-multitype-error.yaml
diff --git a/ppl/archive/ztests/custom-index.yaml b/ppl/lake/ztests/custom-index.yaml
similarity index 100%
rename from ppl/archive/ztests/custom-index.yaml
rename to ppl/lake/ztests/custom-index.yaml
diff --git a/ppl/archive/ztests/deterministic-merge.yaml b/ppl/lake/ztests/deterministic-merge.yaml
similarity index 100%
rename from ppl/archive/ztests/deterministic-merge.yaml
rename to ppl/lake/ztests/deterministic-merge.yaml
diff --git a/ppl/archive/ztests/dirs.yaml b/ppl/lake/ztests/dirs.yaml
similarity index 100%
rename from ppl/archive/ztests/dirs.yaml
rename to ppl/lake/ztests/dirs.yaml
diff --git a/ppl/archive/ztests/find-index-field.yaml b/ppl/lake/ztests/find-index-field.yaml
similarity index 100%
rename from ppl/archive/ztests/find-index-field.yaml
rename to ppl/lake/ztests/find-index-field.yaml
diff --git a/ppl/archive/ztests/find-index-type.yaml b/ppl/lake/ztests/find-index-type.yaml
similarity index 100%
rename from ppl/archive/ztests/find-index-type.yaml
rename to ppl/lake/ztests/find-index-type.yaml
diff --git a/ppl/archive/ztests/findzng.yaml b/ppl/lake/ztests/findzng.yaml
similarity index 100%
rename from ppl/archive/ztests/findzng.yaml
rename to ppl/lake/ztests/findzng.yaml
diff --git a/ppl/archive/ztests/group-by-ts.yaml b/ppl/lake/ztests/group-by-ts.yaml
similarity index 100%
rename from ppl/archive/ztests/group-by-ts.yaml
rename to ppl/lake/ztests/group-by-ts.yaml
diff --git a/ppl/archive/ztests/import-check.yaml b/ppl/lake/ztests/import-check.yaml
similarity index 100%
rename from ppl/archive/ztests/import-check.yaml
rename to ppl/lake/ztests/import-check.yaml
diff --git a/ppl/archive/ztests/index-custom-input.yaml b/ppl/lake/ztests/index-custom-input.yaml
similarity index 100%
rename from ppl/archive/ztests/index-custom-input.yaml
rename to ppl/lake/ztests/index-custom-input.yaml
diff --git a/ppl/archive/ztests/index-drop.yaml b/ppl/lake/ztests/index-drop.yaml
similarity index 100%
rename from ppl/archive/ztests/index-drop.yaml
rename to ppl/lake/ztests/index-drop.yaml
diff --git a/ppl/archive/ztests/index-missing.yaml b/ppl/lake/ztests/index-missing.yaml
similarity index 100%
rename from ppl/archive/ztests/index-missing.yaml
rename to ppl/lake/ztests/index-missing.yaml
diff --git a/ppl/archive/ztests/index-new-data.yaml b/ppl/lake/ztests/index-new-data.yaml
similarity index 100%
rename from ppl/archive/ztests/index-new-data.yaml
rename to ppl/lake/ztests/index-new-data.yaml
diff --git a/ppl/archive/ztests/index-on-compact.yaml b/ppl/lake/ztests/index-on-compact.yaml
similarity index 100%
rename from ppl/archive/ztests/index-on-compact.yaml
rename to ppl/lake/ztests/index-on-compact.yaml
diff --git a/ppl/archive/ztests/ls.yaml b/ppl/lake/ztests/ls.yaml
similarity index 100%
rename from ppl/archive/ztests/ls.yaml
rename to ppl/lake/ztests/ls.yaml
diff --git a/ppl/archive/ztests/map.yaml b/ppl/lake/ztests/map.yaml
similarity index 100%
rename from ppl/archive/ztests/map.yaml
rename to ppl/lake/ztests/map.yaml
diff --git a/ppl/archive/ztests/merge-largest-spaninfo.yaml b/ppl/lake/ztests/merge-largest-spaninfo.yaml
similarity index 100%
rename from ppl/archive/ztests/merge-largest-spaninfo.yaml
rename to ppl/lake/ztests/merge-largest-spaninfo.yaml
diff --git a/ppl/archive/ztests/multikey-index.yaml b/ppl/lake/ztests/multikey-index.yaml
similarity index 100%
rename from ppl/archive/ztests/multikey-index.yaml
rename to ppl/lake/ztests/multikey-index.yaml
diff --git a/ppl/archive/ztests/notcustom-index-multitype-error.yaml b/ppl/lake/ztests/notcustom-index-multitype-error.yaml
similarity index 100%
rename from ppl/archive/ztests/notcustom-index-multitype-error.yaml
rename to ppl/lake/ztests/notcustom-index-multitype-error.yaml
diff --git a/ppl/archive/ztests/overlap.yaml b/ppl/lake/ztests/overlap.yaml
similarity index 100%
rename from ppl/archive/ztests/overlap.yaml
rename to ppl/lake/ztests/overlap.yaml
diff --git a/ppl/archive/ztests/rm.yaml b/ppl/lake/ztests/rm.yaml
similarity index 100%
rename from ppl/archive/ztests/rm.yaml
rename to ppl/lake/ztests/rm.yaml
diff --git a/ppl/archive/ztests/s3/find-index-type.yaml b/ppl/lake/ztests/s3/find-index-type.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/find-index-type.yaml
rename to ppl/lake/ztests/s3/find-index-type.yaml
diff --git a/ppl/archive/ztests/s3/rm.yaml b/ppl/lake/ztests/s3/rm.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/rm.yaml
rename to ppl/lake/ztests/s3/rm.yaml
diff --git a/ppl/archive/ztests/s3/rmdirs.yaml b/ppl/lake/ztests/s3/rmdirs.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/rmdirs.yaml
rename to ppl/lake/ztests/s3/rmdirs.yaml
diff --git a/ppl/archive/ztests/s3/s3root.yaml b/ppl/lake/ztests/s3/s3root.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/s3root.yaml
rename to ppl/lake/ztests/s3/s3root.yaml
diff --git a/ppl/archive/ztests/s3/stat.yaml b/ppl/lake/ztests/s3/stat.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/stat.yaml
rename to ppl/lake/ztests/s3/stat.yaml
diff --git a/ppl/archive/ztests/s3/zq.yaml b/ppl/lake/ztests/s3/zq.yaml
similarity index 100%
rename from ppl/archive/ztests/s3/zq.yaml
rename to ppl/lake/ztests/s3/zq.yaml
diff --git a/ppl/archive/ztests/seek-index-overlap.yaml b/ppl/lake/ztests/seek-index-overlap.yaml
similarity index 100%
rename from ppl/archive/ztests/seek-index-overlap.yaml
rename to ppl/lake/ztests/seek-index-overlap.yaml
diff --git a/ppl/archive/ztests/seek-index.yaml b/ppl/lake/ztests/seek-index.yaml
similarity index 100%
rename from ppl/archive/ztests/seek-index.yaml
rename to ppl/lake/ztests/seek-index.yaml
diff --git a/ppl/archive/ztests/sortmem.yaml b/ppl/lake/ztests/sortmem.yaml
similarity index 100%
rename from ppl/archive/ztests/sortmem.yaml
rename to ppl/lake/ztests/sortmem.yaml
diff --git a/ppl/archive/ztests/stat.yaml b/ppl/lake/ztests/stat.yaml
similarity index 100%
rename from ppl/archive/ztests/stat.yaml
rename to ppl/lake/ztests/stat.yaml
diff --git a/ppl/archive/ztests/zq.yaml b/ppl/lake/ztests/zq.yaml
similarity index 100%
rename from ppl/archive/ztests/zq.yaml
rename to ppl/lake/ztests/zq.yaml
diff --git a/ppl/zqd/apiserver/manager.go b/ppl/zqd/apiserver/manager.go
index 68634c638..1df7d9f85 100644
--- a/ppl/zqd/apiserver/manager.go
+++ b/ppl/zqd/apiserver/manager.go
@@ -9,7 +9,7 @@ import (
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/immcache"
+	"github.com/brimsec/zq/ppl/lake/immcache"
 	"github.com/brimsec/zq/ppl/zqd/auth"
 	"github.com/brimsec/zq/ppl/zqd/db"
 	"github.com/brimsec/zq/ppl/zqd/db/schema"
diff --git a/ppl/zqd/core.go b/ppl/zqd/core.go
index 1896c1406..9252b3d52 100644
--- a/ppl/zqd/core.go
+++ b/ppl/zqd/core.go
@@ -13,12 +13,13 @@ import (
 
 	"github.com/brimsec/zq/api"
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive/immcache"
+	"github.com/brimsec/zq/ppl/lake/immcache"
 	"github.com/brimsec/zq/ppl/zqd/apiserver"
 	"github.com/brimsec/zq/ppl/zqd/db"
 	"github.com/brimsec/zq/ppl/zqd/pcapanalyzer"
 	"github.com/brimsec/zq/ppl/zqd/recruiter"
 	"github.com/brimsec/zq/ppl/zqd/worker"
+	"github.com/go-redis/redis/v8"
 	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -43,6 +44,7 @@ type Config struct {
 	ImmutableCache immcache.Config
 	Logger         *zap.Logger
 	Personality    string
+	Redis          RedisConfig
 	Root           string
 	Version        string
 	Worker         worker.WorkerConfig
@@ -54,6 +56,7 @@ type Config struct {
 type Core struct {
 	auth       *Auth0Authenticator
 	db         db.DB
+	redis      *redis.Client
 	logger     *zap.Logger
 	mgr        *apiserver.Manager
 	registry   *prometheus.Registry
@@ -158,7 +161,11 @@ func (c *Core) addAPIServerRoutes(ctx context.Context, conf Config) (err error)
 	if err != nil {
 		return err
 	}
-	icache, err := immcache.New(conf.ImmutableCache, c.registry)
+	c.redis, err = NewRedisClient(ctx, conf.Redis)
+	if err != nil {
+		return err
+	}
+	icache, err := immcache.New(conf.ImmutableCache, c.redis, c.registry)
 	if err != nil {
 		return err
 	}
diff --git a/ppl/zqd/redis.go b/ppl/zqd/redis.go
new file mode 100644
index 000000000..780ec3338
--- /dev/null
+++ b/ppl/zqd/redis.go
@@ -0,0 +1,66 @@
+package zqd
+
+import (
+	"context"
+	"flag"
+	"fmt"
+
+	"github.com/go-redis/redis/v8"
+)
+
+func NewRedisClient(ctx context.Context, conf RedisConfig) (*redis.Client, error) {
+	if !conf.Enabled {
+		return nil, nil
+	}
+	client := redis.NewClient(&conf.Options)
+	if cmd := client.Ping(ctx); cmd.Err() != nil {
+		return nil, cmd.Err()
+	}
+	return client, nil
+}
+
+type RedisConfig struct {
+	redis.Options
+
+	Enabled bool
+}
+
+func (c *RedisConfig) SetFlags(fs *flag.FlagSet) {
+	fs.Var(c, "redis.url", "redis connection url (i.e. redis://<user>:<password>@<host>:<port>/<db_number>)")
+	fs.BoolVar(&c.Enabled, "redis.enabled", false, "enable use of a redis server")
+	fs.StringVar(&c.Addr, "redis.addr", "localhost:6379", "redis address")
+	fs.StringVar(&c.Username, "redis.user", "", "redis username")
+	fs.StringVar(&c.Password, "redis.password", "", "redis password")
+	fs.IntVar(&c.DB, "redis.database", 0, "redis database number")
+}
+
+func (c *RedisConfig) Set(s string) error {
+	opt, err := redis.ParseURL(s)
+	if err == nil {
+		c.Options = *opt
+	}
+	return err
+}
+
+// redis://<user>:<password>@<host>:<port>/<db_number>
+
+func (c RedisConfig) String() string {
+	if c.IsEmpty() {
+		return ""
+	}
+	str := "redis://"
+	if c.Username != "" {
+		str += c.Username
+		if c.Password != "" {
+			str += ":" + c.Password
+		}
+		str += "@"
+	}
+	str += c.Addr
+	str += fmt.Sprintf("/%d", c.DB)
+	return str
+}
+
+func (c RedisConfig) IsEmpty() bool {
+	return c.Addr == "" && c.Username == "" && c.Password == ""
+}
diff --git a/ppl/zqd/scripts/dkc-services.yaml b/ppl/zqd/scripts/dkc-services.yaml
index d55d4c850..f671e97bd 100644
--- a/ppl/zqd/scripts/dkc-services.yaml
+++ b/ppl/zqd/scripts/dkc-services.yaml
@@ -8,3 +8,7 @@ services:
       - POSTGRES_PASSWORD=test
     ports:
       - "5432:5432"
+  redis:
+    image: "redis:6"
+    ports:
+      - "6379:6379"
diff --git a/ppl/zqd/search/indexsearch.go b/ppl/zqd/search/indexsearch.go
index fca56a47c..33d56af1a 100644
--- a/ppl/zqd/search/indexsearch.go
+++ b/ppl/zqd/search/indexsearch.go
@@ -4,7 +4,7 @@ import (
 	"context"
 
 	"github.com/brimsec/zq/api"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng/resolver"
 )
diff --git a/ppl/zqd/storage/archivestore/archivestore.go b/ppl/zqd/storage/archivestore/archivestore.go
index d3c772654..0d16512d9 100644
--- a/ppl/zqd/storage/archivestore/archivestore.go
+++ b/ppl/zqd/storage/archivestore/archivestore.go
@@ -8,10 +8,10 @@ import (
 	"github.com/brimsec/zq/driver"
 	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/immcache"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/immcache"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/ppl/zqd/storage"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zng/resolver"
@@ -20,19 +20,19 @@ import (
 )
 
 func Load(ctx context.Context, path iosrc.URI, notifier WriteNotifier, cfg *api.ArchiveConfig, immcache immcache.ImmutableCache) (*Storage, error) {
-	co := &archive.CreateOptions{}
+	co := &lake.CreateOptions{}
 	if cfg != nil && cfg.CreateOptions != nil {
 		co.LogSizeThreshold = cfg.CreateOptions.LogSizeThreshold
 	}
-	oo := &archive.OpenOptions{
+	oo := &lake.OpenOptions{
 		ImmutableCache: immcache,
 	}
-	ark, err := archive.CreateOrOpenArchiveWithContext(ctx, path.String(), co, oo)
+	lk, err := lake.CreateOrOpenLakeWithContext(ctx, path.String(), co, oo)
 	if err != nil {
 		return nil, err
 	}
 	return &Storage{
-		ark:      ark,
+		lk:       lk,
 		notifier: notifier,
 	}, nil
 }
@@ -42,12 +42,12 @@ type WriteNotifier interface {
 }
 
 type Storage struct {
-	ark      *archive.Archive
+	lk       *lake.Lake
 	notifier WriteNotifier
 }
 
-func NewStorage(ark *archive.Archive) *Storage {
-	return &Storage{ark: ark}
+func NewStorage(lk *lake.Lake) *Storage {
+	return &Storage{lk: lk}
 }
 
 func (s *Storage) Kind() api.StorageKind {
@@ -55,21 +55,21 @@ func (s *Storage) Kind() api.StorageKind {
 }
 
 func (s *Storage) NativeOrder() zbuf.Order {
-	return s.ark.DataOrder
+	return s.lk.DataOrder
 }
 
 func (s *Storage) MultiSource() driver.MultiSource {
-	return archive.NewMultiSource(s.ark, nil)
+	return lake.NewMultiSource(s.lk, nil)
 }
 
 func (s *Storage) StaticSource(src driver.Source) driver.MultiSource {
-	return archive.NewStaticSource(s.ark, src)
+	return lake.NewStaticSource(s.lk, src)
 }
 
 func (s *Storage) Summary(ctx context.Context) (storage.Summary, error) {
 	var sum storage.Summary
 	sum.Kind = api.ArchiveStore
-	err := archive.Walk(ctx, s.ark, func(chunk chunk.Chunk) error {
+	err := lake.Walk(ctx, s.lk, func(chunk chunk.Chunk) error {
 		info, err := iosrc.Stat(ctx, chunk.Path())
 		if err != nil {
 			return err
@@ -87,7 +87,7 @@ func (s *Storage) Summary(ctx context.Context) (storage.Summary, error) {
 }
 
 func (s *Storage) Write(ctx context.Context, zctx *resolver.Context, zr zbuf.Reader) error {
-	err := archive.Import(ctx, s.ark, zctx, zr)
+	err := lake.Import(ctx, s.lk, zctx, zr)
 	if s.notifier != nil {
 		s.notifier.WriteNotify()
 	}
@@ -95,7 +95,7 @@ func (s *Storage) Write(ctx context.Context, zctx *resolver.Context, zr zbuf.Rea
 }
 
 type Writer struct {
-	*archive.Writer
+	*lake.Writer
 	notifier WriteNotifier
 }
 
@@ -109,7 +109,7 @@ func (w *Writer) Close() error {
 
 // NewWriter returns a writer that will start a compaction when it is closed.
 func (s *Storage) NewWriter(ctx context.Context) (*Writer, error) {
-	w, err := archive.NewWriter(ctx, s.ark)
+	w, err := lake.NewWriter(ctx, s.lk)
 	if err != nil {
 		return nil, err
 	}
@@ -143,19 +143,19 @@ func (s *Storage) IndexCreate(ctx context.Context, req api.IndexPostRequest) err
 		rules = append(rules, rule)
 	}
 	// XXX Eventually this method should provide progress updates.
-	return archive.ApplyRules(ctx, s.ark, nil, rules...)
+	return lake.ApplyRules(ctx, s.lk, nil, rules...)
 }
 
 func (s *Storage) IndexSearch(ctx context.Context, zctx *resolver.Context, query index.Query) (zbuf.ReadCloser, error) {
-	return archive.FindReadCloser(ctx, zctx, s.ark, query, archive.AddPath(archive.DefaultAddPathField, false))
+	return lake.FindReadCloser(ctx, zctx, s.lk, query, lake.AddPath(lake.DefaultAddPathField, false))
 }
 
 func (s *Storage) ArchiveStat(ctx context.Context, zctx *resolver.Context) (zbuf.ReadCloser, error) {
-	return archive.Stat(ctx, zctx, s.ark)
+	return lake.Stat(ctx, zctx, s.lk)
 }
 
 func (s *Storage) Compact(ctx context.Context, logger *zap.Logger) error {
-	if err := archive.Compact(ctx, s.ark, logger); err != nil {
+	if err := lake.Compact(ctx, s.lk, logger); err != nil {
 		return err
 	}
 	select {
@@ -168,6 +168,6 @@ func (s *Storage) Compact(ctx context.Context, logger *zap.Logger) error {
 	// This is a stopgap solution to this problem; a more robust solution
 	// should be architected and implemented.
 	case <-time.After(time.Second * 60):
-		return archive.Purge(ctx, s.ark, logger)
+		return lake.Purge(ctx, s.lk, logger)
 	}
 }
diff --git a/ppl/zqd/worker.go b/ppl/zqd/worker.go
index 8b3c41674..7d0324a6a 100644
--- a/ppl/zqd/worker.go
+++ b/ppl/zqd/worker.go
@@ -4,7 +4,7 @@ import (
 	"net/http"
 
 	"github.com/brimsec/zq/api"
-	"github.com/brimsec/zq/ppl/archive"
+	"github.com/brimsec/zq/ppl/lake"
 	"github.com/brimsec/zq/ppl/zqd/search"
 	"github.com/brimsec/zq/ppl/zqd/storage/archivestore"
 	"github.com/brimsec/zq/zqe"
@@ -51,12 +51,12 @@ func handleWorkerChunkSearch(c *Core, w http.ResponseWriter, httpReq *http.Reque
 		return
 	}
 	ctx := httpReq.Context()
-	ark, err := archive.OpenArchiveWithContext(ctx, req.DataPath, &archive.OpenOptions{})
+	lk, err := lake.OpenLakeWithContext(ctx, req.DataPath, &lake.OpenOptions{})
 	if err != nil {
 		respondError(c, w, httpReq, err)
 		return
 	}
-	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(ark), c.requestLogger(httpReq))
+	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(lk), c.requestLogger(httpReq))
 	if err != nil {
 		respondError(c, w, httpReq, err)
 		return
diff --git a/proc/put/put.go b/proc/put/put.go
index f906b1a12..510e5bdc8 100644
--- a/proc/put/put.go
+++ b/proc/put/put.go
@@ -1,7 +1,10 @@
 package put
 
 import (
+	"fmt"
+
 	"github.com/brimsec/zq/expr"
+	"github.com/brimsec/zq/field"
 	"github.com/brimsec/zq/proc"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zcode"
@@ -10,19 +13,18 @@ import (
 
 // Put is a proc that modifies the record stream with computed values.
 // Each new value is called a clause and consists of a field name and
-// an expression.  Currently, the field name must be a top-level record
-// name, i.e., it cannot be a dotted record access or array reference
-// (XXX this will change when we add more comprehensive expression support).
-// Each put clause either replaces an existing value in the column specified
-// or appends a value as a new column.  Appended values appear as new
-// columns in the order that the clause appears in the put expression.
+// an expression. Each put clause either replaces an existing value in
+// the column specified or appends a value as a new column.  Appended
+// values appear as new columns in the order that the clause appears
+// in the put expression.
 type Proc struct {
 	pctx    *proc.Context
 	parent  proc.Interface
+	builder zcode.Builder
 	clauses []expr.Assignment
 	// vals is a fixed array to avoid re-allocating for every record
 	vals   []zng.Value
-	rules  map[int]*putRule
+	rules  map[int]putRule
 	warned map[string]struct{}
 }
 
@@ -34,28 +36,31 @@ type Proc struct {
 // language.
 type putRule struct {
 	typ         *zng.TypeRecord
-	clauseTypes []clauseType
-	// The clause numbers indexed by input column number that should be
-	// replaced, where -1 indicates no replacement.  Nil if there are no
-	// replacements.
-	replace []int
-	// The clause numbers that should be appeneded to the output record
-	// ordered left to right.  Nil if nothing need be appended.
-	append []int
-}
-
-type clauseType struct {
-	zng.Type
-	container bool
+	clauseTypes []zng.Type
+	step        step
 }
 
 func New(pctx *proc.Context, parent proc.Interface, clauses []expr.Assignment) (proc.Interface, error) {
+	for i, p := range clauses {
+		for j, c := range clauses {
+			if i == j {
+				continue
+			}
+			if p.LHS.Equal(c.LHS) {
+				return nil, fmt.Errorf("put: multiple assignments to %s", p.LHS)
+			}
+			if c.LHS.HasStrictPrefix(p.LHS) {
+				return nil, fmt.Errorf("put: conflicting nested assignments to %s and %s", p.LHS, c.LHS)
+			}
+		}
+	}
+
 	return &Proc{
 		pctx:    pctx,
 		parent:  parent,
 		clauses: clauses,
 		vals:    make([]zng.Value, len(clauses)),
-		rules:   make(map[int]*putRule),
+		rules:   make(map[int]putRule),
 		warned:  make(map[string]struct{}),
 	}, nil
 }
@@ -81,116 +86,278 @@ func (p *Proc) eval(in *zng.Record) ([]zng.Value, error) {
 	return vals, nil
 }
 
-func (p *Proc) buildRule(inType *zng.TypeRecord, vals []zng.Value) (*putRule, error) {
-	n := len(inType.Columns)
-	cols := make([]zng.Column, n, n+len(p.clauses))
-	copy(cols, inType.Columns)
-	clauseTypes := make([]clauseType, len(p.clauses))
-	var nreplace int
-	replace := make([]int, n)
-	for k := range replace {
-		replace[k] = -1
+// A step is a recursive data structure encoding a series of steps to be
+// carried out to construct an output record from an input record and
+// a slice of evaluated clauses.
+type step struct {
+	op        op
+	index     int
+	container bool
+	record    []step // for op == record
+}
+
+func (s *step) append(step step) {
+	s.record = append(s.record, step)
+}
+
+type op int
+
+const (
+	fromInput  op = iota // copy field from input record
+	fromClause           // copy field from put assignment
+	root                 // values are being copied to the root record (put .=)
+	record               // recurse into record below us
+)
+
+func (s step) build(in zcode.Bytes, b *zcode.Builder, vals []zng.Value) (zcode.Bytes, error) {
+	switch s.op {
+	case root:
+		bytes := make(zcode.Bytes, len(vals[s.index].Bytes))
+		copy(bytes, vals[s.index].Bytes)
+		return bytes, nil
+	case record:
+		b.Reset()
+		if err := s.buildRecord(in, b, vals); err != nil {
+			return nil, err
+		}
+		return b.Bytes(), nil
+	default:
+		// top-level op should be root or record
+		panic(fmt.Sprintf("put: unexpected step %v", s.op))
 	}
-	var tail []int
-	for k, cl := range p.clauses {
-		typ := vals[k].Type
-		clauseTypes[k] = clauseType{typ, zng.IsContainerType(typ)}
-		name := cl.LHS.String()
-		col := zng.Column{Name: name, Type: typ}
-		position, hasCol := inType.ColumnOfField(name)
-		if hasCol {
-			nreplace++
-			replace[position] = k
-			cols[position] = col
-		} else {
-			tail = append(tail, k)
-			cols = append(cols, col)
+}
+
+func (s step) buildRecord(in zcode.Bytes, b *zcode.Builder, vals []zng.Value) error {
+	ig := newGetter(in)
+
+	for _, step := range s.record {
+		switch step.op {
+		case fromInput:
+			bytes, err := ig.nth(step.index)
+			if err != nil {
+				return err
+			}
+			if step.container {
+				b.AppendContainer(bytes)
+			} else {
+				b.AppendPrimitive(bytes)
+			}
+		case fromClause:
+			if step.container {
+				b.AppendContainer(vals[step.index].Bytes)
+			} else {
+				b.AppendPrimitive(vals[step.index].Bytes)
+			}
+		case record:
+			b.BeginContainer()
+			bytes, err := in, error(nil)
+			if step.index >= 0 {
+				bytes, err = ig.nth(step.index)
+				if err != nil {
+					return err
+				}
+			}
+			if err := step.buildRecord(bytes, b, vals); err != nil {
+				return err
+			}
+			b.EndContainer()
 		}
 	}
-	if nreplace == 0 {
-		replace = nil
+	return nil
+}
+
+// A getter provides random access to values in a zcode container
+// using zcode.Iter. It uses a cursor to avoid quadratic re-seeks for
+// the common case where values are fetched sequentially.
+type getter struct {
+	cursor    int
+	container zcode.Bytes
+	iter      zcode.Iter
+}
+
+func newGetter(cont zcode.Bytes) getter {
+	return getter{
+		cursor:    -1,
+		container: cont,
+		iter:      cont.Iter(),
+	}
+}
+func (ig *getter) nth(n int) (zcode.Bytes, error) {
+	if n < ig.cursor {
+		ig.iter = ig.container.Iter()
 	}
+	for !ig.iter.Done() {
+		zv, _, err := ig.iter.Next()
+		if err != nil {
+			return nil, err
+		}
+		ig.cursor++
+		if ig.cursor == n {
+			return zv, nil
+		}
+	}
+	return nil, fmt.Errorf("getter.nth: array index %d out of bounds", n)
+}
+
+func findOverwriteClause(path field.Static, clauses []expr.Assignment) (int, field.Static, bool) {
+	for i, cand := range clauses {
+		if path.Equal(cand.LHS) || cand.LHS.HasStrictPrefix(path) {
+			return i, cand.LHS, true
+		}
+	}
+	return -1, nil, false
+}
+
+func (p *Proc) deriveSteps(inType *zng.TypeRecord, vals []zng.Value) (step, *zng.TypeRecord, error) {
+	// special case: assign to root (put .=x)
+	if p.clauses[0].LHS.IsRoot() {
+		recVal, ok := vals[0].Type.(*zng.TypeRecord)
+		if !ok {
+			return step{}, nil, fmt.Errorf("put .=x: cannot put a non-record to .")
+		}
+		typ, err := p.pctx.TypeContext.LookupTypeRecord(recVal.Columns)
+		return step{op: root, index: 0}, typ, err
+	}
+	return p.deriveRecordSteps(field.NewRoot(), inType.Columns, vals)
+}
+
+func (p *Proc) deriveRecordSteps(parentPath field.Static, inCols []zng.Column, vals []zng.Value) (step, *zng.TypeRecord, error) {
+	s := step{op: record}
+	cols := make([]zng.Column, 0)
+
+	// First look at all input columns to see which should
+	// be copied over and which should be overwritten by
+	// assignments.
+	for i, inCol := range inCols {
+		path := append(parentPath, inCol.Name)
+		matchIndex, matchPath, found := findOverwriteClause(path, p.clauses)
+		switch {
+		// input not overwritten by assignment: copy input value.
+		case !found:
+			s.append(step{
+				op:        fromInput,
+				container: zng.IsContainerType(inCol.Type),
+				index:     i,
+			})
+			cols = append(cols, inCol)
+		// input field overwritten by non-nested assignment: copy assignment value.
+		case len(path) == len(matchPath):
+			s.append(step{
+				op:        fromClause,
+				container: zng.IsContainerType(vals[matchIndex].Type),
+				index:     matchIndex,
+			})
+			cols = append(cols, zng.Column{inCol.Name, vals[matchIndex].Type})
+		// input record field overwritten by nested assignment: recurse.
+		case len(path) < len(matchPath) && zng.IsRecordType(inCol.Type):
+			nestedStep, typ, err := p.deriveRecordSteps(path, inCol.Type.(*zng.TypeRecord).Columns, vals)
+			if err != nil {
+				return step{}, nil, err
+			}
+			nestedStep.index = i
+			s.append(nestedStep)
+			cols = append(cols, zng.Column{inCol.Name, typ})
+		// input non-record field overwritten by nested assignment(s): recurse.
+		case len(path) < len(matchPath) && !zng.IsRecordType(inCol.Type):
+			nestedStep, typ, err := p.deriveRecordSteps(path, []zng.Column{}, vals)
+			if err != nil {
+				return step{}, nil, err
+			}
+			nestedStep.index = i
+			s.append(nestedStep)
+			cols = append(cols, zng.Column{inCol.Name, typ})
+		default:
+			panic("faulty logic")
+		}
+	}
+
+	appendClause := func(cl expr.Assignment) bool {
+		if !cl.LHS.HasPrefix(parentPath) {
+			return false
+		}
+		return !hasField(cl.LHS[len(parentPath)], cols)
+	}
+	// Then, look at put assignments to see if there are any new fields to append.
+	for i, cl := range p.clauses {
+		if appendClause(cl) {
+			switch {
+			// Append value at this level
+			case len(cl.LHS) == len(parentPath)+1:
+				s.append(step{
+					op:        fromClause,
+					container: zng.IsContainerType(vals[i].Type),
+					index:     i,
+				})
+				cols = append(cols, zng.Column{cl.LHS[len(parentPath)], vals[i].Type})
+			// Appended and nest. For example, this would happen with "put b.c=1" applied to a record {"a": 1}.
+			case len(cl.LHS) > len(parentPath)+1:
+				path := append(parentPath, cl.LHS[len(parentPath)])
+				nestedStep, typ, err := p.deriveRecordSteps(path, []zng.Column{}, vals)
+				if err != nil {
+					return step{}, nil, err
+				}
+				nestedStep.index = -1
+				cols = append(cols, zng.Column{cl.LHS[len(parentPath)], typ})
+				s.append(nestedStep)
+			}
+		}
+	}
+
 	typ, err := p.pctx.TypeContext.LookupTypeRecord(cols)
-	if err != nil {
-		return nil, err
+	return s, typ, err
+}
+
+func hasField(name string, cols []zng.Column) bool {
+	for _, col := range cols {
+		if col.Name == name {
+			return true
+		}
 	}
-	return &putRule{
-		typ:         typ,
-		clauseTypes: clauseTypes,
-		replace:     replace,
-		append:      tail,
-	}, nil
+	return false
 }
 
-func clauseTypesMatch(types []clauseType, vals []zng.Value) bool {
+func sameTypes(types []zng.Type, vals []zng.Value) bool {
 	for k, typ := range types {
-		if vals[k].Type != typ.Type {
+		if vals[k].Type != typ {
 			return false
 		}
 	}
 	return true
 }
 
-func (p *Proc) lookupRule(inType *zng.TypeRecord, vals []zng.Value) (*putRule, error) {
-	rule := p.rules[inType.ID()]
-	if rule != nil && clauseTypesMatch(rule.clauseTypes, vals) {
+func (p *Proc) lookupRule(inType *zng.TypeRecord, vals []zng.Value) (putRule, error) {
+	rule, ok := p.rules[inType.ID()]
+	if ok && sameTypes(rule.clauseTypes, vals) {
 		return rule, nil
 	}
-	rule, err := p.buildRule(inType, vals)
+	step, typ, err := p.deriveSteps(inType, vals)
+	var clauseTypes []zng.Type
+	for _, val := range vals {
+		clauseTypes = append(clauseTypes, val.Type)
+	}
+	rule = putRule{typ, clauseTypes, step}
 	p.rules[inType.ID()] = rule
 	return rule, err
 }
 
-func (p *Proc) put(in *zng.Record) *zng.Record {
+func (p *Proc) put(in *zng.Record) (*zng.Record, error) {
 	vals, err := p.eval(in)
 	if err != nil {
 		p.maybeWarn(err)
-		return in
+		return in, nil
 	}
 	rule, err := p.lookupRule(in.Type, vals)
 	if err != nil {
 		p.maybeWarn(err)
-		return in
-	}
-	// Start the new output value by either copying or replacing the input values.
-	var bytes zcode.Bytes
-	if rule.replace == nil {
-		// All fields are being appended.
-		bytes = make([]byte, len(in.Raw))
-		copy(bytes, in.Raw)
-	} else {
-		// We're overwriting one or more fields.  Travese the
-		// replacement vector to determine whether each value should
-		// be copied from the input or replaced with a clause result.
-		iter := in.ZvalIter()
-		for _, clause := range rule.replace {
-			item, isContainer, err := iter.Next()
-			if err != nil {
-				panic(err)
-			}
-			if clause >= 0 {
-				item = vals[clause].Bytes
-				isContainer = rule.clauseTypes[clause].container
-			}
-			if isContainer {
-				bytes = zcode.AppendContainer(bytes, item)
-			} else {
-				bytes = zcode.AppendPrimitive(bytes, item)
-			}
-		}
+		return in, nil
 	}
-	// Finish building the output by appending the remaining clauses if any.
-	for _, clause := range rule.append {
-		item := vals[clause].Bytes
-		if rule.clauseTypes[clause].container {
-			bytes = zcode.AppendContainer(bytes, item)
-		} else {
-			bytes = zcode.AppendPrimitive(bytes, item)
-		}
+
+	bytes, err := rule.step.build(in.Raw, &p.builder, vals)
+	if err != nil {
+		return nil, err
 	}
-	return zng.NewRecord(rule.typ, bytes)
+	return zng.NewRecord(rule.typ, bytes), nil
 }
-
 func (p *Proc) Pull() (zbuf.Batch, error) {
 	batch, err := p.parent.Pull()
 	if proc.EOS(batch, err) {
@@ -199,8 +366,12 @@ func (p *Proc) Pull() (zbuf.Batch, error) {
 	recs := make([]*zng.Record, 0, batch.Length())
 	for k := 0; k < batch.Length(); k++ {
 		in := batch.Index(k)
+		rec, err := p.put(in)
+		if err != nil {
+			return nil, err
+		}
 		// Keep is necessary because put can return its argument.
-		recs = append(recs, p.put(in).Keep())
+		recs = append(recs, rec.Keep())
 	}
 	batch.Unref()
 	return zbuf.Array(recs), nil
diff --git a/zio/detector/file.go b/zio/detector/file.go
index 57306867b..cdffb9173 100644
--- a/zio/detector/file.go
+++ b/zio/detector/file.go
@@ -18,8 +18,6 @@ import (
 	"github.com/xitongsys/parquet-go/source"
 )
 
-const StdinPath = "/dev/stdin"
-
 // OpenFile creates and returns zbuf.File for the indicated "path",
 // which can be a local file path, a local directory path, or an S3
 // URL. If the path is neither of these or can't otherwise be opened,


Test patch:
diff --git a/expr/ztests/shape-fill.yaml b/expr/ztests/shape-fill.yaml
new file mode 100644
index 000000000..437b5ff7f
--- /dev/null
+++ b/expr/ztests/shape-fill.yaml
@@ -0,0 +1,9 @@
+zql: put t = fill(., "({a:{b:{a1:string}}})")
+
+input: |
+  #0:record[s:string]
+  0:[x;]
+
+output: |
+  #0:record[s:string,t:record[s:string,a:record[b:record[a1:string]]]]
+  0:[x;[x;-;]]
diff --git a/ppl/archive/chunk/writer_test.go b/ppl/lake/chunk/writer_test.go
similarity index 97%
rename from ppl/archive/chunk/writer_test.go
rename to ppl/lake/chunk/writer_test.go
index 49ac877fa..d60e801ed 100644
--- a/ppl/archive/chunk/writer_test.go
+++ b/ppl/lake/chunk/writer_test.go
@@ -5,7 +5,7 @@ import (
 	"testing"
 
 	"github.com/brimsec/zq/pkg/iosrc"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/tzngio"
 	"github.com/brimsec/zq/zng"
diff --git a/ppl/archive/import_test.go b/ppl/lake/import_test.go
similarity index 86%
rename from ppl/archive/import_test.go
rename to ppl/lake/import_test.go
index 30f4342b9..06eda9b29 100644
--- a/ppl/archive/import_test.go
+++ b/ppl/lake/import_test.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -29,11 +29,11 @@ func testImportStaleDuration(t *testing.T, stale time.Duration, expected uint64)
 0:[1587508850.06466032;202;]`
 
 	// create archive with a 1 ns ImportFlushTimeout
-	ark, err := CreateOrOpenArchive(t.TempDir(), nil, nil)
+	lk, err := CreateOrOpenLake(t.TempDir(), nil, nil)
 	require.NoError(t, err)
 
 	// write one record to an open archive.Writer and do NOT close it.
-	w, err := NewWriter(context.Background(), ark)
+	w, err := NewWriter(context.Background(), lk)
 	require.NoError(t, err)
 	defer w.Close()
 	w.SetStaleDuration(stale)
@@ -43,7 +43,7 @@ func testImportStaleDuration(t *testing.T, stale time.Duration, expected uint64)
 	// flush stale writers and ensure data has been written to archive
 	err = w.flushStaleWriters()
 	require.NoError(t, err)
-	count, err := RecordCount(context.Background(), ark)
+	count, err := RecordCount(context.Background(), lk)
 	require.NoError(t, err)
 	assert.EqualValues(t, expected, count)
 }
diff --git a/ppl/archive/index/definition.go b/ppl/lake/index/definition.go
similarity index 100%
rename from ppl/archive/index/definition.go
rename to ppl/lake/index/definition.go
diff --git a/ppl/archive/index/index.go b/ppl/lake/index/index.go
similarity index 100%
rename from ppl/archive/index/index.go
rename to ppl/lake/index/index.go
diff --git a/ppl/archive/index/index_test.go b/ppl/lake/index/index_test.go
similarity index 100%
rename from ppl/archive/index/index_test.go
rename to ppl/lake/index/index_test.go
diff --git a/ppl/archive/index/multiwriter.go b/ppl/lake/index/multiwriter.go
similarity index 100%
rename from ppl/archive/index/multiwriter.go
rename to ppl/lake/index/multiwriter.go
diff --git a/ppl/archive/index/proc.go b/ppl/lake/index/proc.go
similarity index 100%
rename from ppl/archive/index/proc.go
rename to ppl/lake/index/proc.go
diff --git a/ppl/archive/index/query.go b/ppl/lake/index/query.go
similarity index 100%
rename from ppl/archive/index/query.go
rename to ppl/lake/index/query.go
diff --git a/ppl/archive/index/rule.go b/ppl/lake/index/rule.go
similarity index 100%
rename from ppl/archive/index/rule.go
rename to ppl/lake/index/rule.go
diff --git a/ppl/archive/index/rule_test.go b/ppl/lake/index/rule_test.go
similarity index 100%
rename from ppl/archive/index/rule_test.go
rename to ppl/lake/index/rule_test.go
diff --git a/ppl/archive/index/writer.go b/ppl/lake/index/writer.go
similarity index 100%
rename from ppl/archive/index/writer.go
rename to ppl/lake/index/writer.go
diff --git a/ppl/archive/index/writer_test.go b/ppl/lake/index/writer_test.go
similarity index 100%
rename from ppl/archive/index/writer_test.go
rename to ppl/lake/index/writer_test.go
diff --git a/ppl/archive/archive_test.go b/ppl/lake/lake_test.go
similarity index 82%
rename from ppl/archive/archive_test.go
rename to ppl/lake/lake_test.go
index 5b54e32a4..2b179f2f0 100644
--- a/ppl/archive/archive_test.go
+++ b/ppl/lake/lake_test.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"bytes"
@@ -13,9 +13,9 @@ import (
 	"github.com/brimsec/zq/pkg/nano"
 	"github.com/brimsec/zq/pkg/promtest"
 	"github.com/brimsec/zq/pkg/test"
-	"github.com/brimsec/zq/ppl/archive/chunk"
-	"github.com/brimsec/zq/ppl/archive/immcache"
-	"github.com/brimsec/zq/ppl/archive/index"
+	"github.com/brimsec/zq/ppl/lake/chunk"
+	"github.com/brimsec/zq/ppl/lake/immcache"
+	"github.com/brimsec/zq/ppl/lake/index"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio"
 	"github.com/brimsec/zq/zio/detector"
@@ -29,19 +29,19 @@ import (
 const babble = "../../ztests/suite/data/babble.tzng"
 
 func createArchiveSpace(t *testing.T, datapath string, srcfile string, co *CreateOptions) {
-	ark, err := CreateOrOpenArchive(datapath, co, nil)
+	lk, err := CreateOrOpenLake(datapath, co, nil)
 	require.NoError(t, err)
 
-	importTestFile(t, ark, srcfile)
+	importTestFile(t, lk, srcfile)
 }
 
-func importTestFile(t *testing.T, ark *Archive, srcfile string) {
+func importTestFile(t *testing.T, lk *Lake, srcfile string) {
 	zctx := resolver.NewContext()
 	reader, err := detector.OpenFile(zctx, srcfile, zio.ReaderOpts{})
 	require.NoError(t, err)
 	defer reader.Close()
 
-	err = Import(context.Background(), ark, zctx, reader)
+	err = Import(context.Background(), lk, zctx, reader)
 	require.NoError(t, err)
 }
 
@@ -49,17 +49,17 @@ func indexArchiveSpace(t *testing.T, datapath string, ruledef string) {
 	rule, err := index.NewRule(ruledef)
 	require.NoError(t, err)
 
-	ark, err := OpenArchive(datapath, nil)
+	lk, err := OpenLake(datapath, nil)
 	require.NoError(t, err)
 
-	err = ApplyRules(context.Background(), ark, nil, rule)
+	err = ApplyRules(context.Background(), lk, nil, rule)
 	require.NoError(t, err)
 }
 
-func indexQuery(t *testing.T, ark *Archive, patterns []string, opts ...FindOption) string {
+func indexQuery(t *testing.T, lk *Lake, patterns []string, opts ...FindOption) string {
 	q, err := index.ParseQuery("", patterns)
 	require.NoError(t, err)
-	rc, err := FindReadCloser(context.Background(), resolver.NewContext(), ark, q, opts...)
+	rc, err := FindReadCloser(context.Background(), resolver.NewContext(), lk, q, opts...)
 	require.NoError(t, err)
 	defer rc.Close()
 
@@ -77,13 +77,13 @@ func TestMetadataCache(t *testing.T) {
 	icache, err := immcache.NewLocalCache(128, reg)
 	require.NoError(t, err)
 
-	ark, err := OpenArchive(datapath, &OpenOptions{
+	lk, err := OpenLake(datapath, &OpenOptions{
 		ImmutableCache: icache,
 	})
 	require.NoError(t, err)
 
 	for i := 0; i < 4; i++ {
-		count, err := RecordCount(context.Background(), ark)
+		count, err := RecordCount(context.Background(), lk)
 		require.NoError(t, err)
 		assert.EqualValues(t, 1000, count)
 	}
@@ -105,7 +105,7 @@ func TestSeekIndex(t *testing.T) {
 		ImportStreamRecordsMax = orig
 	}()
 	createArchiveSpace(t, datapath, babble, nil)
-	_, err := OpenArchive(datapath, &OpenOptions{})
+	_, err := OpenLake(datapath, &OpenOptions{})
 	require.NoError(t, err)
 
 	first1 := nano.Ts(1587513592062544400)
diff --git a/ppl/archive/overlap_test.go b/ppl/lake/overlap_test.go
similarity index 93%
rename from ppl/archive/overlap_test.go
rename to ppl/lake/overlap_test.go
index d4c9d601d..2ae8344c3 100644
--- a/ppl/archive/overlap_test.go
+++ b/ppl/lake/overlap_test.go
@@ -1,4 +1,4 @@
-package archive
+package lake
 
 import (
 	"context"
@@ -9,7 +9,7 @@ import (
 	"testing"
 
 	"github.com/brimsec/zq/pkg/nano"
-	"github.com/brimsec/zq/ppl/archive/chunk"
+	"github.com/brimsec/zq/ppl/lake/chunk"
 	"github.com/brimsec/zq/zbuf"
 	"github.com/brimsec/zq/zio/tzngio"
 	"github.com/brimsec/zq/zng/resolver"
@@ -25,10 +25,10 @@ func kid(s string) ksuid.KSUID {
 	return k
 }
 
-func importTzng(t *testing.T, ark *Archive, s string) {
+func importTzng(t *testing.T, lk *Lake, s string) {
 	zctx := resolver.NewContext()
 	reader := tzngio.NewReader(strings.NewReader(s), zctx)
-	err := Import(context.Background(), ark, zctx, reader)
+	err := Import(context.Background(), lk, zctx, reader)
 	require.NoError(t, err)
 }
 
@@ -194,7 +194,7 @@ func TestOverlapWalking(t *testing.T) {
 	require.NoError(t, err)
 	defer os.RemoveAll(datapath)
 
-	ark, err := CreateOrOpenArchive(datapath, &CreateOptions{}, nil)
+	lk, err := CreateOrOpenLake(datapath, &CreateOptions{}, nil)
 	require.NoError(t, err)
 
 	data1 := `
@@ -213,19 +213,19 @@ func TestOverlapWalking(t *testing.T) {
 0:[.000000025;25;]
 `
 	dataChunkSpans := []nano.Span{{Ts: 15, Dur: 11}, {Ts: 10, Dur: 11}, {Ts: 0, Dur: 6}}
-	importTzng(t, ark, data2)
-	importTzng(t, ark, data1)
-	importTzng(t, ark, data3)
+	importTzng(t, lk, data2)
+	importTzng(t, lk, data1)
+	importTzng(t, lk, data3)
 
 	{
 		var chunks []chunk.Chunk
-		err = tsDirVisit(context.Background(), ark, nano.MaxSpan, func(tsd tsDir, c []chunk.Chunk) error {
+		err = tsDirVisit(context.Background(), lk, nano.MaxSpan, func(tsd tsDir, c []chunk.Chunk) error {
 			chunks = append(chunks, c...)
 			return nil
 		})
 		require.NoError(t, err)
 		require.Len(t, chunks, 3)
-		chunk.Sort(ark.DataOrder, chunks)
+		chunk.Sort(lk.DataOrder, chunks)
 		var spans []nano.Span
 		for _, c := range chunks {
 			spans = append(spans, c.Span())
@@ -234,7 +234,7 @@ func TestOverlapWalking(t *testing.T) {
 	}
 	{
 		var chunks []chunk.Chunk
-		err = Walk(context.Background(), ark, func(c chunk.Chunk) error {
+		err = Walk(context.Background(), lk, func(c chunk.Chunk) error {
 			chunks = append(chunks, c)
 			return nil
 		})
@@ -248,13 +248,13 @@ func TestOverlapWalking(t *testing.T) {
 	}
 	{
 		var chunks []chunk.Chunk
-		err = tsDirVisit(context.Background(), ark, nano.Span{Ts: 12, Dur: 20}, func(tsd tsDir, c []chunk.Chunk) error {
+		err = tsDirVisit(context.Background(), lk, nano.Span{Ts: 12, Dur: 20}, func(tsd tsDir, c []chunk.Chunk) error {
 			chunks = append(chunks, c...)
 			return nil
 		})
 		require.NoError(t, err)
 		assert.Len(t, chunks, 2)
-		chunk.Sort(ark.DataOrder, chunks)
+		chunk.Sort(lk.DataOrder, chunks)
 		var spans []nano.Span
 		for _, c := range chunks {
 			spans = append(spans, c.Span())
@@ -267,7 +267,7 @@ func TestOverlapWalking(t *testing.T) {
 			chunkSpans []nano.Span
 		}
 		var sispans []sispan
-		err = SpanWalk(context.Background(), ark, nano.Span{Ts: 12, Dur: 10}, func(si SpanInfo) error {
+		err = SpanWalk(context.Background(), lk, nano.Span{Ts: 12, Dur: 10}, func(si SpanInfo) error {
 			var chunkSpans []nano.Span
 			for _, c := range si.Chunks {
 				chunkSpans = append(chunkSpans, c.Span())
diff --git a/ppl/zqd/db/postgresdb/ztests/duplname.yaml b/ppl/zqd/db/postgresdb/ztests/duplname.yaml
index 429e492a5..5264c0ea9 100644
--- a/ppl/zqd/db/postgresdb/ztests/duplname.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/duplname.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/import.yaml b/ppl/zqd/db/postgresdb/ztests/import.yaml
index 7f8821452..02e43b239 100644
--- a/ppl/zqd/db/postgresdb/ztests/import.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/import.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/info.yaml b/ppl/zqd/db/postgresdb/ztests/info.yaml
index ab0dc36f2..e3dda5490 100644
--- a/ppl/zqd/db/postgresdb/ztests/info.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/info.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/ls.yaml b/ppl/zqd/db/postgresdb/ztests/ls.yaml
index eb329db20..21ceeab09 100644
--- a/ppl/zqd/db/postgresdb/ztests/ls.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/ls.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/rename.yaml b/ppl/zqd/db/postgresdb/ztests/rename.yaml
index 57db1226c..3549b9152 100644
--- a/ppl/zqd/db/postgresdb/ztests/rename.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/rename.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/rm.yaml b/ppl/zqd/db/postgresdb/ztests/rm.yaml
index a42110613..0fecae9e6 100644
--- a/ppl/zqd/db/postgresdb/ztests/rm.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/rm.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 env:
   - PG_TEST
diff --git a/ppl/zqd/db/postgresdb/ztests/tenant.yaml b/ppl/zqd/db/postgresdb/ztests/tenant.yaml
index 31d5f294c..cb5ded710 100644
--- a/ppl/zqd/db/postgresdb/ztests/tenant.yaml
+++ b/ppl/zqd/db/postgresdb/ztests/tenant.yaml
@@ -1,4 +1,4 @@
-tag: postgres
+tag: services
 
 script: |
   # Start an authentication enabled zqd, and create 2 separate access tokens
diff --git a/ppl/zqd/handlers_test.go b/ppl/zqd/handlers_test.go
index 2512b7bd4..dd04ec019 100644
--- a/ppl/zqd/handlers_test.go
+++ b/ppl/zqd/handlers_test.go
@@ -25,7 +25,7 @@ import (
 	"github.com/brimsec/zq/pkg/nano"
 	"github.com/brimsec/zq/pkg/promtest"
 	"github.com/brimsec/zq/pkg/test"
-	"github.com/brimsec/zq/ppl/archive/immcache"
+	"github.com/brimsec/zq/ppl/lake/immcache"
 	"github.com/brimsec/zq/ppl/zqd"
 	"github.com/brimsec/zq/ppl/zqd/pcapanalyzer"
 	"github.com/brimsec/zq/zbuf"
@@ -676,7 +676,10 @@ func TestArchiveInProcessCache(t *testing.T) {
 0:[1000;]`
 
 	core, conn := newCoreWithConfig(t, zqd.Config{
-		ImmutableCache: immcache.Config{LocalCacheSize: 128},
+		ImmutableCache: immcache.Config{
+			Kind:           immcache.KindLocal,
+			LocalCacheSize: 128,
+		},
 	})
 
 	sp, err := conn.SpacePost(context.Background(), api.SpacePostRequest{
diff --git a/ppl/zqd/ztests/cache-local.yaml b/ppl/zqd/ztests/immcache-local.yaml
similarity index 93%
rename from ppl/zqd/ztests/cache-local.yaml
rename to ppl/zqd/ztests/immcache-local.yaml
index f9967f61f..bc152af0d 100644
--- a/ppl/zqd/ztests/cache-local.yaml
+++ b/ppl/zqd/ztests/immcache-local.yaml
@@ -1,5 +1,5 @@
 script: |
-  ZQD_EXTRA_FLAGS=-immcache.localsize=128 source services.sh
+  ZQD_EXTRA_FLAGS=-immcache.kind=local source services.sh
   zapi -h $ZQD_HOST new -d ./root -k archivestore -thresh 20KiB testsp
   zapi -h $ZQD_HOST -s testsp post babble.tzng >/dev/null
   echo ===
diff --git a/ppl/zqd/ztests/redis/immcache.yaml b/ppl/zqd/ztests/redis/immcache.yaml
new file mode 100644
index 000000000..1d13c2045
--- /dev/null
+++ b/ppl/zqd/ztests/redis/immcache.yaml
@@ -0,0 +1,34 @@
+tag: services
+
+script: |
+  source services.sh
+  zapi -h $ZQD_HOST new -k archivestore -thresh 20KiB testsp
+  zapi -h $ZQD_HOST -s testsp post babble.tzng >/dev/null
+  echo ===
+  zapi -h $ZQD_HOST -s testsp get -f text "count()"
+  echo ===
+  zapi -h $ZQD_HOST -s testsp get -f text "count()"
+  echo ===
+  curl $ZQD_HOST/metrics | grep archive_cache
+
+inputs:
+  - name: babble.tzng
+    source: ../../../../ztests/suite/data/babble-sorted.tzng
+  - name: services.sh
+    source: services.sh
+
+outputs:
+  - name: stdout
+    data: |
+      testsp: space created
+      ===
+      1000
+      ===
+      1000
+      ===
+      # HELP archive_cache_hits_total Number of hits for a cache lookup.
+      # TYPE archive_cache_hits_total counter
+      archive_cache_hits_total{kind="metadata"} 4
+      # HELP archive_cache_misses_total Number of misses for a cache lookup.
+      # TYPE archive_cache_misses_total counter
+      archive_cache_misses_total{kind="metadata"} 2
diff --git a/ppl/zqd/ztests/redis/services.sh b/ppl/zqd/ztests/redis/services.sh
new file mode 100644
index 000000000..aece3cb31
--- /dev/null
+++ b/ppl/zqd/ztests/redis/services.sh
@@ -0,0 +1,34 @@
+#!/bin/bash
+
+function awaitfile {
+  file=$1
+  i=0
+  until [ -f $file ]; do
+    let i+=1
+    if [ $i -gt 5 ]; then
+      echo "timed out waiting for file \"$file\" to appear"
+      echo "zqd log:"
+      cat zqd.log
+      exit 1
+    fi
+    sleep 1
+  done
+}
+
+portdir=$(mktemp -d)
+mkdir -p data
+
+zqd listen -l=localhost:0 \
+  -data=data \
+  -loglevel=warn \
+  -portfile="$portdir/zqd" \
+  -redis.enabled \
+  -immcache.kind=redis \
+  &> zqd.log &
+
+zqdpid=$!
+awaitfile $portdir/zqd
+trap "rm -rf $portdir ; kill -9 $zqdpid &>/dev/null ;" EXIT
+
+export ZQD_HOST=localhost:$(cat $portdir/zqd)
+
diff --git a/proc/put/ztests/change-type.yaml b/proc/put/ztests/change-type.yaml
new file mode 100644
index 000000000..10068923c
--- /dev/null
+++ b/proc/put/ztests/change-type.yaml
@@ -0,0 +1,14 @@
+zql: put x=network_of(addr)
+
+input: |
+  #0:record[addr:ip]
+  0:[10.0.0.1;]
+  0:[fe80::215:17ff:fe84:c13f;]
+  0:[10.0.0.1;]
+
+output: |
+  #0:record[addr:ip,x:net]
+  0:[10.0.0.1;10.0.0.0/8;]
+  #1:record[addr:ip,x:error]
+  1:[fe80::215:17ff:fe84:c13f;not an IPv4;]
+  0:[10.0.0.1;10.0.0.0/8;]
diff --git a/proc/put/ztests/nested.yaml b/proc/put/ztests/nested.yaml
new file mode 100644
index 000000000..ba57f3fc9
--- /dev/null
+++ b/proc/put/ztests/nested.yaml
@@ -0,0 +1,28 @@
+script:
+  zq -i zson -f zson -pretty=0 'put pre="pre", a.a=1, a.b=2, a.c.a=3, a.d=4, b=5, post="post"' -
+
+inputs:
+  - name: stdin
+    data: |
+      {}
+      {"a": null}
+      {"a": {}}
+      {"a": {"c": null}}
+      {"a": {"c": {}}}
+      {"a": {"c": {"a": 4}}}
+      {"a": {"c": {"a": {"a": 4}}}}
+      {"pre": "pre","a": {"x": 10,"b": 2,"c": {"a": 3},"d": 4},"b": 5,"post": "post"}
+      {"other": "other"}
+
+outputs:
+  - name: stdout
+    data: |
+      {pre:"pre",a:{a:1,b:2,c:{a:3} (=0),d:4} (=1),b:5,post:"post"} (=2)
+      {a:{a:1,b:2,c:{a:3},d:4} (1),pre:"pre",b:5,post:"post"} (=3)
+      {a:{a:1,b:2,c:{a:3},d:4},pre:"pre",b:5,post:"post"} (3)
+      {a:{c:{a:3} (0),a:1,b:2,d:4} (=4),pre:"pre",b:5,post:"post"} (=5)
+      {a:{c:{a:3},a:1,b:2,d:4},pre:"pre",b:5,post:"post"} (5)
+      {a:{c:{a:3},a:1,b:2,d:4},pre:"pre",b:5,post:"post"} (5)
+      {a:{c:{a:3},a:1,b:2,d:4},pre:"pre",b:5,post:"post"} (5)
+      {pre:"pre",a:{x:10,b:2,c:{a:3} (0),d:4,a:1} (=6),b:5,post:"post"} (=7)
+      {other:"other",pre:"pre",a:{a:1,b:2,c:{a:3},d:4} (1),b:5,post:"post"} (=8)
diff --git a/proc/put/ztests/root.yaml b/proc/put/ztests/root.yaml
new file mode 100644
index 000000000..2623ee1db
--- /dev/null
+++ b/proc/put/ztests/root.yaml
@@ -0,0 +1,8 @@
+zql: put .=a
+
+input: |
+  {"a": {"a": "a", "b": "b"}, "b": "b"}
+
+output: |
+   #0:record[a:string,b:string]
+   0:[a;b;]