[fix](polixy)support drop policy for user or role (#29488)

fe/fe-core/src/main/java/org/apache/doris/analysis/DropPolicyStmt.java

@@ -27,11 +27,12 @@ import org.apache.doris.qe.ConnectContext;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import org.apache.commons.lang3.StringUtils;
 
 /**
  * Drop policy statement.
  * syntax:
- * DROP [ROW] POLICY [IF EXISTS] test_row_policy
+ * DROP [ROW] POLICY [IF EXISTS] test_row_policy ON test_table [FOR user|ROLE role]
  **/
 @AllArgsConstructor
 public class DropPolicyStmt extends DdlStmt {
@@ -45,9 +46,28 @@ public class DropPolicyStmt extends DdlStmt {
     @Getter
     private final String policyName;
 
+    @Getter
+    private final TableName tableName;
+
+    @Getter
+    private final UserIdentity user;
+
+    @Getter
+    private final String roleName;
+
     @Override
     public void analyze(Analyzer analyzer) throws UserException {
         super.analyze(analyzer);
+        switch (type) {
+            case STORAGE:
+                break;
+            case ROW:
+            default:
+                tableName.analyze(analyzer);
+                if (user != null) {
+                    user.analyze();
+                }
+        }
         // check auth
         if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
@@ -62,6 +82,19 @@ public class DropPolicyStmt extends DdlStmt {
             sb.append("IF EXISTS ");
         }
         sb.append(policyName);
+        switch (type) {
+            case STORAGE:
+                break;
+            case ROW:
+            default:
+                sb.append(" ON ").append(tableName.toSql());
+                if (user != null) {
+                    sb.append(" FOR ").append(user.getQualifiedUser());
+                }
+                if (StringUtils.isEmpty(roleName)) {
+                    sb.append(" FOR ROLE ").append(roleName);
+                }
+        }
         return sb.toString();
     }
 }

fe/fe-core/src/main/java/org/apache/doris/policy/DropPolicyLog.java

@@ -18,10 +18,15 @@
 package org.apache.doris.policy;
 
 import org.apache.doris.analysis.DropPolicyStmt;
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+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.gson.annotations.SerializedName;
 import lombok.AllArgsConstructor;
@@ -37,18 +42,44 @@ import java.io.IOException;
 @AllArgsConstructor
 @Getter
 public class DropPolicyLog implements Writable {
+
+    @SerializedName(value = "dbId")
+    private long dbId;
+
+    @SerializedName(value = "tableId")
+    private long tableId;
+
     @SerializedName(value = "type")
     private PolicyTypeEnum type;
 
     @SerializedName(value = "policyName")
     private String policyName;
 
+    @SerializedName(value = "user")
+    private UserIdentity user;
+
+    @SerializedName(value = "roleName")
+    private String roleName;
 
     /**
      * Generate delete logs through stmt.
      **/
     public static DropPolicyLog fromDropStmt(DropPolicyStmt stmt) throws AnalysisException {
-        return new DropPolicyLog(stmt.getType(), stmt.getPolicyName());
+        switch (stmt.getType()) {
+            case STORAGE:
+                return new DropPolicyLog(-1, -1, stmt.getType(), stmt.getPolicyName(), null, null);
+            case ROW:
+                String curDb = stmt.getTableName().getDb();
+                if (curDb == null) {
+                    curDb = ConnectContext.get().getDatabase();
+                }
+                Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException(curDb);
+                Table table = db.getTableOrAnalysisException(stmt.getTableName().getTbl());
+                return new DropPolicyLog(db.getId(), table.getId(), stmt.getType(),
+                        stmt.getPolicyName(), stmt.getUser(), stmt.getRoleName());
+            default:
+                throw new AnalysisException("Invalid policy type: " + stmt.getType().name());
+        }
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/policy/RowPolicy.java

@@ -175,6 +175,13 @@ public class RowPolicy extends Policy {
                 rowPolicy.getPolicyName(), rowPolicy.getUser(), rowPolicy.getRoleName());
     }
 
+    @Override
+    public boolean matchPolicy(DropPolicyLog checkedDropPolicyLogCondition) {
+        return checkMatched(checkedDropPolicyLogCondition.getDbId(), checkedDropPolicyLogCondition.getTableId(),
+                checkedDropPolicyLogCondition.getType(), checkedDropPolicyLogCondition.getPolicyName(),
+                checkedDropPolicyLogCondition.getUser(), checkedDropPolicyLogCondition.getRoleName());
+    }
+
     @Override
     public boolean isInvalid() {
         return (wherePredicate == null);