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..415d7c7f557a 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
@@ -133,6 +133,7 @@
  */
 class MetaStoreDirectSql {
   private static final int NO_BATCHING = -1, DETECT_BATCHING = 0;
+  private static final Set<String> ALLOWED_TABLES_TO_LOCK = Set.of("NOTIFICATION_SEQUENCE");
 
   private static final Logger LOG = LoggerFactory.getLogger(MetaStoreDirectSql.class);
   private final PersistenceManager pm;
@@ -3203,6 +3204,11 @@ private void getStatsTableListResult(
   }
 
   public void lockDbTable(String tableName) throws MetaException {
+    // Only certain tables are allowed to be locked, and the API should restrict them.
+    if (!ALLOWED_TABLES_TO_LOCK.contains(tableName)) {
+      throw new MetaException("Error while locking table " + tableName);
+    }
+
     String lockCommand = "lock table \"" + tableName + "\" in exclusive mode";
     try {
       executeNoResult(lockCommand);
@@ -3243,19 +3249,26 @@ 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;
+    String deleteSql = "delete from " + TAB_COL_STATS + " where \"TBL_ID\" = ?";
+    List<Object> params = new ArrayList<>(colNames == null ? 2 : colNames.size() + 2);
+    params.add(tableId);
+
     if (colNames != null && !colNames.isEmpty()) {
-      deleteSql += " and \"COLUMN_NAME\" in (" + colNames.stream().map(col -> "'" + col + "'").collect(Collectors.joining(",")) + ")";
+      deleteSql += " and \"COLUMN_NAME\" in (" + makeParams(colNames.size()) + ")";
+      params.addAll(colNames);
     }
+
     if (engine != null) {
-      deleteSql += " and \"ENGINE\" = '" + engine + "'";
+      deleteSql += " and \"ENGINE\" = ?";
+      params.add(engine);
     }
-    try {
-      executeNoResult(deleteSql);
-    } catch (SQLException e) {
-      LOG.warn("Error removing table column stats. ", e);
+
+    try (QueryWrapper queryParams = new QueryWrapper(pm.newQuery("javax.jdo.query.SQL", deleteSql))) {
+      executeWithArray(queryParams.getInnerQuery(), params.toArray(), deleteSql);
+    } catch (MetaException e) {
       return false;
     }
+
     return true;
   }
 
@@ -3269,17 +3282,20 @@ public List<Void> run(List<String> input) throws Exception {
             input, Collections.emptyList(), -1);
         if (!partitionIds.isEmpty()) {
           String deleteSql = "delete from " + PART_COL_STATS + " where \"PART_ID\" in ( " + getIdListForIn(partitionIds) + ")";
+          List<Object> params = new ArrayList<>(colNames == null ? 1 : colNames.size() + 1);
+
           if (colNames != null && !colNames.isEmpty()) {
-            deleteSql += " and \"COLUMN_NAME\" in (" + colNames.stream().map(col -> "'" + col + "'").collect(Collectors.joining(",")) + ")";
+            deleteSql += " and \"COLUMN_NAME\" in (" + makeParams(colNames.size()) + ")";
+            params.addAll(colNames);
           }
+
           if (engine != null) {
-            deleteSql += " and \"ENGINE\" = '" + engine + "'";
+            deleteSql += " and \"ENGINE\" = ?";
+            params.add(engine);
           }
-          try {
-            executeNoResult(deleteSql);
-          } catch (SQLException e) {
-            LOG.warn("Error removing partition column stats. ", e);
-            throw new MetaException("Error removing partition column stats: " + e.getMessage());
+
+          try (QueryWrapper queryParams = new QueryWrapper(pm.newQuery("javax.jdo.query.SQL", deleteSql))) {
+            executeWithArray(queryParams.getInnerQuery(), params.toArray(), deleteSql);
           }
         }
         return null;
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..ae96fdaf67b6 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
@@ -919,14 +919,14 @@ public void testTableStatisticsOps() throws Exception {
     List<ColumnStatistics> tabColStats;
     try (AutoCloseable c = deadline()) {
       tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
-          Arrays.asList("test_col1", "test_col2"));
+          Arrays.asList("test_col1", "test_col' 2"));
     }
     Assert.assertEquals(0, tabColStats.size());
 
     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",
+    ColumnStatisticsObj statsObj2 = new ColumnStatisticsObj("test_col' 2", "int",
         new ColumnStatisticsData(ColumnStatisticsData._Fields.DECIMAL_STATS, new DecimalColumnStatsData(200, 2000)));
     ColumnStatistics colStats = new ColumnStatistics(statsDesc, Arrays.asList(statsObj1, statsObj2));
     colStats.setEngine(ENGINE);
@@ -934,7 +934,7 @@ public void testTableStatisticsOps() throws Exception {
 
     try (AutoCloseable c = deadline()) {
       tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
-          Arrays.asList("test_col1", "test_col2"));
+          Arrays.asList("test_col1", "test_col' 2"));
     }
     Assert.assertEquals(1, tabColStats.size());
     Assert.assertEquals(2, tabColStats.get(0).getStatsObjSize());
@@ -942,19 +942,25 @@ public void testTableStatisticsOps() throws Exception {
     objectStore.deleteTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1, "test_col1", ENGINE);
     try (AutoCloseable c = deadline()) {
       tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
-          Arrays.asList("test_col1", "test_col2"));
+          Arrays.asList("test_col1", "test_col' 2"));
     }
     Assert.assertEquals(1, tabColStats.size());
     Assert.assertEquals(1, tabColStats.get(0).getStatsObjSize());
 
