Support MySQL FLUSH statement (#1076)

src/ast/mod.rs

@@ -1777,6 +1777,20 @@ pub enum Statement {
         into: Option<ObjectName>,
     },
     /// ```sql
+    /// FLUSH [NO_WRITE_TO_BINLOG | LOCAL] flush_option [, flush_option] ... | tables_option
+    /// ```
+    ///
+    /// Note: this is a Mysql-specific statement,
+    /// but may also compatible with other SQL.
+    Flush {
+        object_type: FlushType,
+        location: Option<FlushLocation>,
+        channel: Option<String>,
+        read_lock: bool,
+        export: bool,
+        tables: Vec<ObjectName>,
+    },
+    /// ```sql
     /// DISCARD [ ALL | PLANS | SEQUENCES | TEMPORARY | TEMP ]
     /// ```
     ///
@@ -2199,6 +2213,36 @@ impl fmt::Display for Statement {
     #[allow(clippy::cognitive_complexity)]
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
+            Statement::Flush {
+                object_type,
+                location,
+                channel,
+                read_lock,
+                export,
+                tables,
+            } => {
+                write!(f, "FLUSH")?;
+                if let Some(location) = location {
+                    write!(f, " {location}")?;
+                }
+                write!(f, " {object_type}")?;
+
+                if let Some(channel) = channel {
+                    write!(f, " FOR CHANNEL {channel}")?;
+                }
+
+                write!(
+                    f,
+                    "{tables}{read}{export}",
+                    tables = if !tables.is_empty() {
+                        " ".to_string() + &display_comma_separated(tables).to_string()
+                    } else {
+                        "".to_string()
+                    },
+                    export = if *export { " FOR EXPORT" } else { "" },
+                    read = if *read_lock { " WITH READ LOCK" } else { "" }
+                )
+            }
             Statement::Kill { modifier, id } => {
                 write!(f, "KILL ")?;
 
@@ -4818,6 +4862,62 @@ impl fmt::Display for DiscardObject {
     }
 }
 
+#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum FlushType {
+    BinaryLogs,
+    EngineLogs,
+    ErrorLogs,
+    GeneralLogs,
+    Hosts,
+    Logs,
+    Privileges,
+    OptimizerCosts,
+    RelayLogs,
+    SlowLogs,
+    Status,
+    UserResources,
+    Tables,
+}
+
+impl fmt::Display for FlushType {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            FlushType::BinaryLogs => f.write_str("BINARY LOGS"),
+            FlushType::EngineLogs => f.write_str("ENGINE LOGS"),
+            FlushType::ErrorLogs => f.write_str("ERROR LOGS"),
+            FlushType::GeneralLogs => f.write_str("GENERAL LOGS"),
+            FlushType::Hosts => f.write_str("HOSTS"),
+            FlushType::Logs => f.write_str("LOGS"),
+            FlushType::Privileges => f.write_str("PRIVILEGES"),
+            FlushType::OptimizerCosts => f.write_str("OPTIMIZER_COSTS"),
+            FlushType::RelayLogs => f.write_str("RELAY LOGS"),
+            FlushType::SlowLogs => f.write_str("SLOW LOGS"),
+            FlushType::Status => f.write_str("STATUS"),
+            FlushType::UserResources => f.write_str("USER_RESOURCES"),
+            FlushType::Tables => f.write_str("TABLES"),
+        }
+    }
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum FlushLocation {
+    NoWriteToBinlog,
+    Local,
+}
+
+impl fmt::Display for FlushLocation {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            FlushLocation::NoWriteToBinlog => f.write_str("NO_WRITE_TO_BINLOG"),
+            FlushLocation::Local => f.write_str("LOCAL"),
+        }
+    }
+}
+
 /// Optional context modifier for statements that can be or `LOCAL`, or `SESSION`.
 #[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]

src/keywords.rs

