HIVE-28677: Implement direct sql for delete table/partition column stats

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java

@@ -3241,6 +3241,46 @@ public void deleteColumnStatsState(long tbl_id) throws MetaException {
     }
   }
 
+  public boolean deleteTableColumnStatistics(long tableId, String colName, String engine) {
+    String deleteSql = "delete from " + TAB_COL_STATS + " where \"TBL_ID\" = " + tableId;
+    if (colName != null) {
+      deleteSql += " and \"COLUMN_NAME\" = '" + colName + "'";
+    }
+    if (engine != null) {
+      deleteSql += " and \"ENGINE\" = '" + engine + "'";
+    }
+    try {
+      executeNoResult(deleteSql);
+    } catch (SQLException e) {
+      LOG.warn("Error removing table column stats. ", e);
+      return false;
+    }
+    return true;
+  }
+
+  public boolean deletePartitionColumnStats(String catName, String dbName, String tblName,
+      String partName, String colName, String engine) throws MetaException {
+    String sqlFilter = "" + PARTITIONS + ".\"PART_NAME\" = ? ";
+    List<Long> partitionIds = getPartitionIdsViaSqlFilter(catName, dbName, tblName, sqlFilter,
+        Arrays.asList(partName), Collections.emptyList(), -1);
+    assert(partitionIds.size() == 1);
+
+    String deleteSql = "delete from " + PART_COL_STATS + " where \"PART_ID\" = " + partitionIds.get(0);
+    if (colName != null) {
+      deleteSql += " and \"COLUMN_NAME\" = '" + colName + "'";
+    }
+    if (engine != null) {
+      deleteSql += " and \"ENGINE\" = '" + engine + "'";
+    }
+    try {
+      executeNoResult(deleteSql);
+    } catch (SQLException e) {
+      LOG.warn("Error removing partition column stats. ", e);
+      return false;
+    }
+    return true;
+  }
+
   public Map<String, Map<String, String>> updatePartitionColumnStatisticsBatch(
                                                       Map<String, ColumnStatistics> partColStatsMap,
                                                       Table tbl,

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java

@@ -10234,30 +10234,47 @@ protected Integer getJdoResult(GetHelper<Integer> ctx) throws MetaException, NoS
   public boolean deletePartitionColumnStatistics(String catName, String dbName, String tableName,
       String partName, List<String> partVals, String colName, String engine)
       throws NoSuchObjectException, MetaException, InvalidObjectException, InvalidInputException {
-    boolean ret = false;
-    Query query = null;
     dbName = org.apache.commons.lang3.StringUtils.defaultString(dbName,
       Warehouse.DEFAULT_DATABASE_NAME);
     catName = normalizeIdentifier(catName);
     if (tableName == null) {
       throw new InvalidInputException("Table name is null.");
     }
+    // Note: this does not verify ACID state; called internally when removing cols/etc.
+    //       Also called via an unused metastore API that checks for ACID tables.
+    MPartition mPartition = getMPartition(catName, dbName, tableName, partName);
+    if (mPartition == null) {
+      throw new NoSuchObjectException("Partition " + partName
+          + " for which stats deletion is requested doesn't exist");
+    }
+
+    return new GetHelper<Boolean>(catName, dbName, tableName, true, true) {
+      @Override
+      protected String describeResult() {
+        return "delete prtition column stats";
+      }
+
+      @Override
+      protected Boolean getSqlResult(GetHelper<Boolean> ctx) throws MetaException {
+        return directSql.deletePartitionColumnStats(catName, dbName, tableName, partName, colName, engine);
+      }
+
+      @Override
+      protected Boolean getJdoResult(GetHelper<Boolean> ctx)
+          throws MetaException, NoSuchObjectException, InvalidObjectException {
+        return deletePartitionColumnStatisticsVisJDO(catName, dbName, tableName, partName, colName, engine);
+      }
+    }.run(false);
+  }
+
+  private boolean deletePartitionColumnStatisticsVisJDO(String catName, String dbName, String tableName,
+      String partName, String colName, String engine) throws NoSuchObjectException {
+    boolean ret = false;
+    Query query = null;
     try {
       openTransaction();
-      MTable mTable = getMTable(catName, dbName, tableName);
       MPartitionColumnStatistics mStatsObj;
       List<MPartitionColumnStatistics> mStatsObjColl;
-      if (mTable == null) {
-        throw new NoSuchObjectException("Table " + tableName
-            + "  for which stats deletion is requested doesn't exist");
-      }
-      // Note: this does not verify ACID state; called internally when removing cols/etc.
-      //       Also called via an unused metastore API that checks for ACID tables.
-      MPartition mPartition = getMPartition(catName, dbName, tableName, partVals, mTable);
-      if (mPartition == null) {
-        throw new NoSuchObjectException("Partition " + partName
-            + " for which stats deletion is requested doesn't exist");
-      }
       query = pm.newQuery(MPartitionColumnStatistics.class);
       String filter;
       String parameters;
@@ -10334,25 +10351,42 @@ public boolean deletePartitionColumnStatistics(String catName, String dbName, St
   public boolean deleteTableColumnStatistics(String catName, String dbName, String tableName,
       String colName, String engine)
       throws NoSuchObjectException, MetaException, InvalidObjectException, InvalidInputException {
-    boolean ret = false;
-    Query query = null;
     dbName = org.apache.commons.lang3.StringUtils.defaultString(dbName,
       Warehouse.DEFAULT_DATABASE_NAME);
     if (tableName == null) {
       throw new InvalidInputException("Table name is null.");
     }
+
+    // Note: this does not verify ACID state; called internally when removing cols/etc.
+    //       Also called via an unused metastore API that checks for ACID tables.
+    return new GetHelper<Boolean>(catName, dbName, tableName, true, true) {
+
+      @Override
+      protected String describeResult() {
+        return "delete table column stats";
+      }
+
+      @Override
+      protected Boolean getSqlResult(GetHelper<Boolean> ctx) throws MetaException {
+        return directSql.deleteTableColumnStatistics(getTable().getId(), colName, engine);
+      }
+
+      @Override
+      protected Boolean getJdoResult(GetHelper<Boolean> ctx)
+          throws MetaException, NoSuchObjectException, InvalidObjectException {
+            return deleteTableColumnStatisticsViaJdo(catName, dbName, tableName, colName, engine);
+      }
+    }.run(true);
+  }
+
+  private boolean deleteTableColumnStatisticsViaJdo(String catName, String dbName, String tableName,
+      String colName, String engine) throws NoSuchObjectException {
+    boolean ret = false;
+    Query query = null;
     try {
       openTransaction();
-      MTable mTable = getMTable(catName, dbName, tableName);
       MTableColumnStatistics mStatsObj;
       List<MTableColumnStatistics> mStatsObjColl;
-      if (mTable == null) {
-        throw new NoSuchObjectException("Table " +
-            TableName.getQualified(catName, dbName, tableName)
-            + "  for which stats deletion is requested doesn't exist");
-      }
-      // Note: this does not verify ACID state; called internally when removing cols/etc.
-      //       Also called via an unused metastore API that checks for ACID tables.
       query = pm.newQuery(MTableColumnStatistics.class);
       String filter;
       String parameters;

standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestObjectStore.java

@@ -893,7 +893,7 @@ public void testTableStatisticsOps() throws Exception {
   }
 
   @Test
-  public void testGetPartitionStatistics() throws Exception {
+  public void testPartitionStatisticsOps() throws Exception {
     createPartitionedTable(true, true);
 
     List<List<ColumnStatistics>> stat;
@@ -912,6 +912,35 @@ public void testGetPartitionStatistics() throws Exception {
     ColumnStatisticsData expectedStats = new ColStatsBuilder<>(long.class).numNulls(1).numDVs(2)
         .low(3L).high(4L).hll(3, 4).kll(3, 4).build();
     assertEqualStatistics(expectedStats, computedStats);
+
+    objectStore.deletePartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+        "test_part_col=a0", Arrays.asList("a0"), null, ENGINE);
+    try (AutoCloseable c = deadline()) {
+      stat = 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, stat.size());
+    Assert.assertEquals(2, stat.get(0).size());
+
+    objectStore.deletePartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+        "test_part_col=a1", Arrays.asList("a1"), "test_part_col", null);
+    try (AutoCloseable c = deadline()) {
+      stat = 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, stat.size());
+    Assert.assertEquals(1, stat.get(0).size());
+
+    objectStore.deletePartitionColumnStatistics(DEFAULT_CATALOG_NAME, DB1, TABLE1,
+        "test_part_col=a2", Arrays.asList("a2"), null, null);
+    try (AutoCloseable c = deadline()) {
+      stat = 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(0, stat.size());
   }
 
   /**