diff --git a/.prow/config.yaml b/.prow/config.yaml
index 62124578f0..2cdb491831 100644
--- a/.prow/config.yaml
+++ b/.prow/config.yaml
@@ -204,6 +204,27 @@ presubmits:
     skip_branches:
     - ^v0\.(3|4)-branch$
 
+  - name: test-end-to-end-batch-fs-update
+      decorate: true
+      always_run: false
+      spec:
+        volumes:
+          - name: service-account
+            secret:
+              secretName: feast-service-account
+        containers:
+          - image: maven:3.6-jdk-11
+            command: ["infra/scripts/test-end-to-end-batch.sh", "-m", "fs_update"]
+            resources:
+              requests:
+                cpu: "6"
+                memory: "6144Mi"
+            volumeMounts:
+              - name: service-account
+                mountPath: "/etc/service-account"
+      skip_branches:
+        - ^v0\.(3|4)-branch$
+
   - name: test-end-to-end-batch-java-8
     decorate: true
     always_run: true
diff --git a/infra/scripts/test-end-to-end-batch.sh b/infra/scripts/test-end-to-end-batch.sh
index 4b18f6a067..fe24c0df33 100755
--- a/infra/scripts/test-end-to-end-batch.sh
+++ b/infra/scripts/test-end-to-end-batch.sh
@@ -3,6 +3,20 @@
 set -e
 set -o pipefail
 
+PYTEST_MARK='direct_runner' #default
+
+print_usage() {
+  printf "Usage: ./test-end-to-end-batch -m pytest_mark"
+}
+
+while getopts 'm:' flag; do
+  case "${flag}" in
+    m) PYTEST_MARK="${OPTARG}" ;;
+    *) print_usage
+       exit 1 ;;
+  esac
+done
+
 test -z ${GOOGLE_APPLICATION_CREDENTIALS} && GOOGLE_APPLICATION_CREDENTIALS="/etc/service-account/service-account.json"
 test -z ${SKIP_BUILD_JARS} && SKIP_BUILD_JARS="false"
 test -z ${GOOGLE_CLOUD_PROJECT} && GOOGLE_CLOUD_PROJECT="kf-feast"
@@ -254,7 +268,7 @@ ORIGINAL_DIR=$(pwd)
 cd tests/e2e
 
 set +e
-pytest bq-batch-retrieval.py -m direct_runner --gcs_path "gs://${TEMP_BUCKET}/" --junitxml=${LOGS_ARTIFACT_PATH}/python-sdk-test-report.xml
+pytest bq-batch-retrieval.py -m ${PYTEST_MARK} --gcs_path "gs://${TEMP_BUCKET}/" --junitxml=${LOGS_ARTIFACT_PATH}/python-sdk-test-report.xml
 TEST_EXIT_CODE=$?
 
 if [[ ${TEST_EXIT_CODE} != 0 ]]; then
diff --git a/tests/e2e/bq-batch-retrieval.py b/tests/e2e/bq-batch-retrieval.py
index a9a51894c7..35075c92bb 100644
--- a/tests/e2e/bq-batch-retrieval.py
+++ b/tests/e2e/bq-batch-retrieval.py
@@ -445,6 +445,17 @@ def infra_teardown(pytestconfig, core_url, serving_url):
         print("Cleaning up not required")
 
 
+
+'''
+This suite of tests tests the apply feature set - update feature set - retrieve  
+event sequence. It ensures that when a feature set is updated, tombstoned features 
+are no longer retrieved, and added features are null for previously ingested
+rows.
+
+It is marked separately because of the length of time required
+to perform this test, due to bigquery schema caching for streaming writes.
+'''
+
 @pytest.fixture(scope="module")
 def update_featureset_dataframe():
     n_rows = 10
@@ -461,7 +472,7 @@ def update_featureset_dataframe():
     )
 
 
-@pytest.mark.direct_runner
+@pytest.mark.fs_update
 @pytest.mark.run(order=20)
 def test_update_featureset_apply_featureset_and_ingest_first_subset(
     client, update_featureset_dataframe
@@ -494,7 +505,7 @@ def test_update_featureset_apply_featureset_and_ingest_first_subset(
     assert output["update_feature2"].to_list() == subset_df["update_feature2"].to_list()
 
 
-@pytest.mark.direct_runner
+@pytest.mark.fs_update
 @pytest.mark.timeout(600)
 @pytest.mark.run(order=21)
 def test_update_featureset_update_featureset_and_ingest_second_subset(
@@ -548,7 +559,7 @@ def test_update_featureset_update_featureset_and_ingest_second_subset(
     assert output["update_feature4"].to_list() == subset_df["update_feature4"].to_list()
 
 
-@pytest.mark.direct_runner
+@pytest.mark.fs_update
 @pytest.mark.run(order=22)
 def test_update_featureset_retrieve_all_fields(client, update_featureset_dataframe):
     with pytest.raises(Exception):
@@ -564,7 +575,7 @@ def test_update_featureset_retrieve_all_fields(client, update_featureset_datafra
         feature_retrieval_job.result()
 
 
-@pytest.mark.direct_runner
+@pytest.mark.fs_update
 @pytest.mark.run(order=23)
 def test_update_featureset_retrieve_valid_fields(client, update_featureset_dataframe):
     feature_retrieval_job = client.get_batch_features(