ChangeFeed : Adds ChangeFeedStartFrom to support StartTimes x FeedRanges

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/FeedManagement/PartitionSupervisorFactoryCore.cs

@@ -22,21 +22,18 @@ public PartitionSupervisorFactoryCore(
  FeedProcessorFactory<T> partitionProcessorFactory,
  ChangeFeedLeaseOptions options)
  {
- if (observerFactory == null) throw new ArgumentNullException(nameof(observerFactory));
- if (leaseManager == null) throw new ArgumentNullException(nameof(leaseManager));
- if (options == null) throw new ArgumentNullException(nameof(options));
- if (partitionProcessorFactory == null) throw new ArgumentNullException(nameof(partitionProcessorFactory));
-
- this.observerFactory = observerFactory;
- this.leaseManager = leaseManager;
- this.changeFeedLeaseOptions = options;
- this.partitionProcessorFactory = partitionProcessorFactory;
+ this.observerFactory = observerFactory ?? throw new ArgumentNullException(nameof(observerFactory));
+ this.leaseManager = leaseManager ?? throw new ArgumentNullException(nameof(leaseManager));
+ this.changeFeedLeaseOptions = options ?? throw new ArgumentNullException(nameof(options));
+ this.partitionProcessorFactory = partitionProcessorFactory ?? throw new ArgumentNullException(nameof(partitionProcessorFactory));
  }
 
  public override PartitionSupervisor Create(DocumentServiceLease lease)
  {
  if (lease == null)
+ {
  throw new ArgumentNullException(nameof(lease));
+ }
 
  ChangeFeedObserver<T> changeFeedObserver = this.observerFactory.CreateObserver();
  FeedProcessor processor = this.partitionProcessorFactory.Create(lease, changeFeedObserver);

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/Utils/ResultSetIteratorUtils.cs

@@ -16,20 +16,35 @@ public static ChangeFeedPartitionKeyResultSetIteratorCore BuildResultSetIterator
  DateTime? startTime,
  bool startFromBeginning)
  {
- ChangeFeedRequestOptions requestOptions = new ChangeFeedRequestOptions();
- if (startTime.HasValue)
+ FeedRange feedRange = new FeedRangePartitionKeyRange(partitionKeyRangeId);
+
+ ChangeFeedRequestOptions.StartFrom startFrom;
+ if (continuationToken != null)
+ {
+ // For continuation based feed range we need to manufactor a new continuation token that has the partition key range id in it.
+ startFrom = ChangeFeedRequestOptions.StartFrom.CreateFromContinuation(continuationToken);
+ throw new NotImplementedException("Need to implement after I see what the continuation token looks like.");
+ }
+ else if (startTime.HasValue)
  {
- requestOptions.StartTime = startTime.Value;
+ startFrom = ChangeFeedRequestOptions.StartFrom.CreateFromTimeWithRange(startTime.Value, feedRange);
  }
  else if (startFromBeginning)
  {
- requestOptions.StartTime = ChangeFeedRequestOptions.DateTimeStartFromBeginning;
+ startFrom = ChangeFeedRequestOptions.StartFrom.CreateFromBeginningWithRange(feedRange);
  }
+ else
+ {
+ startFrom = ChangeFeedRequestOptions.StartFrom.CreateFromNowWithRange(feedRange);
+ }
+
+ ChangeFeedRequestOptions requestOptions = new ChangeFeedRequestOptions()
+ {
+ MaxItemCount = maxItemCount,
+ From = startFrom,
+ };
 
  return new ChangeFeedPartitionKeyResultSetIteratorCore(
- partitionKeyRangeId: partitionKeyRangeId,
- continuationToken: continuationToken,
- maxItemCount: maxItemCount,
  clientContext: container.ClientContext,
  container: container,
  options: requestOptions);

Microsoft.Azure.Cosmos/src/FeedRange/Continuations/FeedRangeCompositeContinuation.cs

@@ -35,13 +35,6 @@ private FeedRangeCompositeContinuation(
  this.CompositeContinuationTokens = new Queue<CompositeContinuationToken>();
  }
 
- public override void Accept(
- FeedRangeVisitor visitor,
- Action<RequestMessage, string> fillContinuation)
- {
- visitor.Visit(this, fillContinuation);
- }
-
  public FeedRangeCompositeContinuation(
  string containerRid,
  FeedRangeInternal feedRange,
@@ -100,6 +93,21 @@ public FeedRangeCompositeContinuation(
 
  public override string GetContinuation() => this.CurrentToken?.Token;
 
+ public override FeedRange GetFeedRange()
+ {
+ if (this.FeedRange is FeedRangePartitionKeyRange)
+ {
+ return this.FeedRange;
+ }
+
+ if (this.CurrentToken != null)
+ {
+ return new FeedRangeEPK(this.CurrentToken.Range);
+ }
+
+ return null;
+ }
+
  public override string ToString()
  {
  return JsonConvert.SerializeObject(this);
@@ -228,7 +236,7 @@ private static bool TryParseAsCompositeContinuationToken(
  {
  return false;
  }
- } 
+ }
 
  private static CompositeContinuationToken CreateCompositeContinuationTokenForRange(
  string minInclusive,
@@ -252,7 +260,7 @@ private void MoveToNextToken()
  // Consider current range done, if this FeedToken contains multiple ranges due to splits, all of them need to be considered done
  this.CompositeContinuationTokens.Enqueue(recentToken);
  }
- 
+
  this.CurrentToken = this.CompositeContinuationTokens.Count > 0 ? this.CompositeContinuationTokens.Peek() : null;
  }
 
@@ -316,5 +324,7 @@ private void CreateChildRanges(IReadOnlyList<Documents.PartitionKeyRange> keyRan
 
  return keyRanges;
  }
+
+ public override void Accept(IFeedRangeContinuationVisitor visitor) => visitor.Visit(this);
  }
 }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRange.cs

