public List<TransactionStateBatch> getReadyToPublishTxnListBatch() {
        List<TransactionStateBatch> result = new ArrayList<>();
        readLock();

        try {
            List<Long> txnIds = transactionGraph.getTxnsWithoutDependency();
            for (long txnId : txnIds) {
                List<Long> txnsWithDependency = transactionGraph.getTxnsWithTxnDependencyBatch(
                        Config.lake_batch_publish_min_version_num,
                        Config.lake_batch_publish_max_version_num, txnId);
                List<TransactionState> states = txnsWithDependency.stream().map(idToRunningTransactionState::get)
                        .filter(Objects::nonNull)
                        .collect(Collectors.toList());
                if (states.isEmpty()) {
                    continue;
                }
                if (states.size() == 1) { // fast path: no batch
                    result.add(new TransactionStateBatch(states));
                    continue;
                }

                // Only single table transactions will be batched together.
                Preconditions.checkState(states.get(0).getTableIdList().size() == 1);

                long tableId = states.get(0).getTableIdList().get(0);

                // check whether version is consequent
                // for schema change will occupy a version
                Map<Long, PartitionCommitInfo> versions = new HashMap<>();

                outerLoop:
                for (int i = 0; i < states.size(); i++) {
                    TransactionState state = states.get(i);
                    TableCommitInfo tableInfo = state.getTableCommitInfo(tableId);
                    // TableCommitInfo could be null if the table has been dropped before this transaction is committed.
                    if (tableInfo == null) {
                        states = states.subList(0, Math.max(i, 1));
                        break;
                    }
                    Map<Long, PartitionCommitInfo> partitionInfoMap = tableInfo.getIdToPartitionCommitInfo();
                    for (Map.Entry<Long, PartitionCommitInfo> item : partitionInfoMap.entrySet()) {
                        PartitionCommitInfo currTxnInfo = item.getValue();
                        PartitionCommitInfo prevTxnInfo = versions.get(item.getKey());
                        if (prevTxnInfo != null && prevTxnInfo.getVersion() + 1 != currTxnInfo.getVersion()) {
                            assert i > 0;
                            // version is not consecutive
                            // may schema change occupy a version
                            states = states.subList(0, i);
                            break outerLoop;
                        }
                        versions.put(item.getKey(), currTxnInfo);
                    }
                }

                TransactionStateBatch batch = new TransactionStateBatch(states);
                result.add(batch);
            }
        } finally {
            readUnlock();
        }
        return result;
    }
@Test
    public void testGetReadyToPublishTxnListBatch() throws AnalysisException {
        DatabaseTransactionMgr masterDbTransMgr = masterTransMgr.getDatabaseTransactionMgr(GlobalStateMgrTestUtil.testDbId1);
        List<TransactionStateBatch> stateBatchesList = masterDbTransMgr.getReadyToPublishTxnListBatch();
        assertEquals(1, stateBatchesList.size());
        assertEquals(3, stateBatchesList.get(0).size());

        // Table was dropped before committing the txn 'testTxnLabel8'
        masterDbTransMgr.getLabelTransactionState(GlobalStateMgrTestUtil.testTxnLable8)
                .removeTable(GlobalStateMgrTestUtil.testTableId1);
        stateBatchesList = masterDbTransMgr.getReadyToPublishTxnListBatch();
        assertEquals(1, stateBatchesList.size());
        assertEquals(2, stateBatchesList.get(0).size());

        // Table was dropped before committing the txn 'testTxnLabel7'
        masterDbTransMgr.getLabelTransactionState(GlobalStateMgrTestUtil.testTxnLable7)
                .removeTable(GlobalStateMgrTestUtil.testTableId1);
        stateBatchesList = masterDbTransMgr.getReadyToPublishTxnListBatch();
        assertEquals(1, stateBatchesList.size());
        assertEquals(1, stateBatchesList.get(0).size());

        // Table was dropped before committing the txn 'testTxnLabel6'
        masterDbTransMgr.getLabelTransactionState(GlobalStateMgrTestUtil.testTxnLable6)
                .removeTable(GlobalStateMgrTestUtil.testTableId1);
        stateBatchesList = masterDbTransMgr.getReadyToPublishTxnListBatch();
        assertEquals(1, stateBatchesList.size());
        assertEquals(1, stateBatchesList.get(0).size());
    }