Emit column level metadata in catalog and consume it back in

We want to allow users to configure which columns they ingest or don't
ingest in and out of their incident data.

This PR creates column level metadata for each property in a schema (it
doesn't do nested properties).

This lets us say yay or nay on a column by inspecting the metadata and
checking the typical "selected" or "selected-by-default" properties we
have there.

Author: rliddler

tap/catalog.go

@@ -2,6 +2,8 @@ package tap
 
 import (
 	"github.com/incident-io/singer-tap/model"
+	"github.com/pkg/errors"
+	"github.com/samber/lo"
 )
 
 // A catalog can contain several streams or "entries"
@@ -58,6 +60,40 @@ func (c *Catalog) GetEnabledStreams() []CatalogEntry {
 	return enabledStreams
 }
 
+func (c *Catalog) GetDisabledFields(streamName string) (map[string]bool, error) {
+	// Just something to enable quick lookups of fields by name
+	var disabledField = map[string]bool{}
+
+	// For the given stream, get the enabled fields
+	catalogEntry, found := lo.Find(c.Streams, func(stream CatalogEntry) bool { return stream.Stream == streamName })
+	if !found {
+		return nil, errors.Errorf("stream %s not found", streamName)
+	}
+
+	// For this catalog entry, get the metadata, and build a list of all the enabled fields
+	for _, metadata := range *catalogEntry.Metadata {
+		// Ignore the top level metadata
+		if len(metadata.Breadcrumb) == 0 {
+			continue
+		}
+
+		// Check if the metadata has the user input "selected" bool
+		if metadata.Metadata.Selected != nil {
+			// If so, check its set to false!
+			if !*metadata.Metadata.Selected {
+				disabledField[metadata.Breadcrumb[len(metadata.Breadcrumb)-1]] = true
+			}
+		} else {
+			// There's no selected key, so check if WE have set the selected by default
+			if !metadata.Metadata.SelectedByDefault {
+				disabledField[metadata.Breadcrumb[len(metadata.Breadcrumb)-1]] = true
+			}
+		}
+	}
+
+	return disabledField, nil
+}
+
 func NewDefaultCatalog(streams map[string]Stream) *Catalog {
 	entries := []CatalogEntry{}
 

tap/metadata.go

@@ -44,10 +44,10 @@ type MetadataFields struct {
 	// SelectedByDefault: If the user doesn't specify should we
 	// emit this field by default
 	// This really only applies to available inclusion setting
-	SelectedByDefault bool `json:"selected-by-default"`
+	SelectedByDefault bool `json:"selected-by-default,omitempty"`
 
 	// ForcedReplicateMethod: we will set to FULL_TABLE for our tap
-	ForcedReplicationMethod string `json:"forced-replication-method"`
+	ForcedReplicationMethod string `json:"forced-replication-method,omitempty"`
 }
 
 func (m Metadata) DefaultMetadata(schema model.Schema) []Metadata {
@@ -64,5 +64,19 @@ func (m Metadata) DefaultMetadata(schema model.Schema) []Metadata {
 		},
 	}
 
+	// For columns we want to set the inclusion to available for everything - but we set
+	// selected by default to true as well (so unless the user speficially says no, we'll include it)
+	// We might want to get more intelligent later - as this way people could stop themselves
+	// from getting key data by accident
+	for name := range schema.Properties {
+		metadata = append(metadata, Metadata{
+			Breadcrumb: []string{"properties", name},
+			Metadata: MetadataFields{
+				Inclusion:         "available",
+				SelectedByDefault: true,
+			},
+		})
+	}
+
 	return metadata
 }

tap/tap.go

@@ -28,11 +28,25 @@ func Sync(ctx context.Context, logger kitlog.Logger, ol *OutputLogger, cl *clien
 
 		timeExtracted := time.Now().Format(time.RFC3339)
 		logger.Log("msg", "loading records", "time_extracted", timeExtracted)
+
 		records, err := stream.GetRecords(ctx, logger, cl)
 		if err != nil {
 			return err
 		}
 
+		// Get the enabled fields for this stream
+		disabledFields, err := catalog.GetDisabledFields(catalogEntry.Stream)
+		if err != nil {
+			return err
+		}
+
+		// Filter out the disabled fields from each record (ew)
+		for _, record := range records {
+			for fieldName := range disabledFields {
+				delete(record, fieldName)
+			}
+		}
+
 		logger.Log("msg", "outputting records", "count", len(records))
 		for _, record := range records {
 			op := &Output{