This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 de2f596815f Add skipUpsertDelete query option to view valid docs
instead of queryable docs (#19009)
de2f596815f is described below
commit de2f596815f69b6a1ae17e01f898fdab0f20814d
Author: Chaitanya Deepthi <[email protected]>
AuthorDate: Fri Jul 17 23:43:26 2026 -0700
Add skipUpsertDelete query option to view valid docs instead of queryable
docs (#19009)
---
.../common/utils/config/QueryOptionsUtils.java | 4 +
.../org/apache/pinot/core/plan/FilterPlanNode.java | 10 +-
.../core/query/pruner/SegmentPrunerService.java | 19 +--
.../apache/pinot/core/plan/FilterPlanNodeTest.java | 2 +-
...adataAndDictionaryAggregationPlanMakerTest.java | 2 +-
.../query/pruner/SegmentPrunerServiceTest.java | 36 ++++++
.../immutable/ImmutableSegmentImpl.java | 6 +
.../indexsegment/mutable/MutableSegmentImpl.java | 6 +
.../ConcurrentMapTableUpsertMetadataManager.java | 12 +-
.../pinot/segment/local/upsert/UpsertUtils.java | 30 +++++
.../segment/local/upsert/UpsertViewManager.java | 74 ++++++++----
.../BasePartitionUpsertMetadataManagerTest.java | 40 +++----
...oncurrentMapTableUpsertMetadataManagerTest.java | 130 +++++++++++++++++++++
.../segment/local/upsert/UpsertUtilsTest.java | 105 +++++++++++++++++
.../local/upsert/UpsertViewManagerTest.java | 73 +++++++++++-
.../org/apache/pinot/segment/spi/IndexSegment.java | 5 +
.../apache/pinot/segment/spi/SegmentContext.java | 20 +++-
.../apache/pinot/spi/utils/CommonConstants.java | 3 +
18 files changed, 514 insertions(+), 63 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index 9702380739f..4a39ab5d8e7 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -165,6 +165,10 @@ public class QueryOptionsUtils {
return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_UPSERT_VIEW));
}
+ public static boolean isSkipUpsertDelete(Map<String, String> queryOptions) {
+ return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_UPSERT_DELETE));
+ }
+
public static boolean isTraceRuleProductions(Map<String, String>
queryOptions) {
return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.TRACE_RULE_PRODUCTIONS));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
index b69d77bcd2f..0a70e0e6008 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/FilterPlanNode.java
@@ -97,20 +97,20 @@ public class FilterPlanNode implements PlanNode {
@Override
public BaseFilterOperator run() {
- MutableRoaringBitmap queryableDocIdsSnapshot =
_segmentContext.getQueryableDocIdsSnapshot();
+ MutableRoaringBitmap docIdsSnapshot = _segmentContext.getDocIdsSnapshot();
int numDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
if (_filter != null) {
BaseFilterOperator filterOperator = constructPhysicalOperator(_filter,
numDocs);
- if (queryableDocIdsSnapshot != null) {
- BaseFilterOperator validDocFilter = new
BitmapBasedFilterOperator(queryableDocIdsSnapshot, false, numDocs);
+ if (docIdsSnapshot != null) {
+ BaseFilterOperator validDocFilter = new
BitmapBasedFilterOperator(docIdsSnapshot, false, numDocs);
return FilterOperatorUtils.getAndFilterOperator(_queryContext,
Arrays.asList(filterOperator, validDocFilter),
numDocs);
} else {
return filterOperator;
}
- } else if (queryableDocIdsSnapshot != null) {
- return new BitmapBasedFilterOperator(queryableDocIdsSnapshot, false,
numDocs);
+ } else if (docIdsSnapshot != null) {
+ return new BitmapBasedFilterOperator(docIdsSnapshot, false, numDocs);
} else {
return new MatchAllFilterOperator(numDocs);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/SegmentPrunerService.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/SegmentPrunerService.java
index 07aa5601e50..48af9b7395f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/SegmentPrunerService.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/pruner/SegmentPrunerService.java
@@ -136,26 +136,31 @@ public class SegmentPrunerService {
* @param segments the list of segments to be pruned. This is a destructive
operation that may modify this list in an
* undefined way. Therefore, this list should not be used
after calling this method.
* @param query query context; when non-null and skipUpsert=true,
segments with 0 queryable/valid docs are not
- * treated as empty (they contribute replaced rows to the
result). When queryable doc ids exist,
- * emptiness is determined from them; otherwise {@link
IndexSegment#getValidDocIds()} is used.
+ * treated as empty (they contribute replaced rows to the
result). When skipUpsertDelete=true,
+ * emptiness is determined from valid docs (tombstones count
as non-empty); otherwise from
+ * queryable docs.
* @return the new list with filtered elements. This is the list that have
to be used.
*/
private static List<IndexSegment> removeEmptySegments(List<IndexSegment>
segments, QueryContext query) {
int selected = 0;
- boolean skipUpsert =
QueryOptionsUtils.isSkipUpsert(query.getQueryOptions());
+ Map<String, String> queryOptions = query.getQueryOptions();
+ boolean skipUpsert = QueryOptionsUtils.isSkipUpsert(queryOptions);
+ boolean skipUpsertDelete =
QueryOptionsUtils.isSkipUpsertDelete(queryOptions);
for (IndexSegment segment : segments) {
- if (!isEmptySegment(segment, skipUpsert)) {
+ if (!isEmptySegment(segment, skipUpsert, skipUpsertDelete)) {
segments.set(selected++, segment);
}
}
return segments.subList(0, selected);
}
- private static boolean isEmptySegment(IndexSegment segment, boolean
skipUpsert) {
+ private static boolean isEmptySegment(IndexSegment segment, boolean
skipUpsert, boolean skipUpsertDelete) {
if (segment.getSegmentMetadata().getTotalDocs() == 0) {
return true;
}
- // Check if the segment has 0 queryable docIds while skipUpsert=false
- return !skipUpsert && segment.hasNoQueryableDocs();
+ if (skipUpsert) {
+ return false;
+ }
+ return skipUpsertDelete ? segment.hasNoValidDocs() :
segment.hasNoQueryableDocs();
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/plan/FilterPlanNodeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/plan/FilterPlanNodeTest.java
index 3faced128bb..b505c2e8dfb 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/plan/FilterPlanNodeTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/plan/FilterPlanNodeTest.java
@@ -94,7 +94,7 @@ public class FilterPlanNodeTest {
// Result should be invariant - always exactly 3 docs
for (int i = 0; i < 10_000; i++) {
SegmentContext segmentContext = new SegmentContext(segment);
-
segmentContext.setQueryableDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment));
+
segmentContext.setDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment));
assertEquals(getNumberOfFilteredDocs(segmentContext, queryContext), 3);
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
index 672f08a6fbd..769bd57b9d6 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/MetadataAndDictionaryAggregationPlanMakerTest.java
@@ -165,7 +165,7 @@ public class MetadataAndDictionaryAggregationPlanMakerTest {
assertTrue(operatorClass.isInstance(operator));
SegmentContext segmentContext = new SegmentContext(_upsertIndexSegment);
-
segmentContext.setQueryableDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(_upsertIndexSegment));
+
segmentContext.setDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(_upsertIndexSegment));
Operator<?> upsertOperator =
PLAN_MAKER.makeSegmentPlanNode(segmentContext, queryContext).run();
assertTrue(upsertOperatorClass.isInstance(upsertOperator));
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/SegmentPrunerServiceTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/SegmentPrunerServiceTest.java
index f1ae4506907..7f5c6fa4a1a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/SegmentPrunerServiceTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/pruner/SegmentPrunerServiceTest.java
@@ -186,6 +186,42 @@ public class SegmentPrunerServiceTest {
Assert.assertEquals(actual, segments);
}
+ @Test
+ public void emptyQueryableRetainedWithSkipUpsertDelete() {
+ SegmentPrunerService service = new SegmentPrunerService(_emptyPrunerConf);
+ ThreadSafeMutableRoaringBitmap valid = new
ThreadSafeMutableRoaringBitmap(0);
+ ThreadSafeMutableRoaringBitmap queryable = new
ThreadSafeMutableRoaringBitmap();
+ IndexSegment segment = mockUpsertIndexSegment(10, valid, queryable);
+
+ List<IndexSegment> segments = new ArrayList<>();
+ segments.add(segment);
+ QueryContext queryContext =
+ QueryContextConverterUtils.getQueryContext("select col1 from t1
option(skipUpsertDelete=true)");
+
+ List<IndexSegment> actual = service.prune(segments, queryContext, new
SegmentPrunerStatistics());
+
+ Assert.assertEquals(actual, segments);
+ }
+
+ /**
+ * skipUpsertDelete checks valid-docs emptiness specifically, not
skipUpsert's blanket bypass: a segment fully
+ * superseded elsewhere (0 valid docs) is still pruned.
+ */
+ @Test
+ public void emptyValidPrunedWithSkipUpsertDelete() {
+ SegmentPrunerService service = new SegmentPrunerService(_emptyPrunerConf);
+ IndexSegment segment = mockUpsertIndexSegment(10, new
ThreadSafeMutableRoaringBitmap(), null);
+
+ List<IndexSegment> segments = new ArrayList<>();
+ segments.add(segment);
+ QueryContext queryContext =
+ QueryContextConverterUtils.getQueryContext("select col1 from t1
option(skipUpsertDelete=true)");
+
+ List<IndexSegment> actual = service.prune(segments, queryContext, new
SegmentPrunerStatistics());
+
+ Assert.assertEquals(actual, List.of());
+ }
+
@Test
public void nonEmptyQueryableNotPruned() {
SegmentPrunerService service = new SegmentPrunerService(_emptyPrunerConf);
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 9436a49ee35..c0a22c85fcd 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
@@ -40,6 +40,7 @@ import
org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
import
org.apache.pinot.segment.local.segment.virtualcolumn.VirtualColumnContext;
import org.apache.pinot.segment.local.startree.v2.store.StarTreeIndexContainer;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
+import org.apache.pinot.segment.local.upsert.UpsertUtils;
import org.apache.pinot.segment.local.upsert.UpsertViewManager;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.FetchContext;
@@ -382,6 +383,11 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
return validDocIds != null && validDocIds.isEmpty();
}
+ @Override
+ public boolean hasNoValidDocs() {
+ return UpsertUtils.hasNoValidDocs(_partitionUpsertMetadataManager, this);
+ }
+
@Override
public GenericRow getRecord(int docId, GenericRow reuse) {
try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader()) {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index dbf8f935311..a8357d5779b 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -74,6 +74,7 @@ import
org.apache.pinot.segment.local.upsert.ComparisonColumns;
import org.apache.pinot.segment.local.upsert.PartitionUpsertMetadataManager;
import org.apache.pinot.segment.local.upsert.RecordInfo;
import org.apache.pinot.segment.local.upsert.UpsertContext;
+import org.apache.pinot.segment.local.upsert.UpsertUtils;
import org.apache.pinot.segment.local.upsert.UpsertViewManager;
import org.apache.pinot.segment.local.utils.FixedIntArrayOffHeapIdMap;
import org.apache.pinot.segment.local.utils.IdMap;
@@ -1242,6 +1243,11 @@ public class MutableSegmentImpl implements
MutableSegment {
return validDocIds != null && validDocIds.isEmpty();
}
+ @Override
+ public boolean hasNoValidDocs() {
+ return UpsertUtils.hasNoValidDocs(_partitionUpsertMetadataManager, this);
+ }
+
@Override
public GenericRow getRecord(int docId, GenericRow reuse) {
try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader()) {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManager.java
index 9f67a5d3d17..628104273a8 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManager.java
@@ -95,24 +95,28 @@ public class ConcurrentMapTableUpsertMetadataManager
extends BaseTableUpsertMeta
// Otherwise, get queryableDocIds bitmaps as kept by the segment objects
directly as before.
if (_context.getConsistencyMode() == UpsertConfig.ConsistencyMode.NONE ||
QueryOptionsUtils.isSkipUpsertView(
queryOptions)) {
+ // No shared upsert-view lock exists in this branch, so a direct read is
already safe here.
+ boolean skipUpsertDelete =
QueryOptionsUtils.isSkipUpsertDelete(queryOptions);
for (SegmentContext segmentContext : segmentContexts) {
IndexSegment segment = segmentContext.getIndexSegment();
-
segmentContext.setQueryableDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment));
+ segmentContext.setDocIdsSnapshot(skipUpsertDelete
+ ? UpsertUtils.getValidDocIdsSnapshotFromSegment(segment)
+ : UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment));
}
return;
}
- // All segments should have been tracked by partitionMetadataManagers to
provide queries consistent upsert view.
+ // A consistency mode is active: UpsertViewManager knows the locking each
mode requires for skipUpsertDelete too.
_partitionMetadataManagerMap.forEach(
(partitionID, upsertMetadataManager) ->
upsertMetadataManager.getUpsertViewManager()
.setSegmentContexts(segmentContexts, queryOptions));
if (LOGGER.isDebugEnabled()) {
for (SegmentContext segmentContext : segmentContexts) {
IndexSegment segment = segmentContext.getIndexSegment();
- if (segmentContext.getQueryableDocIdsSnapshot() == null) {
+ if (segmentContext.getDocIdsSnapshot() == null) {
LOGGER.debug("No upsert view for segment: {}, type: {}, total: {}",
segment.getSegmentName(),
(segment instanceof ImmutableSegment ? "imm" : "mut"),
segment.getSegmentMetadata().getTotalDocs());
} else {
- int cardCnt =
segmentContext.getQueryableDocIdsSnapshot().getCardinality();
+ int cardCnt = segmentContext.getDocIdsSnapshot().getCardinality();
LOGGER.debug("Got upsert view of segment: {}, type: {}, total: {},
valid: {}", segment.getSegmentName(),
(segment instanceof ImmutableSegment ? "imm" : "mut"),
segment.getSegmentMetadata().getTotalDocs(),
cardCnt);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertUtils.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertUtils.java
index 3978044b878..f8af9bf12a6 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertUtils.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertUtils.java
@@ -57,6 +57,36 @@ public class UpsertUtils {
: (useEmptyForNull ? new MutableRoaringBitmap() : null);
}
+ /// Unlike [#getQueryableDocIdsSnapshotFromSegment], never falls back to
preferring queryable docs.
+ @Nullable
+ public static MutableRoaringBitmap
getValidDocIdsSnapshotFromSegment(IndexSegment segment) {
+ return getValidDocIdsSnapshotFromSegment(segment, false);
+ }
+
+ /// Shared by {@code ImmutableSegmentImpl}/{@code MutableSegmentImpl}'s
`hasNoValidDocs()`. Mirrors
+ /// `hasNoQueryableDocs()`'s consistency-mode-aware/live-fallback split,
against the valid-docs cache instead.
+ public static boolean hasNoValidDocs(@Nullable
PartitionUpsertMetadataManager partitionUpsertMetadataManager,
+ IndexSegment segment) {
+ if (partitionUpsertMetadataManager == null) {
+ return false;
+ }
+ UpsertViewManager viewManager =
partitionUpsertMetadataManager.getUpsertViewManager();
+ if (viewManager != null) {
+ MutableRoaringBitmap validDocIdsSnapshot =
viewManager.getValidDocIdsSnapshot(segment);
+ return validDocIdsSnapshot != null && validDocIdsSnapshot.isEmpty();
+ }
+ ThreadSafeMutableRoaringBitmap validDocIds = segment.getValidDocIds();
+ return validDocIds != null && validDocIds.isEmpty();
+ }
+
+ @Nullable
+ public static MutableRoaringBitmap
getValidDocIdsSnapshotFromSegment(IndexSegment segment,
+ boolean useEmptyForNull) {
+ ThreadSafeMutableRoaringBitmap validDocIds = segment.getValidDocIds();
+ return validDocIds != null ? validDocIds.getMutableRoaringBitmap()
+ : (useEmptyForNull ? new MutableRoaringBitmap() : null);
+ }
+
public static void doReplaceDocId(ThreadSafeMutableRoaringBitmap validDocIds,
@Nullable ThreadSafeMutableRoaringBitmap queryableDocIds, int oldDocId,
int newDocId, RecordInfo recordInfo) {
validDocIds.replace(oldDocId, newDocId);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertViewManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertViewManager.java
index 9b9d52ce2c2..b1df77bcfde 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertViewManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/UpsertViewManager.java
@@ -65,7 +65,12 @@ public class UpsertViewManager {
// The query threads always get _upsertViewTrackedSegmentsLock then
_upsertViewSegmentDocIdsLock to avoid deadlock.
// And the upsert threads never nest the two locks.
private final ReadWriteLock _upsertViewLock = new ReentrantReadWriteLock();
+ // When a segment has no delete column, the same MutableRoaringBitmap
instance is stored in both maps below
+ // (see doBatchRefreshUpsertView) to avoid cloning it twice; never mutate a
bitmap read from either map in place.
private volatile Map<IndexSegment, MutableRoaringBitmap>
_segmentQueryableDocIdsMap;
+ // Refreshed alongside _segmentQueryableDocIdsMap in the same locked pass,
so skipUpsertDelete queries and pruning
+ // can read a consistent cut lock-free instead of taking the write lock per
call.
+ private volatile Map<IndexSegment, MutableRoaringBitmap>
_segmentValidDocIdsMap;
// For SNAPSHOT mode, track segments that get new updates since last refresh
to reduce the overhead of refreshing.
private final Set<IndexSegment> _updatedSegmentsSinceLastRefresh =
ConcurrentHashMap.newKeySet();
@@ -164,10 +169,11 @@ public class UpsertViewManager {
* present, to avoid overwriting the contexts specified at the others places.
*/
public void setSegmentContexts(List<SegmentContext> segmentContexts,
Map<String, String> queryOptions) {
+ boolean skipUpsertDelete =
QueryOptionsUtils.isSkipUpsertDelete(queryOptions);
if (_consistencyMode == UpsertConfig.ConsistencyMode.SYNC) {
_upsertViewLock.readLock().lock();
try {
- setSegmentContexts(segmentContexts);
+ setSegmentContexts(segmentContexts, skipUpsertDelete);
return;
} finally {
_upsertViewLock.readLock().unlock();
@@ -184,21 +190,24 @@ public class UpsertViewManager {
upsertViewFreshnessMs = _upsertViewRefreshIntervalMs;
}
doBatchRefreshUpsertView(upsertViewFreshnessMs, false);
- Map<IndexSegment, MutableRoaringBitmap> currentUpsertView =
_segmentQueryableDocIdsMap;
+ Map<IndexSegment, MutableRoaringBitmap> currentUpsertView =
+ skipUpsertDelete ? _segmentValidDocIdsMap : _segmentQueryableDocIdsMap;
for (SegmentContext segmentContext : segmentContexts) {
IndexSegment segment = segmentContext.getIndexSegment();
MutableRoaringBitmap segmentView = currentUpsertView.get(segment);
if (segmentView != null) {
- segmentContext.setQueryableDocIdsSnapshot(segmentView);
+ segmentContext.setDocIdsSnapshot(segmentView);
}
}
}
- private void setSegmentContexts(List<SegmentContext> segmentContexts) {
+ private void setSegmentContexts(List<SegmentContext> segmentContexts,
boolean skipUpsertDelete) {
for (SegmentContext segmentContext : segmentContexts) {
IndexSegment segment = segmentContext.getIndexSegment();
if (_trackedSegments.contains(segment)) {
-
segmentContext.setQueryableDocIdsSnapshot(UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment,
true));
+ segmentContext.setDocIdsSnapshot(skipUpsertDelete
+ ? UpsertUtils.getValidDocIdsSnapshotFromSegment(segment, true)
+ : UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment,
true));
}
}
}
@@ -212,52 +221,65 @@ public class UpsertViewManager {
@VisibleForTesting
void doBatchRefreshUpsertView(long upsertViewFreshnessMs, boolean
forceRefresh) {
- // Always refresh if the current view is still empty.
- if (!forceRefresh && skipUpsertViewRefresh(upsertViewFreshnessMs) &&
_segmentQueryableDocIdsMap != null) {
+ // Always refresh if either view is still empty. The two maps are swapped
in via two separate volatile writes,
+ // not one atomic publish, so a concurrent reader can observe
_segmentValidDocIdsMap up to one refresh behind
+ // _segmentQueryableDocIdsMap; this is self-correcting next cycle and no
caller combines both maps at once.
+ if (!forceRefresh && skipUpsertViewRefresh(upsertViewFreshnessMs) &&
_segmentQueryableDocIdsMap != null
+ && _segmentValidDocIdsMap != null) {
return;
}
_upsertViewLock.writeLock().lock();
try {
- // Check again with lock, and always refresh if the current view is
still empty.
- Map<IndexSegment, MutableRoaringBitmap> current =
_segmentQueryableDocIdsMap;
- if (!forceRefresh && skipUpsertViewRefresh(upsertViewFreshnessMs) &&
current != null) {
+ // Check again with lock, and always refresh if either view is still
empty.
+ Map<IndexSegment, MutableRoaringBitmap> queryableDocIds =
_segmentQueryableDocIdsMap;
+ Map<IndexSegment, MutableRoaringBitmap> validDocs =
_segmentValidDocIdsMap;
+ if (!forceRefresh && skipUpsertViewRefresh(upsertViewFreshnessMs) &&
queryableDocIds != null
+ && validDocs != null) {
return;
}
if (LOGGER.isDebugEnabled()) {
- if (current == null) {
+ if (queryableDocIds == null) {
LOGGER.debug("Current upsert view is still null");
} else {
- current.forEach(
+ queryableDocIds.forEach(
(segment, bitmap) -> LOGGER.debug("Current upsert view of
segment: {}, type: {}, total: {}, valid: {}",
segment.getSegmentName(), (segment instanceof
ImmutableSegment ? "imm" : "mut"),
segment.getSegmentMetadata().getTotalDocs(),
bitmap.getCardinality()));
}
}
- Map<IndexSegment, MutableRoaringBitmap> updated = new HashMap<>();
+ Map<IndexSegment, MutableRoaringBitmap> updatedQueryable = new
HashMap<>();
+ Map<IndexSegment, MutableRoaringBitmap> updatedValid = new HashMap<>();
for (IndexSegment segment : _trackedSegments) {
// Update bitmap for segment updated since last refresh or not in the
view yet. This also handles segments
// that are tracked by _trackedSegments but not by
_updatedSegmentsSinceLastRefresh, like those didn't update
// any bitmaps as their docs simply lost all the upsert comparisons
with the existing docs.
- if (current == null || current.get(segment) == null ||
_updatedSegmentsSinceLastRefresh.contains(segment)) {
- updated.put(segment,
UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment, true));
+ if (queryableDocIds == null || queryableDocIds.get(segment) == null
+ || _updatedSegmentsSinceLastRefresh.contains(segment)) {
+ MutableRoaringBitmap validSnapshot =
UpsertUtils.getValidDocIdsSnapshotFromSegment(segment, true);
+ updatedValid.put(segment, validSnapshot);
+ // Without delete enabled, queryable docs equal valid docs; reuse
the clone instead of cloning twice.
+ updatedQueryable.put(segment, segment.getQueryableDocIds() != null
+ ? UpsertUtils.getQueryableDocIdsSnapshotFromSegment(segment,
true) : validSnapshot);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Update upsert view of segment: {}, type: {}, total:
{}, valid: {}, reason: {}",
segment.getSegmentName(), (segment instanceof ImmutableSegment
? "imm" : "mut"),
- segment.getSegmentMetadata().getTotalDocs(),
updated.get(segment).getCardinality(),
- current == null || current.get(segment) == null ? "no view
yet" : "bitmap updated");
+ segment.getSegmentMetadata().getTotalDocs(),
updatedQueryable.get(segment).getCardinality(),
+ queryableDocIds == null || queryableDocIds.get(segment) ==
null ? "no view yet" : "bitmap updated");
}
} else {
- updated.put(segment, current.get(segment));
+ updatedQueryable.put(segment, queryableDocIds.get(segment));
+ updatedValid.put(segment, validDocs.get(segment));
}
}
// Swap in the new consistent set of bitmaps.
if (LOGGER.isDebugEnabled()) {
- updated.forEach(
+ updatedQueryable.forEach(
(segment, bitmap) -> LOGGER.debug("Updated upsert view of segment:
{}, type: {}, total: {}, valid: {}",
segment.getSegmentName(), (segment instanceof ImmutableSegment
? "imm" : "mut"),
segment.getSegmentMetadata().getTotalDocs(),
bitmap.getCardinality()));
}
- _segmentQueryableDocIdsMap = updated;
+ _segmentQueryableDocIdsMap = updatedQueryable;
+ _segmentValidDocIdsMap = updatedValid;
_updatedSegmentsSinceLastRefresh.clear();
_lastUpsertViewRefreshTimeMs = System.currentTimeMillis();
} finally {
@@ -307,6 +329,11 @@ public class UpsertViewManager {
return _segmentQueryableDocIdsMap;
}
+ @VisibleForTesting
+ Map<IndexSegment, MutableRoaringBitmap> getSegmentValidDocIdsMap() {
+ return _segmentValidDocIdsMap;
+ }
+
// Returns the queryable doc-id snapshot for the given segment from the most
recent refresh, or
// null if no refresh has happened yet or the segment is not in the current
view.
@Nullable
@@ -315,6 +342,13 @@ public class UpsertViewManager {
return view == null ? null : view.get(segment);
}
+ // Same as getQueryableDocIdsSnapshot, but from the valid-docs (tombstones
included) cache.
+ @Nullable
+ public MutableRoaringBitmap getValidDocIdsSnapshot(IndexSegment segment) {
+ Map<IndexSegment, MutableRoaringBitmap> view = _segmentValidDocIdsMap;
+ return view == null ? null : view.get(segment);
+ }
+
@VisibleForTesting
Set<IndexSegment> getUpdatedSegmentsSinceLastRefresh() {
return _updatedSegmentsSinceLastRefresh;
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 aefd1cfc895..c448fcb5c94 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
@@ -703,15 +703,15 @@ public class BasePartitionUpsertMetadataManagerTest {
ThreadSafeMutableRoaringBitmap validDocIds =
segmentQueryableDocIdsMap.get(sc.getIndexSegment());
assertNotNull(validDocIds);
// SegmentContext holds a clone of the original queryableDocIds held by
the segment object.
- assertNotSame(sc.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
- assertEquals(sc.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertNotSame(sc.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertEquals(sc.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
// docId=0 in seg01 got invalidated.
if (sc.getIndexSegment() == seg01) {
- assertFalse(sc.getQueryableDocIdsSnapshot().contains(0));
+ assertFalse(sc.getDocIdsSnapshot().contains(0));
}
// docId=12 in seg03 was newly added.
if (sc.getIndexSegment() == seg03) {
- assertTrue(sc.getQueryableDocIdsSnapshot().contains(12));
+ assertTrue(sc.getDocIdsSnapshot().contains(12));
}
}
}
@@ -796,21 +796,21 @@ public class BasePartitionUpsertMetadataManagerTest {
for (SegmentContext reuseSC : reuseSegmentContexts) {
for (SegmentContext sc : segmentContexts) {
if (reuseSC.getIndexSegment() == sc.getIndexSegment()) {
- assertSame(reuseSC.getQueryableDocIdsSnapshot(),
sc.getQueryableDocIdsSnapshot());
+ assertSame(reuseSC.getDocIdsSnapshot(), sc.getDocIdsSnapshot());
}
}
ThreadSafeMutableRoaringBitmap validDocIds =
segmentQueryableDocIdsMap.get(reuseSC.getIndexSegment());
assertNotNull(validDocIds);
// The upsert view holds a clone of the original queryableDocIds held by
the segment object.
- assertNotSame(reuseSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
- assertEquals(reuseSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertNotSame(reuseSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertEquals(reuseSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
// docId=0 in seg01 got invalidated.
if (reuseSC.getIndexSegment() == seg01) {
- assertFalse(reuseSC.getQueryableDocIdsSnapshot().contains(0));
+ assertFalse(reuseSC.getDocIdsSnapshot().contains(0));
}
// docId=12 in seg03 was newly added.
if (reuseSC.getIndexSegment() == seg03) {
- assertTrue(reuseSC.getQueryableDocIdsSnapshot().contains(12));
+ assertTrue(reuseSC.getDocIdsSnapshot().contains(12));
}
}
@@ -825,21 +825,21 @@ public class BasePartitionUpsertMetadataManagerTest {
for (SegmentContext refreshSC : refreshSegmentContexts) {
for (SegmentContext sc : segmentContexts) {
if (refreshSC.getIndexSegment() == sc.getIndexSegment()) {
- assertNotSame(refreshSC.getQueryableDocIdsSnapshot(),
sc.getQueryableDocIdsSnapshot());
+ assertNotSame(refreshSC.getDocIdsSnapshot(), sc.getDocIdsSnapshot());
}
}
ThreadSafeMutableRoaringBitmap validDocIds =
segmentQueryableDocIdsMap.get(refreshSC.getIndexSegment());
assertNotNull(validDocIds);
// The upsert view holds a clone of the original queryableDocIds held by
the segment object.
- assertNotSame(refreshSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
- assertEquals(refreshSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertNotSame(refreshSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertEquals(refreshSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
// docId=0 in seg01 got invalidated.
if (refreshSC.getIndexSegment() == seg01) {
- assertFalse(refreshSC.getQueryableDocIdsSnapshot().contains(0));
+ assertFalse(refreshSC.getDocIdsSnapshot().contains(0));
}
// docId=12 in seg03 was newly added.
if (refreshSC.getIndexSegment() == seg03) {
- assertTrue(refreshSC.getQueryableDocIdsSnapshot().contains(12));
+ assertTrue(refreshSC.getDocIdsSnapshot().contains(12));
}
}
}
@@ -896,22 +896,22 @@ public class BasePartitionUpsertMetadataManagerTest {
for (SegmentContext refreshSC : refreshSegmentContexts) {
for (SegmentContext sc : segmentContexts) {
if (refreshSC.getIndexSegment() == sc.getIndexSegment()) {
- assertNotSame(refreshSC.getQueryableDocIdsSnapshot(),
sc.getQueryableDocIdsSnapshot());
+ assertNotSame(refreshSC.getDocIdsSnapshot(), sc.getDocIdsSnapshot());
}
}
ThreadSafeMutableRoaringBitmap validDocIds =
segmentQueryableDocIdsMap.get(refreshSC.getIndexSegment());
assertNotNull(validDocIds);
// The upsert view holds a clone of the original queryableDocIds held by
the segment object.
- assertNotSame(refreshSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
- assertEquals(refreshSC.getQueryableDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
- assertNotNull(refreshSC.getQueryableDocIdsSnapshot());
+ assertNotSame(refreshSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertEquals(refreshSC.getDocIdsSnapshot(),
validDocIds.getMutableRoaringBitmap());
+ assertNotNull(refreshSC.getDocIdsSnapshot());
// docId=0 in seg01 got invalidated.
if (refreshSC.getIndexSegment() == seg01) {
- assertFalse(refreshSC.getQueryableDocIdsSnapshot().contains(0));
+ assertFalse(refreshSC.getDocIdsSnapshot().contains(0));
}
// docId=12 in seg03 was newly added.
if (refreshSC.getIndexSegment() == seg03) {
- assertTrue(refreshSC.getQueryableDocIdsSnapshot().contains(12));
+ assertTrue(refreshSC.getDocIdsSnapshot().contains(12));
}
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManagerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManagerTest.java
new file mode 100644
index 00000000000..9b912bfc721
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapTableUpsertMetadataManagerTest.java
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.upsert;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.segment.local.data.manager.TableDataManager;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.MutableSegment;
+import org.apache.pinot.segment.spi.SegmentContext;
+import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.UpsertConfig;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+
+
+/**
+ * Covers the mode-dispatch branch that a prior review pass regressed:
NONE/skipUpsertView must take the direct,
+ * unlocked read; SYNC/SNAPSHOT must delegate to {@link UpsertViewManager},
which is the only path that respects
+ * segment tracking.
+ */
+public class ConcurrentMapTableUpsertMetadataManagerTest {
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final Schema SCHEMA = new Schema.SchemaBuilder()
+ .setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension("myCol", DataType.STRING)
+ .addDateTimeField("timeCol", DataType.LONG, "TIMESTAMP",
"1:MILLISECONDS")
+ .setPrimaryKeyColumns(List.of("myCol"))
+ .build();
+
+ private ConcurrentMapTableUpsertMetadataManager
createManager(UpsertConfig.ConsistencyMode consistencyMode) {
+ UpsertConfig upsertConfig = new UpsertConfig(UpsertConfig.Mode.FULL);
+ upsertConfig.setConsistencyMode(consistencyMode);
+ TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME)
+ .setTableName(RAW_TABLE_NAME)
+ .setTimeColumnName("timeCol")
+ .setUpsertConfig(upsertConfig)
+ .build();
+ TableDataManager tableDataManager = mock(TableDataManager.class);
+ when(tableDataManager.getTableDataDir()).thenReturn(new
File(RAW_TABLE_NAME));
+ return (ConcurrentMapTableUpsertMetadataManager)
TableUpsertMetadataManagerFactory.create(new PinotConfiguration(),
+ tableConfig, SCHEMA, tableDataManager, null);
+ }
+
+ private IndexSegment mockSegmentWithDistinctBitmaps() {
+ IndexSegment segment = mock(MutableSegment.class);
+ ThreadSafeMutableRoaringBitmap validBitmap =
mock(ThreadSafeMutableRoaringBitmap.class);
+ ThreadSafeMutableRoaringBitmap queryableBitmap =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(validBitmap.getMutableRoaringBitmap()).thenReturn(new
MutableRoaringBitmap());
+ when(queryableBitmap.getMutableRoaringBitmap()).thenReturn(new
MutableRoaringBitmap());
+ when(segment.getValidDocIds()).thenReturn(validBitmap);
+ when(segment.getQueryableDocIds()).thenReturn(queryableBitmap);
+ when(segment.getSegmentName()).thenReturn("seg1");
+ return segment;
+ }
+
+ @Test
+ public void testNoneModeTakesDirectReadEvenWhenUntracked() {
+ ConcurrentMapTableUpsertMetadataManager manager =
createManager(UpsertConfig.ConsistencyMode.NONE);
+ IndexSegment segment = mockSegmentWithDistinctBitmaps();
+ SegmentContext segmentContext = new SegmentContext(segment);
+
+ // Never tracked via any UpsertViewManager, yet a direct read still
populates the snapshot.
+ manager.setSegmentContexts(List.of(segmentContext),
Map.of(QueryOptionKey.SKIP_UPSERT_DELETE, "true"));
+ assertSame(segmentContext.getDocIdsSnapshot(),
segment.getValidDocIds().getMutableRoaringBitmap());
+ }
+
+ @Test
+ public void testSkipUpsertViewTakesDirectReadUnderSyncMode() {
+ ConcurrentMapTableUpsertMetadataManager manager =
createManager(UpsertConfig.ConsistencyMode.SYNC);
+ IndexSegment segment = mockSegmentWithDistinctBitmaps();
+ SegmentContext segmentContext = new SegmentContext(segment);
+
+ manager.setSegmentContexts(List.of(segmentContext),
+ Map.of(QueryOptionKey.SKIP_UPSERT_VIEW, "true",
QueryOptionKey.SKIP_UPSERT_DELETE, "true"));
+ assertSame(segmentContext.getDocIdsSnapshot(),
segment.getValidDocIds().getMutableRoaringBitmap());
+ }
+
+ /// ConcurrentMapTableUpsertMetadataManager's dispatch only special-cases
NONE (see the other tests above);
+ /// SYNC and SNAPSHOT both fall into this same delegate branch, so testing
one mode here is sufficient — their
+ /// differing internal behavior is UpsertViewManager's concern, covered by
UpsertViewManagerTest instead.
+ @Test
+ public void testDelegatesToUpsertViewManagerAndRespectsTracking() {
+ ConcurrentMapTableUpsertMetadataManager manager =
createManager(UpsertConfig.ConsistencyMode.SYNC);
+ IndexSegment segment = mockSegmentWithDistinctBitmaps();
+ SegmentContext segmentContext = new SegmentContext(segment);
+
+ // Force the partition manager (and its UpsertViewManager) to exist, but
deliberately don't track the segment:
+ // a correct delegation to UpsertViewManager leaves the snapshot unset,
since UpsertViewManager only serves
+ // tracked segments. A regression back to a direct-read bypass would
populate it regardless of tracking, as
+ // testNoneModeTakesDirectReadEvenWhenUntracked demonstrates. (Without
forcing the partition manager to exist
+ // first, this assertion would pass vacuously: setSegmentContexts would
iterate zero partition managers.)
+ manager.getOrCreatePartitionManager(0);
+ manager.setSegmentContexts(List.of(segmentContext),
Map.of(QueryOptionKey.SKIP_UPSERT_DELETE, "true"));
+ assertNull(segmentContext.getDocIdsSnapshot());
+
+
manager.getOrCreatePartitionManager(0).getUpsertViewManager().trackSegment(segment);
+ manager.setSegmentContexts(List.of(segmentContext),
Map.of(QueryOptionKey.SKIP_UPSERT_DELETE, "true"));
+ assertSame(segmentContext.getDocIdsSnapshot(),
segment.getValidDocIds().getMutableRoaringBitmap());
+ }
+}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertUtilsTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertUtilsTest.java
index de88a5b27d3..e60a397af28 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertUtilsTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertUtilsTest.java
@@ -20,6 +20,7 @@ package org.apache.pinot.segment.local.upsert;
import java.util.HashMap;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentImpl;
+import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
import org.apache.pinot.segment.spi.store.SegmentDirectory;
@@ -30,11 +31,43 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
public class UpsertUtilsTest {
+ @Test
+ public void testGetValidDocIdsSnapshotFromSegmentReturnsValidDocsDirectly() {
+ IndexSegment segment = mock(IndexSegment.class);
+ ThreadSafeMutableRoaringBitmap validDocIds =
mock(ThreadSafeMutableRoaringBitmap.class);
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ when(validDocIds.getMutableRoaringBitmap()).thenReturn(bitmap);
+ when(segment.getValidDocIds()).thenReturn(validDocIds);
+ // Queryable docs must never be consulted by this method, even when
present.
+
when(segment.getQueryableDocIds()).thenReturn(mock(ThreadSafeMutableRoaringBitmap.class));
+
+ assertSame(UpsertUtils.getValidDocIdsSnapshotFromSegment(segment), bitmap);
+ }
+
+ @Test
+ public void
testGetValidDocIdsSnapshotFromSegmentNullWithoutUseEmptyForNull() {
+ IndexSegment segment = mock(IndexSegment.class);
+ when(segment.getValidDocIds()).thenReturn(null);
+
+ assertNull(UpsertUtils.getValidDocIdsSnapshotFromSegment(segment));
+ }
+
+ @Test
+ public void testGetValidDocIdsSnapshotFromSegmentEmptyWithUseEmptyForNull() {
+ IndexSegment segment = mock(IndexSegment.class);
+ when(segment.getValidDocIds()).thenReturn(null);
+
+ MutableRoaringBitmap result =
UpsertUtils.getValidDocIdsSnapshotFromSegment(segment, true);
+ assertTrue(result.isEmpty());
+ }
+
/// Non-upsert table: `_partitionUpsertMetadataManager` is never set, so
`hasNoQueryableDocs`
/// short-circuits to `false`.
@Test
@@ -135,6 +168,78 @@ public class UpsertUtilsTest {
assertFalse(segment.hasNoQueryableDocs());
}
+ // -------- hasNoValidDocs: mirrors hasNoQueryableDocs, but against the
valid-docs bitmap/cache. --------
+
+ @Test
+ public void testHasNoValidDocsNonUpsert() {
+ ImmutableSegmentImpl segment = newSegment();
+ assertFalse(segment.hasNoValidDocs());
+ }
+
+ @Test
+ public void testHasNoValidDocsLiveEmpty() {
+ ImmutableSegmentImpl segment = newSegment();
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ when(manager.getUpsertViewManager()).thenReturn(null);
+ ThreadSafeMutableRoaringBitmap valid =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(valid.isEmpty()).thenReturn(true);
+ segment.enableUpsert(manager, valid, null);
+ assertTrue(segment.hasNoValidDocs());
+ }
+
+ @Test
+ public void testHasNoValidDocsLiveNonEmpty() {
+ ImmutableSegmentImpl segment = newSegment();
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ when(manager.getUpsertViewManager()).thenReturn(null);
+ ThreadSafeMutableRoaringBitmap valid =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(valid.isEmpty()).thenReturn(false);
+ segment.enableUpsert(manager, valid, null);
+ assertFalse(segment.hasNoValidDocs());
+ }
+
+ @Test
+ public void testHasNoValidDocsConsistencyModeSnapshotEmpty() {
+ ImmutableSegmentImpl segment = newSegment();
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ UpsertViewManager viewManager = mock(UpsertViewManager.class);
+ when(manager.getUpsertViewManager()).thenReturn(viewManager);
+ when(viewManager.getValidDocIdsSnapshot(any())).thenReturn(new
MutableRoaringBitmap());
+ ThreadSafeMutableRoaringBitmap liveValid =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(liveValid.isEmpty()).thenReturn(false);
+ segment.enableUpsert(manager, liveValid, null);
+ assertTrue(segment.hasNoValidDocs());
+ }
+
+ @Test
+ public void testHasNoValidDocsConsistencyModeSnapshotNonEmpty() {
+ ImmutableSegmentImpl segment = newSegment();
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ UpsertViewManager viewManager = mock(UpsertViewManager.class);
+ MutableRoaringBitmap snapshot = new MutableRoaringBitmap();
+ snapshot.add(0);
+ when(manager.getUpsertViewManager()).thenReturn(viewManager);
+ when(viewManager.getValidDocIdsSnapshot(any())).thenReturn(snapshot);
+ ThreadSafeMutableRoaringBitmap liveValid =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(liveValid.isEmpty()).thenReturn(true);
+ segment.enableUpsert(manager, liveValid, null);
+ assertFalse(segment.hasNoValidDocs());
+ }
+
+ @Test
+ public void testHasNoValidDocsConsistencyModeSnapshotAbsent() {
+ // No refresh yet: don't claim empty even if the live bitmap currently is.
+ ImmutableSegmentImpl segment = newSegment();
+ PartitionUpsertMetadataManager manager =
mock(PartitionUpsertMetadataManager.class);
+ UpsertViewManager viewManager = mock(UpsertViewManager.class);
+ when(manager.getUpsertViewManager()).thenReturn(viewManager);
+ when(viewManager.getValidDocIdsSnapshot(any())).thenReturn(null);
+ ThreadSafeMutableRoaringBitmap liveValid =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(liveValid.isEmpty()).thenReturn(true);
+ segment.enableUpsert(manager, liveValid, null);
+ assertFalse(segment.hasNoValidDocs());
+ }
+
/// Returns a minimal {@link ImmutableSegmentImpl} usable for testing the
upsert-aware methods.
/// Mirrors the pattern used in {@code
BasePartitionUpsertMetadataManagerTest#createImmutableSegment}.
private static ImmutableSegmentImpl newSegment() {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertViewManagerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertViewManagerTest.java
index 95d48b66780..3866ed23b43 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertViewManagerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/UpsertViewManagerTest.java
@@ -20,12 +20,14 @@ package org.apache.pinot.segment.local.upsert;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.MutableSegment;
import org.apache.pinot.segment.spi.SegmentContext;
import
org.apache.pinot.segment.spi.index.mutable.ThreadSafeMutableRoaringBitmap;
import org.apache.pinot.spi.config.table.UpsertConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
import org.testng.annotations.Test;
@@ -52,12 +54,79 @@ public class UpsertViewManagerTest {
mgr.trackSegment(seg1);
assertEquals(mgr.getTrackedSegments(), Set.of(seg1));
mgr.setSegmentContexts(List.of(segCtx1), new HashMap<>());
- assertSame(segCtx1.getQueryableDocIdsSnapshot(), mutableRoaringBitmap);
+ assertSame(segCtx1.getDocIdsSnapshot(), mutableRoaringBitmap);
mgr.untrackSegment(seg1);
assertTrue(mgr.getTrackedSegments().isEmpty());
segCtx1 = new SegmentContext(seg1);
mgr.setSegmentContexts(List.of(segCtx1), new HashMap<>());
- assertNull(segCtx1.getQueryableDocIdsSnapshot());
+ assertNull(segCtx1.getDocIdsSnapshot());
+ }
+
+ @Test
+ public void testSkipUpsertDeleteReadsValidBitmapUnderSyncMode() {
+ UpsertViewManager mgr = new
UpsertViewManager(UpsertConfig.ConsistencyMode.SYNC, mock(UpsertContext.class));
+ IndexSegment seg1 = mockSegmentWithDistinctBitmaps();
+ mgr.trackSegment(seg1);
+
+ SegmentContext segCtx = new SegmentContext(seg1);
+ mgr.setSegmentContexts(List.of(segCtx),
Map.of(CommonConstants.Broker.Request.QueryOptionKey.SKIP_UPSERT_DELETE,
+ "true"));
+ assertSame(segCtx.getDocIdsSnapshot(), VALID_RESULT);
+
+ SegmentContext defaultCtx = new SegmentContext(seg1);
+ mgr.setSegmentContexts(List.of(defaultCtx), new HashMap<>());
+ assertSame(defaultCtx.getDocIdsSnapshot(), QUERYABLE_RESULT);
+ }
+
+ @Test
+ public void testSkipUpsertDeleteReadsValidBitmapUnderSnapshotMode() {
+ UpsertViewManager mgr = new
UpsertViewManager(UpsertConfig.ConsistencyMode.SNAPSHOT,
mock(UpsertContext.class));
+ IndexSegment seg1 = mockSegmentWithDistinctBitmaps();
+ mgr.trackSegment(seg1);
+
+ // skipUpsertDelete reads the cached valid-docs map, refreshed alongside
the queryable-docs one on track.
+ SegmentContext segCtx = new SegmentContext(seg1);
+ mgr.setSegmentContexts(List.of(segCtx),
Map.of(CommonConstants.Broker.Request.QueryOptionKey.SKIP_UPSERT_DELETE,
+ "true"));
+ assertSame(segCtx.getDocIdsSnapshot(), VALID_RESULT);
+
+ // Without the option, the cached queryable-docs map is used instead.
+ SegmentContext defaultCtx = new SegmentContext(seg1);
+ mgr.setSegmentContexts(List.of(defaultCtx), new HashMap<>());
+ assertSame(defaultCtx.getDocIdsSnapshot(), QUERYABLE_RESULT);
+ }
+
+ /// Without a delete column, queryable docs == valid docs; the refresh
reuses one clone for both caches instead
+ /// of cloning twice (see doBatchRefreshUpsertView), so both maps must
resolve to the same bitmap instance.
+ @Test
+ public void testNoDeleteColumnReusesSameBitmapInBothCaches() {
+ UpsertViewManager mgr = new
UpsertViewManager(UpsertConfig.ConsistencyMode.SNAPSHOT,
mock(UpsertContext.class));
+ IndexSegment seg1 = mock(MutableSegment.class);
+ ThreadSafeMutableRoaringBitmap validBitmap =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(validBitmap.getMutableRoaringBitmap()).thenReturn(new
MutableRoaringBitmap());
+ when(seg1.getValidDocIds()).thenReturn(validBitmap);
+ when(seg1.getQueryableDocIds()).thenReturn(null);
+ when(seg1.getSegmentName()).thenReturn("seg1");
+
+ mgr.trackSegment(seg1);
+ assertSame(mgr.getQueryableDocIdsSnapshot(seg1),
mgr.getValidDocIdsSnapshot(seg1));
+ }
+
+ private static final MutableRoaringBitmap VALID_RESULT = new
MutableRoaringBitmap();
+ private static final MutableRoaringBitmap QUERYABLE_RESULT = new
MutableRoaringBitmap();
+
+ /// Mocks a segment whose valid-docs and queryable-docs bitmaps resolve to
two different, identity-comparable
+ /// {@link MutableRoaringBitmap} instances, so tests can assert exactly
which one a code path picked.
+ private static IndexSegment mockSegmentWithDistinctBitmaps() {
+ IndexSegment segment = mock(MutableSegment.class);
+ ThreadSafeMutableRoaringBitmap validBitmap =
mock(ThreadSafeMutableRoaringBitmap.class);
+ ThreadSafeMutableRoaringBitmap queryableBitmap =
mock(ThreadSafeMutableRoaringBitmap.class);
+ when(validBitmap.getMutableRoaringBitmap()).thenReturn(VALID_RESULT);
+
when(queryableBitmap.getMutableRoaringBitmap()).thenReturn(QUERYABLE_RESULT);
+ when(segment.getValidDocIds()).thenReturn(validBitmap);
+ when(segment.getQueryableDocIds()).thenReturn(queryableBitmap);
+ when(segment.getSegmentName()).thenReturn("seg1");
+ return segment;
}
}
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
index a491d79457c..acd90638b17 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/IndexSegment.java
@@ -110,6 +110,11 @@ public interface IndexSegment {
return false;
}
+ /// Same as {@link #hasNoQueryableDocs}, against {@link #getValidDocIds}
instead.
+ default boolean hasNoValidDocs() {
+ return false;
+ }
+
/**
* Returns the record for the given document id. Virtual column values are
not returned.
* <p>NOTE: don't use this method for high performance code. Use
PinotSegmentRecordReader when reading multiple
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentContext.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentContext.java
index f2c3b8f8372..993612b5871 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentContext.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/SegmentContext.java
@@ -24,8 +24,9 @@ import org.roaringbitmap.buffer.MutableRoaringBitmap;
public class SegmentContext {
private final IndexSegment _indexSegment;
+ /// Queryable-docs bitmap by default, or valid-docs (tombstones included)
when the query set `skipUpsertDelete`.
@Nullable
- private MutableRoaringBitmap _queryableDocIdsSnapshot = null;
+ private MutableRoaringBitmap _docIdsSnapshot = null;
public SegmentContext(IndexSegment indexSegment) {
_indexSegment = indexSegment;
@@ -35,12 +36,25 @@ public class SegmentContext {
return _indexSegment;
}
+ @Nullable
+ public MutableRoaringBitmap getDocIdsSnapshot() {
+ return _docIdsSnapshot;
+ }
+
+ public void setDocIdsSnapshot(@Nullable MutableRoaringBitmap docIdsSnapshot)
{
+ _docIdsSnapshot = docIdsSnapshot;
+ }
+
+ /// @deprecated Use {@link #getDocIdsSnapshot} instead; kept for binary
compatibility.
+ @Deprecated
@Nullable
public MutableRoaringBitmap getQueryableDocIdsSnapshot() {
- return _queryableDocIdsSnapshot;
+ return getDocIdsSnapshot();
}
+ /// @deprecated Use {@link #setDocIdsSnapshot} instead; kept for binary
compatibility.
+ @Deprecated
public void setQueryableDocIdsSnapshot(@Nullable MutableRoaringBitmap
queryableDocIdsSnapshot) {
- _queryableDocIdsSnapshot = queryableDocIdsSnapshot;
+ setDocIdsSnapshot(queryableDocIdsSnapshot);
}
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index f617e91b74b..c4b6fa71c90 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -741,6 +741,9 @@ public class CommonConstants {
public static final String SKIP_UPSERT = "skipUpsert";
public static final String SKIP_UPSERT_VIEW = "skipUpsertView";
public static final String UPSERT_VIEW_FRESHNESS_MS =
"upsertViewFreshnessMs";
+ /// Default `false` (queryable docs, tombstones excluded). When
`true`, skips the delete/tombstone
+ /// exclusion so rows deleted via `deleteRecordColumn` are included.
+ public static final String SKIP_UPSERT_DELETE = "skipUpsertDelete";
public static final String USE_STAR_TREE = "useStarTree";
/**
* When true, use index-based distinct operators when applicable. This
enables both
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
