ENG-48446: Implementation for new Filter condition and filter conversion

config-service-impl/src/main/java/org/hypertrace/config/service/ConfigServiceGrpcImpl.java

@@ -53,8 +53,7 @@ public void upsertConfig(
     try {
       ConfigResourceContext configResourceContext = getConfigResourceContext(request);
       UpsertedConfig upsertedConfig =
-          configStore.writeConfig(
-              configResourceContext, getUserId(), request.getConfig(), getUserEmail());
+          configStore.writeConfig(configResourceContext, getUserId(), request, getUserEmail());
       UpsertConfigResponse.Builder builder = UpsertConfigResponse.newBuilder();
       builder.setConfig(request.getConfig());
       builder.setCreationTimestamp(upsertedConfig.getCreationTimestamp());

config-service-impl/src/main/java/org/hypertrace/config/service/store/ConfigStore.java

@@ -10,6 +10,7 @@
 import org.hypertrace.config.service.ConfigResourceContext;
 import org.hypertrace.config.service.v1.ContextSpecificConfig;
 import org.hypertrace.config.service.v1.UpsertAllConfigsResponse.UpsertedConfig;
+import org.hypertrace.config.service.v1.UpsertConfigRequest;
 
 /**
  * Abstraction for the backend which stores and serves the configuration data for multiple
@@ -25,7 +26,10 @@ public interface ConfigStore {
    * @return the config written to the store
    */
   UpsertedConfig writeConfig(
-      ConfigResourceContext configResourceContext, String userId, Value config, String userEmail)
+      ConfigResourceContext configResourceContext,
+      String userId,
+      UpsertConfigRequest request,
+      String userEmail)
       throws IOException;
 
   /**

config-service-impl/src/main/java/org/hypertrace/config/service/store/DocumentConfigStore.java

@@ -6,6 +6,7 @@
 import static org.hypertrace.config.service.store.ConfigDocument.RESOURCE_NAMESPACE_FIELD_NAME;
 import static org.hypertrace.config.service.store.ConfigDocument.TENANT_ID_FIELD_NAME;
 import static org.hypertrace.config.service.store.ConfigDocument.VERSION_FIELD_NAME;
+import static org.hypertrace.core.documentstore.Filter.Op.OR;
 
 import com.google.common.collect.Maps;
 import com.google.protobuf.Value;
@@ -29,6 +30,7 @@
 import org.hypertrace.config.service.ConfigServiceUtils;
 import org.hypertrace.config.service.v1.ContextSpecificConfig;
 import org.hypertrace.config.service.v1.UpsertAllConfigsResponse.UpsertedConfig;
+import org.hypertrace.config.service.v1.UpsertConfigRequest;
 import org.hypertrace.core.documentstore.CloseableIterator;
 import org.hypertrace.core.documentstore.Collection;
 import org.hypertrace.core.documentstore.Datastore;
@@ -37,6 +39,7 @@
 import org.hypertrace.core.documentstore.Key;
 import org.hypertrace.core.documentstore.OrderBy;
 import org.hypertrace.core.documentstore.Query;
+import org.hypertrace.core.documentstore.UpdateResult;
 
 /** Document store which stores and serves the configurations. */
 @Slf4j
@@ -46,34 +49,54 @@ public class DocumentConfigStore implements ConfigStore {
 
   private final Datastore datastore;
   private final Collection collection;
+  private final FilterBuilder filterBuilder;
 
   public DocumentConfigStore(Clock clock, Datastore datastore) {
     this.clock = clock;
     this.datastore = datastore;
     this.collection = this.datastore.getCollection(CONFIGURATIONS_COLLECTION);
+    this.filterBuilder = new FilterBuilder();
   }
 
   @Override
   public UpsertedConfig writeConfig(
       ConfigResourceContext configResourceContext,
       String lastUpdatedUserId,
-      Value latestConfig,
+      UpsertConfigRequest request,
       String lastUpdatedUserEmail)
       throws IOException {
     Optional<ConfigDocument> previousConfigDoc = getLatestVersionConfigDoc(configResourceContext);
     Optional<ContextSpecificConfig> optionalPreviousConfig =
         previousConfigDoc.flatMap(this::convertToContextSpecificConfig);
 
+    // reject create config with condition
+    if (optionalPreviousConfig.isEmpty() && request.hasUpsertCondition()) {
+      throw Status.FAILED_PRECONDITION
+          .withDescription("No upsert condition required for creating config")
+          .asRuntimeException();
+    }
+
     Key latestDocKey = new ConfigDocumentKey(configResourceContext);
     ConfigDocument latestConfigDocument =
         buildConfigDocument(
             configResourceContext,
-            latestConfig,
+            request.getConfig(),
             lastUpdatedUserId,
             previousConfigDoc,
             lastUpdatedUserEmail);
 
-    collection.upsert(latestDocKey, latestConfigDocument);
+    if (request.hasUpsertCondition()) {
+      Filter filter = this.filterBuilder.buildDocStoreFilter(request.getUpsertCondition());
+      UpdateResult updateResult = collection.update(latestDocKey, latestConfigDocument, filter);
+      if (updateResult.getUpdatedCount() <= 0) {
+        throw Status.FAILED_PRECONDITION
+            .withDescription("Update failed because upsert condition did not match given record")
+            .asRuntimeException();
+      }
+    } else {
+      collection.createOrReplace(latestDocKey, latestConfigDocument);
+    }
+
     return optionalPreviousConfig
         .map(previousConfig -> this.buildUpsertResult(latestConfigDocument, previousConfig))
         .orElseGet(() -> this.buildUpsertResult(latestConfigDocument));
@@ -262,7 +285,7 @@ private Filter buildConfigResourceContextsFilter(
             .map(this::getConfigResourceFieldContextFilter)
             .collect(Collectors.toUnmodifiableList());
     Filter configResourceFieldContextFilter = new Filter();
-    configResourceFieldContextFilter.setOp(Filter.Op.OR);
+    configResourceFieldContextFilter.setOp(OR);
     configResourceFieldContextFilter.setChildFilters(childFilters.toArray(Filter[]::new));
     Filter tenantIdFilter =
         Filter.eq(

config-service-impl/src/main/java/org/hypertrace/config/service/store/FilterBuilder.java

@@ -0,0 +1,109 @@
+package org.hypertrace.config.service.store;
+
+import static org.hypertrace.config.service.store.ConfigDocument.CONFIG_FIELD_NAME;
+
+import io.grpc.Status;
+import org.hypertrace.config.service.v1.LogicalFilter;
+import org.hypertrace.config.service.v1.RelationalFilter;
+import org.hypertrace.core.documentstore.Filter;
+
+class FilterBuilder {
+
+  public Filter buildDocStoreFilter(org.hypertrace.config.service.v1.Filter filter) {
+    switch (filter.getTypeCase()) {
+      case LOGICAL_FILTER:
+        return evaluateCompositeExpression(filter.getLogicalFilter());
+      case RELATIONAL_FILTER:
+        return evaluateLeafExpression(filter.getRelationalFilter());
+      case TYPE_NOT_SET:
+      default:
+        throw Status.INVALID_ARGUMENT.withDescription("Filter type unset").asRuntimeException();
+    }
+  }
+
+  private Filter evaluateCompositeExpression(LogicalFilter logicalFilter) {
+    switch (logicalFilter.getOperator()) {
+      case LOGICAL_OPERATOR_OR:
+        {
+          Filter[] childFilters =
+              logicalFilter.getOperandsList().stream()
+                  .map(this::buildDocStoreFilter)
+                  .toArray(Filter[]::new);
+          Filter filter = new Filter();
+          filter.setOp(Filter.Op.OR);
+          filter.setChildFilters(childFilters);
+          return filter;
+        }
+      case LOGICAL_OPERATOR_AND:
+        {
+          Filter[] childFilters =
+              logicalFilter.getOperandsList().stream()
+                  .map(this::buildDocStoreFilter)
+                  .toArray(Filter[]::new);
+          Filter filter = new Filter();
+          filter.setOp(Filter.Op.AND);
+          filter.setChildFilters(childFilters);
+          return filter;
+        }
+      case LOGICAL_OPERATOR_UNSPECIFIED:
+      default:
+        throw Status.INVALID_ARGUMENT
+            .withDescription("Unknown logical operator while building filter")
+            .asRuntimeException();
+    }
+  }
+
+  private Filter evaluateLeafExpression(RelationalFilter relationalFilter) {
+    switch (relationalFilter.getOperator()) {
+      case RELATIONAL_OPERATOR_EQ:
+        return new Filter(
+            Filter.Op.EQ,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_NEQ:
+        return new Filter(
+            Filter.Op.NEQ,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_IN:
+        return new Filter(
+            Filter.Op.IN,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_NOT_IN:
+        return new Filter(
+            Filter.Op.NOT_IN,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_LT:
+        return new Filter(
+            Filter.Op.LT,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_GT:
+        return new Filter(
+            Filter.Op.GT,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_LTE:
+        return new Filter(
+            Filter.Op.LTE,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case RELATIONAL_OPERATOR_GTE:
+        return new Filter(
+            Filter.Op.GTE,
+            buildConfigFieldPath(relationalFilter.getConfigJsonPath()),
+            relationalFilter.getValue());
+      case UNRECOGNIZED:
+      default:
+        throw Status.INVALID_ARGUMENT
+            .withDescription("Unknown relational operator while building filter")
+            .asRuntimeException();
+    }
+  }
+
+  private String buildConfigFieldPath(String configJsonPath) {
+    return String.format("%s.%s", CONFIG_FIELD_NAME, configJsonPath);
+  }
+}

config-service-impl/src/test/java/org/hypertrace/config/service/ConfigServiceGrpcImplTest.java

@@ -65,12 +65,15 @@ class ConfigServiceGrpcImplTest {
   void upsertConfig() throws IOException {
     ConfigStore configStore = mock(ConfigStore.class);
     when(configStore.writeConfig(
-            any(ConfigResourceContext.class), anyString(), any(Value.class), anyString()))
+            any(ConfigResourceContext.class),
+            anyString(),
+            any(UpsertConfigRequest.class),
+            anyString()))
         .thenAnswer(
             invocation -> {
-              Value config = invocation.getArgument(2, Value.class);
+              UpsertConfigRequest request = invocation.getArgument(2, UpsertConfigRequest.class);
               return UpsertedConfig.newBuilder()
-                  .setConfig(config)
+                  .setConfig(request.getConfig())
                   .setCreationTimestamp(123)
                   .setUpdateTimestamp(456)
                   .build();
@@ -300,7 +303,11 @@ void deletingNonExistingConfigShouldThrowError() throws IOException {
     assertEquals(Status.NOT_FOUND, ((StatusException) throwable).getStatus());
 
     verify(configStore, never())
-        .writeConfig(any(ConfigResourceContext.class), anyString(), any(Value.class), anyString());
+        .writeConfig(
+            any(ConfigResourceContext.class),
+            anyString(),
+            any(UpsertConfigRequest.class),
+            anyString());
     verify(responseObserver, never()).onNext(any(DeleteConfigResponse.class));
     verify(responseObserver, never()).onCompleted();
   }