Remove validateAppendOnceFlows

Author: sryza

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala

@@ -191,8 +191,6 @@ case class DataflowGraph(flows: Seq[Flow], tables: Seq[Table], views: Seq[View])
     validatePersistedViewSources()
     validateEveryDatasetHasFlow()
     validateTablesAreResettable()
-    validateAppendOnceFlows()
-    // Ensures that all flows are resolved and have a valid schema.
     inferredSchema
   }.failed
 

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala

@@ -192,11 +192,5 @@ class AppendOnceFlow(
     val funcResult: FlowFunctionResult
 ) extends ResolvedFlow {
 
-  /**
-   * Whether the flow was declared as once or not in UnresolvedFlow. If false, then it means the
-   * flow is created from batch query.
-   */
-  val definedAsOnce: Boolean = flow.once
-
   override val once = true
 }

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala

@@ -124,26 +124,6 @@ trait GraphValidations extends Logging {
       }
   }
 
-  /**
-   * Validate if we have any append only flows writing into a streaming table but was created
-   * from a batch query.
-   */
-  protected def validateAppendOnceFlows(): Seq[GraphValidationWarning] = {
-    flows
-      .filter {
-        case af: AppendOnceFlow => !af.definedAsOnce
-        case _ => false
-      }
-      .groupBy(_.destinationIdentifier)
-      .flatMap {
-        case (destination, flows) =>
-          table
-            .get(destination)
-            .map(t => AppendOnceFlowCreatedFromBatchQueryException(t, flows.map(_.identifier)))
-      }
-      .toSeq
-  }
-
   protected def validateUserSpecifiedSchemas(): Unit = {
     flows.flatMap(f => table.get(f.identifier)).foreach { t: TableInput =>
       // The output inferred schema of a table is the declared schema merged with the