add defuse function

book/src/SUMMARY.md

@@ -155,6 +155,7 @@
             - [strftime](super-sql/functions/time/strftime.md)
         - [Types](super-sql/functions/types/intro.md)
             - [cast](super-sql/functions/types/cast.md)
+            - [defuse](super-sql/functions/types/defuse.md)
             - [is](super-sql/functions/types/is.md)
             - [kind](super-sql/functions/types/kind.md)
             - [nameof](super-sql/functions/types/nameof.md)

book/src/super-sql/functions/types/defuse.md

@@ -0,0 +1,53 @@
+# defuse
+
+undo the effects of fuse
+
+## Synopsis
+
+```
+defuse(val any) -> any
+```
+
+## Description
+
+The `defuse` function converts a value `val` from a fused representation
+to a plain representation by
+* replacing any value nested within `val` that is a union type with its
+  underlying non-union value, and
+* eliminating all optional fields from record types within `val` by
+  replacing each optional field with value with its non-optional equivalent
+  and eliminating entirely any optional fields without a value.
+
+## Examples
+
+---
+
+_Remove union types_
+
+```mdtest-spq {data-layout="stacked"}
+# spq
+defuse(this)
+# input
+{a:1::(int64|string)}
+{a:"foo"::(int64|string)}
+# expected output
+{a:1}
+{a:"foo"}
+```
+
+---
+
+_Remove union types_
+
+```mdtest-spq {data-layout="stacked"}
+# spq
+fuse | defuse(this)
+# input
+{x:1}
+{x:2,y:3}
+{x:4,z?:_::int64}
+# expected output
+{x:1}
+{x:2,y:3}
+{x:4}
+```

compiler/semantic/op.go

