address comments

Author: cloud-fan

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2StreamingScanSupportCheck.scala

@@ -30,6 +30,13 @@ object V2StreamingScanSupportCheck extends (LogicalPlan => Unit) {
   import DataSourceV2Implicits._
 
   override def apply(plan: LogicalPlan): Unit = {
+    plan.foreach {
+      case r: StreamingRelationV2 if !r.table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
+        throw new AnalysisException(
+          s"Table ${r.table.name()} does not support either micro-batch or continuous scan.")
+      case _ =>
+    }
+
     val streamingSources = plan.collect {
       case r: StreamingRelationV2 => r.table
     }

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/V2StreamingScanSupportCheckSuite.scala

@@ -58,6 +58,13 @@ class V2StreamingScanSupportCheckSuite extends SparkFunSuite with SharedSparkSes
     V2StreamingScanSupportCheck(Union(plan3, plan4))
   }
 
+  test("table without scan capability") {
+    val e = intercept[AnalysisException] {
+      V2StreamingScanSupportCheck(createStreamingRelation(CapabilityTable(), None))
+    }
+    assert(e.message.contains("does not support either micro-batch or continuous scan"))
+  }
+
   test("mix micro-batch only and continuous only") {
     val plan1 = createStreamingRelation(CapabilityTable(MICRO_BATCH_READ), None)
     val plan2 = createStreamingRelation(CapabilityTable(CONTINUOUS_READ), None)