1515// specific language governing permissions and limitations
1616// under the License.
1717
18- //! Custom [`RelationPlanner`] for Paimon time travel via `FOR SYSTEM_TIME AS OF`.
18+ //! Custom [`RelationPlanner`] for Paimon time travel via `VERSION AS OF` and `TIMESTAMP AS OF`.
1919
2020use std:: collections:: HashMap ;
2121use std:: fmt:: Debug ;
@@ -29,16 +29,16 @@ use datafusion::logical_expr::planner::{
2929 PlannedRelation , RelationPlanner , RelationPlannerContext , RelationPlanning ,
3030} ;
3131use datafusion:: sql:: sqlparser:: ast:: { self , TableFactor , TableVersion } ;
32- use paimon:: spec:: { SCAN_SNAPSHOT_ID_OPTION , SCAN_TAG_NAME_OPTION , SCAN_TIMESTAMP_MILLIS_OPTION } ;
32+ use paimon:: spec:: { SCAN_TIMESTAMP_MILLIS_OPTION , SCAN_VERSION_OPTION } ;
3333
3434use crate :: table:: PaimonTableProvider ;
3535
36- /// A [`RelationPlanner`] that intercepts `FOR SYSTEM_TIME AS OF` clauses
37- /// on Paimon tables and resolves them to time travel options.
36+ /// A [`RelationPlanner`] that intercepts `VERSION AS OF` and `TIMESTAMP AS OF`
37+ /// clauses on Paimon tables and resolves them to time travel options.
3838///
39- /// - Integer literal → sets `scan.snapshot-id ` option on the table.
40- /// - String literal (timestamp) → parsed as a timestamp, sets `scan.timestamp-millis` option .
41- /// - String literal (other) → sets `scan.tag-name` option on the table .
39+ /// - `VERSION AS OF <integer or string>` → sets `scan.version ` option on the table.
40+ /// At scan time, the version is resolved: tag name (if exists) → snapshot id → error .
41+ /// - `TIMESTAMP AS OF <timestamp string>` → parsed as a timestamp, sets `scan.timestamp-millis` .
4242#[ derive( Debug ) ]
4343pub struct PaimonRelationPlanner ;
4444
@@ -67,12 +67,13 @@ impl RelationPlanner for PaimonRelationPlanner {
6767 ..
6868 } = relation
6969 else {
70- return Ok ( RelationPlanning :: Original ( relation) ) ;
70+ return Ok ( RelationPlanning :: Original ( Box :: new ( relation) ) ) ;
7171 } ;
7272
73- let version_expr = match version {
74- Some ( TableVersion :: ForSystemTimeAsOf ( expr) ) => expr. clone ( ) ,
75- _ => return Ok ( RelationPlanning :: Original ( relation) ) ,
73+ let extra_options = match version {
74+ Some ( TableVersion :: VersionAsOf ( expr) ) => resolve_version_as_of ( expr) ?,
75+ Some ( TableVersion :: TimestampAsOf ( expr) ) => resolve_timestamp_as_of ( expr) ?,
76+ _ => return Ok ( RelationPlanning :: Original ( Box :: new ( relation) ) ) ,
7677 } ;
7778
7879 // Resolve the table reference.
@@ -84,10 +85,9 @@ impl RelationPlanner for PaimonRelationPlanner {
8485
8586 // Check if this is a Paimon table.
8687 let Some ( paimon_provider) = provider. as_any ( ) . downcast_ref :: < PaimonTableProvider > ( ) else {
87- return Ok ( RelationPlanning :: Original ( relation) ) ;
88+ return Ok ( RelationPlanning :: Original ( Box :: new ( relation) ) ) ;
8889 } ;
8990
90- let extra_options = resolve_time_travel_options ( & version_expr) ?;
9191 let new_table = paimon_provider. table ( ) . copy_with_options ( extra_options) ;
9292 let new_provider = PaimonTableProvider :: try_new ( new_table) ?;
9393 let new_source = provider_as_source ( Arc :: new ( new_provider) ) ;
@@ -98,7 +98,9 @@ impl RelationPlanner for PaimonRelationPlanner {
9898 } ;
9999
100100 let plan = LogicalPlanBuilder :: scan ( table_ref, new_source, None ) ?. build ( ) ?;
101- Ok ( RelationPlanning :: Planned ( PlannedRelation :: new ( plan, alias) ) )
101+ Ok ( RelationPlanning :: Planned ( Box :: new ( PlannedRelation :: new (
102+ plan, alias,
103+ ) ) ) )
102104 }
103105}
104106
@@ -136,45 +138,47 @@ fn object_name_to_table_reference(
136138 }
137139}
138140
139- /// Resolve `FOR SYSTEM_TIME AS OF <expr>` into table options .
141+ /// Resolve `VERSION AS OF <expr>` into `scan.version` option .
140142///
141- /// - Integer literal → `{"scan.snapshot-id": "N"}`
142- /// - String literal (timestamp `YYYY-MM-DD HH:MM:SS`) → `{"scan.timestamp-millis": "M"}`
143- /// - String literal (other) → `{"scan.tag-name": "S"}`
144- fn resolve_time_travel_options ( expr : & ast:: Expr ) -> DFResult < HashMap < String , String > > {
143+ /// The raw value (integer or string) is passed through as-is.
144+ /// Resolution (tag vs snapshot id) happens at scan time in `TableScan`.
145+ fn resolve_version_as_of ( expr : & ast:: Expr ) -> DFResult < HashMap < String , String > > {
146+ let version = match expr {
147+ ast:: Expr :: Value ( v) => match & v. value {
148+ ast:: Value :: Number ( n, _) => n. clone ( ) ,
149+ ast:: Value :: SingleQuotedString ( s) | ast:: Value :: DoubleQuotedString ( s) => s. clone ( ) ,
150+ _ => {
151+ return Err ( datafusion:: error:: DataFusionError :: Plan ( format ! (
152+ "Unsupported VERSION AS OF expression: {expr}"
153+ ) ) )
154+ }
155+ } ,
156+ _ => {
157+ return Err ( datafusion:: error:: DataFusionError :: Plan ( format ! (
158+ "Unsupported VERSION AS OF expression: {expr}. Expected an integer snapshot id or a tag name."
159+ ) ) )
160+ }
161+ } ;
162+ Ok ( HashMap :: from ( [ ( SCAN_VERSION_OPTION . to_string ( ) , version) ] ) )
163+ }
164+
165+ /// Resolve `TIMESTAMP AS OF <expr>` into `scan.timestamp-millis` option.
166+ fn resolve_timestamp_as_of ( expr : & ast:: Expr ) -> DFResult < HashMap < String , String > > {
145167 match expr {
146168 ast:: Expr :: Value ( v) => match & v. value {
147- ast:: Value :: Number ( n, _) => {
148- // Validate it's a valid integer
149- n. parse :: < i64 > ( ) . map_err ( |e| {
150- datafusion:: error:: DataFusionError :: Plan ( format ! (
151- "Invalid snapshot id '{n}': {e}"
152- ) )
153- } ) ?;
169+ ast:: Value :: SingleQuotedString ( s) | ast:: Value :: DoubleQuotedString ( s) => {
170+ let millis = parse_timestamp_to_millis ( s) ?;
154171 Ok ( HashMap :: from ( [ (
155- SCAN_SNAPSHOT_ID_OPTION . to_string ( ) ,
156- n . clone ( ) ,
172+ SCAN_TIMESTAMP_MILLIS_OPTION . to_string ( ) ,
173+ millis . to_string ( ) ,
157174 ) ] ) )
158175 }
159- ast:: Value :: SingleQuotedString ( s) | ast:: Value :: DoubleQuotedString ( s) => {
160- // Try parsing as timestamp first; fall back to tag name.
161- match parse_timestamp_to_millis ( s) {
162- Ok ( timestamp_millis) => Ok ( HashMap :: from ( [ (
163- SCAN_TIMESTAMP_MILLIS_OPTION . to_string ( ) ,
164- timestamp_millis. to_string ( ) ,
165- ) ] ) ) ,
166- Err ( _) => Ok ( HashMap :: from ( [ (
167- SCAN_TAG_NAME_OPTION . to_string ( ) ,
168- s. clone ( ) ,
169- ) ] ) ) ,
170- }
171- }
172176 _ => Err ( datafusion:: error:: DataFusionError :: Plan ( format ! (
173- "Unsupported time travel expression: {expr}"
177+ "Unsupported TIMESTAMP AS OF expression: {expr}. Expected a timestamp string. "
174178 ) ) ) ,
175179 } ,
176180 _ => Err ( datafusion:: error:: DataFusionError :: Plan ( format ! (
177- "Unsupported time travel expression: {expr}. Expected an integer snapshot id, a timestamp string, or a tag name ."
181+ "Unsupported TIMESTAMP AS OF expression: {expr}. Expected a timestamp string."
178182 ) ) ) ,
179183 }
180184}
0 commit comments