diff --git a/src/Database/Adapter.php b/src/Database/Adapter.php
index 4d2f0ee38f..4acae02a81 100644
--- a/src/Database/Adapter.php
+++ b/src/Database/Adapter.php
@@ -1011,6 +1011,32 @@ abstract public function getSupportForAttributes(): bool;
      */
     abstract public function getSupportForSchemaAttributes(): bool;
 
+    /**
+     * Can a permission be scoped to a single column?
+     *
+     * @return bool
+     */
+    abstract public function getSupportForColumnPermissions(): bool;
+
+    /**
+     * Repoint column-scoped permissions at a renamed column.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string $new
+     * @return array<string> ids of documents whose $permissions changed
+     */
+    abstract public function renameColumnPermissions(Document $collection, string $old, string $new): array;
+
+    /**
+     * Drop every permission scoped to a column that no longer exists.
+     *
+     * @param Document $collection
+     * @param string $column
+     * @return array<string> ids of documents whose $permissions changed
+     */
+    abstract public function deleteColumnPermissions(Document $collection, string $column): array;
+
     /**
      * Are schema indexes supported?
      *
diff --git a/src/Database/Adapter/MariaDB.php b/src/Database/Adapter/MariaDB.php
index 6d2aac8ef7..fec9f5d9d3 100644
--- a/src/Database/Adapter/MariaDB.php
+++ b/src/Database/Adapter/MariaDB.php
@@ -189,11 +189,23 @@ public function createCollection(string $name, array $attributes = [], array $in
         $collection .= ")";
         $collection = $this->trigger(Database::EVENT_COLLECTION_CREATE, $collection);
 
+        // _column scopes a permission to a single column. An empty string means
+        // every column, which is how every permission written before column-level
+        // permissions reads. It is NOT NULL on purpose: MySQL and MariaDB treat
+        // NULLs as distinct in a UNIQUE index, so a nullable _column would let
+        // duplicate permission rows slip past _index1.
+        //
+        // _index1 indexes it by prefix, not in full: these tables are utf8mb4 and the
+        // other four members already cost ~2098 of InnoDB's 3072-byte key limit, so a
+        // full VARCHAR(255) member would take it to ~3120 and the index would fail to
+        // build. MAX_UID_DEFAULT_LENGTH is the longest a column key may be, so the
+        // prefix is full uniqueness for every value that can actually be stored.
         $permissions = "
             CREATE TABLE {$this->getSQLTable($id . '_perms')} (
                 _id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
                 _type VARCHAR(12) NOT NULL,
                 _permission VARCHAR(255) NOT NULL,
+                _column VARCHAR(255) NOT NULL DEFAULT '',
                 _document VARCHAR(255) NOT NULL,
                 PRIMARY KEY (_id),
         ";
@@ -201,12 +213,12 @@ public function createCollection(string $name, array $attributes = [], array $in
         if ($this->sharedTables) {
             $permissions .= "
                 _tenant INT(11) UNSIGNED DEFAULT NULL,
-                UNIQUE INDEX _index1 (_document, _tenant, _type, _permission),
+                UNIQUE INDEX _index1 (_document, _tenant, _type, _permission, _column(" . Database::MAX_UID_DEFAULT_LENGTH . ")),
                 INDEX _permission (_tenant, _permission, _type)
             ";
         } else {
             $permissions .= "
-                UNIQUE INDEX _index1 (_document, _type, _permission),
+                UNIQUE INDEX _index1 (_document, _type, _permission, _column(" . Database::MAX_UID_DEFAULT_LENGTH . ")),
                 INDEX _permission (_permission, _type)
             ";
         }
@@ -895,12 +907,14 @@ public function createDocument(Document $collection, Document $document): Docume
             }
 
             $permissions = [];
+            $permissionBinds = [];
             foreach (Database::PERMISSIONS as $type) {
-                foreach ($document->getPermissionsByType($type) as $permission) {
+                foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                     $tenantBind = $this->sharedTables ? ", :_tenant" : '';
-                    $permission = \str_replace('"', '', $permission);
-                    $permission = "('{$type}', '{$permission}', :_uid {$tenantBind})";
-                    $permissions[] = $permission;
+                    $role = \str_replace('"', '', $permission['role']);
+                    $columnBind = ":_column_{$type}_{$i}";
+                    $permissionBinds[$columnBind] = $permission['column'];
+                    $permissions[] = "('{$type}', '{$role}', {$columnBind}, :_uid {$tenantBind})";
                 }
             }
 
@@ -909,7 +923,7 @@ public function createDocument(Document $collection, Document $document): Docume
                 $permissions = \implode(', ', $permissions);
 
                 $sqlPermissions = "
-                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
+                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _column, _document {$tenantColumn})
                     VALUES {$permissions};
                 ";
 
@@ -918,6 +932,9 @@ public function createDocument(Document $collection, Document $document): Docume
                 if ($this->sharedTables) {
                     $stmtPermissions->bindValue(':_tenant', $document->getTenant());
                 }
+                foreach ($permissionBinds as $key => $value) {
+                    $stmtPermissions->bindValue($key, $value);
+                }
             }
 
             $stmt->execute();
@@ -1007,10 +1024,11 @@ public function updateDocument(Document $collection, string $id, Document $docum
                 $values = [];
                 $binds = [];
                 foreach (Database::PERMISSIONS as $type) {
-                    foreach ($document->getPermissionsByType($type) as $i => $permission) {
+                    foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                         $tenantPlaceholder = $this->sharedTables ? ', :_tenant' : '';
-                        $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i} {$tenantPlaceholder})";
-                        $binds[":_add_{$type}_{$i}"] = $permission;
+                        $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i}, :_addcol_{$type}_{$i} {$tenantPlaceholder})";
+                        $binds[":_add_{$type}_{$i}"] = $permission['role'];
+                        $binds[":_addcol_{$type}_{$i}"] = $permission['column'];
                     }
                 }
 
@@ -1018,7 +1036,7 @@ public function updateDocument(Document $collection, string $id, Document $docum
                     $tenantColumn = $this->sharedTables ? ', _tenant' : '';
 
                     $sql = "
-				    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission {$tenantColumn})
+				    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission, _column {$tenantColumn})
 				    VALUES " . \implode(', ', $values);
 
                     $sql = $this->trigger(Database::EVENT_PERMISSIONS_CREATE, $sql);
@@ -1771,6 +1789,11 @@ public function getSupportForUpsertOnUniqueIndex(): bool
         return true;
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return true;
+    }
+
     public function getSupportForSchemaAttributes(): bool
     {
         return true;
diff --git a/src/Database/Adapter/Memory.php b/src/Database/Adapter/Memory.php
index 5e126a7177..d3ad822c77 100644
--- a/src/Database/Adapter/Memory.php
+++ b/src/Database/Adapter/Memory.php
@@ -2086,6 +2086,30 @@ public function getSchemaIndexes(string $collection): array
         return [];
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return false;
+    }
+
+    /**
+     * Column-level permissions are not supported by this adapter, so a rename
+     * can never have column-scoped permissions to repoint.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string $new
+     * @return array<string>
+     */
+    public function renameColumnPermissions(Document $collection, string $old, string $new): array
+    {
+        return [];
+    }
+
+    public function deleteColumnPermissions(Document $collection, string $column): array
+    {
+        return [];
+    }
+
     public function getTenantQuery(string $collection, string $alias = ''): string
     {
         return '';
diff --git a/src/Database/Adapter/Mongo.php b/src/Database/Adapter/Mongo.php
index 136ebac0f2..7e99b66c9f 100644
--- a/src/Database/Adapter/Mongo.php
+++ b/src/Database/Adapter/Mongo.php
@@ -4202,6 +4202,30 @@ public function decodePolygon(string $wkb): array
         return [];
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return false;
+    }
+
+    /**
+     * Column-level permissions are not supported by this adapter, so a rename
+     * can never have column-scoped permissions to repoint.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string $new
+     * @return array<string>
+     */
+    public function renameColumnPermissions(Document $collection, string $old, string $new): array
+    {
+        return [];
+    }
+
+    public function deleteColumnPermissions(Document $collection, string $column): array
+    {
+        return [];
+    }
+
     /**
      * Get the query to check for tenant when in shared tables mode
      *
@@ -4209,6 +4233,7 @@ public function decodePolygon(string $wkb): array
      * @param string $alias The alias of the parent collection if in a subquery
      * @return string
      */
+
     public function getTenantQuery(string $collection, string $alias = ''): string
     {
         return '';
diff --git a/src/Database/Adapter/Pool.php b/src/Database/Adapter/Pool.php
index 511da2b13a..8a03059fc6 100644
--- a/src/Database/Adapter/Pool.php
+++ b/src/Database/Adapter/Pool.php
@@ -522,6 +522,21 @@ public function getSupportForAttributes(): bool
         return $this->delegate(__FUNCTION__, \func_get_args());
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return $this->delegate(__FUNCTION__, \func_get_args());
+    }
+
+    public function renameColumnPermissions(Document $collection, string $old, string $new): array
+    {
+        return $this->delegate(__FUNCTION__, \func_get_args());
+    }
+
+    public function deleteColumnPermissions(Document $collection, string $column): array
+    {
+        return $this->delegate(__FUNCTION__, \func_get_args());
+    }
+
     public function getSupportForSchemaAttributes(): bool
     {
         return $this->delegate(__FUNCTION__, \func_get_args());
diff --git a/src/Database/Adapter/Postgres.php b/src/Database/Adapter/Postgres.php
index 1241123264..ed2bb255ae 100644
--- a/src/Database/Adapter/Postgres.php
+++ b/src/Database/Adapter/Postgres.php
@@ -256,6 +256,7 @@ public function createCollection(string $name, array $attributes = [], array $in
                 _tenant INTEGER DEFAULT NULL,
                 _type VARCHAR(12) NOT NULL,
                 _permission VARCHAR(255) NOT NULL,
+                _column VARCHAR(255) NOT NULL DEFAULT '',
                 _document VARCHAR(255) NOT NULL
             );
         ";
@@ -265,7 +266,7 @@ public function createCollection(string $name, array $attributes = [], array $in
             $permissionIndex = $this->getShortKey("{$namespace}_{$this->tenant}_{$id}_permission");
             $permissions .= "
                 CREATE UNIQUE INDEX \"{$uniquePermissionIndex}\" 
-                    ON {$this->getSQLTable($id . '_perms')} USING btree (_tenant,_document,_type,_permission);
+                    ON {$this->getSQLTable($id . '_perms')} USING btree (_tenant,_document,_type,_permission,_column);
                 CREATE INDEX \"{$permissionIndex}\" 
                     ON {$this->getSQLTable($id . '_perms')} USING btree (_tenant,_permission,_type); 
             ";
@@ -274,7 +275,7 @@ public function createCollection(string $name, array $attributes = [], array $in
             $permissionIndex = $this->getShortKey("{$namespace}_{$id}_permission");
             $permissions .= "
                 CREATE UNIQUE INDEX \"{$uniquePermissionIndex}\" 
-                    ON {$this->getSQLTable($id . '_perms')} USING btree (_document COLLATE utf8_ci_ai,_type,_permission);
+                    ON {$this->getSQLTable($id . '_perms')} USING btree (_document COLLATE utf8_ci_ai,_type,_permission,_column);
                 CREATE INDEX \"{$permissionIndex}\" 
                     ON {$this->getSQLTable($id . '_perms')} USING btree (_permission,_type); 
             ";
@@ -1046,11 +1047,14 @@ public function createDocument(Document $collection, Document $document): Docume
         }
 
         $permissions = [];
+        $permissionBinds = [];
         foreach (Database::PERMISSIONS as $type) {
-            foreach ($document->getPermissionsByType($type) as $permission) {
-                $permission = \str_replace('"', '', $permission);
+            foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
+                $role = \str_replace('"', '', $permission['role']);
                 $sqlTenant = $this->sharedTables ? ', :_tenant' : '';
-                $permissions[] = "('{$type}', '{$permission}', :_uid {$sqlTenant})";
+                $columnBind = ":_column_{$type}_{$i}";
+                $permissionBinds[$columnBind] = $permission['column'];
+                $permissions[] = "('{$type}', '{$role}', {$columnBind}, :_uid {$sqlTenant})";
             }
         }
 
@@ -1060,7 +1064,7 @@ public function createDocument(Document $collection, Document $document): Docume
             $sqlTenant = $this->sharedTables ? ', _tenant' : '';
 
             $queryPermissions = "
-				INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$sqlTenant})
+				INSERT INTO {$this->getSQLTable($name . '_perms')} (_type, _permission, _column, _document {$sqlTenant})
 				VALUES {$permissions}
 			";
 
