TExecPlanFragmentParams streamLoadPutImpl(TStreamLoadPutRequest request) throws UserException, LockTimeoutException {
        String cluster = request.getCluster();
        if (Strings.isNullOrEmpty(cluster)) {
            cluster = SystemInfoService.DEFAULT_CLUSTER;
        }

        GlobalStateMgr globalStateMgr = GlobalStateMgr.getCurrentState();
        String dbName = request.getDb();
        Database db = globalStateMgr.getDb(dbName);
        if (db == null) {
            throw new UserException("unknown database, database=" + dbName);
        }

        Table table = db.getTable(request.getTbl());
        if (table == null) {
            throw new UserException("unknown table, table=" + request.getTbl());
        }
        if (!(table instanceof OlapTable)) {
            throw new UserException("load table type is not OlapTable, type=" + table.getClass());
        }
        if (table instanceof MaterializedView) {
            throw new UserException(String.format(
                    "The data of '%s' cannot be inserted because '%s' is a materialized view," +
                            "and the data of materialized view must be consistent with the base table.",
                    table.getName(), table.getName()));
        }

        long timeoutMs = request.isSetThrift_rpc_timeout_ms() ? request.getThrift_rpc_timeout_ms() : 5000;
        // Make timeout less than thrift_rpc_timeout_ms.
        // Otherwise, it will result in error like "call frontend service failed"
        timeoutMs = timeoutMs * 3 / 4;

        Locker locker = new Locker();
        if (!locker.tryLockTablesWithIntensiveDbLock(db, Lists.newArrayList(table.getId()), LockType.READ,
                timeoutMs, TimeUnit.MILLISECONDS)) {
            throw new LockTimeoutException(
                    "get database read lock timeout, database=" + dbName + ", timeout=" + timeoutMs + "ms");
        }
        try {
            StreamLoadInfo streamLoadInfo = StreamLoadInfo.fromTStreamLoadPutRequest(request, db);
            StreamLoadPlanner planner = new StreamLoadPlanner(db, (OlapTable) table, streamLoadInfo);
            TExecPlanFragmentParams plan = planner.plan(streamLoadInfo.getId());

            if (plan.query_options.enable_profile) {
                StreamLoadTask streamLoadTask = GlobalStateMgr.getCurrentState().getStreamLoadMgr().
                        getSyncSteamLoadTaskByTxnId(request.getTxnId());
                if (streamLoadTask == null) {
                    throw new UserException("can not find stream load task by txnId " + request.getTxnId());
                }

                streamLoadTask.setTUniqueId(request.getLoadId());

                Coordinator coord = getCoordinatorFactory().createSyncStreamLoadScheduler(planner, getClientAddr());
                streamLoadTask.setCoordinator(coord);

                QeProcessorImpl.INSTANCE.registerQuery(streamLoadInfo.getId(), coord);
            }

            plan.query_options.setLoad_job_type(TLoadJobType.STREAM_LOAD);
            // add table indexes to transaction state
            TransactionState txnState =
                    GlobalStateMgr.getCurrentState().getGlobalTransactionMgr()
                            .getTransactionState(db.getId(), request.getTxnId());
            if (txnState == null) {
                throw new UserException("txn does not exist: " + request.getTxnId());
            }
            txnState.addTableIndexes((OlapTable) table);
            plan.setImport_label(txnState.getLabel());
            plan.setDb_name(dbName);
            plan.setLoad_job_id(request.getTxnId());

            return plan;
        } finally {
            locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(table.getId()), LockType.READ);
        }
    }
@Test
    public void testMetaNotFound() throws UserException {
        FrontendServiceImpl impl = spy(new FrontendServiceImpl(exeEnv));
        TStreamLoadPutRequest request = new TStreamLoadPutRequest();
        request.db = "test";
        request.tbl = "foo";
        request.txnId = 1001L;
        request.setFileType(TFileType.FILE_STREAM);
        request.setLoadId(new TUniqueId(1, 2));

        Exception e = Assert.assertThrows(UserException.class, () -> impl.streamLoadPutImpl(request));
        Assert.assertTrue(e.getMessage().contains("unknown table"));

        request.tbl = "v";
        e = Assert.assertThrows(UserException.class, () -> impl.streamLoadPutImpl(request));
        Assert.assertTrue(e.getMessage().contains("load table type is not OlapTable"));

        request.tbl = "mv";
        e = Assert.assertThrows(UserException.class, () -> impl.streamLoadPutImpl(request));
        Assert.assertTrue(e.getMessage().contains("is a materialized view"));
    }