Merge pull request #66 from citusdata/develop

Version 1.3 release

META.json

@@ -2,15 +2,15 @@
    "name": "cstore_fdw",
    "abstract": "Columnar Store for PostgreSQL",
    "description": "PostgreSQL extension which implements a Columnar Store.",
-   "version": "1.2.0",
-   "maintainer": "Hadi Moshayedi <hadi@citusdata.com>",
+   "version": "1.3.0",
+   "maintainer": "Murat Tuncer <mtuncer@citusdata.com>",
    "license": "apache_2_0",
    "provides": {
       "cstore_fdw": {
          "abstract": "Foreign Data Wrapper for Columnar Store Tables",
-         "file": "cstore_fdw--1.2.sql",
+         "file": "cstore_fdw--1.3.sql",
          "docfile": "README.md",
-         "version": "1.2.0"
+         "version": "1.3.0"
       }
    },
    "prereqs": {

Makefile

@@ -11,9 +11,11 @@ OBJS = cstore.pb-c.o cstore_fdw.o cstore_writer.o cstore_reader.o \
        cstore_metadata_serialization.o
 
 EXTENSION = cstore_fdw
-DATA = cstore_fdw--1.2.sql cstore_fdw--1.1--1.2.sql cstore_fdw--1.0--1.1.sql
+DATA = cstore_fdw--1.3.sql cstore_fdw--1.2--1.3.sql cstore_fdw--1.1--1.2.sql \
+       cstore_fdw--1.0--1.1.sql
 
-REGRESS = create load query analyze data_types functions block_filtering drop insert copyto
+REGRESS = create load query analyze data_types functions block_filtering drop \
+		  insert copyto alter
 EXTRA_CLEAN = cstore.pb-c.h cstore.pb-c.c data/*.cstore data/*.cstore.footer \
               sql/block_filtering.sql sql/create.sql sql/data_types.sql sql/load.sql \
               sql/copyto.sql expected/block_filtering.out expected/create.out \

README.md

@@ -115,13 +115,13 @@ most efficient execution plan for each query.
 commands. We also don't support single row inserts.
 
 
-Updating from version 1.0 or 1.1 to 1.2
----------------------------------------
+Updating from version 1.0, 1.1 or 1.2 to 1.3
+--------------------------------------------
 
-To update your existing cstore_fdw installation from version 1.0 or 1.1 to 1.2,
+To update your existing cstore_fdw installation from version 1.0, 1.1, or 1.2 to 1.3
 you can take the following steps:
 
-* Download and install cstore_fdw version 1.2 using instructions from the "Building"
+* Download and install cstore_fdw version 1.3 using instructions from the "Building"
   section,
 * Restart the PostgreSQL server,
 * Run the ```ALTER EXTENSION cstore_fdw UPDATE;``` command.
@@ -284,6 +284,14 @@ the installation:
 Changeset
 ---------
 
+### Version 1.3
+
+* (Feature) Added support for ```ALTER TABLE ADD COLUMN``` and ```ALTER TABLE DROP COLUMN```.
+* (Feature) Added column list support in ```COPY FROM```.
+* (Optimization) Improve row count estimation, which results in better plans.
+* (Fix) Fix the deadlock issue during concurrent inserts.
+* (Fix) Return correct result when using whole row references.
+
 ### Version 1.2
 
 * (Feature) Added support for ```COPY TO```.

TODO.md

@@ -4,13 +4,10 @@ To see the list of features and bug-fixes planned for next releases, see our
 Requested Features
 ------------------
 
-* Improve query cost estimation
 * Improve write performance
 * Improve read performance
 * Add checksum logic
 * Add new compression methods
-* Enable ALTER FOREIGN TABLE ADD COLUMN
-* Enable ALTER FOREIGN TABLE DROP COLUMN
 * Enable INSERT/DELETE/UPDATE
 * Enable users other than superuser to safely create columnar tables (permissions)
 * Transactional semantics
@@ -27,8 +24,6 @@ Known Issues
   command prints incorrect file size.
 * If two different columnar tables are configured to point to the same file,
   writes to the underlying file aren't protected from each other.
-* Hstore and json types work. However, constructors for hstore and json types
-  applied on a tuple return NULL.
 * When a data load is in progress, concurrent reads on the table overestimate the
   page count.
 * We have a minor memory leak in CStoreEndWrite. We need to also free the
@@ -38,7 +33,9 @@ Known Issues
 * We don't yet incorporate the compression method's impact on disk I/O into cost
   estimates.
 * CitusDB integration errors:
-  * Concurrent staging cstore\_fdw tables doesn't work.
+* Concurrent staging cstore\_fdw tables doesn't work.
+* Setting a default value for column with ALTER TABLE has limited support for
+  existing rows.
 
 [roadmap]: https://github.com/citusdata/cstore_fdw/wiki/Roadmap
 

cstore_fdw--1.2--1.3.sql

@@ -0,0 +1,3 @@
+/* cstore_fdw/cstore_fdw--1.2--1.3.sql */
+
+-- No new functions or definitions were added in 1.3

cstore_fdw--1.2.sql → cstore_fdw--1.3.sql

@@ -1,4 +1,4 @@
-/* cstore_fdw/cstore_fdw--1.2.sql */
+/* cstore_fdw/cstore_fdw--1.3.sql */
 
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION cstore_fdw" to load this file. \quit

cstore_fdw.c

@@ -88,7 +88,7 @@ static ForeignScan * CStoreGetForeignPlan(PlannerInfo *root, RelOptInfo *baserel
 										  List *targetList, List *scanClauses);
 static double TupleCountEstimate(RelOptInfo *baserel, const char *filename);
 static BlockNumber PageCount(const char *filename);
-static List * ColumnList(RelOptInfo *baserel);
+static List * ColumnList(RelOptInfo *baserel, Oid foreignTableId);
 static void CStoreExplainForeignScan(ForeignScanState *scanState,
 									 ExplainState *explainState);
 static void CStoreBeginForeignScan(ForeignScanState *scanState, int executorFlags);
@@ -243,7 +243,7 @@ CStoreProcessUtility(Node *parseTree, const char *queryString,
 		List *droppedTables = DroppedCStoreFilenameList((DropStmt*) parseTree);
 
 		CallPreviousProcessUtility(parseTree, queryString, context,
-				                   paramListInfo, destReceiver, completionTag);
+								   paramListInfo, destReceiver, completionTag);
 
 		foreach(fileListCell, droppedTables)
 		{
@@ -385,23 +385,16 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
 	CStoreFdwOptions *cstoreFdwOptions = NULL;
 	MemoryContext tupleContext = NULL;
 
-	List *columnNameList = copyStatement->attlist;
-	if (columnNameList != NULL)
-	{
-		ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-						errmsg("copy column list is not supported")));
-	}
-
 	/* Only superuser can copy from or to local file */
 	CheckSuperuserPrivilegesForCopy(copyStatement);
 
 	Assert(copyStatement->relation != NULL);
 
 	/*
-	 * Open and lock the relation. We acquire ExclusiveLock to allow concurrent
-	 * reads, but block concurrent writes.
+	 * Open and lock the relation. We acquire ShareUpdateExclusiveLock to allow
+	 * concurrent reads, but block concurrent writes.
 	 */
