Support flush statement SQL bind

infra/binder/core/src/main/java/org/apache/shardingsphere/infra/binder/context/SQLStatementContextFactory.java

@@ -22,6 +22,7 @@
 import org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
 import org.apache.shardingsphere.infra.binder.context.statement.type.CommonSQLStatementContext;
 import org.apache.shardingsphere.infra.binder.context.statement.type.dal.ExplainStatementContext;
+import org.apache.shardingsphere.infra.binder.context.statement.type.dal.FlushStatementContext;
 import org.apache.shardingsphere.infra.binder.context.statement.type.ddl.AlterViewStatementContext;
 import org.apache.shardingsphere.infra.binder.context.statement.type.ddl.CreateProcedureStatementContext;
 import org.apache.shardingsphere.infra.binder.context.statement.type.ddl.CreateViewStatementContext;
@@ -38,6 +39,7 @@
 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.ExplainStatement;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dcl.DCLStatement;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.ddl.CursorStatement;
 import org.apache.shardingsphere.sql.parser.statement.core.statement.type.ddl.DDLStatement;
@@ -132,6 +134,9 @@ private static SQLStatementContext getDALStatementContext(final ShardingSphereMe
         if (sqlStatement instanceof ExplainStatement) {
             return new ExplainStatementContext(metaData, (ExplainStatement) sqlStatement, params, currentDatabaseName);
         }
+        if (sqlStatement instanceof FlushStatement) {
+            return new FlushStatementContext((FlushStatement) sqlStatement);
+        }
         return new CommonSQLStatementContext(sqlStatement);
     }
 }

infra/binder/core/src/main/java/org/apache/shardingsphere/infra/binder/context/statement/type/dal/FlushStatementContext.java

@@ -0,0 +1,42 @@
+/*
+ * 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.context.statement.type.dal;
+
+import lombok.Getter;
+import org.apache.shardingsphere.infra.binder.context.segment.table.TablesContext;
+import org.apache.shardingsphere.infra.binder.context.statement.SQLStatementContext;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.attribute.type.TableSQLStatementAttribute;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
+
+import java.util.Collections;
+
+/**
+ * Flush statement context.
+ */
+@Getter
+public final class FlushStatementContext implements SQLStatementContext {
+
+    private final FlushStatement sqlStatement;
+
+    private final TablesContext tablesContext;
+
+    public FlushStatementContext(final FlushStatement sqlStatement) {
+        this.sqlStatement = sqlStatement;
+        tablesContext = new TablesContext(sqlStatement.getAttributes().findAttribute(TableSQLStatementAttribute.class).map(TableSQLStatementAttribute::getTables).orElse(Collections.emptyList()));
+    }
+}

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

@@ -0,0 +1,70 @@
+/*
+ * 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.sql.parser.statement.core.segment.generic.table.SimpleTableSegment;
+import org.apache.shardingsphere.sql.parser.statement.core.statement.type.dal.FlushStatement;
+
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+/**
+ * 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 = tables.stream()
+                .map(each -> SimpleTableSegmentBinder.bind(each, binderContext, tableBinderContexts))
+                .collect(Collectors.toList());
+
+        return copyFlushStatement(sqlStatement, boundTables);
+    }
+
+    private FlushStatement copyFlushStatement(final FlushStatement sqlStatement, final Collection<SimpleTableSegment> tables) {
+        FlushStatement result = new FlushStatement(sqlStatement.getDatabaseType()) {
+
+            @Override
+            public Collection<SimpleTableSegment> getTables() {
+                return tables;
+            }
+
+            @Override
+            public boolean isFlushTable() {
+                return sqlStatement.isFlushTable();
+            }
+        };
+        SQLStatementCopyUtils.copyAttributes(sqlStatement, result);
+        return result;
+    }
+}

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>