colordiff --git a/doc/src/sgml/ref/copy.sgml b/doc/src/sgml/ref/copy.sgml new file mode 100644 index 6a0fabc..5c58dd2 *** a/doc/src/sgml/ref/copy.sgml --- b/doc/src/sgml/ref/copy.sgml *************** COPY { ta *** 38,43 **** --- 38,44 ---- DELIMITER 'delimiter_character' NULL 'null_string' HEADER [ boolean ] + COMPRESSED [ boolean ] QUOTE 'quote_character' ESCAPE 'escape_character' FORCE_QUOTE { ( column_name [, ...] ) | * } *************** COPY { ta *** 254,259 **** --- 255,271 ---- + + + COMPRESSED + + + Specifies that the file contents are compressed using zlib. On input, + the data should be compressed with zlib (eg: using gzip). On output, + the resulting data will be the compressed contents of the table. + + + QUOTE colordiff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c new file mode 100644 index abd82cf..1f394de *** a/src/backend/commands/copy.c --- b/src/backend/commands/copy.c *************** *** 19,24 **** --- 19,27 ---- #include #include #include + #ifdef HAVE_LIBZ + #include + #endif #include "access/heapam.h" #include "access/htup_details.h" *************** *** 59,66 **** typedef enum CopyDest { COPY_FILE, /* to/from file */ COPY_OLD_FE, /* to/from frontend (2.0 protocol) */ ! COPY_NEW_FE /* to/from frontend (3.0 protocol) */ } CopyDest; /* --- 62,71 ---- typedef enum CopyDest { COPY_FILE, /* to/from file */ + COPY_GZFILE, /* to/from compressed file */ COPY_OLD_FE, /* to/from frontend (2.0 protocol) */ ! COPY_NEW_FE, /* to/from frontend (3.0 protocol) */ ! COPY_NEW_FE_COMPRESSED /* FE 3.0 protocol, compressed */ } CopyDest; /* *************** typedef struct CopyStateData *** 95,100 **** --- 100,110 ---- /* low-level state data */ CopyDest copy_dest; /* type of copy source/destination */ FILE *copy_file; /* used if copy_dest == COPY_FILE */ + #ifdef HAVE_LIBZ + gzFile copy_gzfile; /* used if copy_dest == COPY_GZFILE */ + z_stream *zstrm; /* used if streaming compressed data */ + char *zstrm_buf; /* used if streaming compressed data */ + #endif StringInfo fe_msgbuf; /* used for all dests during COPY TO, only for * dest == COPY_NEW_FE in COPY FROM */ bool fe_eof; /* true if detected end of copy data */ *************** typedef struct CopyStateData *** 109,114 **** --- 119,125 ---- List *attnumlist; /* integer list of attnums to copy */ char *filename; /* filename, or NULL for STDIN/STDOUT */ bool binary; /* binary format? */ + bool compressed; /* compressed file? */ bool oids; /* include OIDs? */ bool freeze; /* freeze rows on loading? */ bool csv_mode; /* Comma Separated Value format? */ *************** static bool CopyGetInt32(CopyState cstat *** 320,325 **** --- 331,437 ---- static void CopySendInt16(CopyState cstate, int16 val); static bool CopyGetInt16(CopyState cstate, int16 *val); + #ifdef HAVE_LIBZ + /* Helper functions for working with zlib */ + static void zstrm_init(CopyState cstate, bool is_from); + void *zstrm_palloc(void *curr_memctx, unsigned int num, unsigned int size); + void zstrm_pfree(void *opaque, void *addr); + + /* + * Define some useful constants. + * We're just using the default windowBits, but we have to combine it + * with 16 to turn on gzip encoding, which we want to use. + */ + #define DEF_WINDOW_BITS 15 + #define GZIP_ENCODING 16 + #define DEF_MEMLEVEL 8 + + /* + * Our wrapper to be like standard palloc(), which we later pass to zlib + * Note that we can't just use palloc() since it's a #define to + * MemoryContextAlloc. + */ + void * + zstrm_palloc(void *curr_memctx, unsigned int num, unsigned int size) + { + return MemoryContextAlloc(curr_memctx, num*size); + } + + /* + * Do-nothing function to pass as free() + * No need to actually pfree() as the memory context will be released. + */ + void + zstrm_pfree(void *opaque, void *addr) + { + opaque = NULL; /* not used */ + addr = NULL; /* will be handled by the context being released */ + + return; + } + + /* Set up zstrm for streaming data */ + static void + zstrm_init(CopyState cstate, bool is_from) + { + /* Allocate storage for our z_stream and buffer */ + cstate->zstrm = palloc(sizeof(z_stream)); + cstate->zstrm_buf = palloc(RAW_BUF_SIZE); + + /* Have zlib use our malloc/free functions */ + cstate->zstrm->zalloc = &zstrm_palloc; + cstate->zstrm->zfree = &zstrm_pfree; + + /* Provide the current memory context, used by MemoryConextAlloc() */ + cstate->zstrm->opaque = CurrentMemoryContext; + + /* Initialize avail_in- there's never data ready right away... */ + cstate->zstrm->avail_in = 0; + + /* + * Initialize our pointers and structures based on if we are going + * doing decompression (from) or compression (to) + */ + if (is_from) + { + /* + * With decompression, we set up the input buffer as where we will + * load our data in to and then ask inflate() to process the + * compressed data and store the uncompressed results in the + * caller's buffer + */ + /* Data will eventually be coming from our buffer */ + cstate->zstrm->next_in = (unsigned char*) cstate->zstrm_buf; + + /* Output space not available yet */ + cstate->zstrm->next_out = Z_NULL; + cstate->zstrm->avail_out = 0; + if (inflateInit2(cstate->zstrm, + DEF_WINDOW_BITS | GZIP_ENCODING) != Z_OK) + elog(ERROR, "Error initializing zlib stream."); + } + else + { + /* + * With compression, we set up the output buffer as where we will + * receive data from the deflate() function, to send to the client; + * the input data comes from the caller sending us uncompressed data. + */ + /* No data available initially */ + cstate->zstrm->next_in = Z_NULL; + + /* Data will eventually be going to our buffer */ + cstate->zstrm->next_out = (unsigned char*) cstate->zstrm_buf; + cstate->zstrm->avail_out = RAW_BUF_SIZE; + if (deflateInit2(cstate->zstrm, Z_DEFAULT_COMPRESSION, + Z_DEFLATED, DEF_WINDOW_BITS | GZIP_ENCODING, + DEF_MEMLEVEL, Z_DEFAULT_STRATEGY) != Z_OK) + elog(ERROR, "Error initializing zlib stream."); + } + + return; + } + #endif /* * Send copy start/stop messages for frontend copies. These have changed *************** SendCopyBegin(CopyState cstate) *** 333,339 **** /* new way */ StringInfoData buf; int natts = list_length(cstate->attnumlist); ! int16 format = (cstate->binary ? 1 : 0); int i; pq_beginmessage(&buf, 'H'); --- 445,451 ---- /* new way */ StringInfoData buf; int natts = list_length(cstate->attnumlist); ! int16 format = (cstate->binary || cstate->compressed ? 1 : 0); int i; pq_beginmessage(&buf, 'H'); *************** SendCopyBegin(CopyState cstate) *** 342,348 **** for (i = 0; i < natts; i++) pq_sendint(&buf, format, 2); /* per-column formats */ pq_endmessage(&buf); ! cstate->copy_dest = COPY_NEW_FE; } else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2) { --- 454,473 ---- for (i = 0; i < natts; i++) pq_sendint(&buf, format, 2); /* per-column formats */ pq_endmessage(&buf); ! ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! cstate->copy_dest = COPY_NEW_FE_COMPRESSED; ! ! /* Initialize our zstream */ ! zstrm_init(cstate,false); ! #else ! elog(ERROR, "Not compiled with zlib support."); ! #endif ! } ! else ! cstate->copy_dest = COPY_NEW_FE; } else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2) { *************** ReceiveCopyBegin(CopyState cstate) *** 378,384 **** /* new way */ StringInfoData buf; int natts = list_length(cstate->attnumlist); ! int16 format = (cstate->binary ? 1 : 0); int i; pq_beginmessage(&buf, 'G'); --- 503,509 ---- /* new way */ StringInfoData buf; int natts = list_length(cstate->attnumlist); ! int16 format = (cstate->binary || cstate->compressed ? 1 : 0); int i; pq_beginmessage(&buf, 'G'); *************** ReceiveCopyBegin(CopyState cstate) *** 387,393 **** for (i = 0; i < natts; i++) pq_sendint(&buf, format, 2); /* per-column formats */ pq_endmessage(&buf); ! cstate->copy_dest = COPY_NEW_FE; cstate->fe_msgbuf = makeStringInfo(); } else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2) --- 512,532 ---- for (i = 0; i < natts; i++) pq_sendint(&buf, format, 2); /* per-column formats */ pq_endmessage(&buf); ! ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! cstate->copy_dest = COPY_NEW_FE_COMPRESSED; ! ! /* Initialize our zstream */ ! zstrm_init(cstate,true); ! #else ! elog(ERROR, "Not compiled with zlib support."); ! #endif ! } ! else ! cstate->copy_dest = COPY_NEW_FE; ! cstate->fe_msgbuf = makeStringInfo(); } else if (PG_PROTOCOL_MAJOR(FrontendProtocol) >= 2) *************** SendCopyEnd(CopyState cstate) *** 424,429 **** --- 563,593 ---- /* Send Copy Done message */ pq_putemptymessage('c'); } + else if (cstate->copy_dest == COPY_NEW_FE_COMPRESSED) + { + #ifdef HAVE_LIBZ + int result; + + /* Finish sending data */ + cstate->zstrm->next_in = (unsigned char*) cstate->fe_msgbuf->data; + cstate->zstrm->avail_in = cstate->fe_msgbuf->len; + + do { + result = deflate(cstate->zstrm, Z_FINISH); + if (result != Z_OK && result != Z_STREAM_END) + elog(ERROR, "Error finishing compressed stream."); + + (void) pq_putmessage('d', (char*) cstate->zstrm_buf, RAW_BUF_SIZE - cstate->zstrm->avail_out); + + /* ensure we empty the buffer */ + } while (result == Z_OK || cstate->zstrm->avail_in > 0); + + /* Send Copy Done message */ + pq_putemptymessage('c'); + #else + elog(ERROR, "Not compiled with zlib support."); + #endif + } else { CopySendData(cstate, "\\.", 2); *************** CopySendEndOfRow(CopyState cstate) *** 486,491 **** --- 650,676 ---- (errcode_for_file_access(), errmsg("could not write to COPY file: %m"))); break; + case COPY_GZFILE: + if (!cstate->binary) + { + /* Default line termination depends on platform */ + #ifndef WIN32 + CopySendChar(cstate, '\n'); + #else + CopySendString(cstate, "\r\n"); + #endif + } + + #ifdef HAVE_LIBZ + if (gzwrite(cstate->copy_gzfile, fe_msgbuf->data, + fe_msgbuf->len) != fe_msgbuf->len) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not write to COPY file: %m"))); + #else + elog(ERROR, "Not compiled with zlib support."); + #endif + break; case COPY_OLD_FE: /* The FE/BE protocol uses \n as newline for all platforms */ if (!cstate->binary) *************** CopySendEndOfRow(CopyState cstate) *** 507,512 **** --- 692,731 ---- /* Dump the accumulated row as one CopyData message */ (void) pq_putmessage('d', fe_msgbuf->data, fe_msgbuf->len); break; + case COPY_NEW_FE_COMPRESSED: + #ifdef HAVE_LIBZ + /* The FE/BE protocol uses \n as newline for all platforms */ + if (!cstate->binary) + CopySendChar(cstate, '\n'); + + /* We should always have some room for more data */ + Assert(cstate->zstrm->avail_out > 0); + + cstate->zstrm->next_in = (unsigned char*) fe_msgbuf->data; + cstate->zstrm->avail_in = fe_msgbuf->len; + + /* Dump the accumulated row as one CopyData message */ + do { + if (deflate(cstate->zstrm, Z_NO_FLUSH) != Z_OK) + elog(ERROR, "Error during compression"); + + /* Once avail_out is full, send it */ + if (cstate->zstrm->avail_out == 0) + { + (void) pq_putmessage('d', (char*) cstate->zstrm_buf, RAW_BUF_SIZE); + + /* Reset the buffer */ + cstate->zstrm->avail_out = RAW_BUF_SIZE; + cstate->zstrm->next_out = (unsigned char*) cstate->zstrm_buf; + } + } while (cstate->zstrm->avail_in > 0); /* ensure we empty buffer */ + #else + elog(ERROR, "Not compiled with zlib support."); + #endif + break; + default: + /* Should never reach here */ + Assert(false); } resetStringInfo(fe_msgbuf); *************** CopyGetData(CopyState cstate, void *data *** 539,544 **** --- 758,774 ---- (errcode_for_file_access(), errmsg("could not read from COPY file: %m"))); break; + case COPY_GZFILE: + #ifdef HAVE_LIBZ + bytesread = gzread(cstate->copy_gzfile, databuf, maxread); + if (bytesread == -1) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from COPY file: %m"))); + #else + elog(ERROR, "Not compiled with zlib support."); + #endif + break; case COPY_OLD_FE: /* *************** CopyGetData(CopyState cstate, void *data *** 557,562 **** --- 787,794 ---- } bytesread = minread; break; + case COPY_NEW_FE_COMPRESSED: + /* Fall through and handle as part of COPY_NEW_FE case */ case COPY_NEW_FE: while (maxread > 0 && bytesread < minread && !cstate->fe_eof) { *************** CopyGetData(CopyState cstate, void *data *** 609,621 **** break; } } ! avail = cstate->fe_msgbuf->len - cstate->fe_msgbuf->cursor; ! if (avail > maxread) ! avail = maxread; ! pq_copymsgbytes(cstate->fe_msgbuf, databuf, avail); ! databuf = (void *) ((char *) databuf + avail); ! maxread -= avail; ! bytesread += avail; } break; } --- 841,918 ---- break; } } ! if (cstate->copy_dest == COPY_NEW_FE_COMPRESSED) ! { ! #ifdef HAVE_LIBZ ! int result; ! ! /* Calculate how much data is *now* available */ ! cstate->zstrm->avail_in = cstate->zstrm->avail_in ! + cstate->fe_msgbuf->len ! - cstate->fe_msgbuf->cursor; ! ! /* ! * The amount of input data should never exceed our buffer ! * size or something has gone awry ! */ ! Assert(((char*) cstate->zstrm->next_in ! + cstate->zstrm->avail_in ! - cstate->zstrm_buf) < RAW_BUF_SIZE); ! ! /* Tack on the next set of input data to our in buffer */ ! pq_copymsgbytes(cstate->fe_msgbuf, ! (char*) cstate->zstrm->next_in, ! cstate->zstrm->avail_in); ! ! /* ! * Tell inflate() to store the uncompressed data into our ! * output buffer ! */ ! cstate->zstrm->next_out = (unsigned char*) databuf; ! cstate->zstrm->avail_out = maxread; ! ! /* decompress input data into our output buffer */ ! result = inflate(cstate->zstrm, Z_NO_FLUSH); ! if (result != Z_OK && result != Z_STREAM_END) ! elog(ERROR, "Error processing compressed input."); ! ! /* ! * Move any left-over data up to the front of our buffer, ! * to make room for the next batch; the amount of left- ! * over data is likely to be pretty small. ! */ ! if (cstate->zstrm->avail_in > 0) ! memmove(cstate->zstrm_buf, ! cstate->zstrm->next_in, ! cstate->zstrm->avail_in); ! ! /* Update our pointer to the next set of input data */ ! cstate->zstrm->next_in = (unsigned char*) cstate->zstrm_buf; ! ! /* Bytes returned from inflate() */ ! avail = maxread - cstate->zstrm->avail_out; ! ! /* ! * Update our databuf pointer, in case we need to provide ! * more data in this call. ! */ ! databuf = (void *) ((char *) databuf + avail); ! maxread -= avail; ! bytesread += avail; ! #else ! elog(ERROR,"Not compiled with zlib support."); ! #endif ! } ! else ! { ! avail = cstate->fe_msgbuf->len - cstate->fe_msgbuf->cursor; ! if (avail > maxread) ! avail = maxread; ! pq_copymsgbytes(cstate->fe_msgbuf, databuf, avail); ! databuf = (void *) ((char *) databuf + avail); ! maxread -= avail; ! bytesread += avail; ! } } break; } *************** ProcessCopyOptions(CopyState cstate, *** 889,894 **** --- 1186,1205 ---- (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("COPY format \"%s\" not recognized", fmt))); } + else if (strcmp(defel->defname, "compressed") == 0) + { + #ifdef HAVE_LIBZ + if (cstate->compressed) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("conflicting or redundant options"))); + cstate->compressed = defGetBoolean(defel); + #else + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("Not compiled with zlib support."))); + #endif + } else if (strcmp(defel->defname, "oids") == 0) { if (cstate->oids) *************** BeginCopy(bool is_from, *** 1379,1385 **** /* See Multibyte encoding comment above */ cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding); ! cstate->copy_dest = COPY_FILE; /* default */ MemoryContextSwitchTo(oldcontext); --- 1690,1699 ---- /* See Multibyte encoding comment above */ cstate->encoding_embeds_ascii = PG_ENCODING_IS_CLIENT_ONLY(cstate->file_encoding); ! if (cstate->compressed) ! cstate->copy_dest = COPY_GZFILE; ! else ! cstate->copy_dest = COPY_FILE; /* default */ MemoryContextSwitchTo(oldcontext); *************** BeginCopy(bool is_from, *** 1392,1402 **** static void EndCopy(CopyState cstate) { ! if (cstate->filename != NULL && FreeFile(cstate->copy_file)) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not close file \"%s\": %m", ! cstate->filename))); MemoryContextDelete(cstate->copycontext); pfree(cstate); --- 1706,1734 ---- static void EndCopy(CopyState cstate) { ! if (cstate->filename != NULL) ! { ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! if (FreeFileGz(cstate->copy_gzfile)) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not close file \"%s\": %m", ! cstate->filename))); ! #else ! elog(ERROR,"Not compiled with zlib support."); ! #endif ! } ! else ! { ! if(FreeFile(cstate->copy_file)) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not close file \"%s\": %m", ! cstate->filename))); ! } ! } MemoryContextDelete(cstate->copycontext); pfree(cstate); *************** BeginCopyTo(Relation rel, *** 1467,1486 **** cstate->filename = pstrdup(filename); oumask = umask(S_IWGRP | S_IWOTH); ! cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W); ! umask(oumask); ! ! if (cstate->copy_file == NULL) ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for writing: %m", ! cstate->filename))); ! fstat(fileno(cstate->copy_file), &st); ! if (S_ISDIR(st.st_mode)) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("\"%s\" is a directory", cstate->filename))); } MemoryContextSwitchTo(oldcontext); --- 1799,1846 ---- cstate->filename = pstrdup(filename); oumask = umask(S_IWGRP | S_IWOTH); ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! cstate->copy_gzfile = AllocateFileGz(cstate->filename, PG_BINARY_W); ! #else ereport(ERROR, ! (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("Not compiled with zlib support."))); ! #endif ! } ! else ! cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_W); ! umask(oumask); ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! if (cstate->copy_gzfile == NULL) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for writing (compressed): %m", ! cstate->filename))); ! #else ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("Not compiled with zlib support."))); ! #endif ! } ! else ! { ! if (cstate->copy_file == NULL) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for writing: %m", ! cstate->filename))); ! ! fstat(fileno(cstate->copy_file), &st); ! if (S_ISDIR(st.st_mode)) ! ereport(ERROR, ! (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("\"%s\" is a directory", cstate->filename))); ! } } MemoryContextSwitchTo(oldcontext); *************** BeginCopyFrom(Relation rel, *** 2419,2437 **** struct stat st; cstate->filename = pstrdup(filename); ! cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R); ! ! if (cstate->copy_file == NULL) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for reading: %m", ! cstate->filename))); ! fstat(fileno(cstate->copy_file), &st); ! if (S_ISDIR(st.st_mode)) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("\"%s\" is a directory", cstate->filename))); } if (!cstate->binary) --- 2779,2816 ---- struct stat st; cstate->filename = pstrdup(filename); ! if (cstate->compressed) ! { ! #ifdef HAVE_LIBZ ! cstate->copy_gzfile = AllocateFileGz(cstate->filename, PG_BINARY_R); ! if (cstate->copy_gzfile == NULL) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for reading: %m", ! cstate->filename))); ! #else ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("Not compiled with zlib support."))); ! #endif ! } ! else ! { ! cstate->copy_file = AllocateFile(cstate->filename, PG_BINARY_R); ! ! if (cstate->copy_file == NULL) ! ereport(ERROR, ! (errcode_for_file_access(), ! errmsg("could not open file \"%s\" for reading: %m", ! cstate->filename))); ! ! fstat(fileno(cstate->copy_file), &st); ! if (S_ISDIR(st.st_mode)) ! ereport(ERROR, ! (errcode(ERRCODE_WRONG_OBJECT_TYPE), ! errmsg("\"%s\" is a directory", cstate->filename))); ! } } if (!cstate->binary) *************** CopyReadLine(CopyState cstate) *** 2823,2829 **** * after \. up to the protocol end of copy data. (XXX maybe better * not to treat \. as special?) */ ! if (cstate->copy_dest == COPY_NEW_FE) { do { --- 3202,3209 ---- * after \. up to the protocol end of copy data. (XXX maybe better * not to treat \. as special?) */ ! if (cstate->copy_dest == COPY_NEW_FE || ! cstate->copy_dest == COPY_NEW_FE_COMPRESSED) { do { colordiff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y new file mode 100644 index 4568876..23ebaa7 *** a/src/backend/parser/gram.y --- b/src/backend/parser/gram.y *************** static void processCASbits(int cas_bits, *** 521,528 **** CACHE CALLED CASCADE CASCADED CASE CAST CATALOG_P CHAIN CHAR_P CHARACTER CHARACTERISTICS CHECK CHECKPOINT CLASS CLOSE CLUSTER COALESCE COLLATE COLLATION COLUMN COMMENT COMMENTS COMMIT ! COMMITTED CONCURRENTLY CONFIGURATION CONNECTION CONSTRAINT CONSTRAINTS ! CONTENT_P CONTINUE_P CONVERSION_P COPY COST CREATE CROSS CSV CURRENT_P CURRENT_CATALOG CURRENT_DATE CURRENT_ROLE CURRENT_SCHEMA CURRENT_TIME CURRENT_TIMESTAMP CURRENT_USER CURSOR CYCLE --- 521,528 ---- CACHE CALLED CASCADE CASCADED CASE CAST CATALOG_P CHAIN CHAR_P CHARACTER CHARACTERISTICS CHECK CHECKPOINT CLASS CLOSE CLUSTER COALESCE COLLATE COLLATION COLUMN COMMENT COMMENTS COMMIT ! COMMITTED COMPRESSED CONCURRENTLY CONFIGURATION CONNECTION CONSTRAINT ! CONSTRAINTS CONTENT_P CONTINUE_P CONVERSION_P COPY COST CREATE CROSS CSV CURRENT_P CURRENT_CATALOG CURRENT_DATE CURRENT_ROLE CURRENT_SCHEMA CURRENT_TIME CURRENT_TIMESTAMP CURRENT_USER CURSOR CYCLE *************** copy_opt_item: *** 2403,2408 **** --- 2403,2412 ---- { $$ = makeDefElem("header", (Node *)makeInteger(TRUE)); } + | COMPRESSED + { + $$ = makeDefElem("compressed", (Node *)makeInteger(TRUE)); + } | QUOTE opt_as Sconst { $$ = makeDefElem("quote", (Node *)makeString($3)); *************** unreserved_keyword: *** 12471,12476 **** --- 12475,12481 ---- | COMMENTS | COMMIT | COMMITTED + | COMPRESSED | CONFIGURATION | CONNECTION | CONSTRAINTS colordiff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c new file mode 100644 index 07ee51c..4547285 *** a/src/backend/storage/file/fd.c --- b/src/backend/storage/file/fd.c *************** *** 66,71 **** --- 66,74 ---- #ifdef HAVE_SYS_RESOURCE_H #include /* for getrlimit */ #endif + #ifdef HAVE_LIBZ + #include + #endif #include "miscadmin.h" #include "access/xact.h" *************** static uint64 temporary_files_size = 0; *** 201,206 **** --- 204,210 ---- typedef enum { AllocateDescFile, + AllocateDescGzFile, AllocateDescDir, AllocateDescRawFD } AllocateDescKind; *************** typedef struct *** 211,216 **** --- 215,223 ---- union { FILE *file; + #ifdef HAVE_LIBZ + gzFile gzfile; + #endif DIR *dir; int fd; } desc; *************** TryAgain: *** 1543,1548 **** --- 1550,1607 ---- return NULL; } + /* + * Like AllocateFile, but uses zlib to open a compressed file. + */ + #ifdef HAVE_LIBZ + gzFile + AllocateFileGz(const char *name, const char *mode) + { + gzFile file; + + DO_DB(elog(LOG, "AllocateFileGz: Allocated %d (%s)", + numAllocatedDescs, name)); + + /* + * The test against MAX_ALLOCATED_DESCS prevents us from overflowing + * allocatedFiles[]; the test against max_safe_fds prevents AllocateFile + * from hogging every one of the available FDs, which'd lead to infinite + * looping. + */ + if (numAllocatedDescs >= MAX_ALLOCATED_DESCS || + numAllocatedDescs >= max_safe_fds - 1) + elog(ERROR, "exceeded MAX_ALLOCATED_DESCS while trying to open file \"%s\"", + name); + + TryAgain: + if ((file = gzopen(name, mode)) != NULL) + { + AllocateDesc *desc = &allocatedDescs[numAllocatedDescs]; + + desc->kind = AllocateDescGzFile; + desc->desc.gzfile = file; + desc->create_subid = GetCurrentSubTransactionId(); + numAllocatedDescs++; + return desc->desc.gzfile; + } + + if (errno == EMFILE || errno == ENFILE) + { + int save_errno = errno; + + ereport(LOG, + (errcode(ERRCODE_INSUFFICIENT_RESOURCES), + errmsg("out of file descriptors: %m; release and retry"))); + errno = 0; + if (ReleaseLruFile()) + goto TryAgain; + errno = save_errno; + } + + return NULL; + } + #endif + /* * Like AllocateFile, but returns an unbuffered fd like open(2) *************** FreeDesc(AllocateDesc *desc) *** 1600,1605 **** --- 1659,1672 ---- case AllocateDescFile: result = fclose(desc->desc.file); break; + case AllocateDescGzFile: + #ifdef HAVE_LIBZ + result = gzclose(desc->desc.gzfile); + #else + elog(ERROR,"AllocateDesc not compiled with zlib support"); + result = 0; /* keep compiler quiet */ + #endif + break; case AllocateDescDir: result = closedir(desc->desc.dir); break; *************** FreeFile(FILE *file) *** 1648,1653 **** --- 1715,1750 ---- } /* + * Close a file returned by AllocateFileGz. + * + * Note we do not check gzclose's return value --- it is up to the caller + * to handle close errors. + */ + #ifdef HAVE_LIBZ + int + FreeFileGz(gzFile file) + { + int i; + + DO_DB(elog(LOG, "FreeFileGz: Allocated %d", numAllocatedDescs)); + + /* Remove file from list of allocated files, if it's present */ + for (i = numAllocatedDescs; --i >= 0;) + { + AllocateDesc *desc = &allocatedDescs[i]; + + if (desc->kind == AllocateDescGzFile && desc->desc.gzfile == file) + return FreeDesc(desc); + } + + /* Only get here if someone passes us a file not in allocatedDescs */ + elog(WARNING, "file passed to FreeFileGz was not obtained from AllocateFile"); + + return gzclose(file); + } + #endif + + /* * Close a file returned by OpenTransientFile. * * Note we do not check close's return value --- it is up to the caller colordiff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h new file mode 100644 index af60dac..2aecd74 *** a/src/include/parser/kwlist.h --- b/src/include/parser/kwlist.h *************** PG_KEYWORD("comment", COMMENT, UNRESERVE *** 85,90 **** --- 85,91 ---- PG_KEYWORD("comments", COMMENTS, UNRESERVED_KEYWORD) PG_KEYWORD("commit", COMMIT, UNRESERVED_KEYWORD) PG_KEYWORD("committed", COMMITTED, UNRESERVED_KEYWORD) + PG_KEYWORD("compressed", COMPRESSED, UNRESERVED_KEYWORD) PG_KEYWORD("concurrently", CONCURRENTLY, TYPE_FUNC_NAME_KEYWORD) PG_KEYWORD("configuration", CONFIGURATION, UNRESERVED_KEYWORD) PG_KEYWORD("connection", CONNECTION, UNRESERVED_KEYWORD) colordiff --git a/src/include/storage/fd.h b/src/include/storage/fd.h new file mode 100644 index 940d9d4..6cd1ae4 *** a/src/include/storage/fd.h --- b/src/include/storage/fd.h *************** *** 41,46 **** --- 41,49 ---- #include + #ifdef HAVE_LIBZ + #include + #endif /* * FileSeek uses the standard UNIX lseek(2) flags. *************** extern char *FilePathName(File file); *** 80,85 **** --- 83,94 ---- extern FILE *AllocateFile(const char *name, const char *mode); extern int FreeFile(FILE *file); + /* Operations for compressed files */ + #ifdef HAVE_LIBZ + extern gzFile AllocateFileGz(const char *name, const char *mode); + extern int FreeFileGz(gzFile file); + #endif + /* Operations to allow use of the library routines */ extern DIR *AllocateDir(const char *dirname); extern struct dirent *ReadDir(DIR *dir, const char *dirname); colordiff --git a/src/test/regress/input/copy.source b/src/test/regress/input/copy.source new file mode 100644 index ab3f508..5f83d98 *** a/src/test/regress/input/copy.source --- b/src/test/regress/input/copy.source *************** copy copytest2 from '@abs_builddir@/resu *** 91,96 **** --- 91,116 ---- select * from copytest except select * from copytest2; + truncate copytest2; + + --- same test but with compression + + copy copytest to '@abs_builddir@/results/copytest.csv.gz' csv compressed; + + copy copytest2 from '@abs_builddir@/results/copytest.csv.gz' csv compressed; + + select * from copytest except select * from copytest2; + + truncate copytest2; + + --- same test but with compression through psql + + \copy copytest to '@abs_builddir@/results/copytest.csv.gz' csv compressed; + + \copy copytest2 from '@abs_builddir@/results/copytest.csv.gz' csv compressed; + + select * from copytest except select * from copytest2; + -- test header line feature colordiff --git a/src/test/regress/output/copy.source b/src/test/regress/output/copy.source new file mode 100644 index febca71..a6730fe *** a/src/test/regress/output/copy.source --- b/src/test/regress/output/copy.source *************** select * from copytest except select * f *** 61,66 **** --- 61,84 ---- -------+------+-------- (0 rows) + truncate copytest2; + --- same test but with compression + copy copytest to '@abs_builddir@/results/copytest.csv.gz' csv compressed; + copy copytest2 from '@abs_builddir@/results/copytest.csv.gz' csv compressed; + select * from copytest except select * from copytest2; + style | test | filler + -------+------+-------- + (0 rows) + + truncate copytest2; + --- same test but with compression through psql + \copy copytest to '@abs_builddir@/results/copytest.csv.gz' csv compressed; + \copy copytest2 from '@abs_builddir@/results/copytest.csv.gz' csv compressed; + select * from copytest except select * from copytest2; + style | test | filler + -------+------+-------- + (0 rows) + -- test header line feature create temp table copytest3 ( c1 int,