Implement sort pushdown.

Author: rjuju

Makefile

@@ -109,7 +109,8 @@ TESTS        = test-$(PYTHON_TEST_VERSION)/sql/multicorn_cache_invalidation.sql
   test-$(PYTHON_TEST_VERSION)/sql/multicorn_sequence_test.sql \
   test-$(PYTHON_TEST_VERSION)/sql/multicorn_test_date.sql \
   test-$(PYTHON_TEST_VERSION)/sql/multicorn_test_dict.sql \
-  test-$(PYTHON_TEST_VERSION)/sql/multicorn_test_list.sql
+  test-$(PYTHON_TEST_VERSION)/sql/multicorn_test_list.sql \
+  test-$(PYTHON_TEST_VERSION)/sql/multicorn_test_sort.sql
 
 ifeq (${UNSUPPORTS_SQLALCHEMY}, 0)
   TESTS += test-$(PYTHON_TEST_VERSION)/sql/multicorn_alchemy_test.sql

python/multicorn/__init__.py

@@ -11,6 +11,7 @@
 """
 
 import sys
+from collections import namedtuple
 
 __version__ = '__VERSION__'
 
@@ -19,6 +20,11 @@
 UNBOUND = object()
 
 
+SortKey = namedtuple("SortKey",
+                              ["attname", "attnum", "is_reversed",
+                               "nulls_first", "collate"])
+
+
 class Qual(object):
     """A Qual describes a postgresql qualifier.
 
@@ -152,6 +158,32 @@ def get_rel_size(self, quals, columns):
         """
         return (100000000, len(columns) * 100)
 
+    def can_sort(self, pathkeys):
+        """
+        Method called from the planner to ask the FDW what are the sorts it can
+        enforced, to avoid PostgreSQL to sort the data after retreiving all the
+        rows. These sorts can come from explicit ORDER BY clauses, but also GROUP
+        BY and DISTINCT clauses.
+
+        The FDW has to inspect every sort, and respond which one are handled.
+        The sorts are cumulatives. For example::
+
+            col1 ASC
+            col2 DESC
+
+        means that the FDW must render the tuples sorted by col1 ascending and
+        col2 descending.
+
+        Args:
+            pathkeys (list): A list of :class:`SortKey`
+                representing all the sorts the query must enforce.
+
+        Return:
+            The list of cumulative SortKey, for which the FDW can
+            enforce the sort.
+        """
+        return []
+
     def get_path_keys(self):
         u"""
         Method called from the planner to add additional Path to the planner.
@@ -209,7 +241,7 @@ def get_path_keys(self):
         """
         return []
 
-    def execute(self, quals, columns):
+    def execute(self, quals, columns, pathkeys=[]):
         """Execute a query in the foreign data wrapper.
 
         This method is called at the first iteration.
@@ -240,13 +272,17 @@ def execute(self, quals, columns):
                 You should return AT LEAST those columns when returning a
                 dict. If returning a sequence, every column from the table
                 should be in the sequence.
+            pathkeys (list): A list of :class:`MulticornDeparsedSortGroup`
+                that the FDW said it can enforce.
 
         Returns:
             An iterable of python objects which can be converted back to PostgreSQL.
             Currently, such objects are:
             - sequences containing exactly as much columns as the
             underlying tables
             - dictionaries mapping column names to their values.
