remove some unnecessary fusion wrappers (#6760)

This commit makes record fusion a bit lighter by recognizing that
fusion wrappers are needed only when field list changes name/order/etc
as any type variation in descendent types will have their own fusion
wrappers and defuse will recursively unravel those.

Author: mccanne

book/src/super-sql/aggregates/fuse.md

@@ -25,5 +25,5 @@ fuse(this)
 {a:1,b:2}
 {a:2,b:"foo"}
 # expected output
-<fusion({a:int64,b:fusion(int64|string)})>
+<{a:int64,b:fusion(int64|string)}>
 ```

book/src/super-sql/operators/fuse.md

@@ -50,8 +50,8 @@ fuse
 {a:1}
 {a:"foo"}
 # expected output
-fusion({a:fusion(1::(int64|string),<int64>)},<{a:int64}>)
-fusion({a:fusion("foo"::(int64|string),<string>)},<{a:string}>)
+{a:fusion(1::(int64|string),<int64>)}
+{a:fusion("foo"::(int64|string),<string>)}
 ```
 
 ---

runtime/sam/expr/agg/fuser.go

@@ -71,7 +71,11 @@ func (f *Fuser) fuse(a, b super.Type) super.Type {
 					fields = append(fields, super.NewFieldWithOpt(field.Name, field.Type, true))
 				}
 			}
-			return f.fusion(f.sctx.MustLookupTypeRecord(fields))
+			fusedRec := f.sctx.MustLookupTypeRecord(fields)
+			if recChanged(a, fusedRec) || recChanged(b, fusedRec) {
+				return f.fusion(fusedRec)
+			}
+			return fusedRec
 		}
 	case *super.TypeArray:
 		if b, ok := b.(*super.TypeArray); ok {
@@ -216,3 +220,21 @@ func indexOfField(fields []super.Field, name string) (int, bool) {
 	}
 	return -1, false
 }
+
+// recChanged returns true iff the two record types are different
+// enough after fusing that they need to be wrapped in a fusion type.
+// As long as all the fields names and optionality are the same, then
+// any type differences in the fused type of the child fields will be
+// captured by a fusion wrapper somewhere in the descendent type.
+func recChanged(a, b *super.TypeRecord) bool {
+	if len(a.Fields) != len(b.Fields) {
+		return true
+	}
+	for k, af := range a.Fields {
+		bf := b.Fields[k]
+		if af.Name != bf.Name || af.Opt != bf.Opt {
+			return true
+		}
+	}
+	return false
+}

runtime/ztests/op/aggregate/fuse.yaml

@@ -14,3 +14,17 @@ output: |
   {key:"a",supertype:<fusion({a?:string,r:fusion({x?:int32,y:int32,z?:int32}),key:string,s?:string,r2?:{x:int32}})>}
   {key:"b",supertype:<fusion({a?:string,r:fusion({x?:int32,y:int32,z?:int32}),key:string,s?:string,r2?:{x:int32}})>}
   {key:"c",supertype:<fusion({a?:string,r:fusion({x?:int32,y:int32,z?:int32}),key:string,s?:string,r2?:{x:int32}})>}
+
+---
+
+spq: fuse(this)
+
+input: |
+  {fixed:{x:1},var:{s:"foo"}}
+  {fixed:{x:2},var:{s:"bar",a:[1,2]}}
+  {fixed:{x:3},var:{s:"bar",a:[3]}}
+  {fixed:{x:4},var:{s:"foo",t:123}}
+  {fixed:{x:5},var:{s:"foo",t:{x:1}}}
+
+output: |
+  <{fixed:{x:int64},var:fusion({s:string,a?:[int64],t?:fusion(int64|{x:int64})})}>