mirror of https://github.com/citusdata/citus.git
Merge branch 'main' into grant_role_2pc
commit
4e5379cc8a
|
@ -2663,7 +2663,6 @@ CreateLocalColocatedIntermediateFile(CitusCopyDestReceiver *copyDest,
|
|||
CreateIntermediateResultsDirectory();
|
||||
|
||||
const int fileFlags = (O_CREAT | O_RDWR | O_TRUNC);
|
||||
const int fileMode = (S_IRUSR | S_IWUSR);
|
||||
|
||||
StringInfo filePath = makeStringInfo();
|
||||
appendStringInfo(filePath, "%s_%ld", copyDest->colocatedIntermediateResultIdPrefix,
|
||||
|
@ -2671,7 +2670,7 @@ CreateLocalColocatedIntermediateFile(CitusCopyDestReceiver *copyDest,
|
|||
|
||||
const char *fileName = QueryResultFileName(filePath->data);
|
||||
shardState->fileDest =
|
||||
FileCompatFromFileStart(FileOpenForTransmit(fileName, fileFlags, fileMode));
|
||||
FileCompatFromFileStart(FileOpenForTransmit(fileName, fileFlags));
|
||||
|
||||
CopyOutState localFileCopyOutState = shardState->copyOutState;
|
||||
bool isBinaryCopy = localFileCopyOutState->binary;
|
||||
|
|
|
@ -295,7 +295,6 @@ PrepareIntermediateResultBroadcast(RemoteFileDestReceiver *resultDest)
|
|||
if (resultDest->writeLocalFile)
|
||||
{
|
||||
const int fileFlags = (O_APPEND | O_CREAT | O_RDWR | O_TRUNC | PG_BINARY);
|
||||
const int fileMode = (S_IRUSR | S_IWUSR);
|
||||
|
||||
/* make sure the directory exists */
|
||||
CreateIntermediateResultsDirectory();
|
||||
|
@ -303,8 +302,7 @@ PrepareIntermediateResultBroadcast(RemoteFileDestReceiver *resultDest)
|
|||
const char *fileName = QueryResultFileName(resultId);
|
||||
|
||||
resultDest->fileCompat = FileCompatFromFileStart(FileOpenForTransmit(fileName,
|
||||
fileFlags,
|
||||
fileMode));
|
||||
fileFlags));
|
||||
}
|
||||
|
||||
WorkerNode *workerNode = NULL;
|
||||
|
@ -606,7 +604,7 @@ CreateIntermediateResultsDirectory(void)
|
|||
{
|
||||
char *resultDirectory = IntermediateResultsDirectory();
|
||||
|
||||
int makeOK = mkdir(resultDirectory, S_IRWXU);
|
||||
int makeOK = MakePGDirectory(resultDirectory);
|
||||
if (makeOK != 0)
|
||||
{
|
||||
if (errno == EEXIST)
|
||||
|
@ -976,7 +974,6 @@ FetchRemoteIntermediateResult(MultiConnection *connection, char *resultId)
|
|||
|
||||
StringInfo copyCommand = makeStringInfo();
|
||||
const int fileFlags = (O_APPEND | O_CREAT | O_RDWR | O_TRUNC | PG_BINARY);
|
||||
const int fileMode = (S_IRUSR | S_IWUSR);
|
||||
|
||||
PGconn *pgConn = connection->pgConn;
|
||||
int socket = PQsocket(pgConn);
|
||||
|
@ -998,7 +995,7 @@ FetchRemoteIntermediateResult(MultiConnection *connection, char *resultId)
|
|||
|
||||
PQclear(result);
|
||||
|
||||
File fileDesc = FileOpenForTransmit(localPath, fileFlags, fileMode);
|
||||
File fileDesc = FileOpenForTransmit(localPath, fileFlags);
|
||||
FileCompat fileCompat = FileCompatFromFileStart(fileDesc);
|
||||
|
||||
while (true)
|
||||
|
|
|
@ -17,6 +17,7 @@
|
|||
#include "pgstat.h"
|
||||
|
||||
#include "commands/defrem.h"
|
||||
#include "common/file_perm.h"
|
||||
#include "libpq/libpq.h"
|
||||
#include "libpq/pqformat.h"
|
||||
#include "storage/fd.h"
|
||||
|
@ -48,8 +49,7 @@ RedirectCopyDataToRegularFile(const char *filename)
|
|||
{
|
||||
StringInfo copyData = makeStringInfo();
|
||||
const int fileFlags = (O_APPEND | O_CREAT | O_RDWR | O_TRUNC | PG_BINARY);
|
||||
const int fileMode = (S_IRUSR | S_IWUSR);
|
||||
File fileDesc = FileOpenForTransmit(filename, fileFlags, fileMode);
|
||||
File fileDesc = FileOpenForTransmit(filename, fileFlags);
|
||||
FileCompat fileCompat = FileCompatFromFileStart(fileDesc);
|
||||
|
||||
SendCopyInStart();
|
||||
|
@ -92,7 +92,7 @@ SendRegularFile(const char *filename)
|
|||
const int fileMode = 0;
|
||||
|
||||
/* we currently do not check if the caller has permissions for this file */
|
||||
File fileDesc = FileOpenForTransmit(filename, fileFlags, fileMode);
|
||||
File fileDesc = FileOpenForTransmitPerm(filename, fileFlags, fileMode);
|
||||
FileCompat fileCompat = FileCompatFromFileStart(fileDesc);
|
||||
|
||||
/*
|
||||
|
@ -136,12 +136,23 @@ FreeStringInfo(StringInfo stringInfo)
|
|||
|
||||
|
||||
/*
|
||||
* FileOpenForTransmit opens file with the given filename and flags. On success,
|
||||
* the function returns the internal file handle for the opened file. On failure
|
||||
* the function errors out.
|
||||
* Open a file with FileOpenForTransmitPerm() and pass default file mode for
|
||||
* the fileMode parameter.
|
||||
*/
|
||||
File
|
||||
FileOpenForTransmit(const char *filename, int fileFlags, int fileMode)
|
||||
FileOpenForTransmit(const char *filename, int fileFlags)
|
||||
{
|
||||
return FileOpenForTransmitPerm(filename, fileFlags, pg_file_create_mode);
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
* FileOpenForTransmitPerm opens file with the given filename and flags. On
|
||||
* success, the function returns the internal file handle for the opened file.
|
||||
* On failure the function errors out.
|
||||
*/
|
||||
File
|
||||
FileOpenForTransmitPerm(const char *filename, int fileFlags, int fileMode)
|
||||
{
|
||||
struct stat fileStat;
|
||||
|
||||
|
|
|
@ -895,22 +895,13 @@ DecrementExternalClientBackendCounterAtExit(int code, Datum arg)
|
|||
static void
|
||||
CreateRequiredDirectories(void)
|
||||
{
|
||||
const char *subdirs[] = {
|
||||
"pg_foreign_file",
|
||||
"pg_foreign_file/cached",
|
||||
("base/" PG_JOB_CACHE_DIR)
|
||||
};
|
||||
const char *subdir = ("base/" PG_JOB_CACHE_DIR);
|
||||
|
||||
for (int dirNo = 0; dirNo < lengthof(subdirs); dirNo++)
|
||||
if (MakePGDirectory(subdir) != 0 && errno != EEXIST)
|
||||
{
|
||||
int ret = mkdir(subdirs[dirNo], S_IRWXU);
|
||||
|
||||
if (ret != 0 && errno != EEXIST)
|
||||
{
|
||||
ereport(ERROR, (errcode_for_file_access(),
|
||||
errmsg("could not create directory \"%s\": %m",
|
||||
subdirs[dirNo])));
|
||||
}
|
||||
ereport(ERROR, (errcode_for_file_access(),
|
||||
errmsg("could not create directory \"%s\": %m",
|
||||
subdir)));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -29,7 +29,7 @@ static bool FileIsLink(const char *filename, struct stat filestat);
|
|||
void
|
||||
CitusCreateDirectory(StringInfo directoryName)
|
||||
{
|
||||
int makeOK = mkdir(directoryName->data, S_IRWXU);
|
||||
int makeOK = MakePGDirectory(directoryName->data);
|
||||
if (makeOK != 0)
|
||||
{
|
||||
ereport(ERROR, (errcode_for_file_access(),
|
||||
|
|
|
@ -126,7 +126,6 @@ TaskFileDestReceiverStartup(DestReceiver *dest, int operation,
|
|||
const char *nullPrintCharacter = "\\N";
|
||||
|
||||
const int fileFlags = (O_APPEND | O_CREAT | O_RDWR | O_TRUNC | PG_BINARY);
|
||||
const int fileMode = (S_IRUSR | S_IWUSR);
|
||||
|
||||
/* use the memory context that was in place when the DestReceiver was created */
|
||||
MemoryContext oldContext = MemoryContextSwitchTo(taskFileDest->memoryContext);
|
||||
|
@ -148,8 +147,7 @@ TaskFileDestReceiverStartup(DestReceiver *dest, int operation,
|
|||
|
||||
taskFileDest->fileCompat = FileCompatFromFileStart(FileOpenForTransmit(
|
||||
taskFileDest->filePath,
|
||||
fileFlags,
|
||||
fileMode));
|
||||
fileFlags));
|
||||
|
||||
if (copyOutState->binary)
|
||||
{
|
||||
|
|
|
@ -21,7 +21,8 @@
|
|||
/* Function declarations for transmitting files between two nodes */
|
||||
extern void RedirectCopyDataToRegularFile(const char *filename);
|
||||
extern void SendRegularFile(const char *filename);
|
||||
extern File FileOpenForTransmit(const char *filename, int fileFlags, int fileMode);
|
||||
extern File FileOpenForTransmit(const char *filename, int fileFlags);
|
||||
extern File FileOpenForTransmitPerm(const char *filename, int fileFlags, int fileMode);
|
||||
|
||||
|
||||
#endif /* TRANSMIT_H */
|
||||
|
|
Loading…
Reference in New Issue