@Override
  @SuppressWarnings("unchecked")
  public int run() throws IOException {
    Preconditions.checkArgument(
        input != null && output != null, "Both input and output parquet file paths are required.");

    Preconditions.checkArgument(codec != null, "The codec cannot be null");

    Path inPath = new Path(input);
    Path outPath = new Path(output);
    CompressionCodecName codecName = Codecs.parquetCodec(codec);

    ParquetMetadata metaData = ParquetFileReader.readFooter(getConf(), inPath, NO_FILTER);
    MessageType schema = metaData.getFileMetaData().getSchema();
    ParquetFileWriter writer = new ParquetFileWriter(getConf(), schema, outPath, ParquetFileWriter.Mode.CREATE);
    writer.start();

    try (TransParquetFileReader reader = new TransParquetFileReader(
        HadoopInputFile.fromPath(inPath, getConf()),
        HadoopReadOptions.builder(getConf()).build())) {
      compressionConverter.processBlocks(
          reader, writer, metaData, schema, metaData.getFileMetaData().getCreatedBy(), codecName);
    } finally {
      writer.end(metaData.getFileMetaData().getKeyValueMetaData());
    }
    return 0;
  }
@Test
  public void testTransCompressionCommand_ZSTD() throws IOException {
    TransCompressionCommand command = new TransCompressionCommand(createLogger());

    command.input = parquetFile().getAbsolutePath();

    File output = new File(getTempFolder(), getClass().getSimpleName() + ".converted-1.ZSTD.parquet");
    command.output = output.getAbsolutePath();
    command.codec = "ZSTD";
    command.setConf(new Configuration());

    Assert.assertEquals(0, command.run());
    Assert.assertTrue(output.exists());
  }