-    objectStore.deleteTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1, "test_col2", ENGINE);
+    objectStore.deleteTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1, "test_col' 2", ENGINE);
     try (AutoCloseable c = deadline()) {
       tabColStats = objectStore.getTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
-          Arrays.asList("test_col1", "test_col2"));
+          Arrays.asList("test_col1", "test_col' 2"));
     }
     Assert.assertEquals(0, tabColStats.size());
   }
 
+  @Test
+  public void testDeleteTableColumnStatisticsWhenEngineHasSpecialCharacter() throws Exception {
+    createPartitionedTable(true, true);
+    objectStore.deleteTableColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1, "test_col1", "special '");
+  }
+
   @Test
   public void testPartitionStatisticsOps() throws Exception {
     createPartitionedTable(true, true);
@@ -1006,6 +1012,14 @@ public void testPartitionStatisticsOps() throws Exception {
     Assert.assertEquals(0, stat.size());
   }
 
+  @Test
+  public void testDeletePartitionColumnStatisticsWhenEngineHasSpecialCharacter() throws Exception {
+    createPartitionedTable(true, true);
+    objectStore.deletePartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+            "test_part_col=a2", List.of("a2"), null, "special '");
+  }
+
+
   @Test
   public void testAggrStatsUseDB() throws Exception {
     Configuration conf2 = MetastoreConf.newMetastoreConf(conf);
@@ -1051,7 +1065,7 @@ private void createPartitionedTable(boolean withPrivileges, boolean withStatisti
             .setDbName(DB1)
             .setTableName(TABLE1)
             .addCol("test_col1", "int")
-            .addCol("test_col2", "int")
+            .addCol("test_col' 2", "int")
             .addPartCol("test_part_col", "int")
             .addCol("test_bucket_col", "int", "test bucket col comment")
             .addCol("test_skewed_col", "int", "test skewed col comment")
@@ -1239,6 +1253,12 @@ protected Database getJdoResult(ObjectStore.GetHelper<Database> ctx) throws Meta
     Assert.assertEquals(1, directSqlErrors.getCount());
   }
 
+  @Test(expected = MetaException.class)
+  public void testLockDbTableThrowsExceptionWhenTableIsNotAllowedToLock() throws Exception {
+    MetaStoreDirectSql metaStoreDirectSql = new MetaStoreDirectSql(objectStore.getPersistenceManager(), conf, null);
+    metaStoreDirectSql.lockDbTable("TBLS");
+  }
+
   @Deprecated
   private static void dropAllStoreObjects(RawStore store)
       throws MetaException, InvalidObjectException, InvalidInputException {