clean up change feed logic from query pipeline

Author: annie-mac

sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py

@@ -1162,7 +1162,6 @@ def _QueryChangeFeed(
             options = {}
         else:
             options = dict(options)
-        options["changeFeed"] = True
 
         resource_key_map = {"Documents": "docs"}
 

sdk/cosmos/azure-cosmos/azure/cosmos/_execution_context/aio/base_execution_context.py

@@ -44,7 +44,6 @@ def __init__(self, client, options):
         """
         self._client = client
         self._options = options
-        self._is_change_feed = "changeFeed" in options and options["changeFeed"] is True
         self._continuation = self._get_initial_continuation()
         self._has_started = False
         self._has_finished = False
@@ -117,10 +116,6 @@ async def _fetch_items_helper_no_retries(self, fetch_function):
         fetched_items = []
         new_options = copy.deepcopy(self._options)
         while self._continuation or not self._has_started:
-            # Check if this is first fetch for read from specific time change feed.
-            # For read specific time the first fetch will return empty even if we have more pages.
-            is_s_time_first_fetch = self._is_change_feed and self._options.get("startTime") and not self._has_started
-
             new_options["continuation"] = self._continuation
 
             response_headers = {}
@@ -129,13 +124,7 @@ async def _fetch_items_helper_no_retries(self, fetch_function):
                 self._has_started = True
 
             continuation_key = http_constants.HttpHeaders.Continuation
-            # Use Etag as continuation token for change feed queries.
-            if self._is_change_feed:
-                continuation_key = http_constants.HttpHeaders.ETag
-            # In change feed queries, the continuation token is always populated. The hasNext() test is whether
-            # there is any items in the response or not.
-            # No initial fetch for start time change feed, so we need to pass continuation token for first fetch
-            if not self._is_change_feed or fetched_items or is_s_time_first_fetch:
+            if fetched_items:
                 self._continuation = response_headers.get(continuation_key)
             else:
                 self._continuation = None

sdk/cosmos/azure-cosmos/azure/cosmos/_execution_context/base_execution_context.py

@@ -42,7 +42,6 @@ def __init__(self, client, options):
         """
         self._client = client
         self._options = options
-        self._is_change_feed = "changeFeed" in options and options["changeFeed"] is True
         self._continuation = self._get_initial_continuation()
         self._has_started = False
         self._has_finished = False
@@ -115,9 +114,6 @@ def _fetch_items_helper_no_retries(self, fetch_function):
         fetched_items = []
         new_options = copy.deepcopy(self._options)
         while self._continuation or not self._has_started:
-            # Check if this is first fetch for read from specific time change feed.
-            # For read specific time the first fetch will return empty even if we have more pages.
-            is_s_time_first_fetch = self._is_change_feed and self._options.get("startTime") and not self._has_started
             if not self._has_started:
                 self._has_started = True
             new_options["continuation"] = self._continuation
@@ -126,13 +122,7 @@ def _fetch_items_helper_no_retries(self, fetch_function):
             (fetched_items, response_headers) = fetch_function(new_options)
 
             continuation_key = http_constants.HttpHeaders.Continuation
-            # Use Etag as continuation token for change feed queries.
-            if self._is_change_feed:
-                continuation_key = http_constants.HttpHeaders.ETag
-            # In change feed queries, the continuation token is always populated. The hasNext() test is whether
-            # there is any items in the response or not.
-            # For start time however we get no initial results, so we need to pass continuation token
-            if not self._is_change_feed or fetched_items or is_s_time_first_fetch:
+            if fetched_items:
                 self._continuation = response_headers.get(continuation_key)
             else:
                 self._continuation = None

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py

@@ -2277,7 +2277,6 @@ def _QueryChangeFeed(
             options = {}
         else:
             options = dict(options)
-        options["changeFeed"] = True
 
         resource_key_map = {"Documents": "docs"}