diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
index 12c78c347e43..4afe8a9900eb 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
@@ -32,6 +32,7 @@
 import static org.apache.hadoop.hive.metastore.ColumnType.VARCHAR_TYPE_NAME;
 
 import java.sql.Connection;
+import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
@@ -366,6 +367,32 @@ private void executeNoResult(final String queryText) throws SQLException {
     }
   }
 
+  @FunctionalInterface
+  private interface PreparedStatementSetter {
+    void setParameters(PreparedStatement ps) throws SQLException;
+  }
+
+  private int executePreparedUpdate(final String baseSql, PreparedStatementSetter setter) throws SQLException {
+    JDOConnection jdoConn = pm.getDataStoreConnection();
+    PreparedStatement ps = null;
+    boolean doTrace = LOG.isDebugEnabled();
+    try {
+      Connection conn = (Connection) jdoConn.getNativeConnection();
+
+      long start = doTrace ? System.nanoTime() : 0;
+      ps = conn.prepareStatement(baseSql);
+      setter.setParameters(ps);
+      int rowsAffected = ps.executeUpdate();
+      MetastoreDirectSqlUtils.timingTrace(doTrace, baseSql, start, doTrace ? System.nanoTime() : 0);
+      return rowsAffected;
+    } finally {
+      if (ps != null) {
+        ps.close();
+      }
+      jdoConn.close();
+    }
+  }
+
   public Database getDatabase(String catName, String dbName) throws MetaException{
     String queryTextDbSelector= "select "
         + "\"DB_ID\", \"NAME\", \"DB_LOCATION_URI\", \"DESC\", "
@@ -3243,20 +3270,36 @@ public void deleteColumnStatsState(long tbl_id) throws MetaException {
   }
 
   public boolean deleteTableColumnStatistics(long tableId, List<String> colNames, String engine) {
-    String deleteSql = "delete from " + TAB_COL_STATS + " where \"TBL_ID\" = " + tableId;
+    StringBuilder deleteSql = new StringBuilder("DELETE FROM " + TAB_COL_STATS + " WHERE \"TBL_ID\" = ?");
+    List<String> inClauseValues = null;
     if (colNames != null && !colNames.isEmpty()) {
-      deleteSql += " and \"COLUMN_NAME\" in (" + colNames.stream().map(col -> "'" + col + "'").collect(Collectors.joining(",")) + ")";
+      // Append a placeholder '?' for each column name
+      String placeholders = colNames.stream().map(col -> "?").collect(Collectors.joining(","));
+      deleteSql.append(" AND \"COLUMN_NAME\" IN (").append(placeholders).append(")");
+      inClauseValues = colNames; // Store the values to be bound later
     }
     if (engine != null) {
-      deleteSql += " and \"ENGINE\" = '" + engine + "'";
+      deleteSql.append(" AND \"ENGINE\" = ?");
     }
+    final List<String> finalInClauseValues = inClauseValues;
     try {
-      executeNoResult(deleteSql);
+      int rowsAffected = executePreparedUpdate(deleteSql.toString(), ps -> {
+        int paramIndex = 1;
+        ps.setLong(paramIndex++, tableId);
+        if (finalInClauseValues != null) {
+          for (String colName : finalInClauseValues) {
+            ps.setString(paramIndex++, colName);
+          }
+        }
+        if (engine != null) {
+          ps.setString(paramIndex++, engine);
+        }
+      });
+      return rowsAffected > 0;
     } catch (SQLException e) {
       LOG.warn("Error removing table column stats. ", e);
       return false;
     }
-    return true;
   }
 
   public boolean deletePartitionColumnStats(String catName, String dbName, String tblName,
@@ -3268,15 +3311,29 @@ public List<Void> run(List<String> input) throws Exception {
         List<Long> partitionIds = getPartitionIdsViaSqlFilter(catName, dbName, tblName, sqlFilter,
             input, Collections.emptyList(), -1);
         if (!partitionIds.isEmpty()) {
-          String deleteSql = "delete from " + PART_COL_STATS + " where \"PART_ID\" in ( " + getIdListForIn(partitionIds) + ")";
+          StringBuilder deleteSql = new StringBuilder("DELETE FROM " + PART_COL_STATS + " WHERE \"PART_ID\" IN (");
+          deleteSql.append(makeParams(partitionIds.size())).append(")");
           if (colNames != null && !colNames.isEmpty()) {
-            deleteSql += " and \"COLUMN_NAME\" in (" + colNames.stream().map(col -> "'" + col + "'").collect(Collectors.joining(",")) + ")";
+            deleteSql.append(" AND \"COLUMN_NAME\" IN (").append(makeParams(colNames.size())).append(")");
           }
           if (engine != null) {
-            deleteSql += " and \"ENGINE\" = '" + engine + "'";
+            deleteSql.append(" AND \"ENGINE\" = ?");
           }
           try {
-            executeNoResult(deleteSql);
+            executePreparedUpdate(deleteSql.toString(), ps -> {
+              int paramIndex = 1;
+              for (Long id : partitionIds) {
+                ps.setLong(paramIndex++, id);
+              }
+              if (colNames != null && !colNames.isEmpty()) {
+                for (String colName : colNames) {
+                  ps.setString(paramIndex++, colName);
+                }
+              }
+              if (engine != null) {
+                ps.setString(paramIndex++, engine);
+              }
+            });
           } catch (SQLException e) {
             LOG.warn("Error removing partition column stats. ", e);
             throw new MetaException("Error removing partition column stats: " + e.getMessage());
diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java
index 4a2408b69e2a..e5436d57e916 100644
--- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java
+++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java
@@ -133,6 +133,9 @@
 import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsNot.not;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 @Category(MetastoreUnitTest.class)
 public class TestObjectStore {
@@ -955,6 +958,56 @@ public void testTableStatisticsOps() throws Exception {
     Assert.assertEquals(0, tabColStats.size());
   }
 
+  @Test
+  public void testDirectSQLDropStatsSQlInject() throws Exception {
+    createPartitionedTable(true, true);
+    List<ColumnStatistics> tabColStats;
+    ColumnStatisticsDesc statsDesc = new ColumnStatisticsDesc(true, DB1, TABLE1);
+    ColumnStatisticsObj statsObj1 = new ColumnStatisticsObj("test_col1", "int",
+        new ColumnStatisticsData(ColumnStatisticsData._Fields.DECIMAL_STATS, new DecimalColumnStatsData(100, 1000)));
+    ColumnStatisticsObj statsObj2 = new ColumnStatisticsObj("test_col2", "int",
+        new ColumnStatisticsData(ColumnStatisticsData._Fields.DECIMAL_STATS, new DecimalColumnStatsData(200, 2000)));
+    ColumnStatistics colStats = new ColumnStatistics(statsDesc, Arrays.asList(statsObj1, statsObj2));
+    colStats.setEngine(ENGINE);
+    objectStore.updateTableColumnStatistics(colStats, null, 0);
+    try (AutoCloseable c = deadline()) {
+      tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+          Arrays.asList("test_col1", "test_col2"));
+    }
+    Assert.assertEquals(1, tabColStats.size());
+    Assert.assertEquals(2, tabColStats.get(0).getStatsObjSize());
+
+    String sqlInjectEngine = "hive' OR 1=1 --'"; // This can delete all the records
+    assertFalse(objectStore.deleteTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1, "test_col1", sqlInjectEngine));
+    try (AutoCloseable c = deadline()) {
+      tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+          Arrays.asList("test_col1", "test_col2"));
+    }
+    Assert.assertEquals(1, tabColStats.size());
+    Assert.assertEquals(2, tabColStats.get(0).getStatsObjSize()); // Verify that sql injection didn't delete any records
+
+    AggrStats aggrStats;
+    try (AutoCloseable c = deadline()) {
+      aggrStats = objectStore.get_aggr_stats_for(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+          Arrays.asList("test_part_col=a0", "test_part_col=a1", "test_part_col=a2"),
+          Collections.singletonList("test_part_col"), ENGINE);
+    }
+    List<ColumnStatisticsObj> stats = aggrStats.getColStats();
+    Assert.assertEquals(1, stats.size());
+    Assert.assertEquals(3, aggrStats.getPartsFound());
+
+    objectStore.deletePartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+        "test_part_col=a0", Arrays.asList("a0"), null, sqlInjectEngine);
+    List<List<ColumnStatistics>> partitionStats;
+    try (AutoCloseable c = deadline()) {
+      partitionStats = objectStore.getPartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+          Arrays.asList("test_part_col=a0", "test_part_col=a1", "test_part_col=a2"),
+          Collections.singletonList("test_part_col"));
+    }
+    Assert.assertEquals(1, partitionStats.size());
+    Assert.assertEquals(3, partitionStats.get(0).size());
+  }
+
   @Test
   public void testPartitionStatisticsOps() throws Exception {
     createPartitionedTable(true, true);