gortiz commented on code in PR #19883:
URL: https://github.com/apache/pinot/pull/19883#discussion_r4293206133
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java:
##########
@@ -2464,11 +2474,97 @@ static void validateRequest(PinotQuery pinotQuery, int
queryResponseLimit) {
"Value for 'LIMIT' (" + limit + ") exceeds maximum allowed value of
" + queryResponseLimit);
}
QueryOptionsUtils.getNumReplicaGroupsToQuery(pinotQuery.getQueryOptions());
+ validateExprMinMax(pinotQuery);
if (pinotQuery.getDataSource().getSubquery() != null) {
validateRequest(pinotQuery.getDataSource().getSubquery(),
queryResponseLimit);
}
}
+ /// Servers cannot execute EXPR_MIN/EXPR_MAX directly. The broker must
rewrite them into parent and child
+ /// aggregations with [ExprMinMaxRewriter], and merge the parent result back
into the child columns with
+ /// [ParentAggregationResultRewriter]. Neither rewriter is enabled by
default.
+ ///
+ /// Without this check, a query that still has EXPR_MIN/EXPR_MAX fails on
the servers, and a query without the
+ /// result rewriter returns the placeholder values of the child aggregations
instead of the expected values.
+ @VisibleForTesting
+ static void validateExprMinMax(PinotQuery pinotQuery) {
+ String exprMinMax = findFunction(pinotQuery, EXPR_MIN_MAX_FUNCTIONS);
+ if (exprMinMax != null) {
+ String functionName = exprMinMax.equals(EXPR_MIN) ? "EXPR_MIN" :
"EXPR_MAX";
+ boolean queryRewriterEnabled =
+
CalciteSqlParser.QUERY_REWRITERS.stream().anyMatch(ExprMinMaxRewriter.class::isInstance);
+ if (!queryRewriterEnabled) {
+ throw new BadQueryRequestException(
+ functionName + " is not enabled on this broker. To enable it, add
'" + ExprMinMaxRewriter.class.getName()
+ + "' to '" +
Broker.CONFIG_OF_BROKER_QUERY_REWRITER_CLASS_NAMES + "' and '"
+ + ParentAggregationResultRewriter.class.getName() + "' to '"
+ + Broker.CONFIG_OF_BROKER_RESULT_REWRITER_CLASS_NAMES + "'");
+ }
+ // The rewriter only handles the top-level expressions of the select
list, so the function is nested in another
+ // expression or has an alias.
+ throw new BadQueryRequestException(
+ functionName + " is only supported as a top-level expression in the
select list, without an alias");
Review Comment:
Good point. In 20b7b64 the message now says that other expressions and the
WHERE, GROUP BY, HAVING and ORDER BY clauses aren't supported either, so a
query like `WHERE EXPR_MIN(a, b) > 5` gets an accurate error.
##########
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:
Agreed, that was fragile. In 20b7b64 an `@AfterMethod` resets both factories
to their defaults after every test, and the per-test `finally` blocks are gone.
Now no test depends on the order of the others.
##########
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:
Yes, that shift is intended: `REQUEST_DESERIALIZATION_EXCEPTIONS` now counts
only Thrift failures, and a compile failure is a query execution error. I added
a release notes section to the PR description and the `release-notes` label so
operators can update their alerts.
##########
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:
Fixed in 20b7b64. The test now asserts the `AggregationFunctionFactory`
message ("Aggregation function: EXPRMIN is only supported in selection without
alias"), so it only passes for the unsupported-function failure.
##########
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:
Added in 20b7b64: WHERE, HAVING and ORDER BY cases, next to the alias and
nested cases, all with the full message.
--
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]
