Skip to content

Commit c6dbf1f

Browse files
robertmhaasrobertmhaas
committed
Stop the executor if no more tuples can be sent from worker to leader.
If a Gather node has read as many tuples as it needs (for example, due to Limit) it may detach the queue connecting it to the worker before reading all of the worker's tuples. Rather than let the worker continue to generate and send all of the results, have it stop after sending the next tuple. More could be done here to stop the worker even quicker, but this is about as well as we can hope to do for 9.6. This is in response to a problem report from Andreas Seltenreich. Commit 44339b8 should be actually be sufficient to fix that example even without this change, but it seems better to do this, too, since we might otherwise waste quite a large amount of effort in one or more workers. Discussion: CAA4eK1KOKGqmz9bGu+Z42qhRwMbm4R5rfnqsLCNqFs9j14jzEA@mail.gmail.com Amit Kapila
1 parent 44339b8 commit c6dbf1f

File tree

14 files changed

+78
-29
lines changed

14 files changed

+78
-29
lines changed

‎src/backend/access/common/printtup.c

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,9 @@
2626

2727
static void printtup_startup(DestReceiver *self, int operation,
2828
TupleDesc typeinfo);
29-
static void printtup(TupleTableSlot *slot, DestReceiver *self);
30-
static void printtup_20(TupleTableSlot *slot, DestReceiver *self);
31-
static void printtup_internal_20(TupleTableSlot *slot, DestReceiver *self);
29+
static bool printtup(TupleTableSlot *slot, DestReceiver *self);
30+
static bool printtup_20(TupleTableSlot *slot, DestReceiver *self);
31+
static bool printtup_internal_20(TupleTableSlot *slot, DestReceiver *self);
3232
static void printtup_shutdown(DestReceiver *self);
3333
static void printtup_destroy(DestReceiver *self);
3434

@@ -299,7 +299,7 @@ printtup_prepare_info(DR_printtup *myState, TupleDesc typeinfo, int numAttrs)
299299
* printtup --- print a tuple in protocol 3.0
300300
* ----------------
301301
*/
302-
static void
302+
static bool
303303
printtup(TupleTableSlot *slot, DestReceiver *self)
304304
{
305305
TupleDesc typeinfo = slot->tts_tupleDescriptor;
@@ -376,13 +376,15 @@ printtup(TupleTableSlot *slot, DestReceiver *self)
376376
/* Return to caller's context, and flush row's temporary memory */
377377
MemoryContextSwitchTo(oldcontext);
378378
MemoryContextReset(myState->tmpcontext);
379+
380+
return true;
379381
}
380382

381383
/* ----------------
382384
* printtup_20 --- print a tuple in protocol 2.0
383385
* ----------------
384386
*/
385-
static void
387+
static bool
386388
printtup_20(TupleTableSlot *slot, DestReceiver *self)
387389
{
388390
TupleDesc typeinfo = slot->tts_tupleDescriptor;
@@ -452,6 +454,8 @@ printtup_20(TupleTableSlot *slot, DestReceiver *self)
452454
/* Return to caller's context, and flush row's temporary memory */
453455
MemoryContextSwitchTo(oldcontext);
454456
MemoryContextReset(myState->tmpcontext);
457+
458+
return true;
455459
}
456460

457461
/* ----------------
@@ -528,7 +532,7 @@ debugStartup(DestReceiver *self, int operation, TupleDesc typeinfo)
528532
* debugtup - print one tuple for an interactive backend
529533
* ----------------
530534
*/
531-
void
535+
bool
532536
debugtup(TupleTableSlot *slot, DestReceiver *self)
533537
{
534538
TupleDesc typeinfo = slot->tts_tupleDescriptor;
@@ -553,6 +557,8 @@ debugtup(TupleTableSlot *slot, DestReceiver *self)
553557
printatt((unsigned) i + 1, typeinfo->attrs[i], value);
554558
}
555559
printf("\t----\n");
560+
561+
return true;
556562
}
557563

