Add first/last group aggregates

knizhnik · knizhnik · commit 37fa849dca7a · 2018-12-06T11:00:40.000+03:00
diff --git a/README.md b/README.md
diff --git a/deparse.c b/deparse.c
@@ -684,6 +684,7 @@ deparseTargetList(StringInfo buf,
 		*retrieved_attrs = lappend_int(*retrieved_attrs,
 									   SelfItemPointerAttributeNumber);
 	}
+#if PG_VERSION_NUM<120000
 	if (bms_is_member(ObjectIdAttributeNumber - FirstLowInvalidHeapAttributeNumber,
 					  attrs_used))
 	{
@@ -700,7 +701,7 @@ deparseTargetList(StringInfo buf,
 		*retrieved_attrs = lappend_int(*retrieved_attrs,
 									   ObjectIdAttributeNumber);
 	}
-
+#endif
 	/* Don't generate bad syntax if no undropped columns */
 	if (first && !is_returning)
 		appendStringInfoString(buf, "NULL");
@@ -1256,12 +1257,14 @@ deparseColumnRef(StringInfo buf, int varno, int varattno, PlannerInfo *root,
 			ADD_REL_QUALIFIER(buf, varno);
 		appendStringInfoString(buf, "ctid");
 	}
+#if PG_VERSION_NUM<120000
 	else if (varattno == ObjectIdAttributeNumber)
 	{
 		if (qualify_col)
 			ADD_REL_QUALIFIER(buf, varno);
 		appendStringInfoString(buf, "oid");
 	}
+#endif
 	else if (varattno < 0)
 	{
 		/*
diff --git a/expected/test.out b/expected/test.out
@@ -360,11 +360,11 @@ select avg((open+close)/2),max(high-low) from stock group by symbol;
 
 set vops.auto_substitute_projections=on;
 explain (costs off) select avg((open+close)/2),max(high-low) from stock group by symbol;
-        QUERY PLAN        
---------------------------
+           QUERY PLAN           
+--------------------------------
  HashAggregate
    Group Key: symbol
-   ->  Seq Scan on vstock
+   ->  Seq Scan on vstock stock
 (3 rows)
 
 select avg((open+close)/2),max(high-low) from stock group by symbol;
@@ -398,6 +398,105 @@ select avg((open+close)/2),max(high-low) from stock group by symbol;
  10.9200000762939 |   1
 (1 row)
 
+create table wiki_data(
+   cat_id bigint,
+   page_id bigint,
+   requests int,
+   size bigint,
+   dyear int,
+   dmonth int,
+   dday int,
+   dhour int
+);
+create table wiki_cat
+(   cat_id bigint primary key,
+    category varchar(20))
+;
+insert into wiki_data values
+(101,1001,123,456),
+(101,1002,789,123),
+(101,1003,456,789),
+(102,2001,123,456),
+(102,2002,789,123),
+(103,3001,456,789);
+insert into wiki_cat values
+(101, 'cat 101'),
+(102, 'cat 102'),
+(103, 'cat 103');
+select create_projection('wiki_data_prj', 'wiki_data', array['page_id','requests','size'],array['cat_id']);
+ create_projection 
+-------------------
+ 
+(1 row)
+
+select wiki_data_prj_refresh();
+ wiki_data_prj_refresh 
+-----------------------
+                     6
+(1 row)
+
+SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+ category | sum  | sum  
+----------+------+------
+ cat 101  | 1368 | 1368
+ cat 103  |  456 |  789
+ cat 102  |  912 |  579
+(3 rows)
+
+set vops.auto_substitute_projections=on;
+SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+ category | sum  | sum  
+----------+------+------
+ cat 101  | 1368 | 1368
+ cat 103  |  456 |  789
+ cat 102  |  912 |  579
+(3 rows)
+
+explain (costs off) SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+                             QUERY PLAN                              
+---------------------------------------------------------------------
+ Limit
+   ->  Sort
+         Sort Key: (sum(wiki_data.size)) DESC
+         ->  HashAggregate
+               Group Key: wiki_cat.category
+               ->  Hash Join
+                     Hash Cond: (wiki_cat.cat_id = wiki_data.cat_id)
+                     ->  Seq Scan on wiki_cat
+                     ->  Hash
+                           ->  Seq Scan on wiki_data_prj wiki_data
+(10 rows)
+
 create table quote(symbol char(5), ts timestamp, ask_price real, ask_size integer, bid_price real, bid_size integer);
 insert into quote values
 ('AAA', '03-12-2018 10:00', 10.0, 100, 10.1, 202),
diff --git a/sql/test.sql b/sql/test.sql
@@ -59,6 +59,77 @@ select avg((open+close)/2),max(high-low) from stock group by symbol;
 set vops.auto_substitute_projections=off;
 select avg((open+close)/2),max(high-low) from stock group by symbol;
 
+create table wiki_data(
+   cat_id bigint,
+   page_id bigint,
+   requests int,
+   size bigint,
+   dyear int,
+   dmonth int,
+   dday int,
+   dhour int
+);
+
+create table wiki_cat
+(   cat_id bigint primary key,
+    category varchar(20))
+;
+
+insert into wiki_data values
+(101,1001,123,456),
+(101,1002,789,123),
+(101,1003,456,789),
+(102,2001,123,456),
+(102,2002,789,123),
+(103,3001,456,789);
+
+insert into wiki_cat values
+(101, 'cat 101'),
+(102, 'cat 102'),
+(103, 'cat 103');
+
+select create_projection('wiki_data_prj', 'wiki_data', array['page_id','requests','size'],array['cat_id']);
+
+select wiki_data_prj_refresh();
+
+SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+
+set vops.auto_substitute_projections=on;
+
+SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+
+explain (costs off) SELECT
+	category,
+	sum( requests ),
+	sum( size )
+FROM
+	wiki_data
+	INNER JOIN wiki_cat
+	 ON wiki_data.cat_id = wiki_cat.cat_id
+GROUP BY
+	category
+ORDER BY 3 DESC limit 5;
+
 create table quote(symbol char(5), ts timestamp, ask_price real, ask_size integer, bid_price real, bid_size integer);
 insert into quote values
 ('AAA', '03-12-2018 10:00', 10.0, 100, 10.1, 202),
@@ -71,3 +142,4 @@ select create_projection('vquote','quote',array['ts','ask_price','ask_size','bid
 select vquote_refresh();
 
 select first(bid_price,ts),last(ask_size,ts) from vquote group by symbol;
+
diff --git a/vops.c b/vops.c
@@ -1037,15 +1037,20 @@ static void
 UserTableUpdateOpenIndexes()
 {
 	List	   *recheckIndexes = NIL;
+#if PG_VERSION_NUM>=120000
+	HeapTuple tuple = ExecFetchSlotHeapTuple(slot, true, NULL);
+#else
+	HeapTuple tuple = slot->tts_tuple;
+#endif
 
 	/* HOT update does not require index inserts */
-	if (HeapTupleIsHeapOnly(slot->tts_tuple))
+	if (HeapTupleIsHeapOnly(tuple))
 		return;
 
 	if (estate->es_result_relation_info->ri_NumIndices > 0)
 	{
 		recheckIndexes = ExecInsertIndexTuples(slot,
-											   &slot->tts_tuple->t_self,
+											   &tuple->t_self,
 											   estate, false, NULL, NIL);
 
 		if (recheckIndexes != NIL)
@@ -1060,7 +1065,7 @@ static void begin_batch_insert(Oid oid)
 {
 	ResultRelInfo *resultRelInfo;
 
-	rel = heap_open(oid, NoLock);
+	rel = heap_open(oid, RowExclusiveLock);
 
 	PushActiveSnapshot(GetTransactionSnapshot());
 
@@ -1075,7 +1080,9 @@ static void begin_batch_insert(Oid oid)
 	estate->es_num_result_relations = 1;
 	estate->es_result_relation_info = resultRelInfo;
 	ExecOpenIndices(estate->es_result_relation_info, false);
-#if PG_VERSION_NUM>=110000
+#if PG_VERSION_NUM>=120000
+	slot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel), &TTSOpsHeapTuple);
+#elif PG_VERSION_NUM>=110000
 	slot = ExecInitExtraTupleSlot(estate, RelationGetDescr(rel));
 #else
 	slot = ExecInitExtraTupleSlot(estate);
@@ -1086,8 +1093,13 @@ static void begin_batch_insert(Oid oid)
 static void insert_tuple(Datum* values, bool* nulls)
 {
 	HeapTuple tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+#if PG_VERSION_NUM>=120000
+	ExecStoreHeapTuple(tup, slot, true);
+	simple_heap_insert(rel, ExecFetchSlotHeapTuple(slot, true, NULL));
+#else
 	ExecStoreTuple(tup, slot, InvalidBuffer, true);
 	simple_heap_insert(rel, slot->tts_tuple);
+#endif
     UserTableUpdateOpenIndexes();
 }
 
@@ -3162,7 +3174,11 @@ Datum vops_unnest(PG_FUNCTION_ARGS)
         user_ctx->nulls = (bool*)palloc(sizeof(bool)*n_attrs);
         user_ctx->types = (vops_type*)palloc(sizeof(vops_type)*n_attrs);
 		user_ctx->tiles = (vops_tile_hdr**)palloc(sizeof(vops_tile_hdr*)*n_attrs);
+#if PG_VERSION_NUM>=120000
+        user_ctx->desc = CreateTemplateTupleDesc(n_attrs);
+#else
         user_ctx->desc = CreateTemplateTupleDesc(n_attrs, false);
+#endif
         func_ctx->user_fctx = user_ctx;
         user_ctx->n_attrs = n_attrs;
 		user_ctx->tile_pos = 0;
@@ -3721,10 +3737,6 @@ vops_pullvars_walker(Node *node, vops_pullvar_context *ctx)
 		ctx->scope = scope;
 		node = (Node*)from->fromlist;
 	}
-	else if (IsA(node, Query))
-	{
-		return query_tree_walker((Query*)node, vops_pullvars_walker, ctx, 0);
-	}
 	(void) expression_tree_walker(node, vops_pullvars_walker, ctx);
 	ctx->scope = scope;
 	return false;
@@ -3838,6 +3850,42 @@ vops_add_literal_type_casts(Node* node, Const** consts)
 	return node;
 }
 