@@ -38,5 +38,7 @@ public static FeedRange FromJsonString(string toStringValue)
 
  return parsedRange;
  }
+
+ public static FeedRange CreateFromPartitionKey(PartitionKey partitionKey) => new FeedRangePartitionKey(partitionKey);
  }
 }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRangeContinuation.cs

@@ -33,12 +33,10 @@ public FeedRangeContinuation(
  this.ContainerRid = containerRid;
  }
 
- public abstract void Accept(
- FeedRangeVisitor visitor,
- Action<RequestMessage, string> fillContinuation);
-
  public abstract string GetContinuation();
 
+ public abstract FeedRange GetFeedRange();
+
  public abstract void ReplaceContinuation(string continuationToken);
 
  public abstract bool IsDone { get; }
@@ -64,5 +62,7 @@ public static bool TryParse(
  ContainerInternal containerCore,
  ResponseMessage responseMessage,
  CancellationToken cancellationToken);
+
+ public abstract void Accept(IFeedRangeContinuationVisitor visitor);
  }
 }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangeContinuationRequestMessagePopulatorVisitor.cs

@@ -0,0 +1,35 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+ using System;
+
+ /// <summary>
+ /// Visitor to populate RequestMessage headers and properties based on FeedRange.
+ /// </summary>
+ internal sealed class FeedRangeContinuationRequestMessagePopulatorVisitor : IFeedRangeContinuationVisitor
+ {
+ private readonly RequestMessage request;
+ private readonly Action<RequestMessage, string> fillContinuation;
+
+ public FeedRangeContinuationRequestMessagePopulatorVisitor(RequestMessage request, Action<RequestMessage, string> fillContinuation)
+ {
+ this.request = request ?? throw new ArgumentNullException(nameof(request));
+ this.fillContinuation = fillContinuation ?? throw new ArgumentNullException(nameof(fillContinuation));
+ }
+
+ public void Visit(FeedRangeCompositeContinuation feedRangeCompositeContinuation)
+ {
+ // In case EPK has already been set by compute
+ if (!this.request.Properties.ContainsKey(HandlerConstants.StartEpkString))
+ {
+ this.request.Properties[HandlerConstants.StartEpkString] = feedRangeCompositeContinuation.CurrentToken.Range.Min;
+ this.request.Properties[HandlerConstants.EndEpkString] = feedRangeCompositeContinuation.CurrentToken.Range.Max;
+ }
+
+ this.fillContinuation(this.request, feedRangeCompositeContinuation.GetContinuation());
+ }
+ }
+}

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangeEPK.cs