@@ -1070,6 +1074,9 @@ public function createDocument(Document $collection, Document $document): Docume
             if ($sqlTenant) {
                 $stmtPermissions->bindValue(':_tenant', $document->getTenant());
             }
+            foreach ($permissionBinds as $key => $value) {
+                $stmtPermissions->bindValue($key, $value);
+            }
         }
 
         try {
@@ -1133,10 +1140,11 @@ public function updateDocument(Document $collection, string $id, Document $docum
             $values = [];
             $binds = [];
             foreach (Database::PERMISSIONS as $type) {
-                foreach ($document->getPermissionsByType($type) as $i => $permission) {
+                foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                     $sqlTenant = $this->sharedTables ? ', :_tenant' : '';
-                    $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i} {$sqlTenant})";
-                    $binds[":_add_{$type}_{$i}"] = $permission;
+                    $values[] = "( :_uid, '{$type}', :_add_{$type}_{$i}, :_addcol_{$type}_{$i} {$sqlTenant})";
+                    $binds[":_add_{$type}_{$i}"] = $permission['role'];
+                    $binds[":_addcol_{$type}_{$i}"] = $permission['column'];
                 }
             }
 
@@ -1144,7 +1152,7 @@ public function updateDocument(Document $collection, string $id, Document $docum
                 $sqlTenant = $this->sharedTables ? ', _tenant' : '';
 
                 $sql = "
-				INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission {$sqlTenant})
+				INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission, _column {$sqlTenant})
 				VALUES " . \implode(', ', $values);
 
                 $sql = $this->trigger(Database::EVENT_PERMISSIONS_CREATE, $sql);
@@ -2090,6 +2098,11 @@ public function getSupportForIntegerBooleans(): bool
      *
      * @return bool
      */
+    public function getSupportForColumnPermissions(): bool
+    {
+        return true;
+    }
+
     public function getSupportForSchemaAttributes(): bool
     {
         return false;
diff --git a/src/Database/Adapter/Redis.php b/src/Database/Adapter/Redis.php
index 81f3350634..8eed228caa 100644
--- a/src/Database/Adapter/Redis.php
+++ b/src/Database/Adapter/Redis.php
@@ -771,6 +771,30 @@ public function setSupportForAttributes(bool $support): bool
         return true;
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return false;
+    }
+
+    /**
+     * Column-level permissions are not supported by this adapter, so a rename
+     * can never have column-scoped permissions to repoint.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string $new
+     * @return array<string>
+     */
+    public function renameColumnPermissions(Document $collection, string $old, string $new): array
+    {
+        return [];
+    }
+
+    public function deleteColumnPermissions(Document $collection, string $column): array
+    {
+        return [];
+    }
+
     public function getSupportForSchemaAttributes(): bool
     {
         return false;
diff --git a/src/Database/Adapter/SQL.php b/src/Database/Adapter/SQL.php
index 9ca4c1aee3..0904b7ed1d 100644
--- a/src/Database/Adapter/SQL.php
+++ b/src/Database/Adapter/SQL.php
@@ -15,6 +15,7 @@
 use Utopia\Database\Exception\Timeout as TimeoutException;
 use Utopia\Database\Exception\Transaction as TransactionException;
 use Utopia\Database\Helpers\ID;
+use Utopia\Database\Helpers\Permission;
 use Utopia\Database\Operator;
 use Utopia\Database\Query;
 
@@ -629,7 +630,7 @@ public function updateDocuments(Document $collection, Document $updates, array $
                 }
 
                 $sql = "
-                    SELECT _type, _permission
+                    SELECT _type, _permission, _column
                     FROM {$this->getSQLTable($name . '_perms')}
                     WHERE _document = :_uid
                     {$this->getTenantQuery($collection)}
@@ -654,14 +655,24 @@ public function updateDocuments(Document $collection, Document $updates, array $
                 }
 
                 $permissions = \array_reduce($permissions, function (array $carry, array $item) {
-                    $carry[$item['_type']][] = $item['_permission'];
+                    $carry[$item['_type']][] = $item['_permission'] . "\0" . ($item['_column'] ?? '');
                     return $carry;
                 }, $initial);
 
+                // Desired state in the same role\0column shape, so a permission that
+                // only changes column still shows up as a removal plus an addition.
+                $desired = [];
+                foreach (Database::PERMISSIONS as $type) {
+                    $desired[$type] = \array_map(
+                        fn (array $permission) => $permission['role'] . "\0" . $permission['column'],
+                        $updates->getPermissionsByTypeWithColumns($type)
+                    );
+                }
+
                 // Get removed Permissions
                 $removals = [];
                 foreach (Database::PERMISSIONS as $type) {
-                    $diff = array_diff($permissions[$type], $updates->getPermissionsByType($type));
+                    $diff = array_diff($permissions[$type], $desired[$type]);
                     if (!empty($diff)) {
                         $removals[$type] = $diff;
                     }
@@ -674,18 +685,25 @@ public function updateDocuments(Document $collection, Document $updates, array $
                         $removeBindKeys[] = ':_uid_' . $index;
                         $removeBindValues[$bindKey] = $document->getId();
 
+                        $pairs = [];
+                        foreach (\array_keys($permissionsToRemove) as $i) {
+                            [$role, $column] = \explode("\0", $permissionsToRemove[$i], 2);
+
+                            $roleBind = 'remove_' . $type . '_' . $index . '_' . $i;
+                            $columnBind = 'removecol_' . $type . '_' . $index . '_' . $i;
+                            $removeBindKeys[] = ':' . $roleBind;
+                            $removeBindKeys[] = ':' . $columnBind;
+                            $removeBindValues[$roleBind] = $role;
+                            $removeBindValues[$columnBind] = $column;
+
+                            $pairs[] = "(_permission = :{$roleBind} AND _column = :{$columnBind})";
+                        }
+
                         $removeQueries[] = "(
                             _document = :_uid_{$index}
                             {$this->getTenantQuery($collection)}
                             AND _type = '{$type}'
-                            AND _permission IN (" . \implode(', ', \array_map(function (string $i) use ($permissionsToRemove, $index, $type, &$removeBindKeys, &$removeBindValues) {
-                            $bindKey = 'remove_' . $type . '_' . $index . '_' . $i;
-                            $removeBindKeys[] = ':' . $bindKey;
-                            $removeBindValues[$bindKey] = $permissionsToRemove[$i];
-
-                            return ':' . $bindKey;
-                        }, \array_keys($permissionsToRemove))) .
-                            ")
+                            AND (" . \implode(' OR ', $pairs) . ")
                         )";
                     }
                 }
