Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line numberDiff line numberDiff line change
Expand Up@@ -37,13 +37,14 @@
import org.apache.doris.transaction.TransactionStatus;

import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.LinkedHashMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Ordering;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
import com.google.common.collect.Table;
import com.google.common.collect.TreeMultimap;
Expand DownExpand Up@@ -135,7 +136,7 @@ public void tabletReport(long backendId, Map<Long, TTablet> backendTablets,
Set<Long> tabletFoundInMeta,
ListMultimap<TStorageMedium, Long> tabletMigrationMap,
Map<Long, Long> partitionVersionSyncMap,
Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish,
Map<Long, SetMultimap<Long, TPartitionVersionInfo>> transactionsToPublish,
ListMultimap<Long, Long> transactionsToClear,
ListMultimap<Long, Long> tabletRecoveryMap,
List<TTabletMetaInfo> tabletToUpdate,
Expand DownExpand Up@@ -314,7 +315,7 @@ && isLocal(tabletMeta.getStorageMedium())) {
}

private void handleBackendTransactions(long backendId, List<Long> transactionIds, long tabletId,
TabletMeta tabletMeta, Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish,
TabletMeta tabletMeta, Map<Long, SetMultimap<Long, TPartitionVersionInfo>> transactionsToPublish,
ListMultimap<Long, Long> transactionsToClear) {
GlobalTransactionMgrIface transactionMgr = Env.getCurrentGlobalTransactionMgr();
long partitionId = tabletMeta.getPartitionId();
Expand DownExpand Up@@ -376,15 +377,15 @@ private TPartitionVersionInfo generatePartitionVersionInfoWhenReport(Transaction
}

private void publishPartition(TransactionState transactionState, long transactionId, TabletMeta tabletMeta,
long partitionId, Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish) {
long partitionId, Map<Long, SetMultimap<Long, TPartitionVersionInfo>> transactionsToPublish) {
TPartitionVersionInfo versionInfo = generatePartitionVersionInfoWhenReport(transactionState,
transactionId, tabletMeta, partitionId);
if (versionInfo != null) {
synchronized (transactionsToPublish) {
ListMultimap<Long, TPartitionVersionInfo> map = transactionsToPublish.get(
SetMultimap<Long, TPartitionVersionInfo> map = transactionsToPublish.get(
transactionState.getDbId());
if (map == null) {
map = ArrayListMultimap.create();
map = LinkedHashMultimap.create();
transactionsToPublish.put(transactionState.getDbId(), map);
}
map.put(transactionId, versionInfo);
Expand Down
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.SetMultimap;
import com.google.common.collect.Sets;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
Expand DownExpand Up@@ -503,7 +504,7 @@ public void tabletReport(long backendId, Map<Long, TTablet> backendTablets,
Map<Long, Long> partitionVersionSyncMap = Maps.newConcurrentMap();

// dbid -> txn id -> [partition info]
Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish = Maps.newHashMap();
Map<Long, SetMultimap<Long, TPartitionVersionInfo>> transactionsToPublish = Maps.newHashMap();
ListMultimap<Long, Long> transactionsToClear = LinkedListMultimap.create();

// db id -> tablet id
Expand DownExpand Up@@ -1147,14 +1148,14 @@ private static void handleMigration(ListMultimap<TStorageMedium, Long> tabletMet
}

private static void handleRepublishVersionInfo(
Map<Long, ListMultimap<Long, TPartitionVersionInfo>> transactionsToPublish, long backendId) {
Map<Long, SetMultimap<Long, TPartitionVersionInfo>> transactionsToPublish, long backendId) {
AgentBatchTask batchTask = new AgentBatchTask();
long createPublishVersionTaskTime = System.currentTimeMillis();
for (Long dbId : transactionsToPublish.keySet()) {
ListMultimap<Long, TPartitionVersionInfo> map = transactionsToPublish.get(dbId);
SetMultimap<Long, TPartitionVersionInfo> map = transactionsToPublish.get(dbId);
for (long txnId : map.keySet()) {
PublishVersionTask task = new PublishVersionTask(backendId, txnId, dbId,
map.get(txnId), createPublishVersionTaskTime);
Lists.newArrayList(map.get(txnId)), createPublishVersionTaskTime);
batchTask.addTask(task);
// add to AgentTaskQueue for handling finish report.
AgentTaskQueue.addTask(task);
Expand Down