feat: parse the MySQL replication statements

CHANGE REPLICATION SOURCE TO, CHANGE REPLICATION FILTER, START REPLICA,
STOP REPLICA, RESET REPLICA, RESET BINARY LOGS AND GTIDS and PURGE BINARY
LOGS failed to parse, so a replica setup script failed at its first
statement.

The new statements of net.sf.jsqlparser.statement.replication follow the
MySQL 8.4 grammar. ReplicationSourceOption extends ReplicationOption, and
each option kind says which value it takes: a string, a number, NULL,
0 or 1, a keyword, the server ID list or an account. ReplicationFilter
keeps database names, tables, patterns or rewrite pairs, and the
statements share one FOR CHANNEL production. RESET reads MySQL's comma
list into ResetReplicationStatement, apart from ResetStatement.
PurgeStatement gains BINARY LOGS TO and BEFORE, which MariaDB also reads,
with its MASTER LOGS; as in MySQL, BEFORE takes no subquery. Without a
dialect a bare RESET REPLICA stays a ResetStatement, and other dialects
keep their readings. STOP becomes a non-reserved keyword and remains
usable as a name, also in Oracle's PURGE TABLESPACE.

Signed-off-by: minleejae <mmj9808@gmail.com>

Author: minleejae

README.md

@@ -150,6 +150,7 @@ and missing syntax gets added on demand — [open an issue](https://github.com/J
 | **MySQL table maintenance** | `ANALYZE` · `CHECK` · `CHECKSUM` · `OPTIMIZE` · `REPAIR TABLE` |
 | **PostgreSQL maintenance** | `VACUUM` · `ANALYZE` · `CLUSTER` · `CHECKPOINT` |
 | **MySQL locks** | `LOCK TABLES` · `UNLOCK TABLES` · `LOCK INSTANCE FOR BACKUP` · `UNLOCK INSTANCE` |
+| **MySQL replication** | `CHANGE REPLICATION SOURCE TO` · `CHANGE REPLICATION FILTER` · `START`/`STOP REPLICA` · `RESET REPLICA` · `RESET BINARY LOGS AND GTIDS` · `PURGE BINARY LOGS` |
 | **MySQL account management** | `CREATE`/`ALTER`/`RENAME USER` · `CREATE ROLE` · `SET ROLE` · `SET DEFAULT ROLE` · `SET PASSWORD` |
 | **Tablespaces** | MySQL and PostgreSQL `CREATE TABLESPACE` · `ALTER TABLESPACE` |
 | **PostgreSQL RLS** | `CREATE POLICY` · `ALTER TABLE … ENABLE`/`DISABLE`/`FORCE`/`NO FORCE ROW LEVEL SECURITY` |

src/main/java/net/sf/jsqlparser/parser/feature/Feature.java

@@ -562,6 +562,12 @@ public enum Feature {
      * @see net.sf.jsqlparser.statement.PurgeStatement
      */
     purge,
+    /**
+     * MySQL/MariaDB "PURGE BINARY LOGS" and MariaDB "PURGE MASTER LOGS" statements are allowed
+     *
+     * @see net.sf.jsqlparser.statement.PurgeStatement
+     */
+    purgeBinaryLogs,
     /**
      * PostgreSQL "REASSIGN OWNED BY ... TO ..." statement is allowed
      *
@@ -782,6 +788,36 @@ public enum Feature {
      * @see BackupLockStatement
      */
     backupLock,
+    /**
+     * MySQL "CHANGE REPLICATION SOURCE TO" statement is allowed
+     *
+     * @see net.sf.jsqlparser.statement.replication.ChangeReplicationSourceStatement
+     */
+    changeReplicationSource,
+    /**
+     * MySQL "CHANGE REPLICATION FILTER" statement is allowed
+     *
+     * @see net.sf.jsqlparser.statement.replication.ChangeReplicationFilterStatement
+     */
+    changeReplicationFilter,
+    /**
+     * MySQL "START REPLICA" statement is allowed
+     *
+     * @see net.sf.jsqlparser.statement.replication.StartReplicaStatement
+     */
+    startReplica,
+    /**
+     * MySQL "STOP REPLICA" statement is allowed
+     *
+     * @see net.sf.jsqlparser.statement.replication.StopReplicaStatement
+     */
+    stopReplica,
+    /**
+     * MySQL "RESET BINARY LOGS AND GTIDS" and "RESET REPLICA" statements are allowed
+     *
+     * @see net.sf.jsqlparser.statement.replication.ResetReplicationStatement
+     */
+    resetReplication,
     /**
      * SQL "COMMENT ON" statement is allowed
      *

src/main/java/net/sf/jsqlparser/statement/PurgeObjectType.java

@@ -15,7 +15,11 @@
  * @author <a href="mailto:andreas@manticore-projects.com">Andreas Reichel</a>
  */
 public enum PurgeObjectType {
-    TABLE, INDEX, RECYCLEBIN, DBA_RECYCLEBIN, TABLESPACE;
+    TABLE, INDEX, RECYCLEBIN, DBA_RECYCLEBIN, TABLESPACE,
+    /** MySQL and MariaDB {@code PURGE BINARY LOGS} */
+    BINARY_LOGS,
+    /** MariaDB {@code PURGE MASTER LOGS}, which MySQL 8.4 removed */
+    MASTER_LOGS;
 
     public static PurgeObjectType from(String type) {
         return Enum.valueOf(PurgeObjectType.class, type.toUpperCase(Locale.ROOT));

src/main/java/net/sf/jsqlparser/statement/PurgeStatement.java

@@ -11,7 +11,9 @@
 package net.sf.jsqlparser.statement;
 
 import java.util.Objects;
+import java.util.function.Consumer;
 
+import net.sf.jsqlparser.expression.Expression;
 import net.sf.jsqlparser.parser.ASTNodeAccessImpl;
 import net.sf.jsqlparser.schema.Table;
 import net.sf.jsqlparser.statement.create.table.Index;
@@ -20,12 +22,23 @@
  * @author <a href="mailto:andreas@manticore-projects.com">Andreas Reichel</a>
  * @see <a href=
  *      "https://docs.oracle.com/cd/B19306_01/server.102/b14200/statements_9018.htm">Purge</a>
+ * @see <a href="https://dev.mysql.com/doc/refman/8.4/en/purge-binary-logs.html">MySQL PURGE BINARY
+ *      LOGS</a>
  */
 
 public class PurgeStatement extends ASTNodeAccessImpl implements Statement {
     private final PurgeObjectType purgeObjectType;
     private final Object object;
     private String userName;
+    private LogBound logBound;
+
+    /** How {@code PURGE BINARY LOGS} gives the first log it keeps. */
+    public enum LogBound {
+        /** {@code TO 'log_name'}: the logs before the named one */
+        TO,
+        /** {@code BEFORE datetime_expr}: the logs older than a point in time */
+        BEFORE
+    }
 
     public PurgeStatement(Table table) {
         this.purgeObjectType = PurgeObjectType.TABLE;
@@ -51,15 +64,43 @@ public PurgeStatement(PurgeObjectType purgeObjectType, String tableSpaceName, St
         this.userName = userName;
     }
 
+    /**
+     * MySQL {@code PURGE {BINARY | MASTER} LOGS {TO 'log_name' | BEFORE expr}}; for {@code TO} the
+     * expression is the string literal of the log name.
+     */
+    public PurgeStatement(PurgeObjectType purgeObjectType, LogBound logBound,
+            Expression logExpression) {
+        if (purgeObjectType != PurgeObjectType.BINARY_LOGS
+                && purgeObjectType != PurgeObjectType.MASTER_LOGS) {
+            throw new IllegalArgumentException(
+                    "Only BINARY LOGS and MASTER LOGS take TO or BEFORE: " + purgeObjectType);
+        }
+        this.purgeObjectType = purgeObjectType;
+        this.logBound = Objects.requireNonNull(logBound,
+                "The TO or BEFORE of the PURGE LOGS statement must not be null.");
+        this.object = Objects.requireNonNull(logExpression,
+                "The log name or point in time of the PURGE LOGS statement must not be null.");
+    }
+
     @Override
     public <T, S> T accept(StatementVisitor<T> statementVisitor, S context) {
         return statementVisitor.visit(this, context);
     }
 
+    public StringBuilder appendTo(StringBuilder builder) {
+        return appendTo(builder, builder::append);
+    }
+
     @SuppressWarnings({"PMD.MissingBreakInSwitch", "PMD.SwitchStmtsShouldHaveDefault",
             "PMD.CyclomaticComplexity"})
-    public StringBuilder appendTo(StringBuilder builder) {
+    public StringBuilder appendTo(StringBuilder builder, Consumer<Expression> expressions) {
         builder.append("PURGE ");
+        if (logBound != null) {
+            builder.append(purgeObjectType == PurgeObjectType.BINARY_LOGS ? "BINARY" : "MASTER")
+                    .append(" LOGS ").append(logBound).append(' ');
+            expressions.accept((Expression) object);
+            return builder;
+        }
 
         switch (purgeObjectType) {
             case RECYCLEBIN:
@@ -106,4 +147,17 @@ public PurgeObjectType getPurgeObjectType() {
     public Object getObject() {
         return object;
     }
+
+    /** TO or BEFORE of {@code PURGE BINARY LOGS}, null for the other object types. */
+    public LogBound getLogBound() {
+        return logBound;
+    }
+
+    /**
+     * The log name literal of {@code TO} or the expression of {@code BEFORE}, null for the other
+     * object types. It is the same object as {@link #getObject()}.
+     */
+    public Expression getLogExpression() {
+        return logBound != null ? (Expression) object : null;
+    }
 }

src/main/java/net/sf/jsqlparser/statement/StatementFeatureVisitor.java

@@ -101,6 +101,11 @@
 import net.sf.jsqlparser.statement.lock.BackupLockStatement;
 import net.sf.jsqlparser.statement.lock.LockStatement;
 import net.sf.jsqlparser.statement.lock.UnlockStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationFilterStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationSourceStatement;
+import net.sf.jsqlparser.statement.replication.ResetReplicationStatement;
+import net.sf.jsqlparser.statement.replication.StartReplicaStatement;
+import net.sf.jsqlparser.statement.replication.StopReplicaStatement;
 import net.sf.jsqlparser.statement.maintenance.CheckpointStatement;
 import net.sf.jsqlparser.statement.maintenance.ClusterStatement;
 import net.sf.jsqlparser.statement.maintenance.VacuumStatement;
@@ -682,6 +687,10 @@ public <S> Void visit(LoadDataStatement loadDataStatement, S context) {
 
     @Override
     public <S> Void visit(PurgeStatement purgeStatement, S context) {
+        if (purgeStatement.getLogBound() != null) {
+            // removes binary log files, a server operation like the replication statements
+            return sessionOnly();
+        }
         analysis.claimTopLevel();
         analysis.certain(StmtFeature.MODIFIES_SCHEMA, StmtFeature.MODIFIES_DATA);
         return null;
@@ -945,6 +954,32 @@ public <S> Void visit(BackupLockStatement backupLock, S context) {
         return transactionOnly();
     }
 
+    // the replication statements change server state, like ALTER SYSTEM, and no data or schema
+    @Override
+    public <S> Void visit(ChangeReplicationSourceStatement changeReplicationSource, S context) {
+        return sessionOnly();
+    }
+
+    @Override
+    public <S> Void visit(ChangeReplicationFilterStatement changeReplicationFilter, S context) {
+        return sessionOnly();
+    }
+
+    @Override
+    public <S> Void visit(StartReplicaStatement startReplica, S context) {
+        return sessionOnly();
+    }
+
+    @Override
+    public <S> Void visit(StopReplicaStatement stopReplica, S context) {
+        return sessionOnly();
+    }
+
+    @Override
+    public <S> Void visit(ResetReplicationStatement resetReplication, S context) {
+        return sessionOnly();
+    }
+
     private Void transactionOnly() {
         analysis.claimTopLevel();
         analysis.certain(StmtFeature.MODIFIES_TRANSACTION);

src/main/java/net/sf/jsqlparser/statement/StatementVisitor.java

@@ -98,6 +98,11 @@
 import net.sf.jsqlparser.statement.maintenance.VacuumStatement;
 import net.sf.jsqlparser.statement.merge.Merge;
 import net.sf.jsqlparser.statement.refresh.RefreshMaterializedViewStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationFilterStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationSourceStatement;
+import net.sf.jsqlparser.statement.replication.ResetReplicationStatement;
+import net.sf.jsqlparser.statement.replication.StartReplicaStatement;
+import net.sf.jsqlparser.statement.replication.StopReplicaStatement;
 import net.sf.jsqlparser.statement.reindex.Reindex;
 import net.sf.jsqlparser.statement.securitylabel.SecurityLabel;
 import net.sf.jsqlparser.statement.select.Select;
@@ -736,6 +741,46 @@ default void visit(BackupLockStatement backupLock) {
         visit(backupLock, null);
     }
 
+    default <S> T visit(ChangeReplicationSourceStatement changeReplicationSource, S context) {
+        return null;
+    }
+
+    default void visit(ChangeReplicationSourceStatement changeReplicationSource) {
+        visit(changeReplicationSource, null);
+    }
+
+    default <S> T visit(ChangeReplicationFilterStatement changeReplicationFilter, S context) {
+        return null;
+    }
+
+    default void visit(ChangeReplicationFilterStatement changeReplicationFilter) {
+        visit(changeReplicationFilter, null);
+    }
+
+    default <S> T visit(StartReplicaStatement startReplica, S context) {
+        return null;
+    }
+
+    default void visit(StartReplicaStatement startReplica) {
+        visit(startReplica, null);
+    }
+
+    default <S> T visit(StopReplicaStatement stopReplica, S context) {
+        return null;
+    }
+
+    default void visit(StopReplicaStatement stopReplica) {
+        visit(stopReplica, null);
+    }
+
+    default <S> T visit(ResetReplicationStatement resetReplication, S context) {
+        return null;
+    }
+
+    default void visit(ResetReplicationStatement resetReplication) {
+        visit(resetReplication, null);
+    }
+
     <S> T visit(CreatePolicy createPolicy, S context);
 
     default void visit(CreatePolicy createPolicy) {

src/main/java/net/sf/jsqlparser/statement/StatementVisitorAdapter.java

@@ -103,6 +103,11 @@
 import net.sf.jsqlparser.statement.merge.MergeOperationVisitor;
 import net.sf.jsqlparser.statement.merge.MergeOperationVisitorAdapter;
 import net.sf.jsqlparser.statement.refresh.RefreshMaterializedViewStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationFilterStatement;
+import net.sf.jsqlparser.statement.replication.ChangeReplicationSourceStatement;
+import net.sf.jsqlparser.statement.replication.ResetReplicationStatement;
+import net.sf.jsqlparser.statement.replication.StartReplicaStatement;
+import net.sf.jsqlparser.statement.replication.StopReplicaStatement;
 import net.sf.jsqlparser.statement.reindex.Reindex;
 import net.sf.jsqlparser.statement.securitylabel.SecurityLabel;
 import net.sf.jsqlparser.statement.select.FromItemVisitor;
@@ -569,6 +574,31 @@ public <S> T visit(BackupLockStatement backupLock, S context) {
         return null;
     }
 
+    @Override
+    public <S> T visit(ChangeReplicationSourceStatement changeReplicationSource, S context) {
+        return null;
+    }
+
+    @Override
+    public <S> T visit(ChangeReplicationFilterStatement changeReplicationFilter, S context) {
+        return null;
+    }
+
+    @Override
+    public <S> T visit(StartReplicaStatement startReplica, S context) {
+        return null;
+    }
+
+    @Override
+    public <S> T visit(StopReplicaStatement stopReplica, S context) {
+        return null;
+    }
+
+    @Override
+    public <S> T visit(ResetReplicationStatement resetReplication, S context) {
+        return null;
+    }
+
     @Override
     public <S> T visit(CreatePolicy createPolicy, S context) {
         fromItemVisitor.visitFromItem(createPolicy.getTable(), context);

src/main/java/net/sf/jsqlparser/statement/replication/ChangeReplicationFilterStatement.java

@@ -0,0 +1,71 @@
+/*-
+ * #%L
+ * JSQLParser library
+ * %%
+ * Copyright (C) 2004 - 2026 JSQLParser
+ * %%
+ * Dual licensed under GNU LGPL 2.1 or Apache License 2.0
+ * #L%
+ */
+package net.sf.jsqlparser.statement.replication;
+
+import java.util.ArrayList;
+import java.util.List;
+import net.sf.jsqlparser.expression.StringValue;
+import net.sf.jsqlparser.parser.ASTNodeAccessImpl;
+import net.sf.jsqlparser.statement.Statement;
+import net.sf.jsqlparser.statement.StatementVisitor;
+
+/**
+ * MySQL {@code CHANGE REPLICATION FILTER filter [, filter] ... [FOR CHANNEL 'channel']}.<br>
+ * Example:<br>
+ * CHANGE REPLICATION FILTER REPLICATE_DO_DB = (d1, d2), REPLICATE_REWRITE_DB = ((a, b))<br>
+ *
+ * @see <a href="https://dev.mysql.com/doc/refman/8.4/en/change-replication-filter.html">MySQL
+ *      CHANGE REPLICATION FILTER</a>
+ */
+public class ChangeReplicationFilterStatement extends ASTNodeAccessImpl implements Statement {
+    private List<ReplicationFilter> filters = new ArrayList<>();
+    private StringValue channel;
+
+    public List<ReplicationFilter> getFilters() {
+        return filters;
+    }
+
+    public void setFilters(List<ReplicationFilter> filters) {
+        this.filters = filters;
+    }
+
+    /** The channel of FOR CHANNEL, or null when the filters apply to every channel. */
+    public StringValue getChannel() {
+        return channel;
+    }
+
+    public void setChannel(StringValue channel) {
+        this.channel = channel;
+    }
+
+    public StringBuilder appendTo(StringBuilder builder) {
+        builder.append("CHANGE REPLICATION FILTER ");
+        for (int i = 0; i < filters.size(); i++) {
+            if (i > 0) {
+                builder.append(", ");
+            }
+            filters.get(i).appendTo(builder);
+        }
+        if (channel != null) {
+            builder.append(" FOR CHANNEL ").append(channel);
+        }
+        return builder;
+    }
+
+    @Override
+    public String toString() {
+        return appendTo(new StringBuilder()).toString();
+    }
+
+    @Override
+    public <T, S> T accept(StatementVisitor<T> statementVisitor, S context) {
+        return statementVisitor.visit(this, context);
+    }
+}