@Override
  public void dropPartition(
      ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists)
      throws PartitionNotExistException, CatalogException {
    checkNotNull(tablePath, "Table path cannot be null");
    checkNotNull(partitionSpec, "CatalogPartitionSpec cannot be null");

    final CatalogBaseTable table;
    try {
      table = getTable(tablePath);
    } catch (TableNotExistException e) {
      if (!ignoreIfNotExists) {
        throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e);
      } else {
        return;
      }
    }
    try (HoodieFlinkWriteClient<?> writeClient = HoodieCatalogUtil.createWriteClient(tablePath, table, hiveConf, this::inferTablePath)) {
      boolean hiveStylePartitioning = Boolean.parseBoolean(table.getOptions().get(FlinkOptions.HIVE_STYLE_PARTITIONING.key()));
      writeClient.deletePartitions(
              Collections.singletonList(HoodieCatalogUtil.inferPartitionPath(hiveStylePartitioning, partitionSpec)),
              writeClient.createNewInstantTime())
          .forEach(writeStatus -> {
            if (writeStatus.hasErrors()) {
              throw new HoodieMetadataException(String.format("Failed to commit metadata table records at file id %s.", writeStatus.getFileId()));
            }
          });

      client.dropPartition(
          tablePath.getDatabaseName(),
          tablePath.getObjectName(),
          HoodieCatalogUtil.getOrderedPartitionValues(
              getName(), getHiveConf(), partitionSpec, ((CatalogTable) table).getPartitionKeys(), tablePath),
          true);
    } catch (NoSuchObjectException e) {
      if (!ignoreIfNotExists) {
        throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e);
      }
    } catch (MetaException | PartitionSpecInvalidException e) {
      throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e);
    } catch (Exception e) {
      throw new CatalogException(
          String.format(
              "Failed to drop partition %s of table %s", partitionSpec, tablePath), e);
    }
  }
@Test
  public void testDropPartition() throws Exception {
    Map<String, String> options = new HashMap<>();
    options.put(FactoryUtil.CONNECTOR.key(), "hudi");
    CatalogTable table =
        new CatalogTableImpl(schema, partitions, options, "hudi table");
    hoodieCatalog.createTable(tablePath, table, false);

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

    Table hiveTable = hoodieCatalog.getHiveTable(tablePath);
    StorageDescriptor partitionSd = new StorageDescriptor(hiveTable.getSd());
    partitionSd.setLocation(new Path(partitionSd.getLocation(), HoodieCatalogUtil.inferPartitionPath(true, partitionSpec)).toString());
    hoodieCatalog.getClient().add_partition(new Partition(Collections.singletonList("20221020"),
        tablePath.getDatabaseName(), tablePath.getObjectName(), 0, 0, partitionSd, null));
    assertNotNull(getHivePartition(partitionSpec));

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

    String tablePathStr = hoodieCatalog.inferTablePath(tablePath, hoodieCatalog.getTable(tablePath));
    HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(hoodieCatalog.getHiveConf()), tablePathStr);
    HoodieInstant latestInstant = metaClient.getActiveTimeline().filterCompletedInstants().lastInstant().orElse(null);
    assertNotNull(latestInstant, "Delete partition commit should be completed");
    HoodieCommitMetadata commitMetadata = WriteProfiles.getCommitMetadata(tablePath.getObjectName(), new org.apache.flink.core.fs.Path(tablePathStr),
        latestInstant, metaClient.getActiveTimeline());
    assertThat(commitMetadata, instanceOf(HoodieReplaceCommitMetadata.class));
    HoodieReplaceCommitMetadata replaceCommitMetadata = (HoodieReplaceCommitMetadata) commitMetadata;
    assertThat(replaceCommitMetadata.getPartitionToReplaceFileIds().size(), is(1));
    assertThrows(NoSuchObjectException.class, () -> getHivePartition(partitionSpec));
  }