addFeedRangesAndUseFeedRangeInQueryChangeFeed

[xinlian12]

This PR merges PR #36930 and PR #37444

Description

In this PR, we added support for getting the feedRanges of the container and added support for getting change feed by using feed ranges.

Getting feed ranges from container

feedRange - represents a set of co-located logical partitions. For each container, customer will be able to get a list of feedRanges(each representing a hash-range of single physical partition). For query(future) and changeFeed(in this PR), SDK will allow customers to config a feedRange(can map to a single physical partition, span of multiple physical partitions or a subset of single physical partition) to filter the results.

Examples:
created_collection.read_feed_ranges()
await created_collection.read_feed_ranges()

Adding feed range support in change feed query

There are few issues with current changeFeed query API:

• It allows customer to pass in a physical partition id for filtering the results, however there is no official public contract exists today for customer to acquire the physical partition id.
• Physical partition id is an internal concept which from SDK perspective, we would like to keep as internal implementation details.
• Currently, the continuation token being returned is simple _lsn, which is not split/merge proof. A new format continuation token will need to be returned
• Does not support all change feed mode(Not included in this PR, will add in following PR)

Based on the above considerations, except adding feedRange support, we are also going to deprecate partition_key_range_id and is_start_from_beginning parameter. The API will be changed into the following:

 @overload
    def query_items_change_feed(
            self,
            *,
            max_item_count: Optional[int] = None,
            start_time: Optional[Union[datetime, Literal["Now", "Beginning"]]] = None,
            partition_key: PartitionKeyType,
            priority: Optional[Literal["High", "Low"]] = None,
            **kwargs: Any
    ) -> ItemPaged[Dict[str, Any]]:

  @overload
    def query_items_change_feed(
            self,
            *,
            feed_range:FeedRange,
            max_item_count: Optional[int] = None,
            start_time: Optional[Union[datetime, Literal["Now", "Beginning"]]] = None,
            priority: Optional[Literal["High", "Low"]] = None,
            **kwargs: Any
    ) -> ItemPaged[Dict[str, Any]]:

    @overload
    def query_items_change_feed(
            self,
            *,
            continuation: str,
            max_item_count: Optional[int] = None,
            priority: Optional[Literal["High", "Low"]] = None,
            **kwargs: Any
    ) -> ItemPaged[Dict[str, Any]]:

    @overload
    def query_items_change_feed(
            self,
            *,
            max_item_count: Optional[int] = None,
            start_time: Optional[Union[datetime, Literal["Now", "Beginning"]]] = None,
            priority: Optional[Literal["High", "Low"]] = None,
            **kwargs: Any
    ) -> ItemPaged[Dict[str, Any]]:

    @distributed_trace
    def query_items_change_feed(
            self,
            *args: Any,
            **kwargs: Any
    ) -> ItemPaged[Dict[str, Any]]:

Notes:
continuation token v1 -> old/existing formatted continuation token -> simple _lsn format, for example "3"
continuation token v2 -> new formatted continuation token ,it will contain information including containerRid, changeFeed mode, changeFeed start from, and a list of continuationTokens for each sub-range
continuation token when filter by pk value

eyJ2IjogInYyIiwiY29udGFpbmVyUmlkIjogIk1TVXRBTEVqc2FzPSIsICJtb2RlIjogIkluY3JlbWVudGFsIiwgInN0YXJ0RnJvbSI6IHsiVHlwZSI6ICJCZWdpbm5pbmcifSwiY29udGludWF0aW9uIjogeyJ2IjogInYyIiwicmlkIjogIk1TVXRBTEVqc2FzPSIsImNvbnRpbnVhdGlvbiI6IFsgeyAidG9rZW4iOiAiXCIyXCIiLCJyYW5nZSI6IHsibWluIjogIiIsICJtYXgiOiAiRkYiLCJpc01pbkluY2x1c2l2ZSI6IHRydWUsImlzTWF4SW5jbHVzaXZlIjogZmFsc2UgfSB9IF0sIlBLIjogJ3BrJyB9fQ==

continuation token when filter by feed range

eyJ2IjogInYyIiwiY29udGFpbmVyUmlkIjogIk1TVXRBTEVqc2FzPSIsIm1vZGUiOiAiSW5jcmVtZW50YWwiLCJzdGFydEZyb20iOiB7IlR5cGUiOiAiQmVnaW5uaW5nIiB9LCJjb250aW51YXRpb24iOiB7InYiOiAidjIiLCJyaWQiOiAiTVNVdEFMRWpzYXM9IiwiY29udGludWF0aW9uIjogWyB7InRva2VuIjogIlwiMlwiIiwicmFuZ2UiOiB7Im1pbiI6ICIiLCJtYXgiOiAiRkYiLCJpc01pbkluY2x1c2l2ZSI6IHRydWUsImlzTWF4SW5jbHVzaXZlIjogZmFsc2UgfSB9XSwiUmFuZ2UiOiB7Im1pbiI6ICIiLCJtYXgiOiAiRkYiLCAiaXNNaW5JbmNsdXNpdmUiOiB0cnVlLCAiaXNNYXhJbmNsdXNpdmUiOiBmYWxzZSB9fX0=