@@ -137,6 +137,7 @@ define_keywords!(
     CENTURY,
     CHAIN,
     CHANGE,
+    CHANNEL,
     CHAR,
     CHARACTER,
     CHARACTERS,
@@ -265,6 +266,7 @@ define_keywords!(
     EXPANSION,
     EXPLAIN,
     EXPLICIT,
+    EXPORT,
     EXTENDED,
     EXTERNAL,
     EXTRACT,
@@ -283,6 +285,7 @@ define_keywords!(
     FLOAT64,
     FLOAT8,
     FLOOR,
+    FLUSH,
     FOLLOWING,
     FOR,
     FORCE,
@@ -302,6 +305,7 @@ define_keywords!(
     FUNCTION,
     FUNCTIONS,
     FUSION,
+    GENERAL,
     GENERATE,
     GENERATED,
     GEOGRAPHY,
@@ -320,6 +324,7 @@ define_keywords!(
     HISTORY,
     HIVEVAR,
     HOLD,
+    HOSTS,
     HOUR,
     HOURS,
     IDENTITY,
@@ -385,6 +390,7 @@ define_keywords!(
     LOCK,
     LOCKED,
     LOGIN,
+    LOGS,
     LOWER,
     LOW_PRIORITY,
     MACRO,
@@ -439,6 +445,7 @@ define_keywords!(
     NOT,
     NOTHING,
     NOWAIT,
+    NO_WRITE_TO_BINLOG,
     NTH_VALUE,
     NTILE,
     NULL,
@@ -458,6 +465,7 @@ define_keywords!(
     OPEN,
     OPERATOR,
     OPTIMIZE,
+    OPTIMIZER_COSTS,
     OPTION,
     OPTIONS,
     OR,
@@ -531,6 +539,7 @@ define_keywords!(
     REGR_SXY,
     REGR_SYY,
     RELATIVE,
+    RELAY,
     RELEASE,
     RENAME,
     REORG,
@@ -581,6 +590,7 @@ define_keywords!(
     SHOW,
     SIMILAR,
     SKIP,
+    SLOW,
     SMALLINT,
     SNAPSHOT,
     SOME,
@@ -598,6 +608,7 @@ define_keywords!(
     START,
     STATIC,
     STATISTICS,
+    STATUS,
     STDDEV_POP,
     STDDEV_SAMP,
     STDIN,
@@ -673,6 +684,7 @@ define_keywords!(
     USAGE,
     USE,
     USER,
+    USER_RESOURCES,
     USING,
     UUID,
     VACUUM,

src/parser/mod.rs

@@ -472,6 +472,7 @@ impl<'a> Parser<'a> {
         match &next_token.token {
             Token::Word(w) => match w.keyword {
                 Keyword::KILL => Ok(self.parse_kill()?),
+                Keyword::FLUSH => Ok(self.parse_flush()?),
                 Keyword::DESCRIBE => Ok(self.parse_explain(true)?),
                 Keyword::EXPLAIN => Ok(self.parse_explain(false)?),
                 Keyword::ANALYZE => Ok(self.parse_analyze()?),
@@ -534,6 +535,93 @@ impl<'a> Parser<'a> {
         }
     }
 
+    pub fn parse_flush(&mut self) -> Result<Statement, ParserError> {
+        let mut channel = None;
+        let mut tables: Vec<ObjectName> = vec![];
+        let mut read_lock = false;
+        let mut export = false;
+
+        if !dialect_of!(self is MySqlDialect | GenericDialect) {
+            return parser_err!("Unsupported statement FLUSH", self.peek_token().location);
+        }
+
+        let location = if self.parse_keyword(Keyword::NO_WRITE_TO_BINLOG) {
+            Some(FlushLocation::NoWriteToBinlog)
+        } else if self.parse_keyword(Keyword::LOCAL) {
+            Some(FlushLocation::Local)
+        } else {
+            None
+        };
+
+        let object_type = if self.parse_keywords(&[Keyword::BINARY, Keyword::LOGS]) {
+            FlushType::BinaryLogs
+        } else if self.parse_keywords(&[Keyword::ENGINE, Keyword::LOGS]) {
+            FlushType::EngineLogs
+        } else if self.parse_keywords(&[Keyword::ERROR, Keyword::LOGS]) {
+            FlushType::ErrorLogs
+        } else if self.parse_keywords(&[Keyword::GENERAL, Keyword::LOGS]) {
+            FlushType::GeneralLogs
+        } else if self.parse_keywords(&[Keyword::HOSTS]) {
+            FlushType::Hosts
+        } else if self.parse_keyword(Keyword::PRIVILEGES) {
+            FlushType::Privileges
+        } else if self.parse_keyword(Keyword::OPTIMIZER_COSTS) {
+            FlushType::OptimizerCosts
+        } else if self.parse_keywords(&[Keyword::RELAY, Keyword::LOGS]) {
+            if self.parse_keywords(&[Keyword::FOR, Keyword::CHANNEL]) {
+                channel = Some(self.parse_object_name().unwrap().to_string());
+            }
+            FlushType::RelayLogs
+        } else if self.parse_keywords(&[Keyword::SLOW, Keyword::LOGS]) {
+            FlushType::SlowLogs
+        } else if self.parse_keyword(Keyword::STATUS) {
+            FlushType::Status
+        } else if self.parse_keyword(Keyword::USER_RESOURCES) {
+            FlushType::UserResources
+        } else if self.parse_keywords(&[Keyword::LOGS]) {
+            FlushType::Logs
+        } else if self.parse_keywords(&[Keyword::TABLES]) {
+            loop {
+                let next_token = self.next_token();
+                match &next_token.token {
+                    Token::Word(w) => match w.keyword {
+                        Keyword::WITH => {
+                            read_lock = self.parse_keywords(&[Keyword::READ, Keyword::LOCK]);
+                        }
+                        Keyword::FOR => {
+                            export = self.parse_keyword(Keyword::EXPORT);
+                        }
+                        Keyword::NoKeyword => {
+                            self.prev_token();
+                            tables = self.parse_comma_separated(Parser::parse_object_name)?;
+                        }
+                        _ => {}
+                    },
+                    _ => {
+                        break;
+                    }
+                }
+            }
+
+            FlushType::Tables
+        } else {
+            return self.expected(
+                "BINARY LOGS, ENGINE LOGS, ERROR LOGS, GENERAL LOGS, HOSTS, LOGS, PRIVILEGES, OPTIMIZER_COSTS,\
+                 RELAY LOGS [FOR CHANNEL channel], SLOW LOGS, STATUS, USER_RESOURCES",
+                self.peek_token(),
+            );
+        };
+
+        Ok(Statement::Flush {
+            object_type,
+            location,
+            channel,
+            read_lock,
+            export,
+            tables,
+        })
+    }
+
     pub fn parse_msck(&mut self) -> Result<Statement, ParserError> {
         let repair = self.parse_keyword(Keyword::REPAIR);
         self.expect_keyword(Keyword::TABLE)?;

tests/sqlparser_mysql.rs

@@ -47,6 +47,176 @@ fn parse_literal_string() {
     );
 }
 
+#[test]
+fn parse_flush() {
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH OPTIMIZER_COSTS"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::OptimizerCosts,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH BINARY LOGS"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::BinaryLogs,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH ENGINE LOGS"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::EngineLogs,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH ERROR LOGS"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::ErrorLogs,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH NO_WRITE_TO_BINLOG GENERAL LOGS"),
+        Statement::Flush {
+            location: Some(FlushLocation::NoWriteToBinlog),
+            object_type: FlushType::GeneralLogs,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH RELAY LOGS FOR CHANNEL test"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::RelayLogs,
+            channel: Some("test".to_string()),
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH LOCAL SLOW LOGS"),
+        Statement::Flush {
+            location: Some(FlushLocation::Local),
+            object_type: FlushType::SlowLogs,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH TABLES `mek`.`table1`, table2"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::Tables,
+            channel: None,
+            read_lock: false,
+            export: false,
+            tables: vec![
+                ObjectName(vec![
+                    Ident {
+                        value: "mek".to_string(),
+                        quote_style: Some('`')
+                    },
+                    Ident {
+                        value: "table1".to_string(),
+                        quote_style: Some('`')
+                    }
+                ]),
+                ObjectName(vec![Ident {
+                    value: "table2".to_string(),
+                    quote_style: None
+                }])
+            ]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH TABLES WITH READ LOCK"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::Tables,
+            channel: None,
+            read_lock: true,
+            export: false,
+            tables: vec![]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH TABLES `mek`.`table1`, table2 WITH READ LOCK"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::Tables,
+            channel: None,
+            read_lock: true,
+            export: false,
+            tables: vec![
+                ObjectName(vec![
+                    Ident {
+                        value: "mek".to_string(),
+                        quote_style: Some('`')
+                    },
+                    Ident {
+                        value: "table1".to_string(),
+                        quote_style: Some('`')
+                    }
+                ]),
+                ObjectName(vec![Ident {
+                    value: "table2".to_string(),
+                    quote_style: None
+                }])
+            ]
+        }
+    );
+    assert_eq!(
+        mysql_and_generic().verified_stmt("FLUSH TABLES `mek`.`table1`, table2 FOR EXPORT"),
+        Statement::Flush {
+            location: None,
+            object_type: FlushType::Tables,
+            channel: None,
+            read_lock: false,
+            export: true,
+            tables: vec![
+                ObjectName(vec![
+                    Ident {
+                        value: "mek".to_string(),
+                        quote_style: Some('`')
+                    },
+                    Ident {
+                        value: "table1".to_string(),
+                        quote_style: Some('`')
+                    }
+                ]),
+                ObjectName(vec![Ident {
+                    value: "table2".to_string(),
+                    quote_style: None
+                }])
+            ]
+        }
+    );
+}
+
 #[test]
 fn parse_show_columns() {
     let table_name = ObjectName(vec![Ident::new("mytable")]);