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

add command execution -2

replace_m_row_insert_select
Onder Kalaci 2020-05-08 15:52:09 +02:00
parent 075940f9d2
commit acd661b715
3 changed files with 16 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/backend/distributed/executor/insert_select_executor.c
View File

@ -128,7 +128,7 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
EState *executorState = ScanStateGetExecutorState(scanState); EState *executorState = ScanStateGetExecutorState(scanState);
ParamListInfo paramListInfo = executorState->es_param_list_info; ParamListInfo paramListInfo = executorState->es_param_list_info;
DistributedPlan *distributedPlan = scanState->distributedPlan; DistributedPlan *distributedPlan = scanState->distributedPlan;
Query *insertSelectQuery = copyObject(distributedPlan->insertSelectQuery); Query *insertSelectQuery = (distributedPlan->insertSelectQuery);
List *insertTargetList = insertSelectQuery->targetList; List *insertTargetList = insertSelectQuery->targetList;
RangeTblEntry *selectRte = ExtractSelectRangeTableEntry(insertSelectQuery); RangeTblEntry *selectRte = ExtractSelectRangeTableEntry(insertSelectQuery);
RangeTblEntry *insertRte = ExtractResultRelationRTE(insertSelectQuery); RangeTblEntry *insertRte = ExtractResultRelationRTE(insertSelectQuery);
@ -175,10 +175,6 @@ CoordinatorInsertSelectExecScanInternal(CustomScanState *node)
// paramListInfo); // paramListInfo);
//elog(ERROR, "selectQuery after: %s", pretty_format_node_dump(nodeToString(selectQuery))); //elog(ERROR, "selectQuery after: %s", pretty_format_node_dump(nodeToString(selectQuery)));
Form_pg_attribute att_tup0 = TupleDescAttr(node->ss.ss_ScanTupleSlot->tts_tupleDescriptor, 0);
Form_pg_attribute att_tup1 = TupleDescAttr(node->ss.ss_ScanTupleSlot->tts_tupleDescriptor, 1);
Form_pg_attribute att_tup2 = TupleDescAttr(node->ss.ss_ScanTupleSlot->tts_tupleDescriptor, 2);
//elog(INFO, "%d-%d-%d", att_tup0->atttypid,att_tup1->atttypid,att_tup2->atttypid); //elog(INFO, "%d-%d-%d", att_tup0->atttypid,att_tup1->atttypid,att_tup2->atttypid);
PlannedStmt *selectPlan = GenerateValuesPlannedStmt(selectQuery); PlannedStmt *selectPlan = GenerateValuesPlannedStmt(selectQuery);
@ -442,7 +438,7 @@ GenerateValuesPlannedStmt(Query *parse)
/* there is only a single relation rte */ /* there is only a single relation rte */
valuesScanNode->values_lists = rte2->values_lists; valuesScanNode->values_lists = rte2->values_lists;
valuesScanNode->values_lists = (List *) eval_const_expressions(NULL, (Node *) valuesScanNode->values_lists); //valuesScanNode->values_lists = (List *) eval_const_expressions(NULL, (Node *) valuesScanNode->values_lists);
plan->targetlist = plan->targetlist =
copyObject(FetchStatementTargetList((Node *) parse)); copyObject(FetchStatementTargetList((Node *) parse));

7
src/backend/distributed/planner/distributed_planner.c
View File

@ -407,17 +407,16 @@ GetSingleValue(Query *insertSelectQuery)
List *valuesLists = valuesRte->values_lists; List *valuesLists = valuesRte->values_lists;
Node *firstValue = linitial(valuesLists); Node *firstValue = linitial(valuesLists);
valuesRte->values_lists = list_make1(firstValue); valuesRte->values_lists = NIL;
Query *copyQuery = copyObject(insertSelectQuery); Query *copyQuery = copyObject(insertSelectQuery);
valuesRte->values_lists = valuesLists;
RangeTblEntry *copySubqueryRte = ExtractSelectRangeTableEntry(copyQuery); RangeTblEntry *copySubqueryRte = ExtractSelectRangeTableEntry(copyQuery);
RangeTblEntry *valuesRteInCopy = linitial(copySubqueryRte->subquery->rtable); RangeTblEntry *valuesRteInCopy = linitial(copySubqueryRte->subquery->rtable);
valuesRteInCopy->values_lists = list_make1(firstValue); valuesRteInCopy->values_lists = list_make1(firstValue);
valuesRte->values_lists = valuesLists;
return copyQuery; return copyQuery;
} }

20
src/backend/distributed/planner/insert_select_planner.c
View File

@ -327,14 +327,6 @@ DistributedInsertSelectSupported(Query *queryTree, RangeTblEntry *insertRte,
Query *subquery = subqueryRte->subquery; Query *subquery = subqueryRte->subquery;
if (!NeedsDistributedPlanning(subquery))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"distributed INSERT ... SELECT can only select from "
"distributed tables",
NULL, NULL);
}
/* we do not expect to see a view in modify target */ /* we do not expect to see a view in modify target */
foreach(rangeTableCell, queryTree->rtable) foreach(rangeTableCell, queryTree->rtable)
{ {
@ -348,6 +340,16 @@ DistributedInsertSelectSupported(Query *queryTree, RangeTblEntry *insertRte,
} }
} }
if (false && !NeedsDistributedPlanning(subquery))
{
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
"distributed INSERT ... SELECT can only select from "
"distributed tables",
NULL, NULL);
}
if (FindNodeCheck((Node *) queryTree, CitusIsVolatileFunction)) if (FindNodeCheck((Node *) queryTree, CitusIsVolatileFunction))
{ {
return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED, return DeferredError(ERRCODE_FEATURE_NOT_SUPPORTED,
@ -1118,7 +1120,7 @@ InsertPartitionColumnMatchesSelect(Query *query, RangeTblEntry *insertRte,
static DistributedPlan * static DistributedPlan *
CreateCoordinatorInsertSelectPlan(uint64 planId, Query *parse) CreateCoordinatorInsertSelectPlan(uint64 planId, Query *parse)
{ {
Query *insertSelectQuery = copyObject(parse); Query *insertSelectQuery = (parse);
RangeTblEntry *insertRte = ExtractResultRelationRTE(insertSelectQuery); RangeTblEntry *insertRte = ExtractResultRelationRTE(insertSelectQuery);
Oid targetRelationId = insertRte->relid; Oid targetRelationId = insertRte->relid;