パラレルロードで使用する場合(MULTI_PROCESS=YES または WRITER=PARALLEL)、以下のことに注意しなければなりません：
+
+pg_bulkload は 2 つの PostgreSQL バックエンドを使います。
+リーダーバックエンドがメインの pg_bulkload() を実行し（入力の読み取り・パース・検証）、
+ライターバックエンドが libpq 経由で起動され、TYPE=TUPLE の pg_bulkload() でテーブルへのダイレクトライトを行います。
+
+PostgreSQL 9.6 以降では、リーダーバックエンドが PostgreSQL のロックグループのリーダーとなり、
+対象テーブルに AccessShareLock を保持したまま動作します。
+ライターバックエンドはテーブルにロックを取得する前にそのロックグループに参加し、
+ダイレクトライトのために AccessExclusiveLock を取得します。
+両バックエンドが同一ロックグループに属するため、これらのロックは互いにブロックせず、
+ライターが動作している間もリーダーは共有ロックを解放しません。
+ロードが進行中はライターの AccessExclusiveLock により、
+他セッションが対象テーブルに対する DDL を実行することもブロックされます。
+そのため、リーダーとライターが一貫したテーブル定義を参照するために、
+対象テーブルについて他セッションにスキーマ変更を避けるよう特別に注意する必要は通常ありません。
+
+PostgreSQL 9.6 より前のバージョン向けにビルドした場合は、
+リーダーバックエンドがライターバックエンドが AccessExclusiveLock を取得する前に
+AccessShareLock を解放します。その間に他セッションがテーブル定義を変更できるため、
+リーダーとライターで異なるテーブル定義を参照する可能性があります。
+ロード中は他セッションから対象テーブルに対する DDL などのスキーマ変更を行わないでください。
+
+PostgreSQL のバージョンにかかわらず、ロードが参照する他のオブジェクト +（例: FILTER 関数が参照する型やテーブル）の定義を、 +ロード中に変更しないでください。 +
MULTI_PROCESS=YESかつロード対象のデータベースにlocalhostから接続するのにパスワードが必要な場合、たとえパスワードを正しくプロンプトに入力しても、パスワード認証に失敗してしまいます。この問題を回避するには、以下のいずれかを設定してください。
kill -9+
+When using MULTI_PROCESS=YES or WRITER=PARALLEL, pg_bulkload uses two PostgreSQL
+backends: a reader backend that runs the main pg_bulkload()
+call (read and parse input, validate rows) and a writer backend started
+via libpq that runs pg_bulkload() with TYPE=TUPLE and performs the
+direct write to the table.
+
+On PostgreSQL 9.6 and later, the reader backend becomes a PostgreSQL lock group
+leader and keeps AccessShareLock on the target table. The writer
+backend joins that lock group before it acquires any lock on the table, then
+takes AccessExclusiveLock for the direct write. Because both
+backends belong to the same lock group, these locks do not block each other and
+the reader does not release its share lock while the writer is running.
+While the load is active, the writer's AccessExclusiveLock also
+blocks other sessions from running DDL on that table, so you do not need an
+extra precaution against concurrent schema changes on the target table for
+the reader and writer to stay consistent.
+
+On PostgreSQL releases before 9.6 (when pg_bulkload is built against them), the
+reader backend releases AccessShareLock before the writer backend
+acquires AccessExclusiveLock. Another session can modify the table
+definition in that interval, and the reader and writer backends may then see
+different table definitions. Do not run DDL or other schema changes on the
+target table from other sessions for the duration of the load.
+
+On any PostgreSQL version, avoid changing the definitions of other objects the +load depends on (for example types or tables referenced by a +FILTER function) while the load is running. +
+When MULTI_PROCESS=YES and password is required to connect from localhost to the database to load, the authentication will fail even if you enter diff --git a/include/reader.h b/include/reader.h index ed8839c..3b4701a 100644 --- a/include/reader.h +++ b/include/reader.h @@ -75,6 +75,9 @@ extern Parser *CreateBinaryParser(void); extern Parser *CreateCSVParser(void); extern Parser *CreateTupleParser(void); extern Parser *CreateFunctionParser(void); +#if PG_VERSION_NUM >= 90600 +extern void SetTupleParserQueue(Parser *parser, void *queue); +#endif #define ParserInit(self, checker, infile, relid, multi_process, collation) ((self)->init((self), (checker), (infile), (relid), (multi_process), (collation))) #define ParserRead(self, checker) ((self)->read((self), (checker))) diff --git a/lib/Makefile b/lib/Makefile index f08bba4..2fcd220 100644 --- a/lib/Makefile +++ b/lib/Makefile @@ -54,7 +54,7 @@ endif endif ifdef USE_PGXS -PG_CONFIG = pg_config +PG_CONFIG ?= pg_config PGXS := $(shell $(PG_CONFIG) --pgxs) include $(PGXS) else diff --git a/lib/parser_tuple.c b/lib/parser_tuple.c index 5e3a18e..6e21e7c 100644 --- a/lib/parser_tuple.c +++ b/lib/parser_tuple.c @@ -54,9 +54,6 @@ CreateTupleParser(void) static void TupleParserInit(TupleParser *self, Checker *checker, const char *infile, TupleDesc desc, bool multi_process, Oid collation) { - unsigned key; - char junk[2]; - if (checker->check_constraints) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("does not support parameter \"CHECK_CONSTRAINTS\" in \"TYPE = TUPLE\""))); @@ -67,10 +64,19 @@ TupleParserInit(TupleParser *self, Checker *checker, const char *infile, TupleDe checker->tchecker = NULL; - if (sscanf(infile, ":%u%1s", &key, junk) != 1) - elog(ERROR, "invalid shmem key format: %s", infile); +#if PG_VERSION_NUM >= 90600 + Assert(self->queue); +#else + { + unsigned key; + char junk[2]; + + if (sscanf(infile, ":%u%1s", &key, junk) != 1) + elog(ERROR, "invalid shmem key format: %s", infile); - self->queue = QueueOpen(key); + self->queue = QueueOpen(key); + } +#endif self->buflen = BLCKSZ; self->buffer = palloc(self->buflen); } @@ -131,3 +137,11 @@ TupleParserDumpRecord(TupleParser *self, FILE *fp, char *filename) { /* parse error does not happen in TupleParser. */ } + +#if PG_VERSION_NUM >= 90600 +void +SetTupleParserQueue(Parser *parser, void *queue) +{ + ((TupleParser *) parser)->queue = queue; +} +#endif diff --git a/lib/pg_bulkload.c b/lib/pg_bulkload.c index 3f928ae..2f4f124 100644 --- a/lib/pg_bulkload.c +++ b/lib/pg_bulkload.c @@ -15,6 +15,7 @@ #include "access/heapam.h" #include "access/reloptions.h" +#include "c.h" #include "catalog/objectaddress.h" #if PG_VERSION_NUM >= 120000 #include "catalog/pg_am.h" @@ -41,6 +42,10 @@ #include "pg_profile.h" #include "pg_strutil.h" #include "pgut/pgut-be.h" +#include "pgut/pgut-ipc.h" +#if PG_VERSION_NUM >= 90600 +#include "storage/proc.h" +#endif PG_MODULE_MAGIC; @@ -485,6 +490,11 @@ ParseOptions(Datum options, Reader **rd, Writer **wt, time_t tm) char *type = NULL; char *writer = NULL; bool multi_process = false; +#if PG_VERSION_NUM >= 90600 + char *infile = NULL; + bool parallel_writer = false; + Queue *queue = NULL; +#endif Assert(*rd == NULL); Assert(*wt == NULL); @@ -505,6 +515,10 @@ ParseOptions(Datum options, Reader **rd, Writer **wt, time_t tm) { ASSERT_ONCE(type == NULL); type = value; +#if PG_VERSION_NUM >= 90600 + if (!pg_strcasecmp(value, "TUPLE")) + parallel_writer = true; +#endif } else if (CompareKeyword(keyword, "WRITER") || CompareKeyword(keyword, "LOADER")) @@ -516,6 +530,13 @@ ParseOptions(Datum options, Reader **rd, Writer **wt, time_t tm) { multi_process = ParseBoolean(value); } +#if PG_VERSION_NUM >= 90600 + else if (CompareKeyword(keyword, "INPUT")) + { + ASSERT_ONCE(infile == NULL); + infile = pstrdup(value); + } +#endif else { rest_defs = lappend(rest_defs, opt); @@ -523,9 +544,52 @@ ParseOptions(Datum options, Reader **rd, Writer **wt, time_t tm) } } +#if PG_VERSION_NUM >= 90600 + /* + * When running as the writer backend of a MULTI_PROCESS load + * (TYPE=TUPLE, INPUT=:), join the reader's lock group before + * DirectWriterParam() acquires a lock via the RangeVarGetRelidExtended() call. + * This is necessary for the lock group mechanism to work. The process must + * join the lock group before the first table lock acquisition to prevent + * conflicts with locks acquired on the table by other group members. + */ + if (parallel_writer) + { + unsigned key; + char junk[2]; + void *leader_pgproc; + int leader_pid; + Assert(infile); + + if (sscanf(infile, ":%u%1s", &key, junk) != 1) + elog(ERROR, "invalid shmem key format: %s", infile); + + queue = QueueOpen(key); + QueueGetLockGroupInfo(queue, &leader_pgproc, &leader_pid); + if (leader_pgproc != NULL && leader_pid != 0) + { + if (!BecomeLockGroupMember((PGPROC *) leader_pgproc, leader_pid)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not join lock group of parallel writer leader (pid %d)", + leader_pid))); + } + } +#endif + *wt = WriterCreate(writer, multi_process); *rd = ReaderCreate(type); +#if PG_VERSION_NUM >= 90600 + if (parallel_writer) + { + Assert(queue); + SetTupleParserQueue((*rd)->parser, queue); + } + + (*rd)->infile = infile; +#endif + foreach (cell, rest_defs) { opt = lfirst(cell); diff --git a/lib/pgut/pgut-ipc.c b/lib/pgut/pgut-ipc.c index 3a7730b..1039375 100644 --- a/lib/pgut/pgut-ipc.c +++ b/lib/pgut/pgut-ipc.c @@ -99,6 +99,10 @@ typedef struct QueueHeader uint32 begin; /* position that begins to read on data */ uint32 end; /* position that begins to write on data */ slock_t mutex; /* locks shared variables begin, end and data */ +#if PG_VERSION_NUM >= 90600 + void *lock_group_leader; /* PGPROC* of the lock group leader */ + int lock_group_leader_pid; /* PID of the lock group leader */ +#endif char data[1]; /* VARIABLE LENGTH ARRAY - MUST BE LAST */ } QueueHeader; @@ -197,6 +201,10 @@ QueueCreate(unsigned *key, uint32 size) header->size = size; header->begin = header->end = 0; SpinLockInit(&header->mutex); +#if PG_VERSION_NUM >= 90600 + header->lock_group_leader = NULL; + header->lock_group_leader_pid = 0; +#endif self = palloc(sizeof(Queue)); self->handle = handle; @@ -370,6 +378,24 @@ QueueRead(Queue *self, void *buffer, uint32 len, bool need_lock) CHECK_FOR_INTERRUPTS(); pg_usleep(SPIN_SLEEP_MSEC * 1000); +#ifndef WIN32 + /* + * Detect if the writer detached without sending the + * terminator (e.g. due to an error on the writer side). + * When shm_nattch drops to 1, only we are still attached; + * the writer is gone and no more data will ever arrive. + */ + { + struct shmid_ds ds; + + if (shmctl(self->handle, IPC_STAT, &ds) < 0 || + ds.shm_nattch <= 1) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("parallel writer has disconnected"))); + } +#endif + goto retry; } @@ -509,3 +535,49 @@ QueueWrite(Queue *self, const struct iovec iov[], int count, uint32 timeout_msec sleep_msec += SPIN_SLEEP_MSEC; goto retry; } + +#if PG_VERSION_NUM >= 90600 +/** + * @brief Store lock group leader information in the queue header. + * + * The reader (lock group leader) calls this after QueueCreate() so + * that the writer, upon opening the same queue, can retrieve the + * information and join the lock group with BecomeLockGroupMember(). + * + * @param self [in] Queue handle returned by QueueCreate(). + * @param leader_pgproc [in] PGPROC pointer of the lock group leader + * (typically MyProc of the calling backend). + * @param leader_pid [in] PID of the lock group leader + * (typically MyProcPid of the calling backend). + */ +void +QueueSetLockGroupInfo(Queue *self, void *leader_pgproc, int leader_pid) +{ + volatile QueueHeader *header = self->header; + + header->lock_group_leader = leader_pgproc; + header->lock_group_leader_pid = leader_pid; +} + +/** + * @brief Retrieve lock group leader information from the queue header. + * + * The writer calls this after QueueOpen() to obtain the PGPROC pointer + * and PID of the lock group leader so it can join the group with + * BecomeLockGroupMember() before acquiring any heavyweight locks. + * + * @param self [in] Queue handle returned by QueueOpen(). + * @param leader_pgproc [out] Receives the PGPROC pointer of the leader, + * or NULL if no lock group info was stored. + * @param leader_pid [out] Receives the PID of the leader, + * or 0 if no lock group info was stored. + */ +void +QueueGetLockGroupInfo(Queue *self, void **leader_pgproc, int *leader_pid) +{ + volatile QueueHeader *header = self->header; + + *leader_pgproc = header->lock_group_leader; + *leader_pid = header->lock_group_leader_pid; +} +#endif diff --git a/lib/pgut/pgut-ipc.h b/lib/pgut/pgut-ipc.h index 02a1950..5a9facf 100644 --- a/lib/pgut/pgut-ipc.h +++ b/lib/pgut/pgut-ipc.h @@ -27,5 +27,9 @@ extern Queue *QueueOpen(unsigned key); extern void QueueClose(Queue *self); extern uint32 QueueRead(Queue *self, void *buffer, uint32 len, bool need_lock); extern bool QueueWrite(Queue *self, const struct iovec iov[], int count, uint32 timeout_msec, bool need_lock); +#if PG_VERSION_NUM >= 90600 +extern void QueueSetLockGroupInfo(Queue *self, void *leader_pgproc, int leader_pid); +extern void QueueGetLockGroupInfo(Queue *self, void **leader_pgproc, int *leader_pid); +#endif #endif /* PGUT_IPC_H */ diff --git a/lib/writer_parallel.c b/lib/writer_parallel.c index d7ecb5f..c84f510 100644 --- a/lib/writer_parallel.c +++ b/lib/writer_parallel.c @@ -20,6 +20,7 @@ #include "miscadmin.h" #include "postmaster/postmaster.h" #include "storage/lmgr.h" +#include "storage/proc.h" #include "utils/guc.h" #include "utils/memutils.h" #include "utils/rel.h" @@ -90,6 +91,16 @@ ParallelWriterInit(ParallelWriter *self) Assert(self->base.truncate == false); +#if PG_VERSION_NUM >= 90600 + /* + * Become lock group leader before acquiring any relation locks. + * The writer process will join this lock group, which allows its + * AccessExclusiveLock to coexist with our AccessShareLock (and the + * RowExclusiveLock held by COPY FROM, if applicable) without conflict. + */ + BecomeLockGroupLeader(); +#endif + /* Initialize information needed to check tuples when reading. */ if (self->base.relid != InvalidOid) { @@ -141,6 +152,17 @@ ParallelWriterInit(ParallelWriter *self) self->queue = QueueCreate(&queryKey, DEFAULT_BUFFER_SIZE); snprintf(queueName, lengthof(queueName), ":%u", queryKey); +#if PG_VERSION_NUM >= 90600 + /* + * Store our lock group identity in the queue header so the writer can + * join the same group before acquiring AccessExclusiveLock. Because + * both processes will belong to the same lock group, the writer's + * AccessExclusiveLock will not conflict with our AccessShareLock (or + * RowExclusiveLock held by COPY FROM). This eliminates the window + * that previously existed between releasing and re-acquiring the lock. + */ + QueueSetLockGroupInfo(self->queue, MyProc, MyProcPid); +#else /* * Connect to a new backend process that will actually perform the writes. * As the new process will take an AccessExclusiveLock on the target @@ -158,6 +180,8 @@ ParallelWriterInit(ParallelWriter *self) */ if (rel) UnlockRelation(rel, AccessShareLock); +#endif + self->conn = connect_to_localhost(); /* start transaction */ diff --git a/util/Makefile b/util/Makefile index 9d64ce4..555f240 100644 --- a/util/Makefile +++ b/util/Makefile @@ -20,7 +20,7 @@ endif endif ifdef USE_PGXS -PG_CONFIG = pg_config +PG_CONFIG ?= pg_config PGXS := $(shell $(PG_CONFIG) --pgxs) include $(PGXS) else