Parse FOR TIMESTAMP AS OF and FOR VERSION AS OF table versions

src/ast/query.rs

@@ -2630,6 +2630,18 @@ pub enum TableVersion {
     /// Databricks supports this syntax.
     /// For example: `SELECT * FROM tbl VERSION AS OF 2`
     VersionAsOf(Expr),
+    /// When the table version is defined using `FOR TIMESTAMP AS OF`.
+    /// Trino reads Iceberg and Delta tables at a point in time this way.
+    /// For example: `SELECT * FROM tbl FOR TIMESTAMP AS OF TIMESTAMP '2026-01-01 00:00:00 UTC'`
+    ///
+    /// See <https://trino.io/docs/current/connector/iceberg.html#time-travel-queries>
+    ForTimestampAsOf(Expr),
+    /// When the table version is defined using `FOR VERSION AS OF`.
+    /// Trino accepts a snapshot id or, for Iceberg, a branch or tag name.
+    /// For example: `SELECT * FROM tbl FOR VERSION AS OF 8954597067493422955`
+    ///
+    /// See <https://trino.io/docs/current/connector/iceberg.html#time-travel-queries>
+    ForVersionAsOf(Expr),
     /// When the table version is defined using a function.
     /// For example: `SELECT * FROM tbl AT(TIMESTAMP => '2020-08-14 09:30:00')`
     Function(Expr),
@@ -2658,6 +2670,8 @@ impl Display for TableVersion {
             TableVersion::ForSystemTimeAsOf(e) => write!(f, "FOR SYSTEM_TIME AS OF {e}")?,
             TableVersion::TimestampAsOf(e) => write!(f, "TIMESTAMP AS OF {e}")?,
             TableVersion::VersionAsOf(e) => write!(f, "VERSION AS OF {e}")?,
+            TableVersion::ForTimestampAsOf(e) => write!(f, "FOR TIMESTAMP AS OF {e}")?,
+            TableVersion::ForVersionAsOf(e) => write!(f, "FOR VERSION AS OF {e}")?,
             TableVersion::Function(func) => write!(f, "{func}")?,
             TableVersion::Changes { changes, at, end } => {
                 write!(f, "{changes} {at}")?;

src/dialect/mod.rs

@@ -31,6 +31,7 @@ mod snowflake;
 mod spark;
 mod sqlite;
 mod teradata;
+mod trino;
 
 use core::any::{Any, TypeId};
 use core::fmt::Debug;
@@ -56,6 +57,7 @@ pub use self::snowflake::SnowflakeDialect;
 pub use self::spark::SparkSqlDialect;
 pub use self::sqlite::SQLiteDialect;
 pub use self::teradata::TeradataDialect;
+pub use self::trino::TrinoDialect;
 
 /// Macro for streamlining the creation of derived `Dialect` objects.
 /// The generated struct includes `new()` and `default()` constructors.
@@ -1412,6 +1414,17 @@ pub trait Dialect: Debug + Any {
         false
     }
 
+    /// Returns true if the dialect spells table versions with a leading
+    /// `FOR`: `FOR TIMESTAMP AS OF <expr>` and `FOR VERSION AS OF <expr>`,
+    /// as Trino does for Iceberg and Delta time travel.
+    ///
+    /// Only consulted when [`Self::supports_table_versioning`] is true.
+    ///
+    /// See <https://trino.io/docs/current/connector/iceberg.html#time-travel-queries>
+    fn supports_for_table_version(&self) -> bool {
+        false
+    }
+
     /// Returns true if this dialect supports the E'...' syntax for string literals
     ///
     /// Postgres: <https://www.postgresql.org/docs/current/sql-syntax-lexical.html#SQL-SYNTAX-STRINGS-ESCAPE>
@@ -1927,6 +1940,7 @@ pub fn dialect_from_str(dialect_name: impl AsRef<str>) -> Option<Box<dyn Dialect
         "spark" | "sparksql" => Some(Box::new(SparkSqlDialect {})),
         "oracle" => Some(Box::new(OracleDialect {})),
         "teradata" => Some(Box::new(TeradataDialect {})),
+        "trino" => Some(Box::new(TrinoDialect {})),
         _ => None,
     }
 }
@@ -1981,6 +1995,7 @@ mod tests {
         assert!(parse_dialect("DataBricks").is::<DatabricksDialect>());
         assert!(parse_dialect("databricks").is::<DatabricksDialect>());
         assert!(parse_dialect("teradata").is::<TeradataDialect>());
+        assert!(parse_dialect("trino").is::<TrinoDialect>());
         assert!(parse_dialect("Teradata").is::<TeradataDialect>());
 
         // error cases

src/dialect/trino.rs

@@ -0,0 +1,115 @@
+// 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.
+
+use crate::dialect::Dialect;
+
+/// A [`Dialect`] for [Trino](https://trino.io/docs/current/language.html),
+/// the distributed SQL engine that descends from Presto.
+///
+/// Trino follows the SQL standard closely: identifiers are delimited with
+/// double quotes only, aggregates take `FILTER (WHERE ...)`, higher-order
+/// functions take `x -> ...` lambdas, and row pattern recognition, table
+/// sampling, table versioning (time travel) and parenthesized `EXPLAIN`
+/// options are all part of the grammar.
+///
+/// See <https://trino.io/docs/current/sql.html>.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
+#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
+pub struct TrinoDialect;
+
+impl Dialect for TrinoDialect {
+    /// Trino delimits identifiers with double quotes only; backquotes are a
+    /// syntax error, the most common mistake of users arriving from
+    /// MySQL, ClickHouse or Spark.
+    ///
+    /// See <https://trino.io/docs/current/language/reserved.html>
+    fn is_delimited_identifier_start(&self, ch: char) -> bool {
+        ch == '"'
+    }
+
+    fn is_identifier_start(&self, ch: char) -> bool {
+        ch.is_ascii_alphabetic() || ch == '_'
+    }
+
+    fn is_identifier_part(&self, ch: char) -> bool {
+        ch.is_ascii_alphanumeric() || ch == '_'
+    }
+
+    /// See <https://trino.io/docs/current/sql/select.html#filter>
+    fn supports_filter_during_aggregation(&self) -> bool {
+        true
+    }
+
+    /// `GROUP BY` accepts arbitrary expressions, `GROUPING SETS`, `CUBE`
+    /// and `ROLLUP`.
+    ///
+    /// See <https://trino.io/docs/current/sql/select.html#group-by-clause>
+    fn supports_group_by_expr(&self) -> bool {
+        true
+    }
+
+    /// See <https://trino.io/docs/current/functions/lambda.html>
+    fn supports_lambda_functions(&self) -> bool {
+        true
+    }
+
+    /// See <https://trino.io/docs/current/sql/match-recognize.html>
+    fn supports_match_recognize(&self) -> bool {
+        true
+    }
+
+    /// Trino reads Iceberg and Delta tables at a point in time with
+    /// `FOR TIMESTAMP AS OF` and `FOR VERSION AS OF`.
+    ///
+    /// See <https://trino.io/docs/current/connector/iceberg.html#time-travel-queries>
+    fn supports_table_versioning(&self) -> bool {
+        true
+    }
+
+    /// See <https://trino.io/docs/current/connector/iceberg.html#time-travel-queries>
+    fn supports_for_table_version(&self) -> bool {
+        true
+    }
+
+    /// `EXPLAIN (TYPE IO, FORMAT JSON) ...` and friends.
+    ///
+    /// See <https://trino.io/docs/current/sql/explain.html>
+    fn supports_explain_with_utility_options(&self) -> bool {
+        true
+    }
+
+    /// Named arguments to table functions use `=>`, e.g.
+    /// `TABLE(sequence(start => 1, stop => 10))`.
+    ///
+    /// See <https://trino.io/docs/current/functions/table.html>
+    fn supports_named_fn_args_with_rarrow_operator(&self) -> bool {
+        true
+    }
+
+    /// See <https://trino.io/docs/current/sql/comment.html>
+    fn supports_comment_on(&self) -> bool {
+        true
+    }
+
+    /// `SHOW TABLES FROM schema LIKE '%x%'`: the source comes before the
+    /// pattern.
+    ///
+    /// See <https://trino.io/docs/current/sql/show-tables.html>
+    fn supports_show_like_before_in(&self) -> bool {
+        false
+    }
+}

src/parser/mod.rs

@@ -17389,6 +17389,22 @@ impl<'a> Parser<'a> {
             {
                 let expr = self.parse_expr()?;
                 return Ok(Some(TableVersion::ForSystemTimeAsOf(expr)));
+            } else if self.dialect.supports_for_table_version()
+                && self.parse_keywords(&[
+                    Keyword::FOR,
+                    Keyword::TIMESTAMP,
+                    Keyword::AS,
+                    Keyword::OF,
+                ])
+            {
+                let expr = self.parse_expr()?;
+                return Ok(Some(TableVersion::ForTimestampAsOf(expr)));
+            } else if self.dialect.supports_for_table_version()
+                && self.parse_keywords(&[Keyword::FOR, Keyword::VERSION, Keyword::AS, Keyword::OF])
+            {
+                // A snapshot id, or a branch/tag name on Iceberg.
+                let expr = self.parse_expr()?;
+                return Ok(Some(TableVersion::ForVersionAsOf(expr)));
             } else if self.peek_keyword(Keyword::CHANGES) {
                 return self.parse_table_version_changes().map(Some);
             } else if self.peek_keyword(Keyword::AT) || self.peek_keyword(Keyword::BEFORE) {