pgsql-v9.2-fix-leaky-view.part-1.v8.patch

 doc/src/sgml/ref/alter_view.sgml             |   20 ++
 doc/src/sgml/ref/create_view.sgml            |   13 +
 doc/src/sgml/rules.sgml                      |   39 +++-
 src/backend/access/common/reloptions.c       |   17 ++-
 src/backend/commands/tablecmds.c             |   66 ++++--
 src/backend/commands/view.c                  |   26 ++-
 src/backend/nodes/copyfuncs.c                |    1 +
 src/backend/nodes/equalfuncs.c               |    1 +
 src/backend/nodes/outfuncs.c                 |    1 +
 src/backend/nodes/readfuncs.c                |    1 +
 src/backend/optimizer/path/allpaths.c        |   14 ++
 src/backend/optimizer/prep/prepjointree.c    |    2 +
 src/backend/optimizer/util/plancat.c         |   88 +++++++
 src/backend/parser/gram.y                    |   10 +-
 src/backend/rewrite/rewriteHandler.c         |    1 +
 src/backend/utils/cache/relcache.c           |    1 +
 src/bin/pg_dump/pg_dump.c                    |    6 +-
 src/include/access/reloptions.h              |    3 +-
 src/include/nodes/parsenodes.h               |    3 +
 src/include/utils/rel.h                      |    9 +
 src/test/regress/expected/create_view.out    |   55 +++++-
 src/test/regress/expected/select_views_1.out |  314 ++++++++++++++++++++++++++
 src/test/regress/sql/create_view.sql         |   33 +++
 src/test/regress/sql/select_views.sql        |  162 +++++++++++++
 24 files changed, 851 insertions(+), 35 deletions(-)

diff --git a/doc/src/sgml/ref/alter_view.sgml b/doc/src/sgml/ref/alter_view.sgml
index c383905..e78176b 100644
--- a/doc/src/sgml/ref/alter_view.sgml
+++ b/doc/src/sgml/ref/alter_view.sgml
@@ -26,6 +26,8 @@ ALTER VIEW <replaceable class="parameter">name</replaceable> ALTER [ COLUMN ] <r
 ALTER VIEW <replaceable class="parameter">name</replaceable> OWNER TO <replaceable class="PARAMETER">new_owner</replaceable>
 ALTER VIEW <replaceable class="parameter">name</replaceable> RENAME TO <replaceable class="parameter">new_name</replaceable>
 ALTER VIEW <replaceable class="parameter">name</replaceable> SET SCHEMA <replaceable class="parameter">new_schema</replaceable>
+ALTER VIEW <replaceable class="parameter">name</replaceable> SET ( <replaceable class="parameter">view_option_name</replaceable> [= <replaceable class="parameter">view_option_value</replaceable>] [, ... ] )
+ALTER VIEW <replaceable class="parameter">name</replaceable> RESET ( <replaceable class="parameter">view_option_name</replaceable> [, ... ] )
 </synopsis>
  </refsynopsisdiv>
 
@@ -102,6 +104,24 @@ ALTER VIEW <replaceable class="parameter">name</replaceable> SET SCHEMA <replace
      </para>
     </listitem>
    </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">view_option_name</replaceable></term>
+    <listitem>
+     <para>
+      The name of a view option to be set or reset.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">view_option_name</replaceable></term>
+    <listitem>
+     <para>
+      The new value for a view option.
+     </para>
+    </listitem>
+   </varlistentry>
   </variablelist>
  </refsect1>
 
diff --git a/doc/src/sgml/ref/create_view.sgml b/doc/src/sgml/ref/create_view.sgml
index 417f8c3..6e86840 100644
--- a/doc/src/sgml/ref/create_view.sgml
+++ b/doc/src/sgml/ref/create_view.sgml
@@ -22,6 +22,7 @@ PostgreSQL documentation
  <refsynopsisdiv>
 <synopsis>
 CREATE [ OR REPLACE ] [ TEMP | TEMPORARY ] VIEW <replaceable class="PARAMETER">name</replaceable> [ ( <replaceable class="PARAMETER">column_name</replaceable> [, ...] ) ]
+    [ WITH ( <replaceable class="PARAMETER">view_option_name</replaceable> [= <replaceable class="PARAMETER">view_option_value</replaceable>] [, ... ] ) ]
     AS <replaceable class="PARAMETER">query</replaceable>
 </synopsis>
  </refsynopsisdiv>
@@ -99,6 +100,18 @@ CREATE [ OR REPLACE ] [ TEMP | TEMPORARY ] VIEW <replaceable class="PARAMETER">n
    </varlistentry>
 
    <varlistentry>
+    <term><literal>WITH ( <replaceable class="PARAMETER">view_option_name</replaceable> [= <replaceable class="PARAMETER">view_option_value</replaceable>] [, ... ] )</literal></term>
+    <listitem>
+     <para>
+      This clause specifies optional parameters for a view; currently, the
+      only suppored parameter name is <literal>security_barrier</literal>,
+      which should be enabled when a view is intended to provide row-level
+      security.  See <xref linkend="rules-privileges"> for full details.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
     <term><replaceable class="parameter">query</replaceable></term>
     <listitem>
      <para>
diff --git a/doc/src/sgml/rules.sgml b/doc/src/sgml/rules.sgml
index 1b06519..896cc64 100644
--- a/doc/src/sgml/rules.sgml
+++ b/doc/src/sgml/rules.sgml
@@ -1822,8 +1822,9 @@ GRANT SELECT ON phone_number TO secretary;
 <para>
     Note that while views can be used to hide the contents of certain
     columns using the technique shown above, they cannot be used to reliably
-    conceal the data in unseen rows.  For example, the following view is
-    insecure:
+    conceal the data in unseen rows unless the
+    <literal>security_barrier</literal> flag has been set.  For example,
+    the following view is insecure:
 <programlisting>
 CREATE VIEW phone_number AS
     SELECT person, phone FROM phone_data WHERE phone NOT LIKE '412%';
@@ -1870,6 +1871,40 @@ SELECT * FROM phone_number WHERE tricky(person, phone);
     which references <literal>shoelace_log</> is an unqualified
     <literal>INSERT</>.  This might not be true in more complex scenarios.
 </para>
+
+<para>
+    When it is necessary for a view to provide row-level security, the
+    <literal>security_barrier</literal> attribute should be applied to
+    the view.  This prevents maliciously-chosen functions and operators from
+    being invoked on rows until afterthe view has done its work.  For
+    example, if the view shown above had been created like this, it would
+    be secure:
+<programlisting>
+CREATE VIEW phone_number WITH (security_barrier) AS
+    SELECT person, phone FROM phone_data WHERE phone NOT LIKE '412%';
+</programlisting>
+    Views created with the <literal>security_barrier</literal> may perform
+    far worse than views created without this option.  In general, there is
+    no way to avoid this: the fastest possible plan must be rejected
+    if it may compromise security.  For this reason, this option is not
+    enabled by default.
+</para>
+
+<para>
+    It is important to understand that even a view created with the
+    <literal>security_barrier</literal> option is intended to be secure only
+    in the limited sense that the contents of the invisible tuples will not
+    passed to possibly-insecure functions.  The user may well have other means
+    of making inferences about the unseen data; for example, they can see the
+    query plan using <command>EXPLAIN</command>, or measure the runtime of
+    queries against the view.  A malicious attacker might be able to infer
+    something about the amount of unseen data, or even gain some information
+    about the data distribution or most common values (since these things may
+    affect the runtime of the plan; or even, since they are also reflected in
+    the optimizer statistics, the choice of plan).  If these types of "covert
+    channel" attacks are of concern, it is probably unwise to grant any access
+    to the data at all.
+</para>
 </sect1>
 
 <sect1 id="rules-status">
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index 240e178..5f39ff3 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -66,6 +66,14 @@ static relopt_bool boolRelOpts[] =
 		},
 		true
 	},
