[GitHub] [spark] MaxGekk commented on a change in pull request #30983: [SPARK-33950][SQL] Refresh cache in v1 `ALTER TABLE .. DROP PARTITION`

[GitBox <gi...@apache.org>]

MaxGekk commented on a change in pull request #30983:
URL: https://github.com/apache/spark/pull/30983#discussion_r550748425



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala
##########
@@ -581,6 +581,7 @@ case class AlterTableDropPartitionCommand(
       table.identifier, normalizedSpecs, ignoreIfNotExists = ifExists, purge = purge,
       retainData = retainData)
 
+    sparkSession.catalog.refreshTable(table.identifier.quotedString)

Review comment:
       We can see the problem when we comment the caching:
   ```scala
     test("SPARK-33941: refresh cache after partition dropping") {
       withNamespaceAndTable("ns", "tbl") { t =>
         sql(s"CREATE TABLE $t (id int, part int) $defaultUsing PARTITIONED BY (part)")
         sql(s"INSERT INTO $t PARTITION (part=0) SELECT 0")
         sql(s"INSERT INTO $t PARTITION (part=1) SELECT 1")
         val df = spark.table(t)
         // df.cache()
         df.collect()
         sql(s"ALTER TABLE $t DROP PARTITION (part=0)")
         df.collect()
       }
     }
   ```
   the test fails on the second `df.collect()`:
   ```
   java.io.FileNotFoundException: File file:.../ns.db/tbl/part=0/part-00000-44671e51-2b9a-4049-b449-4ea454abbe77.c000.snappy.parquet does not exist
   ```
   
   The reason of that is in:
   1. `val df = spark.table(t)` creates `FileSourceScanExec` in which `inputRDD` has not been materialized yet: https://github.com/apache/spark/blob/a093d6feefb0e086d19c86ae53bf92df12ccf2fa/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala#L398
   2. The first `df.collect()` materiazes `inputRDD` -> `createNonBucketedReadRDD()` where `dynamicallySelectedPartitions` are:
       
   
   - PartitionDirectory([0],WrappedArray(LocatedFileStatus{path=file:../ns.db/tbl/part=0 ...
   - PartitionDirectory([1],WrappedArray(LocatedFileStatus{path=file:.../ns.db/tbl/part=1 ...
   
   In this ways, `inputRDD` becomes `FileScanRDD` with 2 partitions that pointed out to 2 folders `part=0` and `part=1`
   3. `sql(s"ALTER TABLE $t DROP PARTITION (part=0)")` deletes the `part=0` folder (and invalidates the cache for partition data after the fix).
   4. Since `df` contains the same `FileSourceScanExec` with already materialized `inputRDD`, the second action `df.collect()` tries to read the deleted folder `part=0`. As a consequence of that, we observe the exception: `java.io.FileNotFoundException: File file:.../ns.db/tbl/part=0/... does not exist`
   
   As you can see, the problem of SPARK-33941 is not only in caching.




----------------------------------------------------------------
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:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org