external
/
citus
mirror of https://github.com/citusdata/citus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for parametrized execution for subquery pushdown (#1356) 

Distributed query planning for subquery pushdown is done on the original
query. This prevents the usage of external parameters on the execution.
To overcome this, we manually replace the parameters on the original
query.
pull/1405/head
Önder Kalacı 2017-05-10 09:38:48 +03:00 committed by GitHub
parent 97334a9123
commit 3ec502b286
7 changed files with 593 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

106
src/backend/distributed/planner/multi_logical_planner.c
View File

@ -132,6 +132,7 @@ static MultiNode * ApplyCartesianProduct(MultiNode *leftNode, MultiNode *rightNo
* Local functions forward declarations for subquery pushdown. Note that these
* functions will be removed with upcoming subqery changes.
*/
static Node * ResolveExternalParams(Node *inputNode, ParamListInfo boundParams);
static MultiNode * MultiSubqueryPlanTree(Query *originalQuery,
Query *queryTree,
PlannerRestrictionContext *
@ -152,6 +153,12 @@ static MultiTable * MultiSubqueryPushdownTable(Query *subquery);
* plan and adds a root node to top of it. The original query is only used for subquery
* pushdown planning.
*
* In order to support external parameters for the queries where planning
* is done on the original query, we need to replace the external parameters
* manually. To achive that for subquery pushdown planning, we pass boundParams
* to this function. We need to do that since Citus currently unable to send
* parameters to the workers on the execution.
*
* We also pass queryTree and plannerRestrictionContext to the planner. They
* are primarily used to decide whether the subquery is safe to pushdown.
* If not, it helps to produce meaningful error messages for subquery
@ -159,7 +166,8 @@ static MultiTable * MultiSubqueryPushdownTable(Query *subquery);
*/
MultiTreeRoot *
MultiLogicalPlanCreate(Query *originalQuery, Query *queryTree,
PlannerRestrictionContext *plannerRestrictionContext)
PlannerRestrictionContext *plannerRestrictionContext,
ParamListInfo boundParams)
{
MultiNode *multiQueryNode = NULL;
MultiTreeRoot *rootNode = NULL;
@ -176,6 +184,9 @@ MultiLogicalPlanCreate(Query *originalQuery, Query *queryTree,
*/
if (SubqueryEntryList(queryTree) != NIL || SublinkList(originalQuery) != NIL)
{
originalQuery = (Query *) ResolveExternalParams((Node *) originalQuery,
boundParams);
multiQueryNode = MultiSubqueryPlanTree(originalQuery, queryTree,
plannerRestrictionContext);
}
@ -192,6 +203,99 @@ MultiLogicalPlanCreate(Query *originalQuery, Query *queryTree,
}
/*
* ResolveExternalParams replaces the external parameters that appears
* in the query with the corresponding entries in the boundParams.
*
* Note that this function is inspired by eval_const_expr() on Postgres.
* We cannot use that function because it requires access to PlannerInfo.
*/
static Node *
ResolveExternalParams(Node *inputNode, ParamListInfo boundParams)
{
/* consider resolving external parameters only when boundParams exists */
if (!boundParams)
{
return inputNode;
}
if (inputNode == NULL)
{
return NULL;
}
if (IsA(inputNode, Param))
{
Param *paramToProcess = (Param *) inputNode;
ParamExternData *correspondingParameterData = NULL;
int numberOfParameters = boundParams->numParams;
int parameterId = paramToProcess->paramid;
int16 typeLength = 0;
bool typeByValue = false;
Datum constValue = 0;
bool paramIsNull = false;
int parameterIndex = 0;
if (paramToProcess->paramkind != PARAM_EXTERN)
{
return inputNode;
}
if (parameterId < 0)
{
return inputNode;
}
/* parameterId starts from 1 */
parameterIndex = parameterId - 1;
if (parameterIndex >= numberOfParameters)
{
return inputNode;
}
correspondingParameterData = &boundParams->params[parameterIndex];
if (!(correspondingParameterData->pflags & PARAM_FLAG_CONST))
{
return inputNode;
}
get_typlenbyval(paramToProcess->paramtype, &typeLength, &typeByValue);
paramIsNull = correspondingParameterData->isnull;
if (paramIsNull)
{
constValue = 0;
}
else if (typeByValue)
{
constValue = correspondingParameterData->value;
}
else
{
/*
* Out of paranoia ensure that datum lives long enough,
* although bind params currently should always live
* long enough.
*/
constValue = datumCopy(correspondingParameterData->value, typeByValue,
typeLength);
}
return (Node *) makeConst(paramToProcess->paramtype, paramToProcess->paramtypmod,
paramToProcess->paramcollid, typeLength, constValue,
paramIsNull, typeByValue);
}
else if (IsA(inputNode, Query))
{
return (Node *) query_tree_mutator((Query *) inputNode, ResolveExternalParams,
boundParams, 0);
}
return expression_tree_mutator(inputNode, ResolveExternalParams, boundParams);
}
/*
* SublinkList finds the subquery nodes in the where clause of the given query. Note
* that the function should be called on the original query given that postgres

3
src/backend/distributed/planner/multi_planner.c
View File

@ -310,7 +310,8 @@ CreateDistributedPlan(PlannedStmt *localPlan, Query *originalQuery, Query *query
if ((!distributedPlan || distributedPlan->planningError) && !hasUnresolvedParams)
{
MultiTreeRoot *logicalPlan = MultiLogicalPlanCreate(originalQuery, query,
plannerRestrictionContext);
plannerRestrictionContext,
boundParams);
MultiLogicalPlanOptimize(logicalPlan);
/*

3
src/include/distributed/multi_logical_planner.h
View File

@ -183,7 +183,8 @@ extern bool SubqueryPushdown;
/* Function declarations for building logical plans */
extern MultiTreeRoot * MultiLogicalPlanCreate(Query *originalQuery, Query *queryTree,
PlannerRestrictionContext *
plannerRestrictionContext);
plannerRestrictionContext,
ParamListInfo boundParams);
extern bool NeedsDistributedPlanning(Query *queryTree);
extern MultiNode * ParentNode(MultiNode *multiNode);
extern MultiNode * ChildNode(MultiUnaryNode *multiNode);

18
src/test/regress/expected/multi_subquery_behavioral_analytics.out
View File

@ -1996,7 +1996,14 @@ FROM (
ORDER BY 2 DESC, 1
LIMIT $1 OFFSET $2;
EXECUTE parametrized_limit(3,3);
ERROR: no value found for parameter 1
DEBUG: push down of limit count: 6
user_id | array_length
---------+--------------
13 | 172
12 | 121
23 | 115
(3 rows)
PREPARE parametrized_offset AS
SELECT user_id, array_length(events_table, 1)
FROM (
@ -2013,7 +2020,14 @@ FROM (
ORDER BY 2 DESC, 1
LIMIT 3 OFFSET $1;
EXECUTE parametrized_offset(3);
ERROR: no value found for parameter 1
DEBUG: push down of limit count: 6
user_id | array_length
---------+--------------
13 | 172
12 | 121
23 | 115
(3 rows)
SET client_min_messages TO DEFAULT;
DROP FUNCTION volatile_func_test();
CREATE FUNCTION test_join_function_2(integer, integer) RETURNS bool

312
src/test/regress/expected/multi_subquery_misc.out Normal file
View File

@ -0,0 +1,312 @@
-- multi subquery pushdown misc aims to test subquery pushdown queries with
-- (i) Prepared statements
-- (ii) PL/PGSQL functions
-- (iii) SQL functions
-- the tables that are used depends to multi_insert_select_behavioral_analytics_create_table.sql
-- We don't need shard id sequence here, so commented out to prevent conflicts with concurrent tests
SET citus.enable_router_execution TO false;
PREPARE prepared_subquery_1 AS
SELECT
user_id,
user_lastseen,
array_length(event_array, 1)
FROM (
SELECT
user_id,
max(u.time) as user_lastseen,
array_agg(event_type ORDER BY u.time) AS event_array
FROM (
SELECT user_id, time
FROM users_table
WHERE
user_id >= 10 AND
user_id <= 70 AND
users_table.value_1 > 10 AND users_table.value_1 < 12
) u LEFT JOIN LATERAL (
SELECT event_type, time
FROM events_table
WHERE user_id = u.user_id AND
events_table.event_type > 10 AND events_table.event_type < 12
) t ON true
GROUP BY user_id
) AS shard_union
ORDER BY user_lastseen DESC, user_id;
EXECUTE prepared_subquery_1;
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
PREPARE prepared_subquery_2(int, int) AS
SELECT
user_id,
user_lastseen,
array_length(event_array, 1)
FROM (
SELECT
user_id,
max(u.time) as user_lastseen,
array_agg(event_type ORDER BY u.time) AS event_array
FROM (
SELECT user_id, time
FROM users_table
WHERE
user_id >= $1 AND
user_id <= $2 AND
users_table.value_1 > 10 AND users_table.value_1 < 12
) u LEFT JOIN LATERAL (
SELECT event_type, time
FROM events_table
WHERE user_id = u.user_id AND
events_table.event_type > 10 AND events_table.event_type < 12
) t ON true
GROUP BY user_id
) AS shard_union
ORDER BY user_lastseen DESC, user_id;
-- should be fine with more than five executions
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
EXECUTE prepared_subquery_2(10, 70);
user_id | user_lastseen | array_length
---------+---------------------------------+--------------
12 | Sun Jan 19 01:49:20.372688 2014 | 1
20 | Sat Jan 18 14:25:31.817903 2014 | 1
42 | Thu Jan 16 07:08:02.651966 2014 | 1
56 | Tue Jan 14 12:11:47.27375 2014 | 1
57 | Mon Jan 13 14:53:50.494836 2014 | 1
65 | Sun Jan 12 03:14:26.810597 2014 | 1
(6 rows)
-- prepared statements with subqueries in WHERE clause
PREPARE prepared_subquery_3(int, int, int, int, int, int) AS
SELECT user_id
FROM users_table
WHERE user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $4 AND value_1 <= $3)
AND user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $5 AND value_1 <= $6)
AND user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $1 AND value_1 <= $2)
GROUP BY
user_id
ORDER BY
user_id DESC
LIMIT 5;
-- enough times (6+) to actually use prepared statements
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
user_id
---------
93
90
88
87
84
(5 rows)
CREATE FUNCTION plpgsql_subquery_test(int, int) RETURNS TABLE(count bigint) AS $$
DECLARE
BEGIN
RETURN QUERY
SELECT
count(*)
FROM
users_table
JOIN
(SELECT
ma.user_id, (GREATEST(coalesce(ma.value_4 / 250, 0.0) + GREATEST(1.0))) / 2 AS prob
FROM
users_table AS ma, events_table as short_list
WHERE
short_list.user_id = ma.user_id and ma.value_1 < $1 and short_list.event_type < 50
) temp
ON users_table.user_id = temp.user_id
WHERE
users_table.value_1 < $2;
END;
$$ LANGUAGE plpgsql;
-- enough times (6+) to actually use prepared statements
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
SELECT plpgsql_subquery_test(10, 20);
plpgsql_subquery_test
-----------------------
1500
(1 row)
-- this should also work, but should return 0 given that int = NULL is always returns false
SELECT plpgsql_subquery_test(10, NULL);
plpgsql_subquery_test
-----------------------
0
(1 row)
CREATE FUNCTION sql_subquery_test(int, int) RETURNS bigint AS $$
SELECT
count(*)
FROM
users_table
JOIN
(SELECT
ma.user_id, (GREATEST(coalesce(ma.value_4 / 250, 0.0) + GREATEST(1.0))) / 2 AS prob
FROM
users_table AS ma, events_table as short_list
WHERE
short_list.user_id = ma.user_id and ma.value_1 < $1 and short_list.event_type < 50
) temp
ON users_table.user_id = temp.user_id
WHERE
users_table.value_1 < $2;
$$ LANGUAGE SQL;
-- should error out
SELECT sql_subquery_test(5,5);
ERROR: could not create distributed plan
DETAIL: Possibly this is caused by the use of parameters in SQL functions, which is not supported in Citus.
HINT: Consider using PL/pgSQL functions instead.
CONTEXT: SQL function "sql_subquery_test" statement 1
DROP FUNCTION plpgsql_subquery_test(int, int);
DROP FUNCTION sql_subquery_test(int, int);

2
src/test/regress/multi_schedule
View File

@ -41,7 +41,7 @@ test: multi_deparse_shard_query
test: multi_basic_queries multi_complex_expressions
test: multi_explain
test: multi_subquery multi_subquery_complex_queries multi_subquery_behavioral_analytics
test: multi_subquery_union multi_subquery_in_where_clause
test: multi_subquery_union multi_subquery_in_where_clause multi_subquery_misc
test: multi_reference_table
test: multi_outer_join_reference
test: multi_single_relation_subquery

155
src/test/regress/sql/multi_subquery_misc.sql Normal file
View File

@ -0,0 +1,155 @@
-- multi subquery pushdown misc aims to test subquery pushdown queries with
-- (i) Prepared statements
-- (ii) PL/PGSQL functions
-- (iii) SQL functions
-- the tables that are used depends to multi_insert_select_behavioral_analytics_create_table.sql
-- We don't need shard id sequence here, so commented out to prevent conflicts with concurrent tests
SET citus.enable_router_execution TO false;
PREPARE prepared_subquery_1 AS
SELECT
user_id,
user_lastseen,
array_length(event_array, 1)
FROM (
SELECT
user_id,
max(u.time) as user_lastseen,
array_agg(event_type ORDER BY u.time) AS event_array
FROM (
SELECT user_id, time
FROM users_table
WHERE
user_id >= 10 AND
user_id <= 70 AND
users_table.value_1 > 10 AND users_table.value_1 < 12
) u LEFT JOIN LATERAL (
SELECT event_type, time
FROM events_table
WHERE user_id = u.user_id AND
events_table.event_type > 10 AND events_table.event_type < 12
) t ON true
GROUP BY user_id
) AS shard_union
ORDER BY user_lastseen DESC, user_id;
EXECUTE prepared_subquery_1;
PREPARE prepared_subquery_2(int, int) AS
SELECT
user_id,
user_lastseen,
array_length(event_array, 1)
FROM (
SELECT
user_id,
max(u.time) as user_lastseen,
array_agg(event_type ORDER BY u.time) AS event_array
FROM (
SELECT user_id, time
FROM users_table
WHERE
user_id >= $1 AND
user_id <= $2 AND
users_table.value_1 > 10 AND users_table.value_1 < 12
) u LEFT JOIN LATERAL (
SELECT event_type, time
FROM events_table
WHERE user_id = u.user_id AND
events_table.event_type > 10 AND events_table.event_type < 12
) t ON true
GROUP BY user_id
) AS shard_union
ORDER BY user_lastseen DESC, user_id;
-- should be fine with more than five executions
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
EXECUTE prepared_subquery_2(10, 70);
-- prepared statements with subqueries in WHERE clause
PREPARE prepared_subquery_3(int, int, int, int, int, int) AS
SELECT user_id
FROM users_table
WHERE user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $4 AND value_1 <= $3)
AND user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $5 AND value_1 <= $6)
AND user_id IN (SELECT user_id FROM users_table WHERE value_1 >= $1 AND value_1 <= $2)
GROUP BY
user_id
ORDER BY
user_id DESC
LIMIT 5;
-- enough times (6+) to actually use prepared statements
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
EXECUTE prepared_subquery_3(50, 60, 20, 10, 30, 40);
CREATE FUNCTION plpgsql_subquery_test(int, int) RETURNS TABLE(count bigint) AS $$
DECLARE
BEGIN
RETURN QUERY
SELECT
count(*)
FROM
users_table
JOIN
(SELECT
ma.user_id, (GREATEST(coalesce(ma.value_4 / 250, 0.0) + GREATEST(1.0))) / 2 AS prob
FROM
users_table AS ma, events_table as short_list
WHERE
short_list.user_id = ma.user_id and ma.value_1 < $1 and short_list.event_type < 50
) temp
ON users_table.user_id = temp.user_id
WHERE
users_table.value_1 < $2;
END;
$$ LANGUAGE plpgsql;
-- enough times (6+) to actually use prepared statements
SELECT plpgsql_subquery_test(10, 20);
SELECT plpgsql_subquery_test(10, 20);
SELECT plpgsql_subquery_test(10, 20);
SELECT plpgsql_subquery_test(10, 20);
SELECT plpgsql_subquery_test(10, 20);
SELECT plpgsql_subquery_test(10, 20);
-- this should also work, but should return 0 given that int = NULL is always returns false
SELECT plpgsql_subquery_test(10, NULL);
CREATE FUNCTION sql_subquery_test(int, int) RETURNS bigint AS $$
SELECT
count(*)
FROM
users_table
JOIN
(SELECT
ma.user_id, (GREATEST(coalesce(ma.value_4 / 250, 0.0) + GREATEST(1.0))) / 2 AS prob
FROM
users_table AS ma, events_table as short_list
WHERE
short_list.user_id = ma.user_id and ma.value_1 < $1 and short_list.event_type < 50
) temp
ON users_table.user_id = temp.user_id
WHERE
users_table.value_1 < $2;
$$ LANGUAGE SQL;
-- should error out
SELECT sql_subquery_test(5,5);
DROP FUNCTION plpgsql_subquery_test(int, int);
DROP FUNCTION sql_subquery_test(int, int);