CloudPartition.java
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.cloud.catalog;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.DistributionInfo;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Table;
import org.apache.doris.cloud.proto.Cloud;
import org.apache.doris.cloud.proto.Cloud.MetaServiceCode;
import org.apache.doris.cloud.rpc.VersionHelper;
import org.apache.doris.common.Config;
import org.apache.doris.common.Pair;
import org.apache.doris.common.profile.SummaryProfile;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.proto.OlapFile;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.qe.VariableMgr;
import org.apache.doris.rpc.RpcException;
import org.apache.doris.service.FrontendOptions;
import com.google.common.annotations.VisibleForTesting;
import com.google.gson.annotations.SerializedName;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
/**
* Internal representation of partition-related metadata.
*/
public class CloudPartition extends Partition {
private static final Logger LOG = LogManager.getLogger(CloudPartition.class);
// not Serialized
@SerializedName(value = "dbId")
private long dbId;
@SerializedName(value = "tableId")
private long tableId;
// This value is set when get the version from meta-service, 0 means version is not cached yet
private volatile long lastVersionCachedTimeMs = 0;
private ReentrantLock lock = new ReentrantLock(true);
public CloudPartition(long id, String name, MaterializedIndex baseIndex,
DistributionInfo distributionInfo, long dbId, long tableId) {
super(id, name, baseIndex, distributionInfo);
super.setVisibleVersion(-1); // cloud partition version is not resident in FE memory, -1 mean unknown
super.nextVersion = -1;
this.dbId = dbId;
this.tableId = tableId;
}
public CloudPartition() {
super();
}
public long getDbId() {
return this.dbId;
}
public void setDbId(long dbId) {
this.dbId = dbId;
}
public long getTableId() {
return this.tableId;
}
public void setTableId(long tableId) {
this.tableId = tableId;
}
protected void setVisibleVersion(long visibleVersion) {
if (LOG.isDebugEnabled()) {
LOG.debug("setVisibleVersion use CloudPartition {}", super.getName());
}
return;
}
public void setCachedVisibleVersion(long version, long versionUpdateTimeMs) {
// we only care the version should increase monotonically and ignore the readers
LOG.debug("setCachedVisibleVersion use CloudPartition {}, version: {}, old version: {}",
super.getId(), version, super.getVisibleVersion());
lock.lock();
if (version > super.getVisibleVersion()) {
super.setVisibleVersionAndTime(version, versionUpdateTimeMs);
}
lock.unlock();
// versionUpdateTimeMs is the version mtime in MS, which is unlikely equal to lastVersionCachedTimeMs in FE
lastVersionCachedTimeMs = System.currentTimeMillis();
}
@Override
public long getCachedVisibleVersion() {
return super.getVisibleVersion();
}
@VisibleForTesting
protected boolean isCachedVersionExpired() {
if (lastVersionCachedTimeMs == 0) {
return true;
}
ConnectContext ctx = ConnectContext.get();
long cacheExpirationMs = ctx == null ? VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs
: ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs;
if (cacheExpirationMs <= 0) { // always expired
return true;
}
return System.currentTimeMillis() - lastVersionCachedTimeMs > cacheExpirationMs;
}
@Override
public long getVisibleVersion() {
if (Env.isCheckpointThread() || Config.enable_check_compatibility_mode) {
return super.getVisibleVersion();
}
if (!isCachedVersionExpired()) {
return getCachedVisibleVersion();
}
return getVisibleVersionFromMs(false);
}
private long getVisibleVersionFromMs(boolean waitForPendingTxns) {
if (LOG.isDebugEnabled()) {
LOG.debug("getVisibleVersionFromMs use CloudPartition {}, waitForPendingTxns: {}",
super.getName(), waitForPendingTxns);
}
Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder()
.setRequestIp(FrontendOptions.getLocalHostAddressCached())
.setDbId(this.dbId)
.setTableId(this.tableId)
.setPartitionId(super.getId())
.setBatchMode(false)
.setWaitForPendingTxn(waitForPendingTxns)
.build();
try {
Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(request);
long version = -1;
long mTime = -1;
if (resp.getStatus().getCode() == MetaServiceCode.OK) {
version = resp.getVersion();
// Cache visible version, see hasData() for details.
mTime = resp.getVersionUpdateTimeMsList().size() == 1 ? resp.getVersionUpdateTimeMs(0) : 0;
} else {
assert resp.getStatus().getCode() == MetaServiceCode.VERSION_NOT_FOUND;
version = Partition.PARTITION_INIT_VERSION;
mTime = System.currentTimeMillis();
}
if (LOG.isDebugEnabled()) {
LOG.debug("get version from meta service, version: {}, partition: {}", version, super.getId());
}
setCachedVisibleVersion(version, mTime);
return version;
} catch (RpcException e) {
throw new RuntimeException("get version from meta service failed");
}
}
// Select the non-empty partitions and return the ids.
public static List<Long> selectNonEmptyPartitionIds(List<CloudPartition> partitions) {
List<Long> nonEmptyPartitionIds = partitions.stream()
.filter(CloudPartition::hasDataCached)
.map(CloudPartition::getId)
.collect(Collectors.toList());
if (nonEmptyPartitionIds.size() == partitions.size()) {
return nonEmptyPartitionIds;
}
List<CloudPartition> unknowns = partitions.stream()
.filter(p -> !p.hasDataCached())
.collect(Collectors.toList());
SummaryProfile profile = getSummaryProfile();
if (profile != null) {
profile.incGetPartitionVersionByHasDataCount();
}
try {
List<Long> versions = CloudPartition.getSnapshotVisibleVersion(unknowns);
int size = versions.size();
for (int i = 0; i < size; i++) {
if (versions.get(i) > Partition.PARTITION_INIT_VERSION) {
nonEmptyPartitionIds.add(unknowns.get(i).getId());
}
}
return nonEmptyPartitionIds;
} catch (RpcException e) {
throw new RuntimeException("get version from meta service failed");
}
}
// Get visible version from the specified partitions;
//
// Return the visible version in order of the specified partition ids
public static List<Long> getSnapshotVisibleVersionFromMs(
List<CloudPartition> partitions, boolean waitForPendingTxns) throws RpcException {
if (partitions.isEmpty()) {
return new ArrayList<>();
}
List<Long> dbIds = new ArrayList<>();
List<Long> tableIds = new ArrayList<>();
List<Long> partitionIds = new ArrayList<>();
List<Long> versionUpdateTimesMs = new ArrayList<>();
for (CloudPartition partition : partitions) {
dbIds.add(partition.getDbId());
tableIds.add(partition.getTableId());
partitionIds.add(partition.getId());
}
List<Long> versions = getSnapshotVisibleVersion(
dbIds, tableIds, partitionIds, versionUpdateTimesMs, waitForPendingTxns);
// Cache visible version, see hasData() for details.
int size = versions.size();
for (int i = 0; i < size; ++i) {
Long version = versions.get(i);
if (version > Partition.PARTITION_INIT_VERSION) {
// For compatibility, the existing partitions may not have mtime
long mTime = versions.size() == versionUpdateTimesMs.size() ? versionUpdateTimesMs.get(i) : 0;
partitions.get(i).setCachedVisibleVersion(versions.get(i), mTime);
} else { // No data has been written to this partition
partitions.get(i).setCachedVisibleVersion(Partition.PARTITION_INIT_VERSION, System.currentTimeMillis());
}
}
return versions;
}
private static List<OlapTable> getTables(List<CloudPartition> partitions) {
Map<Long, OlapTable> tableMap = new HashMap<>();
for (CloudPartition partition : partitions) {
if (tableMap.containsKey(partition.getTableId())) {
continue;
}
Database db = Env.getCurrentInternalCatalog().getDbNullable(partition.getDbId());
if (db == null) {
continue;
}
Table table = db.getTableNullable(partition.getTableId());
if (table == null) {
continue;
}
tableMap.put(partition.getTableId(), (OlapTable) table);
}
List<OlapTable> tables = tableMap.values().stream().collect(Collectors.toCollection(ArrayList::new));
Collections.sort(tables, Comparator.comparingLong(o -> o.getId()));
return tables;
}
/**
* Resolves a historical timestamp to committed partition versions via a single batch RPC.
* Sends one request after partition pruning; returns versions in the same order as
* {@code partitions}. A version of -1 means no data committed at or before {@code timestampMs}.
*/
/**
* Bundles the resolved versions with any rowset manifests returned in the same RPC
* for post-compaction time-travel reads.
*/
public static class VersionAtTimeResult {
public final List<Long> versions;
/** tablet_id → list of serialised RowsetMetaCloudPB bytes, one per compacted version. */
public final java.util.Map<Long, List<byte[]>> tabletManifests;
public VersionAtTimeResult(List<Long> versions,
java.util.Map<Long, List<byte[]>> tabletManifests) {
this.versions = versions;
this.tabletManifests = tabletManifests;
}
}
public static VersionAtTimeResult getVersionsAtTime(List<CloudPartition> partitions,
long timestampMs, int retentionDays,
java.util.function.Function<CloudPartition, List<Long>> tabletIdsProvider)
throws RpcException {
if (partitions.isEmpty()) {
return new VersionAtTimeResult(new ArrayList<>(), java.util.Collections.emptyMap());
}
Cloud.GetVersionAtTimeRequest.Builder req = Cloud.GetVersionAtTimeRequest.newBuilder()
.setCloudUniqueId(Config.cloud_unique_id)
.setRequestIp(FrontendOptions.getLocalHostAddressCached())
.setTimestampMs(timestampMs)
.setRetentionDays(retentionDays)
.setBatchMode(true);
for (CloudPartition p : partitions) {
req.addDbIds(p.getDbId());
req.addTableIds(p.getTableId());
req.addPartitionIds(p.getId());
// Include tablet_ids so the meta service can fetch manifests in the same FDB transaction.
List<Long> tabletIds = tabletIdsProvider != null ? tabletIdsProvider.apply(p)
: java.util.Collections.emptyList();
Cloud.TtPartitionTabletsInfo.Builder info =
Cloud.TtPartitionTabletsInfo.newBuilder();
if (tabletIds != null) {
tabletIds.forEach(info::addTabletIds);
}
req.addTabletInfos(info.build());
}
Cloud.GetVersionAtTimeResponse resp;
try {
resp = org.apache.doris.cloud.rpc.MetaServiceProxy.getInstance()
.getVersionAtTime(req.build());
} catch (RpcException e) {
throw new RpcException("getVersionsAtTime", "RPC failed: " + e.getMessage(), e);
}
if (resp.getStatus().getCode() != MetaServiceCode.OK) {
throw new RpcException("getVersionsAtTime",
"meta service error: " + resp.getStatus().getMsg());
}
int n = partitions.size();
if (resp.getVersionsCount() != n) {
throw new RpcException("getVersionsAtTime",
"wrong number of versions: expected " + n + ", got " + resp.getVersionsCount());
}
List<Long> result = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
result.add(resp.getVersions(i));
}
// Extract per-tablet rowset manifests; serialised RowsetMetaCloudPB bytes passed to BE.
java.util.Map<Long, List<byte[]>> manifests = new java.util.HashMap<>();
for (Cloud.TtTabletRowsetsPB entry : resp.getTabletRowsetsList()) {
List<byte[]> bytesList = new java.util.ArrayList<>();
for (OlapFile.RowsetMetaCloudPB meta : entry.getRowsetMetasList()) {
bytesList.add(meta.toByteArray());
}
// Keep even empty manifests: an empty list signals that this tablet was compacted
// at the query version but had zero rows; BE must use the time-travel read path.
manifests.put(entry.getTabletId(), bytesList);
}
return new VersionAtTimeResult(result, manifests);
}
// Overload without manifest support; retained for existing callers.
public static List<Long> getVersionsAtTime(List<CloudPartition> partitions,
long timestampMs, int retentionDays) throws RpcException {
return getVersionsAtTime(partitions, timestampMs, retentionDays, null).versions;
}
// Get visible version from the specified partitions;
//
// Return the visible version in order of the specified partition ids, -1 means version NOT FOUND.
public static List<Long> getSnapshotVisibleVersion(List<CloudPartition> partitions) throws RpcException {
if (partitions.isEmpty()) {
return Collections.emptyList();
}
long cloudPartitionVersionCacheTtlMs = ConnectContext.get() == null ? 0
: ConnectContext.get().getSessionVariable().cloudPartitionVersionCacheTtlMs;
if (cloudPartitionVersionCacheTtlMs <= 0) { // No cached versions will be used
return getSnapshotVisibleVersionFromMs(partitions, false);
}
// partitionId -> cachedVersion
List<Pair<Long, Long>> allVersions = new ArrayList<>(partitions.size());
List<CloudPartition> expiredPartitions = new ArrayList<>(partitions.size());
List<OlapTable> tables = getTables(partitions);
for (OlapTable table : tables) {
table.versionReadLock();
}
try {
for (CloudPartition partition : partitions) {
long ver = partition.getCachedVisibleVersion();
if (partition.isCachedVersionExpired()) {
expiredPartitions.add(partition);
ver = 0L; // 0 means to be get from meta-service
}
allVersions.add(Pair.of(partition.getId(), ver));
}
} finally {
for (int i = tables.size() - 1; i >= 0; i--) {
tables.get(i).versionReadUnlock();
}
}
if (LOG.isDebugEnabled()) {
LOG.debug("cloudPartitionVersionCacheTtlMs={}, numPartitions={}, numFilteredPartitions={}",
cloudPartitionVersionCacheTtlMs, partitions.size(), partitions.size() - expiredPartitions.size());
}
List<Long> versions = null;
if (!expiredPartitions.isEmpty()) { // Not all partition versions are from cache
versions = getSnapshotVisibleVersionFromMs(
expiredPartitions, /*waitForPendingTxns=*/false); // Get the rest versions from meta-service
}
int verMsIdx = 0;
for (Pair<Long, Long> v : allVersions) { // ATTN: keep the assigning order!!!
if (v.second == 0L && versions != null) {
v.second = versions.get(verMsIdx++);
}
}
if (!expiredPartitions.isEmpty()) { // Not all partition versions are from cache
assert verMsIdx == versions.size() : "size not match, idx=" + verMsIdx + " verSize=" + versions.size();
}
versions = allVersions.stream().map(i -> i.second).collect(Collectors.toList());
return versions;
}
// Get visible versions for the specified partitions.
//
// Return the visible version in order of the specified partition ids
private static List<Long> getSnapshotVisibleVersion(List<Long> dbIds, List<Long> tableIds, List<Long> partitionIds,
List<Long> versionUpdateTimesMs, boolean waitForPendingTxns)
throws RpcException {
assert dbIds.size() == partitionIds.size() :
"partition ids size: " + partitionIds.size() + " should equals to db ids size: " + dbIds.size();
assert tableIds.size() == partitionIds.size() :
"partition ids size: " + partitionIds.size() + " should equals to tablet ids size: " + tableIds.size();
Cloud.GetVersionRequest req = Cloud.GetVersionRequest.newBuilder()
.setRequestIp(FrontendOptions.getLocalHostAddressCached())
.setDbId(-1)
.setTableId(-1)
.setPartitionId(-1)
.setBatchMode(true)
.addAllDbIds(dbIds)
.addAllTableIds(tableIds)
.addAllPartitionIds(partitionIds)
.setWaitForPendingTxn(waitForPendingTxns)
.build();
if (LOG.isDebugEnabled()) {
LOG.debug("getVisibleVersion use CloudPartition {}", partitionIds.toString());
}
Cloud.GetVersionResponse resp = VersionHelper.getVersionFromMeta(req);
if (resp.getStatus().getCode() != MetaServiceCode.OK) {
throw new RpcException("get visible version", "unexpected status " + resp.getStatus());
}
List<Long> versions = resp.getVersionsList();
if (versions.size() != partitionIds.size()) {
throw new RpcException("get visible version",
"wrong number of versions, required " + partitionIds.size() + ", but got " + versions.size());
}
if (LOG.isDebugEnabled()) {
LOG.debug("get version from meta service, partitions: {}, versions: {}", partitionIds, versions);
}
if (versionUpdateTimesMs != null) {
versionUpdateTimesMs.addAll(resp.getVersionUpdateTimeMsList());
}
ArrayList<Long> news = new ArrayList<>();
for (Long v : versions) { // -1 means version NOT FOUND ==> no data has been written
news.add(v == -1 ? Partition.PARTITION_INIT_VERSION : v);
}
return news;
}
@Override
public long getCommittedVersion() {
// Cloud partition version is managed by meta-service, not resident in FE memory.
return -1;
}
@Override
public long getNextVersion() {
// use meta service visibleVersion
if (LOG.isDebugEnabled()) {
LOG.debug("getNextVersion use CloudPartition {}", super.getName());
}
return -1;
}
@Override
public void setNextVersion(long nextVersion) {
// use meta service visibleVersion
if (LOG.isDebugEnabled()) {
LOG.debug("setNextVersion use CloudPartition {} Version {}", super.getName(), nextVersion);
}
return;
}
@Override
public void updateVersionForRestore(long visibleVersion) {
if (LOG.isDebugEnabled()) {
LOG.debug("updateVersionForRestore use CloudPartition {} version for restore: visible: {}",
super.getName(), visibleVersion);
}
return;
}
@Override
public void updateVisibleVersion(long visibleVersion) {
// use meta service visibleVersion
if (LOG.isDebugEnabled()) {
LOG.debug("updateVisibleVersion use CloudPartition {} version for restore: visible: {}",
super.getName(), visibleVersion);
}
return;
}
// Determine whether data this partition has, according to the cached visible version.
public boolean hasDataCached() {
// In order to determine whether a partition is empty, a get_version RPC is issued to
// the meta service. The pruning process will be very slow when there are lots of empty
// partitions. This option disables the empty partition prune optimization to speed SQL
// analysis/plan phase.
if (isEmptyPartitionPruneDisabled()) {
return true;
}
// Every partition starts from version 1, version 1 has no data.
// So as long as version is greater than 1, it can be determined that there is data here.
return super.getVisibleVersion() > Partition.PARTITION_INIT_VERSION;
}
/**
* CloudPartition always has data
*/
@Override
public boolean hasData() {
// To avoid sending an RPC request, see the cached visible version here first.
if (hasDataCached()) {
return true;
}
SummaryProfile profile = getSummaryProfile();
if (profile != null) {
profile.incGetPartitionVersionByHasDataCount();
}
return getVisibleVersion() > Partition.PARTITION_INIT_VERSION;
}
private static boolean isEmptyPartitionPruneDisabled() {
ConnectContext ctx = ConnectContext.get();
if (ctx != null && (ctx.getSessionVariable().getDisableNereidsRules().get(RuleType.valueOf(
"PRUNE_EMPTY_PARTITION").type()) || ctx.getSessionVariable().getDisableEmptyPartitionPrune())) {
return true;
}
return false;
}
private static SummaryProfile getSummaryProfile() {
ConnectContext ctx = ConnectContext.get();
if (ctx != null) {
StmtExecutor executor = ctx.getExecutor();
if (executor != null) {
return executor.getSummaryProfile();
}
}
return null;
}
public boolean equals(Object obj) {
if (!super.equals(obj)) {
return false;
}
if (!(obj instanceof CloudPartition)) {
return false;
}
CloudPartition cloudPartition = (CloudPartition) obj;
return (dbId == cloudPartition.dbId) && (tableId == cloudPartition.tableId);
}
public String toString() {
StringBuilder buffer = new StringBuilder();
buffer.append(super.toString());
buffer.append("dbId: ").append(this.dbId).append("; ");
buffer.append("tableId: ").append(this.tableId).append("; ");
return buffer.toString();
}
}