@Override
  public void dropPartition(ObjectPath tablePath, CatalogPartitionSpec catalogPartitionSpec, boolean ignoreIfNotExists)
      throws PartitionNotExistException, CatalogException {
    if (!tableExists(tablePath)) {
      if (ignoreIfNotExists) {
        return;
      } else {
        throw new PartitionNotExistException(getName(), tablePath, catalogPartitionSpec);
      }
    }

    String tablePathStr = inferTablePath(catalogPathStr, tablePath);
    Map<String, String> options = TableOptionProperties.loadFromProperties(tablePathStr, hadoopConf);
    boolean hiveStylePartitioning = Boolean.parseBoolean(options.getOrDefault(FlinkOptions.HIVE_STYLE_PARTITIONING.key(), "false"));
    String partitionPathStr = HoodieCatalogUtil.inferPartitionPath(hiveStylePartitioning, catalogPartitionSpec);

    if (!StreamerUtil.partitionExists(tablePathStr, partitionPathStr, hadoopConf)) {
      if (ignoreIfNotExists) {
        return;
      } else {
        throw new PartitionNotExistException(getName(), tablePath, catalogPartitionSpec);
      }
    }

    try (HoodieFlinkWriteClient<?> writeClient = HoodieCatalogUtil.createWriteClient(options, tablePathStr, tablePath, hadoopConf)) {
      writeClient.deletePartitions(Collections.singletonList(partitionPathStr),
              writeClient.createNewInstantTime())
          .forEach(writeStatus -> {
            if (writeStatus.hasErrors()) {
              throw new HoodieMetadataException(String.format("Failed to commit metadata table records at file id %s.", writeStatus.getFileId()));
            }
          });
      fs.delete(new Path(tablePathStr, partitionPathStr), true);
    } catch (Exception e) {
      throw new CatalogException(String.format("Dropping partition %s of table %s exception.", partitionPathStr, tablePath), e);
    }
  }
@Test
  public void testDropPartition() throws Exception {
    ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
    // create table
    catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);

    CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(new HashMap<String, String>() {
      {
        put("partition", "par1");
      }
    });
    // drop non-exist partition
    assertThrows(PartitionNotExistException.class,
        () -> catalog.dropPartition(tablePath, partitionSpec, false));

    String tablePathStr = catalog.inferTablePath(catalogPathStr, tablePath);
    Configuration flinkConf = TestConfigurations.getDefaultConf(tablePathStr);
    HoodieTableMetaClient metaClient = HoodieTestUtils
        .createMetaClient(
            new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(flinkConf)), tablePathStr);
    TestData.writeData(TestData.DATA_SET_INSERT, flinkConf);
    assertTrue(catalog.partitionExists(tablePath, partitionSpec));

    // drop partition 'par1'
    catalog.dropPartition(tablePath, partitionSpec, false);

    HoodieInstant latestInstant = metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().orElse(null);
    assertNotNull(latestInstant, "Delete partition commit should be completed");
    HoodieCommitMetadata commitMetadata = WriteProfiles.getCommitMetadata("tb1", new Path(tablePathStr), latestInstant, metaClient.getActiveTimeline());
    assertThat(commitMetadata, instanceOf(HoodieReplaceCommitMetadata.class));
    HoodieReplaceCommitMetadata replaceCommitMetadata = (HoodieReplaceCommitMetadata) commitMetadata;
    assertThat(replaceCommitMetadata.getPartitionToReplaceFileIds().size(), is(1));
    assertFalse(catalog.partitionExists(tablePath, partitionSpec));
  }