From 87cb1d382a14e58fd63cf370137269470b14a543 Mon Sep 17 00:00:00 2001 From: ila Date: Mon, 21 Sep 2026 16:31:25 +0200 Subject: [PATCH] Keep snapshot clauses separate from scan identifiers --- src/cte_nodes.cpp | 17 +++---- src/include/cte_nodes.hpp | 2 + src/include/lpts_ast.hpp | 2 + src/include/lpts_helpers.hpp | 13 ++---- src/include/lpts_pipeline.hpp | 9 +++- src/lpts_ast.cpp | 3 +- src/lpts_ast_builder.cpp | 20 ++++++--- src/lpts_ast_flattener.cpp | 14 +++--- src/lpts_helpers.cpp | 85 +++++++---------------------------- test/sql/dialect_spark.test | 26 +++++++++++ test/sql/time_travel.test | 52 +++++++++++++++++++++ 11 files changed, 136 insertions(+), 107 deletions(-) diff --git a/src/cte_nodes.cpp b/src/cte_nodes.cpp index 1a3f097..85c29d2 100644 --- a/src/cte_nodes.cpp +++ b/src/cte_nodes.cpp @@ -384,17 +384,12 @@ string GetNode::ToQuery(SqlDialect dialect) { get_str << VecToSeparatedList(RenderGetSelectColumns(column_names, column_is_expression, dialect)); } get_str << " FROM "; - string base_table_name; - string snapshot_suffix; - const bool unqualified_snapshot = - catalog.empty() && TrySplitDialectSnapshotSuffix(table_name, dialect, base_table_name, snapshot_suffix); if (!catalog.empty()) { // Fully-qualified: catalog.schema.table (DuckDB / Spark dialect) - get_str << DialectQualifiedTableName(catalog, schema, table_name, dialect); - } else if (unqualified_snapshot) { - // A pinned-snapshot scan rendered unqualified: the qualifier is dialect-specific and must not be - // mistaken for a table-function argument list by the `_tf` aliasing below. - get_str << base_table_name << snapshot_suffix; + get_str << DialectQualifiedTableName(catalog, schema, table_name, dialect) + << RenderSnapshotSuffix(snapshot.get(), dialect); + } else if (snapshot) { + get_str << DialectQuoteIdent(table_name, dialect) << RenderSnapshotSuffix(snapshot.get(), dialect); } else { // A TABLE-argument function: the child CTE is the function's argument, not a lateral input. const size_t table_arg_pos = table_name.find("%LPTS_TABLE_ARG%"); @@ -447,7 +442,9 @@ bool GetNode::BuildSelectParts(SqlDialect dialect, SelectParts &out) const { return false; } out.select_exprs = RenderGetSelectColumns(column_names, column_is_expression, dialect); - out.from = catalog.empty() ? table_name : DialectQualifiedTableName(catalog, schema, table_name, dialect); + out.from = catalog.empty() ? (snapshot ? DialectQuoteIdent(table_name, dialect) : table_name) + : DialectQualifiedTableName(catalog, schema, table_name, dialect); + out.from += RenderSnapshotSuffix(snapshot.get(), dialect); out.where_conds = table_filters; // already complete conditions; the renderer wraps each in parens return true; } diff --git a/src/include/cte_nodes.hpp b/src/include/cte_nodes.hpp index 4a28f82..7ffcad6 100644 --- a/src/include/cte_nodes.hpp +++ b/src/include/cte_nodes.hpp @@ -1,6 +1,7 @@ #pragma once #include "duckdb.hpp" +#include "duckdb/parser/tableref/at_clause.hpp" #include "sql_dialect.hpp" namespace duckdb { @@ -189,6 +190,7 @@ class GetNode : public CteNode { public: /// `_tf(...)` alias list in the table function's output order (empty = use projected column order). vector table_function_alias; + unique_ptr snapshot; /// Parallel to `column_names`: true where the entry is a raw SQL expression (struct field-extraction /// pushdown) emitted verbatim rather than quoted. Empty means "all plain identifiers". vector column_is_expression; diff --git a/src/include/lpts_ast.hpp b/src/include/lpts_ast.hpp index b5946a2..1d46deb 100644 --- a/src/include/lpts_ast.hpp +++ b/src/include/lpts_ast.hpp @@ -1,6 +1,7 @@ #pragma once #include "duckdb.hpp" +#include "duckdb/parser/tableref/at_clause.hpp" #include "duckdb/common/insertion_order_preserving_map.hpp" namespace duckdb { @@ -69,6 +70,7 @@ class AstGetNode : public AstNode { string catalog; string schema; string table_name; + unique_ptr snapshot; size_t table_index; vector column_names; ///< Physical column names (e.g. "age", "name"). vector cte_column_names; ///< CTE-scoped names (e.g. "t0_age", "t0_name"). diff --git a/src/include/lpts_helpers.hpp b/src/include/lpts_helpers.hpp index 1b18534..83838a3 100644 --- a/src/include/lpts_helpers.hpp +++ b/src/include/lpts_helpers.hpp @@ -10,6 +10,9 @@ namespace duckdb { +class AtClause; +string RenderSnapshotSuffix(optional_ptr snapshot, SqlDialect dialect); + /// Convert a vector of strings into a separated list (e.g. "a, b, c"). string VecToSeparatedList(const vector &input_list, const string &separator = ", "); @@ -22,15 +25,6 @@ string QuoteIdentifier(const string &identifier); /// Quote a list of identifiers and join them with a separator. string VecToQuotedIdentifierList(const vector &input_list, const string &separator = ", "); -/// Quote a table name, preserving a DuckDB AT (...) snapshot suffix if present. -string QuoteTableWithOptionalSuffix(const string &table_name); - -/// Split a table name that carries a pinned snapshot (`name AT ( => )`, the DuckDB -/// spelling LPTS uses internally) into `base_name` and the `dialect`-rendered snapshot qualifier -/// (e.g. ` VERSION AS OF 366` for Spark). Returns false and leaves the outputs untouched when -/// `table_name` carries no snapshot. Throws when `dialect` has no verified time-travel syntax. -bool TrySplitDialectSnapshotSuffix(const string &table_name, SqlDialect dialect, string &base_name, string &suffix); - /// Build catalog.schema.table with each identifier quoted when needed. string QualifiedTableName(const string &catalog, const string &schema, const string &table_name); @@ -38,7 +32,6 @@ string QualifiedTableName(const string &catalog, const string &schema, const str /// fall back to `KeywordHelper::WriteOptionallyQuoted` (matching the dialect-blind overloads above). string DialectVecToQuotedIdentifierList(const vector &input_list, SqlDialect dialect, const string &separator = ", "); -string DialectQuoteTableWithOptionalSuffix(const string &table_name, SqlDialect dialect); string DialectQualifiedTableName(const string &catalog, const string &schema, const string &table_name, SqlDialect dialect); diff --git a/src/include/lpts_pipeline.hpp b/src/include/lpts_pipeline.hpp index 9e8380f..ebcebb9 100644 --- a/src/include/lpts_pipeline.hpp +++ b/src/include/lpts_pipeline.hpp @@ -2,13 +2,20 @@ #include "lpts_ast.hpp" #include "cte_nodes.hpp" +#include namespace duckdb { +class TableCatalogEntry; +class AtClause; +// Called by the existing GET visit with the original catalog entry, unaffected by output qualification overrides. +using SnapshotResolver = std::function(const TableCatalogEntry &)>; + /// Phase 1: Convert a DuckDB LogicalOperator tree into a dialect-agnostic AST. /// `dialect` is forwarded to expression serialization for dialect-specific function renaming. unique_ptr LogicalPlanToAst(ClientContext &context, unique_ptr &plan, - SqlDialect dialect = SqlDialect::DUCKDB); + SqlDialect dialect = SqlDialect::DUCKDB, + const SnapshotResolver &snapshot_resolver = {}); /// Phase 2: Convert an AST into a flat CTE list. /// `dialect` controls dialect-specific SQL rendering (default: DuckDB). diff --git a/src/lpts_ast.cpp b/src/lpts_ast.cpp index 70a2102..8813a91 100644 --- a/src/lpts_ast.cpp +++ b/src/lpts_ast.cpp @@ -22,6 +22,7 @@ InsertionOrderPreservingMap AstNode::GetExtraInfo() const { string AstGetNode::ToString(int indent) const { string result = Indent(indent) + "Get"; result += " " + catalog + "." + schema + "." + table_name; + result += snapshot ? " " + snapshot->ToString() : ""; result += " (table_index=" + std::to_string(table_index) + ")"; result += "\n" + Indent(indent + 2) + "columns: ["; for (size_t i = 0; i < column_names.size(); i++) { @@ -49,7 +50,7 @@ string AstGetNode::ToString(int indent) const { InsertionOrderPreservingMap AstGetNode::GetExtraInfo() const { InsertionOrderPreservingMap info; - info.insert("Table", catalog + "." + schema + "." + table_name); + info.insert("Table", catalog + "." + schema + "." + table_name + (snapshot ? " " + snapshot->ToString() : "")); string cols = "["; for (size_t i = 0; i < column_names.size(); i++) { if (i > 0) { diff --git a/src/lpts_ast_builder.cpp b/src/lpts_ast_builder.cpp index 3474a17..016b806 100644 --- a/src/lpts_ast_builder.cpp +++ b/src/lpts_ast_builder.cpp @@ -7,6 +7,7 @@ #include "lpts_expression_renderer.hpp" #include "duckdb/main/connection.hpp" #include "duckdb/parser/keyword_helper.hpp" +#include "duckdb/parser/expression/constant_expression.hpp" #include "duckdb/planner/expression/bound_cast_expression.hpp" #include "duckdb/planner/expression/bound_columnref_expression.hpp" @@ -309,6 +310,7 @@ class AstBuilder { /// Client context for runtime queries (e.g. DuckLake current snapshot). ClientContext &context; + const SnapshotResolver &snapshot_resolver; /// Re-target emitted table references at the destination system's catalog/schema. /// @@ -1125,9 +1127,6 @@ class AstBuilder { schema_name = catalog_entry->schema.name; ApplyOutputQualificationOverrides(catalog_name, schema_name); table_name = catalog_entry.get()->name; - if (is_ducklake_time_travel) { - table_name += " AT (VERSION => " + std::to_string(ducklake_snapshot_id) + ")"; - } } else { // Table function without catalog entry (e.g. range(), read_csv()) std::ostringstream func_str; @@ -1468,6 +1467,12 @@ class AstBuilder { std::move(column_names), std::move(cte_column_names), std::move(table_filters), table_function_output_count); get_node->table_function_alias = std::move(table_function_alias); + if (is_ducklake_time_travel) { + get_node->snapshot = + make_uniq("VERSION", make_uniq(Value::UBIGINT(ducklake_snapshot_id))); + } else if (catalog_entry && snapshot_resolver) { + get_node->snapshot = snapshot_resolver(*catalog_entry); + } // Only carry the flags when at least one column is a struct-extract expression; an all-false // vector is equivalent to "empty" and the renderers treat empty as all plain identifiers. for (auto is_expr : column_is_expr) { @@ -2797,8 +2802,8 @@ class AstBuilder { } public: - AstBuilder(ClientContext &_context, SqlDialect _dialect = SqlDialect::DUCKDB) - : dialect(_dialect), context(_context), + AstBuilder(ClientContext &_context, SqlDialect _dialect, const SnapshotResolver &_snapshot_resolver) + : dialect(_dialect), context(_context), snapshot_resolver(_snapshot_resolver), expression_renderer(_dialect, [this](const ColumnBinding &binding, const char *context) { return FindColumnBinding(binding, context)->ToUniqueColumnName(); }) { @@ -2831,8 +2836,9 @@ class AstBuilder { //============================================================================== // Phase 1 entry point //============================================================================== -unique_ptr LogicalPlanToAst(ClientContext &context, unique_ptr &plan, SqlDialect dialect) { - AstBuilder builder(context, dialect); +unique_ptr LogicalPlanToAst(ClientContext &context, unique_ptr &plan, SqlDialect dialect, + const SnapshotResolver &snapshot_resolver) { + AstBuilder builder(context, dialect, snapshot_resolver); return builder.Build(plan); } diff --git a/src/lpts_ast_flattener.cpp b/src/lpts_ast_flattener.cpp index cf86597..009d205 100644 --- a/src/lpts_ast_flattener.cpp +++ b/src/lpts_ast_flattener.cpp @@ -289,15 +289,11 @@ class AstFlattener { } } sql += " FROM "; - string inline_base_name; - string inline_snapshot_suffix; if (!get.catalog.empty()) { - sql += DialectQualifiedTableName(get.catalog, get.schema, get.table_name, dialect); - } else if (TrySplitDialectSnapshotSuffix(get.table_name, dialect, inline_base_name, - inline_snapshot_suffix)) { - // Pinned-snapshot scan rendered unqualified: the dialect-specific qualifier must not be - // mistaken for a table-function argument list by the `_tf` aliasing below. - sql += inline_base_name + inline_snapshot_suffix; + sql += DialectQualifiedTableName(get.catalog, get.schema, get.table_name, dialect) + + RenderSnapshotSuffix(get.snapshot.get(), dialect); + } else if (get.snapshot) { + sql += DialectQuoteIdent(get.table_name, dialect) + RenderSnapshotSuffix(get.snapshot.get(), dialect); } else { // An in-out (lateral) table function has the delim/correlation source as its AST child: // inline it as the left comma-join input (mirrors GetNode's input_cte_name handling). @@ -743,6 +739,7 @@ class AstFlattener { if (absorb_get) { from_clause = DialectQualifiedTableName(absorb_get->catalog, absorb_get->schema, absorb_get->table_name, dialect); + from_clause += RenderSnapshotSuffix(absorb_get->snapshot.get(), dialect); for (size_t i = 0; i < absorb_get->cte_column_names.size(); i++) { // Struct field-extraction columns are raw SQL expressions, emitted verbatim; plain columns // are quoted identifiers. @@ -1100,6 +1097,7 @@ class AstFlattener { get.table_index, get.table_filters, get.column_names, input_cte_name, get.table_function_output_count); get_node->table_function_alias = get.table_function_alias; + get_node->snapshot = get.snapshot ? get.snapshot->Copy() : nullptr; get_node->column_is_expression = get.column_is_expression; get_node->spark_broadcast_hint = emit_spark_hints && dialect == SqlDialect::SPARK && IsOpenIvmDeltaTable(get.table_name); diff --git a/src/lpts_helpers.cpp b/src/lpts_helpers.cpp index d8b335a..4298657 100644 --- a/src/lpts_helpers.cpp +++ b/src/lpts_helpers.cpp @@ -1,4 +1,5 @@ #include "lpts_helpers.hpp" +#include "duckdb/parser/tableref/at_clause.hpp" #include "lpts_sql_scanner.hpp" #include "duckdb/parser/keyword_helper.hpp" @@ -57,17 +58,8 @@ string VecToQuotedIdentifierList(const vector &input_list, const string return ret_str.str(); } -string QuoteTableWithOptionalSuffix(const string &table_name) { - static const string at_suffix = " AT ("; - auto suffix_pos = table_name.find(at_suffix); - if (suffix_pos == string::npos) { - return QuoteIdentifier(table_name); - } - return QuoteIdentifier(table_name.substr(0, suffix_pos)) + table_name.substr(suffix_pos); -} - string QualifiedTableName(const string &catalog, const string &schema, const string &table_name) { - return QuoteIdentifier(catalog) + "." + QuoteIdentifier(schema) + "." + QuoteTableWithOptionalSuffix(table_name); + return QuoteIdentifier(catalog) + "." + QuoteIdentifier(schema) + "." + QuoteIdentifier(table_name); } string DialectVecToQuotedIdentifierList(const vector &input_list, SqlDialect dialect, const string &separator) { @@ -81,67 +73,20 @@ string DialectVecToQuotedIdentifierList(const vector &input_list, SqlDia return ret_str.str(); } -/// Split `name AT ( => )` — the DuckDB spelling LPTS uses internally to carry a pinned -/// snapshot on a table name — into its base name and `AT` parameter/value. -static bool TrySplitSnapshotSuffix(const string &table_name, string &base_name, string &at_parameter, - string &at_value) { - static const string AT_SUFFIX = " AT ("; - auto suffix_pos = table_name.find(AT_SUFFIX); - if (suffix_pos == string::npos) { - return false; - } - string body = table_name.substr(suffix_pos + AT_SUFFIX.size()); - if (body.empty() || body.back() != ')') { - return false; - } - body.pop_back(); - auto arrow_pos = body.find("=>"); - if (arrow_pos == string::npos) { - return false; +string RenderSnapshotSuffix(optional_ptr snapshot, SqlDialect dialect) { + if (!snapshot) { + return ""; } - base_name = table_name.substr(0, suffix_pos); - at_parameter = TrimCopy(body.substr(0, arrow_pos)); - at_value = TrimCopy(body.substr(arrow_pos + 2)); - return !at_parameter.empty() && !at_value.empty(); -} - -/// Render a pinned-snapshot qualifier in `dialect`. LPTS carries the pin in DuckDB's spelling -/// (`AT (VERSION => 366)`); Spark/Delta spells the same pin `VERSION AS OF 366`. A dialect with no -/// verified time-travel syntax refuses instead of emitting a qualifier the target cannot parse — -/// dropping it would silently turn a pinned scan into a read of the latest snapshot. -static string DialectSnapshotSuffix(const string &base_name, const string &at_parameter, const string &at_value, - SqlDialect dialect) { if (dialect == SqlDialect::DUCKDB) { - return " AT (" + at_parameter + " => " + at_value + ")"; + return " " + snapshot->ToString(); } - string parameter = LowerCopy(at_parameter); + auto parameter = LowerCopy(snapshot->Unit()); + auto value = snapshot->ExpressionMutable()->ToString(); if (dialect == SqlDialect::SPARK && (parameter == "version" || parameter == "timestamp")) { - return (parameter == "version" ? string(" VERSION AS OF ") : string(" TIMESTAMP AS OF ")) + at_value; - } - ThrowLptsNotImplemented("LPTS_UNSUPPORTED_TIME_TRAVEL", dialect, "time_travel", at_parameter + " => " + at_value, - base_name, "no verified time-travel syntax for target dialect"); -} - -string DialectQuoteTableWithOptionalSuffix(const string &table_name, SqlDialect dialect) { - string base_name; - string at_parameter; - string at_value; - if (!TrySplitSnapshotSuffix(table_name, base_name, at_parameter, at_value)) { - return DialectQuoteIdent(table_name, dialect); - } - return DialectQuoteIdent(base_name, dialect) + DialectSnapshotSuffix(base_name, at_parameter, at_value, dialect); -} - -bool TrySplitDialectSnapshotSuffix(const string &table_name, SqlDialect dialect, string &base_name, string &suffix) { - string at_parameter; - string at_value; - string split_base; - if (!TrySplitSnapshotSuffix(table_name, split_base, at_parameter, at_value)) { - return false; + return (parameter == "version" ? string(" VERSION AS OF ") : string(" TIMESTAMP AS OF ")) + value; } - suffix = DialectSnapshotSuffix(split_base, at_parameter, at_value, dialect); - base_name = split_base; - return true; + ThrowLptsNotImplemented("LPTS_UNSUPPORTED_TIME_TRAVEL", dialect, "time_travel", parameter + " => " + value, + "table scan", "no verified time-travel syntax for target dialect"); } string DialectQualifiedTableName(const string &catalog, const string &schema, const string &table_name, @@ -161,16 +106,16 @@ string DialectQualifiedTableName(const string &catalog, const string &schema, co table_path += "."; } table_path += table_name; - return DialectQuoteTableWithOptionalSuffix(table_path, dialect); + return DialectQuoteIdent(table_path, dialect); } if (DialectUsesSchemaQualifiedTableNames(dialect)) { if (schema.empty()) { - return DialectQuoteTableWithOptionalSuffix(table_name, dialect); + return DialectQuoteIdent(table_name, dialect); } - return DialectQuoteIdent(schema, dialect) + "." + DialectQuoteTableWithOptionalSuffix(table_name, dialect); + return DialectQuoteIdent(schema, dialect) + "." + DialectQuoteIdent(table_name, dialect); } return DialectQuoteIdent(catalog, dialect) + "." + DialectQuoteIdent(schema, dialect) + "." + - DialectQuoteTableWithOptionalSuffix(table_name, dialect); + DialectQuoteIdent(table_name, dialect); } [[noreturn]] void ThrowLptsNotImplemented(const string &code, SqlDialect dialect, const string &feature_kind, diff --git a/test/sql/dialect_spark.test b/test/sql/dialect_spark.test index f0012fd..b4dd067 100644 --- a/test/sql/dialect_spark.test +++ b/test/sql/dialect_spark.test @@ -15,6 +15,26 @@ require lpts # whole file; require it so those runners skip instead of erroring. require icu +# A table name is an identifier, even when its characters resemble a snapshot clause. +statement ok +CREATE TABLE "events AT (VERSION => 2)" (id INTEGER); + +statement ok +INSERT INTO "events AT (VERSION => 2)" VALUES (1), (1), (2); + +statement ok +SET lpts_check = true; + +query I rowsort +SELECT id FROM "events AT (VERSION => 2)"; +---- +1 +1 +2 + +statement ok +SET lpts_check = false; + statement ok CREATE TABLE users (id INTEGER, name VARCHAR, "order" INTEGER); @@ -37,6 +57,12 @@ CREATE TABLE compiler_edges (d DECIMAL(4,4), name VARCHAR); statement ok SET lpts_dialect = 'spark'; +query I +SELECT sql LIKE '%`memory`.`main`.`events AT (VERSION => 2)`%' AND sql NOT LIKE '%VERSION AS OF%' +FROM lpts_query('SELECT id FROM "events AT (VERSION => 2)"'); +---- +true + # ============================================================ # Identifier quoting — backticks # ============================================================ diff --git a/test/sql/time_travel.test b/test/sql/time_travel.test index 207a52a..ffd1221 100644 --- a/test/sql/time_travel.test +++ b/test/sql/time_travel.test @@ -112,6 +112,16 @@ SELECT sql FROM lpts_normalize_query('SELECT * FROM t VERSION AS OF 2 p(a, b)'); ---- SELECT * FROM t p(a, b) AT (VERSION => 2) +query T +SELECT sql FROM lpts_normalize_query('SELECT * FROM `a``b` VERSION AS OF 2 AS `p``q`(id)'); +---- +SELECT * FROM "a`b" AS "p`q"(id) AT (VERSION => 2) + +query T +SELECT sql FROM lpts_normalize_query('SELECT * FROM main . t VERSION AS OF 2 AS p(id)'); +---- +SELECT * FROM main . t AS p(id) AT (VERSION => 2) + # Delta path relations behave the same way. query T SELECT sql FROM lpts_normalize_query('SELECT * FROM delta.`/mnt/mv` VERSION AS OF 42 v'); @@ -284,6 +294,18 @@ SELECT COUNT(*) FROM dl.products AS p AT (VERSION => 2), dl.products AS q AT (VE ---- 6 +# The non-fused GET renderer must carry the same typed snapshot as a fused scan. +statement ok +SET lpts_merge_pipeline = false; + +query T +SELECT name FROM dl.products AT (VERSION => 2) WHERE product_id > 1; +---- +Gadget + +statement ok +SET lpts_merge_pipeline = true; + # ------------------------------------------------------------ # Spark input against the same catalog: normalization produces # exactly the executable statements asserted above, and the @@ -343,6 +365,36 @@ FROM lpts_query('SELECT name FROM dl.products VERSION AS OF 2 AS p WHERE p.produ ---- true +statement ok +SET lpts_merge_pipeline = false; + +query I +SELECT sql LIKE '%`dl`.`main`.`products` VERSION AS OF 2%' AND sql NOT LIKE '%AT (VERSION%' +FROM lpts_query('SELECT name FROM dl.products VERSION AS OF 2 WHERE product_id > 1'); +---- +true + +statement ok +SET lpts_output_unqualified = true; + +query I +SELECT sql LIKE '%`products` VERSION AS OF 2%' AND sql NOT LIKE '%_tf(%' +FROM lpts_query('SELECT name FROM dl.products VERSION AS OF 2 WHERE product_id > 1'); +---- +true + +statement ok +SET lpts_merge_pipeline = true; + +query I +SELECT sql LIKE '%`products` VERSION AS OF 2%' AND sql NOT LIKE '%_tf(%' +FROM lpts_query('SELECT name FROM dl.products VERSION AS OF 2 WHERE product_id > 1'); +---- +true + +statement ok +SET lpts_output_unqualified = false; + query I SELECT sql LIKE '%VERSION AS OF 2%' AND sql LIKE '%VERSION AS OF 3%' AS both_pins_render_in_spark FROM lpts_query('SELECT COUNT(*) AS c FROM dl.products VERSION AS OF 2 p, dl.products VERSION AS OF 3 q WHERE p.product_id = q.product_id');