@@ -4,6 +4,7 @@
 
 namespace Microsoft.Azure.Cosmos
 {
+ using System;
  using System.Collections.Generic;
  using System.Linq;
  using System.Threading;
@@ -15,29 +16,28 @@ namespace Microsoft.Azure.Cosmos
  /// </summary>
  internal sealed class FeedRangeEPK : FeedRangeInternal
  {
- public Documents.Routing.Range<string> Range { get; }
-
- public static FeedRangeEPK ForFullRange()
- {
- return new FeedRangeEPK(new Documents.Routing.Range<string>(
- Documents.Routing.PartitionKeyInternal.MinimumInclusiveEffectivePartitionKey,
- Documents.Routing.PartitionKeyInternal.MaximumExclusiveEffectivePartitionKey,
- isMinInclusive: true,
- isMaxInclusive: false));
- }
+ public static readonly FeedRangeEPK FullRange = new FeedRangeEPK(new Documents.Routing.Range<string>(
+ Documents.Routing.PartitionKeyInternal.MinimumInclusiveEffectivePartitionKey,
+ Documents.Routing.PartitionKeyInternal.MaximumExclusiveEffectivePartitionKey,
+ isMinInclusive: true,
+ isMaxInclusive: false));
 
  public FeedRangeEPK(Documents.Routing.Range<string> range)
  {
+ if (range == null)
+ {
+ throw new ArgumentNullException(nameof(range));
+ }
+
  this.Range = range;
  }
 
+ public Documents.Routing.Range<string> Range { get; }
+
  public override Task<List<Documents.Routing.Range<string>>> GetEffectiveRangesAsync(
  IRoutingMapProvider routingMapProvider,
  string containerRid,
- Documents.PartitionKeyDefinition partitionKeyDefinition)
- {
- return Task.FromResult(new List<Documents.Routing.Range<string>>() { this.Range });
- }
+ Documents.PartitionKeyDefinition partitionKeyDefinition) => Task.FromResult(new List<Documents.Routing.Range<string>>() { this.Range });
 
  public override async Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  IRoutingMapProvider routingMapProvider,
@@ -49,10 +49,11 @@ public override async Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  return partitionKeyRanges.Select(partitionKeyRange => partitionKeyRange.Id);
  }
 
- public override void Accept(FeedRangeVisitor visitor)
- {
- visitor.Visit(this);
- }
+ public override void Accept(IFeedRangeVisitor visitor) => visitor.Visit(this);
+
+ public override Task<TResult> AcceptAsync<TResult>(
+ IFeedRangeAsyncVisitor<TResult> visitor,
+ CancellationToken cancellationToken = default) => visitor.VisitAsync(this, cancellationToken);
 
  public override string ToString() => this.Range.ToString();
  }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangeInternal.cs

@@ -25,7 +25,9 @@ public abstract Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  Documents.PartitionKeyDefinition partitionKeyDefinition,
  CancellationToken cancellationToken);
 
- public abstract void Accept(FeedRangeVisitor visitor);
+ public abstract void Accept(IFeedRangeVisitor visitor);
+
+ public abstract Task<TResult> AcceptAsync<TResult>(IFeedRangeAsyncVisitor<TResult> visitor, CancellationToken cancellationToken = default);
 
  public abstract override string ToString();
 

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangePartitionKey.cs

@@ -24,14 +24,12 @@ public FeedRangePartitionKey(PartitionKey partitionKey)
  public override Task<List<Documents.Routing.Range<string>>> GetEffectiveRangesAsync(
  IRoutingMapProvider routingMapProvider,
  string containerRid,
- Documents.PartitionKeyDefinition partitionKeyDefinition)
- {
- return Task.FromResult(new List<Documents.Routing.Range<string>>
+ Documents.PartitionKeyDefinition partitionKeyDefinition) => Task.FromResult(
+ new List<Documents.Routing.Range<string>>
  {
  Documents.Routing.Range<string>.GetPointRange(
  this.PartitionKey.InternalKey.GetEffectivePartitionKeyString(partitionKeyDefinition))
  });
- }
 
  public override async Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  IRoutingMapProvider routingMapProvider,
@@ -44,10 +42,11 @@ public override async Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  return new List<string>() { range.Id };
  }
 
- public override void Accept(FeedRangeVisitor visitor)
- {
- visitor.Visit(this);
- }
+ public override void Accept(IFeedRangeVisitor visitor) => visitor.Visit(this);
+
+ public override Task<TResult> AcceptAsync<TResult>(
+ IFeedRangeAsyncVisitor<TResult> visitor,
+ CancellationToken cancellationToken = default) => visitor.VisitAsync(this, cancellationToken);
 
  public override string ToString() => this.PartitionKey.InternalKey.ToJsonString();
  }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangePartitionKeyRange.cs