+	{
+		{
+			"security_barrier",
+			"View acts as a row security barrier",
+			RELOPT_KIND_VIEW
+		},
+		false
+	},
 	/* list terminator */
 	{{NULL}}
 };
@@ -772,6 +780,7 @@ extractRelOptions(HeapTuple tuple, TupleDesc tupdesc, Oid amoptions)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
+		case RELKIND_VIEW:
 		case RELKIND_UNCATALOGED:
 			options = heap_reloptions(classForm->relkind, datum, false);
 			break;
@@ -1130,7 +1139,9 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 		{"autovacuum_vacuum_scale_factor", RELOPT_TYPE_REAL,
 		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, vacuum_scale_factor)},
 		{"autovacuum_analyze_scale_factor", RELOPT_TYPE_REAL,
-		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)}
+		offsetof(StdRdOptions, autovacuum) +offsetof(AutoVacOpts, analyze_scale_factor)},
+		{"security_barrier", RELOPT_TYPE_BOOL,
+		offsetof(StdRdOptions, security_barrier)},
 	};
 
 	options = parseRelOptions(reloptions, validate, kind, &numoptions);
@@ -1150,7 +1161,7 @@ default_reloptions(Datum reloptions, bool validate, relopt_kind kind)
 }
 
 /*
- * Parse options for heaps and toast tables.
+ * Parse options for heaps, views and toast tables.
  */
 bytea *
 heap_reloptions(char relkind, Datum reloptions, bool validate)
@@ -1172,6 +1183,8 @@ heap_reloptions(char relkind, Datum reloptions, bool validate)
 			return (bytea *) rdopts;
 		case RELKIND_RELATION:
 			return default_reloptions(reloptions, validate, RELOPT_KIND_HEAP);
+		case RELKIND_VIEW:
+			return default_reloptions(reloptions, validate, RELOPT_KIND_VIEW);
 		default:
 			/* other relkinds are not supported */
 			return NULL;
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 1c3fe6a..bfbf5fa 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -364,7 +364,9 @@ static void ATExecDropCluster(Relation rel, LOCKMODE lockmode);
 static void ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel,
 					char *tablespacename, LOCKMODE lockmode);
 static void ATExecSetTableSpace(Oid tableOid, Oid newTableSpace, LOCKMODE lockmode);
-static void ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode);
+static void ATExecSetRelOptions(Relation rel, List *defList,
+								AlterTableType operation,
+								LOCKMODE lockmode);
 static void ATExecEnableDisableTrigger(Relation rel, char *trigname,
 					   char fires_when, bool skip_system, LOCKMODE lockmode);
 static void ATExecEnableDisableRule(Relation rel, char *rulename,
@@ -2785,6 +2787,7 @@ AlterTableGetLockLevel(List *cmds)
 			case AT_DropCluster:
 			case AT_SetRelOptions:
 			case AT_ResetRelOptions:
+			case AT_ReplaceRelOptions:
 			case AT_SetOptions:
 			case AT_ResetOptions:
 			case AT_SetStorage:
@@ -3013,8 +3016,9 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
 			pass = AT_PASS_MISC;	/* doesn't actually matter */
 			break;
 		case AT_SetRelOptions:	/* SET (...) */
-		case AT_ResetRelOptions:		/* RESET (...) */
-			ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX);
+		case AT_ResetRelOptions:	/* RESET (...) */
+		case AT_ReplaceRelOptions:	/* reset them all, then set just these */
+			ATSimplePermissions(rel, ATT_TABLE | ATT_INDEX | ATT_VIEW);
 			/* This command never recurses */
 			/* No command-specific prep needed */
 			pass = AT_PASS_MISC;
@@ -3257,12 +3261,10 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
 			 */
 			break;
 		case AT_SetRelOptions:	/* SET (...) */
-			ATExecSetRelOptions(rel, (List *) cmd->def, false, lockmode);
-			break;
 		case AT_ResetRelOptions:		/* RESET (...) */
-			ATExecSetRelOptions(rel, (List *) cmd->def, true, lockmode);
+		case AT_ReplaceRelOptions:		/* replace entire option list */
+			ATExecSetRelOptions(rel, (List *) cmd->def, cmd->subtype, lockmode);
 			break;
-
 		case AT_EnableTrig:		/* ENABLE TRIGGER name */
 			ATExecEnableDisableTrigger(rel, cmd->name,
 								   TRIGGER_FIRES_ON_ORIGIN, false, lockmode);
@@ -8090,10 +8092,11 @@ ATPrepSetTableSpace(AlteredTableInfo *tab, Relation rel, char *tablespacename, L
 }
 
 /*
- * ALTER TABLE/INDEX SET (...) or RESET (...)
+ * Set, reset, or replace reloptions.
  */
 static void
-ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode)
+ATExecSetRelOptions(Relation rel, List *defList, AlterTableType operation,
+					LOCKMODE lockmode)
 {
 	Oid			relid;
 	Relation	pgclass;
@@ -8107,28 +8110,44 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode
 	bool		repl_repl[Natts_pg_class];
 	static char *validnsps[] = HEAP_RELOPT_NAMESPACES;
 
-	if (defList == NIL)
+	if (defList == NIL && operation != AT_ReplaceRelOptions)
 		return;					/* nothing to do */
 
 	pgclass = heap_open(RelationRelationId, RowExclusiveLock);
 
-	/* Get the old reloptions */
+	/* Fetch heap tuple */
 	relid = RelationGetRelid(rel);
 	tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(relid));
 	if (!HeapTupleIsValid(tuple))
 		elog(ERROR, "cache lookup failed for relation %u", relid);
 
-	datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions, &isnull);
+	if (operation == AT_ReplaceRelOptions)
+	{
+		/*
+		 * If we're supposed to replace the reloptions list, we just pretend
+		 * there were none before.
+		 */
+		datum = (Datum) 0;
+		isnull = true;
+	}
+	else
+	{
+		/* Get the old reloptions */
+		datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
+								&isnull);
+	}
 
 	/* Generate new proposed reloptions (text array) */
 	newOptions = transformRelOptions(isnull ? (Datum) 0 : datum,
-								   defList, NULL, validnsps, false, isReset);
+								   defList, NULL, validnsps, false,
+								   operation == AT_ResetRelOptions);
 
 	/* Validate */
 	switch (rel->rd_rel->relkind)
 	{
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
+		case RELKIND_VIEW:
 			(void) heap_reloptions(rel->rd_rel->relkind, newOptions, true);
 			break;
 		case RELKIND_INDEX:
@@ -8176,15 +8195,30 @@ ATExecSetRelOptions(Relation rel, List *defList, bool isReset, LOCKMODE lockmode
 
 		toastrel = heap_open(toastid, lockmode);
 
-		/* Get the old reloptions */
+		/* Fetch heap tuple */
 		tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(toastid));
 		if (!HeapTupleIsValid(tuple))
 			elog(ERROR, "cache lookup failed for relation %u", toastid);
 
-		datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions, &isnull);
+		if (operation == AT_ReplaceRelOptions)
+		{
+			/*
+			 * If we're supposed to replace the reloptions list, we just
+			 * pretend there were none before.
+			 */
+			datum = (Datum) 0;
+			isnull = true;
+		}
+		else
+		{
+			/* Get the old reloptions */
+			datum = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
+									&isnull);
+		}
 
 		newOptions = transformRelOptions(isnull ? (Datum) 0 : datum,
-								defList, "toast", validnsps, false, isReset);
+								defList, "toast", validnsps, false,
+								operation == AT_ResetRelOptions);
 
 		(void) heap_reloptions(RELKIND_TOASTVALUE, newOptions, true);
 
