add doNotMerge func to collections (#1919)

## Description

Adds a new func to the data.Collection iface:
DoNotMergeItems() tells kopia to skip the
retention of items from prior snapshots when
generating the new snapshot for this collection.
A primary use case for this flag is when a delta
token expires, preventing an incremental lookup
and forcing gc to re-discover all items in the
container.

## Does this PR need a docs update or release note?

- [x] ⛔ No 

## Type of change

- [x] 🌻 Feature

## Issue(s)

* #1914

## Test Plan

- [x] ⚡ Unit test

src/internal/connector/exchange/exchange_data_collection.go

@@ -69,6 +69,9 @@ type Collection struct {
 	prevPath path.Path
 
 	state data.CollectionState
+
+	// doNotMergeItems should only be true if the old delta token expired.
+	doNotMergeItems bool
 }
 
 // NewExchangeDataCollection creates an ExchangeDataCollection.
@@ -156,12 +159,14 @@ func (col Collection) PreviousPath() path.Path {
 	return nil
 }
 
-// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
-// hierarchies.
 func (col Collection) State() data.CollectionState {
 	return col.state
 }
 
+func (col Collection) DoNotMergeItems() bool {
+	return col.doNotMergeItems
+}
+
 // populateByOptionIdentifier is a utility function that uses col.collectionType to be able to serialize
 // all the M365IDs defined in the jobs field. data channel is closed by this function
 func (col *Collection) populateByOptionIdentifier(ctx context.Context) {

src/internal/connector/graph/metadata_collection.go

@@ -127,6 +127,10 @@ func (md MetadataCollection) State() data.CollectionState {
 	return data.NewState
 }
 
+func (md MetadataCollection) DoNotMergeItems() bool {
+	return false
+}
+
 func (md MetadataCollection) Items() <-chan data.Stream {
 	res := make(chan data.Stream)
 

src/internal/connector/mockconnector/mock_data_collection.go

@@ -23,6 +23,7 @@ type MockExchangeDataCollection struct {
 	ColState     data.CollectionState
 	PrevPath     path.Path
 	DeletedItems []bool
+	DoNotMerge   bool
 }
 
 var (
@@ -104,6 +105,10 @@ func (medc MockExchangeDataCollection) State() data.CollectionState {
 	return medc.ColState
 }
 
+func (medc MockExchangeDataCollection) DoNotMergeItems() bool {
+	return medc.DoNotMerge
+}
+
 // Items returns a channel that has the next items in the collection. The
 // channel is closed when there are no more items available.
 func (medc *MockExchangeDataCollection) Items() <-chan data.Stream {

src/internal/connector/onedrive/collection.go

@@ -57,6 +57,9 @@ type Collection struct {
 	statusUpdater support.StatusUpdater
 	itemReader    itemReaderFunc
 	ctrl          control.Options
+
+	// should only be true if the old delta token expired
+	doNotMergeItems bool
 }
 
 // itemReadFunc returns a reader for the specified item
@@ -123,6 +126,10 @@ func (oc Collection) State() data.CollectionState {
 	return data.NewState
 }
 
+func (oc Collection) DoNotMergeItems() bool {
+	return oc.doNotMergeItems
+}
+
 // Item represents a single item retrieved from OneDrive
 type Item struct {
 	id   string

src/internal/connector/sharepoint/collection.go

@@ -81,12 +81,14 @@ func (sc Collection) PreviousPath() path.Path {
 	return nil
 }
 
-// TODO(ashmrtn): Fill in once GraphConnector compares old and new folder
-// hierarchies.
 func (sc Collection) State() data.CollectionState {
 	return data.NewState
 }
 
+func (sc Collection) DoNotMergeItems() bool {
+	return false
+}
+
 func (sc *Collection) Items() <-chan data.Stream {
 	go sc.populate(context.TODO())
 	return sc.data

src/internal/data/data_collection.go

@@ -47,6 +47,15 @@ type Collection interface {
 	// backup along with all items and Collections below them in the hierarchy
 	// unless said items/Collections were moved.
 	State() CollectionState
+	// DoNotMergeItems informs kopia that the collection is rebuilding its contents
+	// from scratch, and that any items currently stored at the previousPath should
+	// be skipped during the process of merging historical data into the new backup.
+	// This flag is normally expected to be false.  It should only be flagged under
+	// specific circumstances.  Example: if the link token used for incremental queries
+	// expires or otherwise becomes unusable, thus requiring the backup producer to
+	// re-discover all data in the container.  This flag only affects the path of the
+	// collection, and does not cascade to subfolders.
+	DoNotMergeItems() bool
 }
 
 // Stream represents a single item within a Collection

src/internal/data/data_collection_test.go

@@ -30,6 +30,10 @@ func (mc mockColl) State() CollectionState {
 	return NewState
 }
 
+func (mc mockColl) DoNotMergeItems() bool {
+	return false
+}
+
 type CollectionSuite struct {
 	suite.Suite
 }

src/internal/kopia/data_collection.go

@@ -43,6 +43,10 @@ func (kdc kopiaDataCollection) State() data.CollectionState {
 	return data.NewState
 }
 
+func (kdc kopiaDataCollection) DoNotMergeItems() bool {
+	return false
+}
+
 type kopiaDataStream struct {
 	reader io.ReadCloser
 	uuid   string

src/internal/streamstore/streamstore.go

@@ -177,6 +177,10 @@ func (dc *streamCollection) State() data.CollectionState {
 	return data.NewState
 }
 
+func (dc *streamCollection) DoNotMergeItems() bool {
+	return false
+}
+
 // Items() always returns a channel with a single data.Stream
 // representing the object to be persisted
 func (dc *streamCollection) Items() <-chan data.Stream {