Skip to content

Unpack item with generated id #153

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
evgeniycheban wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Unpack item with generated id #153

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 34 additions & 13 deletions src/main/java/ru/rt/restream/reindexer/Namespace.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
*/
package ru.rt.restream.reindexer;

import java.util.Optional;

/**
* Provides methods for manipulating Reindexer namespace data.
*
Expand All @@ -32,44 +34,56 @@ public interface Namespace<T> {
/**
* 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<T> 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<T> 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<T> update(String item);

/**
* Deletes the given item data.
Expand Down Expand Up @@ -125,4 +139,11 @@ public interface Namespace<T> {
*/
void updateSql(String query);

/**
* Adds a {@link ReindexerPrecept} to the namespace to be applied to all items persisted in the namespace.
*
* @param precept the {@link ReindexerPrecept} to add, cannot be {@literal null}
*/
void addPrecept(ReindexerPrecept precept);

}
97 changes: 17 additions & 80 deletions src/main/java/ru/rt/restream/reindexer/QueryResultIterator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,10 +129,10 @@ public T next() {
}

private <S> S readItem(ItemReader<S> 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;
}

Expand All @@ -149,10 +149,10 @@ private Query<?> getItemQueryContext(Query<?> defaultQueryContext, int nsId) {
return defaultQueryContext;
}

private <S> S readItemData(ItemParams params, ItemReader<S> reader) {
if (params.cptr != 0) {
ByteBuffer nativeBuffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), params.cptr,
params.nsId);
private <S> S readItemData(QueryResult.ItemParams params, ItemReader<S> reader) {
if (params.getCptr() != 0) {
ByteBuffer nativeBuffer = NativeUtils.getNativeBuffer(queryResult.getResultsPtr(), params.getCptr(),
params.getNsId());
return reader.readItem(nativeBuffer);
}

Expand Down Expand Up @@ -210,7 +210,7 @@ private void readJoinedItemsV1(Object item, int nsId) {
CjsonItemReader<?> joinedItemReader = newItemReader(joinedNamespace.getItemClass(), joinedNamespace);
List<Object> subItems = new ArrayList<>(itemsCount);
for (int j = 0; j < itemsCount; j++) {
ItemParams subItemParams = readItemParams();
QueryResult.ItemParams subItemParams = readItemParams();
subItems.add(readItemData(subItemParams, joinedItemReader));
}

Expand All @@ -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);
}
Expand Down Expand Up @@ -312,42 +312,12 @@ private void writeJoinResult(Object item, String fieldName, List<Object> 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() {
Expand Down Expand Up @@ -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;
}
}

}
Loading
Loading