Column-level-denies DONE

11 files changed, 824 insertions, 127 deletions

diff --git a/mysql-test/suite/deny/columns.result b/mysql-test/suite/deny/columns.result
new file mode 100644
index 00000000000..fd97c27f161
--- /dev/null
+++ b/mysql-test/suite/deny/columns.result
@@ -0,0 +1,125 @@
+create user foo;
+create database deny_db;
+create table deny_db.t1 (a int, b int, secret int);
+create table deny_db.t2 (a2 int, b2 int, secret2 int);
+insert into deny_db.t2 values (100, 200, 300);
+grant all on *.* to foo;
+grant all on deny_db.* to foo;
+grant all on deny_db.t1 to foo;
+grant all on deny_db.t2 to foo;
+grant select (a, b, secret) on deny_db.t1 to foo;
+grant insert (a, b, secret) on deny_db.t1 to foo;
+grant update (a, b, secret) on deny_db.t1 to foo;
+grant references (a, b, secret) on deny_db.t1 to foo;
+grant select (a2, b2, secret2) on deny_db.t2 to foo;
+grant insert (a2, b2, secret2) on deny_db.t2 to foo;
+grant update (a2, b2, secret2) on deny_db.t2 to foo;
+grant references (a2, b2, secret2) on deny_db.t2 to foo;
+#
+# Test select and insert ... select commands.
+#
+deny select (secret) on deny_db.t1 to foo;
+connect  con1,localhost,foo,,;
+select * from deny_db.t1;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+select secret from deny_db.t1;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+#
+# These are different code paths. Check both.
+#
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+insert into deny_db.t1 values (1, 2, 3);
+#
+# Check INSERT SELECT as we need to account for 2 different kinds of
+# privileges.
+#
+insert into deny_db.t1 select * from deny_db.t1;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+insert into deny_db.t1 select * from deny_db.t2;
+show full columns from deny_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a	int(11)	NULL	YES		NULL		select,insert,update,references	
+b	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret	int(11)	NULL	YES		NULL		insert,update,references	
+show full columns from deny_db.t2;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a2	int(11)	NULL	YES		NULL		select,insert,update,references	
+b2	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret2	int(11)	NULL	YES		NULL		select,insert,update,references	
+disconnect con1;
+connection default;
+revoke deny select (secret) on deny_db.t1 from foo;
+deny select(secret2) on deny_db.t2 to foo;
+connect  con1,localhost,foo,,;
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+insert into deny_db.t1 values (1, 2, 3);
+insert into deny_db.t1 select * from deny_db.t1;
+insert into deny_db.t1 select * from deny_db.t2;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret2' in table 't2'
+show full columns from deny_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a	int(11)	NULL	YES		NULL		select,insert,update,references	
+b	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret	int(11)	NULL	YES		NULL		select,insert,update,references	
+show full columns from deny_db.t2;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a2	int(11)	NULL	YES		NULL		select,insert,update,references	
+b2	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret2	int(11)	NULL	YES		NULL		insert,update,references	
+disconnect con1;
+#
+# Test insert commands
+#
+connection default;
+deny insert (secret) on deny_db.t1 to foo;
+connect  con1,localhost,foo,,;
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+ERROR 42000: INSERT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+insert into deny_db.t1 values (1, 2, 3);
+ERROR 42000: INSERT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+insert into deny_db.t1 select * from deny_db.t1;
+ERROR 42000: INSERT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+insert into deny_db.t1 select * from deny_db.t2;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret2' in table 't2'
+show full columns from deny_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a	int(11)	NULL	YES		NULL		select,insert,update,references	
+b	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret	int(11)	NULL	YES		NULL		select,update,references	
+disconnect con1;
+#
+# Test update commands.
+#
+connection default;
+deny select (secret) on deny_db.t1 to foo;
+deny select (secret2) on deny_db.t2 to foo;
+connect  con1, localhost,foo,,;
+update deny_db.t1 set a=111, b=222 where secret=10;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+update deny_db.t1 set a=123 where a = (select secret2 from deny_db.t2);
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for column 'secret2' in table 't2'
+show full columns from deny_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a	int(11)	NULL	YES		NULL		select,insert,update,references	
+b	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret	int(11)	NULL	YES		NULL		update,references	
+show full columns from deny_db.t2;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a2	int(11)	NULL	YES		NULL		select,insert,update,references	
+b2	int(11)	NULL	YES		NULL		select,insert,update,references	
+secret2	int(11)	NULL	YES		NULL		insert,update,references	
+disconnect con1;
+connection default;
+deny update (secret) on deny_db.t1 to foo;
+connect  con1, localhost,foo,,;
+update deny_db.t1 set a=111, b=222;
+update deny_db.t1 set secret=333, a=111;
+ERROR 42000: UPDATE command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+update deny_db.t1 set secret=333;
+ERROR 42000: UPDATE command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+update deny_db.t1 set a=123, secret=333;
+ERROR 42000: UPDATE command denied to user 'foo'@'localhost' for column 'secret' in table 't1'
+disconnect con1;
+connection default;
+drop user foo;
+drop database deny_db;
diff --git a/mysql-test/suite/deny/columns.test b/mysql-test/suite/deny/columns.test
new file mode 100644
index 00000000000..889c33ba2b0
--- /dev/null
+++ b/mysql-test/suite/deny/columns.test
@@ -0,0 +1,120 @@
+--source include/not_embedded.inc
+
+create user foo;
+create database deny_db;
+create table deny_db.t1 (a int, b int, secret int);
+create table deny_db.t2 (a2 int, b2 int, secret2 int);
+insert into deny_db.t2 values (100, 200, 300);
+
+grant all on *.* to foo;
+grant all on deny_db.* to foo;
+grant all on deny_db.t1 to foo;
+grant all on deny_db.t2 to foo;
+grant select (a, b, secret) on deny_db.t1 to foo;
+grant insert (a, b, secret) on deny_db.t1 to foo;
+grant update (a, b, secret) on deny_db.t1 to foo;
+grant references (a, b, secret) on deny_db.t1 to foo;
+
+grant select (a2, b2, secret2) on deny_db.t2 to foo;
+grant insert (a2, b2, secret2) on deny_db.t2 to foo;
+grant update (a2, b2, secret2) on deny_db.t2 to foo;
+grant references (a2, b2, secret2) on deny_db.t2 to foo;
+
+--echo #
+--echo # Test select and insert ... select commands.
+--echo #
+deny select (secret) on deny_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+
+--error ER_COLUMNACCESS_DENIED_ERROR
+select * from deny_db.t1;
+--error ER_COLUMNACCESS_DENIED_ERROR
+select secret from deny_db.t1;
+
+--echo #
+--echo # These are different code paths. Check both.
+--echo #
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+insert into deny_db.t1 values (1, 2, 3);
+--echo #
+--echo # Check INSERT SELECT as we need to account for 2 different kinds of
+--echo # privileges.
+--echo #
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 select * from deny_db.t1;
+insert into deny_db.t1 select * from deny_db.t2;
+
+show full columns from deny_db.t1;
+show full columns from deny_db.t2;
+disconnect con1;
+
+connection default;
+revoke deny select (secret) on deny_db.t1 from foo;
+deny select(secret2) on deny_db.t2 to foo;
+
+--connect (con1,localhost,foo,,)
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+insert into deny_db.t1 values (1, 2, 3);
+insert into deny_db.t1 select * from deny_db.t1;
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 select * from deny_db.t2;
+show full columns from deny_db.t1;
+show full columns from deny_db.t2;
+disconnect con1;
+
+--echo #
+--echo # Test insert commands
+--echo #
+connection default;
+deny insert (secret) on deny_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 (a, b, secret) values (1, 2, 3);
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 values (1, 2, 3);
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 select * from deny_db.t1;
+--error ER_COLUMNACCESS_DENIED_ERROR
+insert into deny_db.t1 select * from deny_db.t2;
+show full columns from deny_db.t1;
+disconnect con1;
+
+--echo #
+--echo # Test update commands.
+--echo #
+
+connection default;
+deny select (secret) on deny_db.t1 to foo;
+deny select (secret2) on deny_db.t2 to foo;
+
+--connect (con1, localhost,foo,,)
+--error ER_COLUMNACCESS_DENIED_ERROR
+update deny_db.t1 set a=111, b=222 where secret=10;
+--error ER_COLUMNACCESS_DENIED_ERROR
+update deny_db.t1 set a=123 where a = (select secret2 from deny_db.t2);
+
+show full columns from deny_db.t1;
+show full columns from deny_db.t2;
+disconnect con1;
+
+connection default;
+deny update (secret) on deny_db.t1 to foo;
+
+--connect (con1, localhost,foo,,)
+update deny_db.t1 set a=111, b=222;
+--error ER_COLUMNACCESS_DENIED_ERROR
+update deny_db.t1 set secret=333, a=111;
+--error ER_COLUMNACCESS_DENIED_ERROR
+update deny_db.t1 set secret=333;
+--error ER_COLUMNACCESS_DENIED_ERROR
+update deny_db.t1 set a=123, secret=333;
+
+disconnect con1;
+
+connection default;
+drop user foo;
+drop database deny_db;
+
diff --git a/mysql-test/suite/deny/deny_datastructures.result b/mysql-test/suite/deny/deny_datastructures.result
index 31453df03dd..8333d893d77 100644
--- a/mysql-test/suite/deny/deny_datastructures.result
+++ b/mysql-test/suite/deny/deny_datastructures.result
@@ -26,9 +26,7 @@ deny select on some_other_db.some_table to foo;
 deny insert on some_other_db.some_table_second to foo;
 deny update on some_other_db.some_table_third to foo;
 deny select(a, b) on some_other_db.some_table_third to foo;
