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 43ecf19f925 Report whether upsert snapshot files can be compared
across replicas (#19728)
43ecf19f925 is described below
commit 43ecf19f925465f13e5a25b0630eb7b3880c1279
Author: Kartik Khare <[email protected]>
AuthorDate: Wed Oct 7 10:13:58 2026 +0530
Report whether upsert snapshot files can be compared across replicas
(#19728)
* Report whether upsert snapshot files can be compared across replicas
A snapshot pass does not block segment adds, replaces or removes, and SRT,
FIT or compaction uploads land on each replica at a different time. So two
replicas' validDocIds snapshot files can differ without any divergence.
After each pass, record a CRC of the sealed segment names, their count, and
whether the pass was a consistent cut: no segment operation overlapped it,
every changed segment got its file written, and no earlier consuming segment
was still waiting to be sealed. Serve it as snapshotPass from
validDocIdsMetadata for the SNAPSHOT types.
* Version each segment by data CRC in the snapshot pass CRC
Minion tasks such as upsert compaction upload a new copy of a segment under
the same name, so a CRC over names alone could call two replicas comparable
when only one had applied the new copy. Hash each sealed segment as name
plus
data CRC, falling back to the full CRC when there is no data CRC.
---------
Co-authored-by: Kartik Khare <[email protected]>
---
.../immutable/ImmutableSegmentImpl.java | 5 +
.../upsert/BasePartitionUpsertMetadataManager.java | 66 +++++++++
.../upsert/PartitionUpsertMetadataManager.java | 14 ++
.../BasePartitionUpsertMetadataManagerTest.java | 153 +++++++++++++++++++++
.../pinot/server/api/resources/TablesResource.java | 16 +++
.../pinot/server/api/TablesResourceTest.java | 15 ++
6 files changed, 269 insertions(+)
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 fbdd9a5ebc6..0cef98c9c97 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
@@ -422,6 +422,11 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
return _multiColumnTextIndex;
}
+ @Nullable
+ public PartitionUpsertMetadataManager getPartitionUpsertMetadataManager() {
+ return _partitionUpsertMetadataManager;
+ }
+
@Nullable
@Override
public ThreadSafeMutableRoaringBitmap getValidDocIds() {
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 09a83c7becd..f88a71434e4 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
@@ -23,18 +23,23 @@ import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.AtomicDouble;
import java.io.File;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
+import java.util.zip.CRC32;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.helix.HelixManager;
@@ -147,6 +152,12 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
private final Map<String, Long> _newlyAddedSegments = new
ConcurrentHashMap<>();
private final long _newSegmentTrackingTimeMs;
+ // Segment adds, preloads, replaces and removes run on Helix threads and can
overlap a snapshot pass. A pass is a
+ // consistent cut only if none was in progress when it started and none
started before it finished.
+ private final AtomicInteger _numSegmentOperationsInProgress = new
AtomicInteger();
+ private final AtomicLong _numSegmentOperationsStarted = new AtomicLong();
+ private volatile SnapshotPass _lastSnapshotPass;
+
protected BasePartitionUpsertMetadataManager(String tableNameWithType, int
partitionId, UpsertContext context) {
_tableNameWithType = tableNameWithType;
_tableType = TableNameBuilder.getTableTypeFromTableName(tableNameWithType);
@@ -303,11 +314,13 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
return;
}
ImmutableSegmentImpl immutableSegment = (ImmutableSegmentImpl) segment;
+ startSegmentOperation();
try {
doAddSegment(immutableSegment);
_trackedSegments.add(immutableSegment);
trackSegmentForSnapshot(immutableSegment);
} finally {
+ finishSegmentOperation();
finishOperation();
}
}
@@ -431,11 +444,13 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
return;
}
ImmutableSegmentImpl immutableSegment = (ImmutableSegmentImpl) segment;
+ startSegmentOperation();
try {
doPreloadSegment(immutableSegment);
_trackedSegments.add(immutableSegment);
trackSegmentForSnapshot(immutableSegment);
} finally {
+ finishSegmentOperation();
finishOperation();
}
}
@@ -585,6 +600,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
_logger.info("Skip replacing segment: {} because metadata manager is
already stopped", segment.getSegmentName());
return;
}
+ startSegmentOperation();
try {
doReplaceSegment(segment, oldSegment);
if (segment instanceof ImmutableSegmentImpl immutableSegment) {
@@ -595,6 +611,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
}
untrackSegment(oldSegment);
} finally {
+ finishSegmentOperation();
finishOperation();
}
}
@@ -791,6 +808,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
_logger.info("Skip removing segment: {} because metadata manager is
already stopped", segmentName);
return;
}
+ startSegmentOperation();
try {
// Skip removing the upsert metadata of segment that is out of metadata
TTL. The expired metadata is removed
// while creating new consuming segment in batches.
@@ -801,6 +819,7 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
}
untrackSegment(segment);
} finally {
+ finishSegmentOperation();
finishOperation();
}
}
@@ -918,19 +937,66 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
_logger.info("Skip taking snapshot because metadata manager is already
stopped");
return;
}
+ // Read the started count before the in-progress count, the reverse of the
order startSegmentOperation() bumps them
+ // in, so an operation that overlaps the pass always shows up in one of
the two
+ long numSegmentOperationsStarted = _numSegmentOperationsStarted.get();
+ boolean noSegmentOperationInProgress =
_numSegmentOperationsInProgress.get() == 0;
+ boolean consistent = false;
try {
long startTime = System.currentTimeMillis();
doTakeSnapshot();
+ consistent = noSegmentOperationInProgress &&
_numSegmentOperationsStarted.get() == numSegmentOperationsStarted;
long duration = System.currentTimeMillis() - startTime;
_serverMetrics.addTimedTableValue(_tableNameWithType,
ServerTimer.UPSERT_SNAPSHOT_TIME_MS, duration,
TimeUnit.MILLISECONDS);
} catch (Exception e) {
_logger.warn("Caught exception while taking snapshot", e);
} finally {
+ recordSnapshotPass(consistent);
finishOperation();
}
}
+ @Nullable
+ @Override
+ public SnapshotPass getLastSnapshotPass() {
+ return _lastSnapshotPass;
+ }
+
+ private void startSegmentOperation() {
+ _numSegmentOperationsInProgress.incrementAndGet();
+ _numSegmentOperationsStarted.incrementAndGet();
+ }
+
+ private void finishSegmentOperation() {
+ _numSegmentOperationsInProgress.decrementAndGet();
+ }
+
+ private void recordSnapshotPass(boolean consistent) {
+ // Name alone is not enough, because minion tasks such as upsert
compaction upload a new copy of a segment under the
+ // same name. It uses the data CRC rather than the full CRC, so a local
build kept on a data CRC match, e.g. with a
+ // Lucene text index, still matches the committed copy.
+ TreeMap<String, Long> segmentVersions = new TreeMap<>();
+ for (IndexSegment segment : _trackedSegments) {
+ if (segment instanceof ImmutableSegmentImpl) {
+ SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
+ long dataCrc = segmentMetadata.getDataCrc();
+ segmentVersions.put(segment.getSegmentName(), dataCrc >= 0 ? dataCrc :
segmentMetadata.getCrc());
+ } else {
+ // The pass runs before the new consuming segment takes its first
record, so a tracked mutable segment is an
+ // earlier one still waiting to be sealed. Its rows may not match the
committed copy yet.
+ consistent = false;
+ }
+ }
+ CRC32 crc = new CRC32();
+ for (Map.Entry<String, Long> entry : segmentVersions.entrySet()) {
+ crc.update((entry.getKey() + ':' + entry.getValue() +
'\n').getBytes(StandardCharsets.UTF_8));
+ }
+ // A changed segment the pass skipped, e.g. on segmentLock contention,
keeps a stale snapshot file
+ _lastSnapshotPass = new SnapshotPass(crc.getValue(),
segmentVersions.size(),
+ consistent && _updatedSegmentsSinceLastSnapshot.isEmpty(),
System.currentTimeMillis());
+ }
+
protected void doTakeSnapshot() {
int numTrackedSegments = _trackedSegments.size();
long numPrimaryKeysInSnapshot = 0L;
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 112b2c17f55..305bc1d63e0 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,20 @@ public interface PartitionUpsertMetadataManager extends
Closeable {
/// before a new consuming segment starts consuming.
void takeSnapshot();
+ /// Returns what the last finished snapshot pass covered, or null if no pass
has finished yet.
+ @Nullable
+ default SnapshotPass getLastSnapshotPass() {
+ return null;
+ }
+
+ // Two replicas whose last passes are consistent and cover the same sealed
segments, by name and data CRC, took their
+ // snapshots from the same inputs, so those snapshots can be compared.
Consistent means no segment was added,
+ // replaced or removed while the pass ran, every changed segment got its
snapshot file written, and no earlier
+ // consuming segment was still waiting to be sealed. A later segment change
either drops a segment's snapshot file or
+ // leaves it alone, so the files left on disk stay as of the pass.
+ record SnapshotPass(long segmentsCrc, int numSegments, boolean consistent,
long finishedAtMs) {
+ }
+
/// Remove the expired primary keys from the metadata when TTL is enabled.
void removeExpiredPrimaryKeys();
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 ca3e2da05a5..6df18ec868b 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
@@ -42,6 +42,7 @@ import
org.apache.pinot.segment.local.data.manager.TableDataManager;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
import org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl;
import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import
org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager.SnapshotPass;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.MutableSegment;
import org.apache.pinot.segment.spi.SegmentContext;
@@ -1039,9 +1040,150 @@ public class BasePartitionUpsertMetadataManagerTest {
return new ThreadSafeMutableRoaringBitmap(bitmap);
}
+ @Test
+ public void testSnapshotPassComparableAcrossReplicas()
+ throws IOException {
+ UpsertContext upsertContext = createSnapshotPassContext();
+ DummyPartitionUpsertMetadataManager replica1 = new
DummyPartitionUpsertMetadataManager("myTable", 0, upsertContext);
+ DummyPartitionUpsertMetadataManager replica2 = new
DummyPartitionUpsertMetadataManager("myTable", 0, upsertContext);
+ assertNull(replica1.getLastSnapshotPass());
+
+ File replica1Dir = new File(TEMP_DIR, "replica1");
+ File replica2Dir = new File(TEMP_DIR, "replica2");
+ addSnapshotSegment(replica1, replica1Dir, "seg01");
+ addSnapshotSegment(replica1, replica1Dir, "seg02");
+ addSnapshotSegment(replica2, replica2Dir, "seg02");
+ addSnapshotSegment(replica2, replica2Dir, "seg01");
+ replica1.takeSnapshot();
+ replica2.takeSnapshot();
+ SnapshotPass pass1 = replica1.getLastSnapshotPass();
+ SnapshotPass pass2 = replica2.getLastSnapshotPass();
+ assertTrue(pass1.consistent());
+ assertTrue(pass2.consistent());
+ assertEquals(pass1.numSegments(), 2);
+ assertEquals(pass2.segmentsCrc(), pass1.segmentsCrc());
+
+ // An upload that reached only one replica so far, e.g. from SRT or FIT,
makes the two passes not comparable
+ addSnapshotSegment(replica1, replica1Dir, "seg03");
+ replica1.takeSnapshot();
+ SnapshotPass passWithUpload = replica1.getLastSnapshotPass();
+ assertTrue(passWithUpload.consistent());
+ assertEquals(passWithUpload.numSegments(), 3);
+ assertNotEquals(passWithUpload.segmentsCrc(), pass2.segmentsCrc());
+
+ // A compacted copy uploaded under the same name, applied on one replica
only, also makes them not comparable
+ DummyPartitionUpsertMetadataManager replica3 = new
DummyPartitionUpsertMetadataManager("myTable", 0, upsertContext);
+ File replica3Dir = new File(TEMP_DIR, "replica3");
+ addSnapshotSegment(replica3, replica3Dir, "seg01");
+ ImmutableSegmentImpl compactedSeg02 = addSnapshotSegment(replica3,
replica3Dir, "seg02");
+ when(compactedSeg02.getSegmentMetadata().getDataCrc()).thenReturn(99L);
+ replica3.takeSnapshot();
+ SnapshotPass passWithCompaction = replica3.getLastSnapshotPass();
+ assertTrue(passWithCompaction.consistent());
+ assertEquals(passWithCompaction.numSegments(), 2);
+ assertNotEquals(passWithCompaction.segmentsCrc(), pass2.segmentsCrc());
+ }
+
+ @Test
+ public void testSnapshotPassNotConsistentWhenSegmentOperationOverlaps()
+ throws Exception {
+ DummyPartitionUpsertMetadataManager upsertMetadataManager =
+ new DummyPartitionUpsertMetadataManager("myTable", 0,
createSnapshotPassContext());
+ File dir = new File(TEMP_DIR, "overlap");
+ addSnapshotSegment(upsertMetadataManager, dir, "seg01");
+
+ // An add that starts before the pass and finishes after it changes valid
docs the pass may half capture
+ CountDownLatch addStarted = new CountDownLatch(1);
+ CountDownLatch finishAdd = new CountDownLatch(1);
+ upsertMetadataManager.setOnAddOrReplaceSegment(() -> {
+ addStarted.countDown();
+ try {
+ finishAdd.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ ImmutableSegmentImpl seg02 = createImmutableSegment("seg02", new File(dir,
"seg02"), new ArrayList<>(), null);
+ seg02.enableUpsert(upsertMetadataManager, createDocIds(0, 1), null);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> addFuture = executor.submit(() ->
upsertMetadataManager.addSegment(seg02));
+ assertTrue(addStarted.await(10, TimeUnit.SECONDS));
+ upsertMetadataManager.takeSnapshot();
+ assertFalse(upsertMetadataManager.getLastSnapshotPass().consistent());
+ finishAdd.countDown();
+ addFuture.get(10, TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ }
+
+ // The next pass with nothing overlapping is consistent again
+ upsertMetadataManager.setOnAddOrReplaceSegment(null);
+ upsertMetadataManager.takeSnapshot();
+ SnapshotPass pass = upsertMetadataManager.getLastSnapshotPass();
+ assertTrue(pass.consistent());
+ assertEquals(pass.numSegments(), 2);
+ }
+
+ @Test
+ public void testSnapshotPassNotConsistentWithStaleFileOrUnsealedSegment()
+ throws IOException {
+ DummyPartitionUpsertMetadataManager upsertMetadataManager =
+ new DummyPartitionUpsertMetadataManager("myTable", 0,
createSnapshotPassContext());
+ File dir = new File(TEMP_DIR, "stale");
+ addSnapshotSegment(upsertMetadataManager, dir, "seg01");
+ // seg02 fails to persist its snapshot once, so its file would lag its
valid docs
+ AtomicBoolean failPersist = new AtomicBoolean(true);
+ ImmutableSegmentImpl seg02 = createImmutableSegment("seg02", new File(dir,
"seg02"), new ArrayList<>(), null,
+ () -> {
+ if (failPersist.getAndSet(false)) {
+ throw new IllegalStateException("Failed to persist snapshot");
+ }
+ });
+ seg02.enableUpsert(upsertMetadataManager, createDocIds(0, 1), null);
+ upsertMetadataManager.addSegment(seg02);
+ upsertMetadataManager.takeSnapshot();
+ assertFalse(upsertMetadataManager.getLastSnapshotPass().consistent());
+ upsertMetadataManager.takeSnapshot();
+ assertTrue(upsertMetadataManager.getLastSnapshotPass().consistent());
+
+ // A consuming segment still waiting to be sealed may not match the
committed copy yet
+ upsertMetadataManager.addRecord(mock(MutableSegmentImpl.class),
mock(RecordInfo.class));
+ upsertMetadataManager.takeSnapshot();
+ assertFalse(upsertMetadataManager.getLastSnapshotPass().consistent());
+ }
+
+ private static UpsertContext createSnapshotPassContext() {
+ UpsertContext upsertContext = mock(UpsertContext.class);
+ when(upsertContext.isSnapshotEnabled()).thenReturn(true);
+ // Partial upsert, so takeSnapshot() does not wait for the first consuming
segment
+ when(upsertContext.getPartialUpsertHandlerSupplier()).thenReturn(() ->
mock(PartialUpsertHandler.class));
+ TableDataManager tdm = mock(TableDataManager.class);
+ when(upsertContext.getTableDataManager()).thenReturn(tdm);
+ when(tdm.getSegmentLock(anyString())).thenAnswer(invocation -> new
ReentrantLock());
+ return upsertContext;
+ }
+
+ private static ImmutableSegmentImpl
addSnapshotSegment(DummyPartitionUpsertMetadataManager upsertMetadataManager,
+ File dir, String segmentName)
+ throws IOException {
+ ImmutableSegmentImpl segment =
+ createImmutableSegment(segmentName, new File(dir, segmentName), new
ArrayList<>(), null);
+ segment.enableUpsert(upsertMetadataManager, createDocIds(0, 1), null);
+ upsertMetadataManager.addSegment(segment);
+ return segment;
+ }
+
private static ImmutableSegmentImpl createImmutableSegment(String segName,
File segDir,
List<String> segmentsTakenSnapshot, @Nullable List<String>
queryableDocIdsSegmentsTaken)
throws IOException {
+ return createImmutableSegment(segName, segDir, segmentsTakenSnapshot,
queryableDocIdsSegmentsTaken, () -> {
+ });
+ }
+
+ private static ImmutableSegmentImpl createImmutableSegment(String segName,
File segDir,
+ List<String> segmentsTakenSnapshot, @Nullable List<String>
queryableDocIdsSegmentsTaken, Runnable onPersist)
+ throws IOException {
FileUtils.forceMkdir(segDir);
SegmentMetadataImpl meta = mock(SegmentMetadataImpl.class);
when(meta.getName()).thenReturn(segName);
@@ -1050,6 +1192,7 @@ public class BasePartitionUpsertMetadataManagerTest {
public void persistDocIdsSnapshot(String fileName,
ThreadSafeMutableRoaringBitmap.CardinalityAndBytes docIdsSnapshot)
throws IOException {
+ onPersist.run();
if (V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME.equals(fileName)) {
segmentsTakenSnapshot.add(segName);
} else if
(V1Constants.QUERYABLE_DOC_IDS_SNAPSHOT_FILE_NAME.equals(fileName)
@@ -1077,10 +1220,16 @@ public class BasePartitionUpsertMetadataManagerTest {
private static class DummyPartitionUpsertMetadataManager extends
BasePartitionUpsertMetadataManager {
+ private volatile Runnable _onAddOrReplaceSegment;
+
protected DummyPartitionUpsertMetadataManager(String tableNameWithType,
int partitionId, UpsertContext context) {
super(tableNameWithType, partitionId, context);
}
+ public void setOnAddOrReplaceSegment(@Nullable Runnable
onAddOrReplaceSegment) {
+ _onAddOrReplaceSegment = onAddOrReplaceSegment;
+ }
+
public void trackSegment(IndexSegment seg) {
_trackedSegments.add(seg);
}
@@ -1102,6 +1251,10 @@ public class BasePartitionUpsertMetadataManagerTest {
protected void doAddOrReplaceSegment(ImmutableSegmentImpl segment,
ThreadSafeMutableRoaringBitmap validDocIds,
@Nullable ThreadSafeMutableRoaringBitmap queryableDocIds,
Iterator<RecordInfo> recordInfoIterator,
@Nullable IndexSegment oldSegment, @Nullable MutableRoaringBitmap
validDocIdsForOldSegment) {
+ Runnable onAddOrReplaceSegment = _onAddOrReplaceSegment;
+ if (onAddOrReplaceSegment != null) {
+ onAddOrReplaceSegment.run();
+ }
}
@Override
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 1cd083fe7ce..d4f878aa94f 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,8 @@ 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.PartitionUpsertMetadataManager;
+import
org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager.SnapshotPass;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
@@ -782,6 +784,20 @@ public class TablesResource {
((ImmutableSegment)
segmentDataManager.getSegment()).getSegmentSizeBytes());
}
validDocIdsMetadata.put("segmentCreationTimeMillis",
indexSegment.getSegmentMetadata().getIndexCreationTime());
+ ValidDocIdsType resolvedValidDocIdsType =
validDocIdSnapshotPair.getLeft();
+ if (resolvedValidDocIdsType == ValidDocIdsType.SNAPSHOT
+ || resolvedValidDocIdsType ==
ValidDocIdsType.SNAPSHOT_WITH_DELETE) {
+ // Lets a caller tell whether two replicas' snapshot files of this
partition can be compared
+ PartitionUpsertMetadataManager partitionUpsertMetadataManager =
+ ((ImmutableSegmentImpl)
indexSegment).getPartitionUpsertMetadataManager();
+ SnapshotPass snapshotPass =
+ partitionUpsertMetadataManager != null ?
partitionUpsertMetadataManager.getLastSnapshotPass() : null;
+ if (snapshotPass != null) {
+ validDocIdsMetadata.put("snapshotPass", Map.of("segmentsCrc",
snapshotPass.segmentsCrc(),
+ "numSegments", snapshotPass.numSegments(), "consistent",
snapshotPass.consistent(), "finishedAtMs",
+ snapshotPass.finishedAtMs()));
+ }
+ }
allValidDocIdsMetadata.add(validDocIdsMetadata);
}
if (nonImmutableSegmentCount > 0) {
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 9b65b08225c..82ceea43fdc 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
@@ -49,6 +49,7 @@ import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoa
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.PartitionUpsertMetadataManager;
+import
org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager.SnapshotPass;
import org.apache.pinot.segment.spi.ImmutableSegment;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentMetadata;
@@ -417,12 +418,25 @@ public class TablesResourceTest extends BaseResourceTest {
((ImmutableSegmentImpl) segment).getSegmentSizeBytes());
assertTrue(validDocIdsMetadata.has("segmentCreationTimeMillis"));
assertTrue(validDocIdsMetadata.get("segmentCreationTimeMillis").asLong() >
0);
+ JsonNode snapshotPass = validDocIdsMetadata.get("snapshotPass");
+ assertEquals(snapshotPass.get("segmentsCrc").asLong(), 1234L);
+ assertEquals(snapshotPass.get("numSegments").asInt(), 2);
+ assertTrue(snapshotPass.get("consistent").asBoolean());
+ assertEquals(snapshotPass.get("finishedAtMs").asLong(), 5678L);
// Verify server status information
assertTrue(validDocIdsMetadata.has("serverStatus"), "Server status should
be included in response");
String serverStatus = validDocIdsMetadata.get("serverStatus").asText();
assertNotNull(serverStatus, "Server status should not be null");
assertEquals(serverStatus, "NOT_STARTED", serverStatus);
+
+ // In-memory bitmaps keep changing, so they carry no snapshot pass
+ String inMemoryResponse = _webTarget.path(validDocIdsMetadataPath)
+ .queryParam("segmentNames", segment.getSegmentName())
+ .queryParam("validDocIdsType", ValidDocIdsType.IN_MEMORY.toString())
+ .request()
+ .post(Entity.json(tableSegments), String.class);
+
assertFalse(JsonUtils.stringToJsonNode(inMemoryResponse).get(0).has("snapshotPass"));
}
@Test
@@ -494,6 +508,7 @@ public class TablesResourceTest extends BaseResourceTest {
String snapshotPath = "/segments/" + tableNameWithType + "/" +
segment.getSegmentName() + "/validDocIdsBitmap";
PartitionUpsertMetadataManager upsertMetadataManager =
mock(PartitionUpsertMetadataManager.class);
+ when(upsertMetadataManager.getLastSnapshotPass()).thenReturn(new
SnapshotPass(1234L, 2, true, 5678L));
ThreadSafeMutableRoaringBitmap validDocIds = new
ThreadSafeMutableRoaringBitmap();
ThreadSafeMutableRoaringBitmap queryableDocIds = new
ThreadSafeMutableRoaringBitmap();
ThreadSafeMutableRoaringBitmap validDocIdsSnapshot = new
ThreadSafeMutableRoaringBitmap();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
