Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 7 additions & 10 deletions src/cte_nodes.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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%");
Expand Down Expand Up @@ -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;
}
Expand Down
2 changes: 2 additions & 0 deletions src/include/cte_nodes.hpp
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
#pragma once

#include "duckdb.hpp"
#include "duckdb/parser/tableref/at_clause.hpp"
#include "sql_dialect.hpp"

namespace duckdb {
Expand Down Expand Up @@ -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<string> table_function_alias;
unique_ptr<AtClause> 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<bool> column_is_expression;
Expand Down
2 changes: 2 additions & 0 deletions src/include/lpts_ast.hpp
Original file line number Diff line number Diff line change
@@ -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 {
Expand Down Expand Up @@ -69,6 +70,7 @@ class AstGetNode : public AstNode {
string catalog;
string schema;
string table_name;
unique_ptr<AtClause> snapshot;
size_t table_index;
vector<string> column_names; ///< Physical column names (e.g. "age", "name").
vector<string> cte_column_names; ///< CTE-scoped names (e.g. "t0_age", "t0_name").
Expand Down
13 changes: 3 additions & 10 deletions src/include/lpts_helpers.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@

namespace duckdb {

class AtClause;
string RenderSnapshotSuffix(optional_ptr<AtClause> snapshot, SqlDialect dialect);

/// Convert a vector of strings into a separated list (e.g. "a, b, c").
string VecToSeparatedList(const vector<string> &input_list, const string &separator = ", ");

Expand All @@ -22,23 +25,13 @@ string QuoteIdentifier(const string &identifier);
/// Quote a list of identifiers and join them with a separator.
string VecToQuotedIdentifierList(const vector<string> &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 (<PARAM> => <value>)`, 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);

/// Dialect-aware variants. Backtick dialects use backtick quoting; other dialects
/// fall back to `KeywordHelper::WriteOptionallyQuoted` (matching the dialect-blind overloads above).
string DialectVecToQuotedIdentifierList(const vector<string> &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);

Expand Down
9 changes: 8 additions & 1 deletion src/include/lpts_pipeline.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,20 @@

#include "lpts_ast.hpp"
#include "cte_nodes.hpp"
#include <functional>

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<unique_ptr<AtClause>(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<AstNode> LogicalPlanToAst(ClientContext &context, unique_ptr<LogicalOperator> &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).
Expand Down
3 changes: 2 additions & 1 deletion src/lpts_ast.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ InsertionOrderPreservingMap<string> 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++) {
Expand Down Expand Up @@ -49,7 +50,7 @@ string AstGetNode::ToString(int indent) const {

InsertionOrderPreservingMap<string> AstGetNode::GetExtraInfo() const {
InsertionOrderPreservingMap<string> 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) {
Expand Down
20 changes: 13 additions & 7 deletions src/lpts_ast_builder.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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.
///
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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<AtClause>("VERSION", make_uniq<ConstantExpression>(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) {
Expand Down Expand Up @@ -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();
}) {
Expand Down Expand Up @@ -2831,8 +2836,9 @@ class AstBuilder {
//==============================================================================
// Phase 1 entry point
//==============================================================================
unique_ptr<AstNode> LogicalPlanToAst(ClientContext &context, unique_ptr<LogicalOperator> &plan, SqlDialect dialect) {
AstBuilder builder(context, dialect);
unique_ptr<AstNode> LogicalPlanToAst(ClientContext &context, unique_ptr<LogicalOperator> &plan, SqlDialect dialect,
const SnapshotResolver &snapshot_resolver) {
AstBuilder builder(context, dialect, snapshot_resolver);
return builder.Build(plan);
}

Expand Down
14 changes: 6 additions & 8 deletions src/lpts_ast_flattener.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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);
Expand Down
85 changes: 15 additions & 70 deletions src/lpts_helpers.cpp
Original file line number Diff line number Diff line change
@@ -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"
Expand Down Expand Up @@ -57,17 +58,8 @@ string VecToQuotedIdentifierList(const vector<string> &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<string> &input_list, SqlDialect dialect, const string &separator) {
Expand All @@ -81,67 +73,20 @@ string DialectVecToQuotedIdentifierList(const vector<string> &input_list, SqlDia
return ret_str.str();
}

/// Split `name AT (<PARAM> => <value>)` — 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<AtClause> 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,
Expand All @@ -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,
Expand Down
26 changes: 26 additions & 0 deletions test/sql/dialect_spark.test
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand All @@ -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
# ============================================================
Expand Down
Loading
Loading