-	relation = heap_openrv(copyStatement->relation, ExclusiveLock);
+	relation = heap_openrv(copyStatement->relation, ShareUpdateExclusiveLock);
 	relationId = RelationGetRelid(relation);
 
 	/* allocate column values and nulls arrays */
@@ -427,7 +420,8 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
 
 	/* init state to read from COPY data source */
 	copyState = BeginCopyFrom(relation, copyStatement->filename,
-							  copyStatement->is_program, NIL,
+							  copyStatement->is_program,
+							  copyStatement->attlist,
 							  copyStatement->options);
 
 	/* init state to write to the cstore file */
@@ -457,7 +451,7 @@ CopyIntoCStoreTable(const CopyStmt *copyStatement, const char *queryString)
 	/* end read/write sessions and close the relation */
 	EndCopyFrom(copyState);
 	CStoreEndWrite(writeState);
-	heap_close(relation, ExclusiveLock);
+	heap_close(relation, ShareUpdateExclusiveLock);
 
 	return processedRowCount;
 }
@@ -1115,7 +1109,7 @@ CStoreGetForeignPaths(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId
 	 * algorithm and using the correlation statistics to detect which columns
 	 * are in stored in sorted order.
 	 */
-	List *queryColumnList = ColumnList(baserel);
+	List *queryColumnList = ColumnList(baserel, foreignTableId);
 	uint32 queryColumnCount = list_length(queryColumnList);
 	BlockNumber relationPageCount = PageCount(cstoreFdwOptions->filename);
 	uint32 relationColumnCount = RelationGetNumberOfAttributes(relation);
@@ -1177,7 +1171,7 @@ CStoreGetForeignPlan(PlannerInfo *root, RelOptInfo *baserel, Oid foreignTableId,
 	 * in executor's callback functions, so we get the column list here and put
 	 * it into foreign scan node's private list.
 	 */
-	columnList = ColumnList(baserel);
+	columnList = ColumnList(baserel, foreignTableId);
 	foreignPrivateList = list_make1(columnList);
 
 	/* create the foreign scan node */
@@ -1213,24 +1207,7 @@ TupleCountEstimate(RelOptInfo *baserel, const char *filename)
 	}
 	else
 	{
-		/*
-		 * Otherwise we have to fake it. We back into this estimate using the
-		 * planner's idea of relation width, which may be inaccurate. For better
-		 * estimates, users need to run ANALYZE.
-		 */
-		struct stat statBuffer;
-		int tupleWidth = 0;
-
-		int statResult = stat(filename, &statBuffer);
-		if (statResult < 0)
-		{
-			/* file may not be there at plan time, so use a default estimate */
-			statBuffer.st_size = 10 * BLCKSZ;
-		}
-
-		tupleWidth = MAXALIGN(baserel->width) + MAXALIGN(sizeof(HeapTupleHeaderData));
-		tupleCountEstimate = (double) statBuffer.st_size / (double) tupleWidth;
-		tupleCountEstimate = clamp_row_est(tupleCountEstimate);
+		tupleCountEstimate = (double) CStoreTableRowCount(filename);
 	}
 
 	return tupleCountEstimate;
@@ -1268,7 +1245,7 @@ PageCount(const char *filename)
  * and returns them in a new list. This function is unchanged from mongo_fdw.
  */
 static List *
-ColumnList(RelOptInfo *baserel)
+ColumnList(RelOptInfo *baserel, Oid foreignTableId)
 {
 	List *columnList = NIL;
 	List *neededColumnList = NIL;
@@ -1277,6 +1254,10 @@ ColumnList(RelOptInfo *baserel)
 	List *targetColumnList = baserel->reltargetlist;
 	List *restrictInfoList = baserel->baserestrictinfo;
 	ListCell *restrictInfoCell = NULL;
+	const AttrNumber wholeRow = 0;
+	Relation relation = heap_open(foreignTableId, AccessShareLock);
+	TupleDesc tupleDescriptor = RelationGetDescr(relation);
+	Form_pg_attribute *attributeFormArray = tupleDescriptor->attrs;
 
 	/* first add the columns used in joins and projections */
 	neededColumnList = list_copy(targetColumnList);
@@ -1311,6 +1292,16 @@ ColumnList(RelOptInfo *baserel)
 				column = neededColumn;
 				break;
 			}
+			else if (neededColumn->varattno == wholeRow)
+			{
+				Form_pg_attribute attributeForm = attributeFormArray[columnIndex - 1];
+				Index tableId = neededColumn->varno;
+
+				column = makeVar(tableId, columnIndex, attributeForm->atttypid,
+								 attributeForm->atttypmod, attributeForm->attcollation,
+								 0);
+				break;
+			}
 		}
 
 		if (column != NULL)
