comments

sunchao · sunchao · commit 98e9edeabc7d · 2020-11-28T10:36:06.000-08:00
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala
@@ -38,10 +38,10 @@ case class AppendDataExecV1(
     table: SupportsWrite,
     writeOptions: CaseInsensitiveStringMap,
     plan: LogicalPlan,
-    afterWrite: () => Unit) extends V1FallbackWriters {
+    refreshCache: () => Unit) extends V1FallbackWriters {
 
   override protected def run(): Seq[InternalRow] = {
-    writeWithV1(newWriteBuilder().buildForV1Write(), afterWrite = afterWrite)
+    writeWithV1(newWriteBuilder().buildForV1Write(), refreshCache = refreshCache)
   }
 }
 
@@ -61,7 +61,7 @@ case class OverwriteByExpressionExecV1(
     deleteWhere: Array[Filter],
     writeOptions: CaseInsensitiveStringMap,
     plan: LogicalPlan,
-    afterWrite: () => Unit) extends V1FallbackWriters {
+    refreshCache: () => Unit) extends V1FallbackWriters {
 
   private def isTruncate(filters: Array[Filter]): Boolean = {
     filters.length == 1 && filters(0).isInstanceOf[AlwaysTrue]
@@ -70,11 +70,11 @@ case class OverwriteByExpressionExecV1(
   override protected def run(): Seq[InternalRow] = {
     newWriteBuilder() match {
       case builder: SupportsTruncate if isTruncate(deleteWhere) =>
-        writeWithV1(builder.truncate().asV1Builder.buildForV1Write(), afterWrite = afterWrite)
+        writeWithV1(builder.truncate().asV1Builder.buildForV1Write(), refreshCache = refreshCache)
 
       case builder: SupportsOverwrite =>
         writeWithV1(builder.overwrite(deleteWhere).asV1Builder.buildForV1Write(),
-          afterWrite = afterWrite)
+          refreshCache = refreshCache)
 
       case _ =>
         throw new SparkException(s"Table does not support overwrite by expression: $table")
@@ -117,11 +117,11 @@ trait SupportsV1Write extends SparkPlan {
 
   protected def writeWithV1(
       relation: InsertableRelation,
-      afterWrite: () => Unit = () => ()): Seq[InternalRow] = {
+      refreshCache: () => Unit = () => ()): Seq[InternalRow] = {
     val session = sqlContext.sparkSession
     // The `plan` is already optimized, we should not analyze and optimize it again.
     relation.insert(AlreadyOptimized.dataFrame(session, plan), overwrite = false)
-    afterWrite()
+    refreshCache()
 
     Nil
   }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -216,11 +216,11 @@ case class AppendDataExec(
     table: SupportsWrite,
     writeOptions: CaseInsensitiveStringMap,
     query: SparkPlan,
-    afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
+    refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
 
   override protected def run(): Seq[InternalRow] = {
     val writtenRows = writeWithV2(newWriteBuilder().buildForBatch())
-    afterWrite()
+    refreshCache()
     writtenRows
   }
 }
@@ -240,7 +240,7 @@ case class OverwriteByExpressionExec(
     deleteWhere: Array[Filter],
     writeOptions: CaseInsensitiveStringMap,
     query: SparkPlan,
-    afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
+    refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
 
   private def isTruncate(filters: Array[Filter]): Boolean = {
     filters.length == 1 && filters(0).isInstanceOf[AlwaysTrue]
@@ -257,7 +257,7 @@ case class OverwriteByExpressionExec(
       case _ =>
         throw new SparkException(s"Table does not support overwrite by expression: $table")
     }
-    afterWrite()
+    refreshCache()
     writtenRows
   }
 }
@@ -276,7 +276,7 @@ case class OverwritePartitionsDynamicExec(
     table: SupportsWrite,
     writeOptions: CaseInsensitiveStringMap,
     query: SparkPlan,
-    afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
+    refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {
 
   override protected def run(): Seq[InternalRow] = {
     val writtenRows = newWriteBuilder() match {
@@ -286,7 +286,7 @@ case class OverwritePartitionsDynamicExec(
       case _ =>
         throw new SparkException(s"Table does not support dynamic partition overwrite: $table")
     }
-    afterWrite()
+    refreshCache()
     writtenRows
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -216,11 +216,11 @@ case class AppendDataExec(`
`216`	`216`	`table: SupportsWrite,`
`217`	`217`	`writeOptions: CaseInsensitiveStringMap,`
`218`	`218`	`query: SparkPlan,`
`219`		`- afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
	`219`	`+ refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
`220`	`220`
`221`	`221`	`override protected def run(): Seq[InternalRow] = {`
`222`	`222`	`val writtenRows = writeWithV2(newWriteBuilder().buildForBatch())`
`223`		`- afterWrite()`
	`223`	`+ refreshCache()`
`224`	`224`	`writtenRows`
`225`	`225`	`}`
`226`	`226`	`}`
`@@ -240,7 +240,7 @@ case class OverwriteByExpressionExec(`
`240`	`240`	`deleteWhere: Array[Filter],`
`241`	`241`	`writeOptions: CaseInsensitiveStringMap,`
`242`	`242`	`query: SparkPlan,`
`243`		`- afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
	`243`	`+ refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
`244`	`244`
`245`	`245`	`private def isTruncate(filters: Array[Filter]): Boolean = {`
`246`	`246`	`filters.length == 1 && filters(0).isInstanceOf[AlwaysTrue]`
`@@ -257,7 +257,7 @@ case class OverwriteByExpressionExec(`
`257`	`257`	`case _ =>`
`258`	`258`	`throw new SparkException(s"Table does not support overwrite by expression: $table")`
`259`	`259`	`}`
`260`		`- afterWrite()`
	`260`	`+ refreshCache()`
`261`	`261`	`writtenRows`
`262`	`262`	`}`
`263`	`263`	`}`
`@@ -276,7 +276,7 @@ case class OverwritePartitionsDynamicExec(`
`276`	`276`	`table: SupportsWrite,`
`277`	`277`	`writeOptions: CaseInsensitiveStringMap,`
`278`	`278`	`query: SparkPlan,`
`279`		`- afterWrite: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
	`279`	`+ refreshCache: () => Unit) extends V2TableWriteExec with BatchWriteHelper {`
`280`	`280`
`281`	`281`	`override protected def run(): Seq[InternalRow] = {`
`282`	`282`	`val writtenRows = newWriteBuilder() match {`
`@@ -286,7 +286,7 @@ case class OverwritePartitionsDynamicExec(`
`286`	`286`	`case _ =>`
`287`	`287`	`throw new SparkException(s"Table does not support dynamic partition overwrite: $table")`
`288`	`288`	`}`
`289`		`- afterWrite()`
	`289`	`+ refreshCache()`
`290`	`290`	`writtenRows`
`291`	`291`	`}`
`292`	`292`	`}`