Pushdown DistinctLimitNode in Pinot Connector

[xiangfu0]

Push down DistinctLimitNode to Pinot Query.

For Presto query: SELECT DISTINCT flightnum FROM airlinestats LIMIT 10,
We will pushdown below query to Pinot:

• SQL format: SELECT FlightNum FROM airlineStats GROUP BY FlightNum LIMIT 10.
• PQL format: SELECT count(*) FROM airlineStats GROUP BY FlightNum TOP 10.

Below is the generated query plan for SQL mode.

presto:default> explain select distinct flightnum from airlinestats limit 10;
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 - Output[flightnum] => [flightnum:integer]
         Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: ?}
     - RemoteStreamingExchange[GATHER] => [flightnum:integer]
             Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: ?}
         - TableScan[TableHandle {connectorId='pinot', connectorHandle='PinotTableHandle{connectorId=pinot, schemaName=default, tableName=airlineStats, isQueryShort=Optional[true], expectedColumnHandles=Optional[[PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}]], pinotQuery=Optional[GeneratedPinotQuery{query=SELECT FlightNum FROM airlineStats GROUP BY FlightNum LIMIT 10, format=SQL, table=airlineStats, expectedColumnIndices=[], groupByClauses=1, haveFilter=false, isQueryShort=true}]}', layout='Optional[PinotTableHandle{connectorId=pinot, schemaName=default, tableName=airlineStats, isQueryShort=Optional[true], expectedColumnHandles=Optional[[PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}]], pinotQuery=Optional[GeneratedPinotQuery{query=SELECT FlightNum FROM airlineStats GROUP BY FlightNum LIMIT 10, format=SQL, table=airlineStats, expectedColumnIndices=[], groupByClauses=1, haveFilter=false, isQueryShort=true}]}]'}] => [flightnum:integer]
                 Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: 0.00}
                 flightnum := PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}

Below is the generated query plan for PQL mode.

presto:default> explain select distinct flightnum from airlinestats limit 10;

-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 - Output[flightnum] => [flightnum:integer]
         Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: ?}
     - RemoteStreamingExchange[GATHER] => [flightnum:integer]
             Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: ?}
         - TableScan[TableHandle {connectorId='pinot', connectorHandle='PinotTableHandle{connectorId=pinot, schemaName=default, tableName=airlineStats, isQueryShort=Optional[true], expectedColumnHandles=Optional[[PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}]], pinotQuery=Optional[GeneratedPinotQuery{query=SELECT count(*) FROM airlineStats GROUP BY FlightNum TOP 10, format=PQL, table=airlineStats, expectedColumnIndices=[0, -1], groupByClauses=1, haveFilter=false, isQueryShort=true}]}', layout='Optional[PinotTableHandle{connectorId=pinot, schemaName=default, tableName=airlineStats, isQueryShort=Optional[true], expectedColumnHandles=Optional[[PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}]], pinotQuery=Optional[GeneratedPinotQuery{query=SELECT count(*) FROM airlineStats GROUP BY FlightNum TOP 10, format=PQL, table=airlineStats, expectedColumnIndices=[0, -1], groupByClauses=1, haveFilter=false, isQueryShort=true}]}]'}] => [flightnum:integer]
                 Estimates: {rows: ? (?), cpu: ?, memory: 0.00, network: 0.00}
                 flightnum := PinotColumnHandle{columnName=FlightNum, dataType=integer, type=REGULAR}

(1 row)

== RELEASE NOTES ==

Pinot Changes
* Pushdown DistinctLimitNode to Pinot Query in SQL mode.

[xiangfu0]

@agrawaldevesh Could you help review this PR, thanks!

[agrawaldevesh]

Still trying to understand the implementation better -- some very high level questions inline.

[agrawaldevesh]

I think I have forgotten your old Pinot-SQL PR :-) -- but didn't we have a special derived class for the SQL generation ? Instead of using if useSqlSyntax

[xiangfu0]

This is for pushing down a new PlanNode: DistinctLimitPlanNode.
Query SELECT DISTINCT colA will be parsed to an AggregationPlanNode,
Query SELECT DISTINCT colA LIMIT X will be parsed to a DistinctLimitPlanNode.
For PQL side, since we still need to use the hidden column count(*), so I use withAggregation(...) for setting.
For SQL side, use withDistinctLimit to set groupbycolumns and limit.

[agrawaldevesh]

PinotQueryGeneratorContext.withAggregation has some special handling for the case when distinctCount is already pushed down. (Search for PINOT_DISTINCT_COUNT_FUNCTION_NAME). Can you make sure that this doesn't conflict with that. Thanks

[agrawaldevesh]

Getting very close

[agrawaldevesh]

PinotQueryGeneratorContext.withAggregation has some special handling for the case when distinctCount is already pushed down. (Search for PINOT_DISTINCT_COUNT_FUNCTION_NAME). Can you make sure that this doesn't conflict with that. Thanks

[xiangfu0]

distinctCount pushdown won't be triggered by this. It's on top of count on a column with distinct mark.

[agrawaldevesh]

LGTM !. Thanks for iterating on this !

cc: @zhenxiao @highker @sachdevs Please have a look, adding you guys as reviewers.

[xiangfu0]

LGTM !. Thanks for iterating on this !

cc: @zhenxiao @highker @sachdevs Please have a look, adding you guys as reviewers.

Thanks @agrawaldevesh

[highker]

lgtm