Add 'change_feed_mode' to 'query_items_change_feed' API

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -3,6 +3,7 @@
 ### 4.9.1 (Unreleased)
 
 #### Features Added
+* Added change feed mode support in `query_items_change_feed`. See [PR 38105](https://github.com/Azure/azure-sdk-for-python/pull/38105)
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos/azure/cosmos/_base.py

@@ -61,7 +61,8 @@
     'supported_query_features': 'supportedQueryFeatures',
     'query_version': 'queryVersion',
     'priority': 'priorityLevel',
-    'no_response': 'responsePayloadOnWriteDisabled'
+    'no_response': 'responsePayloadOnWriteDisabled',
+    'max_item_count': 'maxItemCount',
 }
 
 # Cosmos resource ID validation regex breakdown:
@@ -170,6 +171,7 @@ def GetHeaders(  # pylint: disable=too-many-statements,too-many-branches
     # set consistency level. check if set via options, this will override the default
     if options.get("consistencyLevel"):
         consistency_level = options["consistencyLevel"]
+        # TODO: move this line outside of if-else cause to remove the code duplication
         headers[http_constants.HttpHeaders.ConsistencyLevel] = consistency_level
     elif default_client_consistency_level is not None:
         consistency_level = default_client_consistency_level

sdk/cosmos/azure-cosmos/azure/cosmos/_change_feed/change_feed_state.py

@@ -29,6 +29,8 @@
 from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Optional, Union, List, Any, Dict, Deque
+import logging
+from typing_extensions import Literal
 
 from azure.cosmos import http_constants
 from azure.cosmos._change_feed.change_feed_start_from import ChangeFeedStartFromInternal, \
@@ -176,18 +178,18 @@ def apply_server_response_continuation(self, continuation: str, has_modified_res
 
 class ChangeFeedStateV2(ChangeFeedState):
     container_rid_property_name = "containerRid"
-    change_feed_mode_property_name = "mode"
+    mode_property_name = "mode"
     change_feed_start_from_property_name = "startFrom"
     continuation_property_name = "continuation"
 
-    # TODO: adding change feed mode
     def __init__(
             self,
             container_link: str,
             container_rid: str,
             feed_range: FeedRangeInternal,
             change_feed_start_from: ChangeFeedStartFromInternal,
-            continuation: Optional[FeedRangeCompositeContinuation]
+            continuation: Optional[FeedRangeCompositeContinuation],
+            mode: Optional[Literal["LatestVersion", "AllVersionsAndDeletes"]]
     ) -> None:
 
         self._container_link = container_link
@@ -208,6 +210,8 @@ def __init__(
         else:
             self._continuation = continuation
 
+        self._mode = "LatestVersion" if mode is None else mode
+
         super(ChangeFeedStateV2, self).__init__(ChangeFeedStateVersion.V2)
 
     @property
@@ -218,17 +222,14 @@ def to_dict(self) -> Dict[str, Any]:
         return {
             self.version_property_name: ChangeFeedStateVersion.V2.value,
             self.container_rid_property_name: self._container_rid,
-            self.change_feed_mode_property_name: "LatestVersion",
+            self.mode_property_name: self._mode,
             self.change_feed_start_from_property_name: self._change_feed_start_from.to_dict(),
             self.continuation_property_name: self._continuation.to_dict() if self._continuation is not None else None
         }
 
-    def populate_request_headers(
+    def set_start_from_request_headers(
             self,
-            routing_provider: SmartRoutingMapProvider,
             request_headers: Dict[str, Any]) -> None:
-        request_headers[http_constants.HttpHeaders.AIM] = http_constants.HttpHeaders.IncrementalFeedHeaderValue
-
         # When a merge happens, the child partition will contain documents ordered by LSN but the _ts/creation time
         # of the documents may not be sequential.
         # So when reading the changeFeed by LSN, it is possible to encounter documents with lower _ts.
@@ -243,11 +244,10 @@ def populate_request_headers(
                     self._continuation.current_token.feed_range)
             change_feed_start_from_feed_range_and_etag.populate_request_headers(request_headers)
 
-        # based on the feed range to find the overlapping partition key range id
-        over_lapping_ranges =\
-            routing_provider.get_overlapping_ranges(
-                self._container_link,
-                [self._continuation.current_token.feed_range])
+    def set_pk_range_id_request_headers(
+            self,
+            over_lapping_ranges,
+            request_headers: Dict[str, Any]) -> None:
 
         if len(over_lapping_ranges) > 1:
             raise self.get_feed_range_gone_error(over_lapping_ranges)
@@ -260,53 +260,51 @@ def populate_request_headers(
             # the current token feed range spans less than single physical partition
             # for this case, need to set both the partition key range id and epk filter headers
             request_headers[http_constants.HttpHeaders.PartitionKeyRangeID] = over_lapping_ranges[0]["id"]
-            request_headers[
-                http_constants.HttpHeaders.StartEpkString] = self._continuation.current_token.feed_range.min
-            request_headers[
-                http_constants.HttpHeaders.EndEpkString] = self._continuation.current_token.feed_range.max
+            request_headers[http_constants.HttpHeaders.StartEpkString] = self._continuation.current_token.feed_range.min
+            request_headers[http_constants.HttpHeaders.EndEpkString] = self._continuation.current_token.feed_range.max
 
-    async def populate_request_headers_async(
+    def set_mode_request_headers(
             self,
-            async_routing_provider: AsyncSmartRoutingMapProvider,
             request_headers: Dict[str, Any]) -> None:
-        request_headers[http_constants.HttpHeaders.AIM] = http_constants.HttpHeaders.IncrementalFeedHeaderValue
+        if self._mode == "AllVersionsAndDeletes":
+            request_headers[http_constants.HttpHeaders.AIM] = http_constants.HttpHeaders.FullFidelityFeedHeaderValue
+            request_headers[http_constants.HttpHeaders.ChangeFeedWireFormatVersion] = \
+                http_constants.HttpHeaders.SeparateMetaWithCrts
+        else:
+            request_headers[http_constants.HttpHeaders.AIM] = http_constants.HttpHeaders.IncrementalFeedHeaderValue
 
-        # When a merge happens, the child partition will contain documents ordered by LSN but the _ts/creation time
-        # of the documents may not be sequential.
-        # So when reading the changeFeed by LSN, it is possible to encounter documents with lower _ts.
-        # In order to guarantee we always get the documents after customer's point start time,
-        # we will need to always pass the start time in the header.
-        self._change_feed_start_from.populate_request_headers(request_headers)
+    def populate_request_headers(
+            self,
+            routing_provider: SmartRoutingMapProvider,
+            request_headers: Dict[str, Any]) -> None:
+        self.set_start_from_request_headers(request_headers)
+
+        # based on the feed range to find the overlapping partition key range id
+        over_lapping_ranges = \
+            routing_provider.get_overlapping_ranges(
+                self._container_link,
+                [self._continuation.current_token.feed_range])
+
+        self.set_pk_range_id_request_headers(over_lapping_ranges, request_headers)
+
+        self.set_mode_request_headers(request_headers)
 
-        if self._continuation.current_token is not None and self._continuation.current_token.token is not None:
-            change_feed_start_from_feed_range_and_etag = \
-                ChangeFeedStartFromETagAndFeedRange(
-                    self._continuation.current_token.token,
-                    self._continuation.current_token.feed_range)
-            change_feed_start_from_feed_range_and_etag.populate_request_headers(request_headers)
+
+    async def populate_request_headers_async(
+            self,
+            async_routing_provider: AsyncSmartRoutingMapProvider,
+            request_headers: Dict[str, Any]) -> None:
+        self.set_start_from_request_headers(request_headers)
 
         # based on the feed range to find the overlapping partition key range id
         over_lapping_ranges = \
             await async_routing_provider.get_overlapping_ranges(
                 self._container_link,
                 [self._continuation.current_token.feed_range])
 
-        if len(over_lapping_ranges) > 1:
-            raise self.get_feed_range_gone_error(over_lapping_ranges)
+        self.set_pk_range_id_request_headers(over_lapping_ranges, request_headers)
 
-        overlapping_feed_range = Range.PartitionKeyRangeToRange(over_lapping_ranges[0])
-        if overlapping_feed_range == self._continuation.current_token.feed_range:
-            # exactly mapping to one physical partition, only need to set the partitionKeyRangeId
-            request_headers[http_constants.HttpHeaders.PartitionKeyRangeID] = over_lapping_ranges[0]["id"]
-        else:
-            # the current token feed range spans less than single physical partition
-            # for this case, need to set both the partition key range id and epk filter headers
-            request_headers[http_constants.HttpHeaders.PartitionKeyRangeID] = \
-                over_lapping_ranges[0]["id"]
-            request_headers[http_constants.HttpHeaders.StartEpkString] = \
-                self._continuation.current_token.feed_range.min
-            request_headers[http_constants.HttpHeaders.EndEpkString] = \
-                self._continuation.current_token.feed_range.max
+        self.set_mode_request_headers(request_headers)
 
     def populate_feed_options(self, feed_options: Dict[str, Any]) -> None:
         pass
@@ -367,12 +365,20 @@ def from_continuation(
         if continuation_data is None:
             raise ValueError(f"Invalid continuation: [Missing {ChangeFeedStateV2.continuation_property_name}]")
         continuation = FeedRangeCompositeContinuation.from_json(continuation_data)
-        return ChangeFeedStateV2(
+
+        mode = continuation_json.get(ChangeFeedStateV2.mode_property_name)
+        # All 'continuation_json' from ChangeFeedStateV2 must contain 'mode' property. For the 'continuation_json'
+        # from older ChangeFeedState versions won't even hit this point, since their version is not 'v2'.
+        if mode is None:
+            raise ValueError(f"Invalid continuation: [Missing {ChangeFeedStateV2.mode_property_name}]")
+
+        return cls(
             container_link=container_link,
             container_rid=container_rid,
             feed_range=continuation.feed_range,
             change_feed_start_from=change_feed_start_from,
-            continuation=continuation)
+            continuation=continuation,
+            mode=mode)
 
     @classmethod
     def from_initial_state(
@@ -394,6 +400,7 @@ def from_initial_state(
                 raise ValueError("partitionKey is in the changeFeedStateContext, but missing partitionKeyFeedRange")
         else:
             # default to full range
+            logging.info("'feed_range' empty. Using full range by default.")
             feed_range = FeedRangeInternalEpk(
                 Range(
                 "",
@@ -405,11 +412,12 @@ def from_initial_state(
         change_feed_start_from = (
             ChangeFeedStartFromInternal.from_start_time(change_feed_state_context.get("startTime")))
 
-        if feed_range is not None:
-            return cls(
-                container_link=container_link,
-                container_rid=collection_rid,
-                feed_range=feed_range,
-                change_feed_start_from=change_feed_start_from,
-                continuation=None)
-        raise ValueError("feed_range is empty")
+        mode = change_feed_state_context.get("mode")
+
+        return cls(
+            container_link=container_link,
+            container_rid=collection_rid,
+            feed_range=feed_range,
+            change_feed_start_from=change_feed_start_from,
+            continuation=None,
+            mode=mode)

sdk/cosmos/azure-cosmos/azure/cosmos/_change_feed/change_feed_utils.py

@@ -0,0 +1,146 @@
+# The MIT License (MIT)
+# Copyright (c) 2014 Microsoft Corporation
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+"""Internal Helper functions in the Azure Cosmos database change_feed service.
+"""
+
+import warnings
+from datetime import datetime
+from typing import Any, Dict, Tuple
+
+CHANGE_FEED_MODES = ["LatestVersion", "AllVersionsAndDeletes"]
+
+def add_args_to_kwargs(
+        args: Tuple[Any, ...],
+        kwargs: Dict[str, Any]
+    ) -> None:
+    """Add positional arguments(args) to keyword argument dictionary(kwargs).
+    Since 'query_items_change_feed' method only allows the following 4 positional arguments in the exact order
+    and types, if the order and types don't match, errors will be raised.
+    If the positional arguments are in the correct orders and types, the arguments will be added to keyword arguments.
+
+    4 positional arguments:
+        - str 'partition_key_range_id': [Deprecated] ChangeFeed requests can be executed against specific partition
+            key ranges. This is used to process the change feed in parallel across multiple consumers.
+        - bool 'is_start_from_beginning': [Deprecated] Get whether change feed should start from
+            beginning (true) or from current (false). By default, it's start from current (false).
+        - str 'continuation': e_tag value to be used as continuation for reading change feed.
+        - int 'max_item_count': Max number of items to be returned in the enumeration operation.
+
+    :param args: Positional arguments. Arguments must be in the following order:
+        1. partition_key_range_id
+        2. is_start_from_beginning
+        3. continuation
+        4. max_item_count
+    :type args: Tuple[Any, ...]
+    :param kwargs: Keyword arguments
+    :type kwargs: dict[str, Any]
+    """
+    if len(args) > 4:
+        raise TypeError(f"'query_items_change_feed()' takes 4 positional arguments but {len(args)} were given.")
+
+    if len(args) > 0:
+        keys = [
+            'partition_key_range_id',
+            'is_start_from_beginning',
+            'continuation',
+            'max_item_count',
+        ]
+        for i, value in enumerate(args):
+            key = keys[i]
+
+            if key in kwargs:
+                raise TypeError(f"'query_items_change_feed()' got multiple values for argument '{key}'.")
+
+            kwargs[key] = value
+
+def validate_kwargs(
+        kwargs: Dict[str, Any]
+    ) -> None:
+    """Validate keyword arguments(kwargs).
+    The values of keyword arguments must match the expect type and conditions. If the conditions do not match,
+    errors will be raised with the error messages and possible ways to correct the errors.
+
+    :param kwargs: Keyword arguments to verify for query_items_change_feed API
+    :keyword mode: Must be one of the values in the Enum, 'ChangeFeedMode'.
+        If the value is 'ALL_VERSIONS_AND_DELETES', the following keywords must be in the right condition:
+            - 'partition_key_range_id': Cannot be used at any time
+            - 'is_start_from_beginning': Must be 'False'
+            - 'start_time': Must be "Now"
+    :paramtype mode: Optional[Literal["LatestVersion", "AllVersionsAndDeletes"]]
+    :keyword partition_key_range_id: Deprecated Warning.
+    :paramtype partition_key_range_id: str
+    :keyword is_start_from_beginning: Deprecated Warning. Cannot be used with 'start_time'.
+    :paramtype is_start_from_beginning: bool
+    :keyword start_time: Must be in supported types.
+    :paramtype start_time: Union[~datetime.datetime, Literal["Now", "Beginning"]]
+    :type kwargs: dict[str, Any]
+    """
+    # Filter items with value None
+    kwargs = {key: value for key, value in kwargs.items() if value is not None}
+
+    # Validate the keyword arguments
+    if "mode" in kwargs:
+        mode = kwargs["mode"]
+        if mode not in CHANGE_FEED_MODES:
+            raise ValueError(
+                f"Invalid mode was used: '{kwargs['mode']}'."
+                f" Supported modes are {CHANGE_FEED_MODES}.")
+
+        if mode == 'AllVersionsAndDeletes':
+            if "partition_key_range_id" in kwargs:
+                raise ValueError(
+                    "'AllVersionsAndDeletes' mode is not supported if 'partition_key_range_id'"
+                    " was used. Please use 'feed_range' instead.")
+            if "is_start_from_beginning" in kwargs and kwargs["is_start_from_beginning"] is not False:
+                raise ValueError(
+                    "'AllVersionsAndDeletes' mode is only supported if 'is_start_from_beginning'"
+                    " is 'False'. Please use 'is_start_from_beginning=False' or 'continuation' instead.")
+            if "start_time" in kwargs and kwargs["start_time"] != "Now":
+                raise ValueError(
+                    "'AllVersionsAndDeletes' mode is only supported if 'start_time' is 'Now'."
+                    " Please use 'start_time=\"Now\"' or 'continuation' instead.")
+
+    if "partition_key_range_id" in kwargs:
+        warnings.warn(
+            "'partition_key_range_id' is deprecated. Please pass in 'feed_range' instead.",
+            DeprecationWarning
+        )
+
+    if "is_start_from_beginning" in kwargs:
+        warnings.warn(
+            "'is_start_from_beginning' is deprecated. Please pass in 'start_time' instead.",
+            DeprecationWarning
+        )
+
+        if not isinstance(kwargs["is_start_from_beginning"], bool):
+            raise TypeError(
+                f"'is_start_from_beginning' must be 'bool' type,"
+                f" but given '{type(kwargs['is_start_from_beginning']).__name__}'.")
+
+        if kwargs["is_start_from_beginning"] is True and "start_time" in kwargs:
+            raise ValueError("'is_start_from_beginning' and 'start_time' are exclusive, please only set one of them.")
+
+    if "start_time" in kwargs:
+        if not isinstance(kwargs['start_time'], datetime):
+            if kwargs['start_time'].lower() not in ["now", "beginning"]:
+                raise ValueError(
+                    f"'start_time' must be either 'Now' or 'Beginning', but given '{kwargs['start_time']}'.")