@@ -1319,6 +1310,8 @@ ColumnList(RelOptInfo *baserel)
 		}
 	}
 
+	heap_close(relation, AccessShareLock);
+
 	return columnList;
 }
 
@@ -1509,10 +1502,12 @@ CStoreAcquireSampleRows(Relation relation, int logLevel,
 		Form_pg_attribute attributeForm = attributeFormArray[columnIndex];
 		const Index tableId = 1;
 
-		Var *column = makeVar(tableId, columnIndex + 1, attributeForm->atttypid,
-							  attributeForm->atttypmod, attributeForm->attcollation, 0);
-
-		columnList = lappend(columnList, column);
+		if (!attributeForm->attisdropped)
+		{
+			Var *column = makeVar(tableId, columnIndex + 1, attributeForm->atttypid,
+								  attributeForm->atttypmod, attributeForm->attcollation, 0);
+			columnList = lappend(columnList, column);
+		}
 	}
 
 	/* setup foreign scan plan node */
@@ -1703,7 +1698,7 @@ CStoreBeginForeignModify(ModifyTableState *modifyTableState,
 	Assert (modifyTableState->operation == CMD_INSERT);
 
 	foreignTableOid = RelationGetRelid(relationInfo->ri_RelationDesc);
-	relation = heap_open(foreignTableOid, ExclusiveLock);
+	relation = heap_open(foreignTableOid, ShareUpdateExclusiveLock);
 	cstoreFdwOptions = CStoreGetOptions(foreignTableOid);
 	tupleDescriptor = RelationGetDescr(relationInfo->ri_RelationDesc);
 
@@ -1757,7 +1752,7 @@ CStoreEndForeignModify(EState *executorState, ResultRelInfo *relationInfo)
 		Relation relation = writeState->relation;
 
 		CStoreEndWrite(writeState);
-		heap_close(relation, ExclusiveLock);
+		heap_close(relation, ShareUpdateExclusiveLock);
 	}
 }
 

