Skip to content

Commit 238d968

Browse files
author
Alexander Lavrukov
committed
table-cache: common table methods
1 parent 6fbd377 commit 238d968

File tree

5 files changed

+88
-53
lines changed

5 files changed

+88
-53
lines changed

Diff for: ‎repository-inmemory/src/main/java/tech/ydb/yoj/repository/test/inmemory/InMemoryTable.java

+6
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import tech.ydb.yoj.databind.expression.OrderExpression;
88
import tech.ydb.yoj.databind.schema.ObjectSchema;
99
import tech.ydb.yoj.databind.schema.Schema;
10+
import tech.ydb.yoj.repository.db.CommonTable;
1011
import tech.ydb.yoj.repository.db.Entity;
1112
import tech.ydb.yoj.repository.db.EntityExpressions;
1213
import tech.ydb.yoj.repository.db.EntityIdSchema;
@@ -184,6 +185,11 @@ public <V extends View> V find(Class<V> viewType, Entity.Id<T> id) {
184185
return transaction.doInTransaction("find(" + id + ")", type, shard -> shard.find(id, viewType));
185186
}
186187

188+
@Override
189+
public <ID extends Entity.Id<T>> List<T> find(Set<ID> ids) {
190+
return CommonTable.find(transaction.getTransactionLocal(), this, ids);
191+
}
192+
187193
@Override
188194
@SuppressWarnings("unchecked")
189195
public <ID extends Entity.Id<T>> List<T> find(Range<ID> range) {

Diff for: ‎repository-ydb-v1/src/main/java/tech/ydb/yoj/repository/ydb/table/YdbTable.java

+6-1
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import lombok.NonNull;
88
import tech.ydb.yoj.databind.expression.FilterExpression;
99
import tech.ydb.yoj.databind.expression.OrderExpression;
10+
import tech.ydb.yoj.repository.db.CommonTable;
1011
import tech.ydb.yoj.repository.db.Entity;
1112
import tech.ydb.yoj.repository.db.Entity.Id;
1213
import tech.ydb.yoj.repository.db.EntityIdSchema;
@@ -16,7 +17,6 @@
1617
import tech.ydb.yoj.repository.db.Tx;
1718
import tech.ydb.yoj.repository.db.ViewSchema;
1819
import tech.ydb.yoj.repository.db.bulk.BulkParams;
19-
import tech.ydb.yoj.repository.db.cache.FirstLevelCache;
2020
import tech.ydb.yoj.repository.db.cache.TransactionLocal;
2121
import tech.ydb.yoj.repository.db.readtable.ReadTableParams;
2222
import tech.ydb.yoj.repository.db.statement.Changeset;
@@ -235,6 +235,11 @@ public <V extends View> V find(Class<V> viewType, Entity.Id<T> id) {
235235
return res.isEmpty() ? null : res.get(0);
236236
}
237237

238+
@Override
239+
public <ID extends Id<T>> List<T> find(Set<ID> ids) {
240+
return CommonTable.find(executor.getTransactionLocal(), this, ids);
241+
}
242+
238243
@Override
239244
public <ID extends Entity.Id<T>> List<T> find(Range<ID> range) {
240245
return postLoad(executor.execute(YqlStatement.findRange(type, range), range));

Diff for: ‎repository-ydb-v2/src/main/java/tech/ydb/yoj/repository/ydb/table/YdbTable.java

+6
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import lombok.NonNull;
88
import tech.ydb.yoj.databind.expression.FilterExpression;
99
import tech.ydb.yoj.databind.expression.OrderExpression;
10+
import tech.ydb.yoj.repository.db.CommonTable;
1011
import tech.ydb.yoj.repository.db.Entity;
1112
import tech.ydb.yoj.repository.db.Entity.Id;
1213
import tech.ydb.yoj.repository.db.EntityIdSchema;
@@ -234,6 +235,11 @@ public <V extends View> V find(Class<V> viewType, Entity.Id<T> id) {
234235
return res.isEmpty() ? null : res.get(0);
235236
}
236237

238+
@Override
239+
public <ID extends Id<T>> List<T> find(Set<ID> ids) {
240+
return CommonTable.find(executor.getTransactionLocal(), this, ids);
241+
}
242+
237243
@Override
238244
public <ID extends Entity.Id<T>> List<T> find(Range<ID> range) {
239245
return postLoad(executor.execute(YqlStatement.findRange(type, range), range));

Diff for: ‎repository/src/main/java/tech/ydb/yoj/repository/db/CommonTable.java

+68
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
package tech.ydb.yoj.repository.db;
2+
3+
import com.google.common.collect.Sets;
4+
import lombok.AccessLevel;
5+
import lombok.NoArgsConstructor;
6+
import tech.ydb.yoj.repository.db.cache.TransactionLocal;
7+
8+
import java.util.HashMap;
9+
import java.util.HashSet;
10+
import java.util.List;
11+
import java.util.Optional;
12+
import java.util.Set;
13+
import java.util.function.Function;
14+
import java.util.stream.Collectors;
15+
16+
import static java.util.stream.Collectors.toSet;
17+
18+
@NoArgsConstructor(access = AccessLevel.PRIVATE)
19+
public final class CommonTable {
20+
public static <E extends Entity<E>, ID extends Entity.Id<E>> List<E> find(
21+
TransactionLocal transactionLocal, Table<E> table, Set<ID> ids) {
22+
if (ids.isEmpty()) {
23+
return List.of();
24+
}
25+
26+
var orderBy = EntityExpressions.defaultOrder(table.getType());
27+
var cache = transactionLocal.firstLevelCache();
28+
var isPartialIdMode = ids.iterator().next().isPartial();
29+
30+
var foundInCache = ids.stream()
31+
.filter(cache::containsKey)
32+
.map(cache::peek)
33+
.flatMap(Optional::stream)
34+
.collect(Collectors.toMap(Entity::getId, Function.identity()));
35+
var remainingIds = Sets.difference(ids, foundInCache.keySet());
36+
var foundInDb = table.findUncached(remainingIds, null, orderBy, null);
37+
38+
var merged = new HashMap<Entity.Id<E>, E>();
39+
40+
// some entries found in db with partial id query may already be in cache (after update/delete),
41+
// so we must return actual entries from cache
42+
for (var entry : foundInDb) {
43+
var id = entry.getId();
44+
if (cache.containsKey(id)) {
45+
var cached = cache.peek(id);
46+
cached.ifPresent(t -> merged.put(id, t));
47+
// not present means marked as deleted in cache
48+
} else {
49+
merged.put(id, table.postLoad(entry));
50+
}
51+
}
52+
53+
// add entries found in cache and not fetched from db
54+
for (var pair : foundInCache.entrySet()) {
55+
var id = pair.getKey();
56+
var entry = pair.getValue();
57+
merged.put(id, entry);
58+
}
59+
60+
if (!isPartialIdMode) {
61+
Set<Entity.Id<E>> foundInDbIds = foundInDb.stream().map(Entity::getId).collect(toSet());
62+
Set<Entity.Id<E>> foundInCacheIds = new HashSet<>(foundInCache.keySet());
63+
Sets.difference(Sets.difference(ids, foundInDbIds), foundInCacheIds).forEach(cache::putEmpty);
64+
}
65+
66+
return merged.values().stream().sorted(EntityIdSchema.SORT_ENTITY_BY_ID).collect(Collectors.toList());
67+
}
68+
}

Diff for: ‎repository/src/main/java/tech/ydb/yoj/repository/db/Table.java

+2-52
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
package tech.ydb.yoj.repository.db;
22

3-
import com.google.common.collect.Sets;
43
import lombok.NonNull;
54
import tech.ydb.yoj.databind.expression.FilterExpression;
65
import tech.ydb.yoj.databind.expression.OrderExpression;
@@ -14,8 +13,6 @@
1413
import javax.annotation.CheckForNull;
1514
import javax.annotation.Nullable;
1615
import java.util.Collection;
17-
import java.util.HashMap;
18-
import java.util.HashSet;
1916
import java.util.List;
2017
import java.util.Optional;
2118
import java.util.Set;
@@ -25,7 +22,6 @@
2522
import java.util.stream.Stream;
2623

2724
import static java.util.stream.Collectors.toList;
28-
import static java.util.stream.Collectors.toSet;
2925
import static java.util.stream.Stream.concat;
3026

3127
public interface Table<T extends Entity<T>> {
@@ -42,6 +38,8 @@ public interface Table<T extends Entity<T>> {
4238

4339
<V extends View> V find(Class<V> viewType, Entity.Id<T> id);
4440

41+
<ID extends Entity.Id<T>> List<T> find(Set<ID> ids);
42+
4543
<ID extends Entity.Id<T>> List<T> find(Range<ID> range);
4644

4745
<ID extends Entity.Id<T>> List<ID> findIds(Range<ID> range);
@@ -265,54 +263,6 @@ default <V extends Table.View> ViewListResult<T, V> list(Class<V> viewType, List
265263
return ViewListResult.forPage(request, viewType, nextPage);
266264
}
267265

268-
default <ID extends Entity.Id<T>> List<T> find(Set<ID> ids) {
269-
if (ids.isEmpty()) {
270-
return List.of();
271-
}
272-
273-
var orderBy = EntityExpressions.defaultOrder(getType());
274-
var cache = Tx.Current.get().getRepositoryTransaction().getTransactionLocal().firstLevelCache();
275-
var isPartialIdMode = ids.iterator().next().isPartial();
276-
277-
var foundInCache = ids.stream()
278-
.filter(cache::containsKey)
279-
.map(cache::peek)
280-
.flatMap(Optional::stream)
281-
.collect(Collectors.toMap(Entity::getId, Function.identity()));
282-
var remainingIds = Sets.difference(ids, foundInCache.keySet());
283-
var foundInDb = findUncached(remainingIds, null, orderBy, null);
284-
285-
var merged = new HashMap<Entity.Id<T>, T>();
286-
287-
// some entries found in db with partial id query may already be in cache (after update/delete),
288-
// so we must return actual entries from cache
289-
for (var entry : foundInDb) {
290-
var id = entry.getId();
291-
if (cache.containsKey(id)) {
292-
var cached = cache.peek(id);
293-
cached.ifPresent(t -> merged.put(id, t));
294-
// not present means marked as deleted in cache
295-
} else {
296-
merged.put(id, this.postLoad(entry));
297-
}
298-
}
299-
300-
// add entries found in cache and not fetched from db
301-
for (var pair : foundInCache.entrySet()) {
302-
var id = pair.getKey();
303-
var entry = pair.getValue();
304-
merged.put(id, entry);
305-
}
306-
307-
if (!isPartialIdMode) {
308-
Set<Entity.Id<T>> foundInDbIds = foundInDb.stream().map(Entity::getId).collect(toSet());
309-
Set<Entity.Id<T>> foundInCacheIds = new HashSet<>(foundInCache.keySet());
310-
Sets.difference(Sets.difference(ids, foundInDbIds), foundInCacheIds).forEach(cache::putEmpty);
311-
}
312-
313-
return merged.values().stream().sorted(EntityIdSchema.SORT_ENTITY_BY_ID).collect(Collectors.toList());
314-
}
315-
316266
default void bulkUpsert(List<T> input, BulkParams params) {
317267
throw new UnsupportedOperationException();
318268
}

0 commit comments

Comments
 (0)