copy_compressed.patch
text/x-diff
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 { <replaceable class="parameter">ta
*** 38,43 ****
--- 38,44 ----
DELIMITER '<replaceable class="parameter">delimiter_character</replaceable>'
NULL '<replaceable class="parameter">null_string</replaceable>'
HEADER [ <replaceable class="parameter">boolean</replaceable> ]
+ COMPRESSED [ <replaceable class="parameter">boolean</replaceable> ]
QUOTE '<replaceable class="parameter">quote_character</replaceable>'
ESCAPE '<replaceable class="parameter">escape_character</replaceable>'
FORCE_QUOTE { ( <replaceable class="parameter">column_name</replaceable> [, ...] ) | * }
*************** COPY { <replaceable class="parameter">ta
*** 254,259 ****
--- 255,271 ----
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>COMPRESSED</literal></term>
+ <listitem>
+ <para>
+ 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.
+ </para>
+ </listitem>
+ </varlistentry>
<varlistentry>
<term><literal>QUOTE</literal></term>
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 <sys/stat.h>
#include <netinet/in.h>
#include <arpa/inet.h>
+ #ifdef HAVE_LIBZ
+ #include <zlib.h>
+ #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 <sys/resource.h> /* for getrlimit */
#endif
+ #ifdef HAVE_LIBZ
+ #include <zlib.h>
+ #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 <dirent.h>
+ #ifdef HAVE_LIBZ
+ #include <zlib.h>
+ #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 <dirent.h> 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,