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

Merge pull request #3423 from citusdata/remove-directory-even-if-new-files-added 

CitusRemoveDirectory: loop when directory is not empty
pull/3447/head
Philip Dubé 2020-01-30 20:21:47 +00:00 committed by GitHub
parent 6b43fab325 84a500ffc6
commit d7204c9696
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 81 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/backend/distributed/executor/intermediate_results.c
View File

@ -667,10 +667,7 @@ RemoveIntermediateResultsDirectory(void)
{ {
if (CreatedResultsDirectory) if (CreatedResultsDirectory)
{ {
StringInfo resultsDirectory = makeStringInfo(); CitusRemoveDirectory(IntermediateResultsDirectory());
appendStringInfoString(resultsDirectory, IntermediateResultsDirectory());
CitusRemoveDirectory(resultsDirectory);
CreatedResultsDirectory = false; CreatedResultsDirectory = false;
} }

2
src/backend/distributed/executor/multi_server_executor.c
View File

@ -187,7 +187,7 @@ void
RemoveJobDirectory(uint64 jobId) RemoveJobDirectory(uint64 jobId)
{ {
StringInfo jobDirectoryName = MasterJobDirectoryName(jobId); StringInfo jobDirectoryName = MasterJobDirectoryName(jobId);
CitusRemoveDirectory(jobDirectoryName); CitusRemoveDirectory(jobDirectoryName->data);
ResourceOwnerForgetJobDirectory(CurrentResourceOwner, jobId); ResourceOwnerForgetJobDirectory(CurrentResourceOwner, jobId);
} }

2
src/backend/distributed/test/file_utils.c
View File

@ -23,7 +23,7 @@ citus_rm_job_directory(PG_FUNCTION_ARGS)
appendStringInfo(jobCacheDirectory, "base/%s/%s%0*" INT64_MODIFIER "u", appendStringInfo(jobCacheDirectory, "base/%s/%s%0*" INT64_MODIFIER "u",
PG_JOB_CACHE_DIR, JOB_DIRECTORY_PREFIX, PG_JOB_CACHE_DIR, JOB_DIRECTORY_PREFIX,
MIN_JOB_DIRNAME_WIDTH, jobId); MIN_JOB_DIRNAME_WIDTH, jobId);
CitusRemoveDirectory(jobCacheDirectory); CitusRemoveDirectory(jobCacheDirectory->data);
FreeStringInfo(jobCacheDirectory); FreeStringInfo(jobCacheDirectory);
PG_RETURN_VOID(); PG_RETURN_VOID();

2
src/backend/distributed/worker/task_tracker.c
View File

@ -344,7 +344,7 @@ TrackerCleanupJobDirectories(void)
StringInfo jobCacheDirectory = makeStringInfo(); StringInfo jobCacheDirectory = makeStringInfo();
appendStringInfo(jobCacheDirectory, "base/%s", PG_JOB_CACHE_DIR); appendStringInfo(jobCacheDirectory, "base/%s", PG_JOB_CACHE_DIR);
CitusRemoveDirectory(jobCacheDirectory); CitusRemoveDirectory(jobCacheDirectory->data);
CitusCreateDirectory(jobCacheDirectory); CitusCreateDirectory(jobCacheDirectory);
FreeStringInfo(jobCacheDirectory); FreeStringInfo(jobCacheDirectory);

2
src/backend/distributed/worker/task_tracker_protocol.c
View File

@ -231,7 +231,7 @@ task_tracker_cleanup_job(PG_FUNCTION_ARGS)
* writing to a table within the schema. * writing to a table within the schema.
*/ */
StringInfo jobDirectoryName = JobDirectoryName(jobId); StringInfo jobDirectoryName = JobDirectoryName(jobId);
CitusRemoveDirectory(jobDirectoryName); CitusRemoveDirectory(jobDirectoryName->data);
RemoveJobSchema(jobSchemaName); RemoveJobSchema(jobSchemaName);
UnlockJobResource(jobId, AccessExclusiveLock); UnlockJobResource(jobId, AccessExclusiveLock);

2
src/backend/distributed/worker/worker_merge_protocol.c
View File

@ -97,7 +97,7 @@ worker_repartition_cleanup(PG_FUNCTION_ARGS)
Oid schemaId = get_namespace_oid(jobSchemaName->data, false); Oid schemaId = get_namespace_oid(jobSchemaName->data, false);
EnsureSchemaOwner(schemaId); EnsureSchemaOwner(schemaId);
CitusRemoveDirectory(jobDirectoryName); CitusRemoveDirectory(jobDirectoryName->data);
RemoveJobSchema(jobSchemaName); RemoveJobSchema(jobSchemaName);
PG_RETURN_VOID(); PG_RETURN_VOID();
} }