cstore_fdw.control

@@ -1,5 +1,5 @@
 # cstore_fdw extension
 comment = 'foreign-data wrapper for flat cstore access'
-default_version = '1.2'
+default_version = '1.3'
 module_pathname = '$libdir/cstore_fdw'
 relocatable = true

cstore_fdw.h

@@ -47,7 +47,7 @@
 /* CStore file signature */
 #define CSTORE_MAGIC_NUMBER "citus_cstore"
 #define CSTORE_VERSION_MAJOR 1
-#define CSTORE_VERSION_MINOR 2
+#define CSTORE_VERSION_MINOR 3
 
 /* miscellaneous defines */
 #define CSTORE_FDW_NAME "cstore_fdw"
@@ -334,6 +334,7 @@ extern ColumnBlockData ** CreateEmptyBlockDataArray(uint32 columnCount, bool *co
 													uint32 blockRowCount);
 extern void FreeColumnBlockDataArray(ColumnBlockData **blockDataArray,
 									 uint32 columnCount);
+extern uint64 CStoreTableRowCount(const char *filename);
 
 
 #endif   /* CSTORE_FDW_H */ 

cstore_metadata_serialization.c

@@ -408,6 +408,41 @@ DeserializeBlockCount(StringInfo buffer)
 }
 
 
+/*
+ * DeserializeRowCount deserializes the given column skip list buffer and
+ * returns the total number of rows in block skip list.
+ */
+uint32
+DeserializeRowCount(StringInfo buffer)
+{
+	uint32 rowCount = 0;
+	Protobuf__ColumnBlockSkipList *protobufBlockSkipList = NULL;
+	uint32 blockIndex = 0;
+	uint32 blockCount = 0;
+
+	protobufBlockSkipList =
+		protobuf__column_block_skip_list__unpack(NULL, buffer->len,
+												 (uint8 *) buffer->data);
+	if (protobufBlockSkipList == NULL)
+	{
+		ereport(ERROR, (errmsg("could not unpack column store"),
+						errdetail("invalid skip list buffer")));
+	}
+
+	blockCount = (uint32) protobufBlockSkipList->n_blockskipnodearray;
+	for (blockIndex = 0; blockIndex < blockCount; blockIndex++)
+	{
+		Protobuf__ColumnBlockSkipNode *protobufBlockSkipNode =
+				protobufBlockSkipList->blockskipnodearray[blockIndex];
+		rowCount += protobufBlockSkipNode->rowcount;
+	}
+
+	protobuf__column_block_skip_list__free_unpacked(protobufBlockSkipList, NULL);
+
+	return rowCount;
+}
+
+
 /*
  * DeserializeColumnSkipList deserializes the given buffer and returns the result as
  * a ColumnBlockSkipNode array. If the number of unpacked block skip nodes are not

cstore_metadata_serialization.h

@@ -32,6 +32,7 @@ extern StringInfo SerializeColumnSkipList(ColumnBlockSkipNode *blockSkipNodeArra
 extern void DeserializePostScript(StringInfo buffer, uint64 *tableFooterLength);
 extern TableFooter * DeserializeTableFooter(StringInfo buffer);
 extern uint32 DeserializeBlockCount(StringInfo buffer);
+extern uint32 DeserializeRowCount(StringInfo buffer);
 extern StripeFooter * DeserializeStripeFooter(StringInfo buffer);
 extern ColumnBlockSkipNode * DeserializeColumnSkipList(StringInfo buffer,
 													   bool typeByValue, int typeLength,