[core] Support RollbackAction and Procedure with timestamp #4333
Conversation
xuzifu666
changed the title
| @@ -51,7 +56,16 @@ public void run() throws Exception { | |||
| } | |||
|
|
|||
| if (version.chars().allMatch(Character::isDigit)) { | |||
| table.rollbackTo(Long.parseLong(version)); | |||
| if (isTimestamp != null && isTimestamp) { | |||
There was a problem hiding this comment.
Here may not be boolean, because this argument may be null.
| fileStoreTable | ||
| .snapshotManager() | ||
| .earlierOrEqualTimeMills(Long.parseLong(version)) | ||
| .id()); |
There was a problem hiding this comment.
Method invocation 'id' may produce 'NullPointerException'
| @@ -30,6 +30,8 @@ public class RollbackToActionFactory implements ActionFactory { | |||
|
|
|||
| private static final String VERSION = "version"; | |||
|
|
|||
| private static final String IS_TIMESTAMP = "is_timestamp"; | |||
There was a problem hiding this comment.
You should also modify doc.
| } else { | ||
| FileStoreTable fileStoreTable = (FileStoreTable) table; | ||
| fileStoreTable.rollbackTo( | ||
| fileStoreTable.snapshotManager().earlierOrEqualTimeMills(timestamp).id()); |
There was a problem hiding this comment.
Method invocation 'id' may produce 'NullPointerException'
| @@ -32,14 +33,18 @@ public class RollbackToAction extends TableActionBase { | |||
|
|
|||
| private final String version; | |||
|
|
|||
| private final Boolean isTimestamp; | |||
There was a problem hiding this comment.
Why not use long timestamp uniformly in flink action && flink procedure && spark procedure?
There was a problem hiding this comment.
Firstly I also want to change it like you suggested, but consider older version spark user their procedure sql would not be compatible, so keep it can use in older version first. @wwj6591812
| @@ -51,7 +56,16 @@ public void run() throws Exception { | |||
| } | |||
|
|
|||
| if (version.chars().allMatch(Character::isDigit)) { | |||
| table.rollbackTo(Long.parseLong(version)); | |||
| if (isTimestamp != null && isTimestamp) { | |||
| FileStoreTable fileStoreTable = (FileStoreTable) table; | |||
There was a problem hiding this comment.
Move line 60 from here to line 57 like spark procedure.
| @@ -68,17 +68,31 @@ class RollbackProcedureTest extends PaimonSparkTestBase with StreamTest { | |||
| inputData.addData((2, "b")) | |||
| stream.processAllAvailable() | |||
| checkAnswer(query(), Row(1, "a") :: Row(2, "b") :: Nil) | |||
| val ts = System.currentTimeMillis | |||
| Row(true) :: Nil) | ||
| checkAnswer(query(), Row(1, "a") :: Row(2, "b2") :: Nil) | ||
|
|
||
| // rollback to timestamp |
There was a problem hiding this comment.
rollback with timestamp
|
Id you finish, please delete "[WIP]" in title. |
|
Thanks @wwj6591812 had addressed. |
xuzifu666
changed the title
|
+1 |
OK it is a better way，would change it @JingsongLi thanks |
|
Had link this pr to #4410, close the pr firstly~ @JingsongLi @wwj6591812 |
Purpose
Currently Rollback cannot rollback according timestamp which can make user more easy to rollback，this pr is aim to support it.
due to difference in spark and flink args do two ways to support it.
Linked issue: close #xxx
Tests
API and Format
Documentation