Python API for restoring delta table

 * Add possibility to restore delta table using version or timestamp from pyspark
   Examples:
   ```
   DeltaTable.forPath(spark, path).restoreToVersion(0)
   DeltaTable.forPath(spark, path).restoreToTimestamp('2021-01-01 01:01-01')
   ```

Fixes #890

Signed-off-by: Maksym Dovhal <maksym.dovhal@gmail.com>

python/delta/tables.py

@@ -537,6 +537,46 @@ def upgradeTableProtocol(self, readerVersion: int, writerVersion: int) -> None:
                              type(writerVersion))
         jdt.upgradeTableProtocol(readerVersion, writerVersion)
 
+    @since(1.2)  # type: ignore[arg-type]
+    def restoreToVersion(self, version: int) -> DataFrame:
+        """
+        Restore the DeltaTable to an older version of the table specified by version number.
+
+        Example::
+
+            deltaTable.restoreToVersion(1)
+
+        :param version: target version of restored table
+        :return: Dataframe with metrics of restore operation.
+        :rtype: pyspark.sql.DataFrame
+        """
+
+        return DataFrame(
+            self._jdt.restoreToVersion(version),
+            self._spark._wrapped  # type: ignore[attr-defined]
+        )
+
+    @since(1.2)  # type: ignore[arg-type]
+    def restoreToTimestamp(self, timestamp: str) -> DataFrame:
+        """
+        Restore the DeltaTable to an older version of the table specified by a timestamp.
+        Timestamp can be of the format yyyy-MM-dd or yyyy-MM-dd HH:mm:ss
+
+        Example::
+
+            deltaTable.restoreToTimestamp('2021-01-01')
+            deltaTable.restoreToTimestamp('2021-01-01 01:01:01')
+
+        :param timestamp: target timestamp of restored table
+        :return: Dataframe with metrics of restore operation.
+        :rtype: pyspark.sql.DataFrame
+        """
+
+        return DataFrame(
+            self._jdt.restoreToTimestamp(timestamp),
+            self._spark._wrapped  # type: ignore[attr-defined]
+        )
+
     @staticmethod
     def _dict_to_jmap(
         sparkSession: SparkSession,

python/delta/tests/test_deltatable.py

@@ -791,6 +791,40 @@ def test_protocolUpgrade(self) -> None:
         with self.assertRaisesRegex(ValueError, "writerVersion"):
             dt.upgradeTableProtocol(1, {})  # type: ignore[arg-type]
 
+    def test_restoreToVersion(self) -> None:
+        self.__writeDeltaTable([('a', 1), ('b', 2)])
+        self.__overwriteDeltaTable([('a', 3), ('b', 2)],
+                                   schema=["key_new", "value_new"],
+                                   overwriteSchema='true')
+
+        overwritten = DeltaTable.forPath(self.spark, self.tempFile).toDF()
+        self.__checkAnswer(overwritten, [Row(key_new='a', value_new=3), Row(key_new='b', value_new=2)])
+
+        DeltaTable.forPath(self.spark, self.tempFile).restoreToVersion(0)
+        restored = DeltaTable.forPath(self.spark, self.tempFile).toDF()
+
+        self.__checkAnswer(restored, [Row(key='a', value=1), Row(key='b', value=2)])
+
+    def test_restoreToTimestamp(self) -> None:
+        self.__writeDeltaTable([('a', 1), ('b', 2)])
+        timestampToRestore = DeltaTable.forPath(self.spark, self.tempFile) \
+            .history() \
+            .head() \
+            .timestamp \
+            .strftime('%Y-%m-%d %H:%M:%S.%f')
+
+        self.__overwriteDeltaTable([('a', 3), ('b', 2)],
+                                   schema=["key_new", "value_new"],
+                                   overwriteSchema='true')
+
+        overwritten = DeltaTable.forPath(self.spark, self.tempFile).toDF()
+        self.__checkAnswer(overwritten, [Row(key_new='a', value_new=3), Row(key_new='b', value_new=2)])
+
+        DeltaTable.forPath(self.spark, self.tempFile).restoreToTimestamp(timestampToRestore)
+
+        restored = DeltaTable.forPath(self.spark, self.tempFile).toDF()
+        self.__checkAnswer(restored, [Row(key='a', value=1), Row(key='b', value=2)])
+
     def __checkAnswer(self, df: DataFrame,
                       expectedAnswer: List[Any],
                       schema: Union[StructType, List[str]] = ["key", "value"]) -> None:
@@ -818,9 +852,14 @@ def __writeAsTable(self, datalist: List[Tuple[Any, Any]], tblName: str) -> None:
         df = self.spark.createDataFrame(datalist, ["key", "value"])
         df.write.format("delta").saveAsTable(tblName)
 
-    def __overwriteDeltaTable(self, datalist: List[Tuple[Any, Any]]) -> None:
-        df = self.spark.createDataFrame(datalist, ["key", "value"])
-        df.write.format("delta").mode("overwrite").save(self.tempFile)
+    def __overwriteDeltaTable(self, datalist: List[Tuple[Any, Any]],
+                              schema: Union[StructType, List[str]] = ["key", "value"],
+                              overwriteSchema: str = 'false') -> None:
+        df = self.spark.createDataFrame(datalist, schema)
+        df.write.format("delta") \
+            .option('overwriteSchema', overwriteSchema) \
+            .mode("overwrite") \
+            .save(self.tempFile)
 
     def __createFile(self, fileName: str, content: Any) -> None:
         with open(os.path.join(self.tempFile, fileName), 'w') as f: