feat: Carry FOR VERSION AS OF from the parser into the plan

axiom/common/SchemaTableName.cpp

@@ -32,5 +32,9 @@ size_t std::hash<facebook::axiom::SchemaTableName>::operator()(
   auto hash = folly::hasher<std::string>{}(name.table);
   hash = facebook::velox::bits::hashMix(
       hash, folly::hasher<std::string>{}(name.schema));
+  if (name.snapshotId.has_value()) {
+    hash = facebook::velox::bits::hashMix(
+        hash, folly::hasher<int64_t>{}(*name.snapshotId));
+  }
   return hash;
 }

axiom/common/SchemaTableName.h

@@ -15,6 +15,8 @@
  */
 #pragma once
 
+#include <cstdint>
+#include <optional>
 #include <string>
 
 #include <fmt/format.h>
@@ -30,7 +32,16 @@ struct SchemaTableName {
   std::string schema;
   std::string table;
 
-  /// Returns "schema"."table" for display/logging only.
+  /// Snapshot the reference is pinned to, when the query asked for a specific
+  /// table version (Iceberg time travel, `FOR VERSION AS OF <id>`). Empty for
+  /// an ordinary reference, which reads the current version. Part of the
+  /// identity: two references differing only by snapshot are different tables,
+  /// so a version-aware connector serves each its own state and the optimizer
+  /// caches them apart.
+  std::optional<int64_t> snapshotId;
+
+  /// Returns "schema"."table" for display/logging only, with an `@<id>` suffix
+  /// when pinned to a snapshot.
   std::string toString() const;
 
   bool operator==(const SchemaTableName&) const = default;
@@ -52,6 +63,14 @@ struct fmt::formatter<facebook::axiom::SchemaTableName>
     : fmt::formatter<std::string_view> {
   auto format(const facebook::axiom::SchemaTableName& name, format_context& ctx)
       const {
+    if (name.snapshotId.has_value()) {
+      return fmt::format_to(
+          ctx.out(),
+          R"d("{}"."{}"@{})d",
+          name.schema,
+          name.table,
+          *name.snapshotId);
+    }
     return fmt::format_to(ctx.out(), R"d("{}"."{}")d", name.schema, name.table);
   }
 };

axiom/connectors/ConnectorMetadata.h

@@ -1187,6 +1187,15 @@ class ConnectorMetadata {
   /// @return nullptr if table doesn't exist.
   virtual TablePtr findTable(const SchemaTableName& tableName) = 0;
 
+  /// Returns whether a table reference may pin a snapshot (SchemaTableName's
+  /// snapshotId, `FOR VERSION AS OF <id>`). Defaults to false, so the SQL layer
+  /// rejects a versioned reference to a connector that would otherwise ignore
+  /// the snapshot and read the current version. A connector that honors the
+  /// snapshot in findTable overrides this to true.
+  virtual bool supportsTableTimeTravel() const {
+    return false;
+  }
+
   /// Return a ViewPtr given the view name.
   ///
   /// @return nullptr if view doesn't exist.

axiom/logical_plan/LogicalPlanNode.cpp

@@ -282,6 +282,9 @@ folly::dynamic TableScanNode::serialize() const {
   obj["connectorId"] = connectorId_;
   obj["tableName"] = tableName_.table;
   obj["schema"] = tableName_.schema;
+  if (tableName_.snapshotId.has_value()) {
+    obj["snapshotId"] = *tableName_.snapshotId;
+  }
   obj["columnNames"] =
       serializeVector(columnNames_, [](const std::string& s) { return s; });
   return obj;
@@ -293,6 +296,9 @@ LogicalPlanNodePtr TableScanNode::create(
     void* /*context*/) {
   SchemaTableName tableName{
       obj.getDefault("schema", "").asString(), obj["tableName"].asString()};
+  if (const auto* snapshotId = obj.get_ptr("snapshotId")) {
+    tableName.snapshotId = snapshotId->asInt();
+  }
   return std::make_shared<TableScanNode>(
       obj["id"].asString(),
       deserializeOutputType(obj),

axiom/logical_plan/PlanBuilder.cpp

@@ -241,10 +241,12 @@ PlanBuilder& PlanBuilder::tableScan(
     const std::string& connectorId,
     const std::string& schemaName,
     const std::string& tableName,
-    bool includeHiddenColumns) {
+    bool includeHiddenColumns,
+    std::optional<int64_t> snapshotId) {
   VELOX_USER_CHECK_NULL(node_, "Table scan node must be the leaf node");
 
   SchemaTableName schemaTableName{schemaName, tableName};
+  schemaTableName.snapshotId = snapshotId;
   auto metadata = ConnectorMetadataRegistry::get(connectorId);
   auto table = metadata->findTable(schemaTableName);
   VELOX_USER_CHECK_NOT_NULL(

axiom/logical_plan/PlanBuilder.h

@@ -291,11 +291,16 @@ class PlanBuilder {
   }
 
   /// Equivalent to SELECT * FROM <tableName>.
+  ///
+  /// @param snapshotId Pins the scan to a table snapshot (Iceberg time travel).
+  /// Empty reads the current version. Only a connector whose metadata reports
+  /// supportsTableTimeTravel() honors it.
   PlanBuilder& tableScan(
       const std::string& connectorId,
       const std::string& schemaName,
       const std::string& tableName,
-      bool includeHiddenColumns = false);
+      bool includeHiddenColumns = false,
+      std::optional<int64_t> snapshotId = std::nullopt);
 
   /// @overload Uses the default connector ID from Context.
   PlanBuilder& tableScan(

axiom/sql/presto/PrestoParser.cpp

@@ -722,10 +722,26 @@ class RelationPlanner : public AstVisitor {
   }
 
   void processTable(const Table& table) {
-    processTable(table.location(), *table.name());
+    std::optional<int64_t> snapshotId;
+    if (const auto& version = table.version()) {
+      // The AST builder only produces a VERSION (not TIMESTAMP) version here.
+      const auto* snapshot =
+          dynamic_cast<const LongLiteral*>(version->expression().get());
+      if (snapshot == nullptr) {
+        AXIOM_PRESTO_SEMANTIC_FAIL(
+            version->location(),
+            /*token=*/"",
+            "FOR VERSION AS OF requires an integer snapshot id");
+      }
+      snapshotId = snapshot->value();
+    }
+    processTable(table.location(), *table.name(), snapshotId);
   }
 
-  void processTable(const NodeLocation& location, const QualifiedName& name) {
+  void processTable(
+      const NodeLocation& location,
+      const QualifiedName& name,
+      std::optional<int64_t> snapshotId = std::nullopt) {
     const auto tableName = canonicalizeName(name.suffix());
 
     // Only an unqualified single-part name can name a CTE; a qualified
@@ -736,7 +752,7 @@ class RelationPlanner : public AstVisitor {
     }
 
     // Regular base-table reference.
-    const auto [connectorId, connectorTable] = toConnectorTable(
+    auto [connectorId, connectorTable] = toConnectorTable(
         name,
         context_.defaultConnectorId.value(),
         defaultSchema_,
@@ -745,6 +761,19 @@ class RelationPlanner : public AstVisitor {
     auto metadata =
         facebook::axiom::connector::ConnectorMetadataRegistry::get(connectorId);
 
+    if (snapshotId.has_value()) {
+      // Reject rather than let a connector that ignores the snapshot read the
+      // current version as if no version were given.
+      if (!metadata->supportsTableTimeTravel()) {
+        AXIOM_PRESTO_SEMANTIC_FAIL(
+            location,
+            name.suffix(),
+            "Time travel (FOR VERSION AS OF) is not supported by connector: {}",
+            connectorId);
+      }
+      connectorTable.snapshotId = snapshotId;
+    }
+
     if (metadata->findTable(connectorTable) != nullptr) {
       // Drop display names captured from a sibling FROM relation so
       // this table's columns aren't tagged with them.
@@ -755,7 +784,8 @@ class RelationPlanner : public AstVisitor {
           connectorId,
           connectorTable.schema,
           connectorTable.table,
-          /*includeHiddenColumns=*/true);
+          /*includeHiddenColumns=*/true,
+          connectorTable.snapshotId);
     } else if (auto view = metadata->findView(connectorTable)) {
       views_.emplace(
           facebook::axiom::CatalogSchemaTableName{connectorId, connectorTable},

axiom/sql/presto/ast/AstBuilder.cpp

@@ -544,8 +544,40 @@ std::any AstBuilder::visitAliasedRelation(
 std::any AstBuilder::visitTableName(PrestoSqlParser::TableNameContext* ctx) {
   trace("visitTableName");
 
+  std::shared_ptr<TableVersionExpression> version;
+  if (auto* versionCtx = ctx->tableVersionExpression()) {
+    auto* tableVersion =
+        dynamic_cast<PrestoSqlParser::TableVersionContext*>(versionCtx);
+    VELOX_CHECK_NOT_NULL(tableVersion);
+
+    // Only `FOR VERSION AS OF <id>` is wired below the parser so far. Reject
+    // the other forms the grammar accepts rather than build a Table that
+    // silently reads the current snapshot.
+    if (tableVersion->TIMESTAMP() != nullptr ||
+        tableVersion->SYSTEM_TIME() != nullptr) {
+      AXIOM_PRESTO_SYNTAX_FAIL(
+          getLocation(tableVersion),
+          tableVersion->getText(),
+          "Timestamp time travel is not supported yet; use FOR VERSION AS OF");
+    }
+    if (dynamic_cast<PrestoSqlParser::TableversionbeforeContext*>(
+            tableVersion->tableVersionState()) != nullptr) {
+      AXIOM_PRESTO_SYNTAX_FAIL(
+          getLocation(tableVersion),
+          tableVersion->getText(),
+          "FOR VERSION BEFORE is not supported yet; use FOR VERSION AS OF");
+    }
+
+    version = std::make_shared<TableVersionExpression>(
+        getLocation(tableVersion),
+        TableVersionExpression::TableVersionType::kVersion,
+        visitTyped<Expression>(tableVersion->valueExpression()));
+  }
+
   return std::static_pointer_cast<Relation>(std::make_shared<Table>(
-      getLocation(ctx), getQualifiedName(ctx->qualifiedName())));
+      getLocation(ctx),
+      getQualifiedName(ctx->qualifiedName()),
+      std::move(version)));
 }
 
 std::any AstBuilder::visitSelectAll(PrestoSqlParser::SelectAllContext* ctx) {

axiom/sql/presto/tests/PrestoParserTest.cpp

@@ -1166,6 +1166,29 @@ TEST_F(PrestoParserTest, tablesample) {
   }
 }
 
+TEST_F(PrestoParserTest, rejectsUnsupportedTableVersionForms) {
+  // Timestamp and BEFORE are not wired below the parser, so they are rejected
+  // rather than dropped, which would silently read the current snapshot.
+  AXIOM_EXPECT_PRESTO_SYNTAX_ERROR(
+      parseSql(
+          "SELECT * FROM nation FOR TIMESTAMP AS OF TIMESTAMP '2020-01-01 00:00:00'"),
+      "Timestamp time travel is not supported yet");
+  AXIOM_EXPECT_PRESTO_SYNTAX_ERROR(
+      parseSql("SELECT * FROM nation FOR VERSION BEFORE 8"),
+      "FOR VERSION BEFORE is not supported yet");
+
+  // VERSION AS OF parses, but the test connector does not honor a snapshot, so
+  // the reference is rejected rather than silently reading the current version.
+  AXIOM_EXPECT_PRESTO_SEMANTIC_ERROR(
+      parseSql("SELECT * FROM nation FOR VERSION AS OF 8"),
+      "Time travel (FOR VERSION AS OF) is not supported by connector");
+
+  // A non-integer snapshot id is rejected before the connector is consulted.
+  AXIOM_EXPECT_PRESTO_SEMANTIC_ERROR(
+      parseSql("SELECT * FROM nation FOR VERSION AS OF 'abc'"),
+      "FOR VERSION AS OF requires an integer snapshot id");
+}
+
 TEST_F(PrestoParserTest, everything) {
   auto matcher = matchScan()
                      .join(matchScan().build())