src/backend/cdb/cdbsharedstorageop.c - hawq - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 /*-------------------------------------------------------------------------
  *
  * cdbsharedstorageop.c
  *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"

 #include "access/appendonlytid.h"
 #include "catalog/heap.h"
 #include "cdb/cdbdisp.h"
 #include "cdb/cdbmirroredappendonly.h"
 #include "cdb/cdbmirroredfilesysobj.h"
 #include "cdb/cdbsharedstorageop.h"
 #include "cdb/cdbdispatchresult.h"
 #include "cdb/dispatcher.h"
 #include "cdb/cdbsrlz.h"
 #include "cdb/cdbvars.h"
 #include "cdb/cdbpartition.h"
 #include "nodes/parsenodes.h"
 #include "optimizer/prep.h"
 #include "storage/lmgr.h"

 /*
  * Create a sharedStorageOpTask instance
  */
 SharedStorageOpTasks *CreateSharedStorageOpTasks(void) {
   SharedStorageOpTasks *retval;
   retval = palloc(sizeof(SharedStorageOpTasks));
   retval->numTasks = 0;
   retval->sizeTasks = 1;
   retval->tasks = palloc0(sizeof(SharedStorageOpTask));
   return retval;
 }

 /*
  * Destory a sharedStorageOpTask instance
  */
 void DropSharedStorageOpTasks(SharedStorageOpTasks *task) {
   if (NULL != task) {
     if (task->tasks) {
       int i;
       for (i = 0; i < task->numTasks; ++i) {
         if (task->tasks[i].relname) {
           pfree(task->tasks[i].relname);
         }
       }
       pfree(task->tasks);
     }
     pfree(task);
   }
 }

 /*
  * Before add a task, add the operation to delete pending.
  */
 void SharedStorageOpPreAddTask(RelFileNode *relFileNode, int32 segmentFileNum,
                                const char *relationName,
                                ItemPointer persistentTid,
                                int64 *persistentSerialNum) {

   smgrcreatepending(relFileNode, segmentFileNum,
                     PersistentFileSysRelStorageMgr_AppendOnly,
                     PersistentFileSysRelBufpoolKind_None, (char *) relationName,
                     persistentTid, persistentSerialNum,
                     /*isLocalBuf*/false,
                     /*bufferPoolBulkLoad*/false,
                     /*flushToXLog*/true);
 }

 /*
  * add a task, it will be performed on all segments.
  */
 void SharedStorageOpAddTask(const char * relname, RelFileNode *node,
                             int32 segno, ItemPointer persistentTid,
                             int64 persistentSerialNum,
                             SharedStorageOpTasks *tasks) {
   Assert(NULL != node && NULL != tasks);
   Assert(tasks->sizeTasks >= tasks->numTasks);

   RelFileNode *n;

   if (tasks->sizeTasks == tasks->numTasks) {
     tasks->tasks = repalloc(tasks->tasks,
                             tasks->sizeTasks * sizeof(SharedStorageOpTask) * 2);
     tasks->sizeTasks *= 2;
   }

   n = &tasks->tasks[tasks->numTasks].node;
   n->dbNode = node->dbNode;
   n->relNode = node->relNode;
   n->spcNode = node->spcNode;

   tasks->tasks[tasks->numTasks].segno = segno;
   tasks->tasks[tasks->numTasks].relname = palloc(strlen(relname) + 1);
   strcpy(tasks->tasks[tasks->numTasks].relname, relname);
   ItemPointerCopy(persistentTid, &tasks->tasks[tasks->numTasks].persistentTid);
   tasks->tasks[tasks->numTasks].persistentSerialNum = persistentSerialNum;

   tasks->numTasks++;
 }

 /*
  * Do the tasks on all segments.
  */
 void PerformSharedStorageOpTasks(SharedStorageOpTasks *tasks,
                                  enum SharedStorageOp op) {
   int i, j = 0;

   if (tasks->numTasks == 0)
     return;

   SharedStorageOpStmt *stat = makeNode(SharedStorageOpStmt);

   stat->op = op;

   stat->relFileNode = palloc(sizeof(RelFileNode) * tasks->numTasks);
   stat->segmentFileNum = palloc(sizeof(int) * tasks->numTasks);
   stat->relationName = palloc(sizeof(char *) * tasks->numTasks);

   QueryContextInfo *contextdisp = CreateQueryContextInfo();

   for (i = 0, j = 0; i < tasks->numTasks; ++i) {
     SharedStorageOpTask *task = tasks->tasks + i;
     prepareDispatchedCatalogTablespace(contextdisp, task->node.spcNode);
     stat->relFileNode[j] = task->node;
     stat->segmentFileNum[j] = task->segno;
     stat->relationName[j] = task->relname;

     ++j;
   }

   stat->numTasks = j;

   FinalizeQueryContextInfo(contextdisp);

   int gp_segments_for_planner_before = gp_segments_for_planner;

   QueryResource *resource =
       AllocateResource(QRL_INHERIT, 0, 0, tasks->numTasks, 1, NULL, 0);
   DispatchDataResult result;
   dispatch_statement_node((Node *) stat, contextdisp, resource, &result);
   dispatch_free_result(&result);
   FreeResource(resource);
   DropQueryContextInfo(contextdisp);

   pfree(stat->relFileNode);
   pfree(stat->relationName);
   pfree(stat->segmentFileNum);
   pfree(stat);

   gp_segments_for_planner = gp_segments_for_planner_before;
 }

 /*
  * Do all the tasks on masters
  */
 void PerformSharedStorageOpTasksOnMaster(SharedStorageOpTasks *tasks,
                                          enum SharedStorageOp op) {
   int i;

   switch (op) {
     case Op_CreateSegFile:
       for (i = 0; i < tasks->numTasks; i++) {
         int error = 0;
         MirroredAppendOnly_Create(&tasks->tasks[i].node, tasks->tasks[i].segno,
                                   tasks->tasks[i].relname, &error);

         if (error != 0) {
           ereport(
               ERROR,
               (errcode_for_file_access(), errmsg("could not create relation file '%s', relation name '%s', segno '%d': %s", relpath(tasks->tasks[i].node), tasks->tasks[i].relname, tasks->tasks[i].segno, strerror(error))));
         }
       }
       break;
     case Op_OverWriteSegFile:
       for (i = 0; i < tasks->numTasks; i++) {
         int error = 0;
         AppendOnly_Overwrite(&tasks->tasks[i].node, tasks->tasks[i].segno,
                              &error);

         if (error != 0) {
           ereport(
               ERROR,
               (errcode_for_file_access(), errmsg("could not create relation with overwrite mode: file '%s', relation name '%s', segno '%d': %s", relpath(tasks->tasks[i].node), tasks->tasks[i].relname, tasks->tasks[i].segno, strerror(error))));
         }
       }
       break;
     default:
       Insist(!"unexpected shared storage op type.");
   }
 }

 /*
  * after shared storage operation complete, update persistent table.
  */
 void PostPerformSharedStorageOpTasks(SharedStorageOpTasks *tasks) {
   int i;

   Relation gp_relation_node;

   Assert(NULL != tasks);

   if (tasks->numTasks == 0)
     return;

   gp_relation_node = heap_open(GpRelfileNodeRelationId, RowExclusiveLock);

   for (i = 0; i < tasks->numTasks; ++i) {
     SharedStorageOpTask * task = &tasks->tasks[i];
     InsertGpRelfileNodeTuple(gp_relation_node,
     /* relationId */0,
                              task->relname, task->node.relNode, task->segno,
                              /* updateIndex */true,
                              &task->persistentTid, task->persistentSerialNum);
   }
   heap_close(gp_relation_node, RowExclusiveLock);
 }

 /*
  * Create db directory if necessary.
  */
 void TransactionCreateDatabaseDir(Oid dbid, Oid tsp) {
   DbDirNode justInTimeDbDirNode;

   /* "Fault-in" the database directory in a tablespace if it doesn't exist yet.*/

   justInTimeDbDirNode.tablespace = tsp;
   justInTimeDbDirNode.database = dbid;

   Assert(Gp_role != GP_ROLE_EXECUTE);

   MirroredFileSysObj_JustInTimeDbDirCreate(&justInTimeDbDirNode);
 }

 /*
  * Execute shared storage operation on segments.
  */
 void PerformSharedStorageOp(SharedStorageOpStmt * stat) {
   Assert(NULL != stat);

   int i;
   int error = 0;
   Assert(Gp_role != GP_ROLE_DISPATCH);

   switch (stat->op) {
     case Op_CreateSegFile:
       for (i = 0; i < stat->numTasks; ++i) {
         /*
          * This interface should pass a state object to let the following
          * code access task.
          */
         if (GpIdentity.segindex == i) {
           MirroredAppendOnly_Create(&stat->relFileNode[i],
                                     stat->segmentFileNum[i],
                                     stat->relationName[i], &error);

           if (error != 0) {
             ereport(
                 ERROR,
                 (errcode_for_file_access(), errmsg("could not create relation file '%s', relation name '%s': %s", relpath(stat->relFileNode[i]), stat->relationName[i], strerror(error)), errdetail("%s", HdfsGetLastError())));
           }
         }
       }
       break;
     case Op_OverWriteSegFile:
       for (i = 0; i < stat->numTasks; ++i) {
         /*
          * This interface should pass a state object to let the following
          * code access task.
          */
         if (GpIdentity.segindex == i) {
           AppendOnly_Overwrite(&stat->relFileNode[i], stat->segmentFileNum[i],
                                &error);

           if (error != 0) {
             ereport(
                 ERROR,
                 (errcode_for_file_access(), errmsg("could not create relation with overwrite mode: file '%s', relation name '%s': %s", relpath(stat->relFileNode[i]), stat->relationName[i], strerror(error)), errdetail("%s", HdfsGetLastError())));
           }
         }
       }
       break;
     default:
       Insist(!"unexpected shared storage op type.");
   }
 }

 static void LockSegfilesOnMasterForSingleRel(Relation rel, int32 segno) {
   Insist(Gp_role == GP_ROLE_DISPATCH);

   /*
    * do not lock segfile with content id = -1
    */
   /*
    for (i = 1; i < rel->rd_segfile0_count; ++i)
    {
    if (RelationIsAoRows(rel) || RelationIsParquet(rel))
    {
    LockRelationAppendOnlySegmentFile(&rel->rd_node, segno,
    AccessExclusiveLock, false, i - 1);
    }
    }
    */
   {
     if (RelationIsAoRows(rel) || RelationIsParquet(rel)) {
       LockRelationAppendOnlySegmentFile(&rel->rd_node, segno,
       AccessExclusiveLock,
                                         false);
     }
   }

 }

 void LockSegfilesOnMaster(Relation rel, int32 segno) {
   List *children = NIL;
   ListCell *lc;

   children = find_all_inheritors(rel->rd_id);

   foreach(lc, children)
   {
     Relation child_rel = rel;
     Oid oid = lfirst_oid(lc);

     if (rel_is_partitioned(oid))
       continue;

     /* open/close the child relation. */
     if (oid != rel->rd_id)
       child_rel = heap_open(oid, AccessShareLock);

     LockSegfilesOnMasterForSingleRel(child_rel, segno);

     if (oid != rel->rd_id)
       heap_close(child_rel, AccessShareLock);
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	/*-------------------------------------------------------------------------
	*
	* cdbsharedstorageop.c
	*
	*-------------------------------------------------------------------------
	*/
	#include "postgres.h"

	#include "access/appendonlytid.h"
	#include "catalog/heap.h"
	#include "cdb/cdbdisp.h"
	#include "cdb/cdbmirroredappendonly.h"
	#include "cdb/cdbmirroredfilesysobj.h"
	#include "cdb/cdbsharedstorageop.h"
	#include "cdb/cdbdispatchresult.h"
	#include "cdb/dispatcher.h"
	#include "cdb/cdbsrlz.h"
	#include "cdb/cdbvars.h"
	#include "cdb/cdbpartition.h"
	#include "nodes/parsenodes.h"
	#include "optimizer/prep.h"
	#include "storage/lmgr.h"

	/*
	* Create a sharedStorageOpTask instance
	*/
	SharedStorageOpTasks *CreateSharedStorageOpTasks(void) {
	SharedStorageOpTasks *retval;
	retval = palloc(sizeof(SharedStorageOpTasks));
	retval->numTasks = 0;
	retval->sizeTasks = 1;
	retval->tasks = palloc0(sizeof(SharedStorageOpTask));
	return retval;
	}

	/*
	* Destory a sharedStorageOpTask instance
	*/
	void DropSharedStorageOpTasks(SharedStorageOpTasks *task) {
	if (NULL != task) {
	if (task->tasks) {
	int i;
	for (i = 0; i < task->numTasks; ++i) {
	if (task->tasks[i].relname) {
	pfree(task->tasks[i].relname);
	}
	}
	pfree(task->tasks);
	}
	pfree(task);
	}
	}

	/*
	* Before add a task, add the operation to delete pending.
	*/
	void SharedStorageOpPreAddTask(RelFileNode *relFileNode, int32 segmentFileNum,
	const char *relationName,
	ItemPointer persistentTid,
	int64 *persistentSerialNum) {

	smgrcreatepending(relFileNode, segmentFileNum,
	PersistentFileSysRelStorageMgr_AppendOnly,
	PersistentFileSysRelBufpoolKind_None, (char *) relationName,
	persistentTid, persistentSerialNum,
	/isLocalBuf/false,
	/bufferPoolBulkLoad/false,
	/flushToXLog/true);
	}

	/*
	* add a task, it will be performed on all segments.
	*/
	void SharedStorageOpAddTask(const char * relname, RelFileNode *node,
	int32 segno, ItemPointer persistentTid,
	int64 persistentSerialNum,
	SharedStorageOpTasks *tasks) {
	Assert(NULL != node && NULL != tasks);
	Assert(tasks->sizeTasks >= tasks->numTasks);

	RelFileNode *n;

	if (tasks->sizeTasks == tasks->numTasks) {
	tasks->tasks = repalloc(tasks->tasks,
	tasks->sizeTasks * sizeof(SharedStorageOpTask) * 2);
	tasks->sizeTasks *= 2;
	}

	n = &tasks->tasks[tasks->numTasks].node;
	n->dbNode = node->dbNode;
	n->relNode = node->relNode;
	n->spcNode = node->spcNode;

	tasks->tasks[tasks->numTasks].segno = segno;
	tasks->tasks[tasks->numTasks].relname = palloc(strlen(relname) + 1);
	strcpy(tasks->tasks[tasks->numTasks].relname, relname);
	ItemPointerCopy(persistentTid, &tasks->tasks[tasks->numTasks].persistentTid);
	tasks->tasks[tasks->numTasks].persistentSerialNum = persistentSerialNum;

	tasks->numTasks++;
	}

	/*
	* Do the tasks on all segments.
	*/
	void PerformSharedStorageOpTasks(SharedStorageOpTasks *tasks,
	enum SharedStorageOp op) {
	int i, j = 0;

	if (tasks->numTasks == 0)
	return;

	SharedStorageOpStmt *stat = makeNode(SharedStorageOpStmt);

	stat->op = op;

	stat->relFileNode = palloc(sizeof(RelFileNode) * tasks->numTasks);
	stat->segmentFileNum = palloc(sizeof(int) * tasks->numTasks);
	stat->relationName = palloc(sizeof(char ) tasks->numTasks);

	QueryContextInfo *contextdisp = CreateQueryContextInfo();

	for (i = 0, j = 0; i < tasks->numTasks; ++i) {
	SharedStorageOpTask *task = tasks->tasks + i;
	prepareDispatchedCatalogTablespace(contextdisp, task->node.spcNode);
	stat->relFileNode[j] = task->node;
	stat->segmentFileNum[j] = task->segno;
	stat->relationName[j] = task->relname;

	++j;
	}

	stat->numTasks = j;

	FinalizeQueryContextInfo(contextdisp);

	int gp_segments_for_planner_before = gp_segments_for_planner;

	QueryResource *resource =
	AllocateResource(QRL_INHERIT, 0, 0, tasks->numTasks, 1, NULL, 0);
	DispatchDataResult result;
	dispatch_statement_node((Node *) stat, contextdisp, resource, &result);
	dispatch_free_result(&result);
	FreeResource(resource);
	DropQueryContextInfo(contextdisp);

	pfree(stat->relFileNode);
	pfree(stat->relationName);
	pfree(stat->segmentFileNum);
	pfree(stat);

	gp_segments_for_planner = gp_segments_for_planner_before;
	}

	/*
	* Do all the tasks on masters
	*/
	void PerformSharedStorageOpTasksOnMaster(SharedStorageOpTasks *tasks,
	enum SharedStorageOp op) {
	int i;

	switch (op) {
	case Op_CreateSegFile:
	for (i = 0; i < tasks->numTasks; i++) {
	int error = 0;
	MirroredAppendOnly_Create(&tasks->tasks[i].node, tasks->tasks[i].segno,
	tasks->tasks[i].relname, &error);

	if (error != 0) {
	ereport(
	ERROR,
	(errcode_for_file_access(), errmsg("could not create relation file '%s', relation name '%s', segno '%d': %s", relpath(tasks->tasks[i].node), tasks->tasks[i].relname, tasks->tasks[i].segno, strerror(error))));
	}
	}
	break;
	case Op_OverWriteSegFile:
	for (i = 0; i < tasks->numTasks; i++) {
	int error = 0;
	AppendOnly_Overwrite(&tasks->tasks[i].node, tasks->tasks[i].segno,
	&error);

	if (error != 0) {
	ereport(
	ERROR,
	(errcode_for_file_access(), errmsg("could not create relation with overwrite mode: file '%s', relation name '%s', segno '%d': %s", relpath(tasks->tasks[i].node), tasks->tasks[i].relname, tasks->tasks[i].segno, strerror(error))));
	}
	}
	break;
	default:
	Insist(!"unexpected shared storage op type.");
	}
	}

	/*
	* after shared storage operation complete, update persistent table.
	*/
	void PostPerformSharedStorageOpTasks(SharedStorageOpTasks *tasks) {
	int i;

	Relation gp_relation_node;

	Assert(NULL != tasks);

	if (tasks->numTasks == 0)
	return;

	gp_relation_node = heap_open(GpRelfileNodeRelationId, RowExclusiveLock);

	for (i = 0; i < tasks->numTasks; ++i) {
	SharedStorageOpTask * task = &tasks->tasks[i];
	InsertGpRelfileNodeTuple(gp_relation_node,
	/* relationId */0,
	task->relname, task->node.relNode, task->segno,
	/* updateIndex */true,
	&task->persistentTid, task->persistentSerialNum);
	}
	heap_close(gp_relation_node, RowExclusiveLock);
	}

	/*
	* Create db directory if necessary.
	*/
	void TransactionCreateDatabaseDir(Oid dbid, Oid tsp) {
	DbDirNode justInTimeDbDirNode;

	/* "Fault-in" the database directory in a tablespace if it doesn't exist yet.*/

	justInTimeDbDirNode.tablespace = tsp;
	justInTimeDbDirNode.database = dbid;

	Assert(Gp_role != GP_ROLE_EXECUTE);

	MirroredFileSysObj_JustInTimeDbDirCreate(&justInTimeDbDirNode);
	}

	/*
	* Execute shared storage operation on segments.
	*/
	void PerformSharedStorageOp(SharedStorageOpStmt * stat) {
	Assert(NULL != stat);

	int i;
	int error = 0;
	Assert(Gp_role != GP_ROLE_DISPATCH);

	switch (stat->op) {
	case Op_CreateSegFile:
	for (i = 0; i < stat->numTasks; ++i) {
	/*
	* This interface should pass a state object to let the following
	* code access task.
	*/
	if (GpIdentity.segindex == i) {
	MirroredAppendOnly_Create(&stat->relFileNode[i],
	stat->segmentFileNum[i],
	stat->relationName[i], &error);

	if (error != 0) {
	ereport(
	ERROR,
	(errcode_for_file_access(), errmsg("could not create relation file '%s', relation name '%s': %s", relpath(stat->relFileNode[i]), stat->relationName[i], strerror(error)), errdetail("%s", HdfsGetLastError())));
	}
	}
	}
	break;
	case Op_OverWriteSegFile:
	for (i = 0; i < stat->numTasks; ++i) {
	/*
	* This interface should pass a state object to let the following
	* code access task.
	*/
	if (GpIdentity.segindex == i) {
	AppendOnly_Overwrite(&stat->relFileNode[i], stat->segmentFileNum[i],
	&error);

	if (error != 0) {
	ereport(
	ERROR,
	(errcode_for_file_access(), errmsg("could not create relation with overwrite mode: file '%s', relation name '%s': %s", relpath(stat->relFileNode[i]), stat->relationName[i], strerror(error)), errdetail("%s", HdfsGetLastError())));
	}
	}
	}
	break;
	default:
	Insist(!"unexpected shared storage op type.");
	}
	}

	static void LockSegfilesOnMasterForSingleRel(Relation rel, int32 segno) {
	Insist(Gp_role == GP_ROLE_DISPATCH);

	/*
	* do not lock segfile with content id = -1
	*/
	/*
	for (i = 1; i < rel->rd_segfile0_count; ++i)
	{
	if (RelationIsAoRows(rel) \|\| RelationIsParquet(rel))
	{
	LockRelationAppendOnlySegmentFile(&rel->rd_node, segno,
	AccessExclusiveLock, false, i - 1);
	}
	}
	*/
	{
	if (RelationIsAoRows(rel) \|\| RelationIsParquet(rel)) {
	LockRelationAppendOnlySegmentFile(&rel->rd_node, segno,
	AccessExclusiveLock,
	false);
	}
	}

	}

	void LockSegfilesOnMaster(Relation rel, int32 segno) {
	List *children = NIL;
	ListCell *lc;

	children = find_all_inheritors(rel->rd_id);

	foreach(lc, children)
	{
	Relation child_rel = rel;
	Oid oid = lfirst_oid(lc);

	if (rel_is_partitioned(oid))
	continue;

	/* open/close the child relation. */
	if (oid != rel->rd_id)
	child_rel = heap_open(oid, AccessShareLock);

	LockSegfilesOnMasterForSingleRel(child_rel, segno);

	if (oid != rel->rd_id)
	heap_close(child_rel, AccessShareLock);
	}
	}