Repository: brimsec/zq
Pull request #1984 (2021-01-18 17:07:33)

Add shaping functions
This commit adds the shaping 'primitive' functions cast(), fill(),
crop(), and order() along with fit() (composition of fill and crop)
and shape (composition of all four primitive functions). 

A description of what these functions do is in #1783; the ztests of this commit also provide some illustration.

I will expand the test set after #1932 is done, since a lot of shaping assignments will use that. 


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ddca8d2db..4e0911a9b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,45 @@ These entries focus on changes we think are relevant to users of Brim,
 zq, or pcap.  For all changes to zqd, its API, or to other components in the
 zq repo, check the git log.
 
+## v0.28.0
+**NOTE** - Beginning with this release, a subset of the source code in the
+[github.com/brimsec/zq](https://github.com/brimsec/zq) GitHub repository is
+covered by a source-available style license, the
+[Polyform Perimeter License (PPL)](https://polyformproject.org/licenses/perimeter/1.0.0/).
+We've moved the PPL-covered code under a `ppl/` directory in the repository.
+The majority of our source code retains the existing BSD-3-Clause license.
+
+The overwhelming majority of zq/zqd users and developers will not be impacted
+by this change, including those using zq/zqd in commercial settings. The use of
+the source-available Polyform Perimeter license prevents use cases like
+marketing a work as a "as-a-service" style offering for server components like
+zqd while using material covered under the PPL.
+
+In general, we are making this change to ensure technology giants can't use the
+PPL-covered code to make replacement offerings of our projects. We believe
+users and developers should have access to the source code for our projects,
+and we need a sustainable business model to continue funding our work. Using
+the source-available Polyform Perimeter license on portions of the source code
+lets us realize both. 
+
+For more detail regarding licensing, see the
+[CONTRIBUTING.md](https://github.com/brimsec/zq/blob/master/CONTRIBUTING.md)
+doc, and feel free to come talk to us on
+[Slack](https://www.brimsecurity.com/join-slack/) if you have additional
+questions.
+
+---
+
+* zqd: Update Zeek pointer to [v3.2.1-brim9](https://github.com/brimsec/zeek/releases/tag/v3.2.1-brim9) which provides the latest [geolocation](https://github.com/brimsec/brim/wiki/Geolocation) data (#2010)
+* zqd: Update Suricata pointer to [v5.0.3-brim1](https://github.com/brimsec/build-suricata/releases/tag/v5.0.3-brim1) which disables checksum checks, allowing for alert creation on more types of pcaps (#1975)
+* zson: Update [Zeek Interoperability doc](https://github.com/brimsec/zq/blob/master/zng/docs/zeek-compat.md) to include current ZSON syntax (#1956)
+* zq: Ensure the output from the [`fuse`](https://github.com/brimsec/zq/tree/master/zql/docs/processors#fuse) processor is deterministic (#1958)
+* zq: Fix an issue where the presence of the Greek µ character caused a ZSON read parsing error (#1967)
+* zqd: Fix an issue where Zeek events generated during pcap import and written to an archivestore were only visible after ingest completion (#1973)
+* zqd: Change the logger configuration to output stacktraces on messages of level "warn" and higher (#1990)
+* zq: Update [performance results](https://github.com/brimsec/zq/blob/master/performance/README.md) to include ZSON read/write (#1974)
+
+
 ## v0.27.1
 * zq: Fix an issue where nested nulls caused a panic in CSV output (#1954)
 
diff --git a/compiler/expr.go b/compiler/expr.go
index be578fa04..f24d61d61 100644
--- a/compiler/expr.go
+++ b/compiler/expr.go
@@ -258,6 +258,48 @@ func compileCutter(zctx *resolver.Context, node ast.FunctionCall) (*expr.Cutter,
 	return expr.NewCutter(zctx, lhs, rhs)
 }
 
+func shaperOps(name string) expr.ShaperTransform {
+	switch name {
+	case "cast":
+		return expr.Cast
+	case "crop":
+		return expr.Crop
+	case "fill":
+		return expr.Fill
+	case "fit":
+		return expr.Crop | expr.Fill
+	case "order":
+		return expr.Order
+	case "shape":
+		return expr.Cast | expr.Crop | expr.Fill | expr.Order
+	default:
+		return 0
+	}
+
+}
+
+func isShaperFunc(name string) bool {
+	return shaperOps(name) != 0
+}
+
+func compileShaper(zctx *resolver.Context, node ast.FunctionCall) (*expr.Shaper, error) {
+	if len(node.Args) < 2 {
+		return nil, function.ErrTooFewArgs
+	}
+	if len(node.Args) > 2 {
+		return nil, function.ErrTooManyArgs
+	}
+	field, err := CompileExpr(zctx, node.Args[0])
+	if err != nil {
+		return nil, err
+	}
+	ev, err := CompileExpr(zctx, node.Args[1])
+	if err != nil {
+		return nil, err
+	}
+	return expr.NewShaper(zctx, field, ev, shaperOps(node.Function))
+}
+
 func compileCall(zctx *resolver.Context, node ast.FunctionCall) (expr.Evaluator, error) {
 	// For now, we special case cut and pick here.  We shuold generalize this
 	// as we will add many more stateful functions and also resolve this
@@ -274,6 +316,9 @@ func compileCall(zctx *resolver.Context, node ast.FunctionCall) (expr.Evaluator,
 	if node.Function == "pick" {
 		return compileCutter(zctx, node)
 	}
+	if isShaperFunc(node.Function) {
+		return compileShaper(zctx, node)
+	}
 	nargs := len(node.Args)
 	fn, err := function.New(zctx, node.Function, nargs)
 	if err != nil {
diff --git a/driver/parallel.go b/driver/parallel.go
index f7f4ca08a..2fbb8e20d 100644
--- a/driver/parallel.go
+++ b/driver/parallel.go
@@ -13,6 +13,7 @@ import (
 	"github.com/brimsec/zq/ppl/zqd/recruiter"
 	"github.com/brimsec/zq/proc"
 	"github.com/brimsec/zq/zbuf"
+	"go.uber.org/zap"
 )
 
 type parallelHead struct {
@@ -95,6 +96,7 @@ func (ph *parallelHead) Done() {
 
 type parallelGroup struct {
 	filter     SourceFilter
+	logger     *zap.Logger
 	mcfg       MultiConfig
 	msrc       MultiSource
 	once       sync.Once
@@ -212,6 +214,7 @@ func createParallelGroup(pctx *proc.Context, filter *compiler.Filter, msrc Multi
 			Filter: filter,
 			Span:   mcfg.Span,
 		},
+		logger:     mcfg.Logger,
 		mcfg:       mcfg,
 		msrc:       msrc,
 		pctx:       pctx,
diff --git a/expr/shaper.go b/expr/shaper.go
new file mode 100644
index 000000000..2e4548b55
--- /dev/null
+++ b/expr/shaper.go
@@ -0,0 +1,419 @@
+package expr
+
+import (
+	"fmt"
+
+	"github.com/brimsec/zq/zcode"
+	"github.com/brimsec/zq/zng"
+	"github.com/brimsec/zq/zng/resolver"
+	"github.com/brimsec/zq/zson"
+)
+
+// A ShaperTransform represents one of the different transforms that a
+// shaper can apply.
+type ShaperTransform int
+
+const (
+	Cast ShaperTransform = 1 << iota
+	Fill
+	Crop
+	Order
+)
+
+type step int
+
+const (
+	copyPrimitive step = iota // copy field fromIndex from input record
+	copyContainer
+	castPrimitive // cast field fromIndex from input record
+	null          // write null
+	record        // record into record below us
+)
+
+// A op is a recursive data structure encoding a series of
+// copy/cast steps to be carried out over an input record.
+type op struct {
+	op        step
+	fromIndex int
+	castTypes struct{ from, to zng.Type } // for op == castPrimitive
+	record    []op                        // for op == record
+}
+
+func (s *op) append(step op) {
+	s.record = append(s.record, step)
+}
+
+// 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
+// (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
+// step is inserted.
+func createOp(in, out *zng.TypeRecord) (op, error) {
+	o := op{op: record}
+	for _, outCol := range out.Columns {
+		ind, ok := in.ColumnOfField(outCol.Name)
+		if !ok {
+			o.append(op{op: null})
+			continue
+		}
+
+		inCol := in.Columns[ind]
+
+		switch {
+		case inCol.Type.ID() == outCol.Type.ID():
+			if zng.IsContainerType(inCol.Type) {
+				o.append(op{fromIndex: ind, op: copyContainer})
+			} else {
+				o.append(op{fromIndex: ind, op: copyPrimitive})
+			}
+		case zng.IsRecordType(inCol.Type) && zng.IsRecordType(outCol.Type):
+			step, err := createOp(inCol.Type.(*zng.TypeRecord), outCol.Type.(*zng.TypeRecord))
+			if err != nil {
+				return op{}, err
+			}
+			step.fromIndex = ind
+			o.append(step)
+		case zng.IsPrimitiveType(inCol.Type) && zng.IsPrimitiveType(outCol.Type):
+			step := op{fromIndex: ind, op: castPrimitive, castTypes: struct{ from, to zng.Type }{inCol.Type, outCol.Type}}
+			o.append(step)
+		default:
+			return op{}, fmt.Errorf("createOp incompatible column types %s and %s\n", inCol.Type, outCol.Type)
+		}
+	}
+	return o, nil
+}
+
+func (s op) castPrimitive(in zcode.Bytes, b *zcode.Builder) {
+	pc := LookupPrimitiveCaster(s.castTypes.to)
+	v, err := pc(zng.Value{s.castTypes.from, in})
+	if err != nil {
+		b.AppendNull()
+		return
+	}
+	b.AppendPrimitive(v.Bytes)
+}
+
+func (s op) buildRecord(in zcode.Bytes, b *zcode.Builder) {
+	if s.op != record {
+		panic("bad op")
+	}
+	for _, step := range s.record {
+
+		switch step.op {
+		case null:
+			b.AppendNull()
+			continue
+		}
+		// Using getNthFromContainer means we iterate from the
+		// beginning of the record for each field. An
+		// optimization (for shapes that don't require field
+		// reordering) would be make direct use of a
+		// zcode.Iter along with keeping track of our
+		// position.
+		bytes, err := getNthFromContainer(in, uint(step.fromIndex))
+		if err != nil {
+			panic(err)
+		}
+
+		switch step.op {
+		case copyPrimitive:
+			b.AppendPrimitive(bytes)
+		case copyContainer:
+			b.AppendContainer(bytes)
+		case castPrimitive:
+			step.castPrimitive(bytes, b)
+		case record:
+			step.buildRecord(bytes, b)
+		}
+	}
+}
+
+// A shapeSpec is a per-input type ID "spec" that contains the output
+// type and the op to create an output record.
+type shapeSpec struct {
+	typ *zng.TypeRecord
+	op  op
+}
+
+type Shaper struct {
+	zctx       *resolver.Context
+	b          zcode.Builder
+	fieldExpr  Evaluator
+	typ        *zng.TypeRecord
+	shapeSpecs map[int]shapeSpec // map from type ID to shapeSpec
+	transforms ShaperTransform
+}
+
+// NewShaper returns a shaper that will shape the result of fieldExpr
+// to the provided typExpr. (typExpr should evaluate to a type value,
+// e.g. a value of type TypeType).
+func NewShaper(zctx *resolver.Context, fieldExpr, typExpr Evaluator, tf ShaperTransform) (*Shaper, error) {
+	lit, ok := typExpr.(*Literal)
+	if !ok {
+		return nil, fmt.Errorf("shaping functions (crop, fill, cast, order) take a literal as second parameter")
+	}
+
+	// Z doesn't yet have type value literals, so we accept a
+	// string literal and parse it. When type value lits are in,
+	// this only change will be to require zng.TypeType rather
+	// than zng.TypeString. Since a TypeType value holds the type
+	// as zson, parsing with the zson.TypeTable below will be
+	// identical.
+	//
+	// if lit.zv.Type != zng.TypeType {
+	if lit.zv.Type != zng.TypeString {
+		return nil, fmt.Errorf("shaper needs a type value as second parameter")
+	}
+	tt := zson.NewTypeTable(zctx)
+	shapeToType, err := tt.LookupType(string(lit.zv.Bytes))
+	if err != nil {
+		return nil, fmt.Errorf("shaper could not parse type value literal: %s", err)
+	}
+
+	recType, isRecord := shapeToType.(*zng.TypeRecord)
+	if !isRecord {
+		return nil, fmt.Errorf("shaper needs a record type value as second parameter")
+	}
+	return &Shaper{
+		zctx:       zctx,
+		fieldExpr:  fieldExpr,
+		typ:        recType,
+		shapeSpecs: make(map[int]shapeSpec),
+		transforms: tf,
+	}, nil
+}
+
+func (c *Shaper) Eval(in *zng.Record) (zng.Value, error) {
+	inVal, err := c.fieldExpr.Eval(in)
+	if err != nil {
+		return zng.Value{}, err
+	}
+	inType, ok := inVal.Type.(*zng.TypeRecord)
+	if !ok {
+		return inVal, nil
+	}
+	if _, ok := c.shapeSpecs[in.Type.ID()]; !ok {
+		spec, err := c.createShapeSpec(inType, c.typ)
+		if err != nil {
+			return zng.Value{}, err
+		}
+		c.shapeSpecs[in.Type.ID()] = spec
+	}
+	spec := c.shapeSpecs[in.Type.ID()]
+	if spec.typ.ID() == in.Type.ID() {
+		return inVal, nil
+	}
+
+	c.b.Reset()
+	spec.op.buildRecord(inVal.Bytes, &c.b)
+	return zng.Value{spec.typ, c.b.Bytes()}, nil
+}
+
+func (c *Shaper) createShapeSpec(inType, spec *zng.TypeRecord) (shapeSpec, error) {
+	var err error
+	typ := inType
+	if c.transforms&Cast > 0 {
+		typ, err = c.castRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Crop > 0 {
+		typ, err = c.cropRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Fill > 0 {
+		typ, err = c.fillRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	if c.transforms&Order > 0 {
+		typ, err = c.orderRecordType(typ, spec)
+		if err != nil {
+			return shapeSpec{}, err
+		}
+	}
+	op, err := createOp(inType, typ)
+	return shapeSpec{typ, op}, err
+}
+
+// cropRecordType applies a crop (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) cropRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+	for _, inCol := range input.Columns {
+		ind, ok := spec.ColumnOfField(inCol.Name)
+		if !ok {
+			// 1. Field not in crop: drop.
+			continue
+		}
+
+		specCol := spec.Columns[ind]
+		switch {
+		case !zng.IsRecordType(inCol.Type):
+			// 2. Field is non-record in input: keep (regardless of crop record-ness)
+			cols = append(cols, inCol)
+		case zng.IsRecordType(specCol.Type):
+			// 3. Both records: recurse
+			out, err := c.cropRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord))
+			if err != nil {
+				return nil, err
+			}
+			cols = append(cols, zng.Column{inCol.Name, out})
+		default:
+			// 4. record input but non-record in crop: keep crop
+			cols = append(cols, specCol)
+
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// orderRecordType applies a field order (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) orderRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+	// Simple order algorithm creates a list with all specified
+	// 'order' fields present in input, followed by any other
+	// fields that were in input but not specified in order. Two
+	// examples:
+	//
+	// 1. (a b c d) order (c e b a) -> (c b a d)
+	//
+	// 2. (a b c d) order (d e c b) -> (d c b a)
+	//
+	// The second example with 'a' moving to the back suggests
+	// that we may want to use a different algorithm where
+	// unspecified fields "stay where they were". Specifically we
+	// might prefer that the result be (a d c b). We will learn
+	// with use, so starting with simpler algorithm for now.
+	//
+	for _, specCol := range spec.Columns {
+		if ind, ok := input.ColumnOfField(specCol.Name); ok {
+			inCol := input.Columns[ind]
+			if zng.IsRecordType(inCol.Type) && zng.IsRecordType(specCol.Type) {
+				if nested, err := c.orderRecordType(inCol.Type.(*zng.TypeRecord), specCol.Type.(*zng.TypeRecord)); err != nil {
+					return nil, err
+				} else {
+					cols = append(cols, zng.Column{specCol.Name, nested})
+				}
+				continue
+			}
+			cols = append(cols, inCol)
+		}
+	}
+	for _, inCol := range input.Columns {
+		if !spec.HasField(inCol.Name) {
+			cols = append(cols, inCol)
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// 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 = append(cols, zng.Column{inCol.Name, col})
+
+		case !inIsRec && !specIsRec:
+			cols = append(cols, inCol)
+
+		case inIsRec && !specIsRec:
+			cols = append(cols, inCol)
+
+		case !inIsRec && specIsRec:
+			cols = append(cols, specCol)
+
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
+
+// castRecordType applies a cast (as specified by the record type 'spec')
+// to a record type and returns the resulting record type.
+func (c *Shaper) castRecordType(input, spec *zng.TypeRecord) (*zng.TypeRecord, error) {
+	cols := make([]zng.Column, 0)
+
+	for _, inCol := range input.Columns {
+		// For each input column, check if we have a matching
+		// name in the cast spec.
+		ind, ok := spec.ColumnOfField(inCol.Name)
+		if !ok {
+			// 1. No match: output type unmodified.
+			cols = append(cols, inCol)
+			continue
+		}
+		specCol := spec.Columns[ind]
+
+		inRec, inIsRec := inCol.Type.(*zng.TypeRecord)
+		castRec, castIsRec := specCol.Type.(*zng.TypeRecord)
+
+		switch {
+		case inCol.Type.ID() == specCol.Type.ID():
+			// 2. Field has same type in cast: output type unmodified.
+			cols = append(cols, inCol)
+		case inIsRec && castIsRec:
+			// 3. Matching field is a record: recurse.
+			out, err := c.castRecordType(inRec, castRec)
+			if err != nil {
+				return nil, err
+			}
+			cols = append(cols, zng.Column{inCol.Name, out})
+		case zng.IsPrimitiveType(inCol.Type) && zng.IsPrimitiveType(specCol.Type):
+			// 4. Matching field is a primitive: output type is cast type.
+			if LookupPrimitiveCaster(specCol.Type) == nil {
+				return nil, fmt.Errorf("cast to %s not implemented", specCol.Type)
+			}
+			cols = append(cols, zng.Column{inCol.Name, specCol.Type})
+		default:
+			// 5. Non-castable type pair with at least one
+			// (non-record) container: output column is left
+			// unchanged.  Note that eventually, we should
+			// recognize and cast e.g. array[string] to array[ip].
+			// xxx before merge: file issue for this and mention
+			// it here.
+			cols = append(cols, inCol)
+		}
+	}
+	return c.zctx.LookupTypeRecord(cols)
+}
diff --git a/package.json b/package.json
index b69d08a81..80fe1c42d 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "zq",
-  "version": "0.27.1-dev",
+  "version": "0.28.0-dev",
   "scripts": {
       "prepack": "node npm/build"
   },
@@ -13,6 +13,6 @@
   },
   "brimDependencies": {
       "suricataTag": "v5.0.3-brim1",
-      "zeekTag": "v3.2.1-brim8"
+      "zeekTag": "v3.2.1-brim9"
   }
 }
diff --git a/ppl/archive/chunk/chunk.go b/ppl/archive/chunk/chunk.go
index 05e294877..73cce744a 100644
--- a/ppl/archive/chunk/chunk.go
+++ b/ppl/archive/chunk/chunk.go
@@ -19,9 +19,9 @@ type FileKind string
 
 const (
 	FileKindUnknown  FileKind = ""
-	FileKindData              = "d"
-	FileKindMetadata          = "m"
-	FileKindSeek              = "ts"
+	FileKindData     FileKind = "d"
+	FileKindMetadata FileKind = "m"
+	FileKindSeek     FileKind = "ts"
 )
 
 var fileRegex = regexp.MustCompile(`(d|m)-([0-9A-Za-z]{27}).zng$`)
diff --git a/ppl/zqd/handlers.go b/ppl/zqd/handlers.go
index 894059170..718a1d8b0 100644
--- a/ppl/zqd/handlers.go
+++ b/ppl/zqd/handlers.go
@@ -96,7 +96,7 @@ func handleSearch(c *Core, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	srch, err := search.NewSearchOp(req)
+	srch, err := search.NewSearchOp(req, c.requestLogger(r))
 	if err != nil {
 		respondError(c, w, r, err)
 		return
diff --git a/ppl/zqd/search/search.go b/ppl/zqd/search/search.go
index 8b4aae877..4367f6ded 100644
--- a/ppl/zqd/search/search.go
+++ b/ppl/zqd/search/search.go
@@ -36,11 +36,12 @@ const (
 )
 
 type SearchOp struct {
+	logger  *zap.Logger
 	query   *Query
 	workers int // for distributed queries only
 }
 
-func NewSearchOp(req api.SearchRequest) (*SearchOp, error) {
+func NewSearchOp(req api.SearchRequest, logger *zap.Logger) (*SearchOp, error) {
 	if req.Span.Ts < 0 {
 		return nil, zqe.ErrInvalid("time span must have non-negative timestamp")
 	}
@@ -59,7 +60,10 @@ func NewSearchOp(req api.SearchRequest) (*SearchOp, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &SearchOp{query: query}, nil
+	return &SearchOp{
+		logger: logger,
+		query:  query,
+	}, nil
 }
 
 func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output) (err error) {
@@ -83,9 +87,10 @@ func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output
 	switch st := store.(type) {
 	case *archivestore.Storage:
 		return driver.MultiRun(ctx, d, s.query.Proc, zctx, st.MultiSource(), driver.MultiConfig{
+			Logger:    s.logger,
+			Order:     zbuf.OrderDesc,
 			Span:      s.query.Span,
 			StatsTick: statsTicker.C,
-			Order:     zbuf.OrderDesc,
 		})
 	case *filestore.Storage:
 		rc, err := st.Open(ctx, zctx, s.query.Span)
@@ -95,6 +100,7 @@ func (s *SearchOp) Run(ctx context.Context, store storage.Storage, output Output
 		defer rc.Close()
 
 		return driver.Run(ctx, d, s.query.Proc, zctx, rc, driver.Config{
+			Logger:            s.logger,
 			ReaderSortKey:     "ts",
 			ReaderSortReverse: true,
 			Span:              s.query.Span,
diff --git a/ppl/zqd/search/worker.go b/ppl/zqd/search/worker.go
index 4e69085b9..e55353bf6 100644
--- a/ppl/zqd/search/worker.go
+++ b/ppl/zqd/search/worker.go
@@ -12,16 +12,18 @@ import (
 	"github.com/brimsec/zq/ppl/zqd/storage/archivestore"
 	"github.com/brimsec/zq/zng/resolver"
 	"github.com/brimsec/zq/zqe"
+	"go.uber.org/zap"
 )
 
 type WorkerOp struct {
-	proc  ast.Proc
-	span  nano.Span
-	src   driver.Source
-	store *archivestore.Storage
+	logger *zap.Logger
+	proc   ast.Proc
+	span   nano.Span
+	src    driver.Source
+	store  *archivestore.Storage
 }
 
-func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Storage) (*WorkerOp, error) {
+func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Storage, logger *zap.Logger) (*WorkerOp, error) {
 	// XXX zqd only supports backwards searches, remove once this has been
 	// fixed.
 	if req.Dir == 1 {
@@ -42,7 +44,13 @@ func NewWorkerOp(ctx context.Context, req api.WorkerChunkRequest, st storage.Sto
 	if err != nil {
 		return nil, err
 	}
-	return &WorkerOp{proc: proc, span: req.Span, src: src, store: store}, nil
+	return &WorkerOp{
+		logger: logger,
+		proc:   proc,
+		span:   req.Span,
+		src:    src,
+		store:  store,
+	}, nil
 }
 
 func (w *WorkerOp) Run(ctx context.Context, output Output) (err error) {
@@ -64,6 +72,7 @@ func (w *WorkerOp) Run(ctx context.Context, output Output) (err error) {
 	zctx := resolver.NewContext()
 
 	return driver.MultiRun(ctx, d, w.proc, zctx, w.store.StaticSource(w.src), driver.MultiConfig{
+		Logger:    w.logger,
 		Span:      w.span,
 		StatsTick: statsTicker.C,
 	})
diff --git a/ppl/zqd/worker.go b/ppl/zqd/worker.go
index ccf0d9a32..8b3c41674 100644
--- a/ppl/zqd/worker.go
+++ b/ppl/zqd/worker.go
@@ -22,7 +22,7 @@ func handleWorkerRootSearch(c *Core, w http.ResponseWriter, r *http.Request) {
 		respondError(c, w, r, err)
 		return
 	}
-	srch, err := search.NewSearchOp(req.SearchRequest)
+	srch, err := search.NewSearchOp(req.SearchRequest, c.requestLogger(r))
 	if err != nil {
 		respondError(c, w, r, err)
 		return
@@ -56,7 +56,7 @@ func handleWorkerChunkSearch(c *Core, w http.ResponseWriter, httpReq *http.Reque
 		respondError(c, w, httpReq, err)
 		return
 	}
-	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(ark))
+	work, err := search.NewWorkerOp(ctx, req, archivestore.NewStorage(ark), c.requestLogger(httpReq))
 	if err != nil {
 		respondError(c, w, httpReq, err)
 		return
diff --git a/zng/type.go b/zng/type.go
index b14b08e1e..b8a898f49 100644
--- a/zng/type.go
+++ b/zng/type.go
@@ -337,6 +337,11 @@ func IsUnionType(typ Type) bool {
 	return ok
 }
 
+func IsRecordType(typ Type) bool {
+	_, ok := AliasedType(typ).(*TypeRecord)
+	return ok
+}
+
 func IsContainerType(typ Type) bool {
 	switch typ := typ.(type) {
 	case *TypeAlias:
@@ -348,6 +353,10 @@ func IsContainerType(typ Type) bool {
 	}
 }
 
+func IsPrimitiveType(typ Type) bool {
+	return !IsContainerType(typ)
+}
+
 func AliasTypes(typ Type) []*TypeAlias {
 	var aliases []*TypeAlias
 	switch typ := typ.(type) {


Test patch:
diff --git a/expr/ztests/shape-cast.yaml b/expr/ztests/shape-cast.yaml
new file mode 100644
index 000000000..03ce6b1e6
--- /dev/null
+++ b/expr/ztests/shape-cast.yaml
@@ -0,0 +1,18 @@
+zql: |
+   put
+   // cast to type with same field order
+   id=cast(id, "({orig_h:ip,orig_p:port=(uint16),resp_h:ip,resp_p:port})"),
+   // cast to type with different field order, does not change output order
+   id2=cast(id, "({resp_h:ip,resp_p:port=(uint16),orig_h:ip,orig_p:port})")
+
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:string],other:float64]
+   0:[[ff02::fb;5353;5353;1.2.3.4;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:ip],other:float64,id2:record[orig_h:ip,orig_p:port,resp_p:port,resp_h:ip]]
+   0:[[ff02::fb;5353;5353;1.2.3.4;]123;[ff02::fb;5353;5353;1.2.3.4;]]
+
diff --git a/expr/ztests/shape-crop.yaml b/expr/ztests/shape-crop.yaml
new file mode 100644
index 000000000..da12224f4
--- /dev/null
+++ b/expr/ztests/shape-crop.yaml
@@ -0,0 +1,17 @@
+zql: |
+   put
+   // crop to type with same field order
+   id=crop(id, "({orig_h:ip,orig_p:port=(uint16)})"),
+   // crop to type with different field order, does not change output order
+   id2=crop(id, "({orig_p:port=(uint16),orig_h:ip})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,resp_h:ip,resp_p:port],other:float64]
+   0:[[ff02::fb;5353;1.2.3.4;5353;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port],other:float64,id2:record[orig_h:ip,orig_p:port]]
+   0:[[ff02::fb;5353;]123;[ff02::fb;5353;]]
+
diff --git a/expr/ztests/shape-fill-fit.yaml b/expr/ztests/shape-fill-fit.yaml
new file mode 100644
index 000000000..a5f39db22
--- /dev/null
+++ b/expr/ztests/shape-fill-fit.yaml
@@ -0,0 +1,13 @@
+zql: |
+   cut id_filled=fill(id, "({orig_h:ip,orig_p:port=(uint16)})"),  id_fitted=fit(id, "({orig_h:ip,orig_p:port=(uint16)})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,resp_h:string]]
+   0:[[ff02::fb;ff02::fb;]]
+
+output: |
+   #port=uint16
+   #0:record[id_filled:record[orig_h:string,resp_h:string,orig_p:port],id_fitted:record[orig_h:string,orig_p:port]]
+   0:[[ff02::fb;ff02::fb;-;][ff02::fb;-;]]
+
diff --git a/expr/ztests/shape-order.yaml b/expr/ztests/shape-order.yaml
new file mode 100644
index 000000000..c961fb933
--- /dev/null
+++ b/expr/ztests/shape-order.yaml
@@ -0,0 +1,13 @@
+zql: |
+   put id=order(id, "({orig_h: string, orig_p: port=(uint16), resp_h:ip,resp_p:port=(uint16)})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,orig_p:port,resp_p:port,resp_h:string],other:float64]
+   0:[[ff02::fb;5353;5354;1.2.3.4;]123;]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,orig_p:port,resp_h:string,resp_p:port],other:float64]
+   0:[[ff02::fb;5353;1.2.3.4;5354;]123;]
+
diff --git a/expr/ztests/shape.yaml b/expr/ztests/shape.yaml
new file mode 100644
index 000000000..4543b38d2
--- /dev/null
+++ b/expr/ztests/shape.yaml
@@ -0,0 +1,13 @@
+zql: |
+   put id=shape(id, "({orig_h: ip, orig_p: port=(uint16), vlan: uint16, resp_h:ip,resp_p:port})")
+
+input: |
+   #port=uint16
+   #0:record[id:record[orig_h:string,resp_h:string,orig_p:port,resp_p:port,tag:uint16]]
+   0:[[1.2.3.4;5.6.7.8;1234;53;0;]]
+
+output: |
+   #port=uint16
+   #0:record[id:record[orig_h:ip,orig_p:port,vlan:uint16,resp_h:ip,resp_p:port]]
+   0:[[1.2.3.4;1234;-;5.6.7.8;53;]]
+
diff --git a/ppl/archive/archive_test.go b/ppl/archive/archive_test.go
index 03ce2a5ea..f01d2b21f 100644
--- a/ppl/archive/archive_test.go
+++ b/ppl/archive/archive_test.go
@@ -151,24 +151,19 @@ func TestOpenOptions(t *testing.T) {
 }
 
 func TestSeekIndex(t *testing.T) {
-	datapath, err := ioutil.TempDir("", "")
-	require.NoError(t, err)
-	defer os.RemoveAll(datapath)
+	datapath := t.TempDir()
 
 	orig := ImportStreamRecordsMax
 	ImportStreamRecordsMax = 1
 	defer func() {
 		ImportStreamRecordsMax = orig
 	}()
-	createArchiveSpace(t, datapath, babble, &CreateOptions{
-		// Must use SortAscending: true until zq#1329 is addressed.
-		SortAscending: true,
-	})
-	_, err = OpenArchive(datapath, &OpenOptions{})
+	createArchiveSpace(t, datapath, babble, nil)
+	_, err := OpenArchive(datapath, &OpenOptions{})
 	require.NoError(t, err)
 
-	first1 := nano.Ts(1587508830068523240)
-	var idxUri iosrc.URI
+	first1 := nano.Ts(1587513592062544400)
+	var idxURI iosrc.URI
 	err = filepath.Walk(datapath, func(p string, fi os.FileInfo, err error) error {
 		if err != nil {
 			return err
@@ -179,23 +174,24 @@ func TestSeekIndex(t *testing.T) {
 				return err
 			}
 			uri.Path = path.Dir(uri.Path)
-			chunk, err := chunk.Open(context.Background(), uri, id, zbuf.OrderAsc)
+			chunk, err := chunk.Open(context.Background(), uri, id, zbuf.OrderDesc)
 			if err != nil {
 				return err
 			}
 			if chunk.First == first1 {
-				idxUri = chunk.SeekIndexPath()
+				idxURI = chunk.SeekIndexPath()
 			}
 		}
 		return nil
 	})
 	require.NoError(t, err)
-	finder, err := microindex.NewFinder(context.Background(), resolver.NewContext(), idxUri)
+	finder, err := microindex.NewFinder(context.Background(), resolver.NewContext(), idxURI)
 	require.NoError(t, err)
 	keys, err := finder.ParseKeys("1587508851")
 	require.NoError(t, err)
 	rec, err := finder.ClosestLTE(keys)
 	require.NoError(t, err)
+	require.NoError(t, finder.Close())
 
 	var buf bytes.Buffer
 	w := tzngio.NewWriter(zio.NopCloser(&buf))
@@ -203,7 +199,7 @@ func TestSeekIndex(t *testing.T) {
 
 	exp := `
 #0:record[ts:time,offset:int64]
-0:[1587508850.06466032;202;]
+0:[1587508850.06466032;23795;]
 `
 	require.Equal(t, test.Trim(exp), buf.String())
 }