+static RangeVar*
+vops_get_join_rangevar(Node* node, int* relno)
+{
+	if (IsA(node, JoinExpr))
+	{
+		RangeVar* rv;
+		JoinExpr* join = (JoinExpr*)node;
+		rv = vops_get_join_rangevar(join->larg, relno);
+		if (rv != NULL)
+			return rv;
+		rv = vops_get_join_rangevar(join->rarg, relno);
+		if (rv != NULL)
+			return rv;
+	}
+	else
+	{
+		Assert(IsA(node, RangeVar));
+		if (--*relno == 0)
+			return (RangeVar*)node;
+	}
+	return NULL;
+}
+
+static RangeVar*
+vops_get_rangevar(List* from, int relno)
+{
+	ListCell* cell;
+	foreach (cell, from)
+	{
+		Node* node = (Node*)lfirst(cell);
+		RangeVar* rv = vops_get_join_rangevar(node, &relno);
+		if (rv != NULL)
+			return rv;
+	}
+	Assert(false);
+}
 /*
  * Try to substitute tables with their VOPS projections.
  * Criterias for such substitution:
@@ -3871,7 +3919,7 @@ vops_substitute_tables_with_projections(char const* queryString, Query *query)
 	pullvar_ctx.consts = (Const**)palloc0((strlen(queryString) + 1)*sizeof(Const*));
 	pullvar_ctx.scope = SCOPE_DEFAULT;
 	pullvar_ctx.refs = palloc0(n_rels*sizeof(vops_table_refs));
-	query_tree_walker(query, vops_pullvars_walker, &pullvar_ctx, 0);
+	query_tree_walker(query, vops_pullvars_walker, &pullvar_ctx, QTW_IGNORE_CTE_SUBQUERIES|QTW_IGNORE_RANGE_TABLE);
 
     SPI_connect();
 
@@ -3954,7 +4002,10 @@ vops_substitute_tables_with_projections(char const* queryString, Query *query)
 				}
 				/* Replace table with partition */
 				elog(DEBUG1, "Use projection %s instead of table %d", projectionName, rte->relid);
