Skip to content

Fix druid connector bugs related to pushed down queries #14967

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhenxiao merged 2 commits into from
Aug 7, 2020
Merged

Fix druid connector bugs related to pushed down queries #14967

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e3733e8
Fix unhandled HTTP response error for druid client
weidongduan37 Jul 29, 2020
28da710
Support FilterNode on top of AggregationNode
weidongduan37 Jul 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions presto-druid/src/main/java/com/facebook/presto/druid/DruidClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,13 @@

import javax.inject.Inject;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.List;
import java.util.stream.Collectors;

import static com.facebook.airlift.http.client.HttpUriBuilder.uriBuilderFrom;
import static com.facebook.airlift.http.client.JsonResponseHandler.createJsonResponseHandler;
Expand All @@ -48,6 +51,7 @@
import static com.google.common.net.HttpHeaders.CONTENT_TYPE;
import static com.google.common.net.MediaType.JSON_UTF_8;
import static java.lang.String.format;
import static java.net.HttpURLConnection.HTTP_OK;
import static java.util.Objects.requireNonNull;

public class DruidClient
Expand Down Expand Up @@ -175,6 +179,10 @@ public InputStream handleException(Request request, Exception exception)
public InputStream handle(Request request, com.facebook.airlift.http.client.Response response)
{
try {
if (response.getStatusCode() != HTTP_OK) {
String result = new BufferedReader(new InputStreamReader(response.getInputStream())).lines().collect(Collectors.joining("\n"));
throw new PrestoException(DRUID_BROKER_RESULT_ERROR, result);
}
if (APPLICATION_JSON.equals(response.getHeader(CONTENT_TYPE))) {
return response.getInputStream();
}
Expand Down
6 changes: 5 additions & 1 deletion presto-druid/src/main/java/com/facebook/presto/druid/DruidQueryGeneratorContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static com.facebook.presto.druid.DruidErrorCode.DRUID_PUSHDOWN_UNSUPPORTED_EXPRESSION;
import static com.facebook.presto.druid.DruidErrorCode.DRUID_QUERY_GENERATOR_FAILURE;
import static com.facebook.presto.druid.DruidPushdownUtils.DRUID_COUNT_DISTINCT_FUNCTION_NAME;
import static com.google.common.base.MoreObjects.toStringHelper;
Expand Down Expand Up @@ -104,7 +105,10 @@ private DruidQueryGeneratorContext(

public DruidQueryGeneratorContext withFilter(String filter)
{
checkState(!hasFilter(), "Druid doesn't support filters at multiple levels");
if (hasAggregation()) {
throw new PrestoException(DRUID_PUSHDOWN_UNSUPPORTED_EXPRESSION, "Druid does not support filter on top of AggregationNode.");
Copy link
Copy Markdown
Collaborator

@zhenxiao zhenxiao Aug 5, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think better not through PrestoException here. Could we do:
checkState(!hasAggregation && !hasFilter, ...)

Copy link
Copy Markdown
Contributor Author

@weidongduan37 weidongduan37 Aug 6, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throw PrestoException here is intended here I think. checkState will make query failed directly. if query are like:
select a , sum(b) from druid_table where c = 'xxx' group by 1 having sum(b) > 0;
Throw PrestoException will help make the filter sum(b) > 0(on top of aggregation node) executed in presto(not in druid). The reason not push down such filter right now is because we could not make sure all filters on top of aggregation node could be pushed down. Following code will catch this exception and make queries work(DruidQueryGenerator.java):

    public Optional<DruidQueryGeneratorResult> generate(PlanNode plan, ConnectorSession session)
    {
        try {
            DruidQueryGeneratorContext context = requireNonNull(plan.accept(
                    new DruidQueryPlanVisitor(session),
                    new DruidQueryGeneratorContext()),
                    "Resulting context is null");
            return Optional.of(new DruidQueryGeneratorResult(context.toQuery(), context));
        }
        catch (PrestoException e) {
            log.debug(e, "Possibly benign error when pushing plan into scan node %s", plan);
            return Optional.empty();
        }
    }

Copy link
Copy Markdown
Contributor Author

@weidongduan37 weidongduan37 Aug 6, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After check #14952 changes, I think we are trying to solve different cases. His code focuses on time with timezone. I think our code is compatible.
By the way, the 1st and 3rd example he gave could be pushed down by original code. His change may be helpful for the 2nd example.
First:
select * from druid.wikipedia where __time > timestamp '2016-06-26 18:00:00.000'
Second:
select * from druid.wikipedia where __time > timestamp '2016-06-26 19:00:00.000 UTC';
Third:
select * from druid.wikipedia where __time > CAST('2016-06-26 18:00:00.000' as TIMESTAMP);

And the pushed down dql will be as following by original code:
first:
select col1,col2,.. from druid.wikipedia where __time > '2016-06-26 18:00:00.000'
third:
select col1, col2,.. from druid.wikipedia where __time > '2016-06-26 18:00:00.000'

}
checkState(!hasFilter(), "Druid doesn't support filters at multiple levels under AggregationNode");
return new DruidQueryGeneratorContext(
selections,
from,
Expand Down