@Override
  public Catalog createCatalog(Context context) {
    final FactoryUtil.CatalogFactoryHelper helper =
        FactoryUtil.createCatalogFactoryHelper(this, context);
    helper.validateExcept(HadoopConfigurations.HADOOP_PREFIX);
    String mode = helper.getOptions().get(CatalogOptions.MODE);
    switch (mode.toLowerCase(Locale.ROOT)) {
      case "hms":
        return new HoodieHiveCatalog(
            context.getName(),
            (Configuration) helper.getOptions());
      case "dfs":
        return new HoodieCatalog(
            context.getName(),
            (Configuration) helper.getOptions());
      default:
        throw new HoodieCatalogException(String.format("Invalid catalog mode: %s, supported modes: [hms, dfs].", mode));
    }
  }
@Test
  void testCreateDFSCatalog() {
    final String catalogName = "mycatalog";

    Map<String, String> catalogOptions = new HashMap<>();
    catalogOptions.put(CATALOG_PATH.key(), tempFile.getAbsolutePath());
    catalogOptions.put(DEFAULT_DATABASE.key(), "test_db");
    HoodieCatalog expectedCatalog = new HoodieCatalog(catalogName, Configuration.fromMap(catalogOptions));

    final Map<String, String> options = new HashMap<>();
    options.put(CommonCatalogOptions.CATALOG_TYPE.key(), HoodieCatalogFactory.IDENTIFIER);
    options.put(CATALOG_PATH.key(), tempFile.getAbsolutePath());
    options.put(DEFAULT_DATABASE.key(), "test_db");
    options.put(CatalogOptions.MODE.key(), "dfs");

    final Catalog actualCatalog =
        FactoryUtil.createCatalog(
            catalogName, options, null, Thread.currentThread().getContextClassLoader());

    checkEquals(expectedCatalog, (AbstractCatalog) actualCatalog);
  }