Time travel not work

[lordk911]

I'm testing with spark3.0.1 and cdh5.14 ,iceberg0.9.1. and spark-shell
catalog config is :

spark.sql.catalog.hadoop_prod               org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.hadoop_prod.type          hadoop
spark.sql.catalog.hadoop_prod.warehouse     hdfs://hdfsnamespace/user/hive/warehouse

I tried to create a table :
scala> spark.sql("CREATE TABLE hadoop_prod.ice.icetest (id bigint, data string) USING iceberg PARTITIONED BY (id) ")

insert some value:

scala> spark.sql("INSERT INTO hadoop_prod.ice.icetest VALUES (1, 'a'), (2, 'b'), (3, 'c')")
scala> spark.sql("select * from hadoop_prod.ice.icetest ").show(false)
+---+----+                                                                      
|id |data|
+---+----+
|1  |a   |
|2  |b   |
|3  |c   |
+---+----+

delete a partion:

scala> spark.sql("delete from hadoop_prod.ice.icetest where id=1").show(false)
scala> spark.sql("select * from hadoop_prod.ice.icetest ").show(false)
+---+----+
|id |data|
+---+----+
|2  |b   |
|3  |c   |
+---+----+

insert some value:

scala> spark.sql("INSERT INTO hadoop_prod.ice.icetest VALUES (1, 'a'), (1, 'b'), (1, 'c')")
scala> spark.sql("select * from hadoop_prod.ice.icetest ").show(false)
+---+----+
|id |data|
+---+----+
|2  |b   |
|3  |c   |
|1  |a   |
|1  |b   |
|1  |c   |
+---+----+

show snapshots:

scala> spark.sql("select committed_at, snapshot_id, parent_id, operation from hadoop_prod.ice.icetest.snapshots").show(false)
+-----------------------+-------------------+------------------+---------+
|committed_at           |snapshot_id        |parent_id         |operation|
+-----------------------+-------------------+------------------+---------+
|2020-09-16 13:32:39.952|628886310322778010 |null              |append   |
|2020-09-16 13:42:34.109|598127609483871079 |628886310322778010|delete   |
|2020-09-16 13:43:14.415|6880502734717374864|598127609483871079|append   |
+-----------------------+-------------------+------------------+---------+

but every snapshot I read ,show the same last state of the table:

scala> val df2 = spark.read.option("snapshot-id", 628886310322778010L).table("hadoop_prod.ice.icetest")
df2: org.apache.spark.sql.DataFrame = [id: bigint, data: string]

scala> df2.show
+---+----+                                                                      
| id|data|
+---+----+
|  2|   b|
|  3|   c|
|  1|   a|
|  1|   b|
|  1|   c|
+---+----+

[RussellSpitzer]

Repo'd this, running tests checking to see what the issue is

[RussellSpitzer]

Ok unfortunate thing here.

If you use "table" in your read command it will ignore all other options passed to the builder.
So
spark.read.option("snapshot-id", 628886310322778010L).table("hadoop_prod.ice.icetest") is equivalent to
spark.table("hadoop_prod.ice.icetest")

See
https://github.com/apache/spark/blob/v3.0.1/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala#L795-L798

BUT Good news!
This is fixed in master, the fix version is marked as 3.1.0
https://issues.apache.org/jira/browse/SPARK-32592

https://github.com/apache/spark/blob/56ae95053df4afa9764df3f1d88f300896ca0183/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala#L832-L838

For now you'll need to use the catalog free pathway to load from a snapshot, I'll try to see if there is another approach but

spark.read.format("iceberg").option("snapshot-id", "6159119813997540061").load("/tmp/ice/icetest") // tmp was my warehouse dir

Works

[RussellSpitzer]

@rdblue we'll need to modify the docs here since the examples given will not work. Also something I was mentioning internally, it would be great if we had unit tests for all pieces of example code in the docs to catch things like this.

[rdblue]

Yeah, we should update this to use load instead of table. It's too bad because that goes through the IcebergSource rather than a catalog.

[RussellSpitzer]

Spark 3.1 :) Although honestly I could see the case for back porting that. Discarding user input should at the very least warn, but probably should have been an exception.

[RussellSpitzer]

Doc PR Up

[rdblue]

I merged the PR and deployed docs. Thanks, @RussellSpitzer!

@lordk911, it looks like the problem is in Spark. Can you try with the new examples? If those work, then we can close this issue.

[lordk911]

ok , test passed by use load method:

scala> val df = spark.read.format("iceberg").option("snapshot-id", "7975731372139528551").load("/user/hive/warehouse/ice/icetest")
df: org.apache.spark.sql.DataFrame = [id: bigint, data: string]

scala> df.show(false)
+---+----+                                                                      
|id |data|
+---+----+
|1  |a   |
|2  |b   |
|3  |c   |
+---+----+


scala> val df = spark.read.format("iceberg").option("snapshot-id", "4063381308553626534").load("/user/hive/warehouse/ice/icetest")
df: org.apache.spark.sql.DataFrame = [id: bigint, data: string]

scala> df.show(false)
+---+----+                                                                      
|id |data|
+---+----+
|2  |b   |
|3  |c   |
+---+----+

Thank you !

[rdblue]

Thank you for confirming that this works!