Support flush statement SQL bind

RELEASE-NOTES.md

@@ -119,6 +119,7 @@
 1. SQL Binder: Support AnalyzeTable statement SQL bind - [#35954](https://github.com/apache/shardingsphere/pull/35954)
 1. SQL Binder: Support Comment statement SQL bind - [#36012](https://github.com/apache/shardingsphere/pull/36012)
 1. SQL Binder: Support Prepare statement SQL bind - [#36064](https://github.com/apache/shardingsphere/pull/36064)
+1. SQL Binder: Support Flush statement SQL bind - [#36036](https://github.com/apache/shardingsphere/pull/36036)
 1. SQL Binder: Support Revoke statement SQL bind - [#36124](https://github.com/apache/shardingsphere/pull/36124)
 1. SQL Binder: Add alter table metadata check - [#35877](https://github.com/apache/shardingsphere/pull/35877)
 1. SQL Router: Add check for select with union all routing to multi data sources - [#35037](https://github.com/apache/shardingsphere/pull/35037)

infra/binder/core/src/main/java/org/apache/shardingsphere/infra/binder/engine/statement/dal/FlushStatementBinder.java

@@ -0,0 +1,68 @@
+/*
+ * 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.shardingsphere.infra.binder.engine.statement.dal;
+
+import com.cedarsoftware.util.CaseInsensitiveMap.CaseInsensitiveString;
+import com.google.common.collect.LinkedHashMultimap;
+import com.google.common.collect.Multimap;
+import org.apache.shardingsphere.infra.binder.engine.segment.dml.from.context.TableSegmentBinderContext;
+import org.apache.shardingsphere.infra.binder.engine.segment.dml.from.type.SimpleTableSegmentBinder;
+import org.apache.shardingsphere.infra.binder.engine.statement.SQLStatementBinder;
+import org.apache.shardingsphere.infra.binder.engine.statement.SQLStatementBinderContext;
+import org.apache.shardingsphere.infra.binder.engine.statement.SQLStatementCopyUtils;
+import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import org.apache.shardingsphere.sql.parser.statement.core.segment.generic.table.SimpleTableSegment;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+/**
+ * Flush statement binder.
+ */
+public final class FlushStatementBinder implements SQLStatementBinder<FlushStatement> {
+
+    @Override
+    public FlushStatement bind(final FlushStatement sqlStatement, final SQLStatementBinderContext binderContext) {
+        Collection<SimpleTableSegment> tables = sqlStatement.getTables();
+        if (tables.isEmpty()) {
+            return sqlStatement;
+        }
+        Multimap<CaseInsensitiveString, TableSegmentBinderContext> tableBinderContexts = LinkedHashMultimap.create();
+        Collection<SimpleTableSegment> boundTables = new ArrayList<>();
+        for (SimpleTableSegment each : tables) {
+            boundTables.add(SimpleTableSegmentBinder.bind(each, binderContext, tableBinderContexts));
+        }
+        return copyFlushStatement(sqlStatement, boundTables);
+    }
+
+    private FlushStatement copyFlushStatement(final FlushStatement sqlStatement, final Collection<SimpleTableSegment> tables) {
+        if (tables.equals(sqlStatement.getTables())) {
+            return sqlStatement;
+        }
+        try {
+            FlushStatement result = sqlStatement.getClass()
+                    .getDeclaredConstructor(DatabaseType.class, Collection.class, boolean.class)
+                    .newInstance(sqlStatement.getDatabaseType(), tables, sqlStatement.isFlushTable());
+            SQLStatementCopyUtils.copyAttributes(sqlStatement, result);
+            return result;
+        } catch (final ReflectiveOperationException ex) {
+            return sqlStatement;
+        }
+    }
+}

infra/binder/core/src/main/java/org/apache/shardingsphere/infra/binder/engine/type/DALStatementBindEngine.java

@@ -20,9 +20,11 @@
 import org.apache.shardingsphere.infra.binder.engine.statement.SQLStatementBinderContext;
 import org.apache.shardingsphere.infra.binder.engine.statement.dal.AnalyzeTableStatementBinder;
 import org.apache.shardingsphere.infra.binder.engine.statement.dal.ExplainStatementBinder;
+import org.apache.shardingsphere.infra.binder.engine.statement.dal.FlushStatementBinder;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.AnalyzeTableStatement;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.DALStatement;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.ExplainStatement;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
 
 /**
  * DAL statement bind engine.
@@ -43,6 +45,9 @@ public DALStatement bind(final DALStatement statement, final SQLStatementBinderC
         if (statement instanceof ExplainStatement) {
             return new ExplainStatementBinder().bind((ExplainStatement) statement, binderContext);
         }
+        if (statement instanceof FlushStatement) {
+            return new FlushStatementBinder().bind((FlushStatement) statement, binderContext);
+        }
         return statement;
     }
 }

parser/sql/statement/core/src/main/java/org/apache/shardingsphere/sql/parser/statement/core/statement/type/dal/FlushStatement.java

@@ -0,0 +1,47 @@
+/*
+ * 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.shardingsphere.sql.parser.statement.core.statement.type.dal;
+
+import org.apache.shardingsphere.infra.database.core.type.DatabaseType;
+import org.apache.shardingsphere.sql.parser.statement.core.segment.generic.table.SimpleTableSegment;
+
+import java.util.Collection;
+
+/**
+ * Flush statement.
+ */
+public abstract class FlushStatement extends DALStatement {
+
+    public FlushStatement(final DatabaseType databaseType) {
+        super(databaseType);
+    }
+
+    /**
+     * Get tables.
+     *
+     * @return tables
+     */
+    public abstract Collection<SimpleTableSegment> getTables();
+
+    /**
+     * Is flush table.
+     *
+     * @return true if flush table, false otherwise
+     */
+    public abstract boolean isFlushTable();
+}

parser/sql/statement/type/mysql/src/main/java/org/apache/shardingsphere/sql/parser/statement/mysql/dal/MySQLFlushStatement.java

@@ -22,15 +22,15 @@
 import org.apache.shardingsphere.sql.parser.statement.core.segment.generic.table.SimpleTableSegment;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.attribute.SQLStatementAttributes;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.attribute.type.TableSQLStatementAttribute;
-import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.DALStatement;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
 
 import java.util.Collection;
 
 /**
  * Flush statement for MySQL.
  */
 @Getter
-public final class MySQLFlushStatement extends DALStatement {
+public final class MySQLFlushStatement extends FlushStatement {
 
     private final Collection<SimpleTableSegment> tables;
 

test/it/binder/src/test/resources/cases/dal/flush.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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-parser-test-cases>
+    <flush sql-case-id="flush_tables" flush-table="true"></flush>
+
+    <flush sql-case-id="flush_tables_with_single_table" flush-table="true">
+        <table name="t_order" start-index="13" stop-index="19">
+            <table-bound>
+                <original-database name="foo_db_1" />
+                <original-schema name="foo_db_1" />
+            </table-bound>
+        </table>
+    </flush>
+
+    <flush sql-case-id="flush_tables_with_multiple_tables" flush-table="true">
+        <table name="t_order" start-index="13" stop-index="19">
+            <table-bound>
+                <original-database name="foo_db_1" />
+                <original-schema name="foo_db_1" />
+            </table-bound>
+        </table>
+        <table name="t_order_item" start-index="22" stop-index="33">
+            <table-bound>
+                <original-database name="foo_db_1" />
+                <original-schema name="foo_db_1" />
+            </table-bound>
+        </table>
+    </flush>
+
+    <flush sql-case-id="flush_tables_with_owner" flush-table="true">
+        <table name="t_product" start-delimiter="`" end-delimiter="`" start-index="13" stop-index="34">
+            <owner name="foo_db_2" start-delimiter="`" end-delimiter="`" start-index="13" stop-index="22" />
+            <table-bound>
+                <original-database name="foo_db_2" />
+                <original-schema name="foo_db_2" start-delimiter="`" end-delimiter="`" />
+            </table-bound>
+        </table>
+    </flush>
+
+    <flush sql-case-id="flush_logs" flush-table="false"></flush>
+
+    <flush sql-case-id="flush_privileges" flush-table="false"></flush>
+</sql-parser-test-cases>

test/it/binder/src/test/resources/sqls/dal/flush.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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-cases>
+    <sql-case id="flush_tables" value="FLUSH TABLES" db-types="MySQL" />
+    <sql-case id="flush_tables_with_single_table" value="FLUSH TABLES t_order" db-types="MySQL" />
+    <sql-case id="flush_tables_with_multiple_tables" value="FLUSH TABLES t_order, t_order_item" db-types="MySQL" />
+    <sql-case id="flush_tables_with_owner" value="FLUSH TABLES `foo_db_2`.`t_product`" db-types="MySQL" />
+    <sql-case id="flush_logs" value="FLUSH LOGS" db-types="MySQL" />
+    <sql-case id="flush_privileges" value="FLUSH PRIVILEGES" db-types="MySQL" />
+</sql-cases>