-				rv = list_nth_node(RangeVar, select->fromClause, fromno-1);
+				rv = vops_get_rangevar(select->fromClause, fromno);
+				Assert(rv != NULL);
+				if (rv->alias == NULL)
+					rv->alias = makeAlias(rv->relname, NULL);
 				rv->relname = pstrdup(projectionName);
 
 				/* Update vector/scalar bitmap sets for this query for this projection */
@@ -4135,8 +4186,13 @@ static void vops_explain_hook(Query *query,
 		params == NULL)    /* do not support prepared statements yet */
 	{
 		char* explain = pstrdup(queryString);
-		char* select = strstr(explain, "select");
-		size_t prefix = (select != NULL) ? (select - explain) : 7;
+		char* select;
+		size_t prefix;
+		select = strstr(explain, "select");
+		if (select == NULL) {
+			select = strstr(explain, "SELECT");
+		}
+		prefix = (select != NULL) ? (select - explain) : 7;
 		memset(explain, ' ', prefix);  /* clear "explain" prefix: we need to preseve node locations */
 		vops_substitute_tables_with_projections(explain, query);
 	}
diff --git a/vops_fdw.c b/vops_fdw.c

Original file line number	Diff line number	Diff line change
`@@ -684,6 +684,7 @@ deparseTargetList(StringInfo buf,`
`684`	`684`	`retrieved_attrs = lappend_int(retrieved_attrs,`
`685`	`685`	`SelfItemPointerAttributeNumber);`
`686`	`686`	`}`
	`687`	`+#if PG_VERSION_NUM<120000`