46
src/backend/distributed/worker/worker_partition_protocol.c
View File

@ -85,7 +85,7 @@ static uint32 RangePartitionId(Datum partitionValue, Oid partitionCollation,
static uint32 HashPartitionId(Datum partitionValue, Oid partitionCollation, static uint32 HashPartitionId(Datum partitionValue, Oid partitionCollation,
const void *context); const void *context);
static StringInfo UserPartitionFilename(StringInfo directoryName, uint32 partitionId); static StringInfo UserPartitionFilename(StringInfo directoryName, uint32 partitionId);
static bool FileIsLink(char *filename, struct stat filestat); static bool FileIsLink(const char *filename, struct stat filestat);
/* exports for SQL callable functions */ /* exports for SQL callable functions */
@ -158,7 +158,7 @@ worker_range_partition_table(PG_FUNCTION_ARGS)
/* close partition files and atomically rename (commit) them */ /* close partition files and atomically rename (commit) them */
ClosePartitionFiles(partitionFileArray, fileCount); ClosePartitionFiles(partitionFileArray, fileCount);
CitusRemoveDirectory(taskDirectory); CitusRemoveDirectory(taskDirectory->data);
RenameDirectory(taskAttemptDirectory, taskDirectory); RenameDirectory(taskAttemptDirectory, taskDirectory);
PG_RETURN_VOID(); PG_RETURN_VOID();
@ -232,7 +232,7 @@ worker_hash_partition_table(PG_FUNCTION_ARGS)
/* close partition files and atomically rename (commit) them */ /* close partition files and atomically rename (commit) them */
ClosePartitionFiles(partitionFileArray, fileCount); ClosePartitionFiles(partitionFileArray, fileCount);
CitusRemoveDirectory(taskDirectory); CitusRemoveDirectory(taskDirectory->data);
RenameDirectory(taskAttemptDirectory, taskDirectory); RenameDirectory(taskAttemptDirectory, taskDirectory);
PG_RETURN_VOID(); PG_RETURN_VOID();
@ -698,7 +698,7 @@ FileIsLink(char *filename, struct stat filestat)
#else #else
static bool static bool
FileIsLink(char *filename, struct stat filestat) FileIsLink(const char *filename, struct stat filestat)
{ {
return S_ISLNK(filestat.st_mode); return S_ISLNK(filestat.st_mode);
} }
@ -714,12 +714,15 @@ FileIsLink(char *filename, struct stat filestat)
* system library's remove_all() method. * system library's remove_all() method.
*/ */
void void
CitusRemoveDirectory(StringInfo filename) CitusRemoveDirectory(const char *filename)
{ {
/* files may be added during execution, loop when that occurs */
while (true)
{
struct stat fileStat; struct stat fileStat;
int removed = 0; int removed = 0;
int fileStated = stat(filename->data, &fileStat); int fileStated = stat(filename, &fileStat);
if (fileStated < 0) if (fileStated < 0)
{ {
if (errno == ENOENT) if (errno == ENOENT)
@ -729,7 +732,7 @@ CitusRemoveDirectory(StringInfo filename)
else else
{ {
ereport(ERROR, (errcode_for_file_access(), ereport(ERROR, (errcode_for_file_access(),
errmsg("could not stat file \"%s\": %m", filename->data))); errmsg("could not stat file \"%s\": %m", filename)));
} }
} }
@ -738,9 +741,9 @@ CitusRemoveDirectory(StringInfo filename)
* content, recurse into this function. Also, make sure that we do not * content, recurse into this function. Also, make sure that we do not
* recurse into symbolic links. * recurse into symbolic links.
*/ */
if (S_ISDIR(fileStat.st_mode) && !FileIsLink(filename->data, fileStat)) if (S_ISDIR(fileStat.st_mode) && !FileIsLink(filename, fileStat))
{ {
const char *directoryName = filename->data; const char *directoryName = filename;
DIR *directory = AllocateDir(directoryName); DIR *directory = AllocateDir(directoryName);
if (directory == NULL) if (directory == NULL)
@ -750,8 +753,10 @@ CitusRemoveDirectory(StringInfo filename)
directoryName))); directoryName)));
} }
StringInfo fullFilename = makeStringInfo();
struct dirent *directoryEntry = ReadDir(directory, directoryName); struct dirent *directoryEntry = ReadDir(directory, directoryName);
for (; directoryEntry != NULL; directoryEntry = ReadDir(directory, directoryName)) for (; directoryEntry != NULL; directoryEntry = ReadDir(directory,
directoryName))
{ {
const char *baseFilename = directoryEntry->d_name; const char *baseFilename = directoryEntry->d_name;
@ -762,31 +767,38 @@ CitusRemoveDirectory(StringInfo filename)
continue; continue;
} }
StringInfo fullFilename = makeStringInfo(); resetStringInfo(fullFilename);
appendStringInfo(fullFilename, "%s/%s", directoryName, baseFilename); appendStringInfo(fullFilename, "%s/%s", directoryName, baseFilename);
CitusRemoveDirectory(fullFilename); CitusRemoveDirectory(fullFilename->data);
FreeStringInfo(fullFilename);
} }
FreeStringInfo(fullFilename);
FreeDir(directory); FreeDir(directory);
} }
/* we now have an empty directory or a regular file, remove it */ /* we now have an empty directory or a regular file, remove it */
if (S_ISDIR(fileStat.st_mode)) if (S_ISDIR(fileStat.st_mode))
{ {
removed = rmdir(filename->data); removed = rmdir(filename);
if (errno == ENOTEMPTY || errno == EEXIST)
{
continue;
}
} }
else else
{ {
removed = unlink(filename->data); removed = unlink(filename);
} }
if (removed != 0 && errno != ENOENT) if (removed != 0 && errno != ENOENT)
{ {
ereport(ERROR, (errcode_for_file_access(), ereport(ERROR, (errcode_for_file_access(),
errmsg("could not remove file \"%s\": %m", filename->data))); errmsg("could not remove file \"%s\": %m", filename)));
}
return;
} }
} }

2
src/include/distributed/worker_protocol.h
View File

@ -117,7 +117,7 @@ extern bool CacheDirectoryElement(const char *filename);
extern bool JobDirectoryElement(const char *filename); extern bool JobDirectoryElement(const char *filename);
extern bool DirectoryExists(StringInfo directoryName); extern bool DirectoryExists(StringInfo directoryName);
extern void CitusCreateDirectory(StringInfo directoryName); extern void CitusCreateDirectory(StringInfo directoryName);
extern void CitusRemoveDirectory(StringInfo filename); extern void CitusRemoveDirectory(const char *filename);
extern StringInfo InitTaskDirectory(uint64 jobId, uint32 taskId); extern StringInfo InitTaskDirectory(uint64 jobId, uint32 taskId);
extern void RemoveJobSchema(StringInfo schemaName); extern void RemoveJobSchema(StringInfo schemaName);
extern Datum * DeconstructArrayObject(ArrayType *arrayObject); extern Datum * DeconstructArrayObject(ArrayType *arrayObject);