Improved tool to be able to process duplicated rows

go.mod

@@ -8,8 +8,9 @@ require (
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/afero v1.1.2
 	github.com/spf13/cobra v0.0.5
-	github.com/stretchr/testify v1.4.0
+	github.com/stretchr/testify v1.7.0
 	golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa // indirect
+	golang.org/x/text v0.3.0 // indirect
 )
 
 go 1.13

go.sum

@@ -50,6 +50,8 @@ github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -63,3 +65,5 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/digest/diff.go

@@ -48,7 +48,7 @@ func Diff(baseConfig, deltaConfig Config) (Differences, error) {
 	baseFileDigest := NewFileDigest()
 	for digests := range baseDigestChannel {
 		for _, d := range digests {
-			baseFileDigest.Append(d)
+			baseFileDigest.SafeAppend(d)
 		}
 	}
 
@@ -94,21 +94,48 @@ func streamDifferences(baseFileDigest *FileDigest, digestChannel chan []Digest)
 		for digests := range digestChannel {
 			for _, d := range digests {
 				if baseValue, present := base.Digests[d.Key]; present {
-					if baseValue != d.Value {
-						// Modification
-						msgChannel <- message{_type: modification, current: d.Source, original: base.SourceMap[d.Key]}
+					last := len(baseValue) - 1
+					stopSearch := false
+					for i, v := range baseValue {
+						if v != d.Value && i == last {
+
+							// Modification
+							msgChannel <- message{_type: modification, current: d.Source, original: base.SourceMap[d.Key][i]}
+							stopSearch = true
+						} else if v == d.Value {
+							stopSearch = true
+						} else {
+							stopSearch = false
+						}
+						if stopSearch {
+
+							// delete from sourceMap so that at the end only deletions are left in base
+							sources := base.SourceMap[d.Key]
+							if len(sources) == 1 {
+								delete(base.SourceMap, d.Key)
+								break
+							}
+							if sources == nil {
+								msgChannel <- message{_type: addition, current: d.Source}
+								break
+							}
+							sources = append(sources[:i], sources[i+1:]...)
+							baseValue = append(baseValue[:i], baseValue[i+1:]...)
+							base.SourceMap[d.Key] = sources
+							base.Digests[d.Key] = baseValue
+							break
+						}
 					}
-					// delete from sourceMap so that at the end only deletions are left in base
-					delete(base.SourceMap, d.Key)
 				} else {
 					// Addition
 					msgChannel <- message{_type: addition, current: d.Source}
 				}
 			}
 		}
-
 		for _, value := range base.SourceMap {
-			msgChannel <- message{_type: deletion, current: value}
+			for _, v := range value {
+				msgChannel <- message{_type: deletion, current: v}
+			}
 		}
 
 	}(baseFileDigest, digestChannel, msgChannel)

pkg/digest/digest.go

@@ -30,13 +30,13 @@ const bufferSize = 512
 // Create can create a Digest using the Configurations passed.
 // It returns the digest as a map[uint64]uint64.
 // It can also keep track of the Source line.
-func Create(config *Config) (map[uint64]uint64, map[uint64][]string, error) {
+func Create(config *Config) (map[uint64]uint64, map[uint64][][]string, error) {
 	maxProcs := runtime.NumCPU()
 	reader := csv.NewReader(config.Reader)
 	reader.Comma = config.Separator
 	reader.LazyQuotes = config.LazyQuotes
 	output := make(map[uint64]uint64)
-	sourceMap := make(map[uint64][]string)
+	sourceMap := make(map[uint64][][]string)
 
 	digestChannel := make(chan []Digest, bufferSize*maxProcs)
 	errorChannel := make(chan error)
@@ -47,7 +47,9 @@ func Create(config *Config) (map[uint64]uint64, map[uint64][]string, error) {
 	for digests := range digestChannel {
 		for _, digest := range digests {
 			output[digest.Key] = digest.Value
-			sourceMap[digest.Key] = digest.Source
+			sources :=sourceMap[digest.Key]
+			sources = append(sources, digest.Source)
+			sourceMap[digest.Key] = sources
 		}
 	}
 

pkg/digest/file_digest.go

@@ -6,25 +6,25 @@ import (
 
 // FileDigest represents the digests created from one file
 type FileDigest struct {
-	Digests   map[uint64]uint64
-	SourceMap map[uint64][]string
+	Digests   map[uint64][]uint64
+	SourceMap map[uint64][][]string
 	lock      *sync.Mutex
 }
 
 // NewFileDigest to instantiate a new FileDigest
 func NewFileDigest() *FileDigest {
 	return &FileDigest{
-		Digests:   make(map[uint64]uint64),
-		SourceMap: make(map[uint64][]string),
+		Digests:   make(map[uint64][]uint64),
+		SourceMap: make(map[uint64][][]string),
 		lock:      &sync.Mutex{},
 	}
 }
 
 // Append a Digest to a FileDigest
 // This operation is not thread safe
 func (f *FileDigest) Append(d Digest) {
-	f.Digests[d.Key] = d.Value
-	f.SourceMap[d.Key] = d.Source
+	f.Digests[d.Key] = append(f.Digests[d.Key], d.Value)
+	f.SourceMap[d.Key] = append(f.SourceMap[d.Key], d.Source)
 }
 
 // SafeAppend a Digest to a FileDigest
@@ -33,6 +33,6 @@ func (f *FileDigest) SafeAppend(d Digest) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
-	f.Digests[d.Key] = d.Value
-	f.SourceMap[d.Key] = d.Source
+	f.Digests[d.Key] = append(f.Digests[d.Key], d.Value) // key the same but value is different
+	f.SourceMap[d.Key] = append(f.SourceMap[d.Key], d.Source)
 }

pkg/digest/file_digest_test.go

@@ -25,7 +25,8 @@ func TestFileDigest_Append(t *testing.T) {
 	assert.NotNil(t, fd)
 	assert.Len(t, fd.Digests, 1)
 	assert.Len(t, fd.SourceMap, 1)
-	assert.Len(t, fd.SourceMap[uint64(1)], 0)
+	assert.Len(t, fd.SourceMap[uint64(1)], 1)
+	assert.Len(t, fd.SourceMap[uint64(1)][0], 0)
 }
 
 func TestFileDigest_SafeAppend(t *testing.T) {