Skip to content

Commit 06ba898

Browse files
ohadzeligerohadzeliger
authored
Add store timer events for Lucene serialization (#4044)
This PR adds store timer events for Lucene serialization operations. Resolves #4043
1 parent 8bddbf0 commit 06ba898

3 files changed

Lines changed: 50 additions & 14 deletions

File tree

‎fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneEvents.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,9 @@ public enum Waits implements StoreTimer.Wait {
185185
/** Replay items from queue in a read-only transaction. */
186186
WAIT_LUCENE_REPLAY_QUEUE("lucene replay pending writes queue"),
187187
/** Get the number of entries in the pending writes queue. */
188-
WAIT_LUCENE_GET_QUEUE_SIZE("lucene get pending writes queue size");
188+
WAIT_LUCENE_GET_QUEUE_SIZE("lucene get pending writes queue size"),
189+
WAIT_LUCENE_SERIALIZE("lucene serialize data"),
190+
WAIT_LUCENE_DESERIALIZE("lucene deserialize data");
189191
private final String title;
190192
private final String logKey;
191193

‎fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java‎

Lines changed: 41 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -363,7 +363,7 @@ void writeFieldInfos(long id, byte[] rawBytes) {
363363
throw new RecordCoreArgumentException("FieldInfo id should never be 0");
364364
}
365365
byte[] key = fieldInfosSubspace.pack(id);
366-
byte[] value = serializer.encodeFieldProtobuf(rawBytes);
366+
byte[] value = encodeFieldProtobuf(rawBytes);
367367
agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, key.length + value.length);
368368
if (LOGGER.isTraceEnabled()) {
369369
LOGGER.trace(getLogMessage("Write lucene stored field infos data",
@@ -379,8 +379,8 @@ Stream<NonnullPair<Long, byte[]>> getAllFieldInfosStream() {
379379
agilityContext.apply(aContext -> aContext.ensureActive().getRange(fieldInfosSubspace.range()).asList()))
380380
.stream()
381381
.map(keyValue -> NonnullPair.of(
382-
fieldInfosSubspace.unpack(keyValue.getKey()).getLong(0),
383-
serializer.decodeFieldProtobuf(keyValue.getValue())));
382+
fieldInfosSubspace.unpack(keyValue.getKey()).getLong(0),
383+
decodeFieldProtobuf(keyValue.getValue())));
384384
}
385385

386386
public CompletableFuture<Integer> getFieldInfosCount() {
@@ -425,7 +425,7 @@ public static boolean isStoredFieldsFile(String name) {
425425
*/
426426
public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLuceneFileReference reference) {
427427
final byte[] fileReferenceBytes = reference.getBytes();
428-
final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(fileReferenceBytes));
428+
final byte[] encodedBytes = Objects.requireNonNull(encode(fileReferenceBytes));
429429
agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE_FILE_REFERENCE, encodedBytes.length);
430430
if (LOGGER.isTraceEnabled()) {
431431
LOGGER.trace(getLogMessage("Write lucene file reference",
@@ -447,7 +447,7 @@ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLucene
447447
* @return the actual data size written to database with potential compression and encryption applied
448448
*/
449449
public int writeData(final long id, final int block, @Nonnull final byte[] value) {
450-
final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(value));
450+
final byte[] encodedBytes = Objects.requireNonNull(encode(value));
451451
agilityContext.increment(LuceneEvents.Counts.LUCENE_BLOCK_WRITES);
452452
//This may not be correct transactionally
453453
agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, encodedBytes.length);
@@ -471,7 +471,7 @@ public int writeData(final long id, final int block, @Nonnull final byte[] value
471471
*/
472472
public void writeStoredFields(@Nonnull String segmentName, int docID, @Nonnull final byte[] rawBytes) {
473473
byte[] key = storedFieldsSubspace.pack(Tuple.from(segmentName, docID));
474-
byte[] value = serializer.encodeFieldProtobuf(rawBytes);
474+
byte[] value = encodeFieldProtobuf(rawBytes);
475475
agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE_STORED_FIELDS, key.length + value.length);
476476
if (LOGGER.isTraceEnabled()) {
477477
LOGGER.trace(getLogMessage("Write lucene stored fields data",
@@ -562,7 +562,7 @@ private CompletableFuture<byte[]> readBlock(@Nonnull IndexInput requestingInput,
562562
private CompletableFuture<byte[]> readData(long id, int block) {
563563
return agilityContext.instrument(LuceneEvents.Events.LUCENE_FDB_READ_BLOCK,
564564
agilityContext.get(dataSubspace.pack(Tuple.from(id, block)))
565-
.thenApply(serializer::decode));
565+
.thenApply(this::decode));
566566
}
567567

568568
@Nonnull
@@ -577,7 +577,7 @@ public byte[] readStoredFields(String segmentName, int docId) {
577577
.addLogInfo(LuceneLogMessageKeys.DOC_ID, docId)
578578
.addLogInfo(LogMessageKeys.KEY, ByteArrayUtil2.loggable(key));
579579
}
580-
return Objects.requireNonNull(serializer.decodeFieldProtobuf(rawBytes));
580+
return Objects.requireNonNull(decodeFieldProtobuf(rawBytes));
581581
}
582582

583583
@Nonnull
@@ -591,7 +591,7 @@ public List<byte[]> readAllStoredFields(String segmentName) {
591591
.addLogInfo(LogMessageKeys.RANGE_START, ByteArrayUtil2.loggable(range.begin))
592592
.addLogInfo(LogMessageKeys.RANGE_END, ByteArrayUtil2.loggable(range.end));
593593
}
594-
return list.stream().map(KeyValue::getValue).map(serializer::decodeFieldProtobuf).collect(Collectors.toList());
594+
return list.stream().map(KeyValue::getValue).map(this::decodeFieldProtobuf).collect(Collectors.toList());
595595
}
596596

597597
/**
@@ -648,7 +648,7 @@ private CompletableFuture<Void> loadFileReferenceCacheForMemoization() {
648648
agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_FILES_COUNT, list.size());
649649
list.forEach(kv -> {
650650
String name = metaSubspace.unpack(kv.getKey()).getString(0);
651-
final FDBLuceneFileReference fileReference = Objects.requireNonNull(FDBLuceneFileReference.parseFromBytes(serializer.decode(kv.getValue())));
651+
final FDBLuceneFileReference fileReference = Objects.requireNonNull(FDBLuceneFileReference.parseFromBytes(decode(kv.getValue())));
652652
outMap.put(name, fileReference);
653653
if (fileReference.getFieldInfosId() != 0) {
654654
fieldInfosCount.computeIfAbsent(fileReference.getFieldInfosId(), key -> new AtomicInteger(0))
@@ -939,7 +939,7 @@ public void rename(@Nonnull final String source, @Nonnull final String dest) thr
939939
.addLogInfo(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, serializer.isCompressionEnabled())
940940
.addLogInfo(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, serializer.isEncryptionEnabled());
941941
}
942-
byte[] encodedBytes = serializer.encode(value.getBytes());
942+
byte[] encodedBytes = encode(value.getBytes());
943943
agilityContext.set(metaSubspace.pack(dest), encodedBytes);
944944
agilityContext.clear(key);
945945

@@ -1259,4 +1259,34 @@ public String getIndexOption(@Nonnull String key) {
12591259
private FDBDirectoryLockFactory defaultLockFactory(final AgilityContext agilityContext) {
12601260
return new FDBDirectoryLockFactory(this, Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_FILE_LOCK_TIME_WINDOW_MILLISECONDS), 0));
12611261
}
1262+
1263+
private byte[] encodeFieldProtobuf(final byte[] bytes) {
1264+
long startTime = System.nanoTime();
1265+
byte[] encoded = serializer.encodeFieldProtobuf(bytes);
1266+
agilityContext.recordEvent(LuceneEvents.Waits.WAIT_LUCENE_SERIALIZE, System.nanoTime() - startTime);
1267+
return encoded;
1268+
}
1269+
1270+
private byte[] decodeFieldProtobuf(final byte[] bytes) {
1271+
long startTime = System.nanoTime();
1272+
final byte[] decoded = serializer.decodeFieldProtobuf(bytes);
1273+
agilityContext.recordEvent(LuceneEvents.Waits.WAIT_LUCENE_DESERIALIZE, System.nanoTime() - startTime);
1274+
return decoded;
1275+
}
1276+
1277+
private byte[] encode(final byte[] bytes) {
1278+
long startTime = System.nanoTime();
1279+
final byte[] encoded = serializer.encode(bytes);
1280+
agilityContext.recordEvent(LuceneEvents.Waits.WAIT_LUCENE_SERIALIZE, System.nanoTime() - startTime);
1281+
return encoded;
1282+
}
1283+
1284+
private byte[] decode(final byte[] bytes) {
1285+
long startTime = System.nanoTime();
1286+
final byte[] decoded = serializer.decode(bytes);
1287+
agilityContext.recordEvent(LuceneEvents.Waits.WAIT_LUCENE_DESERIALIZE, System.nanoTime() - startTime);
1288+
return decoded;
1289+
}
1290+
1291+
12621292
}

‎fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/PendingWriteQueue.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -204,15 +204,17 @@ public RecordCursor<QueueEntry> getQueueCursor(
204204
.limitRowsTo(scanProperties.getExecuteProperties().getReturnedRowLimit());
205205

206206
return unsplitter.map(rawRecord ->
207-
toQueueEntry(rawRecord.getPrimaryKey(), rawRecord.getRawRecord()));
207+
toQueueEntry(context, rawRecord.getPrimaryKey(), rawRecord.getRawRecord()));
208208
}
209209

210210
/**
211211
* Convert a raw record back to a queue entry.
212212
*/
213-
private QueueEntry toQueueEntry(Tuple keyTuple, byte[] valueBytes) {
213+
private QueueEntry toQueueEntry(FDBRecordContext context, Tuple keyTuple, byte[] valueBytes) {
214214
try {
215+
long startTime = System.nanoTime();
215216
final byte[] value = serializer.decode(valueBytes);
217+
context.record(LuceneEvents.Waits.WAIT_LUCENE_DESERIALIZE, System.nanoTime() - startTime);
216218
LucenePendingWriteQueueProto.PendingWriteItem item = LucenePendingWriteQueueProto.PendingWriteItem.parseFrom(value);
217219
return new QueueEntry(keyTuple, item, allowIncarnation);
218220
} catch (InvalidProtocolBufferException e) {
@@ -361,7 +363,9 @@ private void enqueueOperationInternal(
361363
Tuple keyTuple = allowIncarnation
362364
? Tuple.from(incarnation, recordVersion.toVersionstamp())
363365
: Tuple.from(recordVersion.toVersionstamp());
366+
long startTime = System.nanoTime();
364367
byte[] value = serializer.encode(builder.build().toByteArray());
368+
context.record(LuceneEvents.Waits.WAIT_LUCENE_SERIALIZE, System.nanoTime() - startTime);
365369
// save with splits
366370
SplitHelper.saveWithSplit(context, queueSubspace, keyTuple, value, null, true, false, false, null, null);
367371

0 commit comments

Comments
 (0)