diff --git a/src/backend/commands/view.c b/src/backend/commands/view.c
index b238199..3edcd33 100644
--- a/src/backend/commands/view.c
+++ b/src/backend/commands/view.c
@@ -32,6 +32,7 @@
 #include "utils/builtins.h"
 #include "utils/lsyscache.h"
 #include "utils/rel.h"
+#include "utils/syscache.h"
 
 
 static void checkViewTupleDesc(TupleDesc newdesc, TupleDesc olddesc);
@@ -98,7 +99,7 @@ isViewOnTempTable_walker(Node *node, void *context)
  */
 static Oid
 DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
-					  Oid namespaceId)
+					  Oid namespaceId, List *options)
 {
 	Oid			viewOid;
 	CreateStmt *createStmt = makeNode(CreateStmt);
@@ -166,6 +167,8 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
 	{
 		Relation	rel;
 		TupleDesc	descriptor;
+		List	   *atcmds = NIL;
+		AlterTableCmd *atcmd;
 
 		/*
 		 * Yes.  Get exclusive lock on the existing view ...
@@ -204,20 +207,26 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
 		checkViewTupleDesc(descriptor, rel->rd_att);
 
 		/*
+		 * The new options list replaces the existing options list, even
+		 * if it's empty.
+		 */
+		atcmd = makeNode(AlterTableCmd);
+		atcmd->subtype = AT_ReplaceRelOptions;
+		atcmd->def = (Node *) options;
+		atcmds = lappend(atcmds, atcmd);
+
+		/*
 		 * If new attributes have been added, we must add pg_attribute entries
 		 * for them.  It is convenient (although overkill) to use the ALTER
 		 * TABLE ADD COLUMN infrastructure for this.
 		 */
 		if (list_length(attrList) > rel->rd_att->natts)
 		{
-			List	   *atcmds = NIL;
 			ListCell   *c;
 			int			skip = rel->rd_att->natts;
 
 			foreach(c, attrList)
 			{
-				AlterTableCmd *atcmd;
-
 				if (skip > 0)
 				{
 					skip--;
@@ -228,9 +237,11 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
 				atcmd->def = (Node *) lfirst(c);
 				atcmds = lappend(atcmds, atcmd);
 			}
-			AlterTableInternal(viewOid, atcmds, true);
 		}
 
+		/* OK, let's do it. */
+		AlterTableInternal(viewOid, atcmds, true);
+
 		/*
 		 * Seems okay, so return the OID of the pre-existing view.
 		 */
@@ -250,7 +261,8 @@ DefineVirtualRelation(const RangeVar *relation, List *tlist, bool replace,
 		createStmt->tableElts = attrList;
 		createStmt->inhRelations = NIL;
 		createStmt->constraints = NIL;
-		createStmt->options = list_make1(defWithOids(false));
+		createStmt->options = options;
+		createStmt->options = lappend(options, defWithOids(false));
 		createStmt->oncommit = ONCOMMIT_NOOP;
 		createStmt->tablespacename = NULL;
 		createStmt->if_not_exists = false;
@@ -513,7 +525,7 @@ DefineView(ViewStmt *stmt, const char *queryString)
 	 * aborted.
 	 */
 	viewOid = DefineVirtualRelation(view, viewParse->targetList,
-									stmt->replace, namespaceId);
+									stmt->replace, namespaceId, stmt->options);
 
 	/*
 	 * The relation we have just created is not visible to any other commands
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index c70a5bd..dd809ef 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -1965,6 +1965,7 @@ _copyRangeTblEntry(RangeTblEntry *from)
 	COPY_SCALAR_FIELD(relid);
 	COPY_SCALAR_FIELD(relkind);
 	COPY_NODE_FIELD(subquery);
+	COPY_SCALAR_FIELD(security_barrier);
 	COPY_SCALAR_FIELD(jointype);
 	COPY_NODE_FIELD(joinaliasvars);
 	COPY_NODE_FIELD(funcexpr);
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index f490a7a..4fe94de 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2228,6 +2228,7 @@ _equalRangeTblEntry(RangeTblEntry *a, RangeTblEntry *b)
 	COMPARE_SCALAR_FIELD(relid);
 	COMPARE_SCALAR_FIELD(relkind);
 	COMPARE_NODE_FIELD(subquery);
+	COMPARE_SCALAR_FIELD(security_barrier);
 	COMPARE_SCALAR_FIELD(jointype);
 	COMPARE_NODE_FIELD(joinaliasvars);
 	COMPARE_NODE_FIELD(funcexpr);
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 31af47f..85e3219 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2321,6 +2321,7 @@ _outRangeTblEntry(StringInfo str, RangeTblEntry *node)
 			break;
 		case RTE_SUBQUERY:
 			WRITE_NODE_FIELD(subquery);
+			WRITE_BOOL_FIELD(security_barrier);
 			break;
 		case RTE_JOIN:
 			WRITE_ENUM_FIELD(jointype, JoinType);
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 3de20ad..fa9ad16 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -1192,6 +1192,7 @@ _readRangeTblEntry(void)
 			break;
 		case RTE_SUBQUERY:
 			READ_NODE_FIELD(subquery);
+			READ_BOOL_FIELD(security_barrier);
 			break;
 		case RTE_JOIN:
 			READ_ENUM_FIELD(jointype, JoinType);
diff --git a/src/backend/optimizer/path/allpaths.c b/src/backend/optimizer/path/allpaths.c
index 815b996..b3777ea 100644
--- a/src/backend/optimizer/path/allpaths.c
+++ b/src/backend/optimizer/path/allpaths.c
@@ -744,6 +744,11 @@ set_subquery_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	 * pseudoconstant clauses; better to have the gating node above the
 	 * subquery.
 	 *
+	 * Also, if the sub-query has "security_barrier" flag, it means the
+	 * sub-query originated from a view that must enforce row-level security.
+	 * We must not push down quals in order to avoid information leaks, either
+	 * via side-effects or error output.
+	 *
 	 * Non-pushed-down clauses will get evaluated as qpquals of the
 	 * SubqueryScan node.
 	 *
@@ -762,7 +767,16 @@ set_subquery_pathlist(PlannerInfo *root, RelOptInfo *rel,
 			RestrictInfo *rinfo = (RestrictInfo *) lfirst(l);
 			Node	   *clause = (Node *) rinfo->clause;
 
+			/*
+			 * XXX.  You might wonder why we're testing rte->security_barrier
+			 * qual-by-qual here rather than hoisting the test up into the
+			 * surrounding if statement; after all, the answer will be the
+			 * same for all quals.  The answer is that we expect to shortly
+			 * change this logic to allow pushing down some quals that use only
+			 * "leakproof" operators even through a security barrier.
+			 */
 			if (!rinfo->pseudoconstant &&
+				!rte->security_barrier &&
 				qual_is_pushdown_safe(subquery, rti, clause, differentTypes))
 			{
 				/* Push it down */
diff --git a/src/backend/optimizer/prep/prepjointree.c b/src/backend/optimizer/prep/prepjointree.c
index 8bb011b..a711c4f 100644
--- a/src/backend/optimizer/prep/prepjointree.c
+++ b/src/backend/optimizer/prep/prepjointree.c
@@ -543,6 +543,7 @@ pull_up_subqueries(PlannerInfo *root, Node *jtnode,
 		 */
 		if (rte->rtekind == RTE_SUBQUERY &&
 			is_simple_subquery(rte->subquery) &&
+			!rte->security_barrier &&
 			(containing_appendrel == NULL ||
 			 is_safe_append_member(rte->subquery)))
 			return pull_up_simple_subquery(root, jtnode, rte,
@@ -712,6 +713,7 @@ pull_up_simple_subquery(PlannerInfo *root, Node *jtnode, RangeTblEntry *rte,
 	 * pull_up_subqueries.
 	 */
 	if (is_simple_subquery(subquery) &&
+		!rte->security_barrier &&
 		(containing_appendrel == NULL || is_safe_append_member(subquery)))
 	{
 		/* good to go */
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
index de629e9..1d87394 100644
--- a/src/backend/optimizer/util/plancat.c
+++ b/src/backend/optimizer/util/plancat.c
@@ -25,8 +25,10 @@
 #include "catalog/heap.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
+#include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
 #include "optimizer/cost.h"
+#include "optimizer/pathnode.h"
 #include "optimizer/plancat.h"
 #include "optimizer/predtest.h"
 #include "optimizer/prep.h"
@@ -998,6 +1000,76 @@ build_index_tlist(PlannerInfo *root, IndexOptInfo *index,
 }
 
 /*
+ * clause_touch_security_barrier
+ *
+ * It returns true, if the supplied clause touches attribute of relation
+ * with 'security_barrier' being set on its RangeTblEntry.
+ * The purpose of this check is to make sure row-level security performs
+ * correctly, even if user defined its own operator with function that
+ * have a side-effect to leak its arguments.
+ */
+static bool
+clause_touch_security_barrier(Node *clause, PlannerInfo *root)
+{
+	if (!clause)
+		return false;
+
+	if (IsA(clause, Var))
+	{
+		Var			   *var = (Var *) clause;
+		RangeTblEntry  *rte = root->simple_rte_array[var->varno];
+
+		if (rte->rtekind == RTE_RELATION)
+		{
+			/*
+			 * XXX - Extension modules may append qualifiers on WHERE clause
+			 * implicitly for row-level security, and set security_barrier of
+			 * RangeTblEnty. In this case, we should not allow operator
+			 * functions to reference statistics data that may originated from
+			 * the tuples to be invisible.
+			 */
+			return rte->security_barrier;
+		}
+		else if (rte->rtekind == RTE_SUBQUERY)
+		{
+			Query		   *subqry = rte->subquery;
+			RelOptInfo	   *rel;
+			TargetEntry	   *ste;
+
+			/*
+			 * This clause touches an attribute of sub-query originated from
+			 * a view with security_barrier option; that means view is defined
+			 * to row-level security. Thus, user given operator functions
+			 * should not reference statistics data that may originated from
+			 * the tuples to be invisible.
+			 */
+			if (rte->security_barrier)
+				return true;
+
+			if (subqry->setOperations ||
+				subqry->groupClause ||
+				subqry->distinctClause)
+				return false;
+
+			rel = find_base_rel(root, var->varno);
+			Assert(rel->subroot && IsA(rel->subroot, PlannerInfo));
+
+			ste = get_tle_by_resno(subqry->targetList, var->varattno);
+			if (ste == NULL || ste->resjunk)
+				elog(ERROR, "subquery %s does not have attribute %d",
+					 rte->eref->aliasname, var->varattno);
+
+			return clause_touch_security_barrier((Node *)ste->expr,
+												 rel->subroot);
+		}
+		return false;
+	}
+	return expression_tree_walker(clause,
+								  clause_touch_security_barrier,
+								  (void *) root);
+}
+
+/*
  * restriction_selectivity
  *
  * Returns the selectivity of a specified restriction operator clause.
@@ -1022,6 +1094,14 @@ restriction_selectivity(PlannerInfo *root,
 	if (!oprrest)
 		return (Selectivity) 0.5;
 
+	/*
+	 * if the operator depends on values originated from security-barrier
+	 * relation, we skip to call estimator function; to ensure row-level
+	 * security.
+	 */
+	if (clause_touch_security_barrier((Node *) args, root))
+		return (Selectivity) 0.5;
+
 	result = DatumGetFloat8(OidFunctionCall4(oprrest,
 											 PointerGetDatum(root),
 											 ObjectIdGetDatum(operatorid),
@@ -1058,6 +1138,14 @@ join_selectivity(PlannerInfo *root,
 	if (!oprjoin)
 		return (Selectivity) 0.5;
 
+	/*
+	 * if the operator depends on values originated from security-barrier
+	 * relation, we skip to call estimator function; to ensure row-level
+	 * security.
+	 */
+	if (clause_touch_security_barrier((Node *) args, root))
+		return (Selectivity) 0.5;
+
 	result = DatumGetFloat8(OidFunctionCall5(oprjoin,
 											 PointerGetDatum(root),
 											 ObjectIdGetDatum(operatorid),
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 2a497d1..9c792eb 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -7281,26 +7281,28 @@ transaction_mode_list_or_empty:
  *
  *****************************************************************************/
 
-ViewStmt: CREATE OptTemp VIEW qualified_name opt_column_list
+ViewStmt: CREATE OptTemp VIEW qualified_name opt_column_list opt_reloptions
 				AS SelectStmt opt_check_option
 				{
 					ViewStmt *n = makeNode(ViewStmt);
 					n->view = $4;
 					n->view->relpersistence = $2;
 					n->aliases = $5;
-					n->query = $7;
+					n->query = $8;
 					n->replace = false;
+					n->options = $6;
 					$$ = (Node *) n;
 				}
-		| CREATE OR REPLACE OptTemp VIEW qualified_name opt_column_list
+		| CREATE OR REPLACE OptTemp VIEW qualified_name opt_column_list opt_reloptions
 				AS SelectStmt opt_check_option
 				{
 					ViewStmt *n = makeNode(ViewStmt);
 					n->view = $6;
 					n->view->relpersistence = $4;
 					n->aliases = $7;
-					n->query = $9;
+					n->query = $10;
 					n->replace = true;
+					n->options = $8;
 					$$ = (Node *) n;
 				}
 		;
diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c
index a6f4141..8448226 100644
--- a/src/backend/rewrite/rewriteHandler.c
+++ b/src/backend/rewrite/rewriteHandler.c
@@ -1382,6 +1382,7 @@ ApplyRetrieveRule(Query *parsetree,
 
 	rte->rtekind = RTE_SUBQUERY;
 	rte->relid = InvalidOid;
+	rte->security_barrier = RelationIsSecurityView(relation);
 	rte->subquery = rule_action;
 	rte->inh = false;			/* must not be set for a subquery */
 
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 603e4c1..d93e079 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -374,6 +374,7 @@ RelationParseRelOptions(Relation relation, HeapTuple tuple)
 		case RELKIND_RELATION:
 		case RELKIND_TOASTVALUE:
 		case RELKIND_INDEX:
+		case RELKIND_VIEW:
 			break;
 		default:
 			return;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index cbe529e..eea6c06 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -12291,8 +12291,10 @@ dumpTableSchema(Archive *fout, TableInfo *tbinfo)
 		if (binary_upgrade)
 			binary_upgrade_set_pg_class_oids(q, tbinfo->dobj.catId.oid, false);
 
-		appendPQExpBuffer(q, "CREATE VIEW %s AS\n    %s\n",
-						  fmtId(tbinfo->dobj.name), viewdef);
+		appendPQExpBuffer(q, "CREATE VIEW %s", fmtId(tbinfo->dobj.name));
+		if (tbinfo->reloptions && strlen(tbinfo->reloptions) > 0)
+			appendPQExpBuffer(q, " WITH (%s)", tbinfo->reloptions);
+		appendPQExpBuffer(q, " AS\n    %s\n", viewdef);
 
 		appendPQExpBuffer(labelq, "VIEW %s",
 						  fmtId(tbinfo->dobj.name));
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index 14f5034..57595b0 100644
--- a/src/include/access/reloptions.h
+++ b/src/include/access/reloptions.h
@@ -42,8 +42,9 @@ typedef enum relopt_kind
 	RELOPT_KIND_GIST = (1 << 5),
 	RELOPT_KIND_ATTRIBUTE = (1 << 6),
 	RELOPT_KIND_TABLESPACE = (1 << 7),
+	RELOPT_KIND_VIEW = (1 << 8),
 	/* if you add a new kind, make sure you update "last_default" too */
-	RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_TABLESPACE,
+	RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_VIEW,
 	/* some compilers treat enums as signed ints, so we can't use 1 << 31 */
 	RELOPT_KIND_MAX = (1 << 30)
 } relopt_kind;
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 9e277c5..aed2310 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -706,6 +706,7 @@ typedef struct RangeTblEntry
 	 * Fields valid for a subquery RTE (else NULL):
 	 */
 	Query	   *subquery;		/* the sub-query */
+	bool		security_barrier;	/* subquery from security_barrier view */
 
 	/*
 	 * Fields valid for a join RTE (else NULL/zero):
@@ -1208,6 +1209,7 @@ typedef enum AlterTableType
 	AT_SetTableSpace,			/* SET TABLESPACE */
 	AT_SetRelOptions,			/* SET (...) -- AM specific parameters */
 	AT_ResetRelOptions,			/* RESET (...) -- AM specific parameters */
+	AT_ReplaceRelOptions,		/* replace reloption list in its entirety */
 	AT_EnableTrig,				/* ENABLE TRIGGER name */
 	AT_EnableAlwaysTrig,		/* ENABLE ALWAYS TRIGGER name */
 	AT_EnableReplicaTrig,		/* ENABLE REPLICA TRIGGER name */
@@ -2275,6 +2277,7 @@ typedef struct ViewStmt
 	List	   *aliases;		/* target column names */
 	Node	   *query;			/* the SELECT query */
 	bool		replace;		/* replace an existing view? */
+	List	   *options;		/* options from WITH clause */
 } ViewStmt;
 
 /* ----------------------
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 173dc16..dce6d65 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -194,6 +194,7 @@ typedef struct StdRdOptions
 	int32		vl_len_;		/* varlena header (do not touch directly!) */
 	int			fillfactor;		/* page fill factor in percent (0..100) */
 	AutoVacOpts autovacuum;		/* autovacuum-related options */
+	bool		security_barrier;	/* for views */
 } StdRdOptions;
 
 #define HEAP_MIN_FILLFACTOR			10
@@ -222,6 +223,14 @@ typedef struct StdRdOptions
 	(BLCKSZ * (100 - RelationGetFillFactor(relation, defaultff)) / 100)
 
 /*
+ * RelationIsSecurityView
+ *		Returns whether the relation is security view, or not
+ */
+#define RelationIsSecurityView(relation)	\
+	((relation)->rd_options ?				\
+	 ((StdRdOptions *) (relation)->rd_options)->security_barrier : false)
+
+/*
  * RelationIsValid
  *		True iff relation descriptor is valid.
  */
diff --git a/src/test/regress/expected/create_view.out b/src/test/regress/expected/create_view.out
index f9490a3..cc93854 100644
--- a/src/test/regress/expected/create_view.out
+++ b/src/test/regress/expected/create_view.out
@@ -239,6 +239,55 @@ And relnamespace IN (SELECT OID FROM pg_namespace WHERE nspname LIKE 'pg_temp%')
      1
 (1 row)
 
+--
+-- CREATE VIEW and WITH(...) clause
+--
+CREATE VIEW mysecview1
+       AS SELECT * FROM tbl1 WHERE a = 0;
+CREATE VIEW mysecview2 WITH (security_barrier=true)
+       AS SELECT * FROM tbl1 WHERE a > 0;
+CREATE VIEW mysecview3 WITH (security_barrier=false)
+       AS SELECT * FROM tbl1 WHERE a < 0;
+CREATE VIEW mysecview4 WITH (security_barrier)
+       AS SELECT * FROM tbl1 WHERE a <> 0;
+CREATE VIEW mysecview5 WITH (security_barrier=100)	-- Error
+       AS SELECT * FROM tbl1 WHERE a > 100;
+ERROR:  invalid value for boolean option "security_barrier": 100
+CREATE VIEW mysecview6 WITH (invalid_option)		-- Error
+       AS SELECT * FROM tbl1 WHERE a < 100;
+ERROR:  unrecognized parameter "invalid_option"
+SELECT relname, relkind, reloptions FROM pg_class
+       WHERE oid in ('mysecview1'::regclass, 'mysecview2'::regclass,
+                     'mysecview3'::regclass, 'mysecview4'::regclass)
+       ORDER BY relname;
+  relname   | relkind |        reloptions        
+------------+---------+--------------------------
+ mysecview1 | v       | 
+ mysecview2 | v       | {security_barrier=true}
+ mysecview3 | v       | {security_barrier=false}
+ mysecview4 | v       | {security_barrier=true}
+(4 rows)
+
+CREATE OR REPLACE VIEW mysecview1
+       AS SELECT * FROM tbl1 WHERE a = 256;
+CREATE OR REPLACE VIEW mysecview2
+       AS SELECT * FROM tbl1 WHERE a > 256;
+CREATE OR REPLACE VIEW mysecview3 WITH (security_barrier=true)
+       AS SELECT * FROM tbl1 WHERE a < 256;
+CREATE OR REPLACE VIEW mysecview4 WITH (security_barrier=false)
+       AS SELECT * FROM tbl1 WHERE a <> 256;
+SELECT relname, relkind, reloptions FROM pg_class
+       WHERE oid in ('mysecview1'::regclass, 'mysecview2'::regclass,
+                     'mysecview3'::regclass, 'mysecview4'::regclass)
+       ORDER BY relname;
+  relname   | relkind |        reloptions        
+------------+---------+--------------------------
+ mysecview1 | v       | 
+ mysecview2 | v       | 
+ mysecview3 | v       | {security_barrier=true}
+ mysecview4 | v       | {security_barrier=false}
+(4 rows)
+
 DROP SCHEMA temp_view_test CASCADE;
 NOTICE:  drop cascades to 22 other objects
 DETAIL:  drop cascades to table temp_view_test.base_table
@@ -264,7 +313,7 @@ drop cascades to view temp_view_test.v8
 drop cascades to sequence temp_view_test.seq1
 drop cascades to view temp_view_test.v9
 DROP SCHEMA testviewschm2 CASCADE;
-NOTICE:  drop cascades to 16 other objects
+NOTICE:  drop cascades to 20 other objects
 DETAIL:  drop cascades to table t1
 drop cascades to view temporal1
 drop cascades to view temporal2
@@ -281,4 +330,8 @@ drop cascades to table tbl3
 drop cascades to table tbl4
 drop cascades to view mytempview
 drop cascades to view pubview
+drop cascades to view mysecview1
+drop cascades to view mysecview2
+drop cascades to view mysecview3
+drop cascades to view mysecview4
 SET search_path to public;
diff --git a/src/test/regress/expected/select_views_1.out b/src/test/regress/expected/select_views_1.out
index 9a972cf..a005b72 100644
--- a/src/test/regress/expected/select_views_1.out
+++ b/src/test/regress/expected/select_views_1.out
@@ -1247,3 +1247,317 @@ SELECT * FROM toyemp WHERE name = 'sharon';
  sharon |  25 | (15,12)  |     12000
 (1 row)
 
+--
+-- Test for Leaky view scenario
+--
+CREATE USER alice;
+CREATE FUNCTION f_leak (text)
+       RETURNS bool LANGUAGE 'plpgsql' COST 0.0000001
+       AS 'BEGIN RAISE NOTICE ''f_leak => %'', $1; RETURN true; END';
+CREATE TABLE customer (
+       cid      int primary key,
+       name     text not null,
+       tel      text,
+       passwd	text
+);
+NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "customer_pkey" for table "customer"
+CREATE TABLE credit_card (
+       cid      int references customer(cid),
+       cnum     text,
+       climit   int
+);
+CREATE TABLE credit_usage (
+       cid      int references customer(cid),
+       ymd      date,
+       usage    int
+);
+INSERT INTO customer
+       VALUES (101, 'alice', '+81-12-3456-7890', 'passwd123'),
+              (102, 'bob',   '+01-234-567-8901', 'beafsteak'),
+              (103, 'eve',   '+49-8765-43210',   'hamburger');
+INSERT INTO credit_card
+       VALUES (101, '1111-2222-3333-4444', 4000),
+              (102, '5555-6666-7777-8888', 3000),
+              (103, '9801-2345-6789-0123', 2000);
+INSERT INTO credit_usage
+       VALUES (101, '2011-09-15', 120),
+       	      (101, '2011-10-05',  90),
+	      (101, '2011-10-18', 110),
+	      (101, '2011-10-21', 200),
+	      (101, '2011-11-10',  80),
+	      (102, '2011-09-22', 300),
+	      (102, '2011-10-12', 120),
+	      (102, '2011-10-28', 200),
+	      (103, '2011-10-15', 480);
+CREATE VIEW my_property_normal AS
+       SELECT * FROM customer WHERE name = current_user;
+CREATE VIEW my_property_secure WITH (security_barrier) AS
+       SELECT * FROM customer WHERE name = current_user;
+CREATE VIEW my_credit_card_normal AS
+       SELECT * FROM customer l NATURAL JOIN credit_card r
+       WHERE l.name = current_user;
+CREATE VIEW my_credit_card_secure WITH (security_barrier) AS
+       SELECT * FROM customer l NATURAL JOIN credit_card r
+       WHERE l.name = current_user;
+CREATE VIEW my_credit_card_usage_normal AS
+       SELECT * FROM my_credit_card_secure l NATURAL JOIN credit_usage r;
+CREATE VIEW my_credit_card_usage_secure WITH (security_barrier) AS
+       SELECT * FROM my_credit_card_secure l NATURAL JOIN credit_usage r;
+GRANT SELECT ON my_property_normal TO public;
+GRANT SELECT ON my_property_secure TO public;
+GRANT SELECT ON my_credit_card_normal TO public;
+GRANT SELECT ON my_credit_card_secure TO public;
+GRANT SELECT ON my_credit_card_usage_normal TO public;
+GRANT SELECT ON my_credit_card_usage_secure TO public;
+--
+-- Run leaky view scenarios
+--
+SET SESSION AUTHORIZATION alice;
+--
+-- scenario: if a qualifier with tiny-cost is given, it shall be launched
+--           prior to the security policy of the view.
+--
+SELECT * FROM my_property_normal WHERE f_leak(passwd);
+NOTICE:  f_leak => passwd123
+NOTICE:  f_leak => beafsteak
+NOTICE:  f_leak => hamburger
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_property_normal WHERE f_leak(passwd);
+                            QUERY PLAN                            
+------------------------------------------------------------------
+ Seq Scan on customer
+   Filter: (f_leak(passwd) AND (name = ("current_user"())::text))
+(2 rows)
+
+SELECT * FROM my_property_secure WHERE f_leak(passwd);
+NOTICE:  f_leak => passwd123
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_property_secure WHERE f_leak(passwd);
+                    QUERY PLAN                     
+---------------------------------------------------
+ Subquery Scan on my_property_secure
+   Filter: f_leak(my_property_secure.passwd)
+   ->  Seq Scan on customer
+         Filter: (name = ("current_user"())::text)
+(4 rows)
+
+--
+-- scenario: if a qualifier references only one-side of a particular join-
+--           tree, it shall be distributed to the most deep scan plan as
+--           possible as we can.
+--
+SELECT * FROM my_credit_card_normal WHERE f_leak(cnum);
+NOTICE:  f_leak => 1111-2222-3333-4444
+NOTICE:  f_leak => 5555-6666-7777-8888
+NOTICE:  f_leak => 9801-2345-6789-0123
+ cid | name  |       tel        |  passwd   |        cnum         | climit 
+-----+-------+------------------+-----------+---------------------+--------
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000
+(1 row)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_normal WHERE f_leak(cnum);
+                       QUERY PLAN                        
+---------------------------------------------------------
+ Hash Join
+   Hash Cond: (r.cid = l.cid)
+   ->  Seq Scan on credit_card r
+         Filter: f_leak(cnum)
+   ->  Hash
+         ->  Seq Scan on customer l
+               Filter: (name = ("current_user"())::text)
+(7 rows)
+
+SELECT * FROM my_credit_card_secure WHERE f_leak(cnum);
+NOTICE:  f_leak => 1111-2222-3333-4444
+ cid | name  |       tel        |  passwd   |        cnum         | climit 
+-----+-------+------------------+-----------+---------------------+--------
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000
+(1 row)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_secure WHERE f_leak(cnum);
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ Subquery Scan on my_credit_card_secure
+   Filter: f_leak(my_credit_card_secure.cnum)
+   ->  Hash Join
+         Hash Cond: (r.cid = l.cid)
+         ->  Seq Scan on credit_card r
+         ->  Hash
+               ->  Seq Scan on customer l
+                     Filter: (name = ("current_user"())::text)
+(8 rows)
+
+--
+-- scenario: an external qualifier can be pushed-down by in-front-of the
+--           views with "security_barrier" attribute
+--
+SELECT * FROM my_credit_card_usage_normal
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+NOTICE:  f_leak => 1111-2222-3333-4444
+ cid | name  |       tel        |  passwd   |        cnum         | climit |    ymd     | usage 
+-----+-------+------------------+-----------+---------------------+--------+------------+-------
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-05-2011 |    90
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-18-2011 |   110
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-21-2011 |   200
+(3 rows)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_usage_normal
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+                                  QUERY PLAN                                  
+------------------------------------------------------------------------------
+ Nested Loop
+   Join Filter: (l.cid = r.cid)
+   ->  Seq Scan on credit_usage r
+         Filter: ((ymd >= '10-01-2011'::date) AND (ymd < '11-01-2011'::date))
+   ->  Materialize
+         ->  Subquery Scan on l
+               Filter: f_leak(l.cnum)
+               ->  Hash Join
+                     Hash Cond: (r.cid = l.cid)
+                     ->  Seq Scan on credit_card r
+                     ->  Hash
+                           ->  Seq Scan on customer l
+                                 Filter: (name = ("current_user"())::text)
+(13 rows)
+
+SELECT * FROM my_credit_card_usage_secure
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+NOTICE:  f_leak => 1111-2222-3333-4444
+NOTICE:  f_leak => 1111-2222-3333-4444
+NOTICE:  f_leak => 1111-2222-3333-4444
+NOTICE:  f_leak => 1111-2222-3333-4444
+NOTICE:  f_leak => 1111-2222-3333-4444
+ cid | name  |       tel        |  passwd   |        cnum         | climit |    ymd     | usage 
+-----+-------+------------------+-----------+---------------------+--------+------------+-------
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-05-2011 |    90
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-18-2011 |   110
+ 101 | alice | +81-12-3456-7890 | passwd123 | 1111-2222-3333-4444 |   4000 | 10-21-2011 |   200
+(3 rows)
+
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_usage_secure
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+                                                                                 QUERY PLAN                                                                                  
+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------
+ Subquery Scan on my_credit_card_usage_secure
+   Filter: (f_leak(my_credit_card_usage_secure.cnum) AND (my_credit_card_usage_secure.ymd >= '10-01-2011'::date) AND (my_credit_card_usage_secure.ymd < '11-01-2011'::date))
+   ->  Hash Join
+         Hash Cond: (r.cid = l.cid)
+         ->  Seq Scan on credit_usage r
+         ->  Hash
+               ->  Hash Join
+                     Hash Cond: (r.cid = l.cid)
+                     ->  Seq Scan on credit_card r
+                     ->  Hash
+                           ->  Seq Scan on customer l
+                                 Filter: (name = ("current_user"())::text)
+(12 rows)
+
+--
+-- Test for the case when security_barrier gets changed between rewriter
+-- and planner stage.
+--
+PREPARE p1 AS SELECT * FROM my_property_normal WHERE f_leak(passwd);
+PREPARE p2 AS SELECT * FROM my_property_secure WHERE f_leak(passwd);
+EXECUTE p1;
+NOTICE:  f_leak => passwd123
+NOTICE:  f_leak => beafsteak
+NOTICE:  f_leak => hamburger
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+EXECUTE p2;
+NOTICE:  f_leak => passwd123
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+RESET SESSION AUTHORIZATION;
+ALTER VIEW my_property_normal SET (security_barrier=true);
+ALTER VIEW my_property_secure SET (security_barrier=false);
+SET SESSION AUTHORIZATION alice;
+EXECUTE p1;		-- To be perform as a view with security-barrier
+NOTICE:  f_leak => passwd123
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+EXECUTE p2;		-- To be perform as a view without security-barrier
+NOTICE:  f_leak => passwd123
+NOTICE:  f_leak => beafsteak
+NOTICE:  f_leak => hamburger
+ cid | name  |       tel        |  passwd   
+-----+-------+------------------+-----------
+ 101 | alice | +81-12-3456-7890 | passwd123
+(1 row)
+
+--
+-- Estimator function might be leakable
+--
+RESET SESSION AUTHORIZATION;
+CREATE TABLE foo (x int);
+INSERT INTO foo (SELECT generate_series(1,10));
+INSERT INTO foo VALUES (3);
+ANALYZE foo;
+CREATE VIEW foo_normal AS SELECT * FROM foo WHERE x > 5;
+CREATE VIEW foo_secure WITH (security_barrier)
+        AS SELECT * FROM foo WHERE x > 5;
+GRANT SELECT ON foo_normal TO public;
+GRANT SELECT ON foo_secure TO public;
+SET SESSION AUTHORIZATION alice;
+CREATE OR REPLACE FUNCTION op_leak (int, int) RETURNS boolean
+AS 'BEGIN RAISE NOTICE ''op_leak => % : %'', $1, $2; RETURN true; END'
+LANGUAGE plpgsql;
+CREATE OPERATOR !! ( procedure = op_leak,
+       leftarg = integer, rightarg = integer, restrict = eqsel);
+EXPLAIN SELECT * FROM foo_normal WHERE x !! 1;
+NOTICE:  op_leak => 3 : 1
+                    QUERY PLAN                     
+---------------------------------------------------
+ Seq Scan on foo  (cost=0.00..3.89 rows=1 width=4)
+   Filter: ((x > 5) AND (x !! 1))
+(2 rows)
+
+EXPLAIN SELECT * FROM foo_secure WHERE x !! 2;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ Subquery Scan on foo_secure  (cost=0.00..2.70 rows=3 width=4)
+   Filter: (foo_secure.x !! 2)
+   ->  Seq Scan on foo  (cost=0.00..1.14 rows=6 width=4)
+         Filter: (x > 5)
+(4 rows)
+
+--
+-- Clean up
+--
+RESET SESSION AUTHORIZATION;
+DROP FUNCTION f_leak(text) CASCADE;
+DROP TABLE customer CASCADE;
+NOTICE:  drop cascades to 8 other objects
+DETAIL:  drop cascades to constraint credit_card_cid_fkey on table credit_card
+drop cascades to constraint credit_usage_cid_fkey on table credit_usage
+drop cascades to view my_property_normal
+drop cascades to view my_property_secure
+drop cascades to view my_credit_card_normal
+drop cascades to view my_credit_card_secure
+drop cascades to view my_credit_card_usage_normal
+drop cascades to view my_credit_card_usage_secure
+DROP TABLE credit_card CASCADE;
+DROP TABLE credit_usage CASCADE;
+DROP TABLE foo CASCADE;
+NOTICE:  drop cascades to 2 other objects
+DETAIL:  drop cascades to view foo_normal
+drop cascades to view foo_secure
+DROP OWNED BY alice;
+DROP USER alice;
diff --git a/src/test/regress/sql/create_view.sql b/src/test/regress/sql/create_view.sql
index 86cfc51..48d8d22 100644
--- a/src/test/regress/sql/create_view.sql
+++ b/src/test/regress/sql/create_view.sql
@@ -191,6 +191,39 @@ AND NOT EXISTS (SELECT g FROM tbl4 LEFT JOIN tmptbl ON tbl4.h = tmptbl.j);
 SELECT count(*) FROM pg_class where relname LIKE 'mytempview'
 And relnamespace IN (SELECT OID FROM pg_namespace WHERE nspname LIKE 'pg_temp%');
 
+--
+-- CREATE VIEW and WITH(...) clause
+--
+CREATE VIEW mysecview1
+       AS SELECT * FROM tbl1 WHERE a = 0;
+CREATE VIEW mysecview2 WITH (security_barrier=true)
+       AS SELECT * FROM tbl1 WHERE a > 0;
+CREATE VIEW mysecview3 WITH (security_barrier=false)
+       AS SELECT * FROM tbl1 WHERE a < 0;
+CREATE VIEW mysecview4 WITH (security_barrier)
+       AS SELECT * FROM tbl1 WHERE a <> 0;
+CREATE VIEW mysecview5 WITH (security_barrier=100)	-- Error
+       AS SELECT * FROM tbl1 WHERE a > 100;
+CREATE VIEW mysecview6 WITH (invalid_option)		-- Error
+       AS SELECT * FROM tbl1 WHERE a < 100;
+SELECT relname, relkind, reloptions FROM pg_class
+       WHERE oid in ('mysecview1'::regclass, 'mysecview2'::regclass,
+                     'mysecview3'::regclass, 'mysecview4'::regclass)
+       ORDER BY relname;
+
+CREATE OR REPLACE VIEW mysecview1
+       AS SELECT * FROM tbl1 WHERE a = 256;
+CREATE OR REPLACE VIEW mysecview2
+       AS SELECT * FROM tbl1 WHERE a > 256;
+CREATE OR REPLACE VIEW mysecview3 WITH (security_barrier=true)
+       AS SELECT * FROM tbl1 WHERE a < 256;
+CREATE OR REPLACE VIEW mysecview4 WITH (security_barrier=false)
+       AS SELECT * FROM tbl1 WHERE a <> 256;
+SELECT relname, relkind, reloptions FROM pg_class
+       WHERE oid in ('mysecview1'::regclass, 'mysecview2'::regclass,
+                     'mysecview3'::regclass, 'mysecview4'::regclass)
+       ORDER BY relname;
+
 DROP SCHEMA temp_view_test CASCADE;
 DROP SCHEMA testviewschm2 CASCADE;
 
diff --git a/src/test/regress/sql/select_views.sql b/src/test/regress/sql/select_views.sql
index 14f1be8..44ba314 100644
--- a/src/test/regress/sql/select_views.sql
+++ b/src/test/regress/sql/select_views.sql
@@ -8,3 +8,165 @@ SELECT * FROM street;
 SELECT name, #thepath FROM iexit ORDER BY 1, 2;
 
 SELECT * FROM toyemp WHERE name = 'sharon';
+
+--
+-- Test for Leaky view scenario
+--
+CREATE USER alice;
+
+CREATE FUNCTION f_leak (text)
+       RETURNS bool LANGUAGE 'plpgsql' COST 0.0000001
+       AS 'BEGIN RAISE NOTICE ''f_leak => %'', $1; RETURN true; END';
+
+CREATE TABLE customer (
+       cid      int primary key,
+       name     text not null,
+       tel      text,
+       passwd	text
+);
+
+CREATE TABLE credit_card (
+       cid      int references customer(cid),
+       cnum     text,
+       climit   int
+);
+
+CREATE TABLE credit_usage (
+       cid      int references customer(cid),
+       ymd      date,
+       usage    int
+);
+
+INSERT INTO customer
+       VALUES (101, 'alice', '+81-12-3456-7890', 'passwd123'),
+              (102, 'bob',   '+01-234-567-8901', 'beafsteak'),
+              (103, 'eve',   '+49-8765-43210',   'hamburger');
+INSERT INTO credit_card
+       VALUES (101, '1111-2222-3333-4444', 4000),
+              (102, '5555-6666-7777-8888', 3000),
+              (103, '9801-2345-6789-0123', 2000);
+INSERT INTO credit_usage
+       VALUES (101, '2011-09-15', 120),
+       	      (101, '2011-10-05',  90),
+	      (101, '2011-10-18', 110),
+	      (101, '2011-10-21', 200),
+	      (101, '2011-11-10',  80),
+	      (102, '2011-09-22', 300),
+	      (102, '2011-10-12', 120),
+	      (102, '2011-10-28', 200),
+	      (103, '2011-10-15', 480);
+
+CREATE VIEW my_property_normal AS
+       SELECT * FROM customer WHERE name = current_user;
+CREATE VIEW my_property_secure WITH (security_barrier) AS
+       SELECT * FROM customer WHERE name = current_user;
+
+CREATE VIEW my_credit_card_normal AS
+       SELECT * FROM customer l NATURAL JOIN credit_card r
+       WHERE l.name = current_user;
+CREATE VIEW my_credit_card_secure WITH (security_barrier) AS
+       SELECT * FROM customer l NATURAL JOIN credit_card r
+       WHERE l.name = current_user;
+
+CREATE VIEW my_credit_card_usage_normal AS
+       SELECT * FROM my_credit_card_secure l NATURAL JOIN credit_usage r;
+CREATE VIEW my_credit_card_usage_secure WITH (security_barrier) AS
+       SELECT * FROM my_credit_card_secure l NATURAL JOIN credit_usage r;
+
+GRANT SELECT ON my_property_normal TO public;
+GRANT SELECT ON my_property_secure TO public;
+GRANT SELECT ON my_credit_card_normal TO public;
+GRANT SELECT ON my_credit_card_secure TO public;
+GRANT SELECT ON my_credit_card_usage_normal TO public;
+GRANT SELECT ON my_credit_card_usage_secure TO public;
+
+--
+-- Run leaky view scenarios
+--
+SET SESSION AUTHORIZATION alice;
+
+--
+-- scenario: if a qualifier with tiny-cost is given, it shall be launched
+--           prior to the security policy of the view.
+--
+SELECT * FROM my_property_normal WHERE f_leak(passwd);
+EXPLAIN (COSTS OFF) SELECT * FROM my_property_normal WHERE f_leak(passwd);
+
+SELECT * FROM my_property_secure WHERE f_leak(passwd);
+EXPLAIN (COSTS OFF) SELECT * FROM my_property_secure WHERE f_leak(passwd);
+
+--
+-- scenario: if a qualifier references only one-side of a particular join-
+--           tree, it shall be distributed to the most deep scan plan as
+--           possible as we can.
+--
+SELECT * FROM my_credit_card_normal WHERE f_leak(cnum);
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_normal WHERE f_leak(cnum);
+
+SELECT * FROM my_credit_card_secure WHERE f_leak(cnum);
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_secure WHERE f_leak(cnum);
+
+--
+-- scenario: an external qualifier can be pushed-down by in-front-of the
+--           views with "security_barrier" attribute
+--
+SELECT * FROM my_credit_card_usage_normal
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_usage_normal
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+
+SELECT * FROM my_credit_card_usage_secure
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+EXPLAIN (COSTS OFF) SELECT * FROM my_credit_card_usage_secure
+       WHERE f_leak(cnum) AND ymd >= '2011-10-01' AND ymd < '2011-11-01';
+
+--
+-- Test for the case when security_barrier gets changed between rewriter
+-- and planner stage.
+--
+PREPARE p1 AS SELECT * FROM my_property_normal WHERE f_leak(passwd);
+PREPARE p2 AS SELECT * FROM my_property_secure WHERE f_leak(passwd);
+EXECUTE p1;
+EXECUTE p2;
+RESET SESSION AUTHORIZATION;
+ALTER VIEW my_property_normal SET (security_barrier=true);
+ALTER VIEW my_property_secure SET (security_barrier=false);
+SET SESSION AUTHORIZATION alice;
+EXECUTE p1;		-- To be perform as a view with security-barrier
+EXECUTE p2;		-- To be perform as a view without security-barrier
+
+--
+-- Estimator function might be leakable
+--
+RESET SESSION AUTHORIZATION;
+CREATE TABLE foo (x int);
+INSERT INTO foo (SELECT generate_series(1,10));
+INSERT INTO foo VALUES (3);
+ANALYZE foo;
+
+CREATE VIEW foo_normal AS SELECT * FROM foo WHERE x > 5;
+CREATE VIEW foo_secure WITH (security_barrier)
+        AS SELECT * FROM foo WHERE x > 5;
+GRANT SELECT ON foo_normal TO public;
+GRANT SELECT ON foo_secure TO public;
+
+SET SESSION AUTHORIZATION alice;
+CREATE OR REPLACE FUNCTION op_leak (int, int) RETURNS boolean
+AS 'BEGIN RAISE NOTICE ''op_leak => % : %'', $1, $2; RETURN true; END'
+LANGUAGE plpgsql;
+CREATE OPERATOR !! ( procedure = op_leak,
+       leftarg = integer, rightarg = integer, restrict = eqsel);
+EXPLAIN SELECT * FROM foo_normal WHERE x !! 1;
+EXPLAIN SELECT * FROM foo_secure WHERE x !! 2;
+
+--
+-- Clean up
+--
+RESET SESSION AUTHORIZATION;
+DROP FUNCTION f_leak(text) CASCADE;
+DROP TABLE customer CASCADE;
+DROP TABLE credit_card CASCADE;
+DROP TABLE credit_usage CASCADE;
+DROP TABLE foo CASCADE;
+DROP OWNED BY alice;
+DROP USER alice;