-ERROR HY000: Column level deny is not supported yet.
 deny select, insert(a, b, c) on some_other_db.some_table_fourth to foo;
-ERROR HY000: Column level deny is not supported yet.
 select user, host, JSON_DETAILED(JSON_EXTRACT(priv, '$.deny')) from mysql.global_priv where user = 'foo';
 user	host	JSON_DETAILED(JSON_EXTRACT(priv, '$.deny'))
 foo	%	{
@@ -50,8 +48,8 @@ foo	%	{
     [
         
         {
-            "name": "`some_other_db`.`some_table_third`",
-            "access": 4
+            "name": "`some_other_db`.`some_table_fourth`",
+            "access": 1
         },
         
         {
@@ -62,6 +60,39 @@ foo	%	{
         {
             "name": "`some_other_db`.`some_table`",
             "access": 1
+        },
+        
+        {
+            "name": "`some_other_db`.`some_table_third`",
+            "access": 4
+        }
+    ],
+    "column": 
+    [
+        
+        {
+            "name": "`some_other_db`.`some_table_fourth`.`b`",
+            "access": 2
+        },
+        
+        {
+            "name": "`some_other_db`.`some_table_fourth`.`c`",
+            "access": 2
+        },
+        
+        {
+            "name": "`some_other_db`.`some_table_third`.`b`",
+            "access": 1
+        },
+        
+        {
+            "name": "`some_other_db`.`some_table_fourth`.`a`",
+            "access": 2
+        },
+        
+        {
+            "name": "`some_other_db`.`some_table_third`.`a`",
+            "access": 1
         }
     ],
     "version_id": VERSION_ID
diff --git a/mysql-test/suite/deny/deny_datastructures.test b/mysql-test/suite/deny/deny_datastructures.test
index 88bbd237498..3667e69bf5d 100644
--- a/mysql-test/suite/deny/deny_datastructures.test
+++ b/mysql-test/suite/deny/deny_datastructures.test
@@ -29,9 +29,7 @@ deny insert on some_other_db.* to bar;
 deny select on some_other_db.some_table to foo;
 deny insert on some_other_db.some_table_second to foo;
 deny update on some_other_db.some_table_third to foo;
---error ER_UNSUPPORTED_DENY
 deny select(a, b) on some_other_db.some_table_third to foo;
---error ER_UNSUPPORTED_DENY
 deny select, insert(a, b, c) on some_other_db.some_table_fourth to foo;
 
 --replace_regex $REGEX_VERSION_ID
diff --git a/mysql-test/suite/deny/show_databases.result b/mysql-test/suite/deny/show_databases.result
index 6b16a9894d2..c8af4d51a1f 100644
--- a/mysql-test/suite/deny/show_databases.result
+++ b/mysql-test/suite/deny/show_databases.result
@@ -645,3 +645,74 @@ disconnect con1;
 connection default;
 drop user foo;
 drop database some_db;
+#
+# Test column level denies masking column level grants.
+#
+connection default;
+create user foo;
+create database some_db;
+create table some_db.t1 (a int, b int);
+grant insert(a) on some_db.t1 to foo;
+deny select(a) on some_db.t1 to foo;
+connect  con1,localhost,foo,,;
+show databases;
+Database
+information_schema
+some_db
+test
+use some_db;
+show tables from some_db;
+Tables_in_some_db
+t1
+show full columns from some_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+a	int(11)	NULL	YES		NULL		insert	
+disconnect con1;
+connection default;
+deny insert(a) on some_db.t1 to foo;
+connect  con1,localhost,foo,,;
+show databases;
+Database
+information_schema
+test
+use some_db;
+ERROR 42000: Access denied for user 'foo'@'%' to database 'some_db'
+show tables from some_db;
+ERROR 42000: Access denied for user 'foo'@'%' to database 'some_db'
+show full columns from some_db.t1;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for table 't1'
+disconnect con1;
+connection default;
+grant select(b) on some_db.t1 to foo;
+connect  con1,localhost,foo,,;
+show databases;
+Database
+information_schema
+some_db
+test
+use some_db;
+show tables from some_db;
+Tables_in_some_db
+t1
+show full columns from some_db.t1;
+Field	Type	Collation	Null	Key	Default	Extra	Privileges	Comment
+b	int(11)	NULL	YES		NULL		select	
+disconnect con1;
+connection default;
+connection default;
+deny select(b) on some_db.t1 to foo;
+connect  con1,localhost,foo,,;
+show databases;
+Database
+information_schema
+test
+use some_db;
+ERROR 42000: Access denied for user 'foo'@'%' to database 'some_db'
+show tables from some_db;
+ERROR 42000: Access denied for user 'foo'@'%' to database 'some_db'
+show full columns from some_db.t1;
+ERROR 42000: SELECT command denied to user 'foo'@'localhost' for table 't1'
+disconnect con1;
+connection default;
+drop user foo;
+drop database some_db;
diff --git a/mysql-test/suite/deny/show_databases.test b/mysql-test/suite/deny/show_databases.test
index 1254a863fcd..e1878eeffee 100644
--- a/mysql-test/suite/deny/show_databases.test
+++ b/mysql-test/suite/deny/show_databases.test
@@ -531,3 +531,63 @@ disconnect con1;
 connection default;
 drop user foo;
 drop database some_db;
+
+--echo #
+--echo # Test column level denies masking column level grants.
+--echo #
+connection default;
+create user foo;
+create database some_db;
+create table some_db.t1 (a int, b int);
+grant insert(a) on some_db.t1 to foo;
+
+deny select(a) on some_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+show databases;
+use some_db;
+show tables from some_db;
+show full columns from some_db.t1;
+disconnect con1;
+connection default;
+
+deny insert(a) on some_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+show databases;
+--error ER_DBACCESS_DENIED_ERROR
+use some_db;
+--error ER_DBACCESS_DENIED_ERROR
+show tables from some_db;
+--error ER_TABLEACCESS_DENIED_ERROR
+show full columns from some_db.t1;
+disconnect con1;
+connection default;
+
+grant select(b) on some_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+show databases;
+use some_db;
+show tables from some_db;
+show full columns from some_db.t1;
+disconnect con1;
+connection default;
+
+connection default;
+
+deny select(b) on some_db.t1 to foo;
+
+--connect (con1,localhost,foo,,)
+show databases;
+--error ER_DBACCESS_DENIED_ERROR
+use some_db;
+--error ER_DBACCESS_DENIED_ERROR
+show tables from some_db;
+--error ER_TABLEACCESS_DENIED_ERROR
+show full columns from some_db.t1;
+disconnect con1;
+
+connection default;
+drop user foo;
+drop database some_db;
diff --git a/mysql-test/suite/deny/show_view.result b/mysql-test/suite/deny/show_view.result
index f4725089716..50a65773c56 100644
--- a/mysql-test/suite/deny/show_view.result
+++ b/mysql-test/suite/deny/show_view.result
@@ -129,7 +129,7 @@ disconnect con1;
 connection default;
 select user, host, JSON_EXTRACT(priv, '$.deny') from mysql.global_priv where user = 'foo';
 user	host	JSON_EXTRACT(priv, '$.deny')
-foo	%	{"global": 0, "version_id": VERSION_ID}
+foo	%	{"version_id": VERSION_ID}
 deny select on some_db.* to foo;
 connect  con1,localhost,foo,,;
 select * from information_schema.views
diff --git a/sql/sql_acl.cc b/sql/sql_acl.cc
index 7dd94206d1e..88ca05d318e 100644
--- a/sql/sql_acl.cc
+++ b/sql/sql_acl.cc
@@ -136,6 +136,12 @@ static bool compare_hostname(const acl_host_and_ip *, const char *, const char *
 /*
   This class represents a set of privilege bits and how they should be applied
   to an ACL_xxx entity.
+
+  TODO(cvicentiu) There is quite a bit of overlap between Grant_privilege
+  and Priv_spec, with Priv_spec covering more.
+  The class exists now so as to not cause additionaly refactoring, however
+  it should be merged with class Grant_privilege and make sure it
+  provides a consistent API that all grant handling functions can use.
 */
 class Priv_spec
 {
@@ -146,33 +152,49 @@ public:
   PRIV_TYPE spec_type;
   const LEX_CSTRING db;
   const LEX_CSTRING table;
+  const List<LEX_COLUMN> *columns;
 
   /* TODO(cvicentiu) clean up initializer list */
   Priv_spec(privilege_t access, bool revoke, bool deny=false):
     revoke{revoke}, deny{deny}, access{access},
     spec_type{PRIV_TYPE::GLOBAL_PRIV},
-    db({.str=nullptr, .length=0}), table({.str=nullptr, .length=0})
+    db({.str=nullptr, .length=0}), table({.str=nullptr, .length=0}),
+    columns{nullptr}
   {}
 
   Priv_spec(privilege_t access, bool revoke, const LEX_CSTRING &db, bool deny=false):
     revoke{revoke}, deny{deny}, access{access},
     spec_type{PRIV_TYPE::DATABASE_PRIV},
     db({db.str, db.length}),
-    table({.str=nullptr, .length=0})
-  { /* TODO(cvicentiu) hack */
+    table({.str=nullptr, .length=0}),
+    columns{nullptr}
+  {
+    /* TODO(cvicentiu) hack */
      if (!db.str)
+     {
        spec_type= PRIV_TYPE::GLOBAL_PRIV;
+     }
   }
 
-  /* TODO(cvicentiu) We'll need to do something with COLUMNS at some point. */
   Priv_spec(privilege_t access, bool revoke, const LEX_CSTRING &db,
-            const LEX_CSTRING &table, bool deny=false):
+            const LEX_CSTRING &routine, PRIV_TYPE routine_type,
+            bool deny=false):
+    revoke{revoke}, deny{deny}, access{access},
+    spec_type{routine_type},
+    db({db.str, db.length}),
+    table({routine.str, routine.length}),
+    columns{nullptr}
+  {}
+
+  Priv_spec(privilege_t access, bool revoke, const LEX_CSTRING &db,
+            const LEX_CSTRING &table, const List<LEX_COLUMN> *columns,
+            bool deny=false):
     revoke{revoke}, deny{deny}, access{access},
     spec_type{PRIV_TYPE::TABLE_PRIV},
     db({db.str, db.length}),
-    table({table.str, table.length})
-  {
-  }
+    table({table.str, table.length}),
+    columns{columns}
+  {}
 };
 
 
@@ -183,6 +205,7 @@ class Deny_spec
   privilege_t global_denies;
   Hash_set<deny_entry> *db_denies;
   Hash_set<deny_entry> *table_denies;
+  Hash_set<deny_entry> *column_denies;
   Hash_set<deny_entry> *procedure_denies;
   Hash_set<deny_entry> *function_denies;
   Hash_set<deny_entry> *package_spec_denies;
@@ -234,14 +257,15 @@ class Deny_spec
   // TODO(cvicentiu) proxy denies.
 public:
   Deny_spec() : specified_denies{NO_PRIV}, global_denies{NO_ACL},
-    db_denies{nullptr}, table_denies{nullptr}, procedure_denies{nullptr},
-    function_denies{nullptr}, package_spec_denies{nullptr},
-    package_body_denies{nullptr} {}
+    db_denies{nullptr}, table_denies{nullptr}, column_denies{nullptr},
+    procedure_denies{nullptr}, function_denies{nullptr},
+    package_spec_denies{nullptr}, package_body_denies{nullptr} {}
 
   ~Deny_spec()
   {
     delete db_denies;
     delete table_denies;
+    delete column_denies;
     delete procedure_denies;
     delete function_denies;
     delete package_spec_denies;
@@ -262,6 +286,13 @@ public:
     return get_hash_value(table_denies, db, table);
   }
 
+  privilege_t get_column_deny(const LEX_CSTRING &db,
+                              const LEX_CSTRING &table,
+                              const LEX_CSTRING &column) const
+  {
+    return get_hash_value(column_denies, db, table, column);
+  }
+
   privilege_t get_procedure_deny(const LEX_CSTRING &db,
                                  const LEX_CSTRING &procedure) const
   {
@@ -287,13 +318,14 @@ public:
 
   static privilege_t get_hash_value(const Hash_set<deny_entry> *hash,
                                     const LEX_CSTRING &db,
-                                    const LEX_CSTRING &object={nullptr, 0})
+                                    const LEX_CSTRING &object={nullptr, 0},
+                                    const LEX_CSTRING &column={nullptr, 0})
   {
     deny_entry *res= nullptr;
     if (hash && db.length)
     {
       String key;
-      construct_identifier(&key, db, object);
+      construct_identifier(&key, db, object, column);
       res= hash->find(key.c_ptr(), key.length());
     }
     if (!res)
@@ -385,6 +417,17 @@ public:
       append_comma= true;
     }
 
+    if (specified_denies & COLUMN_PRIV)
+    {
+      if (append_comma)
+        s->append(',');
+      s->append(STRING_WITH_LEN("\"column\":["));
+      if (print_hash_table(s, column_denies))
+        return true;
+      s->append(']');
+      append_comma= true;
+    }
+
     if (specified_denies & FUNCTION_PRIV)
     {
       if (append_comma)
@@ -439,10 +482,14 @@ public:
     return false;
   }
 
-  static bool update_hash(const Priv_spec &priv_spec,
-                          Hash_set<deny_entry> **hash)
+  static bool update_hash(Hash_set<deny_entry> **hash,
+                          privilege_t access, bool revoke,
+                          const LEX_CSTRING &db, const LEX_CSTRING &table,
+                          const LEX_CSTRING &column={nullptr, 0})
   {
     deny_entry *entry= nullptr;
+    String key;
+    construct_identifier(&key, db, table, column);
     if (!*hash)
     {
       *hash= new Hash_set<deny_entry>(
@@ -454,9 +501,6 @@ public:
     }
     else
     {
-      String key;
-      construct_identifier(&key,
-                           priv_spec.db, priv_spec.table);
       entry= get_hash_entry(**hash, key);
     }
     bool entry_exists= entry != nullptr;
@@ -464,15 +508,12 @@ public:
     {
       /* Shortcut, revoking a non-existing entry does nothing to the
          hash table. */
-      if (priv_spec.revoke)
+      if (revoke)
         return false;
 
       entry= new deny_entry();
       if (!entry)
         return true;
-      String key;
-      construct_identifier(&key,
-                           priv_spec.db, priv_spec.table);
       entry->first.str= my_strdup(PSI_INSTRUMENT_MEM, key.c_ptr(), MYF(0));
       if (!entry->first.str)
       {
@@ -485,10 +526,10 @@ public:
       entry->first.length= key.length();
     }
 
-    if (priv_spec.revoke)
-      entry->second&= ~priv_spec.access;
+    if (revoke)
+      entry->second&= ~access;
     else
-      entry->second|= priv_spec.access;
+      entry->second|= access;
 
     /* A no privilege entry should be removed from the hash. */
     if (entry_exists && !entry->second)
@@ -510,7 +551,6 @@ public:
 
   bool update_deny(const Priv_spec &priv_spec)
   {
-
     Hash_set<deny_entry> **hash= nullptr;
 
     /* Global */
@@ -521,16 +561,52 @@ public:
         global_denies&= ~priv_spec.access;
       else
         global_denies|= priv_spec.access;
+
+      if (global_denies)
+        specified_denies|= GLOBAL_PRIV;
+      else
+      {
+        DBUG_ASSERT(priv_spec.revoke);
+        /* Clear this bit from the specified denies flag to speed up lookups. */
+        specified_denies&= ~GLOBAL_PRIV;
+      }
       break;
     }
     case DATABASE_PRIV:
       hash= &db_denies;
       break;
     case TABLE_PRIV:
-      hash= &table_denies;
-      break;
+    {
+      if (!priv_spec.access)
+        hash= nullptr;
+      else
+        hash= &table_denies;
+      /*
+         Special handling for COLUMNS_PRIV as they come together with TABLE_PRIV
+         potentially.
+      */
+      DBUG_ASSERT(priv_spec.columns);
+      if (priv_spec.columns->elements)
+      {
+        List_iterator<LEX_COLUMN> it{*const_cast<List<LEX_COLUMN>*>(
+                                                    priv_spec.columns)};
+        LEX_COLUMN *column;
+        while ((column= it++))
+        {
+          if (update_hash(&column_denies, column->rights, priv_spec.revoke,
+                          priv_spec.db, priv_spec.table,
+                          column->column.to_lex_cstring()))
+            return true;
+        }
+        if (!column_denies || column_denies->is_empty())
+          specified_denies&= ~COLUMN_PRIV;
+        else
+          specified_denies|= COLUMN_PRIV;
+      }
+    }
+    break;
     case COLUMN_PRIV:
-      // TODO(cvicentiu)
+      /* Handled as part of TABLE_PRIV, never passed directly. */
       DBUG_ASSERT(0);
       break;
     case PROCEDURE_PRIV:
@@ -554,16 +630,22 @@ public:
       DBUG_ASSERT(0);
       return false;
     }
-    if (hash && update_hash(priv_spec, hash))
+    if (!hash)
+      return false;
+
+    if (update_hash(hash, priv_spec.access, priv_spec.revoke,
+                    priv_spec.db, priv_spec.table))
       return true;
-    if (!hash /* Global spec */ || !(*hash)->is_empty())
+
+    if (!(*hash)->is_empty())
       specified_denies|= priv_spec.spec_type;
-    else if ((*hash)->is_empty())
+    else
     {
       DBUG_ASSERT(priv_spec.revoke);
       /* Clear this bit from the specified denies flag to speed up lookups. */
       specified_denies&= ~priv_spec.spec_type;
     }
+
     return false;
   }
 
@@ -677,6 +759,7 @@ public:
 
     if (load_hash_from_json(start, end, "db", &db_denies) ||
         load_hash_from_json(start, end, "table", &table_denies) ||
+        load_hash_from_json(start, end, "column", &column_denies) ||
         load_hash_from_json(start, end, "procedure", &procedure_denies) ||
         load_hash_from_json(start, end, "function", &function_denies) ||
         load_hash_from_json(start, end, "package_spec", &package_spec_denies) ||
@@ -688,6 +771,8 @@ public:
       specified_denies|= DATABASE_PRIV;
     if (table_denies)
       specified_denies|= TABLE_PRIV;
+    if (column_denies)
+      specified_denies|= COLUMN_PRIV;
     if (procedure_denies)
       specified_denies|= PROCEDURE_PRIV;
     if (function_denies)
@@ -697,7 +782,6 @@ public:
     if (package_body_denies)
       specified_denies|= PACKAGE_BODY_PRIV;
 
-    // TODO(cvicentiu) continue loading column denies and other json denies.
     return false;
   }
 };
@@ -2224,7 +2308,6 @@ class User_table_json: public User_table
     String s;
     spec->get_json_repr(&s);
     set_value("deny", s.c_ptr(), s.length(), false);
-    // TODO(cvicentiu) see if this can be ommited.
     set_int_value("version_id", (longlong) MYSQL_VERSION_ID);
     return false;
   }
@@ -5130,6 +5213,7 @@ static privilege_t acl_get_effective_deny_mask_impl(
     const Deny_spec &denies,
     const LEX_CSTRING &db,
     const LEX_CSTRING &object_name,
+    const LEX_CSTRING &column_name,
     enum object_type type)
 {
   privilege_t result= denies.get_global();
@@ -5140,6 +5224,8 @@ static privilege_t acl_get_effective_deny_mask_impl(
     {
       case TABLE_TYPE:
         result|= denies.get_table_deny(db, object_name);
+        if (column_name.str)
+          result|= denies.get_column_deny(db, object_name, column_name);
         break;
       case PROCEDURE_TYPE:
         result|= denies.get_procedure_deny(db, object_name);
@@ -5161,6 +5247,7 @@ static privilege_t acl_get_effective_deny_mask_impl(
 privilege_t acl_get_effective_deny_mask_impl(const Security_context *ctx,
                                              const LEX_CSTRING &db,
                                              const LEX_CSTRING &object_name,
+                                             const LEX_CSTRING &column_name,
                                              enum object_type type)
 {
   privilege_t result;
@@ -5183,6 +5270,13 @@ privilege_t acl_get_effective_deny_mask_impl(const Security_context *ctx,
   if (!object_name.str && !(ctx->denies_active & (GLOBAL_PRIV | DATABASE_PRIV)))
     return NO_ACL;
 
+  /*
+     If only column denies are active and we are not looking for a specific
+     column, no point in looking at the denies.
+  */
+  if (!column_name.str && ctx->denies_active == COLUMN_PRIV)
+    return NO_ACL;
+
   mysql_mutex_lock(&acl_cache->lock);
   ACL_USER *acl_user= find_user_exact(ctx->priv_host, ctx->priv_user);
   if (!acl_user)
@@ -5204,7 +5298,7 @@ privilege_t acl_get_effective_deny_mask_impl(const Security_context *ctx,
   }
 
   result= acl_get_effective_deny_mask_impl(*acl_user->denies,
-                                           db, object_name, type);
+                                           db, object_name, column_name, type);
 
 
   //TODO(cvicentiu) roles
@@ -5215,9 +5309,10 @@ privilege_t acl_get_effective_deny_mask_impl(const Security_context *ctx,
 
 privilege_t acl_get_effective_deny_mask(const Security_context *ctx,
                                         const LEX_CSTRING &db,
-                                        const LEX_CSTRING &table)
+                                        const LEX_CSTRING &table,
+                                        const LEX_CSTRING &column)
 {
-  return acl_get_effective_deny_mask_impl(ctx, db, table, TABLE_TYPE);
+  return acl_get_effective_deny_mask_impl(ctx, db, table, column, TABLE_TYPE);
 }
 
 /*
@@ -6935,7 +7030,8 @@ public:
   uint key_length;
   GRANT_COLUMN(String &c, privilege_t y) :rights (y), init_rights(y)
   {
-    column= (char*) memdup_root(&grant_memroot,c.ptr(), key_length=c.length());
+    key_length=c.length();
+    column= (char*) memdup_root(&grant_memroot, c.c_ptr(), c.length() + 1);
   }
 
   /* this constructor assumes thas source->column is allocated in grant_memroot */
@@ -9754,6 +9850,66 @@ bool grant_reload(THD *thd)
 }
 
 
+/* TODO(cvicentiu) extend this to pass in both user and role denies. */
+static
+bool check_some_grants_remain(const Deny_spec &denies,
+                              privilege_t table_deny_mask, GRANT_TABLE *table)
+{
+  if (!table)
+    return false;
+
+  LEX_CSTRING db{table->db, strlen(table->db)};
+  LEX_CSTRING table_name{table->tname, strlen(table->tname)};
+
+  for (size_t col_idx= 0; col_idx < table->hash_columns.records; col_idx++)
+  {
+    GRANT_COLUMN *grant_column= (GRANT_COLUMN*)
+      my_hash_element(&table->hash_columns, col_idx);
+    const LEX_CSTRING column_name{grant_column->column,
+                                  grant_column->key_length};
+
+    privilege_t col_deny_mask=
+      table_deny_mask | denies.get_column_deny(db, table_name, column_name);
+
+    if (grant_column->rights & ~col_deny_mask)
+      return true;
+  }
+  return false;
+}
+
+
+static
+bool check_some_grants_remain(Security_context *sctx,
+                              const LEX_CSTRING &db, const LEX_CSTRING &table,
+                              GRANT_TABLE *grant_table,
+                              GRANT_TABLE *grant_table_role)
+{
+  bool result= false;
+  if (!grant_table && !grant_table_role)
+    return false;
+  /* TODO(cvicentiu) extend this to pass in both user and role denies. */
+  mysql_mutex_lock(&acl_cache->lock);
+  ACL_USER *acl_user= find_user_exact(sctx->priv_host, sctx->priv_user);
+  if (!acl_user)
+  {
+     mysql_mutex_unlock(&acl_cache->lock);
+     return false;
+  }
+
+  DBUG_ASSERT(sctx->denies_active && acl_user->denies);
+  const Deny_spec& denies= *acl_user->denies;
+  privilege_t deny_mask= denies.get_global() |
+                         denies.get_db_deny(db) |
+                         denies.get_table_deny(db, table);
+
+  result= check_some_grants_remain(denies, deny_mask, grant_table) ||
+          check_some_grants_remain(denies, deny_mask, grant_table_role);
+
+  mysql_mutex_unlock(&acl_cache->lock);
+  return result;
+}
+
+
 /**
   @brief Check table level grants
 
@@ -9770,6 +9926,10 @@ bool grant_reload(THD *thd)
   in the set of COL_ACLS but access was not granted on the table level. As
   a consequence an extra check of column privileges is required.
 
+  It is possible for table->grant.privileges & table->grant.want_privilege
+  to be non-zero, in case of a column level deny. This forces callers
+  to do column level checks as well.
+
   Specifically if this function returns FALSE the user has some kind of
   privilege on a combination of columns in each table.
 
@@ -9889,7 +10049,18 @@ bool check_grant(THD *thd, privilege_t want_access, TABLE_LIST *tables,
                                                        t_ref->get_db_name(),
                                                        t_ref->get_table_name());
 
-    want_access&= ~(sctx->master_access & ~deny_mask);
+    if (sctx->denies_active & COLUMN_PRIV)
+    {
+      /*
+         We do now know if any global level grants are masked by column
+         level denies, so we assume all of them are masked at this point.
+
+         This still allows the shortcuts to below to work.
+      */
+      want_access&= ~(sctx->master_access & ~(deny_mask | COL_ACLS));
+    }
+    else
+      want_access&= ~(sctx->master_access & ~deny_mask);
 
     /*
        Make sure that whatever privileges were inherited from the global
@@ -9924,7 +10095,15 @@ bool check_grant(THD *thd, privilege_t want_access, TABLE_LIST *tables,
       continue;
     }
 
-    if (!(~t_ref->grant.privilege & want_access))
+    privilege_t grant_ensured_privilege= t_ref->grant.privilege;
+    /*
+       Mask column privileges as we don't know if they're all (or some)
+       blocked by column level denies.
+    */
+    if (sctx->denies_active & COLUMN_PRIV)
+      grant_ensured_privilege&= ~COL_ACLS;
+
+    if (!(~grant_ensured_privilege & want_access))
       continue;
 
     if (is_temporary_table(t_ref))
@@ -9975,10 +10154,14 @@ bool check_grant(THD *thd, privilege_t want_access, TABLE_LIST *tables,
     const bool no_grant_table_role_rights=
       !grant_table_role ||
       !((grant_table_role->privs | grant_table_role->cols) & ~deny_mask);
-    //TODO(cvicentiu) column level deny_mask needs to be computed separately
-    //per-column. This will have to be done as an extra step.
 
-    if (no_grant_table_rights && no_grant_table_role_rights)
+    /*
+       We may reach this point because column denies are active, but
+       it doesn't mean that access is denied, only that we need
+       to check individual columns.
+    */
+    if (no_grant_table_rights && no_grant_table_role_rights &&
+        !(sctx->denies_active & COLUMN_PRIV))
     {
       want_access&= ~(t_ref->grant.privilege & ~deny_mask);
       goto err;					// No grants
@@ -9989,7 +10172,25 @@ bool check_grant(THD *thd, privilege_t want_access, TABLE_LIST *tables,
       privileges on any column combination on the table.
     */
     if (any_combination_will_do)
-      continue;
+    {
+      if (!(sctx->denies_active & COLUMN_PRIV))
+        continue;
+      /*
+         Column DENIES active, we have no choice but to go through
+         all columns and see if any remain.
+
+         TODO(cvicentiu) check_grant is much too big of a function with
+         too many shortcuts and one that catches quite a number of edge cases.
+
+         Deconstruct this so it's not "used for everything under the sun".
+      */
+      if (check_some_grants_remain(sctx,
+                                   t_ref->get_db_name(),
+                                   t_ref->get_table_name(),
+                                   grant_table, grant_table_role))
+        continue;
+      goto err; // Impossible.
+    }
 
     t_ref->grant.grant_table_user= grant_table; // Remember for column test
     t_ref->grant.grant_table_role= grant_table_role;
@@ -9998,7 +10199,17 @@ bool check_grant(THD *thd, privilege_t want_access, TABLE_LIST *tables,
     t_ref->grant.privilege|= grant_table_role ? grant_table_role->privs : NO_ACL;
     // TODO(cvicentiu) here the deny mask for the role must be created too.
     t_ref->grant.privilege&= ~deny_mask;
-    t_ref->grant.want_privilege= ((want_access & COL_ACLS) & ~t_ref->grant.privilege);
+
+    /*
+       Take into account column denies and mark all potentially masked
+       COL_ACLS as part of "want_privilege"
+    */
+    grant_ensured_privilege= t_ref->grant.privilege;
+    if (sctx->denies_active & COLUMN_PRIV)
+      grant_ensured_privilege&= ~COL_ACLS;
+
+    t_ref->grant.want_privilege= ((want_access & COL_ACLS) &
+                                  ~grant_ensured_privilege);
 
     if (!(~t_ref->grant.privilege & want_access))
       continue;
@@ -10075,18 +10286,35 @@ bool check_grant_column(const Security_context *sctx,
                         const LEX_CSTRING &table_name,
                         const LEX_CSTRING &field_name)
 {
-  privilege_t want_access(grant->want_privilege & ~grant->privilege);
+  privilege_t want_access(grant->want_privilege);
   DBUG_ENTER("check_grant_column");
   DBUG_PRINT("enter", ("table: %s  want_access: %llx",
                        table_name, (longlong) want_access));
 
+  privilege_t deny_mask= acl_get_effective_deny_mask(sctx, db_name, table_name,
+                                                     field_name);
+  if (want_access & deny_mask)
+    goto err;
+
+  /* TODO(cvicentiu)
+     This is a little bit of a hack in the sense that it relies on this function
+     being called after check_grant has executed.
+     We could not have known within check_grant context which columns effectively
+     are going to be accessed, hence all column-related ACLS were blocked
+     if there are column-level denies active, in preparation for checking of
+     rights here.
+
+     Now we know which column we are actively looking at so we could apply the
+     deny mask on top of the table's privileges. If we got to this point,
+     it means the mask didn't block any of the required privileges,
+     we can now drop our "want_access" request bits if they are covered by
+     table level grants.
+  */
+  want_access&= ~grant->privilege;
+
   if (!want_access)
     DBUG_RETURN(0);				// Already checked
 
-  /* TODO(cvicentiu) column level denies need to be checked here. */
-  privilege_t deny_mask= acl_get_effective_deny_mask(sctx, db_name, table_name);
-  if (want_access & deny_mask)
-    goto err;
 
   mysql_rwlock_rdlock(&LOCK_grant);
 
@@ -10222,6 +10450,10 @@ bool check_grant_all_columns(THD *thd, privilege_t want_access_arg,
   GRANT_INFO *grant;
   GRANT_TABLE *UNINIT_VAR(grant_table);
   GRANT_TABLE *UNINIT_VAR(grant_table_role);
+  const Deny_spec *denies= nullptr;
+  const bool denies_could_interfere=
+    sctx->denies_active & (GLOBAL_PRIV | DATABASE_PRIV | TABLE_PRIV |
+                           COLUMN_PRIV);
   /*
      Flag that gets set if privilege checking has to be performed on column
      level.
@@ -10230,6 +10462,24 @@ bool check_grant_all_columns(THD *thd, privilege_t want_access_arg,
 
   mysql_rwlock_rdlock(&LOCK_grant);
 
+  if (denies_could_interfere)
+  {
+    /*
+       TODO(cvicentiu) this should be somehow moved into a function so we
+       do not repeat logic here and in acl_get_effective_deny_mask.
+    */
+    mysql_mutex_lock(&acl_cache->lock);
+    /* TODO(cvicentiu) modify this for role denies too! */
+    ACL_USER *acl_user= find_user_exact(sctx->priv_host, sctx->priv_user);
+    if (!acl_user)
+      goto err; /* See acl_get_effective_deny_mask_impl, same shortcut. */
+    /*
+       Caching denies here, vs for every field means we don't need to find
+       it again.
+    */
+    denies= acl_user->denies;
+  }
+
   for (; !fields->end_of_fields(); fields->next())
   {
     if (fields->field() &&
@@ -10237,11 +10487,41 @@ bool check_grant_all_columns(THD *thd, privilege_t want_access_arg,
       continue;
     LEX_CSTRING *field_name= fields->name();
 
+    privilege_t field_deny_mask{NO_ACL};
+
+    /*
+       TODO(cvicentiu) This extra `if` here is needed so we avoid doing
+       strlens if denies are *not* used. Once field_iterator returns
+       LEX_CSTRINGs instead of const char *, we can remove it.
+    */
+    if (sctx->denies_active & (GLOBAL_PRIV | DATABASE_PRIV | TABLE_PRIV |
+                               COLUMN_PRIV))
+    {
+      /* TODO(cvicentiu) -> Role denies too... */
+      field_deny_mask=
+        acl_get_effective_deny_mask_impl(
+            *denies,
+            {fields->get_db_name(), strlen(fields->get_db_name())},
+            {fields->get_table_name(), strlen(fields->get_table_name())},
+            *field_name,
+            TABLE_TYPE);
+      /*
+         Denies mask required privileges, we can never grant access for
+         this query.
+      */
+      if (field_deny_mask & want_access_arg)
+      {
+        want_access= want_access_arg;
+        goto err;
+      }
+    }
+
     if (table_name != fields->get_table_name())
     {
       table_name= fields->get_table_name();
       db_name= fields->get_db_name();
       grant= fields->grant();
+
       /* get a fresh one for each table */
       want_access= want_access_arg & ~grant->privilege;
       if (want_access)
@@ -10292,10 +10572,15 @@ bool check_grant_all_columns(THD *thd, privilege_t want_access_arg,
         goto err;
     }
   }
+
+  if (denies_could_interfere)
+    mysql_mutex_unlock(&acl_cache->lock);
   mysql_rwlock_unlock(&LOCK_grant);
   return 0;
 
 err:
+  if (denies_could_interfere)
+    mysql_mutex_unlock(&acl_cache->lock);
   mysql_rwlock_unlock(&LOCK_grant);
 
   char command[128];
@@ -10342,6 +10627,7 @@ static bool check_grant_db_routine(Security_context *sctx,
       {
         LEX_CSTRING routine_name= {item->tname, strlen(item->tname)};
         usable_mask= acl_get_effective_deny_mask_impl(sctx, db, routine_name,
+                                                      {nullptr, 0},
                                                       routine_type);
       }
       if (item->privs & ~usable_mask)
@@ -10375,6 +10661,9 @@ bool check_grant_db(Security_context *sctx,
   char helping2 [SAFE_NAME_LEN + USERNAME_LENGTH+2], *tmp_db;
   uint len, UNINIT_VAR(len2);
   bool error= TRUE;
+  bool denies_at_table_or_column_level=
+    sctx->denies_active & (TABLE_PRIV | COLUMN_PRIV);
+  const Deny_spec *denies= nullptr;
 
   LEX_CSTRING lower_case_db= db;
 
@@ -10409,28 +10698,36 @@ bool check_grant_db(Security_context *sctx,
      We have specific TABLE / Column level denies that need to mask
      privileges from specific GRANT_TABLEs.
   */
-  privilege_t deny_mask= db_deny_mask;
-  //TODO(cvicentiu) deny_mask needs to be updated when table
-  //and column denies are added.
+  if (denies_at_table_or_column_level)
+  {
+    mysql_mutex_lock(&acl_cache->lock);
+    /* TODO(cvicentiu) modify this for role denies too! */
+    ACL_USER *acl_user= find_user_exact(sctx->priv_host, sctx->priv_user);
+    if (!acl_user)
+      goto error; /* See acl_get_effective_deny_mask_impl, same shortcut. */
+    /*
+       Caching denies here, vs for every hash entry means we don't need to find
+       it again.
+    */
+    denies= acl_user->denies;
+  }
 
   for (size_t idx=0 ; idx < column_priv_hash.records ; idx++)
   {
+    privilege_t deny_mask= db_deny_mask;
     GRANT_TABLE *grant_table= (GRANT_TABLE*) my_hash_element(&column_priv_hash,
                                                              idx);
     if (len < grant_table->key_length &&
         !memcmp(grant_table->hash_key, helping, len) &&
         compare_hostname(&grant_table->host, sctx->host, sctx->ip))
     {
-      LEX_CSTRING db_name{grant_table->db, strlen(grant_table->db)};
-      LEX_CSTRING table_name{grant_table->tname, strlen(grant_table->tname)};
+      LEX_CSTRING table_name;
+      if (denies_at_table_or_column_level)
+      {
+        table_name= {grant_table->tname, strlen(grant_table->tname)};
+        deny_mask|= denies->get_table_deny(db, table_name);
+      }
 
-      /*
-         TODO(cvicentiu)
-         This does a find_user_exact in a loop. It can be optimized by
-         fetching the ACL_USER before the for loop and using Denies object
-         directly, see acl_get_effective_deny_mask_impl(Deny_spec &, ...)
-      */
-      deny_mask= acl_get_effective_deny_mask(sctx, db_name, table_name);
       if (grant_table->privs & ~deny_mask)
       {
         error= false; /* Found match. */
@@ -10443,35 +10740,25 @@ bool check_grant_db(Security_context *sctx,
          TODO(cvicentiu) add a test that showcases this shortcut not
          being taken because "cols" is not fully masked.
       */
-      if ((grant_table->cols & ~deny_mask) &&
-          !(sctx->denies_active & COLUMN_PRIV))
-      {
-        error= false;
-        break;
-      }
-
-      /*
-         Table level grants have been denied. There might be
-         column level grants that are present that may not be masked.
-
-         If the deny mask is not null and there are table or column 
-      */
-      if (deny_mask || (sctx->denies_active & COLUMN_PRIV))
+      if (grant_table->cols & ~deny_mask)
       {
+        /* Nothing can mask column grants. */
+        if (!(sctx->denies_active & COLUMN_PRIV))
+        {
+          error= false;
+          break;
+        }
         /*
-           TODO(cvicentiu) deny_mask needs to be updated when column denies are
-           added.
+           Table level grants have been denied. There might be
+           column level grants that are present that may not be masked, however
+           we have column denies active. We must go through each column
+           to check if there is an explicit deny on each of those columns
+           and compute the effective privileges.
         */
-        for (size_t col_idx= 0; col_idx < grant_table->hash_columns.records;
-             col_idx++)
+        if (check_some_grants_remain(*denies, deny_mask, grant_table))
         {
-          GRANT_COLUMN *grant_column= (GRANT_COLUMN*)
-            my_hash_element(&grant_table->hash_columns, col_idx);
-          if (grant_column->rights & ~deny_mask)
-          {
-            error= false; /* Found match. */
-            goto error;
-          }
+          error= false;
+          goto error;
         }
       }
     }
@@ -10489,14 +10776,15 @@ bool check_grant_db(Security_context *sctx,
     }
   }
 
-  //TODO(cvicentiu) implement deny_mask for check_grant_db_routine
   if (error)
-    error= check_grant_db_routine(sctx, lower_case_db, PROCEDURE_TYPE, deny_mask) &&
-           check_grant_db_routine(sctx, lower_case_db, FUNCTION_TYPE, deny_mask) &&
-           check_grant_db_routine(sctx, lower_case_db, PACKAGE_SPEC_TYPE, deny_mask) &&
-           check_grant_db_routine(sctx, lower_case_db, PACKAGE_BODY_TYPE, deny_mask);
+    error= check_grant_db_routine(sctx, lower_case_db, PROCEDURE_TYPE, db_deny_mask) &&
+           check_grant_db_routine(sctx, lower_case_db, FUNCTION_TYPE, db_deny_mask) &&
+           check_grant_db_routine(sctx, lower_case_db, PACKAGE_SPEC_TYPE, db_deny_mask) &&
+           check_grant_db_routine(sctx, lower_case_db, PACKAGE_BODY_TYPE, db_deny_mask);
 
 error:
+  if (denies_at_table_or_column_level)
+    mysql_mutex_unlock(&acl_cache->lock);
   mysql_rwlock_unlock(&LOCK_grant);
 
   return error;
@@ -10544,6 +10832,7 @@ static bool check_grant_routine(THD *thd, privilege_t want_access,
         thd->security_ctx,
         table->get_db_name(),
         table->get_table_name(),
+        {nullptr, 0},
         get_corresponding_object_type(*sph));
 
     GRANT_NAME *grant_proc;
@@ -10614,7 +10903,7 @@ bool check_routine_level_acl(THD *thd,
 
   privilege_t deny_mask=
     acl_get_effective_deny_mask_impl(thd->security_context(),
-                                     db, name,
+                                     db, name, {nullptr, 0},
                                      get_corresponding_object_type(sph));
 
   mysql_rwlock_rdlock(&LOCK_grant);
@@ -10708,12 +10997,8 @@ privilege_t get_column_grant(THD *thd, GRANT_INFO *grant,
 
   mysql_rwlock_rdlock(&LOCK_grant);
 
-  /*
-     TODO(cvicentiu) One idea is to save this alongside grant->version to
-     speed up lookup.
-  */
   deny_mask= acl_get_effective_deny_mask(thd->security_ctx,
-                                         db_name, table_name);
+                                         db_name, table_name, field_name);
 
   /* reload table if someone has modified any grants */
   if (grant->version != grant_version)
@@ -10760,8 +11045,6 @@ privilege_t get_column_grant(THD *thd, GRANT_INFO *grant,
     }
   }
   mysql_rwlock_unlock(&LOCK_grant);
-  /* TODO(cvicentiu) This mask only covers global, database, table denies.
-     Column level denies need dedicated handling. */
   return priv & ~deny_mask;
 }
 
@@ -14098,6 +14381,7 @@ static bool maria_deny(THD *thd, const User_table &user_table,
                MYF(0));
     DBUG_RETURN(true);
   }
+
   /*
      TODO(cvicentiu) column-level checks, etc.
      Write tests for these too.
@@ -14208,9 +14492,8 @@ bool Sql_cmd_grant_sp::execute(THD *thd)
                                              MODE_NO_AUTO_CREATE_USER);
     bool result;
     Priv_spec priv_spec(grants, is_revoke(), m_gp.db(),
-                        table->get_table_name(), true);
-    priv_spec.spec_type = get_corresponding_priv_spec_type(m_sph);
-                          //TODO (cvicentiu) clean this up, use constructor.
+                        table->get_table_name(),
+                        get_corresponding_priv_spec_type(m_sph), true);
     if (tables.open_and_lock(thd, Table_user, TL_WRITE))
       return true;
 
@@ -14250,7 +14533,8 @@ bool Sql_cmd_grant_table::execute_deny(THD *thd, TABLE_LIST *table)
                                            MODE_NO_AUTO_CREATE_USER);
 
   Priv_spec priv_spec{m_gp.object_privilege(), is_revoke(),
-                      m_gp.db(), table->table_name, true};
+                      m_gp.db(), table->table_name, &m_gp.columns(),
+                      true};
 
   privilege_t required_access= m_gp.object_privilege() |
                                m_gp.column_privilege_total() |
@@ -14433,14 +14717,7 @@ bool Sql_cmd_grant_table::execute(THD *thd)
   if (m_deny)
   {
     if (table)
-    {
-      if (m_gp.columns().elements)
-      {
-        my_error(ER_UNSUPPORTED_DENY, MYF(0), "Column");
-        return true;
-      }
       return execute_deny(thd, table);
-    }
     return execute_deny(thd);
   }
 
diff --git a/sql/sql_acl.h b/sql/sql_acl.h
index c6a8372ce74..12376b08121 100644
--- a/sql/sql_acl.h
+++ b/sql/sql_acl.h
@@ -80,7 +80,8 @@ privilege_t acl_get_current_auth(Security_context *ctx, const char *db,
                                  bool db_is_pattern);
 privilege_t acl_get_effective_deny_mask(const Security_context *ctx,
                                         const LEX_CSTRING &db={nullptr, 0},
-                                        const LEX_CSTRING &table={nullptr, 0});
+                                        const LEX_CSTRING &table={nullptr, 0},
+                                        const LEX_CSTRING &column={nullptr, 0});
 bool acl_authenticate(THD *thd, uint com_change_user_pkt_len);
 bool acl_getroot(Security_context *sctx, const char *user, const char *host,
                  const char *ip, const char *db);
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index d31ba5210b6..1d92a1a2a03 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -8389,14 +8389,21 @@ insert_fields(THD *thd, Name_resolution_context *context, const char *db_name,
 
        - the table is a derived table
 
-       - the table is a view with SELECT privilege
+       - the table is a view with SELECT privilege and the
+         SELECT privilege could not have been masked by a DENY via column denies.
 
-       - the table is a base table with SELECT privilege
+       - the table is a base table with SELECT privilege and the
+         SELECT privilege could not have been masked by a DENY via column denies.
+         See check_grant's output for grant.want_privilege
     */
     if (!any_privileges &&
         !tables->is_derived() &&
-        !(tables->is_view() && (tables->grant.privilege & SELECT_ACL)) &&
-        !(table && (table->grant.privilege & SELECT_ACL)))
+        !(tables->is_view() &&
+          (tables->grant.privilege & SELECT_ACL) &&
+          !(tables->grant.want_privilege & SELECT_ACL)) &&
+        !(table &&
+          (table->grant.privilege & SELECT_ACL) &&
+          !(tables->grant.want_privilege & SELECT_ACL)))
     {
       field_iterator.set(tables);
       if (check_grant_all_columns(thd, SELECT_ACL, &field_iterator))
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 74ed078e936..02e22726550 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -488,6 +488,7 @@ int mysql_update(THD *thd,
   old_covering_keys= table->covering_keys;		// Keys used in WHERE
   /* Check the fields we are going to modify */
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
+  /* TODO(cvicentiu) Test this interaction with DENIES and views. */
   table_list->grant.want_privilege= table->grant.want_privilege= want_privilege;
   table_list->register_want_access(want_privilege);
 #endif
@@ -514,8 +515,11 @@ int mysql_update(THD *thd,
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
   /* Check values */
-  table_list->grant.want_privilege= table->grant.want_privilege=
-    (SELECT_ACL & ~table->grant.privilege);
+  if (thd->security_ctx->denies_active & COLUMN_PRIV)
+    table_list->grant.want_privilege= table->grant.want_privilege= SELECT_ACL;
+  else
+    table_list->grant.want_privilege= table->grant.want_privilege=
+      (SELECT_ACL & ~table->grant.privilege);
 #endif
   if (setup_fields(thd, Ref_ptr_array(), values, MARK_COLUMNS_READ, 0, NULL, 0))
   {
@@ -1421,8 +1425,11 @@ bool mysql_prepare_update(THD *thd, TABLE_LIST *table_list,
   DBUG_ENTER("mysql_prepare_update");
 
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
-  table_list->grant.want_privilege= table->grant.want_privilege= 
-    (SELECT_ACL & ~table->grant.privilege);
+  if (thd->security_ctx->denies_active & COLUMN_PRIV)
+    table_list->grant.want_privilege= table->grant.want_privilege= SELECT_ACL;
+  else
+    table_list->grant.want_privilege=
+      table->grant.want_privilege= (SELECT_ACL & ~table->grant.privilege);
   table_list->register_want_access(SELECT_ACL);
 #endif