Table Refactoring - working insert and find

This commit gets the insert and find working in limited cases. Find does not support any clauses, just returns everything. Insert expects the document to have a field called `key` The feature flag to enable tables has been renamed to fit convention, it is now `-Dstargate.tables.enabled=true` Full command to start quarkus in dev against HCD container would be: ./mvnw quarkus:dev -Dstargate.data-store.ignore-bridge=true \ -Dstargate.jsonapi.operations.vectorize-enabled=true \ -Dstargate.jsonapi.operations.database-config.local-datacenter=dc1 \ -Dquarkus.log.console.darken=2 -Dstargate.tables.enabled=true -Poffline This is the 3rd chunk of changes, it is built against the work in ajm/tables-chunk-2 branch. it pulls commits from the ajm/tables branch listed below: commit e35c78b Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 13:40:20 2024 +1200 fmt fix commit 64f2993 Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 13:37:20 2024 +1200 fmt fix -> Skipped - merge commit <- commit d69d2c6 Merge: f7bad22 018680b Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 13:31:19 2024 +1200 Merge branch 'main' into ajm/tables commit f7bad22 Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 11:52:49 2024 +1200 Test fixes for Integration tests ReadDocument now accepts an optional docID see comments in ReadAndUpdateOperation about how we create an upsert doc commit adb2fa9 Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 10:40:14 2024 +1200 clean for InsertAttempt comments up to standard, removed unused getRow() commit 4985a94 Author: Tatu Saloranta <tatu.saloranta@datastax.com> Date: Wed Jul 10 15:24:26 2024 -0700 Replace FeatureFlags with Quarkus config setting (#1257) -> skipped - already in main <- commit 018680b Author: Hazel <hazel.he@datastax.com> Date: Wed Jul 10 18:06:48 2024 -0400 Follow up for PR #1251: Remove `Optional` and centralize validation (#1259) -> SKIPPED - NOT IN MAIN BUT SAME AS 9d977cd below <- commit bd6be18 Author: Hazel <hazel.he@datastax.com> Date: Wed Jul 10 16:41:24 2024 -0400 Improve error message when EGW timeout (#1255) -> SKIPPED - NOT IN MAIN BUT SAME AS 197bfd2 below <- commit 33311d2 Author: Hazel <hazel.he@datastax.com> Date: Wed Jul 10 15:12:45 2024 -0400 Small fix: Improve error message when not providing provider key through `x-embedding-api-key` (#1251) -> SKIPPED - NOT IN MAIN BUT SAME AS dd652c3 below <- commit 486977f Author: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed Jul 10 11:44:53 2024 -0400 Use Stargate v2.1.0-BETA-13 (#1252) Co-authored-by: jeffreyscarpenter <12115970+jeffreyscarpenter@users.noreply.github.com> commit 611282a Author: Aaron Morton <aaron.morton@datastax.com> Date: Thu Jul 11 10:00:03 2024 +1200 Insert working, one and many - some Test failing -> skipped - already in main <- commit 9d977cd Author: Hazel <hazel.he@datastax.com> Date: Wed Jul 10 16:41:24 2024 -0400 Improve error message when EGW timeout (#1255) -> skipped - already in main <- commit 197bfd2 Author: Hazel <hazel.he@datastax.com> Date: Wed Jul 10 15:12:45 2024 -0400 Small fix: Improve error message when not providing provider key through `x-embedding-api-key` (#1251) -> skipped - already in main <- commit dd652c3 Author: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed Jul 10 11:44:53 2024 -0400 Use Stargate v2.1.0-BETA-13 (#1252) Co-authored-by: jeffreyscarpenter <12115970+jeffreyscarpenter@users.noreply.github.com> commit 7e6e99d Author: Aaron Morton <aaron.morton@datastax.com> Date: Wed Jul 10 16:37:47 2024 +1200 refactor Insert results ready for tables commit 588b297 Author: Aaron Morton <aaron.morton@datastax.com> Date: Wed Jul 10 13:47:06 2024 +1200 Refactor ReadOperationPage for reuse with tables and added DocumentSource commit d62f95c Author: Aaron Morton <aaron.morton@datastax.com> Date: Wed Jul 10 10:34:56 2024 +1200 Basic findOne and find for tables just does a select *, in place to find re-use with collection commands commit fbd09e9 Author: Tatu Saloranta <tatu.saloranta@datastax.com> Date: Tue Jul 9 18:08:16 2024 -0700 Enable metadata access to "system" keyspace (for testing purposes)
stargate · Jul 16, 2024 · e29a796 · e29a796
1 parent 680a795
commit e29a796
Show file tree

Hide file tree

Showing 44 changed files with 1,078 additions and 383 deletions.
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/config/ApiTablesConfig.java b/src/main/java/io/stargate/sgv2/jsonapi/config/ApiTablesConfig.java
@@ -0,0 +1,12 @@
+package io.stargate.sgv2.jsonapi.config;
+
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+
+/** Configuration mapping for API Tables feature. */
+@ConfigMapping(prefix = "stargate.tables")
+public interface ApiTablesConfig {
+  /** Setting that determines if the API Tables feature is enabled. */
+  @WithDefault("false")
+  boolean enabled();
+}
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/FeatureFlags.java b/src/main/java/io/stargate/sgv2/jsonapi/service/FeatureFlags.java
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/executor/NamespaceCache.java b/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/executor/NamespaceCache.java
@@ -8,7 +8,6 @@
 import io.stargate.sgv2.jsonapi.api.request.DataApiRequestInfo;
 import io.stargate.sgv2.jsonapi.exception.ErrorCode;
 import io.stargate.sgv2.jsonapi.exception.JsonApiException;
-import io.stargate.sgv2.jsonapi.service.FeatureFlags;
 import io.stargate.sgv2.jsonapi.service.schema.model.JsonapiTableMatcher;
 import java.time.Duration;
 
@@ -24,6 +23,8 @@ public class NamespaceCache {
 
   private final ObjectMapper objectMapper;
 
+  private final boolean apiTablesEnabled;
+
   // TODO: move the settings to config
   // TODO: set the cache loader when creating the cache
   private static final long CACHE_TTL_SECONDS = 300;
@@ -34,10 +35,15 @@ public class NamespaceCache {
           .maximumSize(CACHE_MAX_SIZE)
           .build();
 
-  public NamespaceCache(String namespace, QueryExecutor queryExecutor, ObjectMapper objectMapper) {
+  public NamespaceCache(
+      String namespace,
+      boolean apiTablesEnabled,
+      QueryExecutor queryExecutor,
+      ObjectMapper objectMapper) {
     this.namespace = namespace;
     this.queryExecutor = queryExecutor;
     this.objectMapper = objectMapper;
+    this.apiTablesEnabled = apiTablesEnabled;
   }
 
   protected Uni<SchemaObject> getSchemaObject(
@@ -130,7 +136,7 @@ private Uni<SchemaObject> loadSchemaObject(
                     optionalTable.get(), objectMapper);
               }
 
-              if (FeatureFlags.TABLES_SUPPORTED) {
+              if (apiTablesEnabled) {
                 return new TableSchemaObject(namespace, collectionName);
               }
 

diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/executor/SchemaCache.java b/src/main/java/io/stargate/sgv2/jsonapi/service/cqldriver/executor/SchemaCache.java
@@ -9,6 +9,7 @@
 import com.github.benmanes.caffeine.cache.Caffeine;
 import io.smallrye.mutiny.Uni;
 import io.stargate.sgv2.jsonapi.api.request.DataApiRequestInfo;
+import io.stargate.sgv2.jsonapi.config.ApiTablesConfig;
 import io.stargate.sgv2.jsonapi.config.OperationsConfig;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -25,6 +26,8 @@ public class SchemaCache {
 
   @Inject OperationsConfig operationsConfig;
 
+  @Inject ApiTablesConfig apiTablesConfig;
+
   // TODO: The size of the cache should be in configuration.
   // TODO: set the cache loader when creating the cache
   private final Cache<CacheKey, NamespaceCache> schemaCache =
@@ -65,7 +68,8 @@ public void evictCollectionSettingCacheEntry(
   }
 
   private NamespaceCache addNamespaceCache(CacheKey cacheKey) {
-    return new NamespaceCache(cacheKey.namespace(), queryExecutor, objectMapper);
+    return new NamespaceCache(
+        cacheKey.namespace(), apiTablesConfig.enabled(), queryExecutor, objectMapper);
   }
 
   /**

diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/DocumentSource.java b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/DocumentSource.java
@@ -0,0 +1,29 @@
+package io.stargate.sgv2.jsonapi.service.operation.model;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.function.Supplier;
+
+/**
+ * A source of document that have been read from the DB, the document could be from a collection or
+ * a table.
+ *
+ * <p>The implementation may create the {@link JsonNode} lazy when it is called, or create and
+ * cache.
+ *
+ * <p>Note: this does not implement {@link Supplier} because there are times we use these and
+ * subclasses in a Uni chain which will see the supplier and call it, when what we really want to do
+ * is pass the instance along.
+ */
+@FunctionalInterface
+public interface DocumentSource {
+
+  /**
+   * Get the document as a JsonNode
+   *
+   * <p>Note to implementers, this method must always return a document. It is not valid to return
+   * null.
+   *
+   * @return the document as a JsonNode
+   */
+  JsonNode get();
+}
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/InsertAttempt.java b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/InsertAttempt.java
@@ -0,0 +1,67 @@
+package io.stargate.sgv2.jsonapi.service.operation.model;
+
+import io.stargate.sgv2.jsonapi.service.shredding.DocRowIdentifer;
+import java.util.Optional;
+
+/**
+ * Container for an individual Document or Row insertion attempy.
+ *
+ * <p>Tracks the original input position; document (if available), its id (if available) and
+ * possible processing error.
+ *
+ * <p>Information will be needed to build responses, including the optional detail response see
+ * {@link InsertOperationPage}
+ *
+ * <p>Is {@link Comparable} so that the attempts can be re-sorted into the order provided in the
+ * user request, compares based on the {@link #position()}
+ */
+public interface InsertAttempt extends Comparable<InsertAttempt> {
+
+  /**
+   * The zero based position of the document or row in the request from the user.
+   *
+   * @return integer position
+   */
+  int position();
+
+  /**
+   * The document _id or the row primary key, if known, used to build the response that includes the
+   * Id's of the documents / rows that were successfully inserted or failed.
+   *
+   * <p>Optional as there may be times when the input document / row could not be parsed to get the
+   * ID. And separate to having the doc / row shreddded because we may have the id (such as when
+   * creating a new document _id sever side) but were not able to shred the document / row.
+   *
+   * @return The {@link DocRowIdentifer} that identifies the document or row by ID
+   */
+  Optional<DocRowIdentifer> docRowID();
+
+  /**
+   * The first error that happened trying to run this insert.
+   *
+   * @return
+   */
+  Optional<Throwable> failure();
+
+  /**
+   * Updates the attempt with an error that happened when trying to process the insert.
+   *
+   * <p>Implmentations must only remember the first error that happened.
+   *
+   * @param failure An error that happened when trying to process the insert.
+   * @return Return the updated {@link InsertAttempt}, must be the same instance the method was
+   *     called on.
+   */
+  InsertAttempt maybeAddFailure(Throwable failure);
+
+  /**
+   * Compares the position of this attempt to another.
+   *
+   * @param other the object to be compared.
+   * @return Result of {@link Integer#compare(int, int)}
+   */
+  @Override
+  default int compareTo(InsertAttempt other) {
+    return Integer.compare(position(), other.position());
+  }
+}
diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/InsertOperationPage.java b/src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/InsertOperationPage.java
@@ -0,0 +1,159 @@
+package io.stargate.sgv2.jsonapi.service.operation.model;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandResult;
+import io.stargate.sgv2.jsonapi.api.model.command.CommandStatus;
+import io.stargate.sgv2.jsonapi.exception.mappers.ThrowableToErrorMapper;
+import io.stargate.sgv2.jsonapi.service.shredding.DocRowIdentifer;
+import java.util.*;
+import java.util.function.Supplier;
+
+/**
+ * The internal to insert operation results, keeping ids of successfully and not-successfully
+ * inserted documents.
+ *
+ * <p>Can serve as an aggregator, using the {@link #aggregate} function. TODO: AARON DOCS
+ *
+ * @param allInsertions Attempted insertions
+ * @param returnDocumentResponses Whether to return detailed document responses
+ * @param successfulInsertions Successfully inserted documents, NOTE: this list is mutated after
+ *     creation
+ * @param failedInsertions Failed insertions NOTE: this list is mutated after creation
+ */
+public class InsertOperationPage implements Supplier<CommandResult> {
+
+  // TODO: AARON changed from a record because the successfulInsertions and failedInsertions were
+  // setable in the ctor
+  // but they have to be empty mutable lists
+  private final List<? extends InsertAttempt> allInsertions;
+  private final boolean returnDocumentResponses;
+
+  // The success and failed lists are mutable and are used to build the response
+  // they do not use the ? wild card because they are always added to via {@link #aggregate} which
+  // wants to accept only of the InsertAttempt interface
+  private final List<InsertAttempt> successfulInsertions;
+  private final List<InsertAttempt> failedInsertions;
+
+  public InsertOperationPage(
+      List<? extends InsertAttempt> allAttemptedInsertions, boolean returnDocumentResponses) {
+    this.allInsertions = List.copyOf(allAttemptedInsertions);
+    this.returnDocumentResponses = returnDocumentResponses;
+
+    this.successfulInsertions = new ArrayList<>(allAttemptedInsertions.size());
+    this.failedInsertions = new ArrayList<>(allAttemptedInsertions.size());
+  }
+
+  enum InsertionStatus {
+    OK,
+    ERROR,
+    SKIPPED
+  }
+
+  // TODO AARON - I think this is for the document responses, confirm
+  @JsonPropertyOrder({"_id", "status", "errorsIdx"})
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  record InsertionResult(DocRowIdentifer _id, InsertionStatus status, Integer errorsIdx) {}
+
+  /** {@inheritDoc} */
+  @Override
+  public CommandResult get() {
+    // Sort on the insert position to rebuild the order we of the documents from the insert.
+    // used for both legacy and new style output
+    Collections.sort(failedInsertions);
+    // TODO AARON used to only sort the success list when not returning detaile responses, check OK
+    Collections.sort(successfulInsertions);
+
+    if (!returnDocumentResponses) { // legacy output, limited to ids, error messages
+      List<CommandResult.Error> errors =
+          failedInsertions.isEmpty()
+              ? null
+              : failedInsertions.stream().map(InsertOperationPage::getOldStyleError).toList();
+
+      // Note: See DocRowIdentifer, it has an attribute that will be called for JSON serialization
+      List<DocRowIdentifer> insertedIds =
+          successfulInsertions.stream()
+              .map(InsertAttempt::docRowID)
+              .map(Optional::orElseThrow)
+              .toList();
+      return new CommandResult(null, Map.of(CommandStatus.INSERTED_IDS, insertedIds), errors);
+    }
+
+    // UPTO THIS AARON
+
+    // New style output: detailed responses.
+    InsertionResult[] results = new InsertionResult[allInsertions.size()];
+    List<CommandResult.Error> errors = new ArrayList<>();
+
+    // Results array filled in order: first successful insertions
+    for (InsertAttempt okInsertion : successfulInsertions) {
+      results[okInsertion.position()] =
+          new InsertionResult(okInsertion.docRowID().orElseThrow(), InsertionStatus.OK, null);
+    }
+    // Second: failed insertions; output in order of insertion
+    for (InsertAttempt failedInsertion : failedInsertions) {
+      // TODO AARON - confirm the null handling in the getError
+      CommandResult.Error error = getError(failedInsertion.failure().orElse(null));
+
+      // We want to avoid adding the same error multiple times, so we keep track of the index:
+      // either one exists, use it; or if not, add it and use the new index.
+      int errorIdx = errors.indexOf(error);
+      if (errorIdx < 0) { // new non-dup error; add it
+        errorIdx = errors.size(); // will be appended at the end
+        errors.add(error);
+      }
+      results[failedInsertion.position()] =
+          new InsertionResult(
+              failedInsertion.docRowID().orElseThrow(), InsertionStatus.ERROR, errorIdx);
+    }
+
+    // And third, if any, skipped insertions; those that were not attempted (f.ex due
+    // to failure for ordered inserts)
+    for (int i = 0; i < results.length; i++) {
+      if (null == results[i]) {
+        results[i] =
+            new InsertionResult(
+                allInsertions.get(i).docRowID().orElseThrow(), InsertionStatus.SKIPPED, null);
+      }
+    }
+    return new CommandResult(
+        null, Map.of(CommandStatus.DOCUMENT_RESPONSES, Arrays.asList(results)), errors);
+  }
+
+  private static CommandResult.Error getOldStyleError(InsertAttempt insertAttempt) {
+    String message =
+        "Failed to insert document with _id %s: %s"
+            .formatted(
+                insertAttempt.docRowID().orElseThrow(),
+                insertAttempt
+                    .failure()
+                    .map(Throwable::getMessage)
+                    .orElse("InsertAttempt failure was null."));
+
+    /// TODO: confirm the null hanlding in the getMapperWithMessageFunction
+    // passing null is what would have happened before changing to optional
+    return ThrowableToErrorMapper.getMapperWithMessageFunction()
+        .apply(insertAttempt.failure().orElse(null), message);
+  }
+
+  private static CommandResult.Error getError(Throwable throwable) {
+    // TODO AARON - confirm we have two different error message paths
+    return ThrowableToErrorMapper.getMapperWithMessageFunction()
+        .apply(throwable, throwable.getMessage());
+  }
+
+  /**
+   * Aggregates the result of the insert operation into this object.
+   *
+   * @param insertion Document insertion attempt
+   */
+  public void aggregate(InsertAttempt insertion) {
+    // TODO: AARON: confirm this should not add to the allInsertions list. It would seem better if
+    // it did
+    insertion
+        .failure()
+        .ifPresentOrElse(
+            throwable -> failedInsertions.add(insertion),
+            () -> successfulInsertions.add(insertion));
+  }
+}