+            If the pathkeys wasn't empty, the FDW has to return the data in the
+            expected order.
 
         """
         pass

python/multicorn/testfdw.py

@@ -4,6 +4,7 @@
 from .utils import log_to_postgres, WARNING, ERROR
 from itertools import cycle
 from datetime import datetime
+from operator import itemgetter
 
 
 class TestForeignDataWrapper(ForeignDataWrapper):
@@ -70,14 +71,30 @@ def _as_generator(self, quals, columns):
                                                           index)
             yield line
 
-    def execute(self, quals, columns):
+    def execute(self, quals, columns, pathkeys=[]):
         log_to_postgres(str(sorted(quals)))
         log_to_postgres(str(sorted(columns)))
+        if (len(pathkeys)) > 0:
+            log_to_postgres("requested sort(s): ")
+            for k in pathkeys:
+                log_to_postgres(k)
         if self.test_type == 'None':
             return None
         elif self.test_type == 'iter_none':
             return [None, None]
         else:
+            if (len(pathkeys) > 0):
+                # testfdw don't have tables with more than 2 fields, without
+                # duplicates, so we only need to worry about sorting on 1st
+                # asked column
+                k = pathkeys[0];
+                res = self._as_generator(quals, columns)
+                if (self.test_type == 'sequence'):
+                    return sorted(res, key=itemgetter(k.attnum - 1),
+                                  reverse=k.is_reversed)
+                else:
+                    return sorted(res, key=itemgetter(k.attname),
+                                  reverse=k.is_reversed)
             return self._as_generator(quals, columns)
 
     def get_rel_size(self, quals, columns):
@@ -90,6 +107,10 @@ def get_path_keys(self):
             return [(('test1',), 1)]
         return []
 
+    def can_sort(self, pathkeys):
+        # assume sort pushdown ok for all cols, in any order, any collation
+        return pathkeys
+
     def update(self, rowid, newvalues):
         if self.test_type == 'nowrite':
             super(TestForeignDataWrapper, self).update(rowid, newvalues)

src/multicorn.c

@@ -22,6 +22,7 @@
 #include "catalog/pg_type.h"
 #include "utils/memutils.h"
 #include "miscadmin.h"
+#include "utils/lsyscache.h"
 #include "utils/rel.h"
 #include "parser/parsetree.h"
 
@@ -304,23 +305,65 @@ multicornGetForeignPaths(PlannerInfo *root,
 						 RelOptInfo *baserel,
 						 Oid foreigntableid)
 {
-	Path	   *path;
-	MulticornPlanState *planstate = baserel->fdw_private;
+	List				*pathes; /* List of ForeignPath */
+	MulticornPlanState	*planstate = baserel->fdw_private;
+	ListCell		    *lc;
+
+	/* These lists are used to handle sort pushdown */
+	List				*apply_pathkeys = NULL;
+	List				*deparsed_pathkeys = NULL;
 
 	/* Extract a friendly version of the pathkeys. */
 	List	   *possiblePaths = pathKeys(planstate);
 
-	findPaths(root, baserel, possiblePaths, planstate->startupCost);
-	/* Add a default path */
-	path = (Path *) create_foreignscan_path(root, baserel,
-											baserel->rows,
-											planstate->startupCost,
-											baserel->rows * baserel->width,
-											NIL,		/* no pathkeys */
-											NULL,		/* no outer rel either */
-											(void *) baserel->fdw_private);
-
-	add_path(baserel, path);
+	/* Try to find parameterized paths */
+	pathes = findPaths(root, baserel, possiblePaths, planstate->startupCost,
+			planstate, apply_pathkeys, deparsed_pathkeys);
+
+	/* Add a simple default path */
+	pathes = lappend(pathes, create_foreignscan_path(root, baserel,
+			baserel->rows,
+			planstate->startupCost,
+			baserel->rows * baserel->width,
+			NIL,		/* no pathkeys */
+			NULL,		/* no outer rel either */
+			NULL));
+
+	/* Handle sort pushdown */
+	if (root->query_pathkeys)
+	{
+		List		*deparsed = deparse_sortgroup(root, foreigntableid, baserel);
+
+		if (deparsed)
+		{
+			/* Update the sort_*_pathkeys lists if needed */
+			computeDeparsedSortGroup(deparsed, planstate, &apply_pathkeys,
+					&deparsed_pathkeys);
+		}
+	}
+
+	/* Add each ForeignPath previously found */
+	foreach(lc, pathes)
+	{
+		ForeignPath *path = (ForeignPath *) lfirst(lc);
+
+		/* Add the path without modification */
+		add_path(baserel, (Path *) path);
+
+		/* Add the path with sort pusdown if possible */
+		if (apply_pathkeys && deparsed_pathkeys)
+		{
+			ForeignPath *newpath;
+
+			newpath = create_foreignscan_path(root, baserel, path->path.rows,
+					path->path.startup_cost, path->path.total_cost,
+					apply_pathkeys, NULL,
+					(void *) deparsed_pathkeys);
+
+			newpath->path.param_info = path->path.param_info;
+			add_path(baserel, (Path *) newpath);
+		}
+	}
 	errorCheck();
 }
 
@@ -351,6 +394,7 @@ multicornGetForeignPlan(PlannerInfo *root,
 								&planstate->qual_list);
 		}
 	}
+	planstate->pathkeys = (List *) best_path->fdw_private;
 	return make_foreignscan(tlist,
 							scan_clauses,
 							scan_relid,
@@ -921,6 +965,9 @@ serializePlanState(MulticornPlanState * state)
 	result = lappend(result, makeConst(INT4OID,
 					-1, InvalidOid, -1, state->foreigntableid, false, true));
 	result = lappend(result, state->target_list);
+
+	result = lappend(result, serializeDeparsedSortGroup(state->pathkeys));
+
 	return result;
 }
 
@@ -935,10 +982,13 @@ initializeExecState(void *internalstate)
 	List	   *values = (List *) internalstate;
 	AttrNumber	attnum = ((Const *) linitial(values))->constvalue;
 	Oid			foreigntableid = ((Const *) lsecond(values))->constvalue;
+	List		*pathkeys;
 
 	/* Those list must be copied, because their memory context can become */
 	/* invalid during the execution (in particular with the cursor interface) */
 	execstate->target_list = copyObject(lthird(values));
+	pathkeys = lfourth(values);
+	execstate->pathkeys = deserializeDeparsedSortGroup(pathkeys);
 	execstate->fdw_instance = getInstance(foreigntableid);
 	execstate->buffer = makeStringInfo();
 	execstate->cinfos = palloc0(sizeof(ConversionInfo *) * attnum);

src/multicorn.h

@@ -57,6 +57,7 @@ typedef struct MulticornPlanState
 	List	   *qual_list;
 	int			startupCost;
 	ConversionInfo **cinfos;
+	List	   *pathkeys; /* list of MulticornDeparsedSortGroup) */
 }	MulticornPlanState;
 
 typedef struct MulticornExecState
@@ -74,6 +75,7 @@ typedef struct MulticornExecState
 	StringInfo	buffer;
 	AttrNumber	rowidAttno;
 	char	   *rowidAttrName;
+	List	   *pathkeys; /* list of MulticornDeparsedSortGroup) */
 }	MulticornExecState;
 
 typedef struct MulticornModifyState
@@ -117,6 +119,16 @@ typedef struct MulticornParamQual
 	Expr	   *expr;
 }	MulticornParamQual;
 
+typedef struct MulticornDeparsedSortGroup
+{
+	char			attname[NAMEDATALEN];
+	int				attnum;
+	bool			reversed;
+	bool			nulls_first;
+	char			collate[NAMEDATALEN];
+	PathKey	*key;
+} MulticornDeparsedSortGroup;
+
 /* errors.c */
 void		errorCheck(void);
 
@@ -138,12 +150,14 @@ PyObject   *optionsListToPyDict(List *options);
 const char *getPythonEncodingName(void);
 
 void getRelSize(MulticornPlanState * state,
-		   PlannerInfo *root,
-		   double *rows,
-		   int *width);
+		PlannerInfo *root,
+		double *rows,
+		int *width);
 
 List	   *pathKeys(MulticornPlanState * state);
 
+List	   *canSort(MulticornPlanState * state, List *deparsed);
+
 CacheEntry *getCacheEntry(Oid foreigntableid);
 UserMapping *multicorn_GetUserMapping(Oid userid, Oid serverid);
 
@@ -158,14 +172,27 @@ void extractRestrictions(Relids base_relids,
 					List **quals);
 List	   *extractColumns(List *reltargetlist, List *restrictinfolist);
 void initConversioninfo(ConversionInfo ** cinfo,
-				   AttInMetadata *attinmeta);
+		AttInMetadata *attinmeta);
 
 Value *colnameFromVar(Var *var, PlannerInfo *root,
-			   MulticornPlanState * state);
+		MulticornPlanState * state);
+
+void computeDeparsedSortGroup(List *deparsed, MulticornPlanState *planstate,
+		List **apply_pathkeys,
+		List **deparsed_pathkeys);
 
-void		findPaths(PlannerInfo *root, RelOptInfo *baserel, List *possiblePaths, int startupCost);
+List	*findPaths(PlannerInfo *root, RelOptInfo *baserel, List *possiblePaths,
+		int startupCost,
+		MulticornPlanState *state,
+		List *apply_pathkeys, List *deparsed_pathkeys);
+
+List        *deparse_sortgroup(PlannerInfo *root, Oid foreigntableid, RelOptInfo *rel);
 
 PyObject   *datumToPython(Datum node, Oid typeoid, ConversionInfo * cinfo);
+
+List	*serializeDeparsedSortGroup(List *pathkeys);
+List	*deserializeDeparsedSortGroup(List *items);
+
 #endif   /* PG_MULTICORN_H */
 
 char	   *PyUnicode_AsPgString(PyObject *p_unicode);