-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Insert working, one and many - some Test failing
- Loading branch information
Showing
19 changed files
with
379 additions
and
42 deletions.
There are no files selected for viewing
10 changes: 9 additions & 1 deletion
10
src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/InsertAttempt.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
94 changes: 94 additions & 0 deletions
94
...in/java/io/stargate/sgv2/jsonapi/service/operation/model/tables/InsertTableOperation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package io.stargate.sgv2.jsonapi.service.operation.model.tables; | ||
|
|
||
| import static com.datastax.oss.driver.api.querybuilder.QueryBuilder.*; | ||
|
|
||
| import com.datastax.oss.driver.api.core.CqlIdentifier; | ||
| import com.datastax.oss.driver.api.core.cql.SimpleStatement; | ||
| import com.datastax.oss.driver.api.querybuilder.term.Term; | ||
| import io.smallrye.mutiny.Multi; | ||
| 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.request.DataApiRequestInfo; | ||
| import io.stargate.sgv2.jsonapi.service.cqldriver.executor.QueryExecutor; | ||
| import io.stargate.sgv2.jsonapi.service.cqldriver.executor.TableSchemaObject; | ||
| import io.stargate.sgv2.jsonapi.service.operation.model.InsertOperationPage; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.tables.WriteableTableRow; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.function.Function; | ||
| import java.util.function.Supplier; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| public class InsertTableOperation extends TableMutationOperation { | ||
|
|
||
| private final List<TableInsertAttempt> insertAttempts; | ||
|
|
||
| // TODO AARON JSON to start with, need a document object | ||
| public InsertTableOperation( | ||
| CommandContext<TableSchemaObject> commandContext, List<TableInsertAttempt> insertAttempts) { | ||
| super(commandContext); | ||
| this.insertAttempts = List.copyOf(insertAttempts); | ||
| } | ||
|
|
||
| @Override | ||
| public Uni<Supplier<CommandResult>> execute( | ||
| DataApiRequestInfo dataApiRequestInfo, QueryExecutor queryExecutor) { | ||
|
|
||
| // TODO AARON - this is for unordered, copy from Collection InsertOperation insertUnordered | ||
| return Multi.createFrom() | ||
| .iterable(insertAttempts) | ||
| // merge to make it parallel | ||
| .onItem() | ||
| .transformToUniAndMerge( | ||
| insertion -> insertRow(dataApiRequestInfo, queryExecutor, insertion)) | ||
| // then reduce here | ||
| .collect() | ||
| .in(() -> new InsertOperationPage(insertAttempts, false), InsertOperationPage::aggregate) | ||
| // use object identity to resolve to Supplier<CommandResult> | ||
| // TODO AARON - not sure what this is doing, original was .map(i -> i) | ||
| .map(Function.identity()); | ||
| } | ||
|
|
||
| private Uni<TableInsertAttempt> insertRow( | ||
| DataApiRequestInfo dataApiRequestInfo, | ||
| QueryExecutor queryExecutor, | ||
| TableInsertAttempt insertAttempt) { | ||
|
|
||
| // First things first: did we already fail? If so, propagate | ||
| if (insertAttempt.failure().isPresent()) { | ||
| return Uni.createFrom().failure(insertAttempt.failure().get()); | ||
| } | ||
|
|
||
| // bind and execute | ||
| var boundStatement = buildInsertStatement(queryExecutor, insertAttempt.row().orElseThrow()); | ||
|
|
||
| // TODO: AARON What happens to errors here? | ||
| return queryExecutor | ||
| .executeWrite(dataApiRequestInfo, boundStatement) | ||
| .onItemOrFailure() | ||
| .transform( | ||
| (result, t) -> { | ||
| if (t != null) { | ||
| return (TableInsertAttempt) insertAttempt.maybeAddFailure(t); | ||
| } | ||
| // This is where to check result.wasApplied() if this was a LWT | ||
| return insertAttempt; | ||
| }) | ||
| .onItemOrFailure() | ||
| .transform((ia, throwable) -> (TableInsertAttempt) ia.maybeAddFailure(throwable)); | ||
| } | ||
|
|
||
| private SimpleStatement buildInsertStatement(QueryExecutor queryExecutor, WriteableTableRow row) { | ||
|
|
||
| Map<CqlIdentifier, Term> colValues = | ||
| row.allColumnValues().entrySet().stream() | ||
| .collect(Collectors.toMap(Map.Entry::getKey, e -> literal(e.getValue()))); | ||
|
|
||
| return insertInto( | ||
| commandContext.schemaObject().name.keyspace(), | ||
| commandContext.schemaObject().name.table()) | ||
| .valuesByIds(colValues) | ||
| .build(); | ||
| } | ||
| } |
84 changes: 84 additions & 0 deletions
84
...main/java/io/stargate/sgv2/jsonapi/service/operation/model/tables/TableInsertAttempt.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| package io.stargate.sgv2.jsonapi.service.operation.model.tables; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.google.common.base.Preconditions; | ||
| import io.stargate.sgv2.jsonapi.service.operation.model.InsertAttempt; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.DocRowIdentifer; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.WritableDocRow; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.tables.RowId; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.tables.RowShredder; | ||
| import io.stargate.sgv2.jsonapi.service.shredding.tables.WriteableTableRow; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.stream.IntStream; | ||
|
|
||
| public class TableInsertAttempt implements InsertAttempt { | ||
|
|
||
| private final int position; | ||
| private final RowId rowId; | ||
| private final WriteableTableRow row; | ||
| private Throwable failure; | ||
|
|
||
| private TableInsertAttempt(int position, RowId rowId, WriteableTableRow row) { | ||
| this.position = position; | ||
| this.rowId = rowId; | ||
| this.row = row; | ||
| } | ||
|
|
||
| public static List<TableInsertAttempt> create(RowShredder shredder, JsonNode document) { | ||
| return create(shredder, List.of(document)); | ||
| } | ||
|
|
||
| public static List<TableInsertAttempt> create(RowShredder shredder, List<JsonNode> documents) { | ||
| Preconditions.checkNotNull(shredder, "shredder cannot be null"); | ||
| Preconditions.checkNotNull(documents, "documents cannot be null"); | ||
|
|
||
| return IntStream.range(0, documents.size()) | ||
| .mapToObj( | ||
| i -> { | ||
| WriteableTableRow row; | ||
| try { | ||
| row = shredder.shred(documents.get(i)); | ||
| } catch (Exception e) { | ||
| // TODO: need a shredding base excpetion to catch | ||
| // TODO: we need to get the row id, so we can return it in the response | ||
| return (TableInsertAttempt) | ||
| new TableInsertAttempt(i, null, null).maybeAddFailure(e); | ||
| } | ||
| return new TableInsertAttempt(i, row.id(), row); | ||
| }) | ||
| .toList(); | ||
| } | ||
|
|
||
| public Optional<WriteableTableRow> row() { | ||
| return Optional.ofNullable(row); | ||
| } | ||
|
|
||
| @Override | ||
| public int position() { | ||
| return position; | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<DocRowIdentifer> docRowID() { | ||
| return Optional.ofNullable(rowId); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<WritableDocRow> docRow() { | ||
| return Optional.ofNullable(row); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<Throwable> failure() { | ||
| return Optional.ofNullable(failure); | ||
| } | ||
|
|
||
| @Override | ||
| public InsertAttempt maybeAddFailure(Throwable failure) { | ||
| if (this.failure == null) { | ||
| this.failure = failure; | ||
| } | ||
| return this; | ||
| } | ||
| } |
10 changes: 9 additions & 1 deletion
10
.../java/io/stargate/sgv2/jsonapi/service/operation/model/tables/TableMutationOperation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,12 @@ | ||
| package io.stargate.sgv2.jsonapi.service.operation.model.tables; | ||
|
|
||
| import io.stargate.sgv2.jsonapi.api.model.command.CommandContext; | ||
| import io.stargate.sgv2.jsonapi.service.cqldriver.executor.TableSchemaObject; | ||
|
|
||
| /** For now, a marker class / interface for operations that modify data in a table. */ | ||
| abstract class TableMutationOperation extends TableOperation {} | ||
| abstract class TableMutationOperation extends TableOperation { | ||
|
|
||
| protected TableMutationOperation(CommandContext<TableSchemaObject> commandContext) { | ||
| super(commandContext); | ||
| } | ||
| } |
13 changes: 12 additions & 1 deletion
13
src/main/java/io/stargate/sgv2/jsonapi/service/operation/model/tables/TableOperation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,20 @@ | ||
| package io.stargate.sgv2.jsonapi.service.operation.model.tables; | ||
|
|
||
| import com.google.common.base.Preconditions; | ||
| import io.stargate.sgv2.jsonapi.api.model.command.CommandContext; | ||
| import io.stargate.sgv2.jsonapi.service.cqldriver.executor.TableSchemaObject; | ||
| import io.stargate.sgv2.jsonapi.service.operation.model.Operation; | ||
|
|
||
| /** | ||
| * Base for any operations that works with CQL Tables with rows, rather than Collections of | ||
| * Documents | ||
| */ | ||
| abstract class TableOperation implements Operation {} | ||
| abstract class TableOperation implements Operation { | ||
|
|
||
| protected final CommandContext<TableSchemaObject> commandContext; | ||
|
|
||
| protected TableOperation(CommandContext<TableSchemaObject> commandContext) { | ||
| Preconditions.checkNotNull(commandContext, "commandContext cannot be null"); | ||
| this.commandContext = commandContext; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.