This is an automated email from the ASF dual-hosted git repository.
xiangfu 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 1772b2f527 Make v2 bytes value displayed same as v1 hex string (#10559)
1772b2f527 is described below
commit 1772b2f527158e8979ee1a23f504ab0c6456374e
Author: Xiang Fu <[email protected]>
AuthorDate: Thu Apr 6 11:41:03 2023 -0700
Make v2 bytes value displayed same as v1 hex string (#10559)
---
.../query/service/dispatch/QueryDispatcher.java | 21 +++++++++++++--------
.../pinot/query/runtime/QueryRunnerTestBase.java | 4 ++++
2 files changed, 17 insertions(+), 8 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
index 36c93a5827..43345b3b08 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java
@@ -62,6 +62,7 @@ import
org.apache.pinot.query.runtime.plan.DistributedStagePlan;
import org.apache.pinot.query.runtime.plan.OpChainExecutionContext;
import org.apache.pinot.query.runtime.plan.serde.QueryPlanSerDeUtils;
import org.apache.pinot.query.service.QueryConfig;
+import org.apache.pinot.spi.utils.ByteArray;
import org.roaringbitmap.RoaringBitmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -138,7 +139,7 @@ public class QueryDispatcher {
dispatchCalls++;
_executorService.submit(() -> {
client.submit(Worker.QueryRequest.newBuilder().setStagePlan(
-
QueryPlanSerDeUtils.serialize(constructDistributedStagePlan(queryPlan, stageId,
serverInstance)))
+
QueryPlanSerDeUtils.serialize(constructDistributedStagePlan(queryPlan, stageId,
serverInstance)))
.putMetadata(QueryConfig.KEY_OF_BROKER_REQUEST_ID,
String.valueOf(requestId))
.putMetadata(QueryConfig.KEY_OF_BROKER_REQUEST_TIMEOUT_MS,
String.valueOf(timeoutMs))
.putAllMetadata(queryOptions).build(), stageId,
serverInstance, deadline, dispatchCallbacks::offer);
@@ -149,12 +150,13 @@ public class QueryDispatcher {
int successfulDispatchCalls = 0;
// TODO: Cancel all dispatched requests if one of the dispatch errors out
or deadline is breached.
while (!deadline.isExpired() && successfulDispatchCalls < dispatchCalls) {
- AsyncQueryDispatchResponse resp = dispatchCallbacks.poll(
- DEFAULT_DISPATCHER_CALLBACK_POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ AsyncQueryDispatchResponse resp =
+ dispatchCallbacks.poll(DEFAULT_DISPATCHER_CALLBACK_POLL_TIMEOUT_MS,
TimeUnit.MILLISECONDS);
if (resp != null) {
if (resp.getThrowable() != null) {
- throw new RuntimeException(String.format("Error dispatching query to
server=%s stage=%s",
- resp.getVirtualServer(), resp.getStageId()),
resp.getThrowable());
+ throw new RuntimeException(
+ String.format("Error dispatching query to server=%s stage=%s",
resp.getVirtualServer(),
+ resp.getStageId()), resp.getThrowable());
} else {
Worker.QueryResponse response = resp.getQueryResponse();
if
(response.containsMetadata(QueryConfig.KEY_OF_SERVER_RESPONSE_STATUS_ERROR)) {
@@ -182,8 +184,7 @@ public class QueryDispatcher {
new OpChainExecutionContext(mailboxService, requestId, reduceStageId,
server, timeoutMs, timeoutMs,
queryPlan.getStageMetadataMap());
MailboxReceiveOperator mailboxReceiveOperator =
- createReduceStageOperator(
- reduceNode.getSenderStageId(), reduceStageId,
reduceNode.getDataSchema(), context);
+ createReduceStageOperator(reduceNode.getSenderStageId(),
reduceStageId, reduceNode.getDataSchema(), context);
List<DataBlock> resultDataBlocks =
reduceMailboxReceive(mailboxReceiveOperator, timeoutMs,
statsAggregatorMap, queryPlan, context.getStats());
return toResultTable(resultDataBlocks, queryPlan.getQueryResultFields(),
@@ -260,7 +261,11 @@ public class QueryDispatcher {
row[colId++] = null;
} else {
int colRef = field.left;
- row[colId++] = rawRow[colRef];
+ if (rawRow[colRef] instanceof ByteArray) {
+ row[colId++] = ((ByteArray) rawRow[colRef]).toHexString();
+ } else {
+ row[colId++] = rawRow[colRef];
+ }
}
}
rows.add(row);
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/QueryRunnerTestBase.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/QueryRunnerTestBase.java
index 1650938d67..3bd3156ce6 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/QueryRunnerTestBase.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/QueryRunnerTestBase.java
@@ -59,6 +59,7 @@ import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.BytesUtils;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.StringUtil;
import org.h2.jdbc.JdbcArray;
@@ -181,6 +182,9 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
}
return Double.compare(ld, rd);
} else if (l instanceof String) {
+ if (r instanceof byte[]) {
+ return ((String) l).compareTo(BytesUtils.toHexString((byte[]) r));
+ }
return ((String) l).compareTo((String) r);
} else if (l instanceof Boolean) {
return ((Boolean) l).compareTo((Boolean) r);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
