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

wip

adaptiveExecutor_DestReceiver
SaitTalhaNisanci 2020-02-12 22:27:58 +03:00
parent fc1fe0244e
commit f4828f547f
1 changed files with 23 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/backend/distributed/executor/adaptive_executor.c
View File

@ -163,6 +163,8 @@
#include "utils/lsyscache.h" #include "utils/lsyscache.h"
#include "utils/memutils.h" #include "utils/memutils.h"
#include "utils/timestamp.h" #include "utils/timestamp.h"
#include "executor/tstoreReceiver.h"
#include "executor/tuptable.h"
/* /*
@ -191,7 +193,7 @@ typedef struct DistributedExecution
/* Tuple descriptor and destination for result. Can be NULL. */ /* Tuple descriptor and destination for result. Can be NULL. */
TupleDesc tupleDescriptor; TupleDesc tupleDescriptor;
Tuplestorestate *tupleStore; DestReceiver* destReceiver;
/* list of workers involved in the execution */ /* list of workers involved in the execution */
@ -547,7 +549,7 @@ static DistributedExecution * CreateDistributedExecution(RowModifyLevel modLevel
bool hasReturning, bool hasReturning,
ParamListInfo paramListInfo, ParamListInfo paramListInfo,
TupleDesc tupleDescriptor, TupleDesc tupleDescriptor,
Tuplestorestate *tupleStore, DestReceiver *destReceiver,
int targetPoolSize, int targetPoolSize,
TransactionProperties * TransactionProperties *
xactProperties, xactProperties,
@ -633,6 +635,7 @@ AdaptiveExecutor(CitusScanState *scanState)
EState *executorState = ScanStateGetExecutorState(scanState); EState *executorState = ScanStateGetExecutorState(scanState);
ParamListInfo paramListInfo = executorState->es_param_list_info; ParamListInfo paramListInfo = executorState->es_param_list_info;
TupleDesc tupleDescriptor = ScanStateGetTupleDescriptor(scanState); TupleDesc tupleDescriptor = ScanStateGetTupleDescriptor(scanState);
DestReceiver *tupleStoreDestReceiever = CreateTuplestoreDestReceiver();
bool randomAccess = true; bool randomAccess = true;
bool interTransactions = false; bool interTransactions = false;
int targetPoolSize = MaxAdaptiveExecutorPoolSize; int targetPoolSize = MaxAdaptiveExecutorPoolSize;
@ -668,6 +671,10 @@ AdaptiveExecutor(CitusScanState *scanState)
scanState->tuplestorestate = scanState->tuplestorestate =
tuplestore_begin_heap(randomAccess, interTransactions, work_mem); tuplestore_begin_heap(randomAccess, interTransactions, work_mem);
SetTuplestoreDestReceiverParams(tupleStoreDestReceiever,
scanState->tuplestorestate,
CurrentMemoryContext, false);
TransactionProperties xactProperties = DecideTransactionPropertiesForTaskList( TransactionProperties xactProperties = DecideTransactionPropertiesForTaskList(
distributedPlan->modLevel, taskList, distributedPlan->modLevel, taskList,
hasDependentJobs); hasDependentJobs);
@ -679,7 +686,7 @@ AdaptiveExecutor(CitusScanState *scanState)
distributedPlan->hasReturning, distributedPlan->hasReturning,
paramListInfo, paramListInfo,
tupleDescriptor, tupleDescriptor,
scanState->tuplestorestate, tupleStoreDestReceiever,
targetPoolSize, targetPoolSize,
&xactProperties, &xactProperties,
jobIdList); jobIdList);
@ -889,9 +896,14 @@ ExecuteTaskListExtended(RowModifyLevel modLevel, List *taskList,
targetPoolSize = 1; targetPoolSize = 1;
} }
DestReceiver *tupleStoreDestReceiever = CreateTuplestoreDestReceiver();
SetTuplestoreDestReceiverParams(tupleStoreDestReceiever,
tupleStore,
CurrentMemoryContext, false);
DistributedExecution *execution = DistributedExecution *execution =
CreateDistributedExecution(modLevel, taskList, hasReturning, paramListInfo, CreateDistributedExecution(modLevel, taskList, hasReturning, paramListInfo,
tupleDescriptor, tupleStore, targetPoolSize, tupleDescriptor, tupleStoreDestReceiever, targetPoolSize,
xactProperties, jobIdList); xactProperties, jobIdList);
StartDistributedExecution(execution); StartDistributedExecution(execution);
@ -910,7 +922,7 @@ static DistributedExecution *
CreateDistributedExecution(RowModifyLevel modLevel, List *taskList, CreateDistributedExecution(RowModifyLevel modLevel, List *taskList,
bool hasReturning, bool hasReturning,
ParamListInfo paramListInfo, TupleDesc tupleDescriptor, ParamListInfo paramListInfo, TupleDesc tupleDescriptor,
Tuplestorestate *tupleStore, int targetPoolSize, DestReceiver *destReceiver, int targetPoolSize,
TransactionProperties *xactProperties, List *jobIdList) TransactionProperties *xactProperties, List *jobIdList)
{ {
DistributedExecution *execution = DistributedExecution *execution =
@ -928,7 +940,7 @@ CreateDistributedExecution(RowModifyLevel modLevel, List *taskList,
(DistributedExecutionStats *) palloc0(sizeof(DistributedExecutionStats)); (DistributedExecutionStats *) palloc0(sizeof(DistributedExecutionStats));
execution->paramListInfo = paramListInfo; execution->paramListInfo = paramListInfo;
execution->tupleDescriptor = tupleDescriptor; execution->tupleDescriptor = tupleDescriptor;
execution->tupleStore = tupleStore; execution->destReceiver = destReceiver;
execution->workerList = NIL; execution->workerList = NIL;
execution->sessionList = NIL; execution->sessionList = NIL;
@ -1114,6 +1126,8 @@ StartDistributedExecution(DistributedExecution *execution)
{ {
RecordParallelRelationAccessForTaskList(execution->tasksToExecute); RecordParallelRelationAccessForTaskList(execution->tasksToExecute);
} }
//execution->destReceiver->rStartup(execution->destReceiver, , execution->tupleDescriptor);
} }
@ -1472,6 +1486,7 @@ FinishDistributedExecution(DistributedExecution *execution)
/* prevent copying shards in same transaction */ /* prevent copying shards in same transaction */
XactModificationLevel = XACT_MODIFICATION_DATA; XactModificationLevel = XACT_MODIFICATION_DATA;
} }
execution->destReceiver->rDestroy(execution->destReceiver);
} }
@ -3204,7 +3219,7 @@ ReceiveResults(WorkerSession *session, bool storeRows)
AttInMetadata *attributeInputMetadata = execution->attributeInputMetadata; AttInMetadata *attributeInputMetadata = execution->attributeInputMetadata;
uint32 expectedColumnCount = 0; uint32 expectedColumnCount = 0;
char **columnArray = execution->columnArray; char **columnArray = execution->columnArray;
Tuplestorestate *tupleStore = execution->tupleStore; DestReceiver *destReceiver = execution->destReceiver;
if (tupleDescriptor != NULL) if (tupleDescriptor != NULL)
{ {
@ -3328,8 +3343,7 @@ ReceiveResults(WorkerSession *session, bool storeRows)
columnArray); columnArray);
MemoryContextSwitchTo(oldContextPerRow); MemoryContextSwitchTo(oldContextPerRow);
//destReceiver->receiveSlot(TupleDescGetSlot(), destReceiver);
tuplestore_puttuple(tupleStore, heapTuple);
MemoryContextReset(ioContext); MemoryContextReset(ioContext);
execution->rowsProcessed++; execution->rowsProcessed++;