`687`	`688`	`if (bms_is_member(ObjectIdAttributeNumber - FirstLowInvalidHeapAttributeNumber,`
`688`	`689`	`attrs_used))`
`689`	`690`	`{`
`@@ -700,7 +701,7 @@ deparseTargetList(StringInfo buf,`
`700`	`701`	`retrieved_attrs = lappend_int(retrieved_attrs,`
`701`	`702`	`ObjectIdAttributeNumber);`
`702`	`703`	`}`
`703`		`-`
	`704`	`+#endif`
`704`	`705`	`/* Don't generate bad syntax if no undropped columns */`
`705`	`706`	`if (first && !is_returning)`
`706`	`707`	`appendStringInfoString(buf, "NULL");`
`@@ -1256,12 +1257,14 @@ deparseColumnRef(StringInfo buf, int varno, int varattno, PlannerInfo *root,`
`1256`	`1257`	`ADD_REL_QUALIFIER(buf, varno);`
`1257`	`1258`	`appendStringInfoString(buf, "ctid");`
`1258`	`1259`	`}`
	`1260`	`+#if PG_VERSION_NUM<120000`
`1259`	`1261`	`else if (varattno == ObjectIdAttributeNumber)`
`1260`	`1262`	`{`
`1261`	`1263`	`if (qualify_col)`
`1262`	`1264`	`ADD_REL_QUALIFIER(buf, varno);`
`1263`	`1265`	`appendStringInfoString(buf, "oid");`
`1264`	`1266`	`}`
	`1267`	`+#endif`
`1265`	`1268`	`else if (varattno < 0)`
`1266`	`1269`	`{`
`1267`	`1270`	`/*`