You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by GitBox <gi...@apache.org> on 2022/04/27 16:11:17 UTC
[GitHub] [incubator-doris] morningman commented on a diff in pull request #9206: [feature]: support row policy filter
morningman commented on code in PR #9206:
URL: https://github.com/apache/incubator-doris/pull/9206#discussion_r859958551
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/Expr.java:
##########
@@ -1654,7 +1653,7 @@ public Subquery getSubquery() {
List<Subquery> subqueries = Lists.newArrayList();
collect(Subquery.class, subqueries);
Preconditions.checkState(subqueries.size() == 1,
- "only support one subquery in " + this.toSql());
Review Comment:
Better not to format this file, since you didn't change it
##########
docs/zh-CN/administrator-guide/policy/row-policy.md:
##########
@@ -0,0 +1,60 @@
+---
+{
+"title": "行安全策略",
+"language": "zh-CN"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# 行安全策略
+
+该功能用于配置行级权限，在 sql 执行时添加对应拦截条件，通过 explain 可以看到改写后的过滤条件:
+
+## 语句
+
+目前支持对行策略的创建/删除/查询
+
+创建
+```sql
+CREATE [ROW] POLICY [IF NOT EXISTS] test_row_policy
Review Comment:
Why make ROW as an optional keywords?
I think it can be required.
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/InPredicate.java:
##########
@@ -209,16 +209,16 @@ public void analyzeImpl(Analyzer analyzer) throws AnalysisException {
opcode = isNotIn ? TExprOpcode.FILTER_NOT_IN : TExprOpcode.FILTER_IN;
} else {
fn = getBuiltinFunction(analyzer, isNotIn ? NOT_IN_ITERATE : IN_ITERATE,
- argTypes, Function.CompareMode.IS_NONSTRICT_SUPERTYPE_OF);
+ argTypes, Function.CompareMode.IS_NONSTRICT_SUPERTYPE_OF);
opcode = isNotIn ? TExprOpcode.FILTER_NEW_NOT_IN : TExprOpcode.FILTER_NEW_IN;
}
Reference<SlotRef> slotRefRef = new Reference<SlotRef>();
Reference<Integer> idxRef = new Reference<Integer>();
if (isSingleColumnPredicate(slotRefRef, idxRef)
- && idxRef.getRef() == 0 && slotRefRef.getRef().getNumDistinctValues() > 0) {
+ && idxRef.getRef() == 0 && slotRefRef.getRef().getNumDistinctValues() > 0) {
selectivity = (double) (getChildren().size() - 1) / (double) slotRefRef.getRef()
- .getNumDistinctValues();
+ .getNumDistinctValues();
Review Comment:
Don't format this file
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java:
##########
@@ -20,15 +20,21 @@
package org.apache.doris.analysis;
-import org.apache.doris.catalog.Type;
-import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.TableAliasGenerator;
-import org.apache.doris.common.UserException;
-
import com.google.common.base.Preconditions;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
+import lombok.SneakyThrows;
Review Comment:
Import order, see http://doris.incubator.apache.org/zh-CN/developer-guide/java-format-code.html
##########
fe/fe-core/src/test/java/org/apache/doris/policy/PolicyTest.java:
##########
@@ -0,0 +1,127 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import org.apache.doris.analysis.CreateDbStmt;
+import org.apache.doris.analysis.CreatePolicyStmt;
+import org.apache.doris.analysis.CreateTableStmt;
+import org.apache.doris.analysis.DropPolicyStmt;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.ExceptionChecker;
+import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.utframe.UtFrameUtils;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.UUID;
+
+public class PolicyTest {
+
+ private static String runningDir = "fe/mocked/policyTest/" + UUID.randomUUID() + "/";
+
+ private static ConnectContext connectContext;
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ UtFrameUtils.createDorisCluster(runningDir);
+
+ // create connect context
+ connectContext = UtFrameUtils.createDefaultCtx();
+
+ // create database
+ String createDbStmtStr = "create database test;";
+ CreateDbStmt createDbStmt = (CreateDbStmt) UtFrameUtils.parseAndAnalyzeStmt(createDbStmtStr, connectContext);
+ Catalog.getCurrentCatalog().createDb(createDbStmt);
+ Catalog.getCurrentCatalog().getDb("test");
+
+ MetricRepo.init();
+ Catalog.getCurrentCatalog().changeDb(connectContext, "default_cluster:test");
+ createTable("create table table1\n" +
+ "(k1 int, k2 int) distributed by hash(k1) buckets 1\n" +
+ "properties(\"replication_num\" = \"1\");");
+ createTable("create table table2\n" +
+ "(k1 int, k2 int) distributed by hash(k1) buckets 1\n" +
+ "properties(\"replication_num\" = \"1\");");
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ File file = new File(runningDir);
+ file.delete();
+ }
+
+ @Test
+ public void testReWriteSql() throws Exception {
+ Catalog.getCurrentCatalog().changeDb(connectContext, "default_cluster:test");
+ createPolicy("CREATE POLICY test_row_policy ON test.table1 AS PERMISSIVE TO root USING (k1 = 1)");
+ String queryStr = "EXPLAIN select * from test.table1";
+ String explainString = UtFrameUtils.getSQLPlanOrErrorMsg(connectContext, queryStr);
+ Assert.assertTrue(explainString.contains("`k1` = 1"));
+ }
+
+ @Test
+ public void testDuplicateAddPolicy() throws Exception {
+ createPolicy("CREATE POLICY test_row_policy1 ON test.table1 AS PERMISSIVE TO root USING (k1 = 1)");
+ ExceptionChecker.expectThrowsWithMsg(DdlException.class, "the policy test_row_policy1 already create",
+ () -> createPolicy("CREATE POLICY test_row_policy1 ON test.table1 AS PERMISSIVE TO root USING (k1 = 1)"));
+ }
+
+ @Test
+ public void testShowPolicy() throws Exception {
+ createPolicy("CREATE POLICY test_row_policy3 ON test.table1 AS PERMISSIVE TO root USING (k1 = 1)");
+ createPolicy("CREATE POLICY test_row_policy4 ON test.table1 AS PERMISSIVE TO root USING (k2 = 1)");
+ int firstSize = Catalog.getCurrentCatalog().getPolicyMgr().getDbUserPolicies(ConnectContext.get().getCurrentDbId(), "root").size();
+ Assert.assertTrue(firstSize > 0);
+ dropPolicy("DROP POLICY test_row_policy3 ON test.table1");
+ dropPolicy("DROP POLICY test_row_policy4 ON test.table1");
+ int secondSize = Catalog.getCurrentCatalog().getPolicyMgr().getDbUserPolicies(ConnectContext.get().getCurrentDbId(), "root").size();
+ Assert.assertEquals(2, firstSize - secondSize);
+ }
+
+ @Test
+ public void testMergeFilter() throws Exception {
+ createPolicy("CREATE POLICY test_row_policy1 ON test.table2 AS PERMISSIVE TO root USING (k1 = 1)");
+ createPolicy("CREATE POLICY test_row_policy2 ON test.table2 AS PERMISSIVE TO root USING (k2 = 1)");
+ // filterType use last
+ createPolicy("CREATE POLICY test_row_policy3 ON test.table2 AS RESTRICTIVE TO root USING (k2 = 2)");
+ String queryStr = "EXPLAIN select * from test.table2";
Review Comment:
You need to test more complicated query, like `union`, `join`, `subquery` and `view`
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java:
##########
@@ -1141,5 +1147,32 @@ private static Expr createJoinConjunct(Expr exprWithSubquery, InlineViewRef inli
smap.put(subquery, subquerySubstitute);
return exprWithSubquery.substitute(smap, analyzer, false);
}
+
+ @SneakyThrows
Review Comment:
Why using `SneakyThrows`? I think we should throw explicit exception
##########
fe/fe-core/src/main/java/org/apache/doris/policy/PolicyMgr.java:
##########
@@ -0,0 +1,180 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import org.apache.doris.analysis.CreatePolicyStmt;
+import org.apache.doris.analysis.DropPolicyStmt;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.common.collect.Maps;
+import com.google.gson.annotations.SerializedName;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+public class PolicyMgr implements Writable {
+ private static final Logger LOG = LogManager.getLogger(PolicyMgr.class);
+
+ private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
+
+ @SerializedName(value = "dbIdToPolicyMap")
+ private Map<Long, List<Policy>> dbIdToPolicyMap = Maps.newConcurrentMap();
+
+ private void writeLock() {
+ lock.writeLock().lock();
+ }
+
+ private void writeUnlock() {
+ lock.writeLock().unlock();
+ }
+
+ public void createPolicy(CreatePolicyStmt stmt) throws UserException {
+ writeLock();
+ try {
+ Policy policy = Policy.fromCreateStmt(stmt);
+ if (existPolicy(policy.getDbId(), policy.getTableId(), policy.getType(), policy.getPolicyName())) {
+ if (stmt.isIfNotExists()) {
+ return;
+ }
+ throw new DdlException("the policy " + policy.getPolicyName() + " already create");
+ }
+ unprotectedAdd(policy);
+ Catalog.getCurrentCatalog().getEditLog().logCreatePolicy(policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ public void dropPolicy(DropPolicyStmt stmt) throws DdlException {
+ writeLock();
+ try {
+ DropPolicyLog policy = DropPolicyLog.fromDropStmt(stmt);
+ if (!existPolicy(policy.getDbId(), policy.getTableId(), policy.getType(), policy.getPolicyName())) {
+ if (stmt.isIfExists()) {
+ return;
+ }
+ throw new DdlException("the policy " + policy.getPolicyName() + " not exist");
+ }
+ unprotectedDrop(policy);
+ Catalog.getCurrentCatalog().getEditLog().logDropPolicy(policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private boolean existPolicy(long dbId, long tableId, String type, String policyName) {
+ List<Policy> policies = getDbPolicies(dbId);
+ return policies.stream().anyMatch(policy -> matchPolicy(policy, type, tableId, policyName));
+ }
+
+ public List<Policy> getDbPolicies(long dbId) {
+ if (dbIdToPolicyMap == null) {
+ return new ArrayList<>();
+ }
+ return dbIdToPolicyMap.getOrDefault(dbId, new ArrayList<>());
+ }
+
+ public List<Policy> getDbUserPolicies(long dbId, String user) {
+ if (dbIdToPolicyMap == null) {
+ return new ArrayList<>();
+ }
+ return dbIdToPolicyMap.getOrDefault(dbId, new ArrayList<>()).stream().filter(p -> p.getUser().equals(user)).collect(Collectors.toList());
+ }
+
+ public void replayCreate(Policy policy) {
+ unprotectedAdd(policy);
+ LOG.info("replay create policy: {}", policy);
+ }
+
+ public void unprotectedAdd(Policy policy) {
+ if (policy == null) {
+ return;
+ }
+ long dbId = policy.getDbId();
+ List<Policy> dbPolicies = getDbPolicies(dbId);
+ dbPolicies.add(policy);
+ dbIdToPolicyMap.put(dbId, dbPolicies);
+ }
+
+ public void replayDrop(DropPolicyLog log) {
+ unprotectedDrop(log);
+ LOG.info("replay drop policy log: {}", log);
+ }
+
+ public void unprotectedDrop(DropPolicyLog log) {
+ List<Policy> policies = getDbPolicies(log.getDbId());
+ policies.removeIf(p -> matchPolicy(p, log.getType(), log.getTableId(), log.getPolicyName()));
+ dbIdToPolicyMap.put(log.getDbId(), policies);
+ }
+
+ private boolean matchPolicy(Policy policy, String type, long tableId, String policyName) {
+ return StringUtils.equals(policy.getType(), type)
+ && policy.getTableId() == tableId
+ && StringUtils.equals(policy.getPolicyName(), policyName);
+ }
+
+ public Policy getMatchPolicy(long dbId, long tableId, String user) {
+ if (!dbIdToPolicyMap.containsKey(dbId)) {
+ return null;
+ }
+ List<Policy> policies = dbIdToPolicyMap.get(dbId);
+ List<Policy> userPolicies = policies.stream().filter(policy -> policy.getTableId() == tableId && StringUtils.equals(policy.getUser(), user)).collect(Collectors.toList());
+ if (userPolicies.isEmpty()) {
+ return null;
+ }
+ // op use last
+ Policy lastPolicy = userPolicies.get(userPolicies.size() - 1);
Review Comment:
About the `op`, I think the logic is:
1. get all policy with `RESTRICTIVE`, and combine them with `and`, eg: (A and B and C)
2. get all policy with `PERMISSIVE`, and combine them with `or`, eg: (A and B and C) or D or E
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/Catalog.java:
##########
@@ -2204,6 +2216,11 @@ public long saveSqlBlockRule(CountingDataOutputStream out, long checksum) throws
return checksum;
}
+ public long savePolicy(CountingDataOutputStream out, long checksum) throws IOException {
+ Catalog.getCurrentCatalog().getPolicyMgr().write(out);
Review Comment:
```suggestion
this.policyMgr.write(out);
```
##########
fe/fe-core/src/main/cup/sql_parser.cup:
##########
@@ -275,7 +275,7 @@ terminal String KW_ADD, KW_ADMIN, KW_AFTER, KW_AGGREGATE, KW_ALIAS, KW_ALL, KW_A
KW_UNCOMMITTED, KW_UNBOUNDED, KW_UNION, KW_UNIQUE, KW_UNLOCK, KW_UNSIGNED, KW_USE, KW_USER, KW_USING, KW_UNINSTALL,
KW_VALUE, KW_VALUES, KW_VARCHAR, KW_VARIABLES, KW_VERBOSE, KW_VIEW,
KW_WARNINGS, KW_WEEK, KW_WHEN, KW_WHITELIST, KW_WHERE, KW_WITH, KW_WORK, KW_WRITE,
- KW_YEAR, KW_SQL_BLOCK_RULE;
+ KW_YEAR, KW_SQL_BLOCK_RULE, KW_POLICY;
Review Comment:
need to add `KW_POLICY` to the `keywords :=` region
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java:
##########
@@ -2139,4 +2143,21 @@ private void handleAdminDiagnoseTablet() {
resultSet = new ShowResultSet(showMetaData, resultRowSet);
}
+ public void handleShowPolicy() throws AnalysisException {
+ ShowPolicyStmt showStmt = (ShowPolicyStmt) stmt;
+ List<List<String>> rows = Lists.newArrayList();
+ List<Policy> policies;
+ long currentDbId = ConnectContext.get().getCurrentDbId();
+ if (showStmt.getUser() == null) {
+ policies = Catalog.getCurrentCatalog().getPolicyMgr().getDbPolicies(currentDbId);
+ } else {
+ policies = Catalog.getCurrentCatalog().getPolicyMgr().getDbUserPolicies(currentDbId, showStmt.getUser());
+ }
+ LOG.info("policies={}", policies);
Review Comment:
```suggestion
LOG.debug("policies={}", policies);
```
##########
fe/fe-core/src/main/java/org/apache/doris/policy/Policy.java:
##########
@@ -0,0 +1,112 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import org.apache.doris.analysis.CreatePolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.Lists;
+import com.google.gson.annotations.SerializedName;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+public class Policy implements Writable {
+ private static final Logger LOG = LogManager.getLogger(Policy.class);
+
+ @SerializedName(value = "dbId")
+ private long dbId;
+
+ @SerializedName(value = "tableId")
+ private long tableId;
+
+ @SerializedName(value = "policyName")
+ private String policyName;
+
+ /**
+ * ROW
+ **/
+ @SerializedName(value = "type")
+ private String type;
+
+ /**
+ * PERMISSIVE | RESTRICTIVE, If multiple types exist, the last type prevails
+ **/
+ @SerializedName(value = "filterType")
+ private final FilterType filterType;
+
+ @SerializedName(value = "whereSql")
+ private String whereSql;
+
+ /**
+ * bind user
+ **/
+ @SerializedName(value = "user")
+ private final String user;
Review Comment:
```suggestion
private final UserIdentity user;
```
And modify the read/write method of `UserIdentity` to support GSON
##########
fe/fe-core/src/main/java/org/apache/doris/policy/DropPolicyLog.java:
##########
@@ -0,0 +1,72 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.SneakyThrows;
+import org.apache.doris.analysis.DropPolicyStmt;
Review Comment:
import order
##########
fe/fe-core/src/main/java/org/apache/doris/policy/PolicyMgr.java:
##########
@@ -0,0 +1,180 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import org.apache.doris.analysis.CreatePolicyStmt;
+import org.apache.doris.analysis.DropPolicyStmt;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.common.collect.Maps;
+import com.google.gson.annotations.SerializedName;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+public class PolicyMgr implements Writable {
+ private static final Logger LOG = LogManager.getLogger(PolicyMgr.class);
+
+ private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
+
+ @SerializedName(value = "dbIdToPolicyMap")
+ private Map<Long, List<Policy>> dbIdToPolicyMap = Maps.newConcurrentMap();
+
+ private void writeLock() {
+ lock.writeLock().lock();
+ }
+
+ private void writeUnlock() {
+ lock.writeLock().unlock();
+ }
+
+ public void createPolicy(CreatePolicyStmt stmt) throws UserException {
+ writeLock();
+ try {
+ Policy policy = Policy.fromCreateStmt(stmt);
+ if (existPolicy(policy.getDbId(), policy.getTableId(), policy.getType(), policy.getPolicyName())) {
+ if (stmt.isIfNotExists()) {
+ return;
+ }
+ throw new DdlException("the policy " + policy.getPolicyName() + " already create");
+ }
+ unprotectedAdd(policy);
+ Catalog.getCurrentCatalog().getEditLog().logCreatePolicy(policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ public void dropPolicy(DropPolicyStmt stmt) throws DdlException {
+ writeLock();
+ try {
+ DropPolicyLog policy = DropPolicyLog.fromDropStmt(stmt);
+ if (!existPolicy(policy.getDbId(), policy.getTableId(), policy.getType(), policy.getPolicyName())) {
+ if (stmt.isIfExists()) {
+ return;
+ }
+ throw new DdlException("the policy " + policy.getPolicyName() + " not exist");
+ }
+ unprotectedDrop(policy);
+ Catalog.getCurrentCatalog().getEditLog().logDropPolicy(policy);
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ private boolean existPolicy(long dbId, long tableId, String type, String policyName) {
+ List<Policy> policies = getDbPolicies(dbId);
+ return policies.stream().anyMatch(policy -> matchPolicy(policy, type, tableId, policyName));
+ }
+
+ public List<Policy> getDbPolicies(long dbId) {
+ if (dbIdToPolicyMap == null) {
+ return new ArrayList<>();
+ }
+ return dbIdToPolicyMap.getOrDefault(dbId, new ArrayList<>());
+ }
+
+ public List<Policy> getDbUserPolicies(long dbId, String user) {
+ if (dbIdToPolicyMap == null) {
+ return new ArrayList<>();
+ }
+ return dbIdToPolicyMap.getOrDefault(dbId, new ArrayList<>()).stream().filter(p -> p.getUser().equals(user)).collect(Collectors.toList());
+ }
+
+ public void replayCreate(Policy policy) {
+ unprotectedAdd(policy);
+ LOG.info("replay create policy: {}", policy);
+ }
+
+ public void unprotectedAdd(Policy policy) {
+ if (policy == null) {
+ return;
+ }
+ long dbId = policy.getDbId();
+ List<Policy> dbPolicies = getDbPolicies(dbId);
+ dbPolicies.add(policy);
+ dbIdToPolicyMap.put(dbId, dbPolicies);
+ }
+
+ public void replayDrop(DropPolicyLog log) {
+ unprotectedDrop(log);
+ LOG.info("replay drop policy log: {}", log);
+ }
+
+ public void unprotectedDrop(DropPolicyLog log) {
+ List<Policy> policies = getDbPolicies(log.getDbId());
+ policies.removeIf(p -> matchPolicy(p, log.getType(), log.getTableId(), log.getPolicyName()));
+ dbIdToPolicyMap.put(log.getDbId(), policies);
+ }
+
+ private boolean matchPolicy(Policy policy, String type, long tableId, String policyName) {
+ return StringUtils.equals(policy.getType(), type)
+ && policy.getTableId() == tableId
+ && StringUtils.equals(policy.getPolicyName(), policyName);
+ }
+
+ public Policy getMatchPolicy(long dbId, long tableId, String user) {
+ if (!dbIdToPolicyMap.containsKey(dbId)) {
+ return null;
+ }
+ List<Policy> policies = dbIdToPolicyMap.get(dbId);
Review Comment:
You need read lock to protect this method.
##########
docs/zh-CN/administrator-guide/policy/row-policy.md:
##########
@@ -0,0 +1,60 @@
+---
+{
+"title": "行安全策略",
+"language": "zh-CN"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# 行安全策略
+
+该功能用于配置行级权限，在 sql 执行时添加对应拦截条件，通过 explain 可以看到改写后的过滤条件:
+
+## 语句
+
+目前支持对行策略的创建/删除/查询
+
+创建
+```sql
+CREATE [ROW] POLICY [IF NOT EXISTS] test_row_policy
+ON test_table # 作用的表
+AS {PERMISSIVE|RESTRICTIVE} # 关联条件，以最后一次创建的为准
+TO admin # 绑定的用户
+USING (a = ’xxx‘) # 拦截语句
+```
+- PERMISSIVE: 宽松条件，多条策略通过 OR 连接
+- RESTRICTIVE：限制条件，多条策略通过 AND 连接
+
+创建示例
+```sql
+CREATE POLICY test_row_policy_1 ON test.table1 AS RESTRICTIVE TO root USING (id in (1, 2));
+```
+
+删除
+```sql
+DROP POLICY test_row_policy_1 on table1;
Review Comment:
How to support drop policy for specified user?
##########
fe/fe-core/src/main/java/org/apache/doris/policy/Policy.java:
##########
@@ -0,0 +1,112 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.policy;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import org.apache.doris.analysis.CreatePolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.Lists;
+import com.google.gson.annotations.SerializedName;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+public class Policy implements Writable {
+ private static final Logger LOG = LogManager.getLogger(Policy.class);
+
+ @SerializedName(value = "dbId")
+ private long dbId;
+
+ @SerializedName(value = "tableId")
+ private long tableId;
+
+ @SerializedName(value = "policyName")
+ private String policyName;
+
+ /**
+ * ROW
+ **/
+ @SerializedName(value = "type")
+ private String type;
+
+ /**
+ * PERMISSIVE | RESTRICTIVE, If multiple types exist, the last type prevails
+ **/
+ @SerializedName(value = "filterType")
+ private final FilterType filterType;
+
+ @SerializedName(value = "whereSql")
+ private String whereSql;
+
+ /**
+ * bind user
+ **/
+ @SerializedName(value = "user")
+ private final String user;
+
+ public static Policy fromCreateStmt(CreatePolicyStmt stmt) throws AnalysisException {
+ String curDb = stmt.getTableName().getDb();
+ if (curDb == null) {
+ curDb = ConnectContext.get().getDatabase();
+ }
+ Database db = Catalog.getCurrentCatalog().getDbOrAnalysisException(curDb);
+ Table table = db.getTableOrAnalysisException(stmt.getTableName().getTbl());
+ UserIdentity userIdent = stmt.getUserIdent();
+ userIdent.analyze(ConnectContext.get().getClusterName());
+ return new Policy(db.getId(), table.getId(), stmt.getPolicyName(), stmt.getType(), stmt.getFilterType(), stmt.getWherePredicate().toSql(), userIdent.getQualifiedUser());
Review Comment:
`stmt.getWherePredicate().toSql()` may not restore the origin where predicate.
I suggest to save the entire `create policy` origin stmt string in `Policy` class. And then parse the entire stmt
to extract the `where` part.
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java:
##########
@@ -1141,5 +1147,32 @@ private static Expr createJoinConjunct(Expr exprWithSubquery, InlineViewRef inli
smap.put(subquery, subquerySubstitute);
return exprWithSubquery.substitute(smap, analyzer, false);
}
+
+ @SneakyThrows
+ public static void rewriteByPolicy(SelectStmt selectStmt, Analyzer analyzer) {
+ Catalog currentCatalog = Catalog.getCurrentCatalog();
+ for (int i = 0; i < selectStmt.fromClause_.size(); i++) {
Review Comment:
The `TableRef` is not only in `from` clause.
You need to traverse the top level `selectStmt` to find all of them. maybe in subquery.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org