Skip to content

[feat](fe) Support Cloud partition inverted format rollout #66235

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hoshinojyunn wants to merge 1 commit into
base: master
Choose a base branch
from
+2,136 −81
Open

[feat](fe) Support Cloud partition inverted format rollout #66235

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
99 changes: 92 additions & 7 deletions fe/fe-core/src/main/java/org/apache/doris/alter/CloudSchemaChangeJobV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,21 +34,27 @@
import org.apache.doris.cloud.qe.ComputeGroupException;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.proto.OlapFile;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.service.FrontendOptions;
import org.apache.doris.task.AgentTask;
import org.apache.doris.task.AgentTaskQueue;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.thrift.TTaskType;

import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Table;
import com.google.common.collect.Table.Cell;
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.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
Expand All @@ -58,6 +64,9 @@
public class CloudSchemaChangeJobV2 extends SchemaChangeJobV2 {
private static final Logger LOG = LogManager.getLogger(SchemaChangeJobV2.class);

@SerializedName("fsv")
private Map<TInvertedIndexFileStorageFormat, Integer> baseShadowSchemaVersionByFormat = new HashMap<>();

public CloudSchemaChangeJobV2(String rawSql, long jobId, long dbId, long tableId,
String tableName, long timeoutMs) {
super(rawSql, jobId, dbId, tableId, tableName, timeoutMs);
Expand All @@ -79,6 +88,28 @@ public CloudSchemaChangeJobV2(String rawSql, long jobId, long dbId, long tableId

private CloudSchemaChangeJobV2() {}

public void configureBaseShadowSchemaVersion(OlapTable table) throws DdlException {
long baseShadowIndexId = getBaseShadowIndexId(table.getBaseIndexId());
if (baseShadowIndexId == -1) {
return;
}
int tableBaseSchemaVersion = table.getIndexMetaByIndexId(table.getBaseIndexId()).getSchemaVersion();
baseShadowSchemaVersionByFormat = getBaseShadowSchemaVersions(table, tableBaseSchemaVersion);
Integer currentSchemaVersion = baseShadowSchemaVersionByFormat.get(
table.getPartitionInvertedIndexFileStorageFormat());
Preconditions.checkState(currentSchemaVersion != null);
indexSchemaVersionAndHashMap.get(baseShadowIndexId).schemaVersion = currentSchemaVersion;
}

public Map<TInvertedIndexFileStorageFormat, Integer> getBaseShadowSchemaVersionByFormat() {
return baseShadowSchemaVersionByFormat;
}

public void setBaseShadowSchemaVersionByFormat(
Map<TInvertedIndexFileStorageFormat, Integer> baseShadowSchemaVersionByFormat) {
this.baseShadowSchemaVersionByFormat = baseShadowSchemaVersionByFormat;
}

@Override
protected void commitShadowIndex() throws AlterCancelException {
List<Long> shadowIdxList =
Expand Down Expand Up @@ -227,16 +258,18 @@ private void createShadowIndexReplicaForPartition(OlapTable tbl) throws Exceptio
for (Map.Entry<Long, MaterializedIndex> entry : shadowIndexMap.entrySet()) {
long shadowIdxId = entry.getKey();
MaterializedIndex shadowIdx = entry.getValue();

boolean isBaseShadowIndex = isShadowIndexOfBase(shadowIdxId, tbl);
short shadowShortKeyColumnCount = indexShortKeyMap.get(shadowIdxId);
List<Column> shadowSchema = indexSchemaMap.get(shadowIdxId);
List<Integer> clusterKeyUids = null;
if (shadowIdxId == tbl.getBaseIndexId() || isShadowIndexOfBase(shadowIdxId, tbl)) {
clusterKeyUids = OlapTable.getClusterKeyUids(shadowSchema);
}
int shadowSchemaHash = indexSchemaVersionAndHashMap.get(shadowIdxId).schemaHash;
int shadowSchemaVersion = indexSchemaVersionAndHashMap.get(shadowIdxId).schemaVersion;
long originIndexId = indexIdMap.get(shadowIdxId);
TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat = isBaseShadowIndex
? tbl.getInvertedIndexFileStorageFormatForPartition(partitionId) : null;
int schemaVersion = isBaseShadowIndex
? getBaseShadowSchemaVersion(invertedIndexFileStorageFormat) : shadowSchemaVersion;
List<Integer> clusterKeyUids = isBaseShadowIndex
? OlapTable.getClusterKeyUids(shadowSchema) : null;
KeysType originKeysType = tbl.getKeysTypeByIndexId(originIndexId);
List<Index> tabletIndexes = originIndexId == tbl.getBaseIndexId() ? indexes : null;

Expand All @@ -255,15 +288,15 @@ private void createShadowIndexReplicaForPartition(OlapTable tbl) throws Exceptio
tbl.getStoragePolicy(), tbl.isInMemory(), true,
tbl.getName(), tbl.getTTLSeconds(),
tbl.getEnableUniqueKeyMergeOnWrite(), tbl.storeRowColumn(),
shadowSchemaVersion, tbl.getCompactionPolicy(),
schemaVersion, tbl.getCompactionPolicy(),
tbl.getTimeSeriesCompactionGoalSizeMbytes(),
tbl.getTimeSeriesCompactionFileCountThreshold(),
tbl.getTimeSeriesCompactionTimeThresholdSeconds(),
tbl.getTimeSeriesCompactionEmptyRowsetsThreshold(),
tbl.getTimeSeriesCompactionLevelThreshold(),
tbl.disableAutoCompaction(),
tbl.getRowStoreColumnsUniqueIds(rowStoreColumns),
tbl.getInvertedIndexFileStorageFormat(),
invertedIndexFileStorageFormat,
tbl.rowStorePageSize(),
tbl.variantEnableFlattenNested(), clusterKeyUids,
tbl.storagePageSize(), tbl.getTDEAlgorithmPB(),
Expand All @@ -279,6 +312,58 @@ private void createShadowIndexReplicaForPartition(OlapTable tbl) throws Exceptio
}
}

private long getBaseShadowIndexId(long baseIndexId) {
for (Map.Entry<Long, Long> entry : indexIdMap.entrySet()) {
if (entry.getValue() == baseIndexId) {
return entry.getKey();
}
}
return -1;
}

private int getBaseShadowSchemaVersion(TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat) {
Preconditions.checkState(invertedIndexFileStorageFormat != null,
"missing inverted index storage format for base partition");
Integer schemaVersion = baseShadowSchemaVersionByFormat.get(invertedIndexFileStorageFormat);
Preconditions.checkState(schemaVersion != null,
"missing shadow schema version for format " + invertedIndexFileStorageFormat);
return schemaVersion;
}

private Map<TInvertedIndexFileStorageFormat, Integer> getBaseShadowSchemaVersions(OlapTable table,
int tableBaseSchemaVersion) throws DdlException {
TInvertedIndexFileStorageFormat currentFormat = table.getPartitionInvertedIndexFileStorageFormat();
Set<TInvertedIndexFileStorageFormat> formats = new HashSet<>();
for (long partitionId : partitionIndexMap.rowKeySet()) {
TInvertedIndexFileStorageFormat format = table.getInvertedIndexFileStorageFormatForPartition(partitionId);
if (format == null) {
throw new DdlException("missing inverted index storage format for partition " + partitionId);
}
formats.add(format);
}
formats.add(currentFormat);

List<TInvertedIndexFileStorageFormat> orderedFormats = new ArrayList<>(formats);
orderedFormats.sort((left, right) -> {
if (left == right) {
return 0;
}
if (left == currentFormat) {
return 1;
}
if (right == currentFormat) {
return -1;
}
return left.name().compareTo(right.name());
});

Map<TInvertedIndexFileStorageFormat, Integer> shadowSchemaVersions = new HashMap<>();
for (int i = 0; i < orderedFormats.size(); ++i) {
shadowSchemaVersions.put(orderedFormats.get(i), tableBaseSchemaVersion + i + 1);
}
return shadowSchemaVersions;
}

@Override
protected void ensureCloudClusterExist(List<AgentTask> tasks) throws AlterCancelException {
if (((CloudSystemInfoService) Env.getCurrentSystemInfo())
Expand Down
36 changes: 30 additions & 6 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/CatalogRecycleBin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.doris.persist.RecoverInfo;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.GlobalVariable;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.thrift.TStorageMedium;

import com.google.common.base.Preconditions;
Expand Down Expand Up @@ -234,9 +235,16 @@ public boolean recycleTable(long dbId, Table table, boolean isReplay,
}

public boolean recyclePartition(long dbId, long tableId, String tableName, Partition partition,
Range<PartitionKey> range, PartitionItem listPartitionItem,
DataProperty dataProperty, ReplicaAllocation replicaAlloc,
boolean isInMemory, boolean isMutable) {
Range<PartitionKey> range, PartitionItem listPartitionItem, DataProperty dataProperty,
ReplicaAllocation replicaAlloc, boolean isInMemory, boolean isMutable) {
return recyclePartition(dbId, tableId, tableName, partition, range, listPartitionItem, dataProperty,
replicaAlloc, isInMemory, isMutable, null);
}

public boolean recyclePartition(long dbId, long tableId, String tableName, Partition partition,
Range<PartitionKey> range, PartitionItem listPartitionItem, DataProperty dataProperty,
ReplicaAllocation replicaAlloc, boolean isInMemory, boolean isMutable,
TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat) {
writeLock();
try {
if (idToPartition.containsKey(partition.getId())) {
Expand All @@ -246,7 +254,8 @@ public boolean recyclePartition(long dbId, long tableId, String tableName, Parti

// recycle partition
RecyclePartitionInfo partitionInfo = new RecyclePartitionInfo(dbId, tableId, partition,
range, listPartitionItem, dataProperty, replicaAlloc, isInMemory, isMutable);
range, listPartitionItem, dataProperty, replicaAlloc, isInMemory, isMutable,
invertedIndexFileStorageFormat);
idToRecycleTime.put(partition.getId(), System.currentTimeMillis());
idToPartition.put(partition.getId(), partitionInfo);
dbTblIdPartitionNameToIds.computeIfAbsent(Pair.of(dbId, tableId), k -> new ConcurrentHashMap<>())
Expand Down Expand Up @@ -1051,7 +1060,10 @@ public void recoverPartition(long dbId, OlapTable table, String partitionName,
partitionInfo.setReplicaAllocation(partitionId, recoverPartitionInfo.getReplicaAlloc());
partitionInfo.setIsInMemory(partitionId, recoverPartitionInfo.isInMemory());
partitionInfo.setIsMutable(partitionId, recoverPartitionInfo.isMutable());

if (recoverPartitionInfo.getInvertedIndexFileStorageFormat() != null) {
partitionInfo.setInvertedIndexFileStorageFormat(partitionId,
recoverPartitionInfo.getInvertedIndexFileStorageFormat());
}
// remove from recycle bin
idToPartition.remove(partitionId);
idToRecycleTime.remove(partitionId);
Expand Down Expand Up @@ -1117,6 +1129,10 @@ public void replayRecoverPartition(OlapTable table, long partitionId,
partitionInfo.setReplicaAllocation(partitionId, recyclePartitionInfo.getReplicaAlloc());
partitionInfo.setIsInMemory(partitionId, recyclePartitionInfo.isInMemory());
partitionInfo.setIsMutable(partitionId, recyclePartitionInfo.isMutable());
if (recyclePartitionInfo.getInvertedIndexFileStorageFormat() != null) {
partitionInfo.setInvertedIndexFileStorageFormat(partitionId,
recyclePartitionInfo.getInvertedIndexFileStorageFormat());
}

iterator.remove();
idToRecycleTime.remove(partitionId);
Expand Down Expand Up @@ -1767,6 +1783,8 @@ public class RecyclePartitionInfo {
private boolean isInMemory;
@SerializedName("mu")
private boolean isMutable = true;
@SerializedName("iifsf")
private TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat;

public RecyclePartitionInfo() {
// for persist
Expand All @@ -1775,7 +1793,8 @@ public RecyclePartitionInfo() {
public RecyclePartitionInfo(long dbId, long tableId, Partition partition,
Range<PartitionKey> range, PartitionItem listPartitionItem,
DataProperty dataProperty, ReplicaAllocation replicaAlloc,
boolean isInMemory, boolean isMutable) {
boolean isInMemory, boolean isMutable,
TInvertedIndexFileStorageFormat invertedIndexFileStorageFormat) {
this.dbId = dbId;
this.tableId = tableId;
this.partition = partition;
Expand All @@ -1785,6 +1804,7 @@ public RecyclePartitionInfo(long dbId, long tableId, Partition partition,
this.replicaAlloc = replicaAlloc;
this.isInMemory = isInMemory;
this.isMutable = isMutable;
this.invertedIndexFileStorageFormat = invertedIndexFileStorageFormat;
}

public long getDbId() {
Expand Down Expand Up @@ -1823,6 +1843,10 @@ public boolean isMutable() {
return isMutable;
}

public TInvertedIndexFileStorageFormat getInvertedIndexFileStorageFormat() {
return invertedIndexFileStorageFormat;
}

public void write(DataOutput out) throws IOException {
Text.writeString(out, GsonUtils.GSON.toJson(this));
}
Expand Down
52 changes: 51 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,7 @@
import org.apache.doris.thrift.TGetMetaResult;
import org.apache.doris.thrift.TGetMetaTableMeta;
import org.apache.doris.thrift.TGetMetaTabletMeta;
import org.apache.doris.thrift.TInvertedIndexFileStorageFormat;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TStatus;
import org.apache.doris.thrift.TStatusCode;
Expand Down Expand Up @@ -3812,6 +3813,12 @@ public void addPartitionLike(Database db, String tableName, AddPartitionLikeOp a
getInternalCatalog().addPartitionLike(db, tableName, addPartitionLikeOp);
}

public void addPartitionLikeForInsertOverwrite(Database db, String tableName,
AddPartitionLikeOp addPartitionLikeOp)
throws DdlException {
getInternalCatalog().addPartitionLikeForInsertOverwrite(db, tableName, addPartitionLikeOp);
}

public void replayAddPartition(PartitionPersistInfo info) throws MetaNotFoundException {
getInternalCatalog().replayAddPartition(info);
}
Expand Down Expand Up @@ -4021,6 +4028,13 @@ private static void addOlapTablePropertyInfo(OlapTable olapTable, StringBuilder
// inverted index storage type
sb.append(",\n\"").append(PropertyAnalyzer.PROPERTIES_INVERTED_INDEX_STORAGE_FORMAT).append("\" = \"");
sb.append(olapTable.getInvertedIndexFileStorageFormat()).append("\"");
if (olapTable.getTableProperty() != null
&& olapTable.getTableProperty().getPartitionInvertedIndexFileStorageFormat() != null) {
sb.append(",\n\"").append(PropertyAnalyzer.PROPERTIES_PARTITION_INVERTED_INDEX_STORAGE_FORMAT)
.append("\" = \"")
.append(olapTable.getTableProperty().getPartitionInvertedIndexFileStorageFormat())
.append("\"");
}

// compression type
if (olapTable.getCompressionType() != TCompressionType.valueOf(Config.default_compression_type)) {
Expand Down Expand Up @@ -6346,6 +6360,19 @@ public void modifyTableDefaultReplicaAllocation(Database db, OlapTable table,

// The caller need to hold the table write lock
public void modifyTableProperties(Database db, OlapTable table, Map<String, String> properties) {
modifyTableProperties(db, table, properties, new HashMap<>());
}

// The caller need to hold the table write lock
public void modifyTableProperties(Database db, OlapTable table, Map<String, String> properties,
Map<Long, Integer> indexIdToSchemaVersion) {
modifyTableProperties(db, table, properties, indexIdToSchemaVersion, new HashMap<>());
}

// The caller need to hold the table write lock
public void modifyTableProperties(Database db, OlapTable table, Map<String, String> properties,
Map<Long, Integer> indexIdToSchemaVersion,
Map<Long, TInvertedIndexFileStorageFormat> partitionIdToInvertedIndexFileStorageFormat) {
Preconditions.checkArgument(table.isWriteLockHeldByCurrentThread());
TableProperty tableProperty = table.getTableProperty();
if (tableProperty == null) {
Expand Down Expand Up @@ -6377,9 +6404,17 @@ public void modifyTableProperties(Database db, OlapTable table, Map<String, Stri
table.getPartitionInfo().setStoragePolicy(partition.getId(), tableProperty.getStoragePolicy());
}

for (Map.Entry<Long, Integer> entry : indexIdToSchemaVersion.entrySet()) {
table.getIndexMetaByIndexId(entry.getKey()).setSchemaVersion(entry.getValue());
}
for (Map.Entry<Long, TInvertedIndexFileStorageFormat> entry
: partitionIdToInvertedIndexFileStorageFormat.entrySet()) {
table.getPartitionInfo().setInvertedIndexFileStorageFormat(entry.getKey(), entry.getValue());
}

ModifyTablePropertyOperationLog info =
new ModifyTablePropertyOperationLog(db.getId(), table.getId(), table.getName(),
properties);
properties, indexIdToSchemaVersion, partitionIdToInvertedIndexFileStorageFormat);
editLog.logModifyTableProperties(info);
}

Expand Down Expand Up @@ -6418,6 +6453,21 @@ public void replayModifyTableProperty(short opCode, ModifyTablePropertyOperation
tableProperty.buildProperty(opCode);
}

Map<Long, Integer> indexIdToSchemaVersion = info.getIndexIdToSchemaVersion();
if (indexIdToSchemaVersion != null) {
for (Map.Entry<Long, Integer> entry : indexIdToSchemaVersion.entrySet()) {
olapTable.getIndexMetaByIndexId(entry.getKey()).setSchemaVersion(entry.getValue());
}
}
Map<Long, TInvertedIndexFileStorageFormat> partitionIdToInvertedIndexFileStorageFormat =
info.getPartitionIdToInvertedIndexFileStorageFormat();
if (partitionIdToInvertedIndexFileStorageFormat != null) {
for (Map.Entry<Long, TInvertedIndexFileStorageFormat> entry
: partitionIdToInvertedIndexFileStorageFormat.entrySet()) {
olapTable.getPartitionInfo().setInvertedIndexFileStorageFormat(entry.getKey(), entry.getValue());
}
}

// need to replay partition info meta
switch (opCode) {
case OperationType.OP_MODIFY_TABLE_PROPERTIES:
Expand Down
Loading
Loading