@@ -19,13 +19,13 @@ namespace Microsoft.Azure.Cosmos
  /// </summary>
  internal sealed class FeedRangePartitionKeyRange : FeedRangeInternal
  {
- public string PartitionKeyRangeId { get; }
-
  public FeedRangePartitionKeyRange(string partitionKeyRangeId)
  {
  this.PartitionKeyRangeId = partitionKeyRangeId;
  }
 
+ public string PartitionKeyRangeId { get; }
+
  public override async Task<List<Documents.Routing.Range<string>>> GetEffectiveRangesAsync(
  IRoutingMapProvider routingMapProvider,
  string containerRid,
@@ -74,10 +74,11 @@ public override Task<IEnumerable<string>> GetPartitionKeyRangesAsync(
  return Task.FromResult(partitionKeyRanges);
  }
 
- public override void Accept(FeedRangeVisitor visitor)
- {
- visitor.Visit(this);
- }
+ public override void Accept(IFeedRangeVisitor visitor) => visitor.Visit(this);
+
+ public override Task<TResult> AcceptAsync<TResult>(
+ IFeedRangeAsyncVisitor<TResult> visitor,
+ CancellationToken cancellationToken = default) => visitor.VisitAsync(this, cancellationToken);
 
  public override string ToString() => this.PartitionKeyRangeId;
  }

Microsoft.Azure.Cosmos/src/FeedRange/FeedRanges/FeedRangePartitionKeyRangeExtractor.cs

@@ -0,0 +1,53 @@
+// ------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos
+{
+ using System;
+ using System.Collections.Generic;
+ using System.Linq;
+ using System.Threading;
+ using System.Threading.Tasks;
+ using Microsoft.Azure.Documents;
+
+ internal sealed class FeedRangePartitionKeyRangeExtractor : IFeedRangeAsyncVisitor<IReadOnlyList<Documents.Routing.Range<string>>>
+ {
+ private readonly ContainerInternal container;
+
+ public FeedRangePartitionKeyRangeExtractor(ContainerInternal container)
+ {
+ this.container = container ?? throw new ArgumentNullException(nameof(container));
+ }
+
+ public async Task<IReadOnlyList<Documents.Routing.Range<string>>> VisitAsync(FeedRangePartitionKey feedRange, CancellationToken cancellationToken = default)
+ {
+ Routing.PartitionKeyRangeCache partitionKeyRangeCache = await this.container.ClientContext.DocumentClient.GetPartitionKeyRangeCacheAsync();
+ PartitionKeyDefinition partitionKeyDefinition = await this.container.GetPartitionKeyDefinitionAsync(cancellationToken);
+ return await feedRange.GetEffectiveRangesAsync(
+ partitionKeyRangeCache,
+ await this.container.GetRIDAsync(cancellationToken),
+ partitionKeyDefinition);
+ }
+
+ public async Task<IReadOnlyList<Documents.Routing.Range<string>>> VisitAsync(FeedRangePartitionKeyRange feedRange, CancellationToken cancellationToken = default)
+ {
+ // Migration from PKRangeId scenario
+ Routing.PartitionKeyRangeCache partitionKeyRangeCache = await this.container.ClientContext.DocumentClient.GetPartitionKeyRangeCacheAsync();
+ return await feedRange.GetEffectiveRangesAsync(
+ routingMapProvider: partitionKeyRangeCache,
+ containerRid: await this.container.GetRIDAsync(cancellationToken),
+ partitionKeyDefinition: null);
+ }
+
+ public async Task<IReadOnlyList<Documents.Routing.Range<string>>> VisitAsync(FeedRangeEPK feedRange, CancellationToken cancellationToken = default)
+ {
+ Routing.PartitionKeyRangeCache partitionKeyRangeCache = await this.container.ClientContext.DocumentClient.GetPartitionKeyRangeCacheAsync();
+ IReadOnlyList<PartitionKeyRange> pkRanges = await partitionKeyRangeCache.TryGetOverlappingRangesAsync(
+ collectionRid: await this.container.GetRIDAsync(cancellationToken),
+ range: feedRange.Range,
+ forceRefresh: false);
+ return pkRanges.Select(pkRange => pkRange.ToRange()).ToList();
+ }
+ }
+}