LakshSingla commented on code in PR #13952:
URL: https://github.com/apache/druid/pull/13952#discussion_r1193428989
##########
server/src/main/java/org/apache/druid/server/ClientQuerySegmentWalker.java:
##########
@@ -577,25 +627,80 @@ private static <T, QueryType extends Query<T>>
InlineDataSource toInlineDataSour
);
}
- final RowSignature signature = toolChest.resultArraySignature(query);
+ if (memoryLimitSet && memoryLimitAccumulator.get() >= memoryLimit) {
+ throw ResourceLimitExceededException.withMessage(
+ "Cannot issue subquery, maximum subquery result bytes[%d] reached",
+ memoryLimit
+ );
+ }
- final ArrayList<Object[]> resultList = new ArrayList<>();
+ DataSource dataSource;
+ // Try to serialize the results into a frame only if the memory limit is
set on the server or the query
+ if (memoryLimitSet) {
+ try {
+ Optional<Sequence<FrameSignaturePair>> framesOptional =
toolChest.resultsAsFrames(
+ query,
+ results,
+ memoryLimit - memoryLimitAccumulator.get()
+ );
- toolChest.resultsAsArrays(query, results).accumulate(
- resultList,
- (acc, in) -> {
- if (limitAccumulator.getAndIncrement() >= limitToUse) {
- throw ResourceLimitExceededException.withMessage(
- "Subquery generated results beyond maximum[%d]",
- limitToUse
- );
- }
- acc.add(in);
- return acc;
+ if (!framesOptional.isPresent()) {
+ throw new ISE("The memory of the subqueries cannot be estimated
correctly.");
}
- );
- return InlineDataSource.fromIterable(resultList, signature);
+ Sequence<FrameSignaturePair> frames = framesOptional.get();
+ List<FrameSignaturePair> frameSignaturePairs = new ArrayList<>();
+ frames.forEach(
+ frame -> {
+ if
(memoryLimitAccumulator.addAndGet(frame.getFrame().numBytes()) >= memoryLimit) {
+ throw ResourceLimitExceededException.withMessage(
+ "Subquery generated results beyond maximum[%d] bytes",
+ memoryLimit
+ );
+
+ }
+ if (limitAccumulator.addAndGet(frame.getFrame().numRows()) >=
limitToUse) {
+ throw ResourceLimitExceededException.withMessage(
+ "Subquery generated results beyond maximum[%d] rows",
+ limitToUse
+ );
+ }
+ frameSignaturePairs.add(frame);
+ }
+ );
+ dataSource = new FramesBackedInlineDataSource(frameSignaturePairs,
toolChest.resultArraySignature(query));
+ }
+ catch (ResourceLimitExceededException rlee) {
+ throw rlee;
+ }
+ catch (Exception e) {
+ log.info(
Review Comment:
This seems reasonable to me. I originally added this because of the fact
that columns with null types might exist and this seemed like a simpler and
lazier way to handle them. With the addition of the user flag
`useNestedForUnknownTypesInSubquery` mentioned in the [top-level comment
](https://github.com/apache/druid/pull/13952#pullrequestreview-1409574208)I
the error handling can seem a lot less convoluted since the user/dev has
control of the flag and what to do with it.
Also, the default `useNestedForUnknownTypesInSubquery` = false would make it
so that we are not breaking what exists currently, just limiting the queries
for which we can limit the subquery by bytes till the flag is set to `true`.
--
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]