@@ -693,7 +711,7 @@ public function updateDocuments(Document $collection, Document $updates, array $
                 // Get added Permissions
                 $additions = [];
                 foreach (Database::PERMISSIONS as $type) {
-                    $diff = \array_diff($updates->getPermissionsByType($type), $permissions[$type]);
+                    $diff = \array_diff($desired[$type], $permissions[$type]);
                     if (!empty($diff)) {
                         $additions[$type] = $diff;
                     }
@@ -703,13 +721,18 @@ public function updateDocuments(Document $collection, Document $updates, array $
                 if (!empty($additions)) {
                     foreach ($additions as $type => $permissionsToAdd) {
                         foreach ($permissionsToAdd as $i => $permission) {
+                            [$role, $column] = \explode("\0", $permission, 2);
+
                             $bindKey = '_uid_' . $index;
                             $addBindValues[$bindKey] = $document->getId();
 
                             $bindKey = 'add_' . $type . '_' . $index . '_' . $i;
-                            $addBindValues[$bindKey] = $permission;
+                            $addBindValues[$bindKey] = $role;
+
+                            $columnBindKey = 'addcol_' . $type . '_' . $index . '_' . $i;
+                            $addBindValues[$columnBindKey] = $column;
 
-                            $addQuery .= "(:_uid_{$index}, '{$type}', :{$bindKey}";
+                            $addQuery .= "(:_uid_{$index}, '{$type}', :{$bindKey}, :{$columnBindKey}";
 
                             if ($this->sharedTables) {
                                 $addQuery .= ", :_tenant)";
@@ -749,7 +772,7 @@ public function updateDocuments(Document $collection, Document $updates, array $
 
             if (!empty($addQuery)) {
                 $sqlAddPermissions = "
-                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission
+                    INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission, _column
                 ";
 
                 if ($this->sharedTables) {
@@ -1987,6 +2010,187 @@ protected function getSQLPermissionsCondition(
         )";
     }
 
+    public function getSupportForColumnPermissions(): bool
+    {
+        return false;
+    }
+
+    /**
+     * Repoint column-scoped permissions at a renamed column.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string $new
+     * @return array<string> ids of documents whose $permissions changed
+     * @throws DatabaseException
+     */
+    public function renameColumnPermissions(Document $collection, string $old, string $new): array
+    {
+        return $this->repointColumnPermissions($collection, $old, $new);
+    }
+
+    /**
+     * Drop every permission scoped to a column that no longer exists.
+     *
+     * Required, not hygiene: because permissions name the column by key, leaving
+     * rows behind means re-creating a column under the same name inherits the old
+     * column's grants.
+     *
+     * @param Document $collection
+     * @param string $column
+     * @return array<string> ids of documents whose $permissions changed
+     * @throws DatabaseException
+     */
+    public function deleteColumnPermissions(Document $collection, string $column): array
+    {
+        return $this->repointColumnPermissions($collection, $column, null);
+    }
+
+    /**
+     * Move or drop the permissions scoped to one column.
+     *
+     * The column key lives in two places: _perms._column, which backs permission
+     * queries, and the _permissions JSON on the collection table, which is what
+     * callers read back as $permissions. Both have to change together.
+     *
+     * The _perms lookup runs first and is empty whenever nobody scoped a permission
+     * to this column, in which case there is nothing else to do. When it is not
+     * empty it yields a bounded set of document ids, so the JSON rewrite stays
+     * targeted instead of scanning the whole collection.
+     *
+     * @param Document $collection
+     * @param string $old
+     * @param string|null $new new column key, or null to drop the permissions
+     * @return array<string> ids of documents whose $permissions changed
+     * @throws DatabaseException
+     */
+    private function repointColumnPermissions(Document $collection, string $old, ?string $new): array
+    {
+        $name = $this->filter($collection->getId());
+        $tenantQuery = $this->getTenantQuery($collection->getId());
+
+        $stmt = $this->getPDO()->prepare("
+            SELECT DISTINCT _document
+            FROM {$this->getSQLTable($name . '_perms')}
+            WHERE _column = :_column
+            {$tenantQuery}
+        ");
+        $stmt->bindValue(':_column', $old);
+        if ($this->sharedTables) {
+            $stmt->bindValue(':_tenant', $this->tenant);
+        }
+        $this->execute($stmt);
+
+        $documents = $stmt->fetchAll(\PDO::FETCH_COLUMN);
+        $stmt->closeCursor();
+
+        if (empty($documents)) {
+            return [];
+        }
+
+        if (\is_null($new)) {
+            $stmt = $this->getPDO()->prepare("
+                DELETE FROM {$this->getSQLTable($name . '_perms')}
+                WHERE _column = :_old
+                {$tenantQuery}
+            ");
+        } else {
+            $stmt = $this->getPDO()->prepare("
+                UPDATE {$this->getSQLTable($name . '_perms')}
+                SET _column = :_new
+                WHERE _column = :_old
+                {$tenantQuery}
+            ");
+            $stmt->bindValue(':_new', $new);
+        }
+
+        $stmt->bindValue(':_old', $old);
+        if ($this->sharedTables) {
+            $stmt->bindValue(':_tenant', $this->tenant);
+        }
+        $this->execute($stmt);
+
+        $placeholders = \implode(', ', \array_map(
+            fn ($index) => ":_uid_{$index}",
+            \array_keys($documents)
+        ));
+
+        $select = $this->getPDO()->prepare("
+            SELECT _uid, _permissions
+            FROM {$this->getSQLTable($name)}
+            WHERE _uid IN ({$placeholders})
+            {$tenantQuery}
+        ");
+        foreach ($documents as $index => $id) {
+            $select->bindValue(":_uid_{$index}", $id);
+        }
+        if ($this->sharedTables) {
+            $select->bindValue(':_tenant', $this->tenant);
+        }
+        $this->execute($select);
+
+        $rows = $select->fetchAll();
+        $select->closeCursor();
+
+        $update = $this->getPDO()->prepare("
+            UPDATE {$this->getSQLTable($name)}
+            SET _permissions = :_permissions
+            WHERE _uid = :_uid
+            {$tenantQuery}
+        ");
+
+        $updated = [];
+
+        foreach ($rows as $row) {
+            $permissions = \json_decode($row['_permissions'] ?? '[]', true);
+
+            if (!\is_array($permissions)) {
+                continue;
+            }
+
+            $rewritten = [];
+            $changed = false;
+
+            foreach ($permissions as $permission) {
+                $parsed = Permission::parse($permission);
+
+                if ($parsed->getColumn() !== $old) {
+                    $rewritten[] = $permission;
+                    continue;
+                }
+
+                $changed = true;
+
+                if (\is_null($new)) {
+                    continue;
+                }
+
+                $rewritten[] = (new Permission(
+                    $parsed->getPermission(),
+                    $parsed->getRole(),
+                    $parsed->getIdentifier(),
+                    $parsed->getDimension(),
+                    $new
+                ))->toString();
+            }
+
+            if (!$changed) {
+                continue;
+            }
+
+            $update->bindValue(':_permissions', \json_encode($rewritten));
+            $update->bindValue(':_uid', $row['_uid']);
+            if ($this->sharedTables) {
+                $update->bindValue(':_tenant', $this->tenant);
+            }
+            $this->execute($update);
+
+            $updated[] = $row['_uid'];
+        }
+
+        return $updated;
+    }
+
     /**
      * Get SQL table
      *
@@ -2512,11 +2716,12 @@ public function createDocuments(Document $collection, array $documents): array
                 $batchKeys[] = '(' . \implode(', ', $bindKeys) . ')';
 
                 foreach (Database::PERMISSIONS as $type) {
-                    foreach ($document->getPermissionsByType($type) as $permission) {
+                    foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                         $tenantBind = $this->sharedTables ? ", :_tenant_{$index}" : '';
-                        $permission = \str_replace('"', '', $permission);
-                        $permission = "('{$type}', '{$permission}', :_uid_{$index} {$tenantBind})";
-                        $permissions[] = $permission;
+                        $role = \str_replace('"', '', $permission['role']);
+                        $columnBind = ":_column_{$type}_{$index}_{$i}";
+                        $bindValuesPermissions[$columnBind] = $permission['column'];
+                        $permissions[] = "('{$type}', '{$role}', {$columnBind}, :_uid_{$index} {$tenantBind})";
                         $bindValuesPermissions[":_uid_{$index}"] = $document->getId();
                         if ($this->sharedTables) {
                             $bindValuesPermissions[":_tenant_{$index}"] = $document->getTenant();
@@ -2544,7 +2749,7 @@ public function createDocuments(Document $collection, array $documents): array
                 $permissions = \implode(', ', $permissions);
 
                 $sqlPermissions = "
-                    {$this->getInsertKeyword()} {$this->getSQLTable($name . '_perms')} (_type, _permission, _document {$tenantColumn})
+                    {$this->getInsertKeyword()} {$this->getSQLTable($name . '_perms')} (_type, _permission, _column, _document {$tenantColumn})
                     VALUES {$permissions}
                     {$this->getInsertPermissionsSuffix()}
                 ";
@@ -2836,35 +3041,51 @@ public function upsertDocuments(
                 $old = $change->getOld();
                 $document = $change->getNew();
 
+                // Permissions are compared as role\0column, so a permission that only
+                // changes which column it is scoped to still registers as a change.
+                $flatten = fn (Document $doc, string $type): array => \array_map(
+                    fn (array $permission) => $permission['role'] . "\0" . $permission['column'],
+                    $doc->getPermissionsByTypeWithColumns($type)
+                );
+
                 $current = [];
+                $desired = [];
                 foreach (Database::PERMISSIONS as $type) {
-                    $current[$type] = $old->getPermissionsByType($type);
+                    $current[$type] = $flatten($old, $type);
+                    $desired[$type] = $flatten($document, $type);
                 }
 
                 foreach (Database::PERMISSIONS as $type) {
-                    $toRemove = \array_diff($current[$type], $document->getPermissionsByType($type));
+                    $toRemove = \array_diff($current[$type], $desired[$type]);
                     if (!empty($toRemove)) {
+                        $pairs = [];
+                        foreach (\array_keys($toRemove) as $i) {
+                            [$role, $column] = \explode("\0", $toRemove[$i], 2);
+                            $pairs[] = "(_permission = :remove_{$type}_{$index}_{$i} AND _column = :removecol_{$type}_{$index}_{$i})";
+                            $removeBindValues[":remove_{$type}_{$index}_{$i}"] = $role;
+                            $removeBindValues[":removecol_{$type}_{$index}_{$i}"] = $column;
+                        }
+
                         $removeQueries[] = "(
                             _document = :_uid_{$index}
                             " . ($this->sharedTables ? " AND _tenant = :_tenant_{$index}" : '') . "
                             AND _type = '{$type}'
-                            AND _permission IN (" . \implode(',', \array_map(fn ($i) => ":remove_{$type}_{$index}_{$i}", \array_keys($toRemove))) . ")
+                            AND (" . \implode(' OR ', $pairs) . ")
                         )";
                         $removeBindValues[":_uid_{$index}"] = $document->getId();
                         if ($this->sharedTables) {
                             $removeBindValues[":_tenant_{$index}"] = $document->getTenant();
                         }
-                        foreach ($toRemove as $i => $perm) {
-                            $removeBindValues[":remove_{$type}_{$index}_{$i}"] = $perm;
-                        }
                     }
                 }
 
                 foreach (Database::PERMISSIONS as $type) {
-                    $toAdd = \array_diff($document->getPermissionsByType($type), $current[$type]);
+                    $toAdd = \array_diff($desired[$type], $current[$type]);
 
                     foreach ($toAdd as $i => $permission) {
-                        $addQuery = "(:_uid_{$index}, '{$type}', :add_{$type}_{$index}_{$i}";
+                        [$role, $column] = \explode("\0", $permission, 2);
+
+                        $addQuery = "(:_uid_{$index}, '{$type}', :add_{$type}_{$index}_{$i}, :addcol_{$type}_{$index}_{$i}";
 
                         if ($this->sharedTables) {
                             $addQuery .= ", :_tenant_{$index}";
@@ -2873,7 +3094,8 @@ public function upsertDocuments(
                         $addQuery .= ")";
                         $addQueries[] = $addQuery;
                         $addBindValues[":_uid_{$index}"] = $document->getId();
-                        $addBindValues[":add_{$type}_{$index}_{$i}"] = $permission;
+                        $addBindValues[":add_{$type}_{$index}_{$i}"] = $role;
+                        $addBindValues[":addcol_{$type}_{$index}_{$i}"] = $column;
 
                         if ($this->sharedTables) {
                             $addBindValues[":_tenant_{$index}"] = $document->getTenant();
@@ -2892,7 +3114,7 @@ public function upsertDocuments(
             }
 
             if (!empty($addQueries)) {
-                $sqlAddPermissions = "INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission";
+                $sqlAddPermissions = "INSERT INTO {$this->getSQLTable($name . '_perms')} (_document, _type, _permission, _column";
                 if ($this->sharedTables) {
                     $sqlAddPermissions .= ", _tenant";
                 }
diff --git a/src/Database/Adapter/SQLite.php b/src/Database/Adapter/SQLite.php
index 3880aec167..da66094d53 100644
--- a/src/Database/Adapter/SQLite.php
+++ b/src/Database/Adapter/SQLite.php
@@ -421,6 +421,7 @@ public function createCollection(string $name, array $attributes = [], array $in
 				{$tenantQuery}
 				`_type` VARCHAR(12) NOT NULL,
 				`_permission` VARCHAR(255) NOT NULL,
+				`_column` VARCHAR(255) NOT NULL DEFAULT '',
 				`_document` VARCHAR(255) NOT NULL
 			)
 		";
@@ -440,7 +441,7 @@ public function createCollection(string $name, array $attributes = [], array $in
             $this->createIndex($id, '_created_at', Database::INDEX_KEY, [ '_createdAt'], [], []);
             $this->createIndex($id, '_updated_at', Database::INDEX_KEY, [ '_updatedAt'], [], []);
 
-            $this->createIndex("{$id}_perms", '_index_1', Database::INDEX_UNIQUE, ['_document', '_type', '_permission'], [], []);
+            $this->createIndex("{$id}_perms", '_index_1', Database::INDEX_UNIQUE, ['_document', '_type', '_permission', '_column'], [], []);
             $this->createIndex("{$id}_perms", '_index_2', Database::INDEX_KEY, ['_permission', '_type'], [], []);
 
             if ($this->sharedTables) {
@@ -1206,11 +1207,14 @@ public function createDocument(Document $collection, Document $document): Docume
         }
 
         $permissions = [];
+        $permissionBinds = [];
         foreach (Database::PERMISSIONS as $type) {
-            foreach ($document->getPermissionsByType($type) as $permission) {
-                $permission = \str_replace('"', '', $permission);
+            foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
+                $role = \str_replace('"', '', $permission['role']);
                 $tenantQuery = $this->sharedTables ? ', :_tenant' : '';
-                $permissions[] = "('{$type}', '{$permission}', '{$document->getId()}' {$tenantQuery})";
+                $columnBind = ":_column_{$type}_{$i}";
+                $permissionBinds[$columnBind] = $permission['column'];
+                $permissions[] = "('{$type}', '{$role}', {$columnBind}, '{$document->getId()}' {$tenantQuery})";
             }
         }
 
@@ -1218,13 +1222,17 @@ public function createDocument(Document $collection, Document $document): Docume
             $tenantQuery = $this->sharedTables ? ', _tenant' : '';
 
             $queryPermissions = "
-				INSERT INTO `{$this->getNamespace()}_{$name}_perms` (_type, _permission, _document {$tenantQuery})
+				INSERT INTO `{$this->getNamespace()}_{$name}_perms` (_type, _permission, _column, _document {$tenantQuery})
 				VALUES " . \implode(', ', $permissions);
 
             $queryPermissions = $this->trigger(Database::EVENT_PERMISSIONS_CREATE, $queryPermissions);
 
             $stmtPermissions = $this->getPDO()->prepare($queryPermissions);
 
+            foreach ($permissionBinds as $key => $value) {
+                $stmtPermissions->bindValue($key, $value);
+            }
+
             if ($this->sharedTables) {
                 $stmtPermissions->bindValue(':_tenant', $this->tenant);
             }
@@ -1295,10 +1303,11 @@ public function updateDocument(Document $collection, string $id, Document $docum
             $values = [];
             $binds = [];
             foreach (Database::PERMISSIONS as $type) {
-                foreach ($document->getPermissionsByType($type) as $i => $permission) {
+                foreach ($document->getPermissionsByTypeWithColumns($type) as $i => $permission) {
                     $tenantQuery = $this->sharedTables ? ', :_tenant' : '';
-                    $values[] = "(:_uid, '{$type}', :_add_{$type}_{$i} {$tenantQuery})";
-                    $binds[":_add_{$type}_{$i}"] = $permission;
+                    $values[] = "(:_uid, '{$type}', :_add_{$type}_{$i}, :_addcol_{$type}_{$i} {$tenantQuery})";
+                    $binds[":_add_{$type}_{$i}"] = $permission['role'];
+                    $binds[":_addcol_{$type}_{$i}"] = $permission['column'];
                 }
             }
 
@@ -1306,7 +1315,7 @@ public function updateDocument(Document $collection, string $id, Document $docum
                 $tenantQuery = $this->sharedTables ? ', _tenant' : '';
 
                 $sql = "
-			   INSERT INTO `{$this->getNamespace()}_{$name}_perms` (_document, _type, _permission {$tenantQuery})
+			   INSERT INTO `{$this->getNamespace()}_{$name}_perms` (_document, _type, _permission, _column {$tenantQuery})
 			   VALUES " . \implode(', ', $values);
 
                 $sql = $this->trigger(Database::EVENT_PERMISSIONS_CREATE, $sql);
@@ -1513,6 +1522,11 @@ public function getSupportForGetConnectionId(): bool
      *
      * @return bool
      */
+    public function getSupportForColumnPermissions(): bool
+    {
+        return true;
+    }
+
     public function getSupportForSchemaAttributes(): bool
     {
         return true;
diff --git a/src/Database/Database.php b/src/Database/Database.php
index 1444286658..f2ff23a5f5 100644
--- a/src/Database/Database.php
+++ b/src/Database/Database.php
@@ -419,6 +419,11 @@ class Database
 
     protected bool $inBatchRelationshipPopulation = false;
 
+    /**
+     * Suppresses column masking for reads the library performs on its own behalf.
+     */
+    protected bool $skipColumnMasking = false;
+
     protected bool $filter = true;
 
     /**
@@ -3293,6 +3298,24 @@ public function updateAttribute(string $collection, string $id, ?string $type =
             if (!$updated) {
                 throw new DatabaseException('Failed to update attribute');
             }
+
+            // A column-scoped permission names its column, in _perms._column and
+            // again inside the $permissions JSON, so a rename has to repoint both.
+            // There is no stable column id to hang permissions off: this method
+            // rewrites the attribute's '$id' and 'key' together, so the key is the
+            // only handle there is. The lookup is skipped entirely when no
+            // permission is scoped to this column, which is the common case.
+            if (
+                !\is_null($newKey)
+                && $newKey !== $id
+                && $this->adapter->getSupportForColumnPermissions()
+            ) {
+                $repointed = $this->adapter->renameColumnPermissions($collectionDoc, $id, $newKey);
+
+                foreach ($repointed as $documentId) {
+                    $this->purgeCachedDocument($collection, $documentId);
+                }
+            }
         }
 
         $collectionDoc->setAttribute('attributes', $attributes);
@@ -3441,6 +3464,16 @@ public function deleteAttribute(string $collection, string $id): bool
             // Ignore
         }
 
+        // Permissions name their column by key, so grants left behind would be
+        // inherited by any column later created under the same name.
+        if ($this->adapter->getSupportForColumnPermissions()) {
+            $cleaned = $this->adapter->deleteColumnPermissions($collection, $id);
+
+            foreach ($cleaned as $documentId) {
+                $this->purgeCachedDocument($collection->getId(), $documentId);
+            }
+        }
+
         $this->updateMetadata(
             collection: $collection,
             rollbackOperation: fn () => $this->adapter->createAttribute(
@@ -4984,6 +5017,8 @@ public function getDocument(string $collection, string $id, array $queries = [],
                 }
             }
 
+            $document = $this->maskUnreadableColumns($collection, $document, $documentSecurity);
+
             $this->trigger(self::EVENT_DOCUMENT_READ, $document);
 
             if ($this->isTtlExpired($collection, $document)) {
@@ -5077,11 +5112,363 @@ public function getDocument(string $collection, string $id, array $queries = [],
             }
         }
 
+        $document = $this->maskUnreadableColumns($collection, $document, $documentSecurity);
+
         $this->trigger(self::EVENT_DOCUMENT_READ, $document);
 
         return $document;
     }
 
+    /**
+     * Columns the current roles hold the given permission on, or null when they
+     * hold it on every column.
+     *
+     * Resolved entirely from permissions that already travel with the document,
+     * so this costs no extra query.
+     *
+     * @param Document $collection
+     * @param Document $document
+     * @param bool $documentSecurity
+     * @param string $type
+     * @return array<string>|null
+     */
+    private function getPermittedColumns(
+        Document $collection,
+        Document $document,
+        bool $documentSecurity,
+        string $type
+    ): ?array {
+        if (!$this->authorization->getStatus()) {
+            return null;
+        }
+
+        $permissions = $collection->getPermissionsByTypeWithColumns($type);
+
+        if ($documentSecurity) {
+            $permissions = [
+                ...$permissions,
+                ...$document->getPermissionsByTypeWithColumns($type),
+            ];
+        }
+
+        $columns = [];
+
+        foreach ($permissions as $permission) {
+            if (!$this->authorization->hasRole($permission['role'])) {
+                continue;
+            }
+
+            // An unscoped permission covers every column, so there is nothing to scope.
+            if ($permission['column'] === Permission::COLUMN_ALL) {
+                return null;
+            }
+
+            $columns[$permission['column']] = true;
+        }
+
+        return \array_keys($columns);
+    }
+
+    /**
+     * Run a callback with column masking suppressed.
+     *
+     * Internal reads need the stored document, not the caller's view of it: they feed
+     * permission comparisons and merges, so a masked copy would make the library
+     * delete the columns and grants the caller was never shown.
+     *
+     * @template T
+     * @param callable(): T $callback
+     * @return T
+     */
+    private function unmasked(callable $callback): mixed
+    {
+        $previous = $this->skipColumnMasking;
+        $this->skipColumnMasking = true;
+
+        try {
+            return $callback();
+        } finally {
+            $this->skipColumnMasking = $previous;
+        }
+    }
+
+    /**
+     * The columns the current roles are demonstrably limited to at collection level,
+     * or null when no restriction can be proven.
+     *
+     * Returns null in two different situations, both meaning "do not restrict":
+     * an unscoped grant (every column is allowed), and no collection-level grant at
+     * all (access comes from per-document permissions, which cannot be bounded before
+     * the rows are read). A non-empty list means column-level permissions are
+     * demonstrably in play for this caller.
+     *
+     * @param Document $collection
+     * @param string $type
+     * @return array<string>|null
+     */
+    private function getCollectionColumnRestriction(Document $collection, string $type): ?array
+    {
+        if (!$this->authorization->getStatus()) {
+            return null;
+        }
+
+        $columns = [];
+
+        foreach ($collection->getPermissionsByTypeWithColumns($type) as $permission) {
+            if (!$this->authorization->hasRole($permission['role'])) {
+                continue;
+            }
+
+            if ($permission['column'] === Permission::COLUMN_ALL) {
+                return null;
+            }
+
+            $columns[$permission['column']] = true;
+        }
+
+        return empty($columns) ? null : \array_keys($columns);
+    }
+
+    /**
+     * Reject a query that reads a column the current roles are restricted from.
+     *
+     * Masking removes a column from the response, but a filter, an order or a select
+     * still reaches it: filtering on a hidden column turns the result set into an
+     * oracle for its value, and ordering by one reveals the ranking. A column the
+     * caller cannot read is treated as a column that does not exist for them, which
+     * is what query validation already does for unknown columns.
+     *
+     * @param Document $collection
+     * @param array<Query> $queries
+     * @param string $type
+     * @return void
+     * @throws AuthorizationException
+     */
+    private function assertColumnsQueryable(Document $collection, array $queries, string $type = self::PERMISSION_READ): void
+    {
+        $restriction = $this->getCollectionColumnRestriction($collection, $type);
+
+        if ($restriction === null) {
+            return;
+        }
+
+        foreach ($queries as $query) {
+            $keys = $query->getMethod() === Query::TYPE_SELECT
+                ? $query->getValues()
+                : [$query->getAttribute()];
+
+            foreach ($keys as $key) {
+                // Internal fields are not columns; dotted keys are relationship paths.
+                if (!\is_string($key) || $key === '' || \str_starts_with($key, '$') || \str_contains($key, '.')) {
+                    continue;
+                }
+
+                if (!\in_array($key, $restriction, true)) {
+                    throw new AuthorizationException('Missing "' . $type . '" permission for column "' . $key . '".');
+                }
+            }
+        }
+    }
+
+    /**
+     * Reject a write that touches a column the current roles are restricted from at
+     * collection level.
+     *
+     * Used where no stored document is available to consult: a create (the row does
+     * not exist yet) and a bulk update (one set of changes applied to many rows).
+     *
+     * @param Document $collection
+     * @param Document $document
+     * @param string $type
+     * @return void
+     * @throws AuthorizationException
+     */
+    private function assertColumnsAllowed(Document $collection, Document $document, string $type): void
+    {
+        $columns = $this->getCollectionColumnRestriction($collection, $type);
+
+        if ($columns === null) {
+            return;
+        }
+
+        $relationships = [];
+        foreach ($collection->getAttribute('attributes', []) as $attribute) {
+            if ($attribute['type'] === self::VAR_RELATIONSHIP) {
+                $relationships[$attribute['key']] = true;
+            }
+        }
+
+        foreach ($document as $key => $value) {
+            if (\str_starts_with($key, '$') || isset($relationships[$key])) {
+                continue;
+            }
+
+            if (\is_null($value)) {
+                continue;
+            }
+
+            if (!\in_array($key, $columns, true)) {
+                throw new AuthorizationException('Missing "' . $type . '" permission for column "' . $key . '".');
+            }
+        }
+    }
+
+    /**
+     * Reject an update that changes a column the current roles cannot update.
+     *
+     * Returns immediately unless a column-scoped update permission is what
+     * granted this write, so the ordinary path pays only for resolving the
+     * permission list already loaded with the document.
+     *
+     * @param Document $collection
+     * @param Document $old stored document, whose permissions govern the write
+     * @param Document $document merged new state
+     * @param bool $documentSecurity
+     * @return void
+     * @throws AuthorizationException
+     */
+    private function assertColumnsWritable(
+        Document $collection,
+        Document $old,
+        Document $document,
+        bool $documentSecurity
+    ): void {
+        $columns = $this->getPermittedColumns($collection, $old, $documentSecurity, self::PERMISSION_UPDATE);
+
+        if ($columns === null) {
+            return;
+        }
+
+        $relationships = [];
+        foreach ($collection->getAttribute('attributes', []) as $attribute) {
+            if ($attribute['type'] === self::VAR_RELATIONSHIP) {
+                $relationships[$attribute['key']] = true;
+            }
+        }
+
+        foreach ($document as $key => $value) {
+            // Internal fields are not columns. $permissions is deliberately not
+            // column-scoped: rewriting permissions stays a document-level right.
+            if (\str_starts_with($key, '$')) {
+                continue;
+            }
+
+            if (\in_array($key, $columns, true) || isset($relationships[$key])) {
+                continue;
+            }
+
+            $changed = Operator::isOperator($value) || !self::valuesEqual($value, $old->getAttribute($key));
+
+            if ($changed) {
+                throw new AuthorizationException('Missing "update" permission for column "' . $key . '".');
+            }
+        }
+    }
+
+    /**
+     * Strip columns the current roles cannot read.
+     *
+     * Must run *after* the document has been written to cache. The cache is shared
+     * across roles, so caching a masked copy would serve one role's view to another.
+     *
+     * @param Document $collection
+     * @param Document $document
+     * @param bool $documentSecurity
+     * @return Document
+     */
+    private function maskUnreadableColumns(Document $collection, Document $document, bool $documentSecurity): Document
+    {
+        if ($this->skipColumnMasking || $document->isEmpty() || $collection->getId() === self::METADATA) {
+            return $document;
+        }
+
+        $columns = $this->getPermittedColumns($collection, $document, $documentSecurity, self::PERMISSION_READ);
+
+        if ($columns === null) {
+            return $document;
+        }
+
+        $document = clone $document;
+
+        foreach (\array_keys($document->getArrayCopy()) as $key) {
+            // Internal fields ($id, $createdAt, $permissions, ...) are not columns.
+            if (\str_starts_with($key, '$')) {
+                continue;
+            }
+
+            if (!\in_array($key, $columns, true)) {
+                $document->removeAttribute($key);
+            }
+        }
+
+        // Permission strings name their column, so returning them whole would
+        // disclose the names of columns this caller cannot read. What is hidden here
+        // is restored by preserveHiddenPermissions() if the document is written back.
+        $permissions = [];
+
+        foreach ($document->getPermissions() as $permission) {
+            $parsed = Permission::parse($permission);
+
+            if ($parsed->isForAllColumns() || \in_array($parsed->getColumn(), $columns, true)) {
+                $permissions[] = $permission;
+            }
+        }
+
+        $document->setAttribute('$permissions', $permissions);
+
+        return $document;
+    }
+
+    /**
+     * Put back the permissions a caller was never allowed to see.
+     *
+     * maskUnreadableColumns() strips permissions scoped to columns the caller cannot
+     * read, so a client that reads a document and writes it back would otherwise
+     * delete grants it never received. Runs before change detection, so echoing a
+     * masked document back is correctly seen as no permission change at all.
+     *
+     * @param Document $collection
+     * @param Document $old unmasked stored document
+     * @param Document $document incoming document
+     * @param bool $documentSecurity
+     * @return void
+     */
+    private function preserveHiddenPermissions(
+        Document $collection,
+        Document $old,
+        Document $document,
+        bool $documentSecurity
+    ): void {
+        if (!$document->offsetExists('$permissions')) {
+            return;
+        }
+
+        $columns = $this->getPermittedColumns($collection, $old, $documentSecurity, self::PERMISSION_READ);
+
+        if ($columns === null) {
+            return;
+        }
+
+        $hidden = [];
+
+        foreach ($old->getPermissions() as $permission) {
+            $parsed = Permission::parse($permission);
+
+            if (!$parsed->isForAllColumns() && !\in_array($parsed->getColumn(), $columns, true)) {
+                $hidden[] = $permission;
+            }
+        }
+
+        if (empty($hidden)) {
+            return;
+        }
+
+        $document->setAttribute('$permissions', \array_values(\array_unique([
+            ...$document->getPermissions(),
+            ...$hidden,
+        ])));
+    }
+
     private function isTtlExpired(Document $collection, Document $document): bool
     {
         if (!$this->adapter->getSupportForTTLIndexes()) {
@@ -5740,6 +6127,8 @@ public function createDocument(string $collection, Document $document): Document
             if (!$isValid) {
                 throw new AuthorizationException($this->authorization->getDescription());
             }
+
+            $this->assertColumnsAllowed($collection, $document, self::PERMISSION_CREATE);
         }
 
         $time = DateTime::now();
@@ -5862,6 +6251,10 @@ public function createDocuments(
             if (!$this->authorization->isValid(new Input(self::PERMISSION_CREATE, $collection->getCreate()))) {
                 throw new AuthorizationException($this->authorization->getDescription());
             }
+
+            foreach ($documents as $document) {
+                $this->assertColumnsAllowed($collection, $document, self::PERMISSION_CREATE);
+            }
         }
 
         $time = DateTime::now();
@@ -6314,6 +6707,13 @@ public function updateDocument(string $collection, string $id, Document $documen
                 return new Document();
             }
 
+            $this->preserveHiddenPermissions(
+                $collection,
+                $old,
+                $document,
+                $collection->getAttribute('documentSecurity', false)
+            );
+
             $skipPermissionsUpdate = true;
 
             if ($document->offsetExists('$permissions')) {
@@ -6473,6 +6873,8 @@ public function updateDocument(string $collection, string $id, Document $documen
                     if (!$this->authorization->isValid(new Input(self::PERMISSION_UPDATE, $updatePermissions))) {
                         throw new AuthorizationException($this->authorization->getDescription());
                     }
+
+                    $this->assertColumnsWritable($collection, $old, $document, $documentSecurity);
                 } else {
                     if (!$this->authorization->isValid(new Input(self::PERMISSION_READ, $readPermissions))) {
                         throw new AuthorizationException($this->authorization->getDescription());
@@ -6508,7 +6910,7 @@ public function updateDocument(string $collection, string $id, Document $documen
             }
 
             if ($this->resolveRelationships) {
-                $document = $this->silent(fn () => $this->updateDocumentRelationships($collection, $old, $document));
+                $document = $this->unmasked(fn () => $this->silent(fn () => $this->updateDocumentRelationships($collection, $old, $document)));
             }
 
             $document = $this->adapter->castingBefore($collection, $document);
@@ -6615,6 +7017,11 @@ public function updateDocuments(
             throw new AuthorizationException($this->authorization->getDescription());
         }
 
+        if ($collection->getId() !== self::METADATA) {
+            $this->assertColumnsAllowed($collection, $updates, self::PERMISSION_UPDATE);
+            $this->assertColumnsQueryable($collection, $queries);
+        }
+
         $attributes = $collection->getAttribute('attributes', []);
         $indexes = $collection->getAttribute('indexes', []);
 
@@ -6703,11 +7110,11 @@ public function updateDocuments(
                 $new[] = Query::cursorAfter($last);
             }
 
-            $batch = $this->silent(fn () => $this->find(
+            $batch = $this->unmasked(fn () => $this->silent(fn () => $this->find(
                 $collection->getId(),
                 array_merge($new, $queries),
                 forPermission: Database::PERMISSION_UPDATE
-            ));
+            )));
 
             if (empty($batch)) {
                 break;
@@ -6717,7 +7124,7 @@ public function updateDocuments(
             $currentPermissions = $updates->getPermissions();
             sort($currentPermissions);
 
-            $this->withTransaction(function () use ($collection, $updates, &$batch, $currentPermissions) {
+            $this->withTransaction(function () use ($collection, $updates, &$batch, $currentPermissions, $documentSecurity) {
                 foreach ($batch as $index => $document) {
                     $skipPermissionsUpdate = true;
 
@@ -6737,8 +7144,12 @@ public function updateDocuments(
 
                     $new = new Document(\array_merge($document->getArrayCopy(), $updates->getArrayCopy()));
 
+                    // Per document: the collection-level check cannot see grants that
+                    // individual rows add, so each row is verified against its own.
+                    $this->assertColumnsWritable($collection, $document, $new, $documentSecurity);
+
                     if ($this->resolveRelationships) {
-                        $this->silent(fn () => $this->updateDocumentRelationships($collection, $document, $new));
+                        $this->unmasked(fn () => $this->silent(fn () => $this->updateDocumentRelationships($collection, $document, $new)));
                     }
 
                     $document = $new;
@@ -7510,11 +7921,17 @@ public function upsertDocumentsWithIncrease(
                 if (!$this->authorization->isValid(new Input(self::PERMISSION_CREATE, $collection->getCreate()))) {
                     throw new AuthorizationException($this->authorization->getDescription());
                 }
-            } elseif (!$this->authorization->isValid(new Input(self::PERMISSION_UPDATE, [
-                ...$collection->getUpdate(),
-                ...($documentSecurity ? $old->getUpdate() : [])
-            ]))) {
-                throw new AuthorizationException($this->authorization->getDescription());
+
+                $this->assertColumnsAllowed($collection, $document, self::PERMISSION_CREATE);
+            } else {
+                if (!$this->authorization->isValid(new Input(self::PERMISSION_UPDATE, [
+                    ...$collection->getUpdate(),
+                    ...($documentSecurity ? $old->getUpdate() : [])
+                ]))) {
+                    throw new AuthorizationException($this->authorization->getDescription());
+                }
+
+                $this->assertColumnsWritable($collection, $old, $document, $documentSecurity);
             }
 
             $updatedAt = $document->getUpdatedAt();
@@ -7767,6 +8184,14 @@ public function increaseDocumentAttribute(
                 ]))) {
                     throw new AuthorizationException($this->authorization->getDescription());
                 }
+
+                // This writes one named column, so it needs update permission on that
+                // column specifically. Without this it bypasses the column gate.
+                $columns = $this->getPermittedColumns($collection, $document, $documentSecurity, self::PERMISSION_UPDATE);
+
+                if ($columns !== null && !\in_array($attribute, $columns, true)) {
+                    throw new AuthorizationException('Missing "update" permission for column "' . $attribute . '".');
+                }
             }
 
             if (!\is_null($max) && ($document->getAttribute($attribute) + $value > $max)) {
@@ -7868,6 +8293,14 @@ public function decreaseDocumentAttribute(
                 ]))) {
                     throw new AuthorizationException($this->authorization->getDescription());
                 }
+
+                // This writes one named column, so it needs update permission on that
+                // column specifically. Without this it bypasses the column gate.
+                $columns = $this->getPermittedColumns($collection, $document, $documentSecurity, self::PERMISSION_UPDATE);
+
+                if ($columns !== null && !\in_array($attribute, $columns, true)) {
+                    throw new AuthorizationException('Missing "update" permission for column "' . $attribute . '".');
+                }
             }
 
             if (!\is_null($min) && ($document->getAttribute($attribute) - $value < $min)) {
@@ -8002,10 +8435,10 @@ private function deleteDocumentRelationships(Document $collection, Document $doc
 
             switch ($onDelete) {
                 case Database::RELATION_MUTATE_RESTRICT:
-                    $this->deleteRestrict($relatedCollection, $document, $value, $relationType, $twoWay, $twoWayKey, $side);
+                    $this->unmasked(fn () => $this->deleteRestrict($relatedCollection, $document, $value, $relationType, $twoWay, $twoWayKey, $side));
                     break;
                 case Database::RELATION_MUTATE_SET_NULL:
-                    $this->deleteSetNull($collection, $relatedCollection, $document, $value, $relationType, $twoWay, $twoWayKey, $side);
+                    $this->unmasked(fn () => $this->deleteSetNull($collection, $relatedCollection, $document, $value, $relationType, $twoWay, $twoWayKey, $side));
                     break;
                 case Database::RELATION_MUTATE_CASCADE:
                     foreach ($this->relationshipDeleteStack as $processedRelationship) {
@@ -8052,7 +8485,7 @@ private function deleteDocumentRelationships(Document $collection, Document $doc
                             break 2;
                         }
                     }
-                    $this->deleteCascade($collection, $relatedCollection, $document, $key, $value, $relationType, $twoWayKey, $side, $relationship);
+                    $this->unmasked(fn () => $this->deleteCascade($collection, $relatedCollection, $document, $key, $value, $relationType, $twoWayKey, $side, $relationship));
                     break;
             }
         }
@@ -8443,11 +8876,11 @@ public function deleteDocuments(
             /**
              * @var array<Document> $batch
              */
-            $batch = $this->silent(fn () => $this->find(
+            $batch = $this->unmasked(fn () => $this->silent(fn () => $this->find(
                 $collection->getId(),
                 array_merge($new, $queries),
                 forPermission: Database::PERMISSION_DELETE
-            ));
+            )));
 
             if (empty($batch)) {
                 break;
@@ -8657,6 +9090,8 @@ public function find(string $collection, array $queries = [], string $forPermiss
             throw new AuthorizationException($this->authorization->getDescription());
         }
 
+        $this->assertColumnsQueryable($collection, $queries, $forPermission);
+
         $relationships = \array_filter(
             $collection->getAttribute('attributes', []),
             fn (Document $attribute) => $attribute->getAttribute('type') === self::VAR_RELATIONSHIP
@@ -8780,6 +9215,12 @@ public function find(string $collection, array $queries = [], string $forPermiss
                 $node->setAttribute('$collection', $collection->getId());
             }
 
+            // Not gated on $skipAuth: that flag only means the caller may see every
+            // ROW (it is set by any collection-level read, including a column-scoped
+            // one), so it says nothing about which columns are readable. Masking is
+            // already a no-op when authorization is disabled.
+            $node = $this->maskUnreadableColumns($collection, $node, $documentSecurity);
+
             $results[$index] = $node;
         }
 
@@ -9174,6 +9615,8 @@ public function count(string $collection, array $queries = [], ?int $max = null)
             throw new AuthorizationException($this->authorization->getDescription());
         }
 
+        $this->assertColumnsQueryable($collection, $queries);
+
         $relationships = \array_filter(
             $collection->getAttribute('attributes', []),
             fn (Document $attribute) => $attribute->getAttribute('type') === self::VAR_RELATIONSHIP
@@ -9248,6 +9691,16 @@ public function sum(string $collection, string $attribute, array $queries = [],
             throw new AuthorizationException($this->authorization->getDescription());
         }
 
+        $this->assertColumnsQueryable($collection, $queries);
+
+        // The aggregated column itself is read, so it needs the same permission a
+        // filter on it would. Without this, sum() extracts a masked column in one call.
+        $columns = $this->getCollectionColumnRestriction($collection, self::PERMISSION_READ);
+
+        if ($columns !== null && !\in_array($attribute, $columns, true)) {
+            throw new AuthorizationException('Missing "read" permission for column "' . $attribute . '".');
+        }
+
         $relationships = \array_filter(
             $collection->getAttribute('attributes', []),
             fn (Document $attribute) => $attribute->getAttribute('type') === self::VAR_RELATIONSHIP
diff --git a/src/Database/Document.php b/src/Database/Document.php
index 73bd458cd5..cd624d0ad9 100644
--- a/src/Database/Document.php
+++ b/src/Database/Document.php
@@ -5,6 +5,7 @@
 use ArrayObject;
 use Utopia\Database\Exception as DatabaseException;
 use Utopia\Database\Exception\Structure as StructureException;
+use Utopia\Database\Helpers\Permission;
 
 /**
  * @extends ArrayObject<string, mixed>
@@ -150,14 +151,46 @@ public function getPermissionsByType(string $type): array
     {
         $typePermissions = [];
 
+        foreach ($this->getPermissionsByTypeWithColumns($type) as $permission) {
+            $typePermissions[] = $permission['role'];
+        }
+
+        return \array_unique($typePermissions);
+    }
+
+    /**
+     * Permissions of the given type, split into the role and the column it is scoped to.
+     *
+     * A column of Permission::COLUMN_ALL means the role is granted every column,
+     * which is how every permission written before column-level permissions reads.
+     *
+     * @param string $type
+     * @return array<array{role: string, column: string}>
+     */
+    public function getPermissionsByTypeWithColumns(string $type): array
+    {
+        $typePermissions = [];
+
         foreach ($this->getPermissions() as $permission) {
             if (!\str_starts_with($permission, $type)) {
                 continue;
             }
-            $typePermissions[] = \str_replace([$type . '(', ')', '"', ' '], '', $permission);
+
+            $column = Permission::COLUMN_ALL;
+
+            // Peel off an optional second argument: type("role", "column").
+            if (\preg_match('/^(.*?)\s*,\s*"([^"]*)"\)$/', $permission, $matches) === 1) {
+                $permission = $matches[1] . ')';
+                $column = $matches[2];
+            }
+
+            $typePermissions[] = [
+                'role' => \str_replace([$type . '(', ')', '"', ' '], '', $permission),
+                'column' => $column,
+            ];
         }
 
-        return \array_unique($typePermissions);
+        return $typePermissions;
     }
 
     /**
diff --git a/src/Database/Helpers/Permission.php b/src/Database/Helpers/Permission.php
index 18c4fe5a94..7801f922a3 100644
--- a/src/Database/Helpers/Permission.php
+++ b/src/Database/Helpers/Permission.php
@@ -8,6 +8,15 @@
 
 class Permission
 {
+    /**
+     * Sentinel column value meaning "every column".
+     *
+     * Stored as an empty string rather than NULL: MySQL and MariaDB treat NULLs
+     * as distinct in a UNIQUE index, so a nullable _column would let duplicate
+     * permission rows through the _perms uniqueness guarantee.
+     */
+    public const COLUMN_ALL = '';
+
     private Role $role;
 
     /**
@@ -26,6 +35,7 @@ public function __construct(
         string $role,
         string $identifier = '',
         string $dimension = '',
+        private string $column = self::COLUMN_ALL,
     ) {
         $this->role = new Role($role, $identifier, $dimension);
     }
@@ -37,7 +47,31 @@ public function __construct(
      */
     public function toString(): string
     {
-        return $this->permission . '("' . $this->role->toString() . '")';
+        $permission = $this->permission . '("' . $this->role->toString() . '"';
+
+        if ($this->column !== self::COLUMN_ALL) {
+            $permission .= ', "' . $this->column . '"';
+        }
+
+        return $permission . ')';
+    }
+
+    /**
+     * The column this permission is scoped to, or COLUMN_ALL for every column.
+     *
+     * @return string
+     */
+    public function getColumn(): string
+    {
+        return $this->column;
+    }
+
+    /**
+     * @return bool
+     */
+    public function isForAllColumns(): bool
+    {
+        return $this->column === self::COLUMN_ALL;
     }
 
     /**
@@ -82,6 +116,24 @@ public function getDimension(): string
      */
     public static function parse(string $permission): self
     {
+        $column = self::COLUMN_ALL;
+
+        // Peel off an optional second argument: type("role", "column").
+        // Role identifiers and dimensions never contain a comma, so the lazy
+        // match cannot swallow part of the role.
+        if (\preg_match('/^(.*?)\s*,\s*"([^"]*)"\)$/', $permission, $matches) === 1) {
+            $permission = $matches[1] . ')';
+            $column = $matches[2];
+
+            if ($column === self::COLUMN_ALL) {
+                throw new DatabaseException('Column must not be empty. Omit the argument to grant every column.');
+            }
+
+            if ($column === '*') {
+                throw new DatabaseException('Wildcard column "*" is not supported. Omit the argument to grant every column.');
+            }
+        }
+
         $permissionParts = \explode('("', $permission);
 
         if (\count($permissionParts) !== 2) {
@@ -101,12 +153,12 @@ public static function parse(string $permission): self
         $hasDimension = \str_contains($fullRole, '/');
 
         if (!$hasIdentifier && !$hasDimension) {
-            return new self($permission, $role);
+            return new self($permission, $role, column: $column);
         }
 
         if ($hasIdentifier && !$hasDimension) {
             $identifier = $roleParts[1];
-            return new self($permission, $role, $identifier);
+            return new self($permission, $role, $identifier, column: $column);
         }
 
         if (!$hasIdentifier) {
@@ -121,7 +173,7 @@ public static function parse(string $permission): self
             if (empty($dimension)) {
                 throw new DatabaseException('Dimension must not be empty');
             }
-            return new self($permission, $role, '', $dimension);
+            return new self($permission, $role, '', $dimension, $column);
         }
 
         // Has both identifier and dimension
@@ -137,7 +189,7 @@ public static function parse(string $permission): self
             throw new DatabaseException('Dimension must not be empty');
         }
 
-        return new self($permission, $role, $identifier, $dimension);
+        return new self($permission, $role, $identifier, $dimension, $column);
     }
 
     /**
@@ -169,7 +221,8 @@ public static function aggregate(?array $permissions, array $allowed = Database:
                         $subType,
                         $permission->getRole(),
                         $permission->getIdentifier(),
-                        $permission->getDimension()
+                        $permission->getDimension(),
+                        $permission->getColumn()
                     ))->toString();
                 }
             }
@@ -181,15 +234,17 @@ public static function aggregate(?array $permissions, array $allowed = Database:
      * Create a read permission string from the given Role
      *
      * @param Role $role
+     * @param string $column Restrict to a single column, or COLUMN_ALL for every column
      * @return string
      */
-    public static function read(Role $role): string
+    public static function read(Role $role, string $column = self::COLUMN_ALL): string
     {
         $permission = new self(
             'read',
             $role->getRole(),
             $role->getIdentifier(),
-            $role->getDimension()
+            $role->getDimension(),
+            $column
         );
         return $permission->toString();
     }
@@ -198,15 +253,17 @@ public static function read(Role $role): string
      * Create a create permission string from the given Role
      *
      * @param Role $role
+     * @param string $column Restrict to a single column, or COLUMN_ALL for every column
      * @return string
      */
-    public static function create(Role $role): string
+    public static function create(Role $role, string $column = self::COLUMN_ALL): string
     {
         $permission = new self(
             'create',
             $role->getRole(),
             $role->getIdentifier(),
-            $role->getDimension()
+            $role->getDimension(),
+            $column
         );
         return $permission->toString();
     }
@@ -215,15 +272,17 @@ public static function create(Role $role): string
      * Create an update permission string from the given Role
      *
      * @param Role $role
+     * @param string $column Restrict to a single column, or COLUMN_ALL for every column
      * @return string
      */
-    public static function update(Role $role): string
+    public static function update(Role $role, string $column = self::COLUMN_ALL): string
     {
         $permission = new self(
             'update',
             $role->getRole(),
             $role->getIdentifier(),
-            $role->getDimension()
+            $role->getDimension(),
+            $column
         );
         return $permission->toString();
     }
@@ -232,15 +291,17 @@ public static function update(Role $role): string
      * Create a delete permission string from the given Role
      *
      * @param Role $role
+     * @param string $column Restrict to a single column, or COLUMN_ALL for every column
      * @return string
      */
-    public static function delete(Role $role): string
+    public static function delete(Role $role, string $column = self::COLUMN_ALL): string
     {
         $permission = new self(
             'delete',
             $role->getRole(),
             $role->getIdentifier(),
-            $role->getDimension()
+            $role->getDimension(),
+            $column
         );
         return $permission->toString();
     }
@@ -249,15 +310,17 @@ public static function delete(Role $role): string
      * Create a write permission string from the given Role
      *
      * @param Role $role
+     * @param string $column Restrict to a single column, or COLUMN_ALL for every column
      * @return string
      */
-    public static function write(Role $role): string
+    public static function write(Role $role, string $column = self::COLUMN_ALL): string
     {
         $permission = new self(
             'write',
             $role->getRole(),
             $role->getIdentifier(),
-            $role->getDimension()
+            $role->getDimension(),
+            $column
         );
         return $permission->toString();
     }
diff --git a/src/Database/Validator/Permissions.php b/src/Database/Validator/Permissions.php
index 13e7372050..d03e997741 100644
--- a/src/Database/Validator/Permissions.php
+++ b/src/Database/Validator/Permissions.php
@@ -16,16 +16,26 @@ class Permissions extends Roles
 
     protected int $length;
 
+    /**
+     * @var array<string>
+     */
+    protected array $columns;
+
+    protected Key $key;
+
     /**
      * Permissions constructor.
      *
      * @param int $length maximum amount of permissions. 0 means unlimited.
      * @param array<string> $allowed allowed permissions. Defaults to all available.
+     * @param array<string> $columns known column keys a permission may be scoped to. Empty means any valid key.
      */
-    public function __construct(int $length = 0, array $allowed = [...Database::PERMISSIONS, Database::PERMISSION_WRITE])
+    public function __construct(int $length = 0, array $allowed = [...Database::PERMISSIONS, Database::PERMISSION_WRITE], array $columns = [])
     {
         $this->length = $length;
         $this->allowed = $allowed;
+        $this->columns = $columns;
+        $this->key = new Key();
     }
 
     /**
@@ -96,6 +106,29 @@ public function isValid($permissions): bool
                 return false;
             }
 
+            $column = $permission->getColumn();
+
+            if ($column !== Permission::COLUMN_ALL) {
+                $type = $permission->getPermission();
+
+                // Delete removes the whole row, so scoping it to one column is
+                // meaningless. Write implies delete, so it inherits the same rule.
+                if (\in_array($type, [Database::PERMISSION_DELETE, Database::PERMISSION_WRITE], true)) {
+                    $this->message = 'Permission "' . $type . '" cannot be scoped to a column, it applies to the whole row.';
+                    return false;
+                }
+
+                if (!$this->key->isValid($column)) {
+                    $this->message = 'Column "' . $column . '" is not a valid column key.';
+                    return false;
+                }
+
+                if (!empty($this->columns) && !\in_array($column, $this->columns, true)) {
+                    $this->message = 'Column "' . $column . '" does not exist.';
+                    return false;
+                }
+            }
+
             $role = $permission->getRole();
             $identifier = $permission->getIdentifier();
             $dimension = $permission->getDimension();
diff --git a/tests/unit/ColumnPermissionEnforcementTest.php b/tests/unit/ColumnPermissionEnforcementTest.php
new file mode 100644
index 0000000000..11b8c9ce1a
--- /dev/null
+++ b/tests/unit/ColumnPermissionEnforcementTest.php
@@ -0,0 +1,229 @@
+<?php
+
+namespace Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Cache\Adapter\None as NoCache;
+use Utopia\Cache\Cache;
+use Utopia\Database\Adapter\Memory;
+use Utopia\Database\Database;
+use Utopia\Database\Document;
+use Utopia\Database\Exception\Authorization as AuthorizationException;
+use Utopia\Database\Helpers\Permission;
+use Utopia\Database\Helpers\Role;
+use Utopia\Database\Validator\Authorization;
+
+/**
+ * Column-level permission enforcement.
+ *
+ * Runs on the in-memory adapter: masking and update enforcement are resolved from
+ * the $permissions that already travel with the document, so they need no adapter
+ * support and no running database.
+ */
+class ColumnPermissionEnforcementTest extends TestCase
+{
+    protected Authorization $authorization;
+
+    protected Database $database;
+
+    protected function setUp(): void
+    {
+        $this->authorization = new Authorization();
+
+        $this->database = new Database(new Memory(), new Cache(new NoCache()));
+        $this->database
+            ->setAuthorization($this->authorization)
+            ->setDatabase('columnPermissions')
+            ->setNamespace('cols_' . \uniqid());
+
+        if (!$this->database->exists()) {
+            $this->database->create();
+        }
+
+        $this->authorization->skip(function () {
+            $this->database->createCollection('employees', permissions: [], documentSecurity: true);
+
+            foreach (['name', 'email', 'salary'] as $column) {
+                $this->database->createAttribute('employees', $column, Database::VAR_STRING, 128, false);
+            }
+
+            $this->database->createDocument('employees', new Document([
+                '$id' => 'e1',
+                '$permissions' => [
+                    // Reads and writes only the columns it is granted
+                    Permission::read(Role::user('peer'), 'name'),
+                    Permission::read(Role::user('peer'), 'email'),
+                    Permission::update(Role::user('peer'), 'email'),
+                    // Unscoped, so every column
+                    Permission::read(Role::user('boss')),
+                    Permission::update(Role::user('boss')),
+                ],
+                'name' => 'Bob',
+                'email' => 'bob@example.com',
+                'salary' => '100000',
+            ]));
+        });
+    }
+
+    /**
+     * @return array<string>
+     */
+    private function columnsVisibleTo(string $role): array
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole($role);
+
+        $document = $this->database->getDocument('employees', 'e1');
+
+        return \array_values(\array_filter(
+            \array_keys($document->getArrayCopy()),
+            fn (string $key) => !\str_starts_with($key, '$')
+        ));
+    }
+
+    public function testUnscopedRoleSeesEveryColumn(): void
+    {
+        $this->assertSame(['name', 'email', 'salary'], $this->columnsVisibleTo('user:boss'));
+    }
+
+    public function testColumnScopedRoleSeesOnlyGrantedColumns(): void
+    {
+        $this->assertSame(['name', 'email'], $this->columnsVisibleTo('user:peer'));
+    }
+
+    public function testRoleWithNoReadPermissionSeesNothing(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:stranger');
+
+        $this->assertTrue($this->database->getDocument('employees', 'e1')->isEmpty());
+    }
+
+    public function testFindMasksColumnsToo(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:peer');
+
+        $results = $this->database->find('employees');
+        $this->assertCount(1, $results);
+
+        $columns = \array_values(\array_filter(
+            \array_keys($results[0]->getArrayCopy()),
+            fn (string $key) => !\str_starts_with($key, '$')
+        ));
+
+        $this->assertSame(['name', 'email'], $columns);
+    }
+
+    public function testSkippedAuthorizationIsNotMasked(): void
+    {
+        $columns = $this->authorization->skip(function () {
+            $document = $this->database->getDocument('employees', 'e1');
+
+            return \array_values(\array_filter(
+                \array_keys($document->getArrayCopy()),
+                fn (string $key) => !\str_starts_with($key, '$')
+            ));
+        });
+
+        $this->assertSame(['name', 'email', 'salary'], $columns);
+    }
+
+    /**
+     * A column-scoped grant on the COLLECTION sets $skipAuth, because the roles-only
+     * permission check cannot see the column. That flag means "may see every row",
+     * never "may see every column", so masking must still apply.
+     */
+    public function testCollectionLevelColumnGrantIsStillMaskedInFind(): void
+    {
+        $this->authorization->skip(function () {
+            $this->database->createCollection('public_employees', documentSecurity: true, permissions: [
+                Permission::read(Role::any(), 'name'),
+            ]);
+
+            foreach (['name', 'email', 'salary'] as $column) {
+                $this->database->createAttribute('public_employees', $column, Database::VAR_STRING, 128, false);
+            }
+
+            $this->database->createDocument('public_employees', new Document([
+                '$id' => 'pub1',
+                '$permissions' => [],
+                'name' => 'Bob',
+                'email' => 'bob@example.com',
+                'salary' => '100000',
+            ]));
+        });
+
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('any');
+
+        $results = $this->database->find('public_employees');
+        $this->assertCount(1, $results);
+
+        $columns = \array_values(\array_filter(
+            \array_keys($results[0]->getArrayCopy()),
+            fn (string $key) => !\str_starts_with($key, '$')
+        ));
+
+        $this->assertSame(['name'], $columns, 'find() must mask even when $skipAuth is set');
+    }
+
+    public function testUpdateOfGrantedColumnIsAllowed(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:peer');
+
+        $this->database->updateDocument('employees', 'e1', new Document([
+            'email' => 'new@example.com',
+        ]));
+
+        $stored = $this->authorization->skip(fn () => $this->database->getDocument('employees', 'e1'));
+        $this->assertSame('new@example.com', $stored->getAttribute('email'));
+    }
+
+    public function testUpdateOfUngrantedColumnIsRejected(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:peer');
+
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "update" permission for column "salary"');
+
+        $this->database->updateDocument('employees', 'e1', new Document([
+            'salary' => '999999',
+        ]));
+    }
+
+    public function testUpdateIsRejectedWholesaleWhenOneColumnIsUngranted(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:peer');
+
+        try {
+            $this->database->updateDocument('employees', 'e1', new Document([
+                'email' => 'allowed@example.com',
+                'salary' => '999999',
+            ]));
+            $this->fail('Expected an AuthorizationException');
+        } catch (AuthorizationException) {
+            // The permitted column must not have been written either
+        }
+
+        $stored = $this->authorization->skip(fn () => $this->database->getDocument('employees', 'e1'));
+        $this->assertSame('bob@example.com', $stored->getAttribute('email'));
+        $this->assertSame('100000', $stored->getAttribute('salary'));
+    }
+
+    public function testUnscopedRoleMayUpdateAnyColumn(): void
+    {
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('user:boss');
+
+        $this->database->updateDocument('employees', 'e1', new Document([
+            'salary' => '123456',
+        ]));
+
+        $stored = $this->authorization->skip(fn () => $this->database->getDocument('employees', 'e1'));
+        $this->assertSame('123456', $stored->getAttribute('salary'));
+    }
+}
diff --git a/tests/unit/ColumnPermissionQueryTest.php b/tests/unit/ColumnPermissionQueryTest.php
new file mode 100644
index 0000000000..95539f3643
--- /dev/null
+++ b/tests/unit/ColumnPermissionQueryTest.php
@@ -0,0 +1,193 @@
+<?php
+
+namespace Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Cache\Adapter\None as NoCache;
+use Utopia\Cache\Cache;
+use Utopia\Database\Adapter\Memory;
+use Utopia\Database\Database;
+use Utopia\Database\Document;
+use Utopia\Database\Exception\Authorization as AuthorizationException;
+use Utopia\Database\Helpers\Permission;
+use Utopia\Database\Helpers\Role;
+use Utopia\Database\Query;
+use Utopia\Database\Validator\Authorization;
+
+/**
+ * Query, aggregate and write gates for column-level permissions.
+ *
+ * The collection grants `any` read and create on `name` only, so every caller is
+ * demonstrably column-restricted and the gates are exercised without a stored
+ * document having to be consulted.
+ */
+class ColumnPermissionQueryTest extends TestCase
+{
+    protected Authorization $authorization;
+
+    protected Database $database;
+
+    protected function setUp(): void
+    {
+        $this->authorization = new Authorization();
+
+        $this->database = new Database(new Memory(), new Cache(new NoCache()));
+        $this->database
+            ->setAuthorization($this->authorization)
+            ->setDatabase('columnPermissions')
+            ->setNamespace('colq_' . \uniqid());
+
+        if (!$this->database->exists()) {
+            $this->database->create();
+        }
+
+        $this->authorization->skip(function () {
+            $this->database->createCollection('employees', documentSecurity: true, permissions: [
+                Permission::read(Role::any(), 'name'),
+                Permission::create(Role::any(), 'name'),
+            ]);
+
+            $this->database->createAttribute('employees', 'name', Database::VAR_STRING, 128, false);
+            $this->database->createAttribute('employees', 'salary', Database::VAR_INTEGER, 8, false);
+
+            $this->database->createDocument('employees', new Document([
+                '$id' => 'e1',
+                '$permissions' => [
+                    Permission::read(Role::user('hr'), 'salary'),
+                    Permission::update(Role::any(), 'name'),
+                ],
+                'name' => 'Bob',
+                'salary' => 100000,
+            ]));
+        });
+
+        $this->authorization->cleanRoles();
+        $this->authorization->addRole('any');
+    }
+
+    public function testCreateOfGrantedColumnIsAllowed(): void
+    {
+        $created = $this->database->createDocument('employees', new Document([
+            '$id' => 'c1',
+            'name' => 'Alice',
+        ]));
+
+        $this->assertSame('c1', $created->getId());
+    }
+
+    public function testCreateOfUngrantedColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "create" permission for column "salary"');
+
+        $this->database->createDocument('employees', new Document([
+            '$id' => 'c2',
+            'salary' => 9,
+        ]));
+    }
+
+    public function testFilterOnReadableColumnIsAllowed(): void
+    {
+        $this->assertCount(1, $this->database->find('employees', [Query::equal('name', ['Bob'])]));
+    }
+
+    /**
+     * Masking hides the value, but an unguarded filter turns the result set into an
+     * oracle for it.
+     */
+    public function testFilterOnUnreadableColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "read" permission for column "salary"');
+
+        $this->database->find('employees', [Query::greaterThan('salary', 1)]);
+    }
+
+    public function testOrderByUnreadableColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+
+        $this->database->find('employees', [Query::orderDesc('salary')]);
+    }
+
+    public function testSelectOfUnreadableColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+
+        $this->database->find('employees', [Query::select(['salary'])]);
+    }
+
+    public function testCountFilteredByUnreadableColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+
+        $this->database->count('employees', [Query::equal('salary', [100000])]);
+    }
+
+    /**
+     * Without this guard sum() extracts a masked column in a single call.
+     */
+    public function testSumOfUnreadableColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "read" permission for column "salary"');
+
+        $this->database->sum('employees', 'salary');
+    }
+
+    public function testBulkUpdateOfGrantedColumnIsAllowed(): void
+    {
+        $this->assertSame(1, $this->database->updateDocuments('employees', new Document([
+            'name' => 'Renamed',
+        ])));
+    }
+
+    /**
+     * The collection grants no update at all, so the restriction is only visible on
+     * the document itself: the bulk path has to check each row, not just the schema.
+     */
+    public function testBulkUpdateOfUngrantedColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "update" permission for column "salary"');
+
+        $this->database->updateDocuments('employees', new Document(['salary' => 1]));
+    }
+
+    public function testIncreaseOfUngrantedColumnIsRejected(): void
+    {
+        $this->expectException(AuthorizationException::class);
+        $this->expectExceptionMessage('Missing "update" permission for column "salary"');
+
+        $this->database->increaseDocumentAttribute('employees', 'e1', 'salary', 1);
+    }
+
+    public function testPermissionsScopedToUnreadableColumnsAreMasked(): void
+    {
+        $document = $this->database->getDocument('employees', 'e1');
+
+        $this->assertSame(['update("any", "name")'], $document->getPermissions());
+    }
+
+    /**
+     * Because $permissions is masked, writing a document straight back would delete
+     * the grants the caller never saw.
+     */
+    public function testMaskedPermissionsSurviveARoundTrip(): void
+    {
+        $document = $this->database->getDocument('employees', 'e1');
+
+        $this->database->updateDocument('employees', 'e1', new Document([
+            '$permissions' => $document->getPermissions(),
+            'name' => 'Bob2',
+        ]));
+
+        $stored = $this->authorization->skip(
+            fn () => $this->database->getDocument('employees', 'e1')
+        );
+
+        $this->assertContains('read("user:hr", "salary")', $stored->getPermissions());
+        $this->assertContains('update("any", "name")', $stored->getPermissions());
+        $this->assertSame(100000, $stored->getAttribute('salary'));
+    }
+}
diff --git a/tests/unit/ColumnPermissionTest.php b/tests/unit/ColumnPermissionTest.php
new file mode 100644
index 0000000000..922177365b
--- /dev/null
+++ b/tests/unit/ColumnPermissionTest.php
@@ -0,0 +1,171 @@
+<?php
+
+namespace Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Utopia\Database\Document;
+use Utopia\Database\Exception as DatabaseException;
+use Utopia\Database\Helpers\Permission;
+use Utopia\Database\Helpers\Role;
+use Utopia\Database\Validator\Permissions;
+
+class ColumnPermissionTest extends TestCase
+{
+    public function testParseWithoutColumnGrantsEveryColumn(): void
+    {
+        foreach (['read("any")', 'read("users")', 'read("user:123")', 'read("team:123/admin")'] as $string) {
+            $permission = Permission::parse($string);
+
+            $this->assertSame(Permission::COLUMN_ALL, $permission->getColumn());
+            $this->assertTrue($permission->isForAllColumns());
+            $this->assertSame($string, $permission->toString());
+        }
+    }
+
+    public function testParseWithColumn(): void
+    {
+        $permission = Permission::parse('read("user:123", "salary")');
+
+        $this->assertSame('read', $permission->getPermission());
+        $this->assertSame('user', $permission->getRole());
+        $this->assertSame('123', $permission->getIdentifier());
+        $this->assertSame('', $permission->getDimension());
+        $this->assertSame('salary', $permission->getColumn());
+        $this->assertFalse($permission->isForAllColumns());
+    }
+
+    public function testParseWithColumnAndDimension(): void
+    {
+        $permission = Permission::parse('update("team:abc/owner", "salary")');
+
+        $this->assertSame('update', $permission->getPermission());
+        $this->assertSame('team', $permission->getRole());
+        $this->assertSame('abc', $permission->getIdentifier());
+        $this->assertSame('owner', $permission->getDimension());
+        $this->assertSame('salary', $permission->getColumn());
+    }
+
+    /**
+     * @return array<string, array{string}>
+     */
+    public static function roundTripProvider(): array
+    {
+        return [
+            'no column' => ['read("any")'],
+            'column' => ['read("user:123", "salary")'],
+            'dimension and column' => ['read("team:abc/owner", "salary")'],
+            'update column' => ['update("user:123", "name")'],
+            'create column' => ['create("users", "name")'],
+        ];
+    }
+
+    /**
+     * @dataProvider roundTripProvider
+     */
+    public function testRoundTrip(string $string): void
+    {
+        $this->assertSame($string, Permission::parse($string)->toString());
+    }
+
+    public function testFactories(): void
+    {
+        $this->assertSame('read("user:123")', Permission::read(Role::user('123')));
+        $this->assertSame('read("user:123", "salary")', Permission::read(Role::user('123'), 'salary'));
+        $this->assertSame('update("team:abc/owner", "name")', Permission::update(Role::team('abc', 'owner'), 'name'));
+        $this->assertSame('create("users", "name")', Permission::create(Role::users(), 'name'));
+    }
+
+    public function testAggregatePreservesColumn(): void
+    {
+        $aggregated = Permission::aggregate(['read("user:1", "name")']);
+
+        $this->assertSame(['read("user:1", "name")'], $aggregated);
+    }
+
+    public function testEmptyColumnIsRejected(): void
+    {
+        $this->expectException(DatabaseException::class);
+        Permission::parse('read("user:1", "")');
+    }
+
+    public function testWildcardColumnIsRejected(): void
+    {
+        $this->expectException(DatabaseException::class);
+        Permission::parse('read("user:1", "*")');
+    }
+
+    /**
+     * A column-scoped permission must still resolve to a bare role, or every
+     * existing document-level authorization check silently breaks.
+     */
+    public function testDocumentPermissionsByTypeReturnsRolesOnly(): void
+    {
+        $document = new Document(['$permissions' => [
+            'read("any")',
+            'read("user:1", "salary")',
+            'update("user:1", "name")',
+            'delete("user:1")',
+        ]]);
+
+        $this->assertSame(['any', 'user:1'], \array_values($document->getRead()));
+        $this->assertSame(['user:1'], \array_values($document->getUpdate()));
+        $this->assertSame(['user:1'], \array_values($document->getDelete()));
+    }
+
+    public function testDocumentPermissionsByTypeWithColumns(): void
+    {
+        $document = new Document(['$permissions' => [
+            'read("any")',
+            'read("user:1", "salary")',
+        ]]);
+
+        $this->assertSame([
+            ['role' => 'any', 'column' => Permission::COLUMN_ALL],
+            ['role' => 'user:1', 'column' => 'salary'],
+        ], $document->getPermissionsByTypeWithColumns('read'));
+    }
+
+    public function testValidatorAcceptsColumnScopedReadCreateUpdate(): void
+    {
+        $validator = new Permissions();
+
+        $this->assertTrue($validator->isValid([
+            'read("user:1", "salary")',
+            'create("users", "name")',
+            'update("team:abc/owner", "name")',
+        ]), $validator->getDescription());
+    }
+
+    public function testValidatorRejectsColumnScopedDelete(): void
+    {
+        $validator = new Permissions();
+
+        $this->assertFalse($validator->isValid(['delete("user:1", "salary")']));
+        $this->assertStringContainsString('cannot be scoped to a column', $validator->getDescription());
+    }
+
+    public function testValidatorRejectsColumnScopedWrite(): void
+    {
+        $validator = new Permissions();
+
+        $this->assertFalse($validator->isValid(['write("user:1", "salary")']));
+        $this->assertStringContainsString('cannot be scoped to a column', $validator->getDescription());
+    }
+
+    public function testValidatorRejectsUnknownColumnWhenColumnsGiven(): void
+    {
+        $validator = new Permissions(columns: ['name', 'email']);
+
+        $this->assertTrue($validator->isValid(['read("user:1", "name")']), $validator->getDescription());
+        $this->assertFalse($validator->isValid(['read("user:1", "salary")']));
+        $this->assertStringContainsString('does not exist', $validator->getDescription());
+    }
+
+    public function testValidatorRejectsInvalidColumnKey(): void
+    {
+        $validator = new Permissions();
+
+        $this->assertFalse($validator->isValid(['read("user:1", "_internal")']));
+        $this->assertStringContainsString('not a valid column key', $validator->getDescription());
+    }
+}