aokolnychyi opened a new issue #1485:
URL: https://github.com/apache/iceberg/issues/1485
I think we need to discuss the following two cases.
**Case 1 (refresh of cached relations)**
```
sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg",
tableName);
sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
Dataset<Row> query = spark.sql("SELECT * FROM " + tableName + " WHERE id =
1");
query.createOrReplaceTempView("tmp");
spark.sql("CACHE TABLE tmp");
sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
// can be a procedure call as well
// sql("CALL iceberg.system.rollback_to_snapshot(...)")
// !!! cache must be refreshed !!!
assertEquals("View cache must be invalidated",
ImmutableList.of(row(1L, "a"), row(1L, "a"), row(1L, "a")),
sql("SELECT * FROM tmp"));
```
Currently, Spark does not invalidate cached relations that contain a V2
table we just wrote to. I think that's a bug and Spark must match
`InsertIntoDataSourceCommand` that refreshes the cache for V1 tables.
**Case 2 (stored tables in view definitions)**
```
sql("CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg",
tableName);
sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
Dataset<Row> query = spark.sql("SELECT * FROM " + tableName + " WHERE id =
1");
query.createOrReplaceTempView("tmp");
assertEquals("View should have expected rows",
ImmutableList.of(row(1L, "a")),
sql("SELECT * FROM tmp"));
sql("INSERT INTO TABLE %s VALUES (1, 'a')", tableName);
assertEquals("View should have expected rows",
ImmutableList.of(row(1L, "a"), row(1L, "a")),
sql("SELECT * FROM tmp"));
```
Currently, if we disable caching in Iceberg catalogs, the view will get
stale results as the table instance is always different and there is no way we
can refresh it now. As a user, I was surprised because I would expect my table
to be always up to date if caching is disabled.
I propose the following:
- Spark should refresh all cached datasets that contain a reference to the
table it just wrote just like for V1 tables.
- Procedures must refresh all cached datasets that contain a reference to
the table we modified.
- We should override `equals` and `hashCode` for `SparkTable` and `Table`
and it must be independent from caching in Iceberg catalogs. If I have two
instances of `Table` but they point to the same physical table, I think
`equals` should return true. I believe Spark will call
`spark.sharedState.cacheManager.refreshByPlan()` with an instance of
`DataSourceV2Relation` so proper equality will matter.
- If caching in Iceberg catalogs is disabled, we should always refresh the
table before reading.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
