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

Differentiate worker and master job temporary folders 

This commit enables to create different worker and master temporary folders.
This change is important for citus-mx on task-tracker execution. In simple words,
on citus-mx, the worker could actually be reponsible for the master tasks as well.
Prior to this change, both master and worker logic on task-tracker executor was
accessing and using the same files for different purposes which was dangerous on
certain cases (i.e., when task_tracker_delay is low).
pull/742/head
Onder Kalaci 2016-08-15 15:23:41 +03:00 committed by Eren Basak
parent a8701841b5
commit a533b8e7c1
8 changed files with 73 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -87,7 +87,7 @@ multi_ExecutorStart(QueryDesc *queryDesc, int eflags)
* We create a directory on the master node to keep task execution results.
* We also register this directory for automatic cleanup on portal delete.
*/
jobDirectoryName = JobDirectoryName(workerJob->jobId);
jobDirectoryName = MasterJobDirectoryName(workerJob->jobId);
CreateDirectory(jobDirectoryName);
ResourceOwnerEnlargeJobDirectories(CurrentResourceOwner);

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

@ -514,7 +514,7 @@ ManageTaskExecution(Task *task, TaskExecution *taskExecution,
queryStatus = MultiClientQueryStatus(connectionId);
if (queryStatus == CLIENT_QUERY_COPY)
{
StringInfo jobDirectoryName = JobDirectoryName(task->jobId);
StringInfo jobDirectoryName = MasterJobDirectoryName(task->jobId);
StringInfo taskFilename = TaskFilename(jobDirectoryName, task->taskId);
char *filename = taskFilename->data;

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

@ -204,7 +204,7 @@ MaxMasterConnectionCount(void)
void
RemoveJobDirectory(uint64 jobId)
{
StringInfo jobDirectoryName = JobDirectoryName(jobId);
StringInfo jobDirectoryName = MasterJobDirectoryName(jobId);
RemoveDirectory(jobDirectoryName);
ResourceOwnerForgetJobDirectory(CurrentResourceOwner, jobId);

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

@ -1314,7 +1314,7 @@ ManageTransmitExecution(TaskTracker *transmitTracker,
queryStatus = MultiClientQueryStatus(connectionId);
if (queryStatus == CLIENT_QUERY_COPY)
{
StringInfo jobDirectoryName = JobDirectoryName(task->jobId);
StringInfo jobDirectoryName = MasterJobDirectoryName(task->jobId);
StringInfo taskFilename = TaskFilename(jobDirectoryName, task->taskId);
char *filename = taskFilename->data;

2
src/backend/distributed/planner/multi_master_planner.c
View File

@ -347,7 +347,7 @@ MasterNodeCopyStatementList(MultiPlan *multiPlan)
foreach(workerTaskCell, workerTaskList)
{
Task *workerTask = (Task *) lfirst(workerTaskCell);
StringInfo jobDirectoryName = JobDirectoryName(workerTask->jobId);
StringInfo jobDirectoryName = MasterJobDirectoryName(workerTask->jobId);
StringInfo taskFilename = TaskFilename(jobDirectoryName, workerTask->taskId);
RangeVar *relation = makeRangeVar(NULL, tableName, -1);

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

@ -441,7 +441,31 @@ ClosePartitionFiles(FileOutputStream *partitionFileArray, uint32 fileCount)
}
/* Constructs a standardized job directory path for the given job id. */
/*
* MasterJobDirectoryName constructs a standardized job
* directory path for the given job id on the master node.
*/
StringInfo
MasterJobDirectoryName(uint64 jobId)
{
StringInfo jobDirectoryName = makeStringInfo();
/*
* We use the default tablespace in {datadir}/base. Further, we need to
* apply padding on our 64-bit job id, and hence can't use UINT64_FORMAT.
*/
appendStringInfo(jobDirectoryName, "base/%s/%s%0*" INT64_MODIFIER "u",
PG_JOB_CACHE_DIR, MASTER_JOB_DIRECTORY_PREFIX,
MIN_JOB_DIRNAME_WIDTH, jobId);
return jobDirectoryName;
}
/*
* JobDirectoryName Constructs a standardized job
* directory path for the given job id on the worker nodes.
*/
StringInfo
JobDirectoryName(uint64 jobId)
{

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

@ -26,6 +26,7 @@
/* Directory, file, table name, and UDF related defines for distributed tasks */
#define PG_JOB_CACHE_DIR "pgsql_job_cache"
#define MASTER_JOB_DIRECTORY_PREFIX "master_job_"
#define JOB_DIRECTORY_PREFIX "job_"
#define JOB_SCHEMA_PREFIX "pg_merge_job_"
#define TASK_FILE_PREFIX "task_"
@ -107,6 +108,7 @@ extern StringInfo JobSchemaName(uint64 jobId);
extern StringInfo TaskTableName(uint32 taskId);
extern bool JobSchemaExists(StringInfo schemaName);
extern StringInfo JobDirectoryName(uint64 jobId);
extern StringInfo MasterJobDirectoryName(uint64 jobId);
extern StringInfo TaskDirectoryName(uint64 jobId, uint32 taskId);
extern StringInfo PartitionFilename(StringInfo directoryName, uint32 partitionId);
extern bool CacheDirectoryElement(const char *filename);

78
src/test/regress/expected/multi_query_directory_cleanup.out
View File

@ -97,27 +97,27 @@ DECLARE c_18 CURSOR FOR SELECT sum(l_extendedprice * l_discount) as revenue FROM
DECLARE c_19 CURSOR FOR SELECT sum(l_extendedprice * l_discount) as revenue FROM lineitem;
SELECT * FROM pg_ls_dir('base/pgsql_job_cache') f ORDER BY f;
f
----------
job_1256
job_1257
job_1258
job_1259
job_1260
job_1261
job_1262
job_1263
job_1264
job_1265
job_1266
job_1267
job_1268
job_1269
job_1270
job_1271
job_1272
job_1273
job_1274
job_1275
-----------------
master_job_1256
master_job_1257
master_job_1258
master_job_1259
master_job_1260
master_job_1261
master_job_1262
master_job_1263
master_job_1264
master_job_1265
master_job_1266
master_job_1267
master_job_1268
master_job_1269
master_job_1270
master_job_1271
master_job_1272
master_job_1273
master_job_1274
master_job_1275
(20 rows)
-- close first, 17th (first after re-alloc) and last cursor.
@ -126,24 +126,24 @@ CLOSE c_16;
CLOSE c_19;
SELECT * FROM pg_ls_dir('base/pgsql_job_cache') f ORDER BY f;
f
----------
job_1257
job_1258
job_1259
job_1260
job_1261
job_1262
job_1263
job_1264
job_1265
job_1266
job_1267
job_1268
job_1269
job_1270
job_1271
job_1273
job_1274
-----------------
master_job_1257
master_job_1258
master_job_1259
master_job_1260
master_job_1261
master_job_1262
master_job_1263
master_job_1264
master_job_1265
master_job_1266
master_job_1267
master_job_1268
master_job_1269
master_job_1270
master_job_1271
master_job_1273
master_job_1274
(17 rows)
ROLLBACK;