@@ -371,7 +371,7 @@ func unmarshalHeaders(val super.Value) (map[string][]string, error) {
 	}
 	headers := map[string][]string{}
 	for i, f := range val.Fields() {
-		fieldVal := val.DerefByColumn(i)
+		fieldVal, _ := val.DerefByColumn(i)
 		if fieldVal == nil {
 			continue
 		}

runtime/sam/expr/function/cast.go

@@ -322,7 +322,13 @@ func (u *upcast) toRecord(from super.Value, to *super.TypeRecord) (super.Value,
 	b.BeginContainer()
 	for i, f := range to.Fields {
 		var val2 super.Value
-		if fieldVal := from.Deref(f.Name); fieldVal != nil {
+		fieldVal, none := from.DerefWithNone(f.Name)
+		if none {
+			nones = append(nones, optOff)
+			optOff++
+			continue
+		}
+		if fieldVal != nil {
 			val2 = u.Cast(*fieldVal, f.Type)
 			if f.Opt {
 				optOff++

runtime/sam/expr/function/defuse.go

@@ -0,0 +1,139 @@
+package function
+
+import (
+	"github.com/brimdata/super"
+	"github.com/brimdata/super/scode"
+)
+
+type defuse struct {
+	sctx *super.Context
+}
+
+func NewDefuse(sctx *super.Context) *defuse {
+	return &defuse{sctx}
+}
+
+func (d *defuse) Call(args []super.Value) super.Value {
+	return d.eval(args[0])
+}
+
+func (d *defuse) eval(in super.Value) super.Value {
+	switch typ := in.Type().(type) {
+	case *super.TypeRecord:
+		var fields []super.Field
+		var elems []super.Value
+		it := scode.NewRecordIter(in.Bytes(), typ.Opts)
+		for _, f := range typ.Fields {
+			bytes, none := it.Next(f.Opt)
+			if none {
+				continue
+			}
+			val := d.eval(super.NewValue(f.Type, bytes))
+			elems = append(elems, val)
+			fields = append(fields, super.NewField(f.Name, val.Type()))
+		}
+		var b scode.Builder
+		for _, e := range elems {
+			b.Append(e.Bytes())
+		}
+		return super.NewValue(d.sctx.MustLookupTypeRecord(fields), b.Bytes())
+	case *super.TypeArray:
+		elems := d.parseArrayOrSet(typ.Type, in.Bytes())
+		if len(elems) == 0 {
+			typ := d.sctx.LookupTypeArray(super.TypeNull)
+			return super.NewValue(typ, nil)
+		}
+		elemType, bytes := d.unify(elems)
+		return super.NewValue(d.sctx.LookupTypeArray(elemType), bytes)
+	case *super.TypeSet:
+		elems := d.parseArrayOrSet(typ.Type, in.Bytes())
+		if len(elems) == 0 {
+			typ := d.sctx.LookupTypeArray(super.TypeNull)
+			return super.NewValue(typ, nil)
+		}
+		elemType, bytes := d.unify(elems)
+		return super.NewValue(d.sctx.LookupTypeSet(elemType), bytes)
+	case *super.TypeMap:
+		var keys, vals []super.Value
+		for it := in.Bytes().Iter(); !it.Done(); {
+			keys = append(keys, super.NewValue(typ, it.Next()))
+			vals = append(vals, super.NewValue(typ, it.Next()))
+		}
+		keyType := d.unifyType(keys)
+		valType := d.unifyType(vals)
+		var b scode.Builder
+		for k, key := range keys {
+			if u, ok := keyType.(*super.TypeUnion); ok {
+				super.BuildUnion(&b, u.TagOf(u), key.Bytes())
+			} else {
+				b.Append(key.Bytes())
+			}
+			val := vals[k]
+			if u, ok := valType.(*super.TypeUnion); ok {
+				super.BuildUnion(&b, u.TagOf(u), val.Bytes())
+			} else {
+				b.Append(val.Bytes())
+			}
+		}
+		return super.NewValue(d.sctx.LookupTypeMap(keyType, valType), b.Bytes())
+	case *super.TypeUnion:
+		return d.eval(in.Deunion())
+	default:
+		// primitives, named types, enums
+		// BTW, named types are a barrier to defuse.
+		return in
+	}
+}
+
+func (d *defuse) parseArrayOrSet(typ super.Type, bytes scode.Bytes) []super.Value {
+	var elems []super.Value
+	for it := bytes.Iter(); !it.Done(); {
+		elems = append(elems, d.eval(super.NewValue(typ, it.Next())))
+	}
+	return elems
+}
+
+func (d *defuse) unify(elems []super.Value) (super.Type, scode.Bytes) {
+	seen := make(map[super.Type]struct{})
+	var types []super.Type
+	for _, e := range elems {
+		typ := e.Type()
+		if _, ok := seen[typ]; !ok {
+			seen[typ] = struct{}{}
+			types = append(types, typ)
+		}
+	}
+	if len(types) == 1 {
+		var b scode.Builder
+		for _, e := range elems {
+			b.Append(e.Bytes())
+		}
+		return types[0], b.Bytes()
+	}
+	var b scode.Builder
+	union := d.sctx.LookupTypeUnion(types)
+	for _, e := range elems {
+		super.BuildUnion(&b, union.TagOf(e.Type()), e.Bytes())
+	}
+	return union, b.Bytes()
+}
+
+func (d *defuse) unifyType(vals []super.Value) super.Type {
+	seen := make(map[super.Type]struct{})
+	var types []super.Type
+	for _, e := range vals {
+		typ := e.Type()
+		if _, ok := seen[typ]; !ok {
+			seen[typ] = struct{}{}
+			types = append(types, typ)
+		}
+	}
+	switch len(types) {
+	case 0:
+		return super.TypeNull // XXX should be TypeEmpty
+	case 1:
+		return types[0]
+	default:
+		return d.sctx.LookupTypeUnion(types)
+	}
+}

runtime/sam/expr/function/function.go

@@ -53,6 +53,8 @@ func New(sctx *super.Context, name string, narg int) (expr.Function, error) {
 		argmin = 2
 		argmax = 2
 		f = &DatePart{sctx}
+	case "defuse":
+		f = &defuse{sctx: sctx}
 	case "error":
 		f = &Error{sctx: sctx}
 	case "fields":

runtime/vam/expr/function/function.go

@@ -45,6 +45,8 @@ func New(sctx *super.Context, name string, narg int) (expr.Function, error) {
 		argmin = 2
 		argmax = 2
 		f = &DatePart{sctx}
+	case "defuse":
+		f = &samFunc{function.NewDefuse(sctx)}
 	case "error":
 		f = &Error{sctx}
 	case "fields":

runtime/ztests/expr/defuse.yaml

@@ -0,0 +1,21 @@
+spq: defuse(this)
+
+vector: true
+
+input: |
+  [1,2,"foo"]::[null|int64|string]
+  {x:[1,2]::[null|int64],y?:_::string,z:1::(int64|string)}
+
+output: |
+  [1,2,"foo"]
+  {x:[1,2],z:1}
+---
+
+spq: fuse | defuse(this)
+
+input: &input |
+  {x:1}
+  {x:2,y:3}
+
+output: *input
+

runtime/ztests/expr/function/upcast.yaml

@@ -6,3 +6,13 @@ input: |
 
 output: |
   [error({message:"cannot upcast to int8|string",on:1}),"a"::(int8|string)]
+
+---
+
+spq: upcast(this, <{y?:int64,z?:int64}>)
+
+input: |
+  {z?:_::int64}
+
+output: |
+  {y?:_::int64,z?:_::int64}

sio/tableio/writer.go

@@ -58,7 +58,8 @@ func (w *Writer) Write(r super.Value) error {
 	var out []string
 	for k, f := range r.Fields() {
 		var v string
-		value := r.DerefByColumn(k).MissingAsNull()
+		valPtr, _ := r.DerefByColumn(k)
+		value := valPtr.MissingAsNull()
 		if f.Type == super.TypeTime {
 			if !value.IsNull() {
 				v = super.DecodeTime(value.Bytes()).Time().Format(time.RFC3339Nano)

value.go

@@ -316,42 +316,42 @@ func (r Value) Walk(rv Visitor) error {
 	return Walk(r.Type(), r.Bytes(), rv)
 }
 
-func (r Value) nth(n int) (scode.Bytes, bool) {
+func (r Value) nth(n int) (scode.Bytes, bool, bool) {
 	if typ := TypeRecordOf(r.typ); typ != nil {
 		var elem scode.Bytes
 		var none bool
 		for i, it := 0, scode.NewRecordIter(r.Bytes(), typ.Opts); i <= n; i++ {
 			if it.Done() {
-				return nil, false
+				return nil, false, false
 			}
 			elem, none = it.Next(typ.Fields[i].Opt)
 		}
-		if none {
-			elem = nil
-		}
-		return elem, true
+		return elem, none, true
 	}
 	var zv scode.Bytes
 	for i, it := 0, r.Bytes().Iter(); i <= n; i++ {
 		if it.Done() {
-			return nil, false
+			return nil, false, false
 		}
 		zv = it.Next()
 	}
-	return zv, true
+	return zv, false, true
 }
 
 func (r Value) Fields() []Field {
 	return TypeRecordOf(r.Type()).Fields
 }
 
-func (v *Value) DerefByColumn(col int) *Value {
+func (v *Value) DerefByColumn(col int) (*Value, bool) {
 	if v != nil {
-		if bytes, ok := v.nth(col); ok {
-			return NewValue(v.Fields()[col].Type, bytes).Ptr()
+		if bytes, none, ok := v.nth(col); ok {
+			if none {
+				return nil, true
+			}
+			return NewValue(v.Fields()[col].Type, bytes).Ptr(), false
 		}
 	}
-	return nil
+	return nil, false
 }
 
 func (v Value) IndexOfField(field string) (int, bool) {
@@ -369,6 +369,18 @@ func (v *Value) Deref(field string) *Value {
 	if !ok {
 		return nil
 	}
+	val, _ := v.DerefByColumn(i)
+	return val
+}
+
+func (v *Value) DerefWithNone(field string) (*Value, bool) {
+	if v == nil {
+		return nil, false
+	}
+	i, ok := v.IndexOfField(field)
+	if !ok {
+		return nil, false
+	}
 	return v.DerefByColumn(i)
 }