feat: Supports table delete . (#40)

Author: yuanoOo

spark-connector-oceanbase/spark-connector-oceanbase-base/src/main/scala/com/oceanbase/spark/catalog/OceanBaseTable.scala

@@ -22,12 +22,14 @@ import com.oceanbase.spark.reader.v2.OBJdbcScanBuilder
 import com.oceanbase.spark.utils.OBJdbcUtils
 import com.oceanbase.spark.writer.v2.{DirectLoadWriteBuilderV2, JDBCWriteBuilder}
 
+import org.apache.spark.sql.ExprUtils.compileFilter
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.connector.catalog._
 import org.apache.spark.sql.connector.catalog.TableCapability._
 import org.apache.spark.sql.connector.read.ScanBuilder
 import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
+import org.apache.spark.sql.sources.Filter
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
 
@@ -44,7 +46,8 @@ case class OceanBaseTable(
   extends Table
   with SupportsRead
   with SupportsWrite
-  with TruncatableTable {
+  with TruncatableTable
+  with SupportsDelete {
 
   override def name(): String = ident.toString
 
@@ -84,4 +87,33 @@ case class OceanBaseTable(
       case Failure(_) => false
     }
   }
+
+  override def canDeleteWhere(filters: Array[Filter]): Boolean = {
+    filters.forall(filter => compileFilter(filter, dialect).isDefined)
+  }
+
+  override def deleteWhere(filters: Array[Filter]): Unit = {
+    val filterWhereClause: String =
+      filters
+        .flatMap(compileFilter(_, dialect))
+        .map(p => s"($p)")
+        .mkString(" AND ")
+
+    val whereClause: String = {
+      if (filterWhereClause.nonEmpty) {
+        "WHERE " + filterWhereClause
+      } else {
+        ""
+      }
+    }
+    OBJdbcUtils.withConnection(config) {
+      conn =>
+        {
+          OBJdbcUtils.executeStatement(
+            conn,
+            config,
+            dialect.getDeleteWhereSql(config.getDbTable, whereClause))
+        }
+    }
+  }
 }

spark-connector-oceanbase/spark-connector-oceanbase-base/src/main/scala/com/oceanbase/spark/dialect/OceanBaseDialect.scala

@@ -128,6 +128,12 @@ abstract class OceanBaseDialect extends Logging with Serializable {
     s"TRUNCATE TABLE $table"
   }
 
+  def getDeleteWhereSql(tableName: String, whereClause: String): String = {
+    val sql = s"DELETE FROM $tableName $whereClause"
+    logInfo(s"The generated OceanBase delete sql statement: $sql")
+    sql
+  }
+
   def getJDBCType(dt: DataType): Option[JdbcType] = None
 
   /** Creates a schema. */

spark-connector-oceanbase/spark-connector-oceanbase-base/src/main/scala/com/oceanbase/spark/utils/OBJdbcUtils.scala

@@ -112,6 +112,9 @@ object OBJdbcUtils {
     try {
       statement.setQueryTimeout(config.getJdbcQueryTimeout)
       statement.executeUpdate(sql)
+    } catch {
+      case exception: Exception =>
+        throw new RuntimeException(s"Failed to execute sql: $sql", exception.getCause)
     } finally {
       statement.close()
     }

spark-connector-oceanbase/spark-connector-oceanbase-base/src/test/scala/com/oceanbase/spark/OBCatalogMySQLITCase.scala

@@ -231,6 +231,42 @@ class OBCatalogMySQLITCase extends OceanBaseMySQLTestBase {
     session.stop()
   }
 
+  @Test
+  def testDeleteWhere(): Unit = {
+    val session = SparkSession
+      .builder()
+      .master("local[*]")
+      .config("spark.sql.catalog.ob", OB_CATALOG_CLASS)
+      .config("spark.sql.catalog.ob.url", getJdbcUrl)
+      .config("spark.sql.catalog.ob.username", getUsername)
+      .config("spark.sql.catalog.ob.password", getPassword)
+      .config("spark.sql.catalog.ob.schema-name", getSchemaName)
+      .getOrCreate()
+
+    session.sql("use ob;")
+    insertTestData(session, "products")
+    session.sql("delete from products where 1 = 0")
+    queryAndVerifyTableData(session, "products", expected)
+
+    session.sql("delete from products where id = 1")
+    queryAndVerifyTableData(session, "products", expected)
+
+    session.sql("delete from products where description is null")
+    queryAndVerifyTableData(session, "products", expected)
+
+    session.sql("delete from products where id in (101, 102, 103)")
+    session.sql("delete from products where name = 'hammer'")
+
+    session.sql("delete from products where name like 'rock%'")
+    session.sql("delete from products where name like '%jack%' and id = 108 or weight = 5.3")
+    session.sql("delete from products where id >= 109")
+
+    val expect = new util.ArrayList[String]()
+    queryAndVerifyTableData(session, "products", expect)
+
+    session.stop()
+  }
+
   private def queryAndVerifyTableData(
       session: SparkSession,
       tableName: String,