• Filter by physical partition id will still be supported, but with the continuation token v1 version being returned, it will not split/merge safe
• When filter by partition key, if it starts with continuation token none, then new formatted continuation token will be returned. If continuation token v1 is being used, then continuation token v1 version will continue being returned. Otherwise, it will return continuation token v2 version
• When filter by feed range, if continuation token v1 is being used, an exception will be thrown

Split flow

Continuation token before split:

eyAidiI6ICJ2MiIsImNvbnRhaW5lclJpZCI6ICJNU1V0QUxFanNhcz0iLCJtb2RlIjogIkluY3JlbWVudGFsIiwgInN0YXJ0RnJvbSI6IHsiVHlwZSI6ICJCZWdpbm5pbmcifSwiY29udGludWF0aW9uIjogeyJ2IjogInYyIiwicmlkIjogIk1TVXRBTEVqc2FzPSIsImNvbnRpbnVhdGlvbiI6IFt7InRva2VuIjogIlwiMlwiIiwgInJhbmdlIjogeyJtaW4iOiAiIiwibWF4IjogIkZGIiwiaXNNaW5JbmNsdXNpdmUiOiB0cnVlLCJpc01heEluY2x1c2l2ZSI6IGZhbHNlfSB9XSwiUmFuZ2UiOiB7ICJtaW4iOiAiIiwibWF4IjogIkZGIiwgImlzTWluSW5jbHVzaXZlIjogdHJ1ZSwiaXNNYXhJbmNsdXNpdmUiOiBmYWxzZX19fQ==

Continuation token after split:

eyAidiI6ICJ2MiIsImNvbnRhaW5lclJpZCI6ICJNU1V0QUxFanNhcz0iLCJtb2RlIjogIkluY3JlbWVudGFsIiwic3RhcnRGcm9tIjogeyAiVHlwZSI6ICJCZWdpbm5pbmciIH0sImNvbnRpbnVhdGlvbiI6IHsgInYiOiAidjIiLCJyaWQiOiAiTVNVdEFMRWpzYXM9IiwiY29udGludWF0aW9uIjogWyB7InRva2VuIjogIlwiMTFcIiIsICJyYW5nZSI6IHsibWluIjogIiIsICJtYXgiOiAiMUZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkYiLCJpc01pbkluY2x1c2l2ZSI6IHRydWUsICJpc01heEluY2x1c2l2ZSI6IGZhbHNlfSB9LCB7InRva2VuIjogIlwiMTRcIiIsInJhbmdlIjogeyJtaW4iOiAiMUZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkZGRkYiLCAibWF4IjogIkZGIiwiaXNNaW5JbmNsdXNpdmUiOiB0cnVlLCJpc01heEluY2x1c2l2ZSI6IGZhbHNlIH0gfSBdLCJSYW5nZSI6IHsibWluIjogIiIsICJtYXgiOiAiRkYiLCJpc01pbkluY2x1c2l2ZSI6IHRydWUsImlzTWF4SW5jbHVzaXZlIjogZmFsc2UgfSB9fQ==

What happens when query change feed from multiple partitions

The continuation token will contain a list of tokens for each partition, changes will be read from each partition in round-robin fashion.

using the continuation token from the above split example, it includes to tokens. one for range ["", 1FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF), one for [1FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF, FF). It will read changes from
["", 1FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF), then read changes from [1FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF, FF) and continue the pattern.

Future changes

• Add full fidelity mode support
• Add API so customer can re-load balancing based on the existing continuation tokens

[azure-sdk]

API change check

APIView has identified API level changes in this PR and created following API reviews.

azure-cosmos

[xinlian12]

/azp run python - cosmos - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[xinlian12]

/azp run python - cosmos - tests

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[tvaron3]

LGTM Thanks

[allenkim0129]

Please remove if this is not in use

[allenkim0129]

I believe that the init functions in V1 and V2 are almost identical. Can we add the init function in the base class(ChangeFeedFetcher) and call the base class's init instead? That will reduce the code duplication.

Such as, super().__init__()

[allenkim0129]

Did you want this method to be public?

If yes, please add this method as an abstract method in the abstract class, ChangeFeedFetcher
If not, please add double leading underscore, such as __fetch_change_feed_items(). In Python, even though there isn't a forcible way to make methods private, but adding underscores will help developers to differentiate between private and public methods.

[allenkim0129]

This is not preferrable, since every time when we add new versions, you will have to update this line. Instead, you could just use the base class type: ChangeFeedFetcher.

[allenkim0129]

Are there list of fetch functions?

How will you handle exceptions from them?

[allenkim0129]

If class variables used outside of class, they should be considered as public variables and they shouldn't have a single leading underscore. Please drop the underscores for those variables, such as _continuation and _change_feed_start_from

[allenkim0129]

It is little confusing to me with this dependency:

• feed_options contains change_feed_state
• change_feed_state calls its function with feed_options

[allenkim0129]

Why did you have to pop(), and reassign the item later? It didn't seem to be necessary.

Can we just use self._feed_options["changeFeedState"], whenever we need to update the change feed state?

Also, please don't use the hardcoded text here. Use a static variable like ChangeFeedStateVersion in change_feed_state.py file. It can be CHANGE_FEED_STATE_KEY="changeFeedState"

After that change, we can use self._feed_options[CHANGE_FEED_STATE_KEY]

[allenkim0129]

Not sure why this error handling was required for v2 only

[allenkim0129]

get() will return None if the key doesn't exist, and cast will cast None to a string "None".