This is an automated email from the ASF dual-hosted git repository.
gortiz 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 b6cd43b8d59 Add a generic response-metadata extension point to
BrokerResponse (#19019)
b6cd43b8d59 is described below
commit b6cd43b8d597ce24ac18c836d07611e7fdc7036d
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Tue Aug 18 08:26:07 2026 +0200
Add a generic response-metadata extension point to BrokerResponse (#19019)
---
.../MultiStageBrokerRequestHandler.java | 8 ++++
.../pinot/common/response/BrokerResponse.java | 49 ++++++++++++++++++++++
.../response/broker/BrokerResponseNativeV2.java | 23 +++++++++-
.../response/broker/BrokerResponseNativeTest.java | 17 ++++++++
.../broker/BrokerResponseNativeV2Test.java | 31 ++++++++++++++
.../pinot/spi/query/QueryExecutionContext.java | 44 +++++++++++++++++++
.../apache/pinot/spi/query/QueryThreadContext.java | 31 ++++++++++++++
7 files changed, 202 insertions(+), 1 deletion(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 34562d457fb..07a9497b368 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -881,6 +881,14 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
}
brokerResponse.setTimeUsedMs(totalTimeMs);
+ // Surface any generic response metadata registered during query
handling (e.g. a
+ // degraded-engine note) into the response. Entries are registered via
+ // QueryThreadContext#addResponseBrokerMetadata; the core engine
attaches no semantics to them.
+ // The sink is broker-local for now, so this only picks up entries
registered on the broker.
+ QueryThreadContext queryThreadContext =
QueryThreadContext.getIfAvailable();
+ if (queryThreadContext != null) {
+
queryThreadContext.getExecutionContext().getResponseMetadata().forEach(brokerResponse::putResponseMetadata);
+ }
augmentStatistics(requestContext, brokerResponse);
if (QueryOptionsUtils.shouldDropResults(query.getOptions())) {
brokerResponse.setResultTable(null);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
index 0b1eb3ed068..83e2d945dfc 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
@@ -19,7 +19,9 @@
package org.apache.pinot.common.response;
import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
import java.io.IOException;
import java.io.OutputStream;
import java.util.List;
@@ -320,4 +322,51 @@ public interface BrokerResponse {
default String getMaterializedViewQueried() {
return null;
}
+
+ /// Returns generic, product-agnostic response metadata: a free-form
string-to-[JsonNode] map that
+ /// any component can populate to surface non-fatal, informational notes
about how the query was
+ /// handled (for example that it was executed with an alternate or degraded
strategy). Values are
+ /// arbitrary JSON, so a note can be a scalar, an object, or an array. This
is intentionally a
+ /// generic extension point: the core engine attaches no semantics to the
keys or values, so
+ /// extensions can add their own entries without a dedicated typed field on
this interface.
+ ///
+ /// This is distinct from [#getTraceInfo()] (per-server trace strings, only
populated when tracing
+ /// is enabled) and from [#getExceptions()] (the error/warning list). The
default is an empty,
+ /// unmodifiable map for implementations that do not support response
metadata.
+ ///
+ /// Today entries can only be registered **on the broker**: they are
collected in the broker-side
+ /// `QueryExecutionContext` while the query is being handled, and that sink
is never sent over the
+ /// wire — an entry registered by a worker (server) on its own copy of the
execution context is
+ /// silently dropped. So in practice this map currently carries broker
metadata only.
+ ///
+ /// That is a limitation of the current implementation, not of this
contract: the plumbing may
+ /// later be extended so that workers can contribute entries too, propagated
back to the broker
+ /// with the rest of the per-server metadata. Hence the deliberately generic
name — where an entry
+ /// was produced is an internal detail, and surfacing it in the
client-facing response as a
+ /// `brokerMetadata` field that would eventually need a sibling
`serverMetadata` field would only
+ /// make the response more complex for no benefit to the user. Worker
entries, when supported, will
+ /// go into this same map.
+ ///
+ /// Marked [JsonIgnore] on the interface default so it does not register
`responseMetadata` as a
+ /// known (setterless) property on deserializable implementations such as
`BrokerResponseNative`
+ /// that do not override it. Otherwise Jackson would try to populate the
immutable [Map#of()]
+ /// returned here via USE_GETTERS_AS_SETTERS and fail with an
[UnsupportedOperationException] when
+ /// a legacy response carries a non-empty `responseMetadata`. Concrete
implementations that support
+ /// the field re-expose it by overriding this method with an explicit
[JsonProperty].
+ @JsonIgnore
+ default Map<String, JsonNode> getResponseMetadata() {
+ return Map.of();
+ }
+
+ /// Records a generic response-metadata entry (arbitrary JSON value; see
[#getResponseMetadata()]).
+ /// The default is a no-op, so implementations that do not support response
metadata silently
+ /// ignore it.
+ default void putResponseMetadata(String key, JsonNode value) {
+ }
+
+ /// String convenience for [#putResponseMetadata(String, JsonNode)] — the
common case — wrapping the
+ /// value in a JSON string node.
+ default void putResponseMetadata(String key, String value) {
+ putResponseMetadata(key, TextNode.valueOf(value));
+ }
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
index a4190faee41..efbb57ead5a 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
@@ -18,12 +18,15 @@
*/
package org.apache.pinot.common.response.broker;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -51,7 +54,8 @@ import org.apache.pinot.common.response.ProcessingException;
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
"realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
"pools", "rlsFiltersApplied", "groupsTrimmed",
- "mseLiteLeafStageLimitReached", "mseLiteLeafStageEffectiveLimit",
"mseLiteFanOutAdjustedLimitApplied"
+ "mseLiteLeafStageLimitReached", "mseLiteLeafStageEffectiveLimit",
"mseLiteFanOutAdjustedLimitApplied",
+ "responseMetadata"
})
public class BrokerResponseNativeV2 implements BrokerResponse {
private final StatMap<StatKey> _brokerStats = new StatMap<>(StatKey.class);
@@ -91,6 +95,7 @@ public class BrokerResponseNativeV2 implements BrokerResponse
{
private Integer _mseLiteLeafStageEffectiveLimit;
@Nullable
private Boolean _mseLiteFanOutAdjustedLimitApplied;
+ private final Map<String, JsonNode> _responseMetadata = new HashMap<>();
@JsonInclude(JsonInclude.Include.NON_NULL)
@Nullable
@@ -477,6 +482,22 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
return Map.of();
}
+ // JsonIgnore(false) re-enables the property here: the interface default
getter is @JsonIgnore
+ // (so it does not register responseMetadata as a known setterless property
on legacy impls that
+ // don't override it), and that ignore would otherwise be inherited by this
override.
+ @JsonIgnore(false)
+ @JsonProperty("responseMetadata")
+ @JsonInclude(JsonInclude.Include.NON_EMPTY)
+ @Override
+ public Map<String, JsonNode> getResponseMetadata() {
+ return _responseMetadata;
+ }
+
+ @Override
+ public void putResponseMetadata(String key, JsonNode value) {
+ _responseMetadata.put(key, value);
+ }
+
@Override
public void setPools(Set<Integer> pools) {
_pools = pools;
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
index 5a5f3f2cc5d..7ede9c2932b 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeTest.java
@@ -113,4 +113,21 @@ public class BrokerResponseNativeTest {
public void testServerStatsDefaultsToNull() {
Assert.assertNull(new BrokerResponseNative().getServerStats());
}
+
+ /// Regression test for backward-compatible deserialization:
[BrokerResponseNative] does not
+ /// override `getResponseMetadata()`, so the interface default (marked
`@JsonIgnore`) must keep
+ /// `responseMetadata` an unknown property. Otherwise Jackson would treat it
as a known setterless
+ /// Map property and try to populate the immutable `Map.of()` default via
USE_GETTERS_AS_SETTERS,
+ /// failing with an [UnsupportedOperationException] when a response produced
by a newer broker
+ /// (e.g. `BrokerResponseNativeV2`) carries a non-empty `responseMetadata`.
+ @Test
+ public void testResponseMetadataDeserializationCompatibility()
+ throws IOException {
+ String json = "{\"responseMetadata\":{\"note\":\"executed with fallback
strategy\","
+ + "\"details\":{\"count\":2}},\"numDocsScanned\":5}";
+ BrokerResponseNative actual = BrokerResponseNative.fromJsonString(json);
+ // The unknown field is ignored on the legacy impl; other fields still
deserialize correctly.
+ Assert.assertEquals(actual.getNumDocsScanned(), 5);
+ Assert.assertTrue(actual.getResponseMetadata().isEmpty());
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
index 320db22557a..22cf653d6e5 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2Test.java
@@ -18,14 +18,19 @@
*/
package org.apache.pinot.common.response.broker;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.BooleanNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.apache.pinot.common.datatable.StatMap;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
@@ -47,6 +52,32 @@ public class BrokerResponseNativeV2Test {
assertTrue(brokerResponse.isPartialResult());
}
+ @Test
+ public void testResponseMetadataSerialization()
+ throws Exception {
+ BrokerResponseNativeV2 brokerResponse = new BrokerResponseNativeV2();
+ // Empty by default and omitted from JSON (NON_EMPTY).
+ assertTrue(brokerResponse.getResponseMetadata().isEmpty());
+ JsonNode emptyNode =
JsonUtils.stringToJsonNode(brokerResponse.toJsonString());
+ assertNull(emptyNode.get("responseMetadata"));
+
+ // A boolean via the JsonNode overload, a string via the convenience
overload, and a nested
+ // object to exercise arbitrary (complex) JSON values.
+ brokerResponse.putResponseMetadata("boolEntry", BooleanNode.getTrue());
+ brokerResponse.putResponseMetadata("stringEntry", "hello");
+ ObjectNode nested = JsonUtils.newObjectNode();
+ nested.put("count", 2);
+ nested.put("label", "example");
+ brokerResponse.putResponseMetadata("objectEntry", nested);
+
+ JsonNode node =
JsonUtils.stringToJsonNode(brokerResponse.toJsonString()).get("responseMetadata");
+ assertTrue(node.get("boolEntry").isBoolean());
+ assertTrue(node.get("boolEntry").asBoolean());
+ assertEquals(node.get("stringEntry").asText(), "hello");
+ assertEquals(node.get("objectEntry").get("count").asInt(), 2);
+ assertEquals(node.get("objectEntry").get("label").asText(), "example");
+ }
+
private static Set<String> stringSet(String... values) {
return new LinkedHashSet<>(List.of(values));
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryExecutionContext.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryExecutionContext.java
index 3309ceb0918..6692d5db2da 100644
---
a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryExecutionContext.java
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryExecutionContext.java
@@ -19,10 +19,13 @@
package org.apache.pinot.spi.query;
import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.TextNode;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nullable;
@@ -85,6 +88,27 @@ public class QueryExecutionContext {
/// Guards single-emission of the scan-based killing dry-run log line and
metric for this query
private final AtomicBoolean _scanKillingDryRunEmitted = new
AtomicBoolean(false);
+ /// Generic, product-agnostic response metadata registered during query
handling — a free-form
+ /// string-to-[JsonNode] map that any component can populate to surface an
informational note about
+ /// how the query was handled (for example that it was executed with an
alternate/degraded
+ /// strategy). Values are arbitrary JSON, so a note can be a scalar, an
object, or an array. The
+ /// broker copies these entries into the query response it sends back to the
client.
+ ///
+ /// This context instance is shared by reference across the query's
[QueryThreadContext]-aware
+ /// executors (e.g. the broker's async compile/plan threads re-open the
context with the same
+ /// instance), so a writer on any of those threads is visible to the
response-assembly thread.
+ /// Concurrent because those writes and the final read can happen on
different threads.
+ ///
+ /// This sink is **broker-local**: it is not part of the context state
serialized to workers, and
+ /// nothing propagates it back from a worker, so only entries registered
while running on the
+ /// broker reach the response. An entry registered on a server's copy of the
execution context is
+ /// silently dropped. This is a limitation of the current implementation
rather than a design
+ /// decision — the plumbing may later be extended so workers can contribute
entries as well. The
+ /// registration API records the restriction where it matters
([QueryThreadContext] exposes it as
+ /// `addResponseBrokerMetadata`); this sink and the response field it feeds
stay generic, so worker
+ /// entries can later be merged into the very same map.
+ private final Map<String, JsonNode> _responseMetadata = new
ConcurrentHashMap<>();
+
public QueryExecutionContext(QueryType queryType, long requestId, String
cid, String workloadName, long startTimeMs,
long activeDeadlineMs, long passiveDeadlineMs, String brokerId, String
instanceId, String queryHash) {
_queryType = queryType;
@@ -214,6 +238,26 @@ public class QueryExecutionContext {
return _terminateException;
}
+ /// Registers a generic response-metadata entry (arbitrary JSON value) to be
surfaced in the query
+ /// response. See [#getResponseMetadata()] — in particular, only entries
registered on the broker
+ /// currently reach the response, which is why the [QueryThreadContext]
entry point is named
+ /// [QueryThreadContext#addResponseBrokerMetadata]. Prefer that one from
code that does not already
+ /// hold this context.
+ public void addResponseMetadata(String key, JsonNode value) {
+ _responseMetadata.put(key, value);
+ }
+
+ /// String convenience for [#addResponseMetadata(String, JsonNode)] — the
common case — wrapping the
+ /// value in a JSON string node.
+ public void addResponseMetadata(String key, String value) {
+ _responseMetadata.put(key, TextNode.valueOf(value));
+ }
+
+ /// Returns the generic response metadata registered for this query (never
null; possibly empty).
+ public Map<String, JsonNode> getResponseMetadata() {
+ return _responseMetadata;
+ }
+
@Nullable
public QueryScanCostContext getQueryScanCostContext() {
return _queryScanCostContext;
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
index 1c6c4d0c8c5..52fca64440f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/query/QueryThreadContext.java
@@ -20,6 +20,7 @@ package org.apache.pinot.spi.query;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.annotations.VisibleForTesting;
import java.util.Set;
import java.util.concurrent.Callable;
@@ -227,6 +228,36 @@ public class QueryThreadContext implements AutoCloseable {
return THREAD_LOCAL.get();
}
+ /// Registers a generic response-metadata entry (arbitrary JSON value) on
the current query's
+ /// [QueryExecutionContext], to be surfaced in the query response by the
broker (see
+ /// [QueryExecutionContext#getResponseMetadata()]). This is the entry point
for code — including
+ /// product extensions — that wants to attach an informational note to the
response without a
+ /// dedicated typed field or knowledge of the response object.
+ ///
+ /// Must be called from **broker-side** code: the sink lives in the broker's
execution context and
+ /// is not propagated from workers, so an entry registered while running on
a server would be
+ /// silently dropped. The `Broker` in the name records that restriction,
which is a limit of the
+ /// current implementation rather than of the response field it feeds: when
worker-side metadata
+ /// becomes supported, this method can either be renamed or joined by an
`addResponseMetadata` that
+ /// servers may call too, with both feeding the same generic
`responseMetadata` response field.
+ ///
+ /// No-op when no [QueryThreadContext] is active on the current thread (e.g.
tests or planning paths
+ /// that run outside a query context), so callers never need a null check.
+ public static void addResponseBrokerMetadata(String key, JsonNode value) {
+ QueryThreadContext threadContext = getIfAvailable();
+ if (threadContext != null) {
+ threadContext.getExecutionContext().addResponseMetadata(key, value);
+ }
+ }
+
+ /// String convenience for [#addResponseBrokerMetadata(String, JsonNode)] —
the common case.
+ public static void addResponseBrokerMetadata(String key, String value) {
+ QueryThreadContext threadContext = getIfAvailable();
+ if (threadContext != null) {
+ threadContext.getExecutionContext().addResponseMetadata(key, value);
+ }
+ }
+
/// Returns a new [ExecutorService] whose tasks will be executed with the
[QueryThreadContext] initialized with the
/// state of the thread submitting the tasks. Tasks are registered for
cancellation when the query is terminated.
public static ExecutorService contextAwareExecutorService(ExecutorService
executorService) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
