src/backend/cdb/cdbmirroredappendonly.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.
  */

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

 #include <unistd.h>
 #include <signal.h>
 #include <fcntl.h>
 #include <sys/file.h>

 #include "access/xlogmm.h"
 #include "utils/palloc.h"
 #include "cdb/cdbmirroredappendonly.h"
 #include "storage/fd.h"
 #include "catalog/catalog.h"
 #include "cdb/cdbpersistenttablespace.h"
 #include "storage/smgr.h"
 #include "storage/lwlock.h"
 #include "utils/guc.h"
 #include "cdb/cdbpersistentfilesysobj.h"
 #include "cdb/cdbpersistentstore.h"
 #include "cdb/cdbpersistentrecovery.h"
 #include "cdb/cdbvars.h"
 #include "cdb/cdbmetadatacache.h"

 /*
  * Open a relation for mirrored write.
  */
 static void MirroredAppendOnly_DoOpen(
 	MirroredAppendOnlyOpen 		*open,
 			/* The resulting open struct. */

 	RelFileNode					*relFileNode,
 			/* The tablespace, database, and relation OIDs for the open. */

 	int32						segmentFileNum,

 	char						*relationName,
 					/* For tracing only.  Can be NULL in some execution paths. */

 	int64						logicalEof,
 				/* The file name path. */

 	bool						create,

 	bool						readOnly,

 	int 						*primaryError)
 {
 	int		fileFlags;
 	int		fileMode = 0600;
 						/*
 						 * File mode is S_IRUSR 00400 user has read permission
 						 *               + S_IWUSR 00200 user has write permission
 						 */

 	char *filespaceLocation = NULL;

 	Assert(open != NULL);

 	*primaryError = 0;

 	MemSet(open, 0, sizeof(MirroredAppendOnlyOpen));

 	open->relFileNode = *relFileNode;

 	open->segmentFileNum = segmentFileNum;

 	open->create = create;

 	PersistentTablespace_GetFilespacePath(
 										relFileNode->spcNode,
 										TRUE,
 										&filespaceLocation);

 	char *path;

 	path = (char*)palloc(MAXPGPATH + 1);

 	/*
 	 * Do the primary work first so we don't leave files on the mirror or have an
 	 * open to clean up.
 	 */
 	FormRelfilePath(path,
 					 filespaceLocation,
 					 relFileNode,
 					 segmentFileNum);

 	errno = 0;


 	if (create)
 		fileFlags = O_CREAT | O_SYNC;
 	else if (readOnly)
 		fileFlags = O_RDONLY;
 	else
 		fileFlags = (O_WRONLY | O_APPEND | O_SYNC);

 	open->primaryFile = PathNameOpenFile(path, fileFlags, fileMode);

 	if (open->primaryFile < 0)
 	{
 		*primaryError = errno;
 	}

 	pfree(path);

 	if (*primaryError != 0)
 	{
 		open->isActive = false;
 	}
 	else
 	{
 		open->isActive = true;
 	}

 	if (filespaceLocation != NULL)
 		pfree(filespaceLocation);
 }

 void AppendOnly_Overwrite(RelFileNode *relFileNode, int32 segmentFileNum, int *primaryError)
 {
 	int		fileFlags, file;
 	int		fileMode = 0600;
 	char *primaryFilespaceLocation = NULL;

 	Assert(open != NULL);

 	*primaryError = 0;

 	PersistentTablespace_GetFilespacePath(
 										relFileNode->spcNode,
 										TRUE,
 										&primaryFilespaceLocation);

 	char *path;

 	path = (char*)palloc(MAXPGPATH + 1);

 	FormRelfilePath(path,
 					 primaryFilespaceLocation,
 					 relFileNode,
 					 segmentFileNum);
 	errno = 0;

 	fileFlags = O_WRONLY | O_SYNC;

 	file = PathNameOpenFile(path, fileFlags, fileMode);

 	if (file < 0)
 		*primaryError = errno;

 	pfree(path);

 	if (primaryFilespaceLocation != NULL)
 		pfree(primaryFilespaceLocation);

 	FileClose(file);
 }

 /*
  * Call MirroredAppendOnly_Create with the MirroredLock already held.
  */
 void MirroredAppendOnly_Create(
 	RelFileNode 				*relFileNode,
 				/* The tablespace, database, and relation OIDs for the open. */

 	int32						segmentFileNum,
 				/* Which segment file. */

 	char						*relationName,
 					/* For tracing only.  Can be NULL in some execution paths. */

 	int 						*primaryError)
 {
 	MirroredAppendOnlyOpen mirroredOpen;

 	*primaryError = 0;

 	MirroredAppendOnly_DoOpen(
 					&mirroredOpen,
 					relFileNode,
 					segmentFileNum,
 					relationName,
 					/* logicalEof */ 0,
 					/* create */ true,
 					/* readOnly */ false,
 					primaryError);
 	if (*primaryError != 0)
 		return;

 	MirroredAppendOnly_FlushAndClose(
 							&mirroredOpen,
 							primaryError);
 }

 /*
  * MirroredAppendOnly_OpenReadWrite will acquire and release the MirroredLock.
  *
  * Use aaa after writing data to determine if mirror loss occurred and
  * mirror catchup must be performed.
  */
 void MirroredAppendOnly_OpenReadWrite(
 	MirroredAppendOnlyOpen		*open,
 				/* The resulting open struct. */

 	RelFileNode 				*relFileNode,
 				/* The tablespace, database, and relation OIDs for the open. */

 	int32						segmentFileNum,
 				/* Which segment file. */

 	char						*relationName,
 					/* For tracing only.  Can be NULL in some execution paths. */

 	int64						logicalEof,
 				/* The logical EOF to begin appending the new data. */

 	bool						readOnly,

 	int 						*primaryError)
 {
 	*primaryError = 0;

 	MirroredAppendOnly_DoOpen(
 							open,
 							relFileNode,
 							segmentFileNum,
 							relationName,
 							logicalEof,
 							/* create */ false,
 							/* readOnly */ readOnly,
 							primaryError);
 }

 bool MirroredAppendOnly_IsActive(
 	MirroredAppendOnlyOpen *open)
 					/* The open struct. */
 {
 	return open->isActive;
 }

 /*
  * Flush and close a bulk relation file.
  *
  * If the flush is unable to complete on the mirror, then this relation will be marked in the
  * commit, distributed commit, distributed prepared and commit prepared records as having
  * un-mirrored bulk initial data.
  */
 void MirroredAppendOnly_FlushAndClose(
 	MirroredAppendOnlyOpen 	*open,
 				/* The open struct. */

 	int						*primaryError)

 {
 	Assert(open != NULL);
 	Assert(open->isActive);

 	*primaryError = 0;

 	{
 		int		ret;

 		errno = 0;

 		ret = FileSync(open->primaryFile);
 		if (ret != 0)
 			*primaryError = errno;
 	}

 	FileClose(open->primaryFile);

 	open->isActive = false;
 	open->primaryFile = 0;
 }

 /*
  * Flush and close a bulk relation file.
  *
  * If the flush is unable to complete on the mirror, then this relation will be marked in the
  * commit, distributed commit, distributed prepared and commit prepared records as having
  * un-mirrored bulk initial data.
  */
 void MirroredAppendOnly_Flush(
 	MirroredAppendOnlyOpen 	*open,
 				/* The open struct. */

 	int						*primaryError)
 {
 	Assert(open != NULL);
 	Assert(open->isActive);

 	*primaryError = 0;

 	int		ret;

 	errno = 0;

 	ret = FileSync(open->primaryFile);
 	if (ret != 0)
 	{
 		*primaryError = errno;
 	}
 }

 /*
  * Close a bulk relation file.
  *
  */
 void MirroredAppendOnly_Close(
 	MirroredAppendOnlyOpen 	*open
 				/* The open struct. */)
 {
 	Assert(open != NULL);
 	Assert(open->isActive);

 	// No primary error to report.
 	errno = 0;

 	FileClose(open->primaryFile);

 	open->isActive = false;
 	open->primaryFile = 0;
 }

 static void MirroredAppendOnly_DoDrop(
 	RelFileNode					*relFileNode,

 	int32						segmentFileNum,

 	char						*relationName,
 					/* For tracing only.  Can be NULL in some execution paths. */

 	int							*primaryError)
 {
 	char *filespaceLocation = NULL;

 	*primaryError = 0;

 	PersistentTablespace_GetFilespacePath(relFileNode->spcNode,
 													   TRUE,
 													   &filespaceLocation);

     HdfsFileInfo *file_info;


 	char *path;

 	path = (char*)palloc(MAXPGPATH + 1);

 	FormRelfilePath(path,
 					 filespaceLocation,
 					 relFileNode,
 					 segmentFileNum);

 	errno = 0;

 	if (RemovePath(path, 0) < 0)
 	{
 		*primaryError = errno;
 	}

     if (!IsLocalPath(path) && Gp_role == GP_ROLE_DISPATCH)
     {
         // Remove Hdfs block locations info in Metadata Cache
         file_info = CreateHdfsFileInfo(*relFileNode, segmentFileNum);
         LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);
         RemoveHdfsFileBlockLocations(file_info);
         LWLockRelease(MetadataCacheLock);
         DestroyHdfsFileInfo(file_info);
     }

 	pfree(path);

 	if (filespaceLocation != NULL)
 		pfree(filespaceLocation);

 }

 void MirroredAppendOnly_Drop(
 	RelFileNode					*relFileNode,

 	int32						segmentFileNum,

 	char						*relationName,
 					/* For tracing only.  Can be NULL in some execution paths. */

 	int							*primaryError)
 {
 	MirroredAppendOnly_DoDrop(
 						relFileNode,
 						segmentFileNum,
 						relationName,
 						primaryError);
 }

 // -----------------------------------------------------------------------------
 // Append
 // -----------------------------------------------------------------------------

 /*
  * Write bulk mirrored.
  */
 void MirroredAppendOnly_Append(
 	MirroredAppendOnlyOpen *open,
 				/* The open struct. */

 	void					*buffer,
 				/* Pointer to the buffer. */

 	int32					bufferLen,
 				/* Byte length of buffer. */

 	int 					*primaryError)
 {
 	Assert(open != NULL);
 	Assert(open->isActive);

 	*primaryError = 0;
 	/**mirrorDataLossOccurred = false;*/

 	int	ret;
 	errno = 0;

 	ret = FileWrite(open->primaryFile, buffer, bufferLen);

 	if (ret != bufferLen)
 	{
 		/* if write didn't set errno, assume problem is no disk space */
 		if (errno == 0)
 			errno = ENOSPC;
 		*primaryError = errno;
 	}
 }

 // -----------------------------------------------------------------------------
 // Truncate
 // ----------------------------------------------------------------------------
 void MirroredAppendOnly_Truncate(
 	MirroredAppendOnlyOpen *open,
 				/* The open struct. */

 	int64		position,
 				/* The position to cutoff the data. */

 	int 		*primaryError)
 {
 	*primaryError = 0;
 		errno = 0;

 	if (FileTruncate(open->primaryFile, position) < 0)
 		*primaryError = errno;
 }

 /*
  * Read an append only file in sequential way.
  * The routine is used for Resync.
  */
 int MirroredAppendOnly_Read(
 	MirroredAppendOnlyOpen *open,
 		/* The open struct. */

 	void		*buffer,
 		/* Pointer to the buffer. */

 	int32		bufferLen)
 		/* Byte length of buffer. */

 {
 	int	ret;

 	Assert(open != NULL);
 	Assert(buffer != NULL);
 	Assert(open->isActive);

 	errno = 0;

 	ret = FileRead(open->primaryFile, buffer, bufferLen);

 	return ret;
 }
	/*
	* 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.
	*/

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

	#include <unistd.h>
	#include <signal.h>
	#include <fcntl.h>
	#include <sys/file.h>

	#include "access/xlogmm.h"
	#include "utils/palloc.h"
	#include "cdb/cdbmirroredappendonly.h"
	#include "storage/fd.h"
	#include "catalog/catalog.h"
	#include "cdb/cdbpersistenttablespace.h"
	#include "storage/smgr.h"
	#include "storage/lwlock.h"
	#include "utils/guc.h"
	#include "cdb/cdbpersistentfilesysobj.h"
	#include "cdb/cdbpersistentstore.h"
	#include "cdb/cdbpersistentrecovery.h"
	#include "cdb/cdbvars.h"
	#include "cdb/cdbmetadatacache.h"

	/*
	* Open a relation for mirrored write.
	*/
	static void MirroredAppendOnly_DoOpen(
	MirroredAppendOnlyOpen *open,
	/* The resulting open struct. */

	RelFileNode *relFileNode,
	/* The tablespace, database, and relation OIDs for the open. */

	int32 segmentFileNum,

	char *relationName,
	/* For tracing only. Can be NULL in some execution paths. */

	int64 logicalEof,
	/* The file name path. */

	bool create,

	bool readOnly,

	int *primaryError)
	{
	int fileFlags;
	int fileMode = 0600;
	/*
	* File mode is S_IRUSR 00400 user has read permission
	* + S_IWUSR 00200 user has write permission
	*/

	char *filespaceLocation = NULL;

	Assert(open != NULL);

	*primaryError = 0;

	MemSet(open, 0, sizeof(MirroredAppendOnlyOpen));

	open->relFileNode = *relFileNode;

	open->segmentFileNum = segmentFileNum;

	open->create = create;

	PersistentTablespace_GetFilespacePath(
	relFileNode->spcNode,
	TRUE,
	&filespaceLocation);

	char *path;

	path = (char*)palloc(MAXPGPATH + 1);

	/*
	* Do the primary work first so we don't leave files on the mirror or have an
	* open to clean up.
	*/
	FormRelfilePath(path,
	filespaceLocation,
	relFileNode,
	segmentFileNum);

	errno = 0;


	if (create)
	fileFlags = O_CREAT \| O_SYNC;
	else if (readOnly)
	fileFlags = O_RDONLY;
	else
	fileFlags = (O_WRONLY \| O_APPEND \| O_SYNC);

	open->primaryFile = PathNameOpenFile(path, fileFlags, fileMode);

	if (open->primaryFile < 0)
	{
	*primaryError = errno;
	}

	pfree(path);

	if (*primaryError != 0)
	{
	open->isActive = false;
	}
	else
	{
	open->isActive = true;
	}

	if (filespaceLocation != NULL)
	pfree(filespaceLocation);
	}

	void AppendOnly_Overwrite(RelFileNode relFileNode, int32 segmentFileNum, int primaryError)
	{
	int fileFlags, file;
	int fileMode = 0600;
	char *primaryFilespaceLocation = NULL;

	Assert(open != NULL);

	*primaryError = 0;

	PersistentTablespace_GetFilespacePath(
	relFileNode->spcNode,
	TRUE,
	&primaryFilespaceLocation);

	char *path;

	path = (char*)palloc(MAXPGPATH + 1);

	FormRelfilePath(path,
	primaryFilespaceLocation,
	relFileNode,
	segmentFileNum);
	errno = 0;

	fileFlags = O_WRONLY \| O_SYNC;

	file = PathNameOpenFile(path, fileFlags, fileMode);

	if (file < 0)
	*primaryError = errno;

	pfree(path);

	if (primaryFilespaceLocation != NULL)
	pfree(primaryFilespaceLocation);

	FileClose(file);
	}

	/*
	* Call MirroredAppendOnly_Create with the MirroredLock already held.
	*/
	void MirroredAppendOnly_Create(
	RelFileNode *relFileNode,
	/* The tablespace, database, and relation OIDs for the open. */

	int32 segmentFileNum,
	/* Which segment file. */

	char *relationName,
	/* For tracing only. Can be NULL in some execution paths. */

	int *primaryError)
	{
	MirroredAppendOnlyOpen mirroredOpen;

	*primaryError = 0;

	MirroredAppendOnly_DoOpen(
	&mirroredOpen,
	relFileNode,
	segmentFileNum,
	relationName,
	/* logicalEof */ 0,
	/* create */ true,
	/* readOnly */ false,
	primaryError);
	if (*primaryError != 0)
	return;

	MirroredAppendOnly_FlushAndClose(
	&mirroredOpen,
	primaryError);
	}

	/*
	* MirroredAppendOnly_OpenReadWrite will acquire and release the MirroredLock.
	*
	* Use aaa after writing data to determine if mirror loss occurred and
	* mirror catchup must be performed.
	*/
	void MirroredAppendOnly_OpenReadWrite(
	MirroredAppendOnlyOpen *open,
	/* The resulting open struct. */

	RelFileNode *relFileNode,
	/* The tablespace, database, and relation OIDs for the open. */

	int32 segmentFileNum,
	/* Which segment file. */

	char *relationName,
	/* For tracing only. Can be NULL in some execution paths. */

	int64 logicalEof,
	/* The logical EOF to begin appending the new data. */

	bool readOnly,

	int *primaryError)
	{
	*primaryError = 0;

	MirroredAppendOnly_DoOpen(
	open,
	relFileNode,
	segmentFileNum,
	relationName,
	logicalEof,
	/* create */ false,
	/* readOnly */ readOnly,
	primaryError);
	}

	bool MirroredAppendOnly_IsActive(
	MirroredAppendOnlyOpen *open)
	/* The open struct. */
	{
	return open->isActive;
	}

	/*
	* Flush and close a bulk relation file.
	*
	* If the flush is unable to complete on the mirror, then this relation will be marked in the
	* commit, distributed commit, distributed prepared and commit prepared records as having
	* un-mirrored bulk initial data.
	*/
	void MirroredAppendOnly_FlushAndClose(
	MirroredAppendOnlyOpen *open,
	/* The open struct. */

	int *primaryError)

	{
	Assert(open != NULL);
	Assert(open->isActive);

	*primaryError = 0;

	{
	int ret;

	errno = 0;

	ret = FileSync(open->primaryFile);
	if (ret != 0)
	*primaryError = errno;
	}

	FileClose(open->primaryFile);

	open->isActive = false;
	open->primaryFile = 0;
	}

	/*
	* Flush and close a bulk relation file.
	*
	* If the flush is unable to complete on the mirror, then this relation will be marked in the
	* commit, distributed commit, distributed prepared and commit prepared records as having
	* un-mirrored bulk initial data.
	*/
	void MirroredAppendOnly_Flush(
	MirroredAppendOnlyOpen *open,
	/* The open struct. */

	int *primaryError)
	{
	Assert(open != NULL);
	Assert(open->isActive);

	*primaryError = 0;

	int ret;

	errno = 0;

	ret = FileSync(open->primaryFile);
	if (ret != 0)
	{
	*primaryError = errno;
	}
	}

	/*
	* Close a bulk relation file.
	*
	*/
	void MirroredAppendOnly_Close(
	MirroredAppendOnlyOpen *open
	/* The open struct. */)
	{
	Assert(open != NULL);
	Assert(open->isActive);

	// No primary error to report.
	errno = 0;

	FileClose(open->primaryFile);

	open->isActive = false;
	open->primaryFile = 0;
	}

	static void MirroredAppendOnly_DoDrop(
	RelFileNode *relFileNode,

	int32 segmentFileNum,

	char *relationName,
	/* For tracing only. Can be NULL in some execution paths. */

	int *primaryError)
	{
	char *filespaceLocation = NULL;

	*primaryError = 0;

	PersistentTablespace_GetFilespacePath(relFileNode->spcNode,
	TRUE,
	&filespaceLocation);

	HdfsFileInfo *file_info;


	char *path;

	path = (char*)palloc(MAXPGPATH + 1);

	FormRelfilePath(path,
	filespaceLocation,
	relFileNode,
	segmentFileNum);

	errno = 0;

	if (RemovePath(path, 0) < 0)
	{
	*primaryError = errno;
	}

	if (!IsLocalPath(path) && Gp_role == GP_ROLE_DISPATCH)
	{
	// Remove Hdfs block locations info in Metadata Cache
	file_info = CreateHdfsFileInfo(*relFileNode, segmentFileNum);
	LWLockAcquire(MetadataCacheLock, LW_EXCLUSIVE);
	RemoveHdfsFileBlockLocations(file_info);
	LWLockRelease(MetadataCacheLock);
	DestroyHdfsFileInfo(file_info);
	}

	pfree(path);

	if (filespaceLocation != NULL)
	pfree(filespaceLocation);

	}

	void MirroredAppendOnly_Drop(
	RelFileNode *relFileNode,

	int32 segmentFileNum,

	char *relationName,
	/* For tracing only. Can be NULL in some execution paths. */

	int *primaryError)
	{
	MirroredAppendOnly_DoDrop(
	relFileNode,
	segmentFileNum,
	relationName,
	primaryError);
	}

	// -----------------------------------------------------------------------------
	// Append
	// -----------------------------------------------------------------------------

	/*
	* Write bulk mirrored.
	*/
	void MirroredAppendOnly_Append(
	MirroredAppendOnlyOpen *open,
	/* The open struct. */

	void *buffer,
	/* Pointer to the buffer. */

	int32 bufferLen,
	/* Byte length of buffer. */

	int *primaryError)
	{
	Assert(open != NULL);
	Assert(open->isActive);

	*primaryError = 0;
	/*mirrorDataLossOccurred = false;/

	int ret;
	errno = 0;

	ret = FileWrite(open->primaryFile, buffer, bufferLen);

	if (ret != bufferLen)
	{
	/* if write didn't set errno, assume problem is no disk space */
	if (errno == 0)
	errno = ENOSPC;
	*primaryError = errno;
	}
	}

	// -----------------------------------------------------------------------------
	// Truncate
	// ----------------------------------------------------------------------------
	void MirroredAppendOnly_Truncate(
	MirroredAppendOnlyOpen *open,
	/* The open struct. */

	int64 position,
	/* The position to cutoff the data. */

	int *primaryError)
	{
	*primaryError = 0;
	errno = 0;

	if (FileTruncate(open->primaryFile, position) < 0)
	*primaryError = errno;
	}

	/*
	* Read an append only file in sequential way.
	* The routine is used for Resync.
	*/
	int MirroredAppendOnly_Read(
	MirroredAppendOnlyOpen *open,
	/* The open struct. */

	void *buffer,
	/* Pointer to the buffer. */

	int32 bufferLen)
	/* Byte length of buffer. */

	{
	int ret;

	Assert(open != NULL);
	Assert(buffer != NULL);
	Assert(open->isActive);

	errno = 0;

	ret = FileRead(open->primaryFile, buffer, bufferLen);

	return ret;
	}