Add blacklist functionality and test

ingestion/src/test/java/feast/ingestion/transform/FeatureRowToStoreAllocatorTest.java

@@ -19,7 +19,10 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import feast.proto.core.StoreProto;
+import feast.proto.core.StoreProto.Store.Subscription;
 import feast.proto.types.FeatureRowProto;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
 import org.apache.beam.sdk.testing.PAssert;
@@ -41,6 +44,10 @@ private StoreProto.Store newStore(String s) {
         .build();
   }
 
+  private StoreProto.Store newStore(List<StoreProto.Store.Subscription> subscriptionList) {
+    return StoreProto.Store.newBuilder().addAllSubscriptions(subscriptionList).build();
+  }
+
   @Test
   public void featureRowShouldBeAllocatedToStoreTagsAccordingToSubscription() {
     StoreProto.Store bqOnlyStore = newStore("bq*");
@@ -96,4 +103,55 @@ public void featureRowShouldBeAllocatedToStoreTagsAccordingToSubscription() {
 
     p.run();
   }
+
+  @Test
+  public void featureRowShouldBeAllocatedToStoreTagsAccordingToSubscriptionBlacklist() {
+    Subscription subscription1 = Subscription.newBuilder().setProject("*").setName("*").build();
+    Subscription subscription2 =
+        Subscription.newBuilder().setProject("project1").setName("fs_2").build();
+    Subscription subscription3 =
+        Subscription.newBuilder().setProject("project1").setName("fs_1").setExclude(true).build();
+    Subscription subscription4 =
+        Subscription.newBuilder().setProject("project2").setName("*").setExclude(true).build();
+
+    List<Subscription> testStoreSubscriptions1 =
+        Arrays.asList(subscription1, subscription2, subscription3);
+    StoreProto.Store testStore1 = newStore(testStoreSubscriptions1);
+
+    List<Subscription> testStoreSubscriptions2 = Arrays.asList(subscription1, subscription4);
+    StoreProto.Store testStore2 = newStore(testStoreSubscriptions2);
+
+    Map<StoreProto.Store, TupleTag<FeatureRowProto.FeatureRow>> storeTags =
+        ImmutableMap.of(
+            testStore1, new TupleTag<>(),
+            testStore2, new TupleTag<>());
+
+    PCollectionTuple allocatedRows =
+        p.apply(
+                Create.of(
+                    FeatureRowProto.FeatureRow.newBuilder().setFeatureSet("project1/fs_1").build(),
+                    FeatureRowProto.FeatureRow.newBuilder().setFeatureSet("project2/fs_1").build(),
+                    FeatureRowProto.FeatureRow.newBuilder().setFeatureSet("project2/fs_2").build()))
+            .apply(
+                FeatureRowToStoreAllocator.newBuilder()
+                    .setStoreTags(storeTags)
+                    .setStores(ImmutableList.of(testStore1, testStore2))
+                    .build());
+
+    PAssert.that(
+            allocatedRows
+                .get(storeTags.get(testStore1))
+                .setCoder(ProtoCoder.of(FeatureRowProto.FeatureRow.class))
+                .apply("CountStore1", Count.globally()))
+        .containsInAnyOrder(2L);
+
+    PAssert.that(
+            allocatedRows
+                .get(storeTags.get(testStore2))
+                .setCoder(ProtoCoder.of(FeatureRowProto.FeatureRow.class))
+                .apply("CountStore2", Count.globally()))
+        .containsInAnyOrder(1L);
+
+    p.run();
+  }
 }

protos/feast/core/Store.proto

@@ -149,6 +149,9 @@ message Store {
     // - my-feature-set-6 can be used to select a single feature set
     string name = 1;
 
+    // Boolean to filter retrieval of subscriptions
+    bool exclude = 4;
+
     // Feature set version was removed in v0.5.0.
     reserved 2;
   }