diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index 4981db5537a7..14ec64f874c3 100644
--- a/datafusion/sql/src/statement.rs
+++ b/datafusion/sql/src/statement.rs
@@ -1108,12 +1108,8 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                     plan_err!("Delete-order-by clause not yet supported")?;
                 }
 
-                if limit.is_some() {
-                    plan_err!("Delete-limit clause not yet supported")?;
-                }
-
                 let table_name = self.get_delete_target(from)?;
-                self.delete_to_plan(&table_name, selection)
+                self.delete_to_plan(&table_name, selection, limit)
             }
 
             Statement::StartTransaction {
@@ -1322,7 +1318,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 let function_body = match function_body {
                     Some(r) => Some(self.sql_to_expr(
                         match r {
-                            // `link_symbol` indicates if the primary expression contains the name of shared library file. 
+                            // `link_symbol` indicates if the primary expression contains the name of shared library file.
                             ast::CreateFunctionBody::AsBeforeOptions{body: expr, link_symbol: _link_symbol} => expr,
                             ast::CreateFunctionBody::AsAfterOptions(expr) => expr,
                             ast::CreateFunctionBody::Return(expr) => expr,
@@ -2072,6 +2068,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         &self,
         table_name: &ObjectName,
         predicate_expr: Option<SQLExpr>,
+        limit: Option<SQLExpr>,
     ) -> Result<LogicalPlan> {
         // Do a table lookup to verify the table exists
         let table_ref = self.object_name_to_table_reference(table_name.clone())?;
@@ -2085,7 +2082,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                 .build()?;
         let mut planner_context = PlannerContext::new();
 
-        let source = match predicate_expr {
+        let mut source = match predicate_expr {
             None => scan,
             Some(predicate_expr) => {
                 let filter_expr =
@@ -2102,6 +2099,14 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             }
         };
 
+        if let Some(limit) = limit {
+            let empty_schema = DFSchema::empty();
+            let limit = self.sql_to_expr(limit, &empty_schema, &mut planner_context)?;
+            source = LogicalPlanBuilder::from(source)
+                .limit_by_expr(None, Some(limit))?
+                .build()?
+        }
+
         let plan = LogicalPlan::Dml(DmlStatement::new(
             table_ref,
             table_source,
diff --git a/datafusion/sqllogictest/test_files/delete.slt b/datafusion/sqllogictest/test_files/delete.slt
index e86343b6bf5f..b01eb6f5e9ec 100644
--- a/datafusion/sqllogictest/test_files/delete.slt
+++ b/datafusion/sqllogictest/test_files/delete.slt
@@ -113,3 +113,30 @@ logical_plan
 05)--------TableScan: t2
 06)----TableScan: t1
 physical_plan_error This feature is not implemented: Physical plan does not support logical expression InSubquery(InSubquery { expr: Column(Column { relation: Some(Bare { table: "t1" }), name: "a" }), subquery: <subquery>, negated: false })
+
+
+# Delete with limit
+
+query TT
+explain delete from t1 limit 10
+----
+logical_plan
+01)Dml: op=[Delete] table=[t1]
+02)--Limit: skip=0, fetch=10
+03)----TableScan: t1
+physical_plan
+01)CooperativeExec
+02)--DmlResultExec: rows_affected=0
+
+
+query TT
+explain delete from t1 where a = 1 and b = '2' limit 10
+----
+logical_plan
+01)Dml: op=[Delete] table=[t1]
+02)--Limit: skip=0, fetch=10
+03)----Filter: CAST(t1.a AS Int64) = Int64(1) AND t1.b = CAST(Utf8("2") AS Utf8View)
+04)------TableScan: t1
+physical_plan
+01)CooperativeExec
+02)--DmlResultExec: rows_affected=0