Skip to content

Add consuming metadata to broker response #9092

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snleee merged 2 commits into from
Jul 27, 2022
Merged

Add consuming metadata to broker response #9092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0fa893c
Add consuming metadata to broker response
ravishankar15 Jul 22, 2022
fda1240
Fix compatibility failures
ravishankar15 Jul 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 62 additions & 62 deletions ...atibility-verifier/sample-test-suite/config/query-results/feature-test-1-rest-sql.results
View file
Open in desktop

Large diffs are not rendered by default.

62 changes: 31 additions & 31 deletions ...ility-verifier/sample-test-suite/config/query-results/feature-test-2-sql-realtime.results
View file
Open in desktop

Large diffs are not rendered by default.

46 changes: 23 additions & 23 deletions ...ility-verifier/sample-test-suite/config/query-results/feature-test-3-sql-realtime.results
View file
Open in desktop

Large diffs are not rendered by default.

4 changes: 3 additions & 1 deletion ...broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -670,14 +670,16 @@ private void logBrokerResponse(long requestId, String query, RequestContext requ
if (_queryLogRateLimiter.tryAcquire() || forceLog(brokerResponse, totalTimeMs)) {
// Table name might have been changed (with suffix _OFFLINE/_REALTIME appended)
LOGGER.info("requestId={},table={},timeMs={},docs={}/{},entries={}/{},"
+ "segments(queried/processed/matched/consuming/unavailable):{}/{}/{}/{}/{},consumingFreshnessTimeMs={},"
+ "segments(queried/processed/matched/consumingQueried/consumingMatched/consumingProcessed/unavailable):"
+ "{}/{}/{}/{}/{}/{}/{},consumingFreshnessTimeMs={},"
+ "servers={}/{},groupLimitReached={},brokerReduceTimeMs={},exceptions={},serverStats={},"
+ "offlineThreadCpuTimeNs(total/thread/sysActivity/resSer):{}/{}/{}/{},"
+ "realtimeThreadCpuTimeNs(total/thread/sysActivity/resSer):{}/{}/{}/{},query={}", requestId, tableName,
totalTimeMs, brokerResponse.getNumDocsScanned(), brokerResponse.getTotalDocs(),
brokerResponse.getNumEntriesScannedInFilter(), brokerResponse.getNumEntriesScannedPostFilter(),
brokerResponse.getNumSegmentsQueried(), brokerResponse.getNumSegmentsProcessed(),
brokerResponse.getNumSegmentsMatched(), brokerResponse.getNumConsumingSegmentsQueried(),
brokerResponse.getNumConsumingSegmentsProcessed(), brokerResponse.getNumConsumingSegmentsMatched(),
numUnavailableSegments, brokerResponse.getMinConsumingFreshnessTimeMs(),
brokerResponse.getNumServersResponded(), brokerResponse.getNumServersQueried(),
brokerResponse.isNumGroupsLimitReached(), requestContext.getReduceTimeMillis(),
Expand Down
10 changes: 10 additions & 0 deletions pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,16 @@ String toJsonString()
*/
long getNumConsumingSegmentsQueried();

/**
* Get number of consuming segments processed by server after server side pruning
*/
long getNumConsumingSegmentsProcessed();

/**
* Get number of consuming segments that had at least one matching document
*/
long getNumConsumingSegmentsMatched();

/**
* Get the minimum freshness timestamp across consuming segments that were queried
*/
Expand Down
31 changes: 27 additions & 4 deletions pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,10 +42,10 @@
*/
@JsonPropertyOrder({
"resultTable", "exceptions", "numServersQueried", "numServersResponded", "numSegmentsQueried",
"numSegmentsProcessed", "numSegmentsMatched", "numConsumingSegmentsQueried", "numDocsScanned",
"numEntriesScannedInFilter", "numEntriesScannedPostFilter", "numGroupsLimitReached", "totalDocs", "timeUsedMs",
"offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs", "offlineSystemActivitiesCpuTimeNs",
"realtimeSystemActivitiesCpuTimeNs", "offlineResponseSerializationCpuTimeNs",
"numSegmentsProcessed", "numSegmentsMatched", "numConsumingSegmentsQueried", "numConsumingSegmentsProcessed",
"numConsumingSegmentsMatched", "numDocsScanned", "numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numGroupsLimitReached", "totalDocs", "timeUsedMs", "offlineThreadCpuTimeNs", "realtimeThreadCpuTimeNs",
"offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs", "offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs", "realtimeTotalCpuTimeNs", "segmentStatistics",
"traceInfo"
})
Expand All @@ -66,6 +66,8 @@ public class BrokerResponseNative implements BrokerResponse {
private long _numSegmentsProcessed = 0L;
private long _numSegmentsMatched = 0L;
private long _numConsumingSegmentsQueried = 0L;
private long _numConsumingSegmentsProcessed = 0L;
private long _numConsumingSegmentsMatched = 0L;
// the timestamp indicating the freshness of the data queried in consuming segments.
// This can be ingestion timestamp if provided by the stream, or the last index time
private long _minConsumingFreshnessTimeMs = 0L;
Expand Down Expand Up @@ -430,6 +432,27 @@ public void setNumConsumingSegmentsQueried(long numConsumingSegmentsQueried) {
_numConsumingSegmentsQueried = numConsumingSegmentsQueried;
}

@JsonProperty("numConsumingSegmentsProcessed")
@Override
public long getNumConsumingSegmentsProcessed() {
return _numConsumingSegmentsProcessed;
}
@JsonProperty("numConsumingSegmentsProcessed")
public void setNumConsumingSegmentsProcessed(long numConsumingSegmentsProcessed) {
_numConsumingSegmentsProcessed = numConsumingSegmentsProcessed;
}

@JsonProperty("numConsumingSegmentsMatched")
@Override
public long getNumConsumingSegmentsMatched() {
return _numConsumingSegmentsMatched;
}

@JsonProperty("numConsumingSegmentsMatched")
public void setNumConsumingSegmentsMatched(long numConsumingSegmentsMatched) {
_numConsumingSegmentsMatched = numConsumingSegmentsMatched;
}

@JsonProperty("minConsumingFreshnessTimeMs")
@Override
public long getMinConsumingFreshnessTimeMs() {
Expand Down
2 changes: 2 additions & 0 deletions pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ enum MetadataKey {
NUM_SEGMENTS_PROCESSED("numSegmentsProcessed", MetadataValueType.INT),
NUM_SEGMENTS_MATCHED("numSegmentsMatched", MetadataValueType.INT),
NUM_CONSUMING_SEGMENTS_QUERIED("numConsumingSegmentsQueried", MetadataValueType.INT),
NUM_CONSUMING_SEGMENTS_PROCESSED("numConsumingSegmentsProcessed", MetadataValueType.INT),
Copy link
Contributor

@Jackie-Jiang Jackie-Jiang Jul 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(MAJOR) Please revert this PR, and re-check it in after it passes the compatibility test. We cannot add MetadataKey in the middle because we use ordinal to encode the key. See the comment of this enum

Copy link
Contributor

@siddharthteotia siddharthteotia Jul 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1. This is a breaking change

NUM_CONSUMING_SEGMENTS_MATCHED("numConsumingSegmentsMatched", MetadataValueType.INT),
MIN_CONSUMING_FRESHNESS_TIME_MS("minConsumingFreshnessTimeMs", MetadataValueType.LONG),
TOTAL_DOCS("totalDocs", MetadataValueType.LONG),
NUM_GROUPS_LIMIT_REACHED("numGroupsLimitReached", MetadataValueType.STRING),
Expand Down
24 changes: 24 additions & 0 deletions pinot-core/src/main/java/org/apache/pinot/core/operator/blocks/IntermediateResultsBlock.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@ public class IntermediateResultsBlock implements Block {
private long _numTotalDocs;
private int _numSegmentsProcessed;
private int _numSegmentsMatched;
private int _numConsumingSegmentsProcessed;
private int _numConsumingSegmentsMatched;
private boolean _numGroupsLimitReached;
private int _numResizes;
private long _resizeTimeMs;
Expand Down Expand Up @@ -269,6 +271,24 @@ public void setNumSegmentsMatched(int numSegmentsMatched) {
_numSegmentsMatched = numSegmentsMatched;
}

@VisibleForTesting
public int getNumConsumingSegmentsProcessed() {
return _numConsumingSegmentsProcessed;
}

public void setNumConsumingSegmentsProcessed(int numConsumingSegmentsProcessed) {
_numConsumingSegmentsProcessed = numConsumingSegmentsProcessed;
}

@VisibleForTesting
public int getNumConsumingSegmentsMatched() {
return _numConsumingSegmentsMatched;
}

public void setNumConsumingSegmentsMatched(int numConsumingSegmentsMatched) {
_numConsumingSegmentsMatched = numConsumingSegmentsMatched;
}

@VisibleForTesting
public long getNumTotalDocs() {
return _numTotalDocs;
Expand Down Expand Up @@ -468,6 +488,10 @@ private DataTable attachMetadataToDataTable(DataTable dataTable) {
.put(MetadataKey.NUM_ENTRIES_SCANNED_POST_FILTER.getName(), String.valueOf(_numEntriesScannedPostFilter));
dataTable.getMetadata().put(MetadataKey.NUM_SEGMENTS_PROCESSED.getName(), String.valueOf(_numSegmentsProcessed));
dataTable.getMetadata().put(MetadataKey.NUM_SEGMENTS_MATCHED.getName(), String.valueOf(_numSegmentsMatched));
dataTable.getMetadata().put(MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED.getName(),
String.valueOf(_numConsumingSegmentsProcessed));
dataTable.getMetadata().put(MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED.getName(),
String.valueOf(_numConsumingSegmentsMatched));
dataTable.getMetadata().put(MetadataKey.NUM_RESIZES.getName(), String.valueOf(_numResizes));
dataTable.getMetadata().put(MetadataKey.RESIZE_TIME_MS.getName(), String.valueOf(_resizeTimeMs));

Expand Down
14 changes: 14 additions & 0 deletions pinot-core/src/main/java/org/apache/pinot/core/operator/combine/CombineOperatorUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.apache.pinot.core.common.Operator;
import org.apache.pinot.core.operator.ExecutionStatistics;
import org.apache.pinot.core.operator.blocks.IntermediateResultsBlock;
import org.apache.pinot.segment.spi.MutableSegment;


@SuppressWarnings("rawtypes")
Expand Down Expand Up @@ -56,6 +57,8 @@ public static void setExecutionStatistics(IntermediateResultsBlock resultsBlock,
long threadCpuTimeNs, int numServerThreads) {
int numSegmentsProcessed = operators.size();
int numSegmentsMatched = 0;
int numConsumingSegmentsProcessed = 0;
int numConsumingSegmentsMatched = 0;
long numDocsScanned = 0;
long numEntriesScannedInFilter = 0;
long numEntriesScannedPostFilter = 0;
Expand All @@ -65,13 +68,24 @@ public static void setExecutionStatistics(IntermediateResultsBlock resultsBlock,
if (executionStatistics.getNumDocsScanned() > 0) {
numSegmentsMatched++;
}
try {
Copy link
Contributor

@snleee snleee Jul 23, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you remove the try-catch block?

Copy link
Contributor Author

@ravishankar15 ravishankar15 Jul 25, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@snleee I am not sure if we can remove that for queries with limit 0 like SELECT * FROM testTable LIMIT 0 we are passing EmptySelectionOperator Instance in there which does not implement the getIndexSegment() hence there will be an exception we have a explicit test case to prevent that SelectionCombineOperatorTest.testSelectionLimit0 Since we are using the interface (Operator) to reference in the for loop and interface has explicit check to throw the exception I think it is better to enclose it within the try catch block.

Also is the regression failing because of this try catch ? I could not see any logs associated with the same ? Can you please help me understand the failure so that I can fix the issue. Thanks

Copy link
Contributor

@snleee snleee Jul 25, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ravishankar15 I thought getIndexSegment() would return null but the interface actually throw UnsupportedOperationException and you're handling it correctly.

Let me take a look on the regression test and try to check why it fails.

if (operator.getIndexSegment() instanceof MutableSegment) {
numConsumingSegmentsProcessed += 1;
if (executionStatistics.getNumDocsScanned() > 0) {
numConsumingSegmentsMatched++;
}
}
} catch (UnsupportedOperationException ignored) { }

numDocsScanned += executionStatistics.getNumDocsScanned();
numEntriesScannedInFilter += executionStatistics.getNumEntriesScannedInFilter();
numEntriesScannedPostFilter += executionStatistics.getNumEntriesScannedPostFilter();
numTotalDocs += executionStatistics.getNumTotalDocs();
}
resultsBlock.setNumSegmentsProcessed(numSegmentsProcessed);
resultsBlock.setNumSegmentsMatched(numSegmentsMatched);
resultsBlock.setNumConsumingSegmentsProcessed(numConsumingSegmentsProcessed);
resultsBlock.setNumConsumingSegmentsMatched(numConsumingSegmentsMatched);
resultsBlock.setNumDocsScanned(numDocsScanned);
resultsBlock.setNumEntriesScannedInFilter(numEntriesScannedInFilter);
resultsBlock.setNumEntriesScannedPostFilter(numEntriesScannedPostFilter);
Expand Down
2 changes: 2 additions & 0 deletions pinot-core/src/main/java/org/apache/pinot/core/query/executor/ServerQueryExecutorV1Impl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,8 @@ private DataTable processQuery(List<IndexSegment> indexSegments, QueryContext qu
metadata.put(MetadataKey.NUM_ENTRIES_SCANNED_POST_FILTER.getName(), "0");
metadata.put(MetadataKey.NUM_SEGMENTS_PROCESSED.getName(), "0");
metadata.put(MetadataKey.NUM_SEGMENTS_MATCHED.getName(), "0");
metadata.put(MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED.getName(), "0");
metadata.put(MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED.getName(), "0");
metadata.put(MetadataKey.NUM_SEGMENTS_PRUNED_BY_SERVER.getName(), String.valueOf(totalSegments));
addPrunerStats(metadata, prunerStats);
return dataTable;
Expand Down
14 changes: 14 additions & 0 deletions pinot-core/src/main/java/org/apache/pinot/core/query/reduce/BaseReduceService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,8 @@ protected static class ExecutionStatsAggregator {
private long _numSegmentsProcessed = 0L;
private long _numSegmentsMatched = 0L;
private long _numConsumingSegmentsQueried = 0L;
private long _numConsumingSegmentsProcessed = 0L;
private long _numConsumingSegmentsMatched = 0L;
private long _minConsumingFreshnessTimeMs = Long.MAX_VALUE;
private long _numTotalDocs = 0L;
private long _offlineThreadCpuTimeNs = 0L;
Expand Down Expand Up @@ -200,6 +202,16 @@ protected synchronized void aggregate(ServerRoutingInstance routingInstance, Dat
_numConsumingSegmentsQueried += Long.parseLong(numConsumingSegmentsQueriedString);
}

String numConsumingSegmentsProcessed = metadata.get(MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED.getName());
if (numConsumingSegmentsProcessed != null) {
_numConsumingSegmentsProcessed += Long.parseLong(numConsumingSegmentsProcessed);
}

String numConsumingSegmentsMatched = metadata.get(MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED.getName());
if (numConsumingSegmentsMatched != null) {
_numConsumingSegmentsMatched += Long.parseLong(numConsumingSegmentsMatched);
}

String minConsumingFreshnessTimeMsString = metadata.get(MetadataKey.MIN_CONSUMING_FRESHNESS_TIME_MS.getName());
if (minConsumingFreshnessTimeMsString != null) {
_minConsumingFreshnessTimeMs =
Expand Down Expand Up @@ -300,6 +312,8 @@ protected void setStats(String rawTableName, BrokerResponseNative brokerResponse
brokerResponseNative.setNumConsumingSegmentsQueried(_numConsumingSegmentsQueried);
brokerResponseNative.setMinConsumingFreshnessTimeMs(_minConsumingFreshnessTimeMs);
}
brokerResponseNative.setNumConsumingSegmentsProcessed(_numConsumingSegmentsProcessed);
brokerResponseNative.setNumConsumingSegmentsMatched(_numConsumingSegmentsMatched);

// Update broker metrics.
if (brokerMetrics != null) {
Expand Down
8 changes: 7 additions & 1 deletion pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/QueryScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,10 @@ protected byte[] processQueryAndSerialize(ServerQueryRequest queryRequest, Execu
MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE.getName(), INVALID_SEGMENTS_COUNT));
long numSegmentsConsuming = Long.parseLong(
dataTableMetadata.getOrDefault(MetadataKey.NUM_CONSUMING_SEGMENTS_QUERIED.getName(), INVALID_SEGMENTS_COUNT));
long numConsumingSegmentsProcessed = Long.parseLong(
dataTableMetadata.getOrDefault(MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED.getName(), INVALID_SEGMENTS_COUNT));
long numConsumingSegmentsMatched = Long.parseLong(
dataTableMetadata.getOrDefault(MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED.getName(), INVALID_SEGMENTS_COUNT));
long minConsumingFreshnessMs = Long.parseLong(
dataTableMetadata.getOrDefault(MetadataKey.MIN_CONSUMING_FRESHNESS_TIME_MS.getName(), INVALID_FRESHNESS_MS));
int numResizes =
Expand Down Expand Up @@ -242,11 +246,13 @@ protected byte[] processQueryAndSerialize(ServerQueryRequest queryRequest, Execu
// Please add new entries at the end
if (_queryLogRateLimiter.tryAcquire() || forceLog(schedulerWaitMs, numDocsScanned, numSegmentsPrunedInvalid)) {
LOGGER.info("Processed requestId={},table={},"
+ "segments(queried/processed/matched/consuming/invalid/limit/value)={}/{}/{}/{}/{}/{}/{},"
+ "segments(queried/processed/matched/consumingQueried/consumingProcessed/consumingMatched/"
+ "invalid/limit/value)={}/{}/{}/{}/{}/{}/{}/{}/{},"
+ "schedulerWaitMs={},reqDeserMs={},totalExecMs={},resSerMs={},totalTimeMs={},minConsumingFreshnessMs={},"
+ "broker={},numDocsScanned={},scanInFilter={},scanPostFilter={},sched={},"
+ "threadCpuTimeNs(total/thread/sysActivity/resSer)={}/{}/{}/{}", requestId, tableNameWithType,
numSegmentsQueried, numSegmentsProcessed, numSegmentsMatched, numSegmentsConsuming,
numConsumingSegmentsProcessed, numConsumingSegmentsMatched,
numSegmentsPrunedInvalid, numSegmentsPrunedByLimit, numSegmentsPrunedByValue, schedulerWaitMs,
timerContext.getPhaseDurationMs(ServerQueryPhase.REQUEST_DESERIALIZATION),
timerContext.getPhaseDurationMs(ServerQueryPhase.QUERY_PROCESSING),
Expand Down
Loading