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 showPipes() throws Exception {
        PipeManager pm = ctx.getGlobalStateMgr().getPipeManager();

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

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

        // show
        String sql = "show pipes";
        ShowPipeStmt showPipeStmt = (ShowPipeStmt) UtFrameUtils.parseStmtWithNewParser(sql, ctx);
        ShowResultSet result = ShowExecutor.execute(showPipeStmt, ctx);
        Assert.assertEquals(
                Arrays.asList("show_1", "RUNNING", "pipe_test_db.tbl1",
                        "{\"loadedFiles\":0,\"loadedBytes\":0,\"loadingFiles\":0}", null),
                result.getResultRows().get(0).subList(2, result.numColumns() - 1));
        Assert.assertEquals(
                Arrays.asList("show_2", "RUNNING", "pipe_test_db.tbl1",
                        "{\"loadedFiles\":0,\"loadedBytes\":0,\"loadingFiles\":0}", null),
                result.getResultRows().get(1).subList(2, result.numColumns() - 1));

        // desc
        sql = "desc pipe show_1";
        DescPipeStmt descPipeStmt = (DescPipeStmt) UtFrameUtils.parseStmtWithNewParser(sql, ctx);
        result = ShowExecutor.execute(descPipeStmt, ctx);
        Assert.assertEquals(
                Arrays.asList("show_1", "FILE", "pipe_test_db.tbl1", "FILE_SOURCE(path=fake://pipe)",
                        "insert into tbl1 select * from files('path'='fake://pipe', 'format'='parquet')", ""),
                result.getResultRows().get(0).subList(2, result.numColumns())
        );
    }