xiangfu0 commented on code in PR #19883:
URL: https://github.com/apache/pinot/pull/19883#discussion_r4289852099
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -123,20 +124,31 @@ protected void channelRead0(ChannelHandlerContext ctx,
ByteBuf msg) {
_serverMetrics.addMeteredGlobalValue(ServerMeter.NETTY_CONNECTION_BYTES_RECEIVED,
requestSize);
byte[] requestBytes = new byte[requestSize];
msg.readBytes(requestBytes);
- ServerQueryRequest queryRequest;
+ InstanceRequest instanceRequest = new InstanceRequest();
try {
- InstanceRequest instanceRequest = new InstanceRequest();
THREAD_LOCAL_T_DESERIALIZER.get().deserialize(instanceRequest,
requestBytes);
- queryRequest = new ServerQueryRequest(instanceRequest, _serverMetrics,
queryArrivalTimeMs);
- _serverMetrics.addMeteredTableValue(queryRequest.getTableNameWithType(),
ServerMeter.QUERIES_ON_TABLE, 1);
-
queryRequest.getTimerContext().startNewPhaseTimer(ServerQueryPhase.REQUEST_DESERIALIZATION,
queryArrivalTimeMs)
- .stopAndRecord();
} catch (Exception e) {
// Do not send error response because request id is unknown.
LOGGER.error("Failed to deserialize instance request: {}",
BytesUtils.toHexString(requestBytes), e);
_serverMetrics.addMeteredGlobalValue(ServerMeter.REQUEST_DESERIALIZATION_EXCEPTIONS,
1);
return;
}
+ ServerQueryRequest queryRequest;
+ try {
+ queryRequest = new ServerQueryRequest(instanceRequest, _serverMetrics,
queryArrivalTimeMs);
+ } catch (Exception e) {
+ // The request id is known, so reply with the error. Without a response,
the broker waits until the query times
+ // out and reports the server as not responded, which hides the actual
error (e.g. an unsupported function).
+ long requestId = instanceRequest.getRequestId();
+ String tableNameWithType = getTableNameWithType(instanceRequest);
+ LOGGER.error("Failed to compile query request: {} on table: {}",
requestId, tableNameWithType, e);
+ sendErrorResponse(ctx, requestId, tableNameWithType, queryArrivalTimeMs,
Review Comment:
Minor: this changes metric semantics. Before, a failure in `new
ServerQueryRequest(...)` fell into the deserialization catch and counted as
`REQUEST_DESERIALIZATION_EXCEPTIONS`; now it sends a response and counts as
`QUERY_EXECUTION_EXCEPTIONS` (via `sendErrorResponse`'s finally). That's
arguably more correct, but it's a silent shift — worth a release note so
operators can adjust any alerting on those meters. Non-blocking.
##########
pinot-core/src/test/java/org/apache/pinot/core/transport/InstanceRequestHandlerTest.java:
##########
@@ -144,6 +156,39 @@ public void testWriteFailureClosesChannel()
verify(ctx).close();
}
+ @Test
+ public void testRepliesWithErrorWhenQueryCannotBeCompiled()
+ throws Exception {
+ PinotConfiguration config = new PinotConfiguration();
+ QueryScheduler queryScheduler = createQueryScheduler(config, new
CountDownLatch(0));
+ InstanceRequestHandler handler =
+ new InstanceRequestHandler("server01", config, queryScheduler,
mock(AccessControl.class),
+ ThreadAccountantUtils.getNoOpAccountant());
+
+ // EXPR_MIN cannot be executed without the broker-side ExprMinMaxRewriter,
so the server fails to build the
+ // aggregation function while it compiles the request.
+ BrokerRequest brokerRequest = new BrokerRequest();
+ brokerRequest.setPinotQuery(CalciteSqlParser.compileToPinotQuery("SELECT
EXPR_MIN(a, b) FROM testTable_OFFLINE"));
+ InstanceRequest instanceRequest = new InstanceRequest(123L, brokerRequest);
+ byte[] requestBytes = new TSerializer(new
TCompactProtocol.Factory()).serialize(instanceRequest);
+
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ when(ctx.writeAndFlush(any())).thenReturn(mock(ChannelFuture.class));
+ handler.channelRead0(ctx, Unpooled.wrappedBuffer(requestBytes));
+
+ ArgumentCaptor<ByteBuf> responseCaptor =
ArgumentCaptor.forClass(ByteBuf.class);
+ verify(ctx).writeAndFlush(responseCaptor.capture());
+ ByteBuf response = responseCaptor.getValue();
+ byte[] responseBytes = new byte[response.readableBytes()];
+ response.readBytes(responseBytes);
+ DataTable dataTable = DataTableFactory.getDataTable(responseBytes);
+
assertEquals(dataTable.getMetadata().get(MetadataKey.REQUEST_ID.getName()),
"123");
+ Map<Integer, String> exceptions = dataTable.getExceptions();
+ assertEquals(exceptions.size(), 1);
+ String message = exceptions.get(QueryErrorCode.QUERY_EXECUTION.getId());
+ assertTrue(message.contains("EXPRMIN"), message);
Review Comment:
Minor: `contains("EXPRMIN")` is a pretty loose check — it would still pass
on an unrelated error that happens to mention the name. Since you already
assert the request id and exception count, pinning a bit more of the expected
message (or at least that it's the unsupported-function failure) would make
this fail for the right reason. Non-blocking.
##########
pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryValidationTest.java:
##########
@@ -167,6 +171,51 @@ public void testReplicaGroupToQueryInvalidQuery() {
() -> BaseSingleStageBrokerRequestHandler.validateRequest(pinotQuery,
10));
}
+ @Test
+ public void testExprMinMaxWithoutRewriters() {
Review Comment:
Minor: `findFunction` also scans WHERE / GROUP BY / HAVING / ORDER BY, but
every new test case puts EXPR_MIN/EXPR_MAX in the select list. A case like `...
WHERE EXPR_MIN(a,b) > 1` or `... ORDER BY EXPR_MIN(a,b)` asserting the
rejection would guard those clause-scan branches against a future regression.
Non-blocking.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
