This is an automated email from the ASF dual-hosted git repository.
KKcorps pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new ee2ee576f8b Persist upsert snapshot diagnostics for replica divergence
triage (#19637)
ee2ee576f8b is described below
commit ee2ee576f8b1af4aba2bd56cf0b4fa35c3a07538
Author: Kartik Khare <[email protected]>
AuthorDate: Fri Oct 9 13:20:29 2026 +0530
Persist upsert snapshot diagnostics for replica divergence triage (#19637)
* Add opt-in upsert bitmap diagnostics to server metadata API
* Persist upsert diagnostics with bitmap snapshots
* Rename upsert snapshot diagnostics to say what they hold
- snapshotTriggerStartOffset -> snapshotConsumedUpToOffset. The value is
the new
consuming segment's start offset, so it marks where the snapshot's data
ends.
- validDocIdsCrc32 -> docIdsCrc32. The same field is written for the
queryable
doc ids file, so "valid" was wrong for SNAPSHOT_WITH_DELETE.
- snapshotTriggerSegmentName -> snapshotConsumingSegmentName, to pair with
the
offset and not read like the bitmap's own segment.
* Name the snapshot CRC docIdsCrc and record which bitmap it covers
The same trailer is written for the valid and queryable doc ids files, so
each
file now carries docIdsType (VALID_DOC_IDS or QUERYABLE_DOC_IDS) next to its
CRC, and the API returns both.
* Ignore unknown snapshot diagnostics fields on read
The default mapper fails on unknown fields, so a field added by a newer
server
would make an older server drop all diagnostics for that file.
* Record where each replica's consumer stopped in the snapshot offset
When the previous consuming segment releases the partition before it is
sealed (parallel build or download), its rows are still in the upsert
metadata when the next consumer snapshots. Hand its stop offset over
through the ConsumerCoordinator so snapshotConsumedUpToOffset shows what
this replica actually covered, not the shared start offset.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* Pass the snapshot trigger explicitly and ignore stale unsealed offsets
The unsealed release is now used only when it comes from the segment right
before the new consumer. A record left by an older segment that was never
offloaded made the replica look far behind, although it consumed every
segment
since. A predecessor that fell behind still reports where it stopped.
The trigger now flows through doTakeSnapshot(Trigger) instead of a
ThreadLocal,
the capture time is taken inside getBytesAndCardinality() under the
bitmap's own
lock, and the trailer is written after the bitmap instead of copying the
bitmap
into a new array.
* Keep the two-arg CardinalityAndBytes constructor for binary compatibility
---------
Co-authored-by: Kartik Khare <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../data/manager/realtime/ConsumerCoordinator.java | 27 ++++
.../realtime/RealtimeSegmentDataManager.java | 21 +++-
.../manager/realtime/ConsumerCoordinatorTest.java | 25 ++++
.../realtime/RealtimeSegmentDataManagerTest.java | 84 +++++++++++++
.../immutable/ImmutableSegmentImpl.java | 22 ++--
.../upsert/BasePartitionUpsertMetadataManager.java | 29 ++++-
.../pinot/segment/local/upsert/DocIdsSnapshot.java | 130 ++++++++++++++++++++
.../upsert/PartitionUpsertMetadataManager.java | 5 +
.../BasePartitionUpsertMetadataManagerTest.java | 58 ++++++++-
...rrentMapPartitionUpsertMetadataManagerTest.java | 3 +-
.../segment/local/upsert/DocIdsSnapshotTest.java | 136 +++++++++++++++++++++
.../mutable/ThreadSafeMutableRoaringBitmap.java | 16 ++-
.../pinot/server/api/resources/TablesResource.java | 30 +++--
.../pinot/server/api/TablesResourceTest.java | 54 ++++++++
14 files changed, 607 insertions(+), 33 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java
index d81bce64fa5..68135ef0f56 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java
@@ -24,9 +24,11 @@ import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import javax.annotation.Nullable;
import org.apache.helix.model.IdealState;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerGauge;
@@ -35,6 +37,7 @@ import org.apache.pinot.common.metrics.ServerTimer;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,6 +63,13 @@ public class ConsumerCoordinator {
private volatile int _maxSequenceNumberRegistered = -1;
+ // Set when a consuming segment releases the semaphore before it is sealed,
so its rows are still in the partition
+ // metadata when the next consumer starts. Cleared when that segment is
offloaded.
+ private final AtomicReference<UnsealedRelease> _unsealedRelease = new
AtomicReference<>();
+
+ public record UnsealedRelease(String segmentName, StreamPartitionMsgOffset
stoppedAtOffset) {
+ }
+
public ConsumerCoordinator(boolean enforceConsumptionInOrder,
RealtimeTableDataManager realtimeTableDataManager) {
_enforceConsumptionInOrder = enforceConsumptionInOrder;
_realtimeTableDataManager = realtimeTableDataManager;
@@ -110,6 +120,23 @@ public class ConsumerCoordinator {
_semaphore.release();
}
+ /// Releases the semaphore while the segment is still mutable, recording the
offset it stopped at (exclusive).
+ public void releaseUnsealed(String segmentName, StreamPartitionMsgOffset
stoppedAtOffset) {
+ _unsealedRelease.set(new UnsealedRelease(segmentName, stoppedAtOffset));
+ _semaphore.release();
+ }
+
+ /// Clears the unsealed release recorded by the given segment, once that
segment is offloaded.
+ public void clearUnsealedRelease(String segmentName) {
+ _unsealedRelease.updateAndGet(
+ release -> release != null &&
release.segmentName().equals(segmentName) ? null : release);
+ }
+
+ @Nullable
+ public UnsealedRelease getUnsealedRelease() {
+ return _unsealedRelease.get();
+ }
+
@VisibleForTesting
Semaphore getSemaphore() {
return _semaphore;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index b330c781dff..c5209138ff4 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -831,13 +831,23 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
// persisted.
// Take upsert snapshot before starting consuming events
if (_partitionUpsertMetadataManager != null) {
+ // If the previous consuming segment released the semaphore to build
or download in parallel, it can still be
+ // mutable in the metadata, so this replica's snapshot covers its
rows up to where it stopped. Only use a
+ // record left by the segment right before this one. An older
segment's offset would make this replica look
+ // far behind, although it consumed every segment since.
+ ConsumerCoordinator.UnsealedRelease unsealedRelease =
_consumerCoordinator.getUnsealedRelease();
+ LLCSegmentName releasedSegment =
+ unsealedRelease != null ?
LLCSegmentName.of(unsealedRelease.segmentName()) : null;
+ String consumedUpToOffset =
+ releasedSegment != null && releasedSegment.getSequenceNumber()
== _llcSegmentName.getSequenceNumber() - 1
+ ? unsealedRelease.stoppedAtOffset().toString() :
_startOffset.toString();
if (_partitionUpsertMetadataManager.getContext().getMetadataTTL() >
0) {
// If upsertMetadataTTL is enabled, we will remove expired primary
keys from upsertMetadata
// AFTER taking a snapshot. Taking the snapshot first is crucial
to capture the final
// state of each key before it exits the TTL window. Out-of-TTL
segments are skipped in
// the doAddSegment flow, and the snapshot is used to enableUpsert
on the immutable out-of-TTL segment.
// If no snapshot is found, the entire segment is marked as valid
and queryable.
- _partitionUpsertMetadataManager.takeSnapshot();
+ _partitionUpsertMetadataManager.takeSnapshot(_segmentNameStr,
consumedUpToOffset);
_partitionUpsertMetadataManager.removeExpiredPrimaryKeys();
} else {
// We should remove deleted-keys first and then take a snapshot.
This is because the deletedKeysTTL
@@ -845,7 +855,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
// after this process, we save one commit cycle, ensuring that the
deletion of valid doc IDs is reflected
// immediately
_partitionUpsertMetadataManager.removeExpiredPrimaryKeys();
- _partitionUpsertMetadataManager.takeSnapshot();
+ _partitionUpsertMetadataManager.takeSnapshot(_segmentNameStr,
consumedUpToOffset);
}
}
@@ -1461,7 +1471,11 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
@VisibleForTesting
void closeStreamConsumerAndReleaseSemaphore() {
closeStreamConsumer();
- releaseConsumerSemaphore();
+ if (_consumerSemaphoreAcquired.compareAndSet(true, false)) {
+ _segmentLogger.info("Releasing consumer semaphore before the segment is
sealed, stopped at offset: {}",
+ _currentOffset);
+ _consumerCoordinator.releaseUnsealed(_segmentNameStr, _currentOffset);
+ }
}
private void closePartitionGroupConsumer() {
@@ -1821,6 +1835,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
_realtimeSegment.offload();
} finally {
releaseConsumerSemaphore();
+ _consumerCoordinator.clearUnsealedRelease(_segmentNameStr);
cleanupMetrics();
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinatorTest.java
index 0650f059c32..836e6ccebac 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinatorTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinatorTest.java
@@ -28,6 +28,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.util.TestUtils;
import org.mockito.Mockito;
@@ -422,6 +423,30 @@ public class ConsumerCoordinatorTest {
Assert.assertFalse(consumerCoordinator.getSemaphore().hasQueuedThreads());
}
+ @Test
+ public void testUnsealedRelease()
+ throws Exception {
+ RealtimeTableDataManager realtimeTableDataManager =
Mockito.mock(RealtimeTableDataManager.class);
+
Mockito.when(realtimeTableDataManager.getTableName()).thenReturn("tableTest_REALTIME");
+ ConsumerCoordinator consumerCoordinator = new ConsumerCoordinator(false,
realtimeTableDataManager);
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(getSegmentName(101));
+ Assert.assertNotNull(llcSegmentName);
+ consumerCoordinator.acquire(llcSegmentName);
+
+ consumerCoordinator.releaseUnsealed(getSegmentName(101), new
LongMsgOffset(250));
+ Assert.assertEquals(consumerCoordinator.getSemaphore().availablePermits(),
1);
+ ConsumerCoordinator.UnsealedRelease unsealedRelease =
consumerCoordinator.getUnsealedRelease();
+ Assert.assertNotNull(unsealedRelease);
+ Assert.assertEquals(unsealedRelease.segmentName(), getSegmentName(101));
+ Assert.assertEquals(unsealedRelease.stoppedAtOffset().toString(), "250");
+
+ // Only the segment that recorded the release can clear it.
+ consumerCoordinator.clearUnsealedRelease(getSegmentName(100));
+ Assert.assertNotNull(consumerCoordinator.getUnsealedRelease());
+ consumerCoordinator.clearUnsealedRelease(getSegmentName(101));
+ Assert.assertNull(consumerCoordinator.getUnsealedRelease());
+ }
+
@Test
public void testPreviousSegment() {
RealtimeTableDataManager realtimeTableDataManager =
Mockito.mock(RealtimeTableDataManager.class);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index 6dd607bb91c..5ab454edf28 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -58,6 +58,7 @@ import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationD
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
+import org.apache.pinot.segment.local.upsert.UpsertContext;
import org.apache.pinot.segment.local.utils.SegmentLocks;
import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache;
import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
@@ -78,9 +79,11 @@ import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.mockito.InOrder;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.mockito.ArgumentMatchers.any;
@@ -92,11 +95,13 @@ import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
// TODO Re-write this test using the stream abstraction
@@ -288,6 +293,78 @@ public class RealtimeSegmentDataManagerTest {
SegmentBuildTimeLeaseExtender.shutdownExecutor();
}
+ @Test
+ public void testSnapshotTriggerContextBeforeConsumption()
+ throws Exception {
+ for (double metadataTtl : new double[]{0, 1}) {
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ PartitionUpsertMetadataManager manager =
mockUpsertMetadataManager(segmentDataManager, metadataTtl);
+ doAnswer(invocation -> {
+ assertEquals(segmentDataManager.getCurrentOffset().toString(),
START_OFFSET.toString());
+ return null;
+ }).when(manager).takeSnapshot(SEGMENT_NAME_STR,
START_OFFSET.toString());
+ segmentDataManager._consumeOffsets.add(new
LongMsgOffset(START_OFFSET_VALUE + 1));
+ segmentDataManager.createPartitionConsumer().run();
+ InOrder order = inOrder(manager);
+ if (metadataTtl > 0) {
+ order.verify(manager).takeSnapshot(SEGMENT_NAME_STR,
START_OFFSET.toString());
+ order.verify(manager).removeExpiredPrimaryKeys();
+ } else {
+ order.verify(manager).removeExpiredPrimaryKeys();
+ order.verify(manager).takeSnapshot(SEGMENT_NAME_STR,
START_OFFSET.toString());
+ }
+ }
+ }
+ }
+
+ @DataProvider
+ public Object[][] unsealedReleases() {
+ long ahead = START_OFFSET_VALUE + 50;
+ long behind = START_OFFSET_VALUE - 50;
+ return new Object[][]{
+ // The previous segment discarded its copy after consuming past this
segment's start offset.
+ {SEQUENCE_ID - 1, ahead, ahead},
+ // The previous segment could not catch up and is downloading, so its
rows end before this segment's start.
+ {SEQUENCE_ID - 1, behind, behind},
+ // An older segment was never offloaded, but this replica consumed the
segments after it.
+ {SEQUENCE_ID - 2, behind, START_OFFSET_VALUE}
+ };
+ }
+
+ @Test(dataProvider = "unsealedReleases")
+ public void testSnapshotTriggerUsesWhereUnsealedPreviousSegmentStopped(int
releasedSequence, long stoppedAt,
+ long expectedConsumedUpTo)
+ throws Exception {
+ _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID);
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ // The released segment is still mutable in the upsert metadata when
this segment starts consuming.
+ ConsumerCoordinator consumerCoordinator =
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID);
+ Assert.assertTrue(consumerCoordinator.getSemaphore().tryAcquire());
+ String releasedSegment =
+ new LLCSegmentName(RAW_TABLE_NAME, PARTITION_GROUP_ID,
releasedSequence, SEG_TIME_MS).getSegmentName();
+ consumerCoordinator.releaseUnsealed(releasedSegment, new
LongMsgOffset(stoppedAt));
+ PartitionUpsertMetadataManager manager =
mockUpsertMetadataManager(segmentDataManager, 0);
+ segmentDataManager._consumeOffsets.add(new
LongMsgOffset(START_OFFSET_VALUE + 1));
+ segmentDataManager.createPartitionConsumer().run();
+ verify(manager).takeSnapshot(SEGMENT_NAME_STR,
Long.toString(expectedConsumedUpTo));
+ } finally {
+ _partitionGroupIdToConsumerCoordinatorMap.remove(PARTITION_GROUP_ID);
+ }
+ }
+
+ private static PartitionUpsertMetadataManager mockUpsertMetadataManager(
+ FakeRealtimeSegmentDataManager segmentDataManager, double metadataTtl)
+ throws Exception {
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ UpsertContext context = mock(UpsertContext.class);
+ when(manager.getContext()).thenReturn(context);
+ when(context.getMetadataTTL()).thenReturn(metadataTtl);
+ Field managerField =
RealtimeSegmentDataManager.class.getDeclaredField("_partitionUpsertMetadataManager");
+ managerField.setAccessible(true);
+ managerField.set(segmentDataManager, manager);
+ return manager;
+ }
+
// Test that we are in HOLDING state as long as the controller responds HOLD
to our segmentConsumed() message.
// we should not consume when holding.
@Test
@@ -765,6 +842,11 @@ public class RealtimeSegmentDataManagerTest {
Assert.assertEquals(semaphore.availablePermits(), 1,
"Consumer semaphore must be released for parallel consumption");
Assert.assertFalse(segmentDataManager.getConsumerSemaphoreAcquired().get());
+ ConsumerCoordinator.UnsealedRelease unsealedRelease =
+
_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getUnsealedRelease();
+ Assert.assertNotNull(unsealedRelease, "The next consumer must see where
this still mutable segment stopped");
+ Assert.assertEquals(unsealedRelease.segmentName(), SEGMENT_NAME_STR);
+ Assert.assertEquals(unsealedRelease.stoppedAtOffset().toString(),
START_OFFSET.toString());
MutableSegmentImpl realtimeSegment = spy((MutableSegmentImpl)
segmentDataManager.getSegment());
doAnswer(invocation -> null).when(realtimeSegment).offload();
@@ -774,6 +856,8 @@ public class RealtimeSegmentDataManagerTest {
verify(realtimeSegment).offload();
Assert.assertEquals(semaphore.availablePermits(), 1, "Offload must not
release the consumer semaphore twice");
+
Assert.assertNull(_partitionGroupIdToConsumerCoordinatorMap.get(PARTITION_GROUP_ID).getUnsealedRelease(),
+ "Offload must clear the unsealed release");
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
index 65c06d5af59..b0ac2d625dd 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
@@ -24,7 +24,6 @@ import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -45,6 +44,7 @@ import
org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import
org.apache.pinot.segment.local.segment.virtualcolumn.VirtualColumnContext;
import org.apache.pinot.segment.local.startree.v2.store.StarTreeIndexContainer;
+import org.apache.pinot.segment.local.upsert.DocIdsSnapshot;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import org.apache.pinot.segment.local.upsert.UpsertUtils;
import org.apache.pinot.segment.local.upsert.UpsertViewManager;
@@ -72,7 +72,6 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.OpenStructNaming;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
-import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -203,14 +202,20 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
@Nullable
public MutableRoaringBitmap loadDocIdsFromSnapshot(String fileName) {
+ DocIdsSnapshot snapshot = loadDocIdsSnapshot(fileName);
+ return snapshot != null ? snapshot.docIds() : null;
+ }
+
+ @Nullable
+ public DocIdsSnapshot loadDocIdsSnapshot(String fileName) {
File docIdsSnapshotFile = getSnapshotFile(fileName);
if (docIdsSnapshotFile.exists()) {
try {
byte[] bytes = FileUtils.readFileToByteArray(docIdsSnapshotFile);
- MutableRoaringBitmap docIds = new
ImmutableRoaringBitmap(ByteBuffer.wrap(bytes)).toMutableRoaringBitmap();
+ DocIdsSnapshot snapshot = DocIdsSnapshot.fromBytes(bytes);
LOGGER.info("Loaded docIds from snapshot for segment: {} with: {}
docs", getSegmentName(),
- docIds.getCardinality());
- return docIds;
+ snapshot.docIds().getCardinality());
+ return snapshot;
} catch (Exception e) {
LOGGER.warn("Caught exception while loading docIds from snapshot file:
{}, ignoring the snapshot",
docIdsSnapshotFile);
@@ -219,8 +224,9 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
return null;
}
- /// Persists the doc ids bitmap snapshot into the given file.
- public void persistDocIdsSnapshot(String fileName,
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIdsSnapshot)
+ /// Persists the doc ids bitmap snapshot into the given file, followed by
its diagnostics when given.
+ public void persistDocIdsSnapshot(String fileName,
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIdsSnapshot,
+ @Nullable DocIdsSnapshot.Metadata metadata)
throws IOException {
File tmpFile =
new
File(SegmentDirectoryPaths.findSegmentDirectory(_segmentMetadata.getIndexDir()),
fileName + "_tmp");
@@ -229,7 +235,7 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
FileUtils.deleteQuietly(tmpFile);
}
try (FileOutputStream fos = new FileOutputStream(tmpFile)) {
- fos.write(docIdsSnapshot.getBytes());
+ DocIdsSnapshot.write(fos, docIdsSnapshot.getBytes(), metadata);
}
File docIdsSnapshotFile = getSnapshotFile(fileName);
Preconditions.checkState(tmpFile.renameTo(docIdsSnapshotFile),
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
index f077c59aac7..3a2c00aeab5 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
@@ -54,6 +54,7 @@ import
org.apache.pinot.segment.local.indexsegment.immutable.EmptyIndexSegment;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
+import org.apache.pinot.segment.local.upsert.DocIdsSnapshot.DocIdsType;
import org.apache.pinot.segment.local.utils.HashUtils;
import org.apache.pinot.segment.local.utils.SegmentPreloadUtils;
import org.apache.pinot.segment.local.utils.WatermarkUtils;
@@ -924,6 +925,15 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
@Override
public void takeSnapshot() {
+ takeSnapshot(null);
+ }
+
+ @Override
+ public void takeSnapshot(String consumingSegmentName, String
consumedUpToOffset) {
+ takeSnapshot(new DocIdsSnapshot.Trigger(consumingSegmentName,
consumedUpToOffset));
+ }
+
+ private void takeSnapshot(@Nullable DocIdsSnapshot.Trigger trigger) {
if (!_enableSnapshot) {
return;
}
@@ -944,7 +954,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
boolean consistent = false;
try {
long startTime = System.currentTimeMillis();
- doTakeSnapshot();
+ doTakeSnapshot(trigger);
consistent = noSegmentOperationInProgress &&
_numSegmentOperationsStarted.get() == numSegmentOperationsStarted;
long duration = System.currentTimeMillis() - startTime;
_serverMetrics.addTimedTableValue(_tableNameWithType,
ServerTimer.UPSERT_SNAPSHOT_TIME_MS, duration,
@@ -998,6 +1008,11 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
}
protected void doTakeSnapshot() {
+ doTakeSnapshot(null);
+ }
+
+ /// The trigger is the consumer startup that asked for this snapshot, stored
with each bitmap written in this round.
+ protected void doTakeSnapshot(@Nullable DocIdsSnapshot.Trigger trigger) {
int numTrackedSegments = _trackedSegments.size();
long numPrimaryKeysInSnapshot = 0L;
long numQueryableDocIdsInSnapshot = 0L;
@@ -1068,7 +1083,8 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
// NOTE: Segment out of TTL without snapshot might have null
validDocIds
if (validDocIds != null) {
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes
validDocIdsSnapshot = validDocIds.getBytesAndCardinality();
-
segment.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
validDocIdsSnapshot);
+
segment.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
validDocIdsSnapshot,
+ DocIdsSnapshot.metadata(validDocIdsSnapshot,
DocIdsType.VALID_DOC_IDS, trigger));
numPrimaryKeysInSnapshot += validDocIdsSnapshot.getCardinality();
}
if (_deleteRecordColumn != null) {
@@ -1076,7 +1092,8 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
if (queryableDocIds != null) {
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes
queryableDocIdsSnapshot =
queryableDocIds.getBytesAndCardinality();
-
segment.persistDocIdsSnapshot(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME,
queryableDocIdsSnapshot);
+
segment.persistDocIdsSnapshot(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME,
queryableDocIdsSnapshot,
+ DocIdsSnapshot.metadata(queryableDocIdsSnapshot,
DocIdsType.QUERYABLE_DOC_IDS, trigger));
numQueryableDocIdsInSnapshot +=
queryableDocIdsSnapshot.getCardinality();
}
}
@@ -1118,7 +1135,8 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
if (validDocIds != null) {
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes
validDocIdsSnapshot =
validDocIds.getBytesAndCardinality();
-
segment.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
validDocIdsSnapshot);
+
segment.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
validDocIdsSnapshot,
+ DocIdsSnapshot.metadata(validDocIdsSnapshot,
DocIdsType.VALID_DOC_IDS, trigger));
// The segment has its validDocIds snapshot file on disk now, so
handle it as a segment with snapshot
// from now on, even if persisting the queryableDocIds snapshot
below fails.
_segmentsWithSnapshot.add(segment);
@@ -1129,7 +1147,8 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
if (queryableDocIds != null) {
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes
queryableDocIdsSnapshot =
queryableDocIds.getBytesAndCardinality();
-
segment.persistDocIdsSnapshot(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME,
queryableDocIdsSnapshot);
+
segment.persistDocIdsSnapshot(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME,
queryableDocIdsSnapshot,
+ DocIdsSnapshot.metadata(queryableDocIdsSnapshot,
DocIdsType.QUERYABLE_DOC_IDS, trigger));
numQueryableDocIdsInSnapshot +=
queryableDocIdsSnapshot.getCardinality();
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java
new file mode 100644
index 00000000000..45fd8069ab0
--- /dev/null
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.pinot.segment.local.upsert;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.zip.CRC32;
+import javax.annotation.Nullable;
+import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.roaringbitmap.IntIterator;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// A persisted bitmap and its optional capture diagnostics, read from the
same file generation.
+/// The Roaring bytes remain a prefix readable by existing snapshot readers. A
versioned metadata trailer is published
+/// with that prefix through the existing temporary-file rename, so a
concurrent reader cannot pair different captures.
+public record DocIdsSnapshot(MutableRoaringBitmap docIds, @Nullable Metadata
metadata) {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DocIdsSnapshot.class);
+ private static final int MAGIC = 0x5044534D; // PDSM: Pinot doc ids snapshot
metadata
+ private static final int VERSION = 1;
+
+ /// The consumer startup that triggered a snapshot. consumedUpToOffset is
exclusive and per replica: the consumer's
+ /// start offset, or where the previous segment stopped if it was still
unsealed. It is not a consistency watermark.
+ public record Trigger(String consumingSegmentName, String
consumedUpToOffset) {
+ }
+
+ /// Which bitmap a snapshot file holds, so its CRC says what it covers.
+ public enum DocIdsType {
+ VALID_DOC_IDS, QUERYABLE_DOC_IDS
+ }
+
+ // Ignore unknown fields so an older server still reads diagnostics written
by a newer one.
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ public record Metadata(long docIdsCrc, DocIdsType docIdsType, long
snapshotCapturedAtMs,
+ @Nullable String snapshotConsumingSegmentName,
@Nullable String snapshotConsumedUpToOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("docIdsCrc", docIdsCrc);
+ response.put("docIdsType", docIdsType.name());
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotConsumingSegmentName != null && snapshotConsumedUpToOffset
!= null) {
+ response.put("snapshotConsumingSegmentName",
snapshotConsumingSegmentName);
+ response.put("snapshotConsumedUpToOffset", snapshotConsumedUpToOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Builds the diagnostics for a bitmap read by
[ThreadSafeMutableRoaringBitmap#getBytesAndCardinality()], which
+ /// also took the capture time under the bitmap's lock. Hashes the detached
bytes outside that lock.
+ public static Metadata
metadata(ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIds, DocIdsType
docIdsType,
+ @Nullable Trigger trigger) {
+ ImmutableRoaringBitmap captured = new
ImmutableRoaringBitmap(ByteBuffer.wrap(docIds.getBytes()));
+ CRC32 crc = new CRC32();
+ IntIterator iterator = captured.getIntIterator();
+ // Canonical ascending doc IDs, four big-endian bytes per ID: independent
of Roaring container layout.
+ while (iterator.hasNext()) {
+ int docId = iterator.next();
+ crc.update(docId >>> 24);
+ crc.update(docId >>> 16);
+ crc.update(docId >>> 8);
+ crc.update(docId);
+ }
+ return new Metadata(crc.getValue(), docIdsType, docIds.getCapturedAtMs(),
+ trigger != null ? trigger.consumingSegmentName() : null, trigger !=
null ? trigger.consumedUpToOffset() : null);
+ }
+
+ /// Writes the bitmap and then its diagnostics trailer, without copying the
bitmap bytes. Writes only the bitmap, as
+ /// legacy files do, when metadata is null.
+ public static void write(OutputStream out, byte[] docIdsBytes, @Nullable
Metadata metadata)
+ throws IOException {
+ out.write(docIdsBytes);
+ if (metadata != null) {
+ out.write(ByteBuffer.allocate(2 *
Integer.BYTES).putInt(MAGIC).putInt(VERSION).array());
+
out.write(JsonUtils.objectToString(metadata).getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ public static DocIdsSnapshot fromBytes(byte[] bytes) {
+ ImmutableRoaringBitmap bitmap = new
ImmutableRoaringBitmap(ByteBuffer.wrap(bytes));
+ Metadata metadata = null;
+ int bitmapSize = bitmap.serializedSizeInBytes();
+ if (bytes.length >= bitmapSize + 2 * Integer.BYTES) {
+ ByteBuffer trailer = ByteBuffer.wrap(bytes, bitmapSize, bytes.length -
bitmapSize);
+ if (trailer.getInt() == MAGIC && trailer.getInt() == VERSION) {
+ try {
+ Metadata parsed = JsonUtils.stringToObject(
+ new String(bytes, trailer.position(), trailer.remaining(),
StandardCharsets.UTF_8), Metadata.class);
+ if (parsed.docIdsType() != null && parsed.snapshotCapturedAtMs() > 0
&& parsed.docIdsCrc() >= 0
+ && parsed.docIdsCrc() <= 0xFFFFFFFFL) {
+ metadata = parsed;
+ }
+ } catch (Exception e) {
+ // Diagnostics must not prevent recovery of a valid bitmap.
+ LOGGER.warn("Ignoring unreadable doc ids snapshot diagnostics", e);
+ }
+ }
+ }
+ return new DocIdsSnapshot(bitmap.toMutableRoaringBitmap(), metadata);
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
index 305bc1d63e0..417863ee479 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
@@ -89,6 +89,11 @@ public interface PartitionUpsertMetadataManager extends
Closeable {
/// before a new consuming segment starts consuming.
void takeSnapshot();
+ /// Takes a snapshot with the triggering consumer's startup context. Older
implementations may ignore the context.
+ default void takeSnapshot(String consumingSegmentName, String
consumedUpToOffset) {
+ takeSnapshot();
+ }
+
/// Returns what the last finished snapshot pass covered, or null if no pass
has finished yet.
@Nullable
default SnapshotPass getLastSnapshotPass() {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java
index 6df18ec868b..9b1da299e0e 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java
@@ -104,6 +104,59 @@ public class BasePartitionUpsertMetadataManagerTest {
}
}
+ @Test
+ public void testPersistedSnapshotDiagnosticsRetainCaptureContext()
+ throws IOException {
+ UpsertContext context = mock(UpsertContext.class);
+ when(context.isSnapshotEnabled()).thenReturn(true);
+ TableDataManager tableDataManager = mock(TableDataManager.class);
+ when(context.getTableDataManager()).thenReturn(tableDataManager);
+ Lock segmentLock = mock(Lock.class);
+ when(tableDataManager.getSegmentLock(anyString())).thenReturn(segmentLock);
+ when(segmentLock.tryLock()).thenReturn(true);
+ DummyPartitionUpsertMetadataManager manager = new
DummyPartitionUpsertMetadataManager("myTable", 0, context);
+ setDeleteRecordColumn(manager, "deleted");
+ ImmutableSegmentImpl segment = createImmutableSegment("seg", new
File(TEMP_DIR, "seg"), new ArrayList<>(), null);
+ ThreadSafeMutableRoaringBitmap valid = createDocIds(1, 2, 3);
+ segment.enableUpsert(manager, valid, createDocIds(1, 3));
+ manager.trackSegment(segment);
+ manager.markSegmentAsUpdated(segment);
+ manager.addRecord(mock(MutableSegmentImpl.class), mock(RecordInfo.class));
+
+ manager.takeSnapshot("myTable__0__2__0", "100");
+ DocIdsSnapshot first =
segment.loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME);
+ DocIdsSnapshot queryable =
segment.loadDocIdsSnapshot(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME);
+ assertEquals(first.metadata().snapshotConsumedUpToOffset(), "100");
+ assertEquals(queryable.metadata().snapshotConsumedUpToOffset(), "100");
+ assertEquals(first.metadata().docIdsType(),
DocIdsSnapshot.DocIdsType.VALID_DOC_IDS);
+ assertEquals(queryable.metadata().docIdsType(),
DocIdsSnapshot.DocIdsType.QUERYABLE_DOC_IDS);
+ assertNotEquals(first.metadata().docIdsCrc(),
queryable.metadata().docIdsCrc());
+ assertEquals(first.docIds(), valid.getMutableRoaringBitmap());
+
+ // An unchanged segment keeps its saved timestamp and trigger, even if a
newer consumer triggers a snapshot round.
+ manager.takeSnapshot("myTable__0__3__0", "200");
+
assertEquals(segment.loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME).metadata(),
first.metadata());
+ valid.remove(2);
+ manager.markSegmentAsUpdated(segment);
+ when(segmentLock.tryLock()).thenReturn(false);
+ manager.takeSnapshot("myTable__0__3__0", "200");
+
assertEquals(segment.loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME).metadata(),
first.metadata());
+
+ // Retrying publishes the new bitmap and its own trigger together.
+ when(segmentLock.tryLock()).thenReturn(true);
+ manager.takeSnapshot("myTable__0__4__0", "300");
+ DocIdsSnapshot retried =
segment.loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME);
+ assertEquals(retried.metadata().snapshotConsumedUpToOffset(), "300");
+ assertEquals(retried.docIds(), valid.getMutableRoaringBitmap());
+ assertNotEquals(retried.metadata().docIdsCrc(),
first.metadata().docIdsCrc());
+
+ // Shutdown or explicit snapshots without a consumer must not inherit an
earlier startup's offset.
+ manager.markSegmentAsUpdated(segment);
+ manager.takeSnapshot();
+
assertNull(segment.loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME)
+ .metadata().snapshotConsumedUpToOffset());
+ }
+
@Test
public void testTakeSnapshotInOrder()
throws IOException {
@@ -1189,8 +1242,9 @@ public class BasePartitionUpsertMetadataManagerTest {
when(meta.getName()).thenReturn(segName);
when(meta.getIndexDir()).thenReturn(segDir);
return new ImmutableSegmentImpl(mock(SegmentDirectory.class), meta, new
HashMap<>(), null) {
+ @Override
public void persistDocIdsSnapshot(String fileName,
- ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIdsSnapshot)
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIdsSnapshot,
@Nullable DocIdsSnapshot.Metadata metadata)
throws IOException {
onPersist.run();
if (V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME.equals(fileName)) {
@@ -1199,7 +1253,7 @@ public class BasePartitionUpsertMetadataManagerTest {
&& queryableDocIdsSegmentsTaken != null) {
queryableDocIdsSegmentsTaken.add(segName);
}
- super.persistDocIdsSnapshot(fileName, docIdsSnapshot);
+ super.persistDocIdsSnapshot(fileName, docIdsSnapshot, metadata);
}
};
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
index 1b331cf03f7..35e2adb96d4 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java
@@ -358,7 +358,8 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerTest {
ThreadSafeMutableRoaringBitmap snapshot = new
ThreadSafeMutableRoaringBitmap();
snapshot.add(0);
snapshot.add(2);
-
segment2.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
snapshot.getBytesAndCardinality());
+
segment2.persistDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
snapshot.getBytesAndCardinality(),
+ null);
upsertMetadataManager.replaceSegment(segment2, segment1);
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshotTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshotTest.java
new file mode 100644
index 00000000000..963a298acf0
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshotTest.java
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.pinot.segment.local.upsert;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.apache.pinot.segment.local.upsert.DocIdsSnapshot.DocIdsType;
+import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+public class DocIdsSnapshotTest {
+ @Test
+ public void testCaptureAndLegacyReaders()
+ throws Exception {
+ ThreadSafeMutableRoaringBitmap bitmap = bitmap(1, 4, 6, 10, 15, 17, 18,
20);
+ long before = System.currentTimeMillis();
+ byte[] bytes = fileBytes(bitmap, new
DocIdsSnapshot.Trigger("table__0__2__0", "123"));
+ long after = System.currentTimeMillis();
+ DocIdsSnapshot snapshot = DocIdsSnapshot.fromBytes(bytes);
+ assertEquals(snapshot.docIds(), bitmap.getMutableRoaringBitmap());
+ assertEquals(snapshot.metadata().docIdsCrc(), 3650129781L);
+ assertEquals(snapshot.metadata().docIdsType(), DocIdsType.VALID_DOC_IDS);
+ assertTrue(snapshot.metadata().snapshotCapturedAtMs() >= before);
+ assertTrue(snapshot.metadata().snapshotCapturedAtMs() <= after);
+ assertEquals(snapshot.metadata().snapshotConsumedUpToOffset(), "123");
+ assertEquals(snapshot.metadata().snapshotConsumingSegmentName(),
"table__0__2__0");
+ // Both reader styles used by old servers and snapshot consumers ignore
the trailer.
+ assertEquals(new
ImmutableRoaringBitmap(ByteBuffer.wrap(bytes)).toMutableRoaringBitmap(),
snapshot.docIds());
+ MutableRoaringBitmap legacy = new MutableRoaringBitmap();
+ legacy.deserialize(new DataInputStream(new ByteArrayInputStream(bytes)));
+ assertEquals(legacy, snapshot.docIds());
+ }
+
+ @Test
+ public void
testChecksumTracksMembershipRatherThanCardinalityOrContainerLayout()
+ throws Exception {
+ assertNotEquals(crc(bitmap(1, 2)), crc(bitmap(1, 3)));
+ MutableRoaringBitmap array = new MutableRoaringBitmap();
+ for (int i = 0; i < 1000; i++) {
+ array.add(i);
+ }
+ MutableRoaringBitmap runs = array.clone();
+ assertTrue(runs.runOptimize());
+ assertNotEquals(array.serializedSizeInBytes(),
runs.serializedSizeInBytes());
+ assertEquals(crc(new ThreadSafeMutableRoaringBitmap(array)), crc(new
ThreadSafeMutableRoaringBitmap(runs)));
+ assertEquals(crc(bitmap()), 0L);
+ }
+
+ @Test
+ public void testLegacyAndUnreadableDiagnosticsDoNotPreventBitmapRecovery()
+ throws Exception {
+ ThreadSafeMutableRoaringBitmap bitmap = bitmap(1, 2, 3);
+ byte[] legacy = bitmap.getBytes();
+ assertNull(DocIdsSnapshot.fromBytes(legacy).metadata());
+ byte[] bytes = fileBytes(bitmap, null);
+ byte[] truncated = Arrays.copyOf(bytes, bytes.length - 1);
+ assertNull(DocIdsSnapshot.fromBytes(truncated).metadata());
+ assertEquals(DocIdsSnapshot.fromBytes(truncated).docIds(),
bitmap.getMutableRoaringBitmap());
+ ByteBuffer.wrap(bytes).putInt(legacy.length + Integer.BYTES, 99);
+ assertNull(DocIdsSnapshot.fromBytes(bytes).metadata());
+ assertEquals(DocIdsSnapshot.fromBytes(bytes).docIds(),
bitmap.getMutableRoaringBitmap());
+ }
+
+ @Test
+ public void testUnknownDiagnosticsFieldsAreIgnored()
+ throws Exception {
+ byte[] bytes = fileBytes(bitmap(1, 2, 3), null);
+ // A newer server may add a field to the trailer. Insert one before the
closing brace.
+ byte[] extra = ",\"futureField\":1}".getBytes(StandardCharsets.UTF_8);
+ byte[] newer = Arrays.copyOf(bytes, bytes.length - 1 + extra.length);
+ System.arraycopy(extra, 0, newer, bytes.length - 1, extra.length);
+ assertNotNull(DocIdsSnapshot.fromBytes(newer).metadata());
+ assertEquals(DocIdsSnapshot.fromBytes(newer).metadata(),
DocIdsSnapshot.fromBytes(bytes).metadata());
+ }
+
+ @Test
+ public void testAgeAndAbsentTrigger() {
+ DocIdsSnapshot.Metadata metadata = new DocIdsSnapshot.Metadata(1L,
DocIdsType.QUERYABLE_DOC_IDS, 100L, null, null);
+ assertEquals(metadata.toResponse(130L).get("snapshotAgeMs"), 30L);
+ assertEquals(metadata.toResponse(130L).get("docIdsType"),
"QUERYABLE_DOC_IDS");
+ assertFalse(metadata.toResponse(99L).containsKey("snapshotAgeMs"));
+
assertFalse(metadata.toResponse(130L).containsKey("snapshotConsumedUpToOffset"));
+
assertFalse(metadata.toResponse(130L).containsKey("snapshotConsumingSegmentName"));
+ }
+
+ private static long crc(ThreadSafeMutableRoaringBitmap bitmap)
+ throws Exception {
+ return DocIdsSnapshot.fromBytes(fileBytes(bitmap,
null)).metadata().docIdsCrc();
+ }
+
+ private static byte[] fileBytes(ThreadSafeMutableRoaringBitmap bitmap,
DocIdsSnapshot.Trigger trigger)
+ throws Exception {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIds =
bitmap.getBytesAndCardinality();
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ DocIdsSnapshot.write(out, docIds.getBytes(),
DocIdsSnapshot.metadata(docIds, DocIdsType.VALID_DOC_IDS, trigger));
+ return out.toByteArray();
+ }
+
+ private static ThreadSafeMutableRoaringBitmap bitmap(int... ids) {
+ ThreadSafeMutableRoaringBitmap bitmap = new
ThreadSafeMutableRoaringBitmap();
+ for (int id : ids) {
+ bitmap.add(id);
+ }
+ return bitmap;
+ }
+}
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/ThreadSafeMutableRoaringBitmap.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/ThreadSafeMutableRoaringBitmap.java
index d57800f3ab7..aced62a49b5 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/ThreadSafeMutableRoaringBitmap.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/mutable/ThreadSafeMutableRoaringBitmap.java
@@ -80,10 +80,10 @@ public class ThreadSafeMutableRoaringBitmap {
return serialize();
}
- /// Returns a consistent point-in-time snapshot containing both cardinality
and serialized bytes captured under a
- /// single lock.
+ /// Returns a consistent point-in-time snapshot containing the cardinality,
the serialized bytes and the time they
+ /// were read, all captured under a single lock.
public synchronized CardinalityAndBytes getBytesAndCardinality() {
- return new CardinalityAndBytes(_mutableRoaringBitmap.getCardinality(),
serialize());
+ return new CardinalityAndBytes(_mutableRoaringBitmap.getCardinality(),
serialize(), System.currentTimeMillis());
}
private byte[] serialize() {
@@ -95,10 +95,16 @@ public class ThreadSafeMutableRoaringBitmap {
public static final class CardinalityAndBytes {
private final int _cardinality;
private final byte[] _bytes;
+ private final long _capturedAtMs;
public CardinalityAndBytes(int cardinality, byte[] bytes) {
+ this(cardinality, bytes, System.currentTimeMillis());
+ }
+
+ public CardinalityAndBytes(int cardinality, byte[] bytes, long
capturedAtMs) {
_cardinality = cardinality;
_bytes = bytes;
+ _capturedAtMs = capturedAtMs;
}
public int getCardinality() {
@@ -108,5 +114,9 @@ public class ThreadSafeMutableRoaringBitmap {
public byte[] getBytes() {
return _bytes;
}
+
+ public long getCapturedAtMs() {
+ return _capturedAtMs;
+ }
}
}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
index d4f878aa94f..969692312a9 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
@@ -104,6 +104,7 @@ import
org.apache.pinot.segment.local.data.manager.SegmentDataManager;
import org.apache.pinot.segment.local.data.manager.StaleSegment;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
+import org.apache.pinot.segment.local.upsert.DocIdsSnapshot;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import
org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager.SnapshotPass;
import org.apache.pinot.segment.spi.ColumnMetadata;
@@ -664,10 +665,11 @@ public class TablesResource {
}
ServiceStatus.Status status =
ServiceStatus.getServiceStatus(_instanceId);
- final Pair<ValidDocIdsType, MutableRoaringBitmap>
validDocIdsSnapshotPair =
+ final Pair<ValidDocIdsType, DocIdsSnapshot> validDocIdsSnapshotPair =
getValidDocIds(indexSegment, validDocIdsType);
ValidDocIdsType finalValidDocIdsType = validDocIdsSnapshotPair.getLeft();
- MutableRoaringBitmap validDocIdSnapshot =
validDocIdsSnapshotPair.getRight();
+ MutableRoaringBitmap validDocIdSnapshot =
+ validDocIdsSnapshotPair.getRight() != null ?
validDocIdsSnapshotPair.getRight().docIds() : null;
if (validDocIdSnapshot == null) {
String msg = String.format(
@@ -747,10 +749,11 @@ public class TablesResource {
continue;
}
- final Pair<ValidDocIdsType, MutableRoaringBitmap>
validDocIdSnapshotPair =
+ final Pair<ValidDocIdsType, DocIdsSnapshot> validDocIdSnapshotPair =
getValidDocIds(indexSegment, validDocIdsType);
String finalValidDocIdsType =
validDocIdSnapshotPair.getLeft().toString();
- MutableRoaringBitmap validDocIdsSnapshot =
validDocIdSnapshotPair.getRight();
+ DocIdsSnapshot snapshot = validDocIdSnapshotPair.getRight();
+ MutableRoaringBitmap validDocIdsSnapshot = snapshot != null ?
snapshot.docIds() : null;
if (validDocIdsSnapshot == null) {
if (LOGGER.isDebugEnabled()) {
String msg = String.format(
@@ -764,6 +767,9 @@ public class TablesResource {
}
Map<String, Object> validDocIdsMetadata = new HashMap<>();
+ if (snapshot.metadata() != null) {
+ validDocIdsMetadata.put("diagnostics",
snapshot.metadata().toResponse(System.currentTimeMillis()));
+ }
int totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
int totalValidDocs = validDocIdsSnapshot.getCardinality();
int totalInvalidDocs = totalDocs - totalValidDocs;
@@ -823,32 +829,34 @@ public class TablesResource {
return dataCrc >= 0 ? Long.toString(dataCrc) : null;
}
- private Pair<ValidDocIdsType, MutableRoaringBitmap>
getValidDocIds(IndexSegment indexSegment,
+ private Pair<ValidDocIdsType, DocIdsSnapshot> getValidDocIds(IndexSegment
indexSegment,
String validDocIdsTypeStr) {
if (validDocIdsTypeStr == null) {
// By default, we read the valid doc ids from snapshot.
return Pair.of(ValidDocIdsType.SNAPSHOT,
- ((ImmutableSegmentImpl)
indexSegment).loadDocIdsFromSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
+ ((ImmutableSegmentImpl)
indexSegment).loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
}
ValidDocIdsType validDocIdsType =
ValidDocIdsType.valueOf(validDocIdsTypeStr.toUpperCase());
switch (validDocIdsType) {
case SNAPSHOT:
return Pair.of(validDocIdsType,
- ((ImmutableSegmentImpl)
indexSegment).loadDocIdsFromSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
+ ((ImmutableSegmentImpl)
indexSegment).loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
case SNAPSHOT_WITH_DELETE:
return Pair.of(validDocIdsType,
- ((ImmutableSegmentImpl) indexSegment).loadDocIdsFromSnapshot(
+ ((ImmutableSegmentImpl) indexSegment).loadDocIdsSnapshot(
V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME));
case IN_MEMORY:
- return Pair.of(validDocIdsType,
indexSegment.getValidDocIds().getMutableRoaringBitmap());
+ return Pair.of(validDocIdsType,
+ new
DocIdsSnapshot(indexSegment.getValidDocIds().getMutableRoaringBitmap(), null));
case IN_MEMORY_WITH_DELETE:
- return Pair.of(validDocIdsType,
indexSegment.getQueryableDocIds().getMutableRoaringBitmap());
+ return Pair.of(validDocIdsType,
+ new
DocIdsSnapshot(indexSegment.getQueryableDocIds().getMutableRoaringBitmap(),
null));
default:
// By default, we read the valid doc ids from snapshot.
LOGGER.warn("Invalid validDocIdsType: {}. Using default
validDocIdsType: {}", validDocIdsType,
ValidDocIdsType.SNAPSHOT);
return Pair.of(ValidDocIdsType.SNAPSHOT,
- ((ImmutableSegmentImpl)
indexSegment).loadDocIdsFromSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
+ ((ImmutableSegmentImpl)
indexSegment).loadDocIdsSnapshot(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME));
}
}
diff --git
a/pinot-server/src/test/java/org/apache/pinot/server/api/TablesResourceTest.java
b/pinot-server/src/test/java/org/apache/pinot/server/api/TablesResourceTest.java
index 82ceea43fdc..40efd3a9400 100644
---
a/pinot-server/src/test/java/org/apache/pinot/server/api/TablesResourceTest.java
+++
b/pinot-server/src/test/java/org/apache/pinot/server/api/TablesResourceTest.java
@@ -48,6 +48,7 @@ import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImp
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.upsert.DocIdsSnapshot;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import
org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager.SnapshotPass;
import org.apache.pinot.segment.spi.ImmutableSegment;
@@ -76,6 +77,7 @@ import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
@@ -409,6 +411,7 @@ public class TablesResourceTest extends BaseResourceTest {
.post(Entity.json(tableSegments), String.class);
JsonNode validDocIdsMetadata = JsonUtils.stringToJsonNode(response).get(0);
+ assertFalse(validDocIdsMetadata.has("diagnostics"));
assertEquals(validDocIdsMetadata.get("totalDocs").asInt(), 200000);
assertEquals(validDocIdsMetadata.get("totalValidDocs").asInt(), 8);
assertEquals(validDocIdsMetadata.get("totalInvalidDocs").asInt(), 199992);
@@ -458,6 +461,7 @@ public class TablesResourceTest extends BaseResourceTest {
.post(Entity.json(tableSegments), String.class);
JsonNode validDocIdsMetadata = JsonUtils.stringToJsonNode(response).get(0);
+ assertFalse(validDocIdsMetadata.has("diagnostics"));
assertEquals(validDocIdsMetadata.get("totalDocs").asInt(), 200000);
assertEquals(validDocIdsMetadata.get("totalValidDocs").asInt(), 8);
assertEquals(validDocIdsMetadata.get("totalInvalidDocs").asInt(), 199992);
@@ -475,6 +479,56 @@ public class TablesResourceTest extends BaseResourceTest {
assertEquals(serverStatus, "NOT_STARTED", serverStatus);
}
+ @DataProvider
+ public Object[][] diagnosticBitmapTypes() {
+ return new Object[][] {{null}, {"SNAPSHOT"}, {"SNAPSHOT_WITH_DELETE"},
{"IN_MEMORY"}, {"IN_MEMORY_WITH_DELETE"}};
+ }
+
+ @Test(dataProvider = "diagnosticBitmapTypes")
+ public void testValidDocIdsMetadataDiagnostics(String bitmapType)
+ throws IOException {
+ ImmutableSegmentImpl segment = (ImmutableSegmentImpl)
_realtimeIndexSegments.get(0);
+ downLoadAndVerifyValidDocIdsSnapshotBitmap(REALTIME_TABLE_NAME, segment);
+ // Upgrade both legacy fixtures to real persisted captures, without
changing the live bitmaps.
+ DocIdsSnapshot.Trigger trigger = new
DocIdsSnapshot.Trigger("testTable__0__2__0", "123");
+ Map<String, DocIdsSnapshot.DocIdsType> fileTypes =
+ Map.of(V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME,
DocIdsSnapshot.DocIdsType.VALID_DOC_IDS,
+ V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME,
DocIdsSnapshot.DocIdsType.QUERYABLE_DOC_IDS);
+ for (Map.Entry<String, DocIdsSnapshot.DocIdsType> fileType :
fileTypes.entrySet()) {
+ ThreadSafeMutableRoaringBitmap bitmap =
+ new
ThreadSafeMutableRoaringBitmap(segment.loadDocIdsFromSnapshot(fileType.getKey()));
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIds =
bitmap.getBytesAndCardinality();
+ segment.persistDocIdsSnapshot(fileType.getKey(), docIds,
+ DocIdsSnapshot.metadata(docIds, fileType.getValue(), trigger));
+ }
+ long beforeRequest = System.currentTimeMillis();
+ String response = _webTarget.path("/tables/" + REALTIME_TABLE_NAME +
"/validDocIdsMetadata")
+ .queryParam("validDocIdsType", bitmapType)
+ .request().post(Entity.json(new
TableSegments(List.of(segment.getSegmentName()))), String.class);
+ long afterRequest = System.currentTimeMillis();
+ JsonNode metadata = JsonUtils.stringToJsonNode(response).get(0);
+ assertEquals(metadata.get("totalValidDocs").asInt(), 8);
+ String expectedType = bitmapType != null ? bitmapType : "SNAPSHOT";
+ assertEquals(metadata.get("validDocIdsType").asText(), expectedType);
+ if (expectedType.startsWith("IN_MEMORY")) {
+ assertFalse(metadata.has("diagnostics"));
+ return;
+ }
+ JsonNode diagnostics = metadata.get("diagnostics");
+ String file = expectedType.equals("SNAPSHOT") ?
V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME
+ : V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME;
+ long capturedAt =
segment.loadDocIdsSnapshot(file).metadata().snapshotCapturedAtMs();
+ assertEquals(diagnostics.get("snapshotCapturedAtMs").asLong(), capturedAt);
+ assertTrue(diagnostics.get("snapshotAgeMs").asLong() >= beforeRequest -
capturedAt);
+ assertTrue(diagnostics.get("snapshotAgeMs").asLong() <= afterRequest -
capturedAt);
+ assertEquals(diagnostics.get("snapshotConsumingSegmentName").asText(),
trigger.consumingSegmentName());
+ assertEquals(diagnostics.get("snapshotConsumedUpToOffset").asText(),
trigger.consumedUpToOffset());
+ assertEquals(diagnostics.get("docIdsType").asText(),
+ expectedType.equals("SNAPSHOT") ? "VALID_DOC_IDS" :
"QUERYABLE_DOC_IDS");
+ assertEquals(diagnostics.get("docIdsCrc").asLong(),
+ expectedType.equals("SNAPSHOT") ? 4200314552L : 569535174L);
+ }
+
// Verify metadata file from segments.
private void downloadAndVerifySegmentContent(String tableNameWithType,
IndexSegment segment)
throws IOException, ConfigurationException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
