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 executeAutoIngest() throws Exception {
        mockRepoExecutor();
        mockTaskExecution(Constants.TaskRunState.SUCCESS);
        // auto_ingest=false
        String pipeP3 = "p3";
        String p3Sql = "create pipe p3 properties('auto_ingest'='false') as " +
                "insert into tbl1 select * from files('path'='fake://pipe', 'format'='parquet')";
        createPipe(p3Sql);
        Pipe pipe = getPipe(pipeP3);
        Assert.assertEquals(Pipe.State.RUNNING, pipe.getState());
        pipe.poll();
        pipe.schedule();
        pipe.schedule();
        pipe.poll();
        // schedule task
        pipe.schedule();
        // finalize task
        pipe.schedule();
        // trigger eos
        pipe.schedule();
        Assert.assertTrue(pipe.getPipeSource().eos());
        Assert.assertEquals(Pipe.State.FINISHED, pipe.getState());

        // auto_ingest=true
        String pipeP4 = "p4";
        String p4Sql = "create pipe p4 properties('auto_ingest'='true') as " +
                "insert into tbl1 select * from files('path'='fake://pipe', 'format'='parquet')";
        createPipe(p4Sql);
        pipe = getPipe(pipeP4);
        Assert.assertEquals(Pipe.State.RUNNING, pipe.getState());
        pipe.poll();
        pipe.schedule();
        pipe.poll();
        pipe.schedule();
        Assert.assertFalse(pipe.getPipeSource().eos());
        Assert.assertEquals(Pipe.State.RUNNING, pipe.getState());
    }