Fix #2570,#2572: Allow upserts to recognize $and clauses

src/main/java/io/stargate/sgv2/jsonapi/api/model/command/clause/update/UpdateOperation.java

@@ -20,7 +20,13 @@ protected UpdateOperation(List<A> actions) {
     this.actions = actions;
   }
 
-  public List<A> actions() {
+  /**
+   * @apiNote Doesn't return {@code List<A>}, otherwise an operation like {@code
+   *     actions().map(ActionWithLocator::locator)} may error trying to call a package-private
+   *     implementation (e.g. {@code SetOperation.Action::locator})
+   * @return List of actions that this update operation will apply to document
+   */
+  public List<? extends ActionWithLocator> actions() {
     return actions;
   }
 

src/main/java/io/stargate/sgv2/jsonapi/exception/UpdateException.java

@@ -20,6 +20,7 @@ public enum Code implements ErrorCode<UpdateException> {
     UNKNOWN_TABLE_COLUMNS,
     UNSUPPORTED_COLUMN_TYPES,
     UNSUPPORTED_OVERLAPPING_UPDATE_OPERATIONS,
+    UNSUPPORTED_OVERLAPPING_UPSERT_PATHS,
     UNSUPPORTED_UPDATE_DATA_TYPE, // from ErrorCodeV1
     UNSUPPORTED_UPDATE_FOR_PRIMARY_KEY_COLUMNS,
     UNSUPPORTED_UPDATE_OPERATIONS_FOR_TABLE,

src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/FindCollectionOperation.java

@@ -5,27 +5,31 @@
 import com.bpodgursky.jbool_expressions.Expression;
 import com.datastax.oss.driver.api.core.cql.SimpleStatement;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.Lists;
 import io.smallrye.mutiny.Uni;
 import io.stargate.sgv2.jsonapi.api.model.command.CommandContext;
 import io.stargate.sgv2.jsonapi.api.model.command.CommandResult;
 import io.stargate.sgv2.jsonapi.api.model.command.clause.sort.SortExpression;
+import io.stargate.sgv2.jsonapi.api.model.command.clause.update.ActionWithLocator;
+import io.stargate.sgv2.jsonapi.api.model.command.clause.update.SetOperation;
 import io.stargate.sgv2.jsonapi.api.request.RequestContext;
 import io.stargate.sgv2.jsonapi.exception.SchemaException;
+import io.stargate.sgv2.jsonapi.exception.UpdateException;
 import io.stargate.sgv2.jsonapi.service.cql.builder.Query;
 import io.stargate.sgv2.jsonapi.service.cql.builder.QueryBuilder;
 import io.stargate.sgv2.jsonapi.service.cqldriver.executor.QueryExecutor;
 import io.stargate.sgv2.jsonapi.service.operation.ReadOperationPage;
 import io.stargate.sgv2.jsonapi.service.operation.builder.BuiltCondition;
 import io.stargate.sgv2.jsonapi.service.operation.filters.collection.CollectionFilter;
 import io.stargate.sgv2.jsonapi.service.operation.filters.collection.IDCollectionFilter;
-import io.stargate.sgv2.jsonapi.service.operation.query.DBFilterBase;
 import io.stargate.sgv2.jsonapi.service.operation.query.DBLogicalExpression;
 import io.stargate.sgv2.jsonapi.service.projection.DocumentProjector;
 import io.stargate.sgv2.jsonapi.service.schema.collections.CollectionSchemaObject;
 import io.stargate.sgv2.jsonapi.service.schema.collections.spec.SuperShreddingMetadata;
-import io.stargate.sgv2.jsonapi.service.shredding.collections.DocumentId;
+import io.stargate.sgv2.jsonapi.util.PathMatchLocator;
 import java.util.*;
+import java.util.function.Predicate;
 import java.util.function.Supplier;
 
 /** Operation that returns the documents or its key based on the filter condition. */
@@ -454,41 +458,102 @@ public Uni<FindResponse> getDocuments(
   }
 
   /**
-   * An operation method which can return ReadDocument with an empty document, if the filter
-   * condition has _id filter it will return document with this field added.
+   * Creates a new empty document with no fields. Intended as a placeholder for when the find
+   * operation does not return any documents, but the caller still needs a document to work with
+   * (e.g. an upsert).
+   *
+   * @return A new empty document with no fields.
    */
-  public ReadDocument getNewDocument() {
+  public ReadDocument newEmptyDocument() {
+    return ReadDocument.from(null, null, objectMapper().createObjectNode());
+  }
 
+  /**
+   * Reconstructs a document from the filter used in the find operation, intended for upserts.
+   *
+   * <p>Contradictory filters (e.g. {@code {"a": 1, "a": 2}}) will cause an error to be thrown.
+   *
+   * @param pathFilter A predicate to filter which paths should be included in the reconstructed
+   *     document. Contradictory paths not matched in the document won't cause an error.
+   * @return The reconstructed document as an {@code ObjectNode}.
+   */
+  public ObjectNode reconstructDocumentFromFilter(Predicate<String> pathFilter) {
     final var rootNode = objectMapper().createObjectNode();
-    DocumentId documentId = null;
+    final var paths = new ArrayList<PathMatchLocator>();
+    final var ops = new ArrayList<SetOperation>();
+
     final var stack = new Stack<DBLogicalExpression>();
     stack.push(dbLogicalExpression);
 
     while (!stack.empty()) {
       var currentDbLogicalExpression = stack.pop();
 
-      for (DBFilterBase filter : dbLogicalExpression.filters()) {
-        // every filter must be a collection filter, because we are making a new document,
-        // and we only do this for docs
-        if (filter instanceof IDCollectionFilter idFilter) {
-          documentId = idFilter.getSingularDocumentId();
-          idFilter
-              .updateForNewDocument(objectMapper().getNodeFactory())
-              .ifPresent(setOperation -> setOperation.updateDocument(rootNode));
-        } else if (filter instanceof CollectionFilter collectionFilter) {
-          collectionFilter
-              .updateForNewDocument(objectMapper().getNodeFactory())
-              .ifPresent(setOperation -> setOperation.updateDocument(rootNode));
-        } else {
-          throw new IllegalArgumentException(
-              "Unsupported filter type in getNewDocument: %s"
-                  .formatted(filter.getClass().getName()));
+      if (currentDbLogicalExpression.operator() != DBLogicalExpression.DBLogicalOperator.AND) {
+        continue;
+      }
+
+      for (var filter : currentDbLogicalExpression.filters()) {
+        switch (filter) {
+          case CollectionFilter cf ->
+              cf.updateForNewDocument(objectMapper().getNodeFactory())
+                  .ifPresent(
+                      op -> {
+                        var beforeSize = paths.size();
+
+                        op.actions().stream()
+                            .map(ActionWithLocator::locator)
+                            .filter(l -> pathFilter.test(l.path()))
+                            .forEach(paths::add);
+
+                        if (paths.size() > beforeSize) {
+                          ops.add(op);
+                        }
+                      });
+          default ->
+              throw new IllegalArgumentException(
+                  "Unsupported filter type in getNewDocument: %s"
+                      .formatted(filter.getClass().getName()));
         }
       }
 
       currentDbLogicalExpression.subExpressions().forEach(stack::push);
     }
-    return ReadDocument.from(documentId, null, rootNode);
+
+    validateUpsertPaths(paths);
+    ops.forEach(op -> op.updateDocument(rootNode));
+
+    return rootNode;
+  }
+
+  /**
+   * Validates that the given paths do not overlap or contradict each other.
+   *
+   * <p>Throws an {@code UNSUPPORTED_OVERLAPPING_UPSERT_PATHS} on error.
+   *
+   * @param paths The list of paths to validate.
+   */
+  private void validateUpsertPaths(List<PathMatchLocator> paths) {
+    Collections.sort(paths);
+
+    for (var i = 0; i < paths.size() - 1; i++) {
+      final var current = paths.get(i);
+      final var next = paths.get(i + 1);
+
+      if (current.compareTo(next) == 0) {
+        throwUnsupportedOverlappingUpsertPaths(
+            "Path '%s' is matched more than once".formatted(current));
+      }
+
+      if (next.isSubPathOf(current)) {
+        throwUnsupportedOverlappingUpsertPaths(
+            "Both paths '%s' and '%s' are matched".formatted(current, next));
+      }
+    }
+  }
+
+  private void throwUnsupportedOverlappingUpsertPaths(String message) {
+    throw UpdateException.Code.UNSUPPORTED_OVERLAPPING_UPSERT_PATHS.get(
+        errVars(commandContext.schemaObject(), map -> map.put("errorMessage", message)));
   }
 
   /**

src/main/java/io/stargate/sgv2/jsonapi/service/operation/collections/ReadAndUpdateCollectionOperation.java

@@ -77,14 +77,7 @@ public Uni<Supplier<CommandResult>> execute(
               pageStateReference.set(findResponse.pageState());
               final List<ReadDocument> docs = findResponse.docs();
               if (upsert() && docs.isEmpty() && matchedCount.get() == 0) {
-                // TODO: creating the new document here, with the defaults from the filter, makes it
-                // harder because the new document created here may nto have an _id if there was
-                // none in the filter. A better approach may be to have the documentUpdater create
-                // the upsert document totally in once place.
-                // Currently creating to upsert document is in multiple places. To do this we would
-                // create UpdateOperations from the filter and give them to the document updated
-                // when it is created.
-                return Multi.createFrom().item(findCollectionOperation().getNewDocument());
+                return Multi.createFrom().item(findCollectionOperation().newEmptyDocument());
               } else {
                 matchedCount.addAndGet(docs.size());
                 return Multi.createFrom().items(docs.stream());
@@ -182,7 +175,11 @@ private Uni<UpdatedDocument> processUpdate(
               JsonNode originalDocument = upsert ? null : readDocument.get();
 
               DocumentUpdater.DocumentUpdaterResponse documentUpdaterResponse =
-                  documentUpdater().apply(readDocument.get().deepCopy(), upsert);
+                  documentUpdater()
+                      .apply(
+                          readDocument.get().deepCopy(),
+                          upsert,
+                          findCollectionOperation()::reconstructDocumentFromFilter);
 
               return documentUpdaterResponse
                   .updateEmbeddingVector(

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/collection/ArrayEqualsCollectionFilter.java

@@ -19,7 +19,10 @@ public ArrayEqualsCollectionFilter(
 
   @Override
   protected Optional<JsonNode> jsonNodeForNewDocument(JsonNodeFactory nodeFactory) {
-    return Optional.of(toJsonNode(nodeFactory, arrayValue));
+    if (Operator.MAP_EQUALS.equals(operator)) {
+      return Optional.of(toJsonNode(nodeFactory, arrayValue));
+    }
+    return Optional.empty();
   }
 
   //    @Override

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/collection/IsNullCollectionFilter.java

@@ -17,7 +17,10 @@ public IsNullCollectionFilter(String path, Operator operator) {
 
   @Override
   protected Optional<JsonNode> jsonNodeForNewDocument(JsonNodeFactory nodeFactory) {
-    return Optional.of(toJsonNode(nodeFactory));
+    if (Operator.CONTAINS.equals(operator)) {
+      return Optional.of(toJsonNode(nodeFactory));
+    }
+    return Optional.empty();
   }
 
   //    @Override

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/collection/MatchCollectionFilter.java

@@ -29,7 +29,7 @@ public BuiltCondition get() {
   }
 
   protected Optional<JsonNode> jsonNodeForNewDocument(JsonNodeFactory nodeFactory) {
-    return Optional.of(toJsonNode(nodeFactory, value));
+    return Optional.empty();
   }
 
   @Override

src/main/java/io/stargate/sgv2/jsonapi/service/operation/filters/collection/SubDocEqualsCollectionFilter.java

@@ -28,7 +28,10 @@ public SubDocEqualsCollectionFilter(
    */
   @Override
   protected Optional<JsonNode> jsonNodeForNewDocument(JsonNodeFactory nodeFactory) {
-    return Optional.of(toJsonNode(nodeFactory, subDocValue));
+    if (Operator.MAP_EQUALS.equals(operator)) {
+      return Optional.of(toJsonNode(nodeFactory, subDocValue));
+    }
+    return Optional.empty();
   }
 
   //    @Override

src/main/java/io/stargate/sgv2/jsonapi/service/updater/DocumentUpdater.java

@@ -17,6 +17,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
+import java.util.function.Predicate;
 
 /** Updates the document read from the database with the updates came as part of the request. */
 public record DocumentUpdater(
@@ -45,55 +47,69 @@ public static DocumentUpdater construct(ObjectNode replaceDocument) {
     return new DocumentUpdater(null, replaceDocument, replaceDocumentId, UpdateType.REPLACE);
   }
 
+  /**
+   * Some function which creates a document given a predicate to filter the fields to include.
+   *
+   * <p>Intended to be satisfied by {@link
+   * io.stargate.sgv2.jsonapi.service.operation.collections.FindCollectionOperation#reconstructDocumentFromFilter}
+   */
+  @FunctionalInterface
+  public interface DocumentReconstructor extends Function<Predicate<String>, ObjectNode> {}
+
   /**
    * This method is the entrance for first level update or replace. First level means it won't
    * vectorize if needed, but will warp an EmbeddingUpdateOperation in the DocumentUpdaterResponse
    * to do the following embedding update.
    *
    * @param readDocument Document to update
-   * @param docInserted True if document was just created (inserted); false if updating existing
-   *     document
+   * @param isNew True if document was just created (inserted); false if updating existing document
+   * @param reconstructor A function which creates a reconstructed document from the search filter
    */
-  public DocumentUpdaterResponse apply(JsonNode readDocument, boolean docInserted) {
-    ObjectNode docToUpdate = (ObjectNode) readDocument;
+  public DocumentUpdaterResponse apply(
+      JsonNode readDocument, boolean isNew, DocumentReconstructor reconstructor) {
+    final var docToUpdate = (ObjectNode) readDocument;
+
     if (UpdateType.UPDATE == updateType) {
-      return update(docToUpdate, docInserted);
+      return update(docToUpdate, isNew, reconstructor);
     } else {
-      return replace(docToUpdate, docInserted);
+      return replace(docToUpdate, isNew, reconstructor);
     }
   }
 
   /**
    * Will be used for update commands. This is first level replace. This method will replace the
    * document, but won't re-vectorize yet(detail in updateEmbeddingVector method)
-   *
-   * @param docToUpdate
-   * @param docInserted
-   * @return
    */
-  private DocumentUpdaterResponse update(ObjectNode docToUpdate, boolean docInserted) {
-    boolean modified = false;
+  private DocumentUpdaterResponse update(
+      ObjectNode docToUpdate, boolean isNew, DocumentReconstructor reconstructor) {
+    if (isNew) {
+      docToUpdate = reconstructor.apply(path -> true);
+    }
+
+    boolean modified = isNew;
     List<EmbeddingUpdateOperation> embeddingUpdateOperationList = new ArrayList<>();
-    for (UpdateOperation updateOperation : updateOperations) {
-      if (updateOperation.shouldApplyIf(docInserted)) {
-        final UpdateOperation.UpdateOperationResult updateOperationResult =
-            updateOperation.updateDocument(docToUpdate);
+
+    for (var updateOperation : updateOperations) {
+      if (updateOperation.shouldApplyIf(isNew)) {
+        final var updateOperationResult = updateOperation.updateDocument(docToUpdate);
         modified |= updateOperationResult.modified();
         embeddingUpdateOperationList.addAll(updateOperationResult.embeddingUpdateOperations());
       }
     }
+
     return new DocumentUpdaterResponse(docToUpdate, modified, embeddingUpdateOperationList);
   }
 
   /**
    * Will be used for findOneAndReplace. This is first level replace. This method will replace the
    * document, but won't re-vectorize yet(detail in updateEmbeddingVector method)
-   *
-   * @param docToUpdate
-   * @param docInserted
-   * @return
    */
-  private DocumentUpdaterResponse replace(ObjectNode docToUpdate, boolean docInserted) {
+  private DocumentUpdaterResponse replace(
+      ObjectNode docToUpdate, boolean isNew, DocumentReconstructor reconstructor) {
+    if (isNew) {
+      docToUpdate = reconstructor.apply(DocumentConstants.Fields.DOC_ID::equals);
+    }
+
     // Do deep clone so we can remove _id field and check
     ObjectNode compareDoc = docToUpdate.deepCopy();
 
@@ -102,6 +118,10 @@ private DocumentUpdaterResponse replace(ObjectNode docToUpdate, boolean docInser
     // from the replacement document to use later, future work needed to so we can
     // reliably go back and forth between JsonNode and DocumentId without losing the benefits of
     // both.
+    //
+    // addendum (toptobes) - instead of removing the field from the doc we can just overwrite the
+    // _id in the replaceDocument (if idNode != null) as they're proven to be already equivalent
+    // (i.e. no compareDoc, no deep clone, and use .get() instead of .remove())
     JsonNode idNode = compareDoc.remove(DocumentConstants.Fields.DOC_ID);
 
     // The replace document cannot specify an _id value that differs from the replaced document.

src/main/resources/errors.yaml

@@ -925,6 +925,16 @@ request-errors:
 
       Resend the command using a single update operation for each column.
 
+  - scope: UPDATE
+    code: UNSUPPORTED_OVERLAPPING_UPSERT_PATHS
+    title: Paths cannot be inferred for upsert due to overlap
+    body: |-
+      The command included overlapping paths, making it impossible to infer the document structure for the upsert.
+
+      ${errorMessage}
+
+      Resend the command ensuring paths are not contradictory or overlapping.
+
 # Note UNSUPPORTED_VECTORIZE_WHEN_MISSING_VECTORIZE_DEFINITION is a duplicate for Document scope, this one is used for Update scope.
   - scope: UPDATE
     code: UNSUPPORTED_VECTORIZE_WHEN_MISSING_VECTORIZE_DEFINITION