Fix record import field mapping & validation

* Allow undefined record ref validator fn
* Remove field mapping fallback if mapping is provided

pkg/envoy/resource/compose_record_shaper.go

@@ -52,7 +52,7 @@ func (crt *composeRecordShaper) toResource(def *ComposeRecordTemplate, dt *Resou
 			// Get the bits in order
 			rRaw := &ComposeRecordRaw{
 				ID:     crt.getKey(mr, def.Key),
-				Values: crt.mapValues(mr, def.FieldMap),
+				Values: crt.mapValues(mr, def.FieldMap, def.Defaultable),
 			}
 
 			crt.getTimestamps(rRaw)
@@ -70,7 +70,7 @@ func (crt *composeRecordShaper) toResource(def *ComposeRecordTemplate, dt *Resou
 }
 
 // mapValues maps original values based on the provided mapping
-func (crt *composeRecordShaper) mapValues(ov map[string]string, fm MappingTplSet) map[string]string {
+func (crt *composeRecordShaper) mapValues(ov map[string]string, fm MappingTplSet, defaultable bool) map[string]string {
 	nv := make(map[string]string)
 
 	// Mappings are provided as a slice since we'll want to do some additional conditional mappings.
@@ -80,13 +80,27 @@ func (crt *composeRecordShaper) mapValues(ov map[string]string, fm MappingTplSet
 		mx[m.Cell] = m
 	}
 
-	for k, v := range ov {
-		if m, has := mx[k]; has {
-			if !m.IsIgnored() {
-				nv[m.Field] = v
+	if defaultable {
+		for k, v := range ov {
+			if m, has := mx[k]; has {
+				if !m.IsIgnored() {
+					nv[m.Field] = v
+				}
+			} else {
+				nv[k] = v
+			}
+		}
+	} else {
+		if len(mx) > 0 {
+			for k, v := range ov {
+				if m, has := mx[k]; has && !m.IsIgnored() {
+					nv[m.Field] = v
+				}
 			}
 		} else {
-			nv[k] = v
+			for k, v := range ov {
+				nv[k] = v
+			}
 		}
 	}
 

pkg/envoy/resource/compose_record_template.go

@@ -27,19 +27,21 @@ type (
 		// Name is the source name; topically file name
 		Name string
 		// Key determines what defines an identifier
-		Key      []string
-		FieldMap MappingTplSet
+		Key         []string
+		FieldMap    MappingTplSet
+		Defaultable bool
 	}
 )
 
 // NewComposeRecordTemplate returns a record template based on the provided parameters.
 //
 // The template is only valid up until the resource shaping; it is not allowed after the fact.
-func NewComposeRecordTemplate(modRef, nsRef, name string, fieldMap MappingTplSet, key ...string) *ComposeRecordTemplate {
+func NewComposeRecordTemplate(modRef, nsRef, name string, defaultable bool, fieldMap MappingTplSet, key ...string) *ComposeRecordTemplate {
 	r := &ComposeRecordTemplate{base: &base{}}
 	r.Name = name
 	r.Key = key
 	r.FieldMap = fieldMap
+	r.Defaultable = defaultable
 
 	r.NsRef = r.AddRef(COMPOSE_NAMESPACE_RESOURCE_TYPE, nsRef)
 	r.ModRef = r.AddRef(COMPOSE_MODULE_RESOURCE_TYPE, modRef).Constraint(r.NsRef)

pkg/envoy/yaml/compose_module.go

@@ -14,8 +14,9 @@ type (
 	composeRecordTpl struct {
 		Source string `yaml:"from"`
 
-		Key     []string
-		Mapping mappingTplSet
+		Key         []string
+		Mapping     mappingTplSet
+		Defaultable bool
 	}
 
 	composeModule struct {

pkg/envoy/yaml/compose_module_unmarshal.go

@@ -134,7 +134,7 @@ func (wrap composeModule) MarshalEnvoy() ([]resource.Interface, error) {
 		for _, m := range s.Mapping {
 			mtt = append(mtt, &m.MappingTpl)
 		}
-		crs = resource.NewComposeRecordTemplate(rs.Identifiers().First(), wrap.refNamespace, s.Source, mtt, s.Key...)
+		crs = resource.NewComposeRecordTemplate(rs.Identifiers().First(), wrap.refNamespace, s.Source, s.Defaultable, mtt, s.Key...)
 	}
 
 	return envoy.CollectNodes(
@@ -161,6 +161,9 @@ func (rt *composeRecordTpl) UnmarshalYAML(n *yaml.Node) error {
 				})
 			}
 
+		case "defaultable":
+			return y7s.DecodeScalar(v, "defaultable field mapping", &rt.Defaultable)
+
 		case "mapping", "map":
 			rt.Mapping = make(mappingTplSet, 0, 20)
 			// When provided as a sequence node, map the fields based on the index.