Skip to content

Commit b72ad1a

Browse files
funbringerfunbringer
committed
check quals, early prototype seems to work somehow
1 parent 998022d commit b72ad1a

File tree

2 files changed

+194
-13
lines changed

2 files changed

+194
-13
lines changed

‎pickyappend.c

Lines changed: 187 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,128 @@ CustomPathMethods pickyappend_path_methods;
1616
CustomScanMethods pickyappend_plan_methods;
1717
CustomExecMethods pickyappend_exec_methods;
1818

19+
20+
static void
21+
clear_plan_states(ChildScanCommon *selected_plans, int n)
22+
{
23+
int i;
24+
25+
if (!selected_plans)
26+
return;
27+
28+
for (i = 0; i < n; i++)
29+
{
30+
ChildScanCommon child = selected_plans[i];
31+
32+
ExecEndNode(child->content.plan_state);
33+
}
34+
}
35+
36+
static void
37+
transform_plans_into_states(ChildScanCommon *selected_plans, int n, EState *estate)
38+
{
39+
int i;
40+
41+
for (i = 0; i < n; i++)
42+
{
43+
ChildScanCommon child = selected_plans[i];
44+
45+
child->content_type = CHILD_PLAN_STATE;
46+
child->content.plan_state = ExecInitNode(child->content.plan, estate, 0);
47+
}
48+
}
49+
50+
static ChildScanCommon *
51+
select_required_plans(ChildScanCommon *children, int nchildren,
52+
Oid *parts, int nparts,
53+
int *nres)
54+
{
55+
ChildScanCommon *children_end = children + nchildren;
56+
Oid *parts_end = parts + nparts;
57+
int allocated = 10;
58+
int used = 0;
59+
ChildScanCommon *result = palloc(10 * sizeof(ChildScanCommon));
60+
61+
while (children < children_end && parts < parts_end)
62+
{
63+
if ((*children)->relid < *parts)
64+
children++;
65+
else
66+
{
67+
if (!(*parts < (*children)->relid))
68+
{
69+
ChildScanCommon child = palloc(sizeof(ChildScanCommonData));
70+
71+
if (allocated <= used)
72+
{
73+
allocated *= 2;
74+
result = repalloc(result, allocated * sizeof(ChildScanCommon));
75+
}
76+
77+
child->content_type = CHILD_PLAN;
78+
child->content.plan = (*children)->content.plan;
79+
child->relid = (*children)->relid;
80+
81+
result[used++] = child;
82+
children++;
83+
}
84+
parts++;
85+
}
86+
}
87+
88+
*nres = used;
89+
return result;
90+
}
91+
92+
/* qsort comparison function for oids */
93+
static int
94+
oid_cmp(const void *p1, const void *p2)
95+
{
96+
Oid v1 = *((const Oid *) p1);
97+
Oid v2 = *((const Oid *) p2);
98+
99+
if (v1 < v2)
100+
return -1;
101+
if (v1 > v2)
102+
return 1;
103+
return 0;
104+
}
105+
106+
static Oid *
107+
get_partition_oids(List *ranges, int *n, PartRelationInfo *prel)
108+
{
109+
ListCell *range_cell;
110+
int allocated = 10;
111+
int used = 0;
112+
Oid *result = palloc(allocated * sizeof(Oid));
113+
Oid *children = dsm_array_get_pointer(&prel->children);
114+
115+
foreach (range_cell, ranges)
116+
{
117+
int i;
118+
int a = irange_lower(lfirst_irange(range_cell));
119+
int b = irange_upper(lfirst_irange(range_cell));
120+
121+
for (i = a; i <= b; i++)
122+
{
123+
if (allocated <= used)
124+
{
125+
allocated *= 2;
126+
result = repalloc(result, allocated * sizeof(Oid));
127+
}
128+
129+
Assert(i < prel->children_count);
130+
result[used++] = children[i];
131+
}
132+
}
133+
134+
if (used > 1)
135+
qsort(result, used, sizeof(Oid), oid_cmp);
136+
137+
*n = used;
138+
return result;
139+
}
140+
19141
static int
20142
cmp_child_scan_common(const void *a, const void *b)
21143
{
@@ -175,7 +297,7 @@ pathman_join_pathlist_hook(PlannerInfo *root,
175297
}
176298

177299
static void
178-
save_pickyappend_private(CustomScan *cscan, PickyAppendPath *path, PlannerInfo *root)
300+
save_pickyappend_private(CustomScan *cscan, PickyAppendPath *path)
179301
{
180302
ChildScanCommon *children = path->children;
181303
int nchildren = path->nchildren;
@@ -205,8 +327,6 @@ unpack_pickyappend_private(PickyAppendState *scan_state, CustomScan *cscan)
205327
ListCell *cur_plan;
206328
int i;
207329

208-
scan_state->relid = linitial_oid(linitial(cscan->custom_private));
209-
210330
i = 0;
211331
forboth (cur_oid, custom_oids, cur_plan, cscan->custom_plans)
212332
{
@@ -218,6 +338,10 @@ unpack_pickyappend_private(PickyAppendState *scan_state, CustomScan *cscan)
218338

219339
children[i++] = child;
220340
}
341+
342+
scan_state->relid = linitial_oid(linitial(cscan->custom_private));
343+
scan_state->children = children;
344+
scan_state->nchildren = nchildren;
221345
}
222346

223347
Plan *
@@ -235,10 +359,11 @@ create_pickyappend_plan(PlannerInfo *root, RelOptInfo *rel,
235359
cscan->scan.scanrelid = 0;
236360

237361
cscan->custom_exprs = gpath->cpath.custom_private;
362+
cscan->custom_plans = custom_plans;
238363

239364
cscan->methods = &pickyappend_plan_methods;
240365

241-
save_pickyappend_private(cscan, gpath, root);
366+
save_pickyappend_private(cscan, gpath);
242367

243368
return &cscan->scan.plan;
244369
}
@@ -257,23 +382,61 @@ pickyappend_create_scan_state(CustomScan *node)
257382

258383
scan_state->prel = get_pathman_relation_info(scan_state->relid, NULL);
259384

385+
scan_state->cur_plans = NULL;
386+
scan_state->ncur_plans = 0;
387+
scan_state->running_idx = 0;
388+
260389
return (Node *) scan_state;
261390
}
262391

263392
void
264393
pickyappend_begin(CustomScanState *node, EState *estate, int eflags)
265394
{
395+
PickyAppendState *scan_state = (PickyAppendState *) node;
396+
397+
scan_state->custom_expr_states = (List *) ExecInitExpr((Expr *) scan_state->custom_exprs,
398+
(PlanState *) scan_state);
266399
}
267400

268401
TupleTableSlot *
269402
pickyappend_exec(CustomScanState *node)
270403
{
404+
PickyAppendState *scan_state = (PickyAppendState *) node;
405+
406+
while (scan_state->running_idx < scan_state->ncur_plans)
407+
{
408+
ChildScanCommon child = scan_state->cur_plans[scan_state->running_idx];
409+
PlanState *state = child->content.plan_state;
410+
TupleTableSlot *slot = NULL;
411+
bool quals;
412+
413+
for (;;)
414+
{
415+
slot = ExecProcNode(state);
416+
417+
if (TupIsNull(slot))
418+
break;
419+
420+
node->ss.ps.ps_ExprContext->ecxt_scantuple = slot;
421+
quals = ExecQual(scan_state->custom_expr_states,
422+
node->ss.ps.ps_ExprContext, false);
423+
424+
if (quals)
425+
return slot;
426+
}
427+
428+
scan_state->running_idx++;
429+
}
430+
271431
return NULL;
272432
}
273433

274434
void
275435
pickyappend_end(CustomScanState *node)
276436
{
437+
PickyAppendState *scan_state = (PickyAppendState *) node;
438+
439+
clear_plan_states(scan_state->cur_plans, scan_state->ncur_plans);
277440
}
278441

279442
void
@@ -284,26 +447,38 @@ pickyappend_rescan(CustomScanState *node)
284447
PartRelationInfo *prel = scan_state->prel;
285448
List *ranges;
286449
ListCell *lc;
450+
Oid *parts;
451+
int nparts;
287452

288453
ranges = list_make1_int(make_irange(0, prel->children_count - 1, false));
289454

290455
foreach (lc, scan_state->custom_exprs)
291456
{
292-
WrapperNode *wn;
293-
WalkerContext wcxt;
457+
WrapperNode *wn;
458+
WalkerContext wcxt;
294459

295460
wcxt.econtext = econtext;
296461
wn = walk_expr_tree(&wcxt, (Expr *) lfirst(lc), prel);
297462

298463
ranges = irange_list_intersect(ranges, wn->rangeset);
299464
}
300465

301-
foreach (lc, ranges)
302-
{
303-
elog(LOG, "lower: %d, upper: %d",
304-
irange_lower(lfirst_irange(lc)),
305-
irange_upper(lfirst_irange(lc)));
306-
}
466+
parts = get_partition_oids(ranges, &nparts, prel);
467+
468+
clear_plan_states(scan_state->cur_plans, scan_state->ncur_plans);
469+
470+
scan_state->cur_plans = select_required_plans(scan_state->children,
471+
scan_state->nchildren,
472+
parts, nparts,
473+
&scan_state->ncur_plans);
474+
pfree(parts);
475+
476+
transform_plans_into_states(scan_state->cur_plans, scan_state->ncur_plans,
477+
scan_state->css.ss.ps.state);
478+
479+
scan_state->running_idx = 0;
480+
481+
/* elog(LOG, "nparts: %d, plans: %d", nparts, nplans); */
307482
}
308483

309484
void

‎pickyappend.h

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,10 +41,16 @@ typedef struct
4141
CustomScanState css;
4242
Oid relid;
4343
PartRelationInfo *prel;
44+
4445
List *custom_exprs;
46+
List *custom_expr_states;
4547

46-
ChildScanCommon *chilren;
48+
ChildScanCommon *children;
4749
int nchildren;
50+
51+
ChildScanCommon *cur_plans;
52+
int ncur_plans;
53+
int running_idx;
4854
} PickyAppendState;
4955

5056
extern set_join_pathlist_hook_type set_join_pathlist_next;

0 commit comments

Comments
 (0)