diff --git a/src/main/java/ru/rt/restream/reindexer/Namespace.java b/src/main/java/ru/rt/restream/reindexer/Namespace.java index ecbb3163..62ba9818 100644 --- a/src/main/java/ru/rt/restream/reindexer/Namespace.java +++ b/src/main/java/ru/rt/restream/reindexer/Namespace.java @@ -15,6 +15,8 @@ */ package ru.rt.restream.reindexer; +import java.util.Optional; + /** * Provides methods for manipulating Reindexer namespace data. * @@ -32,44 +34,56 @@ public interface Namespace { /** * Inserts the given item data. * - * @param item the item data + * @param item the item data + * @return the inserted {@link T} item */ - void insert(T item); + T insert(T item); /** - * Inserts the given json-formatted item data + * Inserts the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * - * @param item the json-formatted item data + * @param item the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - void insert(String item); + Optional insert(String item); /** * Inserts or updates the given item data. * - * @param item the item data + * @param item the item data + * @return the upserted {@link T} item */ - void upsert(T item); + T upsert(T item); /** * Inserts or updates the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * - * @param item the json-formatted item data + * @param item the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - void upsert(String item); + Optional upsert(String item); /** * Updates the given item data. * - * @param item the item data + * @param item the item data + * @return the updated {@link T} item */ - void update(T item); + T update(T item); /** * Updates the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * - * @param item the json-formatted item data + * @param item the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - void update(String item); + Optional update(String item); /** * Deletes the given item data. diff --git a/src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java b/src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java index 9d1e06d5..ef045d10 100644 --- a/src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java +++ b/src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java @@ -129,10 +129,10 @@ public T next() { } private S readItem(ItemReader reader, Query queryContext) { - ItemParams params = readItemParams(); - Query itemQueryContext = getItemQueryContext(queryContext, params.nsId); + QueryResult.ItemParams params = readItemParams(); + Query itemQueryContext = getItemQueryContext(queryContext, params.getNsId()); S item = readItemData(params, reader); - readJoinedItems(item, itemQueryContext, params.nsId); + readJoinedItems(item, itemQueryContext, params.getNsId()); return item; } @@ -149,10 +149,10 @@ private Query getItemQueryContext(Query defaultQueryContext, int nsId) { return defaultQueryContext; } - private S readItemData(ItemParams params, ItemReader reader) { - if (params.cptr != 0) { - ByteBuffer nativeBuffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), params.cptr, - params.nsId); + private S readItemData(QueryResult.ItemParams params, ItemReader reader) { + if (params.getCptr() != 0) { + ByteBuffer nativeBuffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), params.getCptr(), + params.getNsId()); return reader.readItem(nativeBuffer); } @@ -210,7 +210,7 @@ private void readJoinedItemsV1(Object item, int nsId) { CjsonItemReader joinedItemReader = newItemReader(joinedNamespace.getItemClass(), joinedNamespace); List subItems = new ArrayList<>(itemsCount); for (int j = 0; j < itemsCount; j++) { - ItemParams subItemParams = readItemParams(); + QueryResult.ItemParams subItemParams = readItemParams(); subItems.add(readItemData(subItemParams, joinedItemReader)); } @@ -230,19 +230,19 @@ private void readJoinedItemsV1(Object item, int nsId) { private void skipJoinedItems(int itemsCount) { for (int i = 0; i < itemsCount; i++) { - ItemParams itemParams = readItemParams(); - if (itemParams.cptr == 0) { + QueryResult.ItemParams itemParams = readItemParams(); + if (itemParams.getCptr() == 0) { int length = (int) buffer.getUInt32(); buffer.skip(length); } - readJoinedItems(null, null, itemParams.nsId); + readJoinedItems(null, null, itemParams.getNsId()); } } private void skipJoinedItemsV1(int itemsCount) { for (int i = 0; i < itemsCount; i++) { - ItemParams itemParams = readItemParams(); - if (itemParams.cptr == 0) { + QueryResult.ItemParams itemParams = readItemParams(); + if (itemParams.getCptr() == 0) { int length = (int) buffer.getUInt32(); buffer.skip(length); } @@ -312,42 +312,12 @@ private void writeJoinResult(Object item, String fieldName, List subItem } } - private ItemParams readItemParams() { - ItemParams params = new ItemParams(); - - if (queryResult.isWithItemId()) { - params.id = buffer.getVarUInt(); - params.version = buffer.getVarUInt(); - } - - if (queryResult.isWithNsId()) { - params.nsId = (int) buffer.getVarUInt(); - } - + private QueryResult.ItemParams readItemParams() { + QueryResult.ItemParams itemParams = queryResult.readItemParams(); if (queryResult.isWithRank()) { - //used for full-text search - if (queryResult.getRankFormat() == 0) { - params.rank = buffer.getFloat(); - currentRank = params.rank; - } else { - params.rank = buffer.getVarUInt(); - } + currentRank = itemParams.getRank(); } - - if (queryResult.isWithShardId()) { - if (queryResult.getShardId() != Consts.SHARDING_PROXY_OFF) { - params.shardId = queryResult.getShardId(); - } else { - params.shardId = (int) buffer.getVarUInt(); - } - - } - - if (queryResult.isWithResultsPtr()) { - params.cptr = buffer.getUInt64(); - } - - return params; + return itemParams; } private boolean needFetch() { @@ -395,37 +365,4 @@ public void close() { closed = true; } - private static class ItemParams { - private float rank = -1; - private long id = -1; - private int shardId = Consts.SHARDING_PROXY_OFF; - private long version = -1; - private long cptr; - private int nsId = 0; - - public float getRank() { - return rank; - } - - public long getId() { - return id; - } - - public int getShardId() { - return shardId; - } - - public long getVersion() { - return version; - } - - public long getCptr() { - return cptr; - } - - public int getNsId() { - return nsId; - } - } - } diff --git a/src/main/java/ru/rt/restream/reindexer/Reindexer.java b/src/main/java/ru/rt/restream/reindexer/Reindexer.java index 5469ea7a..46711901 100644 --- a/src/main/java/ru/rt/restream/reindexer/Reindexer.java +++ b/src/main/java/ru/rt/restream/reindexer/Reindexer.java @@ -22,15 +22,21 @@ import ru.rt.restream.reindexer.binding.Consts; import ru.rt.restream.reindexer.binding.QueryResult; import ru.rt.restream.reindexer.binding.RequestContext; +import ru.rt.restream.reindexer.binding.cproto.ByteBuffer; import ru.rt.restream.reindexer.binding.cproto.ItemSerializer; +import ru.rt.restream.reindexer.binding.cproto.cjson.CjsonItemReader; +import ru.rt.restream.reindexer.binding.cproto.cjson.CtagMatcher; import ru.rt.restream.reindexer.binding.cproto.cjson.PayloadType; import ru.rt.restream.reindexer.binding.definition.IndexDefinition; import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; import ru.rt.restream.reindexer.exceptions.IndexConflictException; import ru.rt.restream.reindexer.exceptions.StateInvalidatedException; +import ru.rt.restream.reindexer.expression.PreceptExpression; +import ru.rt.restream.reindexer.util.NativeUtils; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; public class Reindexer implements AutoCloseable { @@ -105,6 +111,8 @@ private ReindexerNamespace doOpenNamespace(String name, NamespaceOptions .indexes(reindexScanner.parseIndexes(itemClass)) .reindexer(this) .build(); + // Initialize the namespace instance. + namespace.initialize(); try { binding.openNamespace(NamespaceDefinition.fromNamespace(namespace)); @@ -142,19 +150,23 @@ public void dropNamespace(String namespaceName) { * @param the item type * @param namespaceName the namespace name * @param item the item data + * @return the inserted {@link T} item */ - public void insert(String namespaceName, T item) { - modifyItem(namespaceName, item, MODE_INSERT, Consts.FORMAT_C_JSON); + public T insert(String namespaceName, T item) { + return modifyItem(namespaceName, item, MODE_INSERT, Consts.FORMAT_C_JSON); } /** * Inserts the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * * @param namespaceName the namespace name * @param json the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - public void insert(String namespaceName, String json) { - modifyItem(namespaceName, json, MODE_INSERT, Consts.FORMAT_JSON); + public Optional insert(String namespaceName, String json) { + return Optional.ofNullable(modifyItem(namespaceName, json, MODE_INSERT, Consts.FORMAT_JSON)); } /** @@ -163,19 +175,23 @@ public void insert(String namespaceName, String json) { * @param the item type * @param namespaceName the namespace name * @param item the item data + * @return the updated {@link T} item */ - public void update(String namespaceName, T item) { - modifyItem(namespaceName, item, MODE_UPDATE, Consts.FORMAT_C_JSON); + public T update(String namespaceName, T item) { + return modifyItem(namespaceName, item, MODE_UPDATE, Consts.FORMAT_C_JSON); } /** * Updates the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * * @param namespaceName the namespace name * @param json the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - public void update(String namespaceName, String json) { - modifyItem(namespaceName, json, MODE_UPDATE, Consts.FORMAT_JSON); + public Optional update(String namespaceName, String json) { + return Optional.ofNullable(modifyItem(namespaceName, json, MODE_UPDATE, Consts.FORMAT_JSON)); } /** @@ -184,19 +200,23 @@ public void update(String namespaceName, String json) { * @param the item type * @param namespaceName the namespace name * @param item the item data + * @return the upserted {@link T} item */ - public void upsert(String namespaceName, T item) { - modifyItem(namespaceName, item, MODE_UPSERT, Consts.FORMAT_C_JSON); + public T upsert(String namespaceName, T item) { + return modifyItem(namespaceName, item, MODE_UPSERT, Consts.FORMAT_C_JSON); } /** * Inserts or updates the given json-formatted item data. + * Returns the server-generated {@link T} item, which may contain fields not present + * in the input JSON, or {@link Optional#empty()} if the server returns no data. * * @param namespaceName the namespace name * @param json the json-formatted item data + * @return the server-generated {@link T} item, or {@link Optional#empty()} if no data was returned */ - public void upsert(String namespaceName, String json) { - modifyItem(namespaceName, json, MODE_UPSERT, Consts.FORMAT_JSON); + public Optional upsert(String namespaceName, String json) { + return Optional.ofNullable(modifyItem(namespaceName, json, MODE_UPSERT, Consts.FORMAT_JSON)); } /** @@ -253,14 +273,7 @@ public QueryResultIterator execSql(String query, Class itemClass) { PayloadType pt = namespace.getPayloadType(); long[] ptVersions = pt == null ? new long[]{0} : new long[]{pt.getVersion() ^ pt.getStateToken()}; RequestContext ctx = binding.select(query, false, Integer.MAX_VALUE, ptVersions); - QueryResult queryResult = ctx.getQueryResult(); - for (PayloadType payloadType : queryResult.getPayloadTypes()) { - PayloadType currentPayloadType = namespace.getPayloadType(); - if (currentPayloadType == null || currentPayloadType.getVersion() < payloadType.getVersion() - || currentPayloadType.getStateToken() != payloadType.getStateToken()) { - namespace.updatePayloadType(payloadType); - } - } + updatePayloadType(ctx.getQueryResult(), namespace); return new QueryResultIterator<>(namespace, itemClass, ctx, null, Integer.MAX_VALUE); } @@ -296,6 +309,17 @@ public void dropIndex(String namespaceName, String indexName) { binding.dropIndex(namespaceName, indexName); } + /** + * Registers a {@link ReindexerPrecept} to the namespace to apply to all items persisted in the namespace. + * + * @param namespaceName the namespace name, cannot be {@literal null} + * @param precept the {@link ReindexerPrecept} to register, cannot be {@literal null} + */ + public void registerPrecept(String namespaceName, ReindexerPrecept precept) { + ReindexerNamespace namespace = getNamespace(namespaceName); + namespace.registerPrecept(precept); + } + private ReindexerNamespace getNamespace(String namespaceName, Class itemClass) { ReindexerNamespace namespace = namespaceMap.get(namespaceName); if (namespace == null) { @@ -320,21 +344,75 @@ private ReindexerNamespace getNamespace(String namespaceName) { } @SuppressWarnings("unchecked") - private void modifyItem(String namespaceName, T item, int mode, int format) { - ReindexerNamespace namespace = getNamespace(namespaceName); - String[] percepts = namespace.getPrecepts(); + private E modifyItem(String namespaceName, T item, int mode, int format) { + ReindexerNamespace namespace = getNamespace(namespaceName); + PreceptExpression[] precepts = namespace.getPrecepts(item, format); for (int i = 0; i < 2; i++) { try { PayloadType payloadType = namespace.getPayloadType(); int stateToken = payloadType == null ? 0 : payloadType.getStateToken(); ItemSerializer serializer = ItemSerializer.getInstance(item.getClass(), payloadType); byte[] data = serializer.serialize(item); - binding.modifyItem(namespace.getName(), data, format, mode, percepts, stateToken); + QueryResult queryResult = binding.modifyItem(namespace.getName(), data, format, mode, precepts, stateToken); + updatePayloadType(queryResult, namespace); + if (precepts.length > 0 && queryResult.getCount() > 0) { + return unpackItem(item, queryResult, format, namespace); + } break; } catch (StateInvalidatedException e) { updatePayloadType(namespace); } } + if (format == Consts.FORMAT_C_JSON) { + return (E) item; + } + // No server-generated item was returned for JSON-formatted item. + return null; + } + + private void updatePayloadType(QueryResult queryResult, ReindexerNamespace namespace) { + for (PayloadType payloadType : queryResult.getPayloadTypes()) { + PayloadType currentPayloadType = namespace.getPayloadType(); + if (currentPayloadType == null || currentPayloadType.getVersion() < payloadType.getVersion() + || currentPayloadType.getStateToken() != payloadType.getStateToken()) { + namespace.updatePayloadType(payloadType); + } + } + } + + private T unpackItem(Object item, QueryResult queryResult, int format, ReindexerNamespace namespace) { + QueryResult.ItemParams itemParams = queryResult.readItemParams(); + CjsonItemReader reader = createItemReader(item, format, namespace); + if (queryResult.isWithResultsPtr()) { + ByteBuffer buffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), itemParams.getCptr(), itemParams.getNsId()); + try { + return reader.readItem(buffer); + } finally { + NativeUtils.freeNativeBuffer(queryResult.getResultsPtr()); + } + } + ByteBuffer buffer = queryResult.getBuffer(); + int length = (int) buffer.getUInt32(); + return reader.readItem(new ByteBuffer(buffer.getBytes(length)).rewind()); + } + + @SuppressWarnings("unchecked") + private CjsonItemReader createItemReader(Object item, int format, ReindexerNamespace namespace) { + CtagMatcher ctagMatcher = new CtagMatcher(); + PayloadType payloadType = namespace.getPayloadType(); + if (payloadType == null && format == Consts.FORMAT_JSON) { + // When using JSON, Reindexer does not include payloadType in the response + // or validate stateToken. Therefore, StateInvalidatedException is never thrown, + // and payloadType must be explicitly requested to correctly unpack the result. + updatePayloadType(namespace); + payloadType = namespace.getPayloadType(); + } + if (payloadType != null) { + ctagMatcher.read(payloadType); + } + return format == Consts.FORMAT_JSON + ? new CjsonItemReader<>(namespace.getItemClass(), ctagMatcher) + : new CjsonItemReader<>(() -> (T) item, namespace.getItemClass(), ctagMatcher); } private void updatePayloadType(ReindexerNamespace namespace) { diff --git a/src/main/java/ru/rt/restream/reindexer/ReindexerIndex.java b/src/main/java/ru/rt/restream/reindexer/ReindexerIndex.java index a36a6c1e..d53ca95f 100644 --- a/src/main/java/ru/rt/restream/reindexer/ReindexerIndex.java +++ b/src/main/java/ru/rt/restream/reindexer/ReindexerIndex.java @@ -88,7 +88,7 @@ public class ReindexerIndex { * Precept is a special reindexer embedded function, such as serial(), now(). * {@link ru.rt.restream.reindexer.annotations.Serial} */ - private String precept; + private ReindexerPrecept precept; private boolean isArray; @@ -155,4 +155,5 @@ public IndexConfig getConfig() { return null; } } + } diff --git a/src/main/java/ru/rt/restream/reindexer/ReindexerNamespace.java b/src/main/java/ru/rt/restream/reindexer/ReindexerNamespace.java index eea962ac..8fd2a511 100644 --- a/src/main/java/ru/rt/restream/reindexer/ReindexerNamespace.java +++ b/src/main/java/ru/rt/restream/reindexer/ReindexerNamespace.java @@ -15,19 +15,41 @@ */ package ru.rt.restream.reindexer; +import com.jayway.jsonpath.Configuration; +import com.jayway.jsonpath.DocumentContext; +import com.jayway.jsonpath.JsonPath; +import com.jayway.jsonpath.Option; import lombok.AccessLevel; import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import ru.rt.restream.reindexer.binding.Consts; import ru.rt.restream.reindexer.binding.cproto.cjson.PayloadType; +import ru.rt.restream.reindexer.expression.PreceptExpression; +import ru.rt.restream.reindexer.util.BeanPropertyUtils; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Predicate; /** * Contains the reindexer namespace configuration and methods for manipulating the linked reindexer namespace data. */ +@Slf4j @Getter public class ReindexerNamespace implements Namespace { + /** + * Read/Write lock. + */ + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + @Getter(value = AccessLevel.NONE) private final Reindexer reindexer; @@ -81,7 +103,8 @@ public class ReindexerNamespace implements Namespace { * *

Precept is a special reindexer embedded function, such as serial(), now(). */ - private final String[] precepts; + @Getter(AccessLevel.NONE) + private final Map precepts = new HashMap<>(); /** * Current namespace payload type. @@ -111,13 +134,19 @@ private ReindexerNamespace(Builder builder) { this.disableObjCache = builder.disableObjCache; this.objCacheItemsCount = builder.objCacheItemsCount; this.indexes = builder.indexes; - this.precepts = indexes.stream() - .map(ReindexerIndex::getPrecept) - .filter(Objects::nonNull) - .toArray(String[]::new); this.reindexer = builder.reindexer; } + void initialize() { + // Initialize precepts. + for (ReindexerIndex index : indexes) { + ReindexerPrecept precept = index.getPrecept(); + if (precept != null) { + registerPrecept(precept); + } + } + } + /** * Update the current namespace payload type. If current payload type version is lower than passed or payload type * state token is lower than passed. Payload type will be updated. @@ -137,33 +166,33 @@ public Transaction beginTransaction() { } @Override - public void insert(T item) { - reindexer.insert(name, item); + public T insert(T item) { + return reindexer.insert(name, item); } @Override - public void insert(String item) { - reindexer.insert(name, item); + public Optional insert(String item) { + return reindexer.insert(name, item); } @Override - public void upsert(T item) { - reindexer.upsert(name, item); + public T upsert(T item) { + return reindexer.upsert(name, item); } @Override - public void upsert(String item) { - reindexer.upsert(name, item); + public Optional upsert(String item) { + return reindexer.upsert(name, item); } @Override - public void update(T item) { - reindexer.update(name, item); + public T update(T item) { + return reindexer.update(name, item); } @Override - public void update(String item) { - reindexer.update(name, item); + public Optional update(String item) { + return reindexer.update(name, item); } @Override @@ -201,6 +230,70 @@ public void updateSql(String query) { reindexer.updateSql(query); } + void registerPrecept(ReindexerPrecept precept) { + Objects.requireNonNull(precept, "precept cannot be null"); + lock.writeLock().lock(); + try { + ReindexerPrecept existing = precepts.put(precept.getFieldName(), precept); + if (existing == null) { + log.trace("Registered precept {} to namespace {}", precept, name); + } else { + log.trace("Precept {} was overridden to {} for namespace {}", existing, precept, name); + } + } finally { + lock.writeLock().unlock(); + } + } + + PreceptExpression[] getPrecepts(Object item, int format) { + Predicate isNullPredicate = getIsNullPredicate(item, format); + List precepts = new ArrayList<>(); + for (ReindexerPrecept precept : getPrecepts()) { + applyPrecept(precept, isNullPredicate, precepts); + } + return precepts.toArray(new PreceptExpression[0]); + } + + private Predicate getIsNullPredicate(Object item, int format) { + if (format == Consts.FORMAT_JSON) { + Configuration configuration = Configuration.defaultConfiguration() + .addOptions(Option.DEFAULT_PATH_LEAF_TO_NULL); + DocumentContext document = JsonPath.using(configuration).parse((String) item); + // Lookup for a field value using json-path e.g., "$.id". + return it -> document.read("$." + it.getJsonPath()) == null; + } + // Lookup for a field value using a field name. + return it -> BeanPropertyUtils.getProperty(item, it.getFieldName()) == null; + } + + private Collection getPrecepts() { + lock.readLock().lock(); + try { + return precepts.values(); + } finally { + lock.readLock().unlock(); + } + } + + private void applyPrecept(ReindexerPrecept precept, Predicate isNullPredicate, + List precepts) { + // Apply precept according to the strategy. + switch (precept.getApplyStrategy()) { + case ALWAYS: + // Apply precept to all items. + precepts.add(precept.getExpression()); + break; + case WHEN_NULL: + // Apply precept only to items that have null value for the specified field. + if (isNullPredicate.test(precept)) { + precepts.add(precept.getExpression()); + } + break; + default: + throw new IllegalArgumentException("Unknown apply strategy: " + precept.getApplyStrategy()); + } + } + /** * Reindexer namespace builder. */ diff --git a/src/main/java/ru/rt/restream/reindexer/ReindexerPrecept.java b/src/main/java/ru/rt/restream/reindexer/ReindexerPrecept.java new file mode 100644 index 00000000..f9c65ec8 --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/ReindexerPrecept.java @@ -0,0 +1,79 @@ +/* + * Copyright 2020-present Restream + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package ru.rt.restream.reindexer; + +import lombok.AccessLevel; +import lombok.Builder; +import lombok.Getter; +import lombok.NonNull; +import lombok.RequiredArgsConstructor; +import lombok.ToString; +import ru.rt.restream.reindexer.expression.PreceptExpression; + +/** + * Precept is a special reindexer embedded function, such as {@code serial()}, {@code now()}. + * + * @see ru.rt.restream.reindexer.annotations.Serial + */ +@Getter +@Builder +@ToString +@RequiredArgsConstructor(access = AccessLevel.PRIVATE) +public final class ReindexerPrecept { + + /** + * A field name to apply precept to. + */ + @NonNull + private final String fieldName; + + /** + * A JSON path to apply precept to. + */ + @NonNull + private final String jsonPath; + + /** + * A precept expression. + */ + @NonNull + private final PreceptExpression expression; + + /** + * A precept apply strategy. Default is {@link ApplyStrategy#ALWAYS}. + */ + @NonNull + @Builder.Default + private final ApplyStrategy applyStrategy = ApplyStrategy.ALWAYS; + + /** + * Precept apply strategy. + */ + public enum ApplyStrategy { + + /** + * Apply precept always. + */ + ALWAYS, + + /** + * Apply precept only when the field is {@literal null}. + */ + WHEN_NULL + + } + +} diff --git a/src/main/java/ru/rt/restream/reindexer/Transaction.java b/src/main/java/ru/rt/restream/reindexer/Transaction.java index 9c01778f..9d5e20a8 100644 --- a/src/main/java/ru/rt/restream/reindexer/Transaction.java +++ b/src/main/java/ru/rt/restream/reindexer/Transaction.java @@ -23,6 +23,7 @@ import ru.rt.restream.reindexer.binding.cproto.cjson.PayloadType; import ru.rt.restream.reindexer.exceptions.ReindexerExceptionFactory; import ru.rt.restream.reindexer.exceptions.StateInvalidatedException; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.util.ArrayList; import java.util.List; @@ -339,7 +340,7 @@ private CompletableFuture modifyItemAsync(E item, int mode, int format) { private CompletableFuture modifyItemAsyncInternal(E item, int mode, int format, int retryCount) { LOGGER.debug("rx: transaction modifyItemAsync, params=[{}, {}], retryCount={}", item, mode, retryCount); - String[] precepts = namespace.getPrecepts(); + PreceptExpression[] precepts = namespace.getPrecepts(item, format); PayloadType payloadType = namespace.getPayloadType(); int stateToken = payloadType == null ? 0 : payloadType.getStateToken(); ItemSerializer serializer = ItemSerializer.getInstance(item.getClass(), payloadType); @@ -370,7 +371,7 @@ private CompletableFuture failedFuture(Throwable t) { private void modifyItem(E item, int mode, int format) { LOGGER.debug("rx: transaction modifyItem, params=[{}, {}]", item, mode); - String[] precepts = namespace.getPrecepts(); + PreceptExpression[] precepts = namespace.getPrecepts(item, format); for (int i = 0; i < 2; i++) { try { PayloadType payloadType = namespace.getPayloadType(); diff --git a/src/main/java/ru/rt/restream/reindexer/annotations/ReindexAnnotationScanner.java b/src/main/java/ru/rt/restream/reindexer/annotations/ReindexAnnotationScanner.java index 3141a7f4..9a579dfe 100644 --- a/src/main/java/ru/rt/restream/reindexer/annotations/ReindexAnnotationScanner.java +++ b/src/main/java/ru/rt/restream/reindexer/annotations/ReindexAnnotationScanner.java @@ -21,10 +21,12 @@ import ru.rt.restream.reindexer.IndexType; import ru.rt.restream.reindexer.ReindexScanner; import ru.rt.restream.reindexer.ReindexerIndex; +import ru.rt.restream.reindexer.ReindexerPrecept; import ru.rt.restream.reindexer.convert.FieldConverter; import ru.rt.restream.reindexer.convert.util.ConversionUtils; import ru.rt.restream.reindexer.convert.FieldConverterRegistryFactory; import ru.rt.restream.reindexer.exceptions.IndexConflictException; +import ru.rt.restream.reindexer.expression.Expression; import ru.rt.restream.reindexer.fulltext.FullTextConfig; import ru.rt.restream.reindexer.util.BeanPropertyUtils; import ru.rt.restream.reindexer.convert.util.ResolvableType; @@ -141,12 +143,20 @@ && getFieldTypeByClass(fieldInfo.componentType) == COMPOSITE) { String collate = reindex.collate(); CollateMode collateMode = getCollateMode(collate); String sortOrder = getSortOrder(collateMode, collate); - String precept = null; + ReindexerPrecept precept = null; if (field.isAnnotationPresent(Serial.class)) { if (!fieldInfo.isInt()) { throw new IllegalStateException("@Serial is allowed only for int types (i.e INT, INT64)"); } - precept = field.getName() + "=serial()"; + ReindexerPrecept.ApplyStrategy applyStrategy = reindex.isPrimaryKey() + ? ReindexerPrecept.ApplyStrategy.WHEN_NULL + : ReindexerPrecept.ApplyStrategy.ALWAYS; + precept = ReindexerPrecept.builder() + .fieldName(field.getName()) + .jsonPath(jsonPath) + .expression(Expression.string(field.getName() + "=serial()")) + .applyStrategy(applyStrategy) + .build(); } FullTextConfig fullTextConfig = getFullTextConfig(field, reindex.type()); validateUuidIndexHasTypeUuidOrString(reindex, fieldInfo); @@ -317,7 +327,7 @@ private CollateMode getCollateMode(String collate) { private ReindexerIndex createIndex(String reindexPath, List jsonPath, IndexType indexType, FieldType fieldType, boolean isDense, boolean isSparse, boolean isPk, - boolean isArray, CollateMode collateMode, String sortOrder, String precept, + boolean isArray, CollateMode collateMode, String sortOrder, ReindexerPrecept precept, FullTextConfig textConfig, boolean isUuid, boolean isAppendable, boolean isNoColumn, HnswConfig hnswConfig, IvfConfig ivfConfig, VecBfConfig vecBfConfig) { ReindexerIndex index = new ReindexerIndex(); diff --git a/src/main/java/ru/rt/restream/reindexer/annotations/Serial.java b/src/main/java/ru/rt/restream/reindexer/annotations/Serial.java index c4386bba..4a3cb5f9 100644 --- a/src/main/java/ru/rt/restream/reindexer/annotations/Serial.java +++ b/src/main/java/ru/rt/restream/reindexer/annotations/Serial.java @@ -21,9 +21,8 @@ import java.lang.annotation.Target; /** - * Sequence of integer, useful for uniq ID generation. - * This always updates the index value, so use a separate DTO for the update - * if the index is used as the primary key. + * Sequence of integer, useful for unique ID generation. + * When used on primary-key index, the generation is applied only for items with absent id field. */ @Target(ElementType.FIELD) @Retention(RetentionPolicy.RUNTIME) diff --git a/src/main/java/ru/rt/restream/reindexer/binding/Binding.java b/src/main/java/ru/rt/restream/reindexer/binding/Binding.java index d672db53..b75bbc8a 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/Binding.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/Binding.java @@ -17,6 +17,7 @@ import ru.rt.restream.reindexer.binding.definition.IndexDefinition; import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; +import ru.rt.restream.reindexer.expression.PreceptExpression; /** * Binding to Reindexer instance. @@ -133,8 +134,9 @@ public interface Binding { * @param mode modify mode (UPDATE, INSERT, UPSERT, DELETE) * @param precepts precepts (i.e. "id=serial()", "updated_at=now()") * @param stateToken state token + * @return the {@link QueryResult} to use */ - void modifyItem(String namespaceName, byte[] data, int format, int mode, String[] precepts, int stateToken); + QueryResult modifyItem(String namespaceName, byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken); /** * Drop a namespace by name. diff --git a/src/main/java/ru/rt/restream/reindexer/binding/QueryResult.java b/src/main/java/ru/rt/restream/reindexer/binding/QueryResult.java index a817708d..87e20d10 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/QueryResult.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/QueryResult.java @@ -127,4 +127,54 @@ public class QueryResult { */ private int queryFormatVersion = Consts.QUERY_FORMAT_V1; + /** + * Query result item params. + */ + private ItemParams itemParams; + + /** + * Reads an {@link ItemParams} from the query result buffer. + * + * @return the {@link ItemParams} to use + */ + public ItemParams readItemParams() { + ItemParams itemParams = new ItemParams(); + if (withItemId) { + itemParams.id = buffer.getVarUInt(); + itemParams.version = buffer.getVarUInt(); + } + if (withNsId) { + itemParams.nsId = (int) buffer.getVarUInt(); + } + if (withRank) { + // used for full-text search + if (rankFormat == 0) { + itemParams.rank = buffer.getFloat(); + } else { + itemParams.rank = buffer.getVarUInt(); + } + } + if (withShardId) { + if (shardId != Consts.SHARDING_PROXY_OFF) { + itemParams.shardId = shardId; + } else { + itemParams.shardId = (int) buffer.getVarUInt(); + } + } + if (withResultsPtr) { + itemParams.cptr = buffer.getUInt64(); + } + return itemParams; + } + + @Getter + public static final class ItemParams { + private float rank = -1; + private long id = -1; + private int shardId = Consts.SHARDING_PROXY_OFF; + private long version = -1; + private long cptr; + private int nsId = 0; + } + } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/TransactionContext.java b/src/main/java/ru/rt/restream/reindexer/binding/TransactionContext.java index ccf9f410..2a8bdbf7 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/TransactionContext.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/TransactionContext.java @@ -16,6 +16,7 @@ package ru.rt.restream.reindexer.binding; import ru.rt.restream.reindexer.ReindexerResponse; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.util.concurrent.CompletableFuture; @@ -33,7 +34,7 @@ public interface TransactionContext { * @param precepts precepts (i.e. "id=serial()", "updated_at=now()") * @param stateToken state token */ - void modifyItem(byte[] data, int format, int mode, String[] precepts, int stateToken); + void modifyItem(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken); /** * Modifies the item data in the transaction that is associated with the context asynchronously. @@ -45,7 +46,7 @@ public interface TransactionContext { * @param stateToken state token * @return the {@link CompletableFuture} */ - CompletableFuture modifyItemAsync(byte[] data, int format, int mode, String[] precepts, int stateToken); + CompletableFuture modifyItemAsync(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken); /** * Invoke select query in the transaction that is associated with the context. diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java index d9a65ecc..da802007 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/builtin/Builtin.java @@ -23,12 +23,14 @@ import org.slf4j.LoggerFactory; import ru.rt.restream.reindexer.ReindexerResponse; import ru.rt.restream.reindexer.binding.Binding; +import ru.rt.restream.reindexer.binding.QueryResult; import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; import ru.rt.restream.reindexer.binding.cproto.ByteBuffer; import ru.rt.restream.reindexer.binding.definition.IndexDefinition; import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; import ru.rt.restream.reindexer.exceptions.ReindexerExceptionFactory; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.net.URI; import java.nio.charset.StandardCharsets; @@ -121,18 +123,20 @@ public void dropIndex(String namespace, String indexName) { } @Override - public void modifyItem(String namespaceName, byte[] data, int format, int mode, String[] precepts, int stateToken) { + public QueryResult modifyItem(String namespaceName, byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { ByteBuffer args = new ByteBuffer() .putVString(namespaceName) .putVarUInt32(format) .putVarUInt32(mode) .putVarInt32(stateToken); args.putVarUInt32(precepts.length); - for (String precept : precepts) { - args.putVString(precept); + for (PreceptExpression precept : precepts) { + precept.serializePrecept(args); } ReindexerResponse response = adapter.modifyItem(rx, next.getAndIncrement(), timeout.toMillis(), args.bytes(), data); checkResponse(response); + BuiltinRequestContext context = new BuiltinRequestContext(response); + return context.getQueryResult(); } @Override diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinTransactionContext.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinTransactionContext.java index c924314e..fce9e3b9 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinTransactionContext.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/builtin/BuiltinTransactionContext.java @@ -19,11 +19,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.rt.restream.reindexer.ReindexerResponse; -import ru.rt.restream.reindexer.binding.Consts; import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; import ru.rt.restream.reindexer.binding.cproto.ByteBuffer; import ru.rt.restream.reindexer.exceptions.ReindexerExceptionFactory; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.time.Duration; import java.util.concurrent.CompletableFuture; @@ -65,24 +65,24 @@ public BuiltinTransactionContext(BuiltinAdapter adapter, long rx, long transacti } @Override - public void modifyItem(byte[] data, int format, int mode, String[] precepts, int stateToken) { + public void modifyItem(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { ReindexerResponse response = modifyItemInternal(data, format, mode, precepts, stateToken); checkResponse(response); } @Override - public CompletableFuture modifyItemAsync(byte[] data, int format, int mode, String[] precepts, int stateToken) { + public CompletableFuture modifyItemAsync(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { return CompletableFuture.supplyAsync(() -> modifyItemInternal(data, format, mode, precepts, stateToken)); } - private ReindexerResponse modifyItemInternal(byte[] data, int format, int mode, String[] precepts, int stateToken) { + private ReindexerResponse modifyItemInternal(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { ByteBuffer args = new ByteBuffer() .putVarUInt32(format) .putVarUInt32(mode) .putVarInt32(stateToken); args.putVarUInt32(precepts.length); - for (String precept : precepts) { - args.putVString(precept); + for (PreceptExpression precept : precepts) { + precept.serializePrecept(args); } return adapter.modifyItemTx(rx, transactionId, args.bytes(), data); } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java b/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java index 6416908d..a022177e 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/builtin/server/BuiltinServer.java @@ -20,6 +20,7 @@ import org.slf4j.LoggerFactory; import ru.rt.restream.reindexer.ReindexerResponse; import ru.rt.restream.reindexer.binding.Binding; +import ru.rt.restream.reindexer.binding.QueryResult; import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; import ru.rt.restream.reindexer.binding.builtin.Builtin; @@ -28,6 +29,7 @@ import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; import ru.rt.restream.reindexer.exceptions.ReindexerException; import ru.rt.restream.reindexer.exceptions.ReindexerExceptionFactory; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.io.BufferedReader; import java.io.IOException; @@ -142,8 +144,8 @@ public void dropIndex(String namespace, String indexName) { } @Override - public void modifyItem(String namespaceName, byte[] data, int format, int mode, String[] precepts, int stateToken) { - builtin.modifyItem(namespaceName, data, format, mode, precepts, stateToken); + public QueryResult modifyItem(String namespaceName, byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { + return builtin.modifyItem(namespaceName, data, format, mode, precepts, stateToken); } @Override diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java index cb890569..4852428b 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/Cproto.java @@ -21,11 +21,13 @@ import ru.rt.restream.reindexer.ReindexerResponse; import ru.rt.restream.reindexer.binding.Binding; import ru.rt.restream.reindexer.binding.Consts; +import ru.rt.restream.reindexer.binding.QueryResult; import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; import ru.rt.restream.reindexer.binding.cproto.util.ConnectionUtils; import ru.rt.restream.reindexer.binding.definition.IndexDefinition; import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.nio.charset.StandardCharsets; import java.time.Duration; @@ -99,21 +101,24 @@ public void dropIndex(String namespace, String indexName) { * {@inheritDoc} */ @Override - public void modifyItem(String namespaceName, byte[] data, int format, int mode, String[] precepts, - int stateToken) { + public QueryResult modifyItem(String namespaceName, byte[] data, int format, int mode, PreceptExpression[] precepts, + int stateToken) { byte[] packedPercepts = packPrecepts(precepts); - rpcCallNoResults(MODIFY_ITEM, namespaceName, format, data, mode, + Connection connection = pool.getConnection(); + ReindexerResponse rpcResponse = ConnectionUtils.rpcCall(connection, MODIFY_ITEM, namespaceName, format, data, mode, packedPercepts, stateToken, 0); + CprotoRequestContext context = new CprotoRequestContext(rpcResponse, connection, false, connection.queryFormatVersion()); + return context.getQueryResult(); } - private byte[] packPrecepts(String[] precepts) { + private byte[] packPrecepts(PreceptExpression[] precepts) { if (precepts.length == 0) { return EMPTY_BYTE_ARRAY; } ByteBuffer buffer = new ByteBuffer(); buffer.putVarUInt32(precepts.length); - for (String precept : precepts) { - buffer.putVString(precept); + for (PreceptExpression precept : precepts) { + precept.serializePrecept(buffer); } return buffer.bytes(); } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java index e123140e..5d9eadd9 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java @@ -23,6 +23,7 @@ import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; import ru.rt.restream.reindexer.binding.cproto.util.ConnectionUtils; +import ru.rt.restream.reindexer.expression.PreceptExpression; import java.util.concurrent.CompletableFuture; @@ -51,25 +52,25 @@ public CprotoTransactionContext(long transactionId, Connection connection) { } @Override - public void modifyItem(byte[] data, int format, int mode, String[] precepts, int stateToken) { + public void modifyItem(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { byte[] packedPrecepts = packPrecepts(precepts); ConnectionUtils.rpcCallNoResults(connection, Binding.ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, transactionId); } @Override - public CompletableFuture modifyItemAsync(byte[] data, int format, int mode, String[] precepts, int stateToken) { + public CompletableFuture modifyItemAsync(byte[] data, int format, int mode, PreceptExpression[] precepts, int stateToken) { byte[] packedPrecepts = packPrecepts(precepts); return connection.rpcCallAsync(Binding.ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, transactionId); } - private byte[] packPrecepts(String[] precepts) { + private byte[] packPrecepts(PreceptExpression[] precepts) { byte[] packedPrecepts = new byte[0]; if (precepts.length > 0) { ByteBuffer buffer = new ByteBuffer(); buffer.putVarUInt32(precepts.length); - for (String precept : precepts) { - buffer.putVString(precept); + for (PreceptExpression precept : precepts) { + precept.serializePrecept(buffer); } packedPrecepts = buffer.bytes(); } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/cjson/CjsonItemReader.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/cjson/CjsonItemReader.java index 44652698..53a5e15c 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/cjson/CjsonItemReader.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/cjson/CjsonItemReader.java @@ -35,17 +35,25 @@ import java.util.Collection; import java.util.List; import java.util.UUID; +import java.util.function.Supplier; /** * Reads items from a {@link ByteBuffer}, that contains cjson-encoded data. */ public class CjsonItemReader implements ItemReader { + private final Supplier instanceSupplier; + private final Class itemClass; private final CtagMatcher ctagMatcher; public CjsonItemReader(Class itemClass, CtagMatcher ctagMatcher) { + this(() -> createInstance(itemClass), itemClass, ctagMatcher); + } + + public CjsonItemReader(Supplier instanceSupplier, Class itemClass, CtagMatcher ctagMatcher) { + this.instanceSupplier = instanceSupplier; this.itemClass = itemClass; this.ctagMatcher = ctagMatcher; } @@ -57,11 +65,15 @@ public T readItem(ByteBuffer buffer) { if (!element.isObject()) { throw new IllegalArgumentException("Read object is not an item"); } - return readObject(element.getAsCjsonObject(), itemClass); + return readObject(instanceSupplier, element.getAsCjsonObject(), itemClass); } private V readObject(CjsonObject cjsonObject, Class itemClass) { - V instance = createInstance(itemClass); + return readObject(() -> createInstance(itemClass), cjsonObject, itemClass); + } + + private V readObject(Supplier instanceSupplier, CjsonObject cjsonObject, Class itemClass) { + V instance = instanceSupplier.get(); List fields = BeanPropertyUtils.getInheritedFields(itemClass); for (Field field : fields) { Json json = field.getAnnotation(Json.class); diff --git a/src/main/java/ru/rt/restream/reindexer/expression/PreceptExpression.java b/src/main/java/ru/rt/restream/reindexer/expression/PreceptExpression.java new file mode 100644 index 00000000..c4f2123b --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/expression/PreceptExpression.java @@ -0,0 +1,32 @@ +/* + * Copyright 2020-present Restream + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package ru.rt.restream.reindexer.expression; + +import ru.rt.restream.reindexer.binding.cproto.ByteBuffer; + +/** + * A precept expression. + */ +public interface PreceptExpression extends Expression { + + /** + * Serializes the expression as a precept to the given {@code buffer}. + * + * @param buffer the {@link ByteBuffer} to use + */ + void serializePrecept(ByteBuffer buffer); + +} diff --git a/src/main/java/ru/rt/restream/reindexer/expression/StringExpression.java b/src/main/java/ru/rt/restream/reindexer/expression/StringExpression.java index b3ac8328..5484fee2 100644 --- a/src/main/java/ru/rt/restream/reindexer/expression/StringExpression.java +++ b/src/main/java/ru/rt/restream/reindexer/expression/StringExpression.java @@ -23,7 +23,7 @@ /** * Represents a string expression e.g., {@code now() - 1 * 24 * 60 * 60}. */ -public final class StringExpression implements SetExpression { +public final class StringExpression implements SetExpression, PreceptExpression { private final String expressionString; @@ -44,6 +44,11 @@ public void serializeSet(ByteBuffer buffer) { buffer.putVString(expressionString); } + @Override + public void serializePrecept(ByteBuffer buffer) { + buffer.putVString(expressionString); + } + @Override public String toString() { return expressionString; diff --git a/src/test/java/ru/rt/restream/reindexer/connector/ReindexerTest.java b/src/test/java/ru/rt/restream/reindexer/connector/ReindexerTest.java index 1cf77055..4841a02a 100644 --- a/src/test/java/ru/rt/restream/reindexer/connector/ReindexerTest.java +++ b/src/test/java/ru/rt/restream/reindexer/connector/ReindexerTest.java @@ -21,6 +21,7 @@ import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; +import lombok.SneakyThrows; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; @@ -30,6 +31,7 @@ import ru.rt.restream.reindexer.NamespaceOptions; import ru.rt.restream.reindexer.Query; import ru.rt.restream.reindexer.QueryResultJsonIterator; +import ru.rt.restream.reindexer.ReindexerPrecept; import ru.rt.restream.reindexer.ResultIterator; import ru.rt.restream.reindexer.TimeUnit; import ru.rt.restream.reindexer.Transaction; @@ -73,11 +75,15 @@ import static org.hamcrest.Matchers.containsInAnyOrder; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.empty; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.hamcrest.Matchers.hasSize; import static org.hamcrest.Matchers.in; import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; +import static org.hamcrest.Matchers.sameInstance; import static org.hamcrest.Matchers.startsWith; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -138,7 +144,7 @@ public void testInsertItem() { @Test public void testInsertJsonItem() { String namespaceName = "items"; - db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); TestItem testItem = new TestItem(); testItem.setId(123); @@ -147,9 +153,10 @@ public void testInsertJsonItem() { String jsonItem = JsonSerializer.toJson(testItem); - db.insert(namespaceName, jsonItem); + Optional inserted = ns.insert(jsonItem); + assertThat(inserted.isPresent(), is(false)); - Iterator iterator = db.query(namespaceName, TestItem.class) + Iterator iterator = ns.query() .where("id", EQ, 123) .execute(); @@ -191,21 +198,22 @@ public void testUpdateItem() { @Test public void testUpdateJsonItem() { String namespaceName = "items"; - db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); TestItem testItem = new TestItem(); testItem.setId(123); testItem.setName("TestName"); testItem.setNonIndex("testNonIndex"); - db.insert(namespaceName, testItem); + ns.insert(testItem); testItem.setName("TestNameUpdated"); String jsonItem = JsonSerializer.toJson(testItem); - db.update(namespaceName, jsonItem); + Optional updated = ns.update(jsonItem); + assertThat(updated.isPresent(), is(false)); - Iterator iterator = db.query(namespaceName, TestItem.class) + Iterator iterator = ns.query() .where("id", EQ, 123) .execute(); assertThat(iterator.hasNext(), is(true)); @@ -240,7 +248,7 @@ public void testUpsertItem() { @Test public void testUpsertJsonItem() { String namespaceName = "items"; - db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); TestItem testItem = new TestItem(); testItem.setId(123); @@ -249,9 +257,10 @@ public void testUpsertJsonItem() { String jsonItem = JsonSerializer.toJson(testItem); - db.upsert(namespaceName, jsonItem); + Optional upserted = ns.upsert(jsonItem); + assertThat(upserted.isPresent(), is(false)); - Iterator iterator = db.query(namespaceName, TestItem.class) + Iterator iterator = ns.query() .where("id", EQ, 123) .execute(); assertThat(iterator.hasNext(), is(true)); @@ -1816,6 +1825,147 @@ public void testUpsertWithSerialId() { assertThat(ids, contains(1, 2, 3)); } + @Test + public void testInsertWithSerialIdReturnsGeneratedId() { + String namespaceName = "items"; + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), SerialIdTestItem.class); + + int currentVersion = 0; + for (int i = 0; i < 3; i++) { + SerialIdTestItem inserted = ns.insert(new SerialIdTestItem()); + assertThat(inserted, notNullValue()); + assertThat(inserted.id, is(i + 1)); + assertThat(inserted.version, is(++currentVersion)); + SerialIdTestItem updated = ns.update(inserted); + assertThat(updated, sameInstance(inserted)); + assertThat(inserted.id, is(i + 1)); + assertThat(inserted.version, is(++currentVersion)); + } + } + + @Test + public void testInsertJsonWithSerialIdReturnsGeneratedId() { + String namespaceName = "items"; + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), SerialIdTestItem.class); + + int currentVersion = 0; + for (int i = 0; i < 3; i++) { + SerialIdTestItem inserted = ns.insert(JsonSerializer.toJson(new SerialIdTestItem())) + .orElse(null); + assertThat(inserted, notNullValue()); + assertThat(inserted.id, is(i + 1)); + assertThat(inserted.version, is(++currentVersion)); + SerialIdTestItem updated = ns.update(JsonSerializer.toJson(inserted)) + .orElse(null); + assertThat(updated, notNullValue()); + assertThat(updated.id, is(i + 1)); + assertThat(updated.version, is(++currentVersion)); + } + } + + @Test + public void testUpsertWithSerialIdReturnsGeneratedId() { + String namespaceName = "items"; + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), SerialIdTestItem.class); + + int currentVersion = 0; + for (int i = 0; i < 3; i++) { + SerialIdTestItem upserted1 = ns.upsert(new SerialIdTestItem()); + assertThat(upserted1, notNullValue()); + assertThat(upserted1.id, is(i + 1)); + assertThat(upserted1.version, is(++currentVersion)); + SerialIdTestItem upserted2 = ns.upsert(upserted1); + assertThat(upserted2, sameInstance(upserted1)); + assertThat(upserted2.id, is(i + 1)); + assertThat(upserted2.version, is(++currentVersion)); + } + } + + @Test + public void testUpsertJsonWithSerialIdReturnsGeneratedId() { + String namespaceName = "items"; + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), SerialIdTestItem.class); + + int currentVersion = 0; + for (int i = 0; i < 3; i++) { + SerialIdTestItem upserted1 = ns.upsert(JsonSerializer.toJson(new SerialIdTestItem())) + .orElse(null); + assertThat(upserted1, notNullValue()); + assertThat(upserted1.id, is(i + 1)); + assertThat(upserted1.version, is(++currentVersion)); + SerialIdTestItem upserted2 = ns.upsert(JsonSerializer.toJson(upserted1)) + .orElse(null); + assertThat(upserted2, notNullValue()); + assertThat(upserted2.id, is(i + 1)); + assertThat(upserted2.version, is(++currentVersion)); + } + } + + @Test + public void testOverridingPreceptApplyStrategyToWhenNullGeneratesVersionOnlyOnce() { + String namespaceName = "items"; + Namespace ns = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), SerialIdTestItem.class); + ReindexerPrecept precept = ReindexerPrecept.builder() + .fieldName("version") + .jsonPath("version") + .expression(Expression.string("version=serial()")) + .applyStrategy(ReindexerPrecept.ApplyStrategy.WHEN_NULL) + .build(); + db.registerPrecept(namespaceName, precept); + + for (int i = 0; i < 3; i++) { + SerialIdTestItem inserted = ns.insert(new SerialIdTestItem()); + assertThat(inserted, notNullValue()); + assertThat(inserted.id, is(i + 1)); + assertThat(inserted.version, is(i + 1)); + SerialIdTestItem updated = ns.update(inserted); + assertThat(updated, sameInstance(inserted)); + assertThat(inserted.id, is(i + 1)); + assertThat(inserted.version, is(i + 1)); + } + } + + @SneakyThrows + @Test + public void testWithNowPreceptReturnsGeneratedTimestamp() { + FieldConverterRegistryFactory.INSTANCE.registerFieldConverter(TestItemAuditingFields.class, "createdAt", new LocalDateTimeUnitFieldConverter(TimeUnit.MILLIS)); + FieldConverterRegistryFactory.INSTANCE.registerFieldConverter(TestItemAuditingFields.class, "updatedAt", new LocalDateTimeUnitFieldConverter(TimeUnit.MILLIS)); + String namespaceName = "items"; + db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItemAuditingFields.class); + ReindexerPrecept createdAtPrecept = ReindexerPrecept.builder() + .fieldName("createdAt") + .jsonPath("createdAt") + .expression(Expression.string("createdAt=now(msec)")) + .applyStrategy(ReindexerPrecept.ApplyStrategy.WHEN_NULL) + .build(); + db.registerPrecept(namespaceName, createdAtPrecept); + ReindexerPrecept updatedAtPrecept = ReindexerPrecept.builder() + .fieldName("updatedAt") + .jsonPath("updatedAt") + .expression(Expression.string("updatedAt=now(msec)")) + .applyStrategy(ReindexerPrecept.ApplyStrategy.ALWAYS) + .build(); + db.registerPrecept(namespaceName, updatedAtPrecept); + + TestItemAuditingFields item = new TestItemAuditingFields(1, null, null); + TestItemAuditingFields inserted = db.insert(namespaceName, item); + assertThat(inserted, sameInstance(item)); + assertThat(inserted.id, is(1)); + LocalDateTime createdAt = inserted.createdAt; + LocalDateTime updatedAt = inserted.updatedAt; + LocalDateTime now = LocalDateTime.now(); + assertThat(createdAt, lessThanOrEqualTo(now)); + assertThat(updatedAt, lessThanOrEqualTo(createdAt)); + + // Small delay to shift the updatedAt precept. + Thread.sleep(1000); + TestItemAuditingFields updated = db.update(namespaceName, inserted); + assertThat(updated, sameInstance(inserted)); + assertThat(updated.id, is(1)); + assertThat(updated.createdAt, is(createdAt)); + assertThat(updated.updatedAt, greaterThan(updatedAt)); + } + @Test public void testTransactionUpsertWithSerialId() { String namespaceName = "items"; @@ -3116,7 +3266,10 @@ private static Stream flatArrayLengthExpressionArguments() { public static class SerialIdTestItem { @Serial @Reindex(name = "id", isPrimaryKey = true) - private int id; + private Integer id; + @Serial + @Reindex(name = "version") + private Integer version; } @Getter @@ -3371,6 +3524,21 @@ public static class TestItemCreatedAt { private LocalDateTime createdAt; } + @Getter + @Setter + @NoArgsConstructor + @AllArgsConstructor + public static class TestItemAuditingFields { + @Reindex(name = "id", isPrimaryKey = true) + private Integer id; + + @Reindex(name = "createdAt") + private LocalDateTime createdAt; + + @Reindex(name = "updatedAt") + private LocalDateTime updatedAt; + } + @Getter @Setter @NoArgsConstructor