558564
/* ----------------
@@ -564,7 +570,7 @@ debugtup(TupleTableSlot *slot, DestReceiver *self)
564570
* This is largely same as printtup_20, except we use binary formatting.
565571
* ----------------
566572
*/
567-
static void
573+
static bool
568574
printtup_internal_20(TupleTableSlot *slot, DestReceiver *self)
569575
{
570576
TupleDesc typeinfo = slot->tts_tupleDescriptor;
@@ -636,4 +642,6 @@ printtup_internal_20(TupleTableSlot *slot, DestReceiver *self)
636642
/* Return to caller's context, and flush row's temporary memory */
637643
MemoryContextSwitchTo(oldcontext);
638644
MemoryContextReset(myState->tmpcontext);
645+
646+
return true;
639647
}

‎src/backend/commands/copy.c

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4454,7 +4454,7 @@ copy_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
44544454
/*
44554455
* copy_dest_receive --- receive one tuple
44564456
*/
4457-
static void
4457+
static bool
44584458
copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
44594459
{
44604460
DR_copy *myState = (DR_copy *) self;
@@ -4466,6 +4466,8 @@ copy_dest_receive(TupleTableSlot *slot, DestReceiver *self)
44664466
/* And send the data */
44674467
CopyOneRowTo(cstate, InvalidOid, slot->tts_values, slot->tts_isnull);
44684468
myState->processed++;
4469+
4470+
return true;
44694471
}
44704472

44714473
/*

‎src/backend/commands/createas.c

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ typedef struct
6262
static ObjectAddress CreateAsReladdr = {InvalidOid, InvalidOid, 0};
6363

6464
static void intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
65-
static void intorel_receive(TupleTableSlot *slot, DestReceiver *self);
65+
static bool intorel_receive(TupleTableSlot *slot, DestReceiver *self);
6666
static void intorel_shutdown(DestReceiver *self);
6767
static void intorel_destroy(DestReceiver *self);
6868

@@ -482,7 +482,7 @@ intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
482482
/*
483483
* intorel_receive --- receive one tuple
484484
*/
485-
static void
485+
static bool
486486
intorel_receive(TupleTableSlot *slot, DestReceiver *self)
487487
{
488488
DR_intorel *myState = (DR_intorel *) self;
@@ -507,6 +507,8 @@ intorel_receive(TupleTableSlot *slot, DestReceiver *self)
507507
myState->bistate);
508508

509509
/* We know this is a newly created relation, so there are no indexes */
510+
511+
return true;
510512
}
511513

512514
/*

‎src/backend/commands/matview.c

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ typedef struct
5656
static int matview_maintenance_depth = 0;
5757

5858
static void transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
59-
static void transientrel_receive(TupleTableSlot *slot, DestReceiver *self);
59+
static bool transientrel_receive(TupleTableSlot *slot, DestReceiver *self);
6060
static void transientrel_shutdown(DestReceiver *self);
6161
static void transientrel_destroy(DestReceiver *self);
6262
static void refresh_matview_datafill(DestReceiver *dest, Query *query,
@@ -467,7 +467,7 @@ transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
467467
/*
468468
* transientrel_receive --- receive one tuple
469469
*/
470-
static void
470+
static bool
471471
transientrel_receive(TupleTableSlot *slot, DestReceiver *self)
472472
{
473473
DR_transientrel *myState = (DR_transientrel *) self;
@@ -486,6 +486,8 @@ transientrel_receive(TupleTableSlot *slot, DestReceiver *self)
486486
myState->bistate);
487487

488488
/* We know this is a newly created relation, so there are no indexes */
489+
490+
return true;
489491
}
490492

491493
/*

‎src/backend/executor/execMain.c

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1593,7 +1593,15 @@ ExecutePlan(EState *estate,
15931593
* practice, this is probably always the case at this point.)
15941594
*/
15951595
if (sendTuples)
1596-
(*dest->receiveSlot) (slot, dest);
1596+
{
1597+
/*
1598+
* If we are not able to send the tuple, we assume the destination
1599+
* has closed and no more tuples can be sent. If that's the case,
1600+
* end the loop.
1601+
*/
1602+
if (!((*dest->receiveSlot) (slot, dest)))
1603+
break;
1604+
}
15971605

15981606
/*
15991607
* Count tuples processed, if this is a SELECT. (For other operation

‎src/backend/executor/execTuples.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1266,7 +1266,7 @@ do_tup_output(TupOutputState *tstate, Datum *values, bool *isnull)
12661266
ExecStoreVirtualTuple(slot);
12671267

12681268
/* send the tuple to the receiver */
1269-
(*tstate->dest->receiveSlot) (slot, tstate->dest);
1269+
(void) (*tstate->dest->receiveSlot) (slot, tstate->dest);
12701270

12711271
/* clean up */
12721272
ExecClearTuple(slot);

‎src/backend/executor/functions.c

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ static Datum postquel_get_single_result(TupleTableSlot *slot,
167167
static void sql_exec_error_callback(void *arg);
168168
static void ShutdownSQLFunction(Datum arg);
169169
static void sqlfunction_startup(DestReceiver *self, int operation, TupleDesc typeinfo);
170-
static void sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self);
170+
static bool sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self);
171171
static void sqlfunction_shutdown(DestReceiver *self);
172172
static void sqlfunction_destroy(DestReceiver *self);
173173

@@ -1904,7 +1904,7 @@ sqlfunction_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
19041904
/*
19051905
* sqlfunction_receive --- receive one tuple
19061906
*/
1907-
static void
1907+
static bool
19081908
sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self)
19091909
{
19101910
DR_sqlfunction *myState = (DR_sqlfunction *) self;
@@ -1914,6 +1914,8 @@ sqlfunction_receive(TupleTableSlot *slot, DestReceiver *self)
19141914

19151915
/* Store the filtered tuple into the tuplestore */
19161916
tuplestore_puttupleslot(myState->tstore, slot);
1917+
1918+
return true;
19171919
}
19181920

19191921
/*

‎src/backend/executor/spi.c

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1774,7 +1774,7 @@ spi_dest_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
17741774
* store tuple retrieved by Executor into SPITupleTable
17751775
* of current SPI procedure
17761776
*/
1777-
void
1777+
bool
17781778
spi_printtup(TupleTableSlot *slot, DestReceiver *self)
17791779
{
17801780
SPITupleTable *tuptable;
@@ -1809,6 +1809,8 @@ spi_printtup(TupleTableSlot *slot, DestReceiver *self)
18091809
(tuptable->free)--;
18101810

18111811
MemoryContextSwitchTo(oldcxt);
1812+
1813+
return true;
18121814
}
18131815

18141816
/*

‎src/backend/executor/tqueue.c

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,12 +115,13 @@ static RemapInfo *BuildRemapInfo(TupleDesc tupledesc);
115115
* type over a range type over a range type over an array type over a record,
116116
* or something like that.
117117
*/
118-
static void
118+
static bool
119119
tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
120120
{
121121
TQueueDestReceiver *tqueue = (TQueueDestReceiver *) self;
122122
TupleDesc tupledesc = slot->tts_tupleDescriptor;
123123
HeapTuple tuple;
124+
shm_mq_result result;
124125

125126
/*
126127
* Test to see whether the tupledesc has changed; if so, set up for the
@@ -195,7 +196,16 @@ tqueueReceiveSlot(TupleTableSlot *slot, DestReceiver *self)
195196
}
196197

197198
/* Send the tuple itself. */
198-
shm_mq_send(tqueue->handle, tuple->t_len, tuple->t_data, false);
199+
result = shm_mq_send(tqueue->handle, tuple->t_len, tuple->t_data, false);
200+
201+
if (result == SHM_MQ_DETACHED)
202+
return false;
203+
else if (result != SHM_MQ_SUCCESS)
204+
ereport(ERROR,
205+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
206+
errmsg("unable to send tuples")));
207+
208+
return true;
199209
}
200210

201211
/*

‎src/backend/executor/tstoreReceiver.c

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,8 @@ typedef struct
3737
} TStoreState;
3838

3939

40-
static void tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self);
41-
static void tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self);
40+
static bool tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self);
41+
static bool tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self);
4242

4343

4444
/*
@@ -90,19 +90,21 @@ tstoreStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)
9090
* Receive a tuple from the executor and store it in the tuplestore.
9191
* This is for the easy case where we don't have to detoast.
9292
*/
93-
static void
93+
static bool
9494
tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self)
9595
{
9696
TStoreState *myState = (TStoreState *) self;
9797

9898
tuplestore_puttupleslot(myState->tstore, slot);
99+
100+
return true;
99101
}
100102

101103
/*
102104
* Receive a tuple from the executor and store it in the tuplestore.
103105
* This is for the case where we have to detoast any toasted values.
104106
*/
105-
static void
107+
static bool
106108
tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self)
107109
{
108110
TStoreState *myState = (TStoreState *) self;
@@ -152,6 +154,8 @@ tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self)
152154
/* And release any temporary detoasted values */
153155
for (i = 0; i < nfree; i++)
154156
pfree(DatumGetPointer(myState->tofree[i]));
157+
158+
return true;
155159
}
156160

157161
/*

‎src/backend/tcop/dest.c

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,10 @@
4545
* dummy DestReceiver functions
4646
* ----------------
4747
*/
48-
static void
48+
static bool
4949
donothingReceive(TupleTableSlot *slot, DestReceiver *self)
5050
{
51+
return true;
5152
}
5253

5354
static void

‎src/backend/tcop/pquery.c

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1109,7 +1109,13 @@ RunFromStore(Portal portal, ScanDirection direction, uint64 count,
11091109
if (!ok)
11101110
break;
11111111

1112-
(*dest->receiveSlot) (slot, dest);
1112+
/*
1113+
* If we are not able to send the tuple, we assume the destination
1114+
* has closed and no more tuples can be sent. If that's the case,
1115+
* end the loop.
1116+
*/
1117+
if (!((*dest->receiveSlot) (slot, dest)))
1118+
break;
11131119

11141120
ExecClearTuple(slot);
11151121

‎src/include/access/printtup.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,11 @@ extern void SendRowDescriptionMessage(TupleDesc typeinfo, List *targetlist,
2525

2626
extern void debugStartup(DestReceiver *self, int operation,
2727
TupleDesc typeinfo);
28-
extern void debugtup(TupleTableSlot *slot, DestReceiver *self);
28+
extern bool debugtup(TupleTableSlot *slot, DestReceiver *self);
2929

3030
/* XXX these are really in executor/spi.c */
3131
extern void spi_dest_startup(DestReceiver *self, int operation,
3232
TupleDesc typeinfo);
33-
extern void spi_printtup(TupleTableSlot *slot, DestReceiver *self);
33+
extern bool spi_printtup(TupleTableSlot *slot, DestReceiver *self);
3434

3535
#endif /* PRINTTUP_H */

‎src/include/tcop/dest.h

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,15 +104,17 @@ typedef enum
104104
* pointers that the executor must call.
105105
*
106106
* Note: the receiveSlot routine must be passed a slot containing a TupleDesc
107-
* identical to the one given to the rStartup routine.
107+
* identical to the one given to the rStartup routine. It returns bool where
108+
* a "true" value means "continue processing" and a "false" value means
109+
* "stop early, just as if we'd reached the end of the scan".
108110
* ----------------
109111
*/
110112
typedef struct _DestReceiver DestReceiver;
111113

112114
struct _DestReceiver
113115
{
114116
/* Called for each tuple to be output: */
115-
void (*receiveSlot) (TupleTableSlot *slot,
117+
bool (*receiveSlot) (TupleTableSlot *slot,
116118
DestReceiver *self);
117119
/* Per-executor-run initialization and shutdown: */
118120
void (*rStartup) (DestReceiver *self,

0 commit comments

Comments
 (0)