fix: don't throw on invalid columns in DropColumns

core/src/main/scala/com/microsoft/azure/synapse/ml/stages/DropColumns.scala

@@ -38,28 +38,14 @@ class DropColumns(val uid: String) extends Transformer with Wrappable with Defau
     */
   override def transform(dataset: Dataset[_]): DataFrame = {
     logTransform[DataFrame]({
-      verifySchema(dataset.schema)
       dataset.toDF().drop(getCols: _*)
     })
   }
 
   def transformSchema(schema: StructType): StructType = {
-    verifySchema(schema)
     val droppedCols = getCols.toSet
     StructType(schema.fields.filter(f => !droppedCols(f.name)))
   }
 
   def copy(extra: ParamMap): DropColumns = defaultCopy(extra)
-
-  private def verifySchema(schema: StructType): Unit = {
-    val providedCols = schema.fields.map(_.name).toSet
-    val invalidCols = getCols.filter(!providedCols(_))
-
-    if (invalidCols.length > 0) {
-      throw new NoSuchElementException(
-        s"DataFrame does not contain specified columns: ${invalidCols.reduce(_ + "," + _)}")
-    }
-
-  }
-
 }

core/src/test/scala/com/microsoft/azure/synapse/ml/stages/DropColumnsSuite.scala

@@ -42,10 +42,13 @@ class DropColumnsSuite extends TestBase with TransformerFuzzing[DropColumns] {
 
   test("Invalid column specified") {
     try {
-      new DropColumns().setCol("four").transform(makeBasicDF())
-      fail()
+        val df = makeBasicDF()
+        new DropColumns().setCol("four").transform(df)
+        val result = new DropColumns().setCol("four").transform(df)
+        assert(df.schema == result.schema)
     } catch {
-      case _: NoSuchElementException =>
+      case _: Exception =>
+        fail("DropColumns should not throw when for invalid column input")
     }
   }