Fix #2468 alterCollection, approach 2 (like "alterTable") -- PRIMARY option #2471
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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,20 @@ | ||
| package io.stargate.sgv2.jsonapi.api.model.command.impl; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonTypeName; | ||
| import io.stargate.sgv2.jsonapi.api.model.command.CollectionCommand; | ||
| import io.stargate.sgv2.jsonapi.api.model.command.CommandName; | ||
| import io.stargate.sgv2.jsonapi.api.model.command.NoOptionsCommand; | ||
| import org.eclipse.microprofile.openapi.annotations.media.Schema; | ||
|
|
||
| @Schema( | ||
| description = | ||
| "Command that alters mutable settings of an existing collection. Currently supports enabling the 'lexical' feature.") | ||
| @JsonTypeName(CommandName.Names.ALTER_COLLECTION) | ||
| public record AlterCollectionCommand(AlterCollectionOperation operation) | ||
| implements CollectionCommand, NoOptionsCommand { | ||
|
|
||
| @Override | ||
| public CommandName commandName() { | ||
| return CommandName.ALTER_COLLECTION; | ||
| } | ||
| } | ||
This file contains hidden or 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,14 @@ | ||
| package io.stargate.sgv2.jsonapi.api.model.command.impl; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonSubTypes; | ||
| import com.fasterxml.jackson.annotation.JsonTypeInfo; | ||
|
|
||
| /** | ||
| * Polymorphic operation payload for {@link AlterCollectionCommand}. Each operation is represented | ||
| * by a record implementing this interface; Jackson selects the concrete subtype by the wrapper key | ||
| * (e.g. {@code "enableLexical"}). Mirrors {@link AlterTableOperation}. | ||
| */ | ||
| @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.WRAPPER_OBJECT) | ||
| @JsonSubTypes({@JsonSubTypes.Type(value = AlterCollectionOperationImpl.EnableLexical.class)}) | ||
| public sealed interface AlterCollectionOperation | ||
|
Contributor
Author
There was a problem hiding this comment. Modeled after |
||
| permits AlterCollectionOperationImpl.EnableLexical {} | ||
This file contains hidden or 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,27 @@ | ||
| package io.stargate.sgv2.jsonapi.api.model.command.impl; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonInclude; | ||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import com.fasterxml.jackson.annotation.JsonTypeName; | ||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import java.util.Map; | ||
| import javax.annotation.Nullable; | ||
| import org.eclipse.microprofile.openapi.annotations.media.Schema; | ||
|
|
||
| /** Each operation that {@link AlterCollectionCommand} understands is represented by a record. */ | ||
| public class AlterCollectionOperationImpl { | ||
|
Contributor
Author
There was a problem hiding this comment. and this is modeled after |
||
|
|
||
| @Schema(description = "Operation to enable the lexical search feature on a collection.") | ||
| @JsonTypeName("enableLexical") | ||
| public record EnableLexical( | ||
| @Schema( | ||
| description = | ||
| "Analyzer to use for '$lexical' field: either String (name of a pre-defined analyzer), or JSON Object to specify custom one. Default: 'standard'.", | ||
| defaultValue = "standard", | ||
| oneOf = {String.class, Map.class}) | ||
| @JsonInclude(JsonInclude.Include.NON_NULL) | ||
| @JsonProperty("analyzer") | ||
| @Nullable | ||
| JsonNode analyzerDef) | ||
| implements AlterCollectionOperation {} | ||
| } | ||
This file contains hidden or 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 hidden or 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 hidden or 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,238 @@ | ||
| package io.stargate.sgv2.jsonapi.service.operation.collections; | ||
|
|
||
| import static io.stargate.sgv2.jsonapi.exception.ErrorFormatters.errVars; | ||
|
|
||
| import com.datastax.oss.driver.api.core.CqlIdentifier; | ||
| import com.datastax.oss.driver.api.core.cql.AsyncResultSet; | ||
| import com.datastax.oss.driver.api.core.cql.SimpleStatement; | ||
| import com.datastax.oss.driver.api.core.metadata.Metadata; | ||
| import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata; | ||
| import com.fasterxml.jackson.core.JacksonException; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| 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.RequestContext; | ||
| import io.stargate.sgv2.jsonapi.config.DatabaseLimitsConfig; | ||
| import io.stargate.sgv2.jsonapi.config.constants.DocumentConstants; | ||
| import io.stargate.sgv2.jsonapi.config.constants.TableCommentConstants; | ||
| import io.stargate.sgv2.jsonapi.exception.DatabaseException; | ||
| import io.stargate.sgv2.jsonapi.exception.SchemaException; | ||
| import io.stargate.sgv2.jsonapi.service.cqldriver.executor.QueryExecutor; | ||
| import io.stargate.sgv2.jsonapi.service.operation.Operation; | ||
| import io.stargate.sgv2.jsonapi.service.schema.collections.CollectionLexicalDef; | ||
| import io.stargate.sgv2.jsonapi.service.schema.collections.CollectionSchemaObject; | ||
| import io.stargate.sgv2.jsonapi.service.schema.collections.CollectionTableComment; | ||
| import java.time.Duration; | ||
| import java.util.Optional; | ||
| import java.util.function.Supplier; | ||
|
|
||
| /** | ||
| * Operation that enables the lexical feature on an existing collection by adding the {@code | ||
| * query_lexical_value} column, creating an analyzed SAI index on it, and updating the table | ||
| * "comment" JSON to record the new lexical config. | ||
| * | ||
| * <p>When {@link #noOp} is true the operation returns success without executing any DDL: this is | ||
| * used for the "already enabled with same settings" case. | ||
| * | ||
| * <p><b>No rollback on partial failure.</b> If e.g. ADD COLUMN succeeds but CREATE INDEX fails, the | ||
| * column is left in place and the failure is propagated to the caller. This matches {@link | ||
| * CreateCollectionOperation}'s behavior and is intentional: | ||
| * | ||
| * <ul> | ||
| * <li>Reverse DDL (DROP COLUMN, DROP INDEX) is itself fallible — a rollback that fails leaves the | ||
| * schema in a worse state than the original partial failure and obscures the root cause. | ||
| * <li>The operation is retry-safe: existence is checked against freshly-fetched metadata, so ADD | ||
| * COLUMN is skipped when the column already exists, CREATE INDEX uses {@code IF NOT EXISTS}, | ||
| * and the comment write is a plain overwrite. Re-issuing the same {@code alterCollection} | ||
| * command after the underlying issue is resolved completes the unfinished steps without | ||
| * failing on the finished ones. (The backend does not support {@code ADD IF NOT EXISTS}, so | ||
| * the skip relies on the metadata check.) | ||
| * <li>Users get a consistent mental model with {@code createCollection}, which has the same | ||
| * partial-failure semantics. | ||
| * </ul> | ||
| * | ||
| * <p>The comment is updated last, so an interrupted run can leave the column/index present while | ||
| * {@code findCollections} still reports lexical as disabled; a successful retry reconciles this | ||
| * (see {@code trulyEnabled} in {@code AlterCollectionCommandResolver}). | ||
| */ | ||
| public record AlterCollectionLexicalOperation( | ||
| CommandContext<CollectionSchemaObject> commandContext, | ||
| ObjectMapper objectMapper, | ||
| DatabaseLimitsConfig dbLimitsConfig, | ||
| int ddlDelayMillis, | ||
| CollectionLexicalDef newLexicalConfig, | ||
| boolean noOp) | ||
| implements Operation<CollectionSchemaObject> { | ||
|
|
||
| private static final CqlIdentifier LEXICAL_COLUMN = | ||
| CqlIdentifier.fromInternal(DocumentConstants.Columns.LEXICAL_INDEX_COLUMN_NAME); | ||
|
|
||
| @Override | ||
| public Uni<Supplier<CommandResult>> execute( | ||
| RequestContext requestContext, QueryExecutor queryExecutor) { | ||
|
|
||
| if (noOp) { | ||
| return Uni.createFrom().<Supplier<CommandResult>>item(new SchemaChangeResult(true)); | ||
| } | ||
|
|
||
| final CollectionSchemaObject schemaObject = commandContext.schemaObject(); | ||
| final String keyspace = schemaObject.tableMetadata().getKeyspace().asInternal(); | ||
| final String table = schemaObject.tableMetadata().getName().asInternal(); | ||
|
|
||
| final String newComment; | ||
| try { | ||
| newComment = buildUpdatedComment(schemaObject); | ||
| } catch (JacksonException | RuntimeException e) { | ||
| // Resolver guarantees a V1 comment; if reading/updating still fails, surface a clean error | ||
| // rather than a raw JacksonException/IllegalStateException. | ||
| return Uni.createFrom() | ||
| .failure( | ||
| DatabaseException.Code.CORRUPTED_COLLECTION_SCHEMA.get( | ||
| errVars( | ||
| schemaObject, | ||
| map -> | ||
| map.put( | ||
| "errorMessage", | ||
| "Unable to update collection 'comment' to enable lexical: " | ||
| + e.getMessage())))); | ||
| } | ||
|
|
||
| // Base all existence decisions on freshly-fetched metadata rather than the resolve-time | ||
| // snapshot, so a column/index left by an interrupted prior run (or a concurrent op) is seen | ||
| // here. This is also where we pre-flight the DB-wide index limit, before running any DDL. | ||
| return queryExecutor | ||
| .getDriverMetadata(requestContext) | ||
| .map(Metadata::getKeyspaces) | ||
| .flatMap( | ||
| allKeyspaces -> { | ||
| final TableMetadata currentTable = | ||
| Optional.ofNullable(allKeyspaces.get(schemaObject.tableMetadata().getKeyspace())) | ||
| .flatMap(ks -> ks.getTable(schemaObject.tableMetadata().getName())) | ||
| .orElse(schemaObject.tableMetadata()); | ||
|
|
||
| final boolean columnExists = currentTable.getColumn(LEXICAL_COLUMN).isPresent(); | ||
| final boolean indexExists = | ||
| currentTable | ||
| .getIndexes() | ||
| .containsKey( | ||
| CqlIdentifier.fromInternal( | ||
| CreateCollectionOperation.lexicalIndexName(table))); | ||
|
|
||
| // Only an absent index is net-new, so only then enforce the limit (mirrors | ||
| // CreateCollectionOperation): going over fails with TOO_MANY_INDEXES_FOR_COLLECTION | ||
| // before any DDL, not a generic error from a failed CREATE INDEX. | ||
| if (!indexExists) { | ||
| final int saisUsed = | ||
| allKeyspaces.values().stream() | ||
| .flatMap(ks -> ks.getTables().values().stream()) | ||
| .mapToInt(t -> t.getIndexes().size()) | ||
| .sum(); | ||
| // enableLexical adds exactly one SAI (the analyzed lexical index). | ||
| if (saisUsed + 1 > dbLimitsConfig.indexesAvailablePerDatabase()) { | ||
| return Uni.createFrom() | ||
| .<Supplier<CommandResult>>failure( | ||
| SchemaException.Code.TOO_MANY_INDEXES_FOR_COLLECTION.get( | ||
| errVars(schemaObject, map -> map.put("indexesPerCollection", "1")))); | ||
| } | ||
| } | ||
|
|
||
| return executeLexicalDdl( | ||
| requestContext, queryExecutor, keyspace, table, newComment, columnExists); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Runs the enable-lexical DDL: ADD COLUMN (skipped when it already exists), CREATE CUSTOM INDEX | ||
| * IF NOT EXISTS, then ALTER TABLE WITH comment, spaced by {@link #ddlDelayMillis}. The {@code | ||
| * columnAlreadyExists} flag is derived from freshly-fetched metadata so a leftover column is | ||
| * skipped rather than failing the (plain) ADD — the backend does not support {@code ADD IF NOT | ||
| * EXISTS}. | ||
| */ | ||
| private Uni<Supplier<CommandResult>> executeLexicalDdl( | ||
| RequestContext requestContext, | ||
| QueryExecutor queryExecutor, | ||
| String keyspace, | ||
| String table, | ||
| String newComment, | ||
| boolean columnAlreadyExists) { | ||
|
|
||
| SimpleStatement createIndexStmt = | ||
| CreateCollectionOperation.buildLexicalIndexStatement( | ||
| keyspace, table, newLexicalConfig, /* ifNotExists */ true); | ||
|
|
||
| // Cassandra does not accept bind parameters for table options like `comment`, so the comment | ||
| // JSON is embedded directly into the CQL (as createCollection does); single quotes are doubled | ||
| // to keep the string literal valid. | ||
| SimpleStatement alterCommentStmt = | ||
| SimpleStatement.newInstance( | ||
| "ALTER TABLE \"%s\".\"%s\" WITH comment = '%s'" | ||
| .formatted(keyspace, table, newComment.replace("'", "''"))); | ||
|
|
||
| final Duration delay = Duration.ofMillis(ddlDelayMillis > 0 ? ddlDelayMillis : 100); | ||
|
|
||
| Uni<AsyncResultSet> pipeline; | ||
| if (columnAlreadyExists) { | ||
| pipeline = queryExecutor.executeCreateSchemaChange(requestContext, createIndexStmt); | ||
| } else { | ||
| SimpleStatement addColumnStmt = | ||
| SimpleStatement.newInstance( | ||
| "ALTER TABLE \"%s\".\"%s\" ADD %s text" | ||
| .formatted(keyspace, table, DocumentConstants.Columns.LEXICAL_INDEX_COLUMN_NAME)); | ||
| pipeline = | ||
| queryExecutor | ||
| .executeCreateSchemaChange(requestContext, addColumnStmt) | ||
| .onItem() | ||
| .delayIt() | ||
| .by(delay) | ||
| .onItem() | ||
| .transformToUni( | ||
| r1 -> queryExecutor.executeCreateSchemaChange(requestContext, createIndexStmt)); | ||
| } | ||
|
|
||
| return pipeline | ||
| .onItem() | ||
| .delayIt() | ||
| .by(delay) | ||
| .onItem() | ||
| .transformToUni( | ||
| r2 -> queryExecutor.executeCreateSchemaChange(requestContext, alterCommentStmt)) | ||
| .map(r3 -> new SchemaChangeResult(true)); | ||
| } | ||
|
|
||
| /** | ||
| * Reads the current table comment JSON and surgically replaces the {@code | ||
| * collection.options.lexical} sub-node, leaving all other options (vector / indexing / id / | ||
| * rerank / unknown fields) untouched. | ||
| * | ||
| * <p>The resolver guarantees we are operating on a V1-shaped comment (legacy/V0 collections are | ||
| * rejected before reaching the operation). | ||
| */ | ||
| private String buildUpdatedComment(CollectionSchemaObject schemaObject) throws JacksonException { | ||
| final String comment = CollectionTableComment.rawComment(schemaObject.tableMetadata()); | ||
| if (comment == null || comment.isBlank()) { | ||
| // Defensive: resolver should have rejected this case. | ||
| throw new IllegalStateException( | ||
| "Cannot alter collection: table comment is empty; expected V1 schema"); | ||
| } | ||
|
|
||
| final ObjectNode rootNode = (ObjectNode) objectMapper.readTree(comment); | ||
| final ObjectNode collectionNode = | ||
| (ObjectNode) rootNode.get(TableCommentConstants.TOP_LEVEL_KEY); | ||
| if (collectionNode == null) { | ||
| // Defensive: resolver should have rejected this case. | ||
| throw new IllegalStateException( | ||
| "Cannot alter collection: comment does not have '" | ||
| + TableCommentConstants.TOP_LEVEL_KEY | ||
| + "' node"); | ||
| } | ||
| ObjectNode optionsNode = (ObjectNode) collectionNode.get(TableCommentConstants.OPTIONS_KEY); | ||
| if (optionsNode == null) { | ||
| optionsNode = objectMapper.createObjectNode(); | ||
| collectionNode.set(TableCommentConstants.OPTIONS_KEY, optionsNode); | ||
| } | ||
| optionsNode.putPOJO(TableCommentConstants.COLLECTION_LEXICAL_CONFIG_KEY, newLexicalConfig); | ||
| return objectMapper.writeValueAsString(rootNode); | ||
| } | ||
| } |
This file contains hidden or 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.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Modeled after
AlterTableCommand, same approach with polymorphich operation types.