public void createPipe(CreatePipeStmt stmt) throws DdlException {
        try {
            lock.writeLock().lock();
            Pair<Long, String> dbIdAndName = resolvePipeNameUnlock(stmt.getPipeName());
            boolean existed = nameToId.containsKey(dbIdAndName);
            if (existed) {
                if (!stmt.isIfNotExists() && !stmt.isReplace()) {
                    ErrorReport.reportSemanticException(ErrorCode.ERR_PIPE_EXISTS);
                }
                if (stmt.isIfNotExists()) {
                    return;
                } else if (stmt.isReplace()) {
                    LOG.info("Pipe {} already exist, replace it with a new one", stmt.getPipeName());
                    Pipe pipe = pipeMap.get(nameToId.get(dbIdAndName));
                    dropPipeImpl(pipe);
                }
            }

            // Add pipe
            long id = GlobalStateMgr.getCurrentState().getNextId();
            Pipe pipe = Pipe.fromStatement(id, stmt);
            putPipe(pipe);

            repo.addPipe(pipe);
        } finally {
            lock.writeLock().unlock();
        }

    }
@Test
    public void testListPipeFiles() throws Exception {
        ExecuteEnv env = Mockito.mock(ExecuteEnv.class);
        FrontendServiceImpl impl = new FrontendServiceImpl(env);

        PipeManager pm = GlobalStateMgr.getCurrentState().getPipeManager();
        String createSql =
                "create pipe list_p2 as insert into tbl1 select * from files('path'='fake://pipe', 'format'='parquet')";
        CreatePipeStmt createStmt = (CreatePipeStmt) UtFrameUtils.parseStmtWithNewParser(createSql, ctx);
        pm.createPipe(createStmt);

        List<PipeFileRecord> records = Arrays.asList(
                new PipeFileRecord(1, "file1", "version1", 1024),
                new PipeFileRecord(1, "file2", "version1", 1024),
                new PipeFileRecord(1, "file3", "version1", 1024)
        );
        new Expectations(RepoAccessor.getInstance()) {
            {
                RepoAccessor.getInstance().listAllFiles();
                result = records;
            }
        };

        TListPipeFilesParams params = new TListPipeFilesParams();

        // without identify
        Assert.assertThrows(TException.class, () -> impl.listPipeFiles(params));

        // normal
        TUserIdentity identity = new TUserIdentity();
        identity.setUsername("root");
        params.setUser_ident(identity);
        TListPipeFilesResult result = impl.listPipeFiles(params);
        Assert.assertFalse(result.pipe_files.isEmpty());
    }