You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "navina (via GitHub)" <gi...@apache.org> on 2026/05/28 20:56:17 UTC
[PR] Add merge-only reducer API for intercepting intermediate results [pinot]
navina opened a new pull request, #18621:
URL: https://github.com/apache/pinot/pull/18621
## Summary
Adds a merge-only reduction surface that performs the cross-server merge step
**without** finalizing — producing a single intermediate `DataTable` that
carries non-finalized aggregation state, byte-shape identical to a single
server's partial response. A downstream consumer can intercept this merged
intermediate, custom-handle it, and later re-inject it through the standard
reduce path for finalization. The read path is unchanged; tests verify a
re-injected intermediate finalizes to the same result as a direct reduce.
No caller is wired in by this change — this is the API surface only.
## API
```java
// DataTableReducer (new default method, throws by default)
default DataTable mergeDataTablesOnly(String tableName, DataSchema dataSchema,
Map<ServerRoutingInstance, DataTable> dataTableMap,
DataTableReducerContext reducerContext, BrokerMetrics brokerMetrics);
// BrokerReduceService (merge-only counterpart of reduceOnDataTable)
@Nullable
public DataTable mergeOnDataTable(BrokerRequest serverBrokerRequest,
Map<ServerRoutingInstance, DataTable> dataTableMap, long reduceTimeOutMs,
BrokerMetrics brokerMetrics);
```
### Supported reducers
- `AggregationDataTableReducer`, `GroupByDataTableReducer`,
`DistinctDataTableReducer` — implement `mergeDataTablesOnly`.
- `SelectionDataTableReducer`, `ExplainPlanDataTableReducer` — inherit the
default-throwing implementation (out of scope).
- Aggregation and group-by additionally throw `UnsupportedOperationException`
when `_queryContext.isServerReturnFinalResult()` is true — inputs are then
finalized rather than intermediate, so the contract can't be honored.
### Surfaced on the merged DataTable's metadata
- Group-by completeness: `GROUPS_TRIMMED`, `NUM_GROUPS_LIMIT_REACHED`.
- New `PARTIAL_INTERMEDIATE_RESULT` flag — set when one or more input
DataTables are dropped during merge (e.g. schema conflict). The
`BrokerMeter.RESPONSE_MERGE_EXCEPTIONS` meter is also bumped, symmetric
with how the regular reduce path surfaces conflicting-schema servers.
- Execution stats round-trip via the new
`ExecutionStatsAggregator#setStatsOnMergedDataTable`: additive longs
(`numDocsScanned`, `numSegments*`, `threadCpuTimeNs`, ...), MIN-reduced
`minConsumingFreshnessTimeMs`, boolean flags, per-server exceptions, and
JSON-encoded trace info (when `trace=true`). Method does **not** bump
broker meters/timers (the merge path is expected to run off the
request-serving path; metric increments fire on eventual re-reduce).
### Limitations of the round-trip
- CPU/mem stats lose the offline-vs-realtime split visible on
`BrokerResponseNative` — wire format has only combined keys, so a
re-reduce dumps the whole value into one bucket based on the synthetic
server's tableType.
- Exception attribution to original servers is lost; collisions on the
same error code are last-write-wins (wire format is `Map<Integer, String>`).
- Per-server trace info is JSON-encoded into one `TRACE_INFO` entry; the
downstream aggregator reads it back as one trace blob under the synthetic
server's name.
## Server-side consolidations
Two helpers extracted to avoid duplication between the existing server-side
serialization and the new broker-side merge-only path:
- `AggregationFunctionUtils.setIntermediateResult` (now also used by
`AggregationResultsBlock`).
- `DataTableBuilderUtils.setColumn` (now also used by `GroupByResultsBlock`).
These are behavior-preserving — private methods promoted to shared statics
with identical bodies.
## Wire format
Adds one new `DataTable.MetadataKey` entry (`PARTIAL_INTERMEDIATE_RESULT`,
id=43). Older readers ignore unknown metadata keys, so this is
forward-compatible.
## Test plan
- [x] `MergeDataTablesOnlyTest` (23 tests): round-trip equivalence for
Aggregation/GroupBy/Distinct (including OBJECT-column DISTINCTCOUNT),
`LIMIT 0` group-by round-trip, intermediate-schema preservation,
conflicting-schema partial-merge flag, `serverReturnFinalResult`
rejection, additive stats end-to-end (`numDocsScanned` survives
round-trip), MIN freshness reduction, exception/trace propagation,
null-handling round-trip, selection-throws, empty/no-input null returns.
- [x] `./mvnw -pl pinot-core test-compile`, spotless, checkstyle, license all clean.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "navina (via GitHub)" <gi...@apache.org>.
navina commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3343591503
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
Review Comment:
discussed offline that we will stick to this approach for now and improve later on, if needed.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "navina (via GitHub)" <gi...@apache.org>.
navina commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3331895671
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
Review Comment:
> Is this dropping EARLY_TERMINATION_REASON for distinct queries that was added recently?
I wasn't aware of the recent change. I believe @xiangfu0 made the same comment. I have fixed it. Please check.
> nother option is to collapse the two paths (this and setStats) into one by refactoring it into a pattern like forEachStat((MetadataKey key, long value) -> …)
I considered this kind of approach. But we are updating 2 different types and that is just unnecessarily introducing a new type with some 20 odd methods. Not sure if it is worth the complexity.
Alternative is to introduce another type that encapsulates the merged Datatable and the agrgegated stats.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "navina (via GitHub)" <gi...@apache.org>.
navina commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3331895671
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
Review Comment:
> Is this dropping EARLY_TERMINATION_REASON for distinct queries that was added recently?
I wasn't aware of the recent change. I believe @xiangfu0 made the same comment. I have fixed it. Please check.
> nother option is to collapse the two paths (this and setStats) into one by refactoring it into a pattern like forEachStat((MetadataKey key, long value) -> …)
I considered this kind of approach. But we are updating 2 different types and that is just unnecessarily introducing a new type with some 20 odd methods. Not sure if it is worth the complexity.
Alternative is to introduce another type that encapsulates the merged Datatable and the agrgegated stats.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3327048559
##########
pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTable.java:
##########
@@ -156,11 +156,15 @@ enum MetadataKey {
WORKLOAD_NAME(40, "workloadName", MetadataValueType.STRING),
// Needed so that we can track query id in Netty channel response.
QUERY_ID(41, "queryId", MetadataValueType.STRING),
- EARLY_TERMINATION_REASON(42, "earlyTerminationReason", MetadataValueType.STRING);
+ EARLY_TERMINATION_REASON(42, "earlyTerminationReason", MetadataValueType.STRING),
+ // Set on a merged-only DataTable when one or more input server DataTables were dropped during the
+ // merge (e.g., due to a schema conflict). Signals to a downstream consumer that the merge is
+ // partial; how to react (skip, retry, accept with annotation) is the consumer's policy.
+ PARTIAL_INTERMEDIATE_RESULT(43, "partialIntermediateResult", MetadataValueType.STRING);
Review Comment:
This looks a bit odd, aren't all the intermediate results partial? Should it be something like `INCOMPLETE_MERGE` instead?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
Review Comment:
Is this dropping `EARLY_TERMINATION_REASON` for distinct queries that was added recently?
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
Review Comment:
This is also going to be prone to becoming stale in the future - I'd prefer avoiding this pattern altogether. Maybe we can just carry the execution stats aggregate alongside the data table? Another option is to collapse the two paths (this and setStats) into one by refactoring it into a pattern like `forEachStat((MetadataKey key, long value) -> …)` that both consumers drive? One writes to the response, the other to metadata.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3330194082
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
+ Map<String, String> metadata = dataTable.getMetadata();
+
+ // Additive long stats: mirror setStats()'s pattern of unconditional writes. Accumulators are
+ // initialized to 0, so a 0 here is indistinguishable to downstream from "absent" — the
+ // downstream aggregator's Long.parseLong("0") + null-check both produce 0.
+ putLong(metadata, DataTable.MetadataKey.NUM_DOCS_SCANNED, _numDocsScanned);
+ putLong(metadata, DataTable.MetadataKey.NUM_ENTRIES_SCANNED_IN_FILTER, _numEntriesScannedInFilter);
+ putLong(metadata, DataTable.MetadataKey.NUM_ENTRIES_SCANNED_POST_FILTER, _numEntriesScannedPostFilter);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_QUERIED, _numSegmentsQueried);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PROCESSED, _numSegmentsProcessed);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_MATCHED, _numSegmentsMatched);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_QUERIED, _numConsumingSegmentsQueried);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED, _numConsumingSegmentsProcessed);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED, _numConsumingSegmentsMatched);
+ putLong(metadata, DataTable.MetadataKey.TOTAL_DOCS, _numTotalDocs);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_SERVER, _numSegmentsPrunedByServer);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_INVALID, _numSegmentsPrunedInvalid);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT, _numSegmentsPrunedByLimit);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE, _numSegmentsPrunedByValue);
+ putLong(metadata, DataTable.MetadataKey.EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS,
+ _explainPlanNumEmptyFilterSegments);
+ putLong(metadata, DataTable.MetadataKey.EXPLAIN_PLAN_NUM_MATCH_ALL_FILTER_SEGMENTS,
+ _explainPlanNumMatchAllFilterSegments);
+ // Collapse offline+realtime decomposition back to the combined wire-format keys.
+ putLong(metadata, DataTable.MetadataKey.THREAD_CPU_TIME_NS,
+ _offlineThreadCpuTimeNs + _realtimeThreadCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS,
+ _offlineSystemActivitiesCpuTimeNs + _realtimeSystemActivitiesCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.RESPONSE_SER_CPU_TIME_NS,
+ _offlineResponseSerializationCpuTimeNs + _realtimeResponseSerializationCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.THREAD_MEM_ALLOCATED_BYTES,
+ _offlineThreadMemAllocatedBytes + _realtimeThreadMemAllocatedBytes);
+ putLong(metadata, DataTable.MetadataKey.RESPONSE_SER_MEM_ALLOCATED_BYTES,
+ _offlineResponseSerMemAllocatedBytes + _realtimeResponseSerMemAllocatedBytes);
+
+ // MIN_CONSUMING_FRESHNESS_TIME_MS: sentinel-guarded. Long.MAX_VALUE means "no input had a real
+ // freshness reading"; writing the sentinel would mislead downstream observability.
+ if (_minConsumingFreshnessTimeMs != Long.MAX_VALUE) {
+ metadata.put(DataTable.MetadataKey.MIN_CONSUMING_FRESHNESS_TIME_MS.getName(),
+ Long.toString(_minConsumingFreshnessTimeMs));
+ }
+
+ // Boolean flags: OR-reduced; only write the key when true (a "false" entry is noise and the
+ // existing reduce path treats absent as false).
+ if (_groupsTrimmed) {
Review Comment:
This round-trip drops DISTINCT early-termination state. `aggregate()` sets `_maxRowsInDistinctReached`, `_maxRowsWithoutChangeInDistinctReached`, and `_maxExecutionTimeInDistinctReached` from `EARLY_TERMINATION_REASON`, but `setStatsOnMergedDataTable()` never writes that metadata back onto the merged DataTable. After reinjection, `reduceOnDataTable()` will treat the merged DISTINCT response as complete and clear those partial-result flags. Please serialize the equivalent `EARLY_TERMINATION_REASON` here as well.
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya merged PR #18621:
URL: https://github.com/apache/pinot/pull/18621
--
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 merge-only reducer API for intercepting intermediate results [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18621:
URL: https://github.com/apache/pinot/pull/18621#issuecomment-4568627652
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18621?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 `63.60656%` with `111 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.84%. Comparing base ([`4ecb1d4`](https://app.codecov.io/gh/apache/pinot/commit/4ecb1d4ad88c638083258a6d4b27f92981777c34?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0a6c99d`](https://app.codecov.io/gh/apache/pinot/commit/0a6c99d897506fae1a67041f4c6112f634407f64?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 5 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18621?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...not/core/query/reduce/GroupByDataTableReducer.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Freduce%2FGroupByDataTableReducer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9yZWR1Y2UvR3JvdXBCeURhdGFUYWJsZVJlZHVjZXIuamF2YQ==) | 41.79% | [34 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...t/core/common/datatable/DataTableBuilderUtils.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fcommon%2Fdatatable%2FDataTableBuilderUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9jb21tb24vZGF0YXRhYmxlL0RhdGFUYWJsZUJ1aWxkZXJVdGlscy5qYXZh) | 28.26% | [32 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...core/query/reduce/AggregationDataTableReducer.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Freduce%2FAggregationDataTableReducer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9yZWR1Y2UvQWdncmVnYXRpb25EYXRhVGFibGVSZWR1Y2VyLmphdmE=) | 74.50% | [9 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...aggregation/function/AggregationFunctionUtils.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Faggregation%2Ffunction%2FAggregationFunctionUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9hZ2dyZWdhdGlvbi9mdW5jdGlvbi9BZ2dyZWdhdGlvbkZ1bmN0aW9uVXRpbHMuamF2YQ==) | 52.94% | [7 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...e/pinot/core/query/reduce/BrokerReduceService.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Freduce%2FBrokerReduceService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9yZWR1Y2UvQnJva2VyUmVkdWNlU2VydmljZS5qYXZh) | 88.52% | [0 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/core/query/reduce/ExecutionStatsAggregator.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Freduce%2FExecutionStatsAggregator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9yZWR1Y2UvRXhlY3V0aW9uU3RhdHNBZ2dyZWdhdG9yLmphdmE=) | 83.33% | [4 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/core/query/reduce/DistinctDataTableReducer.java](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fquery%2Freduce%2FDistinctDataTableReducer.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9xdWVyeS9yZWR1Y2UvRGlzdGluY3REYXRhVGFibGVSZWR1Y2VyLmphdmE=) | 66.66% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18621 +/- ##
============================================
+ Coverage 56.79% 56.84% +0.04%
+ Complexity 7 1 -6
============================================
Files 2567 2567
Lines 149143 149291 +148
Branches 24111 24135 +24
============================================
+ Hits 84707 84861 +154
+ Misses 57241 57240 -1
+ Partials 7195 7190 -5
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18621/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/18621/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/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <63.60%> (+0.04%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <63.60%> (+0.04%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <63.60%> (+0.04%)` | :arrow_up: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18621/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <63.60%> (+0.04%)` | :arrow_up: |
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 Sentry](https://app.codecov.io/gh/apache/pinot/pull/18621?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] Add merge-only reducer API for intercepting intermediate results [pinot]
Posted by "navina (via GitHub)" <gi...@apache.org>.
navina commented on code in PR #18621:
URL: https://github.com/apache/pinot/pull/18621#discussion_r3331912543
##########
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/ExecutionStatsAggregator.java:
##########
@@ -346,4 +348,107 @@ private void withNotNullLongMetadata(Map<String, String> metadata, DataTable.Met
consumer.accept(Long.parseLong(strValue));
}
}
+
+ /**
+ * Writes the accumulated execution stats onto the given DataTable's metadata (and exception map),
+ * so a merged-only DataTable can be re-injected into the regular reduce path with the same
+ * downstream totals as a direct reduce of the original inputs would have produced.
+ *
+ * <p>Unlike {@link #setStats(String, BrokerResponseNative, BrokerMetrics)}, this method does NOT
+ * bump broker meters or timers. The merge-only path is expected to run off the request-serving
+ * path; meter increments fire when the result is eventually re-reduced.
+ *
+ * <p>Limitations of the round-trip via DataTable metadata:
+ * <ul>
+ * <li>CPU and memory stats round-trip as a single combined value per key
+ * ({@link DataTable.MetadataKey#THREAD_CPU_TIME_NS}, etc.) because the wire format has no
+ * per-tableType keys. In the standard reduce path the aggregator attributes each server's
+ * value to offline vs realtime based on {@code routingInstance.getTableType()} and surfaces
+ * them as separate fields on {@link BrokerResponseNative}; on a re-reduce of the merged
+ * DataTable the whole combined value lands in one bucket — whichever tableType the caller
+ * assigned to the synthetic server response. So the per-tableType split visible on
+ * BrokerResponse is lost across the round-trip, even though the total is preserved.
+ * <li>Per-server exceptions are written via {@link DataTable#addException(int, String)} which
+ * backs a {@code Map<Integer, String>} keyed by error code; if two inputs reported the
+ * same error code the merged DataTable carries last-write-wins for the message.
+ * <li>Per-server trace info is JSON-encoded into a single
+ * {@link DataTable.MetadataKey#TRACE_INFO} entry; the downstream aggregator reads it back
+ * as one trace blob attributed to the synthetic server.
+ * </ul>
+ */
+ public void setStatsOnMergedDataTable(DataTable dataTable) {
+ Map<String, String> metadata = dataTable.getMetadata();
+
+ // Additive long stats: mirror setStats()'s pattern of unconditional writes. Accumulators are
+ // initialized to 0, so a 0 here is indistinguishable to downstream from "absent" — the
+ // downstream aggregator's Long.parseLong("0") + null-check both produce 0.
+ putLong(metadata, DataTable.MetadataKey.NUM_DOCS_SCANNED, _numDocsScanned);
+ putLong(metadata, DataTable.MetadataKey.NUM_ENTRIES_SCANNED_IN_FILTER, _numEntriesScannedInFilter);
+ putLong(metadata, DataTable.MetadataKey.NUM_ENTRIES_SCANNED_POST_FILTER, _numEntriesScannedPostFilter);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_QUERIED, _numSegmentsQueried);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PROCESSED, _numSegmentsProcessed);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_MATCHED, _numSegmentsMatched);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_QUERIED, _numConsumingSegmentsQueried);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_PROCESSED, _numConsumingSegmentsProcessed);
+ putLong(metadata, DataTable.MetadataKey.NUM_CONSUMING_SEGMENTS_MATCHED, _numConsumingSegmentsMatched);
+ putLong(metadata, DataTable.MetadataKey.TOTAL_DOCS, _numTotalDocs);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_SERVER, _numSegmentsPrunedByServer);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_INVALID, _numSegmentsPrunedInvalid);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_LIMIT, _numSegmentsPrunedByLimit);
+ putLong(metadata, DataTable.MetadataKey.NUM_SEGMENTS_PRUNED_BY_VALUE, _numSegmentsPrunedByValue);
+ putLong(metadata, DataTable.MetadataKey.EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS,
+ _explainPlanNumEmptyFilterSegments);
+ putLong(metadata, DataTable.MetadataKey.EXPLAIN_PLAN_NUM_MATCH_ALL_FILTER_SEGMENTS,
+ _explainPlanNumMatchAllFilterSegments);
+ // Collapse offline+realtime decomposition back to the combined wire-format keys.
+ putLong(metadata, DataTable.MetadataKey.THREAD_CPU_TIME_NS,
+ _offlineThreadCpuTimeNs + _realtimeThreadCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.SYSTEM_ACTIVITIES_CPU_TIME_NS,
+ _offlineSystemActivitiesCpuTimeNs + _realtimeSystemActivitiesCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.RESPONSE_SER_CPU_TIME_NS,
+ _offlineResponseSerializationCpuTimeNs + _realtimeResponseSerializationCpuTimeNs);
+ putLong(metadata, DataTable.MetadataKey.THREAD_MEM_ALLOCATED_BYTES,
+ _offlineThreadMemAllocatedBytes + _realtimeThreadMemAllocatedBytes);
+ putLong(metadata, DataTable.MetadataKey.RESPONSE_SER_MEM_ALLOCATED_BYTES,
+ _offlineResponseSerMemAllocatedBytes + _realtimeResponseSerMemAllocatedBytes);
+
+ // MIN_CONSUMING_FRESHNESS_TIME_MS: sentinel-guarded. Long.MAX_VALUE means "no input had a real
+ // freshness reading"; writing the sentinel would mislead downstream observability.
+ if (_minConsumingFreshnessTimeMs != Long.MAX_VALUE) {
+ metadata.put(DataTable.MetadataKey.MIN_CONSUMING_FRESHNESS_TIME_MS.getName(),
+ Long.toString(_minConsumingFreshnessTimeMs));
+ }
+
+ // Boolean flags: OR-reduced; only write the key when true (a "false" entry is noise and the
+ // existing reduce path treats absent as false).
+ if (_groupsTrimmed) {
Review Comment:
I have addressed it. PTAL
--
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