You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "KKcorps (via GitHub)" <gi...@apache.org> on 2026/09/23 05:01:31 UTC
[PR] Add opt-in upsert bitmap diagnostics to server metadata API [pinot]
KKcorps opened a new pull request, #19637:
URL: https://github.com/apache/pinot/pull/19637
Replica differences in upsert results are hard to investigate from valid-document counts alone: two replicas can report the same count while selecting different rows, and a count mismatch can reflect different ingestion progress or persisted snapshot files.
This adds opt-in diagnostics to the **server** `POST /tables/{tableNameWithType}/validDocIdsMetadata` endpoint. Pass `includeDiagnostics=true` and query the selected segments on each replica. The existing counts, segment CRCs, bitmap type and server identity remain the context for comparison.
### How the fields help triage
The additional fields are grouped under `diagnostics` in each segment entry:
| Field | Question it helps answer |
| --- | --- |
| `validDocIdsCrc32` | Do equal valid-document counts hide different document IDs? Compare it only for the same segment contents and bitmap type. A mismatch identifies a segment whose bitmap membership needs inspection. |
| `captureStartTimeMs`, `captureEndTimeMs` | When did this server collect the observation, and how long did collection span? Use the window to correlate differences with ingestion, restart, replacement and cleanup logs. These use the server clock. |
| `snapshotFileAgeMs` | Are we comparing persisted files with different write ages? This is the age of the selected local snapshot file at capture completion. It can direct investigation toward snapshot/recovery history, but an old unchanged file can still be valid. Omitted for in-memory reads or when file identity/mtime cannot be established across the read. |
| `consumingSegmentOffsetsBefore`, `consumingSegmentOffsetsAfter` | Was an observed consumer advancing during collection, or were the replicas at different consumption positions? The maps preserve the consumer segment name and its next offset to consume, including overlapping consumers in one partition. |
The offsets are current observations around the API bitmap read, **not the offsets at which a persisted snapshot was created**. They are included for segments with an LLC partition identity; empty maps mean no usable offsets were observed, and absent maps mean the partition could not be identified. Only consumers acquired for this request are observed; matching offsets do not rule out concurrent lifecycle work or an in-progress record update.
### Manual use
```bash
curl -sS -X POST \
"$SERVER/tables/T_REALTIME/validDocIdsMetadata?validDocIdsType=IN_MEMORY&includeDiagnostics=true" \
-H 'Content-Type: application/json' \
-d '{"segments":["SEGMENT_NAME"]}'
```
Query each replica's server admin endpoint. Compare the same segment contents and bitmap type; use `SNAPSHOT` for persisted state and the existing `*_WITH_DELETE` modes for queryable IDs. Download mismatched bitmaps through `validDocIdsBitmap` for further inspection.
The controller aggregate endpoint currently retains one response per segment. This change stays on the server endpoint.
### Semantics and cost
- Count and CRC use the same detached bitmap. CRC32 hashes ascending IDs as four big-endian bytes, independent of Roaring representation. Different CRCs identify different sampled memberships; equal CRCs can collide and say nothing about row payloads or a common replica boundary.
- File age is omitted when identity/size/mtime changes across the read, file identity is unavailable, or mtime is invalid/future. It is not time since the last successful partition snapshot attempt.
- Default behavior is unchanged. Opting in scans selected bitmap IDs and reads local file attributes/consumer offsets. There is no per-record ingestion work, stream RPC, forced snapshot, or ingestion lock.
### Validation
- 14 targeted tests passed: bitmap membership versus cardinality, representation-independent CRCs, fixed encoding, capture windows, same-partition/overlapping consumers, unknown offsets, selected snapshot file age, replacement/missing/future files, all four HTTP bitmap modes, and unchanged default responses.
- Server-module Spotless, license formatting/check and Checkstyle passed.
- Server-module `test-compile` with `-Dmaven.compiler.showDeprecation=true -Dmaven.compiler.showWarnings=true` passed with no warnings on added lines. Dependencies were first built and installed from this checkout. The broader warning-enabled reactor attempt hit the existing `ZstandardDecompressor.java:51` missing `org.jetbrains.annotations.NotNull` dependency; the standard reactor build and targeted tests passed.
Targeted test command:
```bash
./mvnw -pl pinot-server -am \
-Dtest='ValidDocIdsMetadataDiagnosticsTest,TablesResourceTest#testValidDocIdsMetadata*' \
-Dsurefire.failIfNoSpecifiedTests=false \
-Dcheckstyle.skip -Dspotless.check.skip -Dlicense.skip test
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4203804415
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -831,21 +831,26 @@ public void run() {
// 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.
+ ConsumerCoordinator.UnsealedRelease unsealedRelease = _consumerCoordinator.getUnsealedRelease();
+ String consumedUpToOffset =
+ unsealedRelease != null ? unsealedRelease.stoppedAtOffset().toString() : _startOffset.toString();
Review Comment:
`_unsealedRelease` is only cleared by the segment that set it, in its own `doOffload()`. So if N's build or download stalls and N+1 commits sealed (plain `release()`, which sets nothing), N+2 still reads N's record and publishes an offset from two segments back — which reads as "this replica is far behind" when it isn't. That's the exact misreading this field exists to prevent.
Can we only use it when it's actually ahead of where this segment starts? `StreamPartitionMsgOffset` is `Comparable`:
```java
String consumedUpToOffset = unsealedRelease != null
&& unsealedRelease.stoppedAtOffset().compareTo(_startOffset) > 0
? unsealedRelease.stoppedAtOffset().toString() : _startOffset.toString();
```
That also covers the race where the predecessor offloads between acquire and snapshot, which today silently falls back to `_startOffset` and makes the value nondeterministic.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java:
##########
@@ -97,6 +98,10 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps
protected final ServerMetrics _serverMetrics;
protected final Logger _logger;
+ // Scoped to the caller so concurrent snapshot triggers cannot exchange provenance. Preserve the existing virtual
+ // takeSnapshot()/doTakeSnapshot() hooks used by custom managers.
+ private final ThreadLocal<DocIdsSnapshot.Trigger> _snapshotTrigger = new ThreadLocal<>();
Review Comment:
The ThreadLocal returns null without any error if `doTakeSnapshot()` ever runs on a different thread, so the diagnostics would just quietly stop appearing. The constraint you're working around is real (subclasses override `doTakeSnapshot()`), but an overload keeps those working and makes the dataflow explicit:
```java
protected void doTakeSnapshot() { doTakeSnapshot(null); }
protected void doTakeSnapshot(@Nullable DocIdsSnapshot.Trigger trigger) { ... }
```
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ DocIdsType docIdsType, @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
Review Comment:
This works only because `ThreadSafeMutableRoaringBitmap.getBytesAndCardinality()` is `public synchronized`, i.e. locks on `this`. If that class moves to a private lock, the timestamp silently stops being atomic with the capture and nothing fails to compile. Worth putting the combined accessor on `ThreadSafeMutableRoaringBitmap` itself.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ DocIdsType docIdsType, @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), docIdsType, capturedAtMs,
+ trigger != null ? trigger.consumingSegmentName() : null, trigger != null ? trigger.consumedUpToOffset() : null);
+ byte[] json = JsonUtils.objectToString(metadata).getBytes(StandardCharsets.UTF_8);
+ byte[] bytes = ByteBuffer.allocate(snapshot.getBytes().length + 2 * Integer.BYTES + json.length)
Review Comment:
Nit: this copies the whole serialized bitmap a second time — an extra MB-scale array per segment per snapshot round on large partitions. Writing the bitmap and the trailer as two writes in `persistDocIdsSnapshot` would avoid it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4130449122
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), capturedAtMs, trigger != null ? trigger.segmentName() : null,
Review Comment:
Renamed it to `consumedUpToOffset` so it reads as where the snapshot's data ends (exclusive), not a start. Also renamed the CRC to `docIdsCrc` with a `docIdsType` saying valid or queryable, and the segment field to `consumingSegmentName`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-5920767316
Thinking more about this: Does it satisfy the case where we have sorted column? When getting the details and showing the api results, do we ensure that there is no other segment replacements from SRT or FIT invalidating at different pace across replicas?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-5883549060
> `/tables/{tableNameWithType}/validDocIdsMetadata` controller API in `PinotTableRestletResource` is going to pick only one replica per segment and give us the details of the metadata. Are we going to add any other api to detect the differences between the replicas?
Yes Deepthi, that will be the next step
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4125530519
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
Review Comment:
Why do we need `snapshotCapturedAtMs`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add opt-in upsert bitmap diagnostics to server metadata API [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-5789697147
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19637?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `87.50000%` with `9 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 39.60%. Comparing base ([`222b418`](https://app.codecov.io/gh/apache/pinot/commit/222b41877c15dbb993b7f3d1087e522925c93c73?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c060741`](https://app.codecov.io/gh/apache/pinot/commit/c060741b0aa2a02df4fa3c60e9a28ae82a812327?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19637?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../api/resources/ValidDocIdsMetadataDiagnostics.java](https://app.codecov.io/gh/apache/pinot/pull/19637?src=pr&el=tree&filepath=pinot-server%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fserver%2Fapi%2Fresources%2FValidDocIdsMetadataDiagnostics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvYXBpL3Jlc291cmNlcy9WYWxpZERvY0lkc01ldGFkYXRhRGlhZ25vc3RpY3MuamF2YQ==) | 90.74% | [0 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19637?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/pinot/server/api/resources/TablesResource.java](https://app.codecov.io/gh/apache/pinot/pull/19637?src=pr&el=tree&filepath=pinot-server%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fserver%2Fapi%2Fresources%2FTablesResource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvYXBpL3Jlc291cmNlcy9UYWJsZXNSZXNvdXJjZS5qYXZh) | 77.77% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19637?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (222b418) and HEAD (c060741). Click for more details.
>
> <details><summary>HEAD has 21 uploads less than BASE</summary>
>
>| Flag | BASE (222b418) | HEAD (c060741) |
>|------|------|------|
>|unittests1|1|0|
>|java-25|5|1|
>|temurin|5|1|
>|lane-a|2|0|
>|integration|4|0|
>|integration1|2|0|
>|lane-b|2|0|
>|integration2|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19637 +/- ##
=============================================
- Coverage 58.11% 39.60% -18.52%
- Complexity 7 1443 +1436
=============================================
Files 2700 3509 +809
Lines 165942 227337 +61395
Branches 27031 35981 +8950
=============================================
- Hits 96437 90030 -6407
- Misses 61403 129153 +67750
- Partials 8102 8154 +52
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `39.60% <87.50%> (-18.52%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `39.60% <87.50%> (-18.52%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `39.60% <87.50%> (-18.52%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19637/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `39.60% <87.50%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19637?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-6033898386
Thanks for the review Deepthi! All four comments are addressed in d0fa749552, and the description's offset row now matches the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4204605592
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -831,21 +831,26 @@ public void run() {
// 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.
+ ConsumerCoordinator.UnsealedRelease unsealedRelease = _consumerCoordinator.getUnsealedRelease();
+ String consumedUpToOffset =
+ unsealedRelease != null ? unsealedRelease.stoppedAtOffset().toString() : _startOffset.toString();
Review Comment:
Good catch! It now uses the record only when it comes from the segment right before this one, so a stuck older segment falls back to `_startOffset`. I skipped the `> _startOffset` check since it would also hide a replica that fell behind and is downloading, and once the predecessor is offloaded `_startOffset` is already the right value.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java:
##########
@@ -97,6 +98,10 @@ public abstract class BasePartitionUpsertMetadataManager implements PartitionUps
protected final ServerMetrics _serverMetrics;
protected final Logger _logger;
+ // Scoped to the caller so concurrent snapshot triggers cannot exchange provenance. Preserve the existing virtual
+ // takeSnapshot()/doTakeSnapshot() hooks used by custom managers.
+ private final ThreadLocal<DocIdsSnapshot.Trigger> _snapshotTrigger = new ThreadLocal<>();
Review Comment:
Makes sense, switched to the `doTakeSnapshot(@Nullable Trigger)` overload and dropped the ThreadLocal.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ DocIdsType docIdsType, @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
Review Comment:
Done, `getBytesAndCardinality()` now reads the clock under its own lock, so `DocIdsSnapshot` no longer syncs on the bitmap.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ DocIdsType docIdsType, @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), docIdsType, capturedAtMs,
+ trigger != null ? trigger.consumingSegmentName() : null, trigger != null ? trigger.consumedUpToOffset() : null);
+ byte[] json = JsonUtils.objectToString(metadata).getBytes(StandardCharsets.UTF_8);
+ byte[] bytes = ByteBuffer.allocate(snapshot.getBytes().length + 2 * Integer.BYTES + json.length)
Review Comment:
Done, `persistDocIdsSnapshot` now writes the bitmap and then the trailer, so there is no second copy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-6077038534
Documentation follow-up merged: https://github.com/pinot-contrib/pinot-docs/pull/1094 — snapshot diagnostic fields and replica-comparison caveats.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4130449122
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), capturedAtMs, trigger != null ? trigger.segmentName() : null,
Review Comment:
Renamed it to `consumedUpToOffset` so it reads as where the snapshot's data ends (exclusive), not a start. Also renamed the CRC to `docIdsCrc32` since it covers queryable doc ids too, and the segment field to `consumingSegmentName`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4129573204
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
Review Comment:
This is mostly to determine how stale the valid doc id snapshot is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-5924379802
> Thinking more about this: Does it satisfy the case where we have sorted column? When getting the details and showing the api results, do we ensure that there is no other segment replacements from SRT or FIT invalidating at different pace across replicas?
So the details added in this PR mostly make sense when using on valid doc id snapshots and not on live bitmaps.
Valid doc id snapshots should be consistent across replicas if taken at single point in time.
I am planning to use this crc introduced here along with new data crc here https://github.com/startreedata/startree-pinot/pull/4597/changes to check for divergence
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4193915414
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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, the first offset that consumer
+ /// reads, and is not a partition 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);
Review Comment:
Good catch, it was the consuming segment's start offset so every replica showed the same value. Now if the previous segment was still unsealed at snapshot time (parallel build or download), it records where this replica's own consumer stopped, so a replica that ran ahead or fell behind shows its real offset.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps merged PR #19637:
URL: https://github.com/apache/pinot/pull/19637
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4129589577
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), capturedAtMs, trigger != null ? trigger.segmentName() : null,
Review Comment:
This is actually end offset only. It's the start offset of current consuming segment at the beginning of which we are taking snapshot
let me rename this variable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4159664228
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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.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, the first offset that consumer
+ /// reads, and is not a partition 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);
Review Comment:
The offset gives the same value across all the replicas. Can we somehow capture the actual offset until which each replica consumed as well? Rest of the changes looks good to me!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on code in PR #19637:
URL: https://github.com/apache/pinot/pull/19637#discussion_r4125438791
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/DocIdsSnapshot.java:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.IOException;
+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;
+
+ /// This identifies the consumer startup that triggered a snapshot, not a partition consistency watermark.
+ public record Trigger(String segmentName, String startOffset) {
+ }
+
+ public record Metadata(long validDocIdsCrc32, long snapshotCapturedAtMs,
+ @Nullable String snapshotTriggerSegmentName, @Nullable String snapshotTriggerStartOffset) {
+ public Map<String, Object> toResponse(long nowMs) {
+ Map<String, Object> response = new HashMap<>();
+ response.put("validDocIdsCrc32", validDocIdsCrc32);
+ response.put("snapshotCapturedAtMs", snapshotCapturedAtMs);
+ // Do not turn clock skew into an apparently fresh snapshot.
+ if (nowMs >= snapshotCapturedAtMs) {
+ response.put("snapshotAgeMs", nowMs - snapshotCapturedAtMs);
+ }
+ if (snapshotTriggerSegmentName != null && snapshotTriggerStartOffset != null) {
+ response.put("snapshotTriggerSegmentName", snapshotTriggerSegmentName);
+ response.put("snapshotTriggerStartOffset", snapshotTriggerStartOffset);
+ }
+ return response;
+ }
+ }
+
+ /// Capture the timestamp with the bitmap under its existing lock, then hash the detached bitmap outside that lock.
+ public static ThreadSafeMutableRoaringBitmap.CardinalityAndBytes capture(ThreadSafeMutableRoaringBitmap bitmap,
+ @Nullable Trigger trigger)
+ throws IOException {
+ ThreadSafeMutableRoaringBitmap.CardinalityAndBytes snapshot;
+ long capturedAtMs;
+ synchronized (bitmap) {
+ snapshot = bitmap.getBytesAndCardinality();
+ capturedAtMs = System.currentTimeMillis();
+ }
+ ImmutableRoaringBitmap captured = new ImmutableRoaringBitmap(ByteBuffer.wrap(snapshot.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);
+ }
+ Metadata metadata = new Metadata(crc.getValue(), capturedAtMs, trigger != null ? trigger.segmentName() : null,
Review Comment:
Should we also capture end offset?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Persist upsert snapshot diagnostics for replica divergence triage [pinot]
Posted by "deepthi912 (via GitHub)" <gi...@apache.org>.
deepthi912 commented on PR #19637:
URL: https://github.com/apache/pinot/pull/19637#issuecomment-5876114404
`/tables/{tableNameWithType}/validDocIdsMetadata` controller API in `PinotTableRestletResource` is going to pick only one replica per segment and give us the details of the metadata. Are we going to add any other api to detect the differences between the replicas?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org