add doNotMergeItems func to collections

Adds a new func to the data.Collections iface:
DoNotMergeItems.  This propagates a flag that
tells kopia when it should not add items from
previous snapshots into the current snapshot
for the given collection.  This is needed in cases
like delta token expiration, where we are forced
to re-discover all items in the container instead
of performing an incremental lookup.

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

@@ -15,14 +15,15 @@ import (
 
 // MockExchangeDataCollection represents a mock exchange mailbox
 type MockExchangeDataCollection struct {
-	fullPath     path.Path
+	fullPath        path.Path
-	messageCount int
+	messageCount    int
-	Data         [][]byte
+	Data            [][]byte
-	Names        []string
+	Names           []string
-	ModTimes     []time.Time
+	ModTimes        []time.Time
-	ColState     data.CollectionState
+	ColState        data.CollectionState
-	PrevPath     path.Path
+	PrevPath        path.Path
-	DeletedItems []bool
+	DeletedItems    []bool
+	doNotMergeItems bool
 }
 
 var (
@@ -104,6 +105,10 @@ func (medc MockExchangeDataCollection) State() data.CollectionState {
 	return medc.ColState
 }
 
+func (medc MockExchangeDataCollection) DoNotMergeItems() bool {
+	return medc.doNotMergeItems
+}
+
 // 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 in that collection 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 {