Skip to content

Commit

Permalink
[Enhancement] Support move truncated old data to recycle bin
Browse files Browse the repository at this point in the history
  • Loading branch information
Vallishp committed Nov 23, 2024
1 parent 362efda commit fb2b3aa
Show file tree
Hide file tree
Showing 16 changed files with 557 additions and 85 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -864,7 +864,7 @@ unsupportedUseStatement
;

unsupportedDmlStatement
: TRUNCATE TABLE multipartIdentifier specifiedPartition? #truncateTable
: TRUNCATE TABLE multipartIdentifier specifiedPartition? FORCE? #truncateTable
| COPY INTO name=multipartIdentifier columns=identifierList? FROM
(stageAndPattern | (LEFT_PAREN SELECT selectColumnClause
FROM stageAndPattern whereClause? RIGHT_PAREN))
Expand Down
4 changes: 2 additions & 2 deletions fe/fe-core/src/main/cup/sql_parser.cup
Original file line number Diff line number Diff line change
Expand Up @@ -7976,9 +7976,9 @@ admin_stmt ::=
;

truncate_stmt ::=
KW_TRUNCATE KW_TABLE base_table_ref:tblRef
KW_TRUNCATE KW_TABLE base_table_ref:tblRef opt_force:force
{:
RESULT = new TruncateTableStmt(tblRef);
RESULT = new TruncateTableStmt(tblRef, force);
:}
;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,21 @@
public class TruncateTableStmt extends DdlStmt implements NotFallbackInParser {

private TableRef tblRef;
private boolean forceDrop;

public TruncateTableStmt(TableRef tblRef) {
public TruncateTableStmt(TableRef tblRef, boolean forceDrop) {
this.tblRef = tblRef;
this.forceDrop = forceDrop;
}

public TableRef getTblRef() {
return tblRef;
}

public boolean isForceDrop() {
return forceDrop;
}

@Override
public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
super.analyze(analyzer);
Expand Down Expand Up @@ -75,6 +81,9 @@ public String toSql() {
if (tblRef.getPartitionNames() != null) {
sb.append(tblRef.getPartitionNames().toSql());
}
if (isForceDrop()) {
sb.append(" FORCE");
}
return sb.toString();
}

Expand All @@ -83,6 +92,9 @@ public String toSqlWithoutTable() {
if (tblRef.getPartitionNames() != null) {
sb.append(tblRef.getPartitionNames().toSql());
}
if (isForceDrop()) {
sb.append(" FORCE");
}
return sb.toString();
}

Expand Down
154 changes: 96 additions & 58 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
Original file line number Diff line number Diff line change
Expand Up @@ -1127,63 +1127,9 @@ private Partition dropPartition(long dbId, String partitionName, boolean isForce
if (partition != null) {
idToPartition.remove(partition.getId());
nameToPartition.remove(partitionName);

if (!isForceDrop) {
// recycle partition
if (partitionInfo.getType() == PartitionType.RANGE) {
Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
partitionInfo.getItem(partition.getId()).getItems(),
new ListPartitionItem(Lists.newArrayList(new PartitionKey())),
partitionInfo.getDataProperty(partition.getId()),
partitionInfo.getReplicaAllocation(partition.getId()),
partitionInfo.getIsInMemory(partition.getId()),
partitionInfo.getIsMutable(partition.getId()));

} else if (partitionInfo.getType() == PartitionType.LIST) {
// construct a dummy range
List<Column> dummyColumns = new ArrayList<>();
dummyColumns.add(new Column("dummy", PrimitiveType.INT));
PartitionKey dummyKey = null;
try {
dummyKey = PartitionKey.createInfinityPartitionKey(dummyColumns, false);
} catch (AnalysisException e) {
LOG.warn("should not happen", e);
}
Range<PartitionKey> dummyRange = Range.open(new PartitionKey(), dummyKey);

Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
dummyRange,
partitionInfo.getItem(partition.getId()),
partitionInfo.getDataProperty(partition.getId()),
partitionInfo.getReplicaAllocation(partition.getId()),
partitionInfo.getIsInMemory(partition.getId()),
partitionInfo.getIsMutable(partition.getId()));
} else {
// unpartition
// construct a dummy range and dummy list.
List<Column> dummyColumns = new ArrayList<>();
dummyColumns.add(new Column("dummy", PrimitiveType.INT));
PartitionKey dummyKey = null;
try {
dummyKey = PartitionKey.createInfinityPartitionKey(dummyColumns, false);
} catch (AnalysisException e) {
LOG.warn("should not happen", e);
}
Range<PartitionKey> dummyRange = Range.open(new PartitionKey(), dummyKey);
Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
dummyRange,
new ListPartitionItem(Lists.newArrayList(new PartitionKey())),
partitionInfo.getDataProperty(partition.getId()),
partitionInfo.getReplicaAllocation(partition.getId()),
partitionInfo.getIsInMemory(partition.getId()),
partitionInfo.getIsMutable(partition.getId()));
}
} else if (!reserveTablets) {
Env.getCurrentEnv().onErasePartition(partition);
}

// drop partition info
partitionInfo.dropPartition(partition.getId());
RecyclePartitionParam recyclePartitionParam = new RecyclePartitionParam();
fillInfo(partition, recyclePartitionParam);
dropPartitionCommon(dbId, isForceDrop, recyclePartitionParam, partition, reserveTablets);
}
return partition;
}
Expand All @@ -1196,6 +1142,81 @@ public Partition dropPartition(long dbId, String partitionName, boolean isForceD
return dropPartition(dbId, partitionName, isForceDrop, !isForceDrop);
}

private void dropPartitionCommon(long dbId, boolean isForceDrop,
RecyclePartitionParam recyclePartitionParam,
Partition partition,
boolean reserveTablets) {
if (!isForceDrop) {
// recycle partition
if (partitionInfo.getType() == PartitionType.RANGE) {
Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
recyclePartitionParam.partitionItem.getItems(),
new ListPartitionItem(Lists.newArrayList(new PartitionKey())),
recyclePartitionParam.dataProperty,
recyclePartitionParam.replicaAlloc,
recyclePartitionParam.isInMemory,
recyclePartitionParam.isMutable);

} else if (partitionInfo.getType() == PartitionType.LIST) {
// construct a dummy range
List<Column> dummyColumns = new ArrayList<>();
dummyColumns.add(new Column("dummy", PrimitiveType.INT));
PartitionKey dummyKey = null;
try {
dummyKey = PartitionKey.createInfinityPartitionKey(dummyColumns, false);
} catch (AnalysisException e) {
LOG.warn("should not happen", e);
}
Range<PartitionKey> dummyRange = Range.open(new PartitionKey(), dummyKey);

Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
dummyRange,
recyclePartitionParam.partitionItem,
recyclePartitionParam.dataProperty,
recyclePartitionParam.replicaAlloc,
recyclePartitionParam.isInMemory,
recyclePartitionParam.isMutable);
} else {
// unpartition
// construct a dummy range and dummy list.
List<Column> dummyColumns = new ArrayList<>();
dummyColumns.add(new Column("dummy", PrimitiveType.INT));
PartitionKey dummyKey = null;
try {
dummyKey = PartitionKey.createInfinityPartitionKey(dummyColumns, false);
} catch (AnalysisException e) {
LOG.warn("should not happen", e);
}
Range<PartitionKey> dummyRange = Range.open(new PartitionKey(), dummyKey);
Env.getCurrentRecycleBin().recyclePartition(dbId, id, name, partition,
dummyRange,
new ListPartitionItem(Lists.newArrayList(new PartitionKey())),
recyclePartitionParam.dataProperty,
recyclePartitionParam.replicaAlloc,
recyclePartitionParam.isInMemory,
recyclePartitionParam.isMutable);
}
} else if (!reserveTablets) {
Env.getCurrentEnv().onErasePartition(partition);
}

// drop partition info
partitionInfo.dropPartition(partition.getId());
}

public Partition dropPartitionForTruncate(long dbId, boolean isForceDrop,
RecyclePartitionParam recyclePartitionParam) {
// 1. If "isForceDrop" is false, the partition will be added to the Catalog Recyle bin, and all tablets of this
// partition will not be deleted.
// 2. If "ifForceDrop" is true, the partition will be dropped immediately
Partition partition = recyclePartitionParam.partition;
if (partition != null) {
idToPartition.remove(partition.getId());
dropPartitionCommon(dbId, isForceDrop, recyclePartitionParam, partition, false);
}
return partition;
}

/*
* A table may contain both formal and temporary partitions.
* There are several methods to get the partition of a table.
Expand Down Expand Up @@ -2027,13 +2048,24 @@ public static OlapTable read(DataInput in) throws IOException {
return GsonUtils.GSON.fromJson(Text.readString(in), OlapTable.class);
}


public void fillInfo(Partition partition, RecyclePartitionParam recyclePartitionParam) {
recyclePartitionParam.dataProperty = partitionInfo.getDataProperty(partition.getId());
recyclePartitionParam.replicaAlloc = partitionInfo.getReplicaAllocation(partition.getId());
recyclePartitionParam.isInMemory = partitionInfo.getIsInMemory(partition.getId());
recyclePartitionParam.isMutable = partitionInfo.getIsMutable(partition.getId());
recyclePartitionParam.partitionItem = partitionInfo.getItem(partition.getId());
recyclePartitionParam.partition = partition;
}

/*
* this method is currently used for truncating table(partitions).
* the new partition has new id, so we need to change all 'id-related' members
*
* return the old partition.
*/
public Partition replacePartition(Partition newPartition) {
public Partition replacePartition(Partition newPartition,
RecyclePartitionParam recyclePartitionParam) {
Partition oldPartition = nameToPartition.remove(newPartition.getName());
idToPartition.remove(oldPartition.getId());

Expand All @@ -2044,6 +2076,12 @@ public Partition replacePartition(Partition newPartition) {
ReplicaAllocation replicaAlloc = partitionInfo.getReplicaAllocation(oldPartition.getId());
boolean isInMemory = partitionInfo.getIsInMemory(oldPartition.getId());
boolean isMutable = partitionInfo.getIsMutable(oldPartition.getId());
recyclePartitionParam.dataProperty = dataProperty;
recyclePartitionParam.replicaAlloc = replicaAlloc;
recyclePartitionParam.isInMemory = isInMemory;
recyclePartitionParam.isMutable = isMutable;
recyclePartitionParam.partitionItem = partitionInfo.getItem(oldPartition.getId());
recyclePartitionParam.partition = oldPartition;

if (partitionInfo.getType() == PartitionType.RANGE
|| partitionInfo.getType() == PartitionType.LIST) {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
// 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.catalog;

public class RecyclePartitionParam {
public Partition partition;
public PartitionItem partitionItem;
public DataProperty dataProperty;
public ReplicaAllocation replicaAlloc;
public boolean isInMemory;
public boolean isMutable = true;

public RecyclePartitionParam() {
// do nothing.
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,7 @@
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.RandomDistributionInfo;
import org.apache.doris.catalog.RangePartitionItem;
import org.apache.doris.catalog.RecyclePartitionParam;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Replica.ReplicaState;
import org.apache.doris.catalog.ReplicaAllocation;
Expand Down Expand Up @@ -3720,13 +3721,17 @@ public void truncateTable(TruncateTableStmt truncateTableStmt) throws DdlExcepti
throw new DdlException("Table[" + copiedTbl.getName() + "]'s meta has been changed. try again.");
}

// replace
oldPartitions = truncateTableInternal(olapTable, newPartitions, truncateEntireTable);
boolean isForceDrop = truncateTableStmt.isForceDrop();
//replace
Map<Long, RecyclePartitionParam> recyclePartitionParamMap = new HashMap<>();
oldPartitions = truncateTableInternal(olapTable, newPartitions,
truncateEntireTable, recyclePartitionParamMap, isForceDrop);

// write edit log
TruncateTableInfo info =
new TruncateTableInfo(db.getId(), db.getFullName(), olapTable.getId(), olapTable.getName(),
newPartitions, truncateEntireTable, truncateTableStmt.toSqlWithoutTable(), oldPartitions);
newPartitions, truncateEntireTable,
truncateTableStmt.toSqlWithoutTable(), oldPartitions, isForceDrop);
Env.getCurrentEnv().getEditLog().logTruncateTable(info);
} catch (DdlException e) {
failedCleanCallback.run();
Expand All @@ -3737,8 +3742,6 @@ public void truncateTable(TruncateTableStmt truncateTableStmt) throws DdlExcepti
}
}

erasePartitionDropBackendReplicas(oldPartitions);

PartitionNames partitionNames = truncateEntireTable ? null
: new PartitionNames(false, tblRef.getPartitionNames().getPartitionNames());
Env.getCurrentEnv().getAnalysisManager().dropStats(olapTable, partitionNames);
Expand All @@ -3747,47 +3750,49 @@ public void truncateTable(TruncateTableStmt truncateTableStmt) throws DdlExcepti
}

private List<Partition> truncateTableInternal(OlapTable olapTable, List<Partition> newPartitions,
boolean isEntireTable) {
boolean isEntireTable, Map<Long, RecyclePartitionParam> recyclePartitionParamMap, boolean isforceDrop) {
// use new partitions to replace the old ones.
List<Partition> oldPartitions = Lists.newArrayList();
Set<Long> oldTabletIds = Sets.newHashSet();
for (Partition newPartition : newPartitions) {
Partition oldPartition = olapTable.replacePartition(newPartition);
RecyclePartitionParam recyclePartitionParam = new RecyclePartitionParam();
Partition oldPartition = olapTable.replacePartition(newPartition, recyclePartitionParam);
oldPartitions.add(oldPartition);
// save old tablets to be removed
for (MaterializedIndex index : oldPartition.getMaterializedIndices(IndexExtState.ALL)) {
index.getTablets().forEach(t -> {
oldTabletIds.add(t.getId());
});
}
recyclePartitionParamMap.put(oldPartition.getId(), recyclePartitionParam);
}

if (isEntireTable) {
Set<Long> oldPartitionsIds = oldPartitions.stream().map(Partition::getId).collect(Collectors.toSet());
for (Partition partition : olapTable.getAllTempPartitions()) {
if (!oldPartitionsIds.contains(partition.getId())) {
RecyclePartitionParam recyclePartitionParam = new RecyclePartitionParam();
olapTable.fillInfo(partition, recyclePartitionParam);
oldPartitions.add(partition);
recyclePartitionParamMap.put(partition.getId(), recyclePartitionParam);
// clear temp partition from memory.
// tablet may be moved to recycle bin or deleted inside
// dropPartitionForTruncate function.
olapTable.dropTempPartition(partition.getName(), false);
}
}
// drop all temp partitions
olapTable.dropAllTempPartitions();
}

// remove the tablets in old partitions
for (Long tabletId : oldTabletIds) {
Env.getCurrentInvertedIndex().deleteTablet(tabletId);
for (Map.Entry<Long, RecyclePartitionParam> pair : recyclePartitionParamMap.entrySet()) {
olapTable.dropPartitionForTruncate(olapTable.getDatabase().getId(), isforceDrop, pair.getValue());
}

return oldPartitions;
}

public void replayTruncateTable(TruncateTableInfo info) throws MetaNotFoundException {
boolean isForceDrop = info.getForce();
List<Partition> oldPartitions = Lists.newArrayList();
Database db = (Database) getDbOrMetaException(info.getDbId());
OlapTable olapTable = (OlapTable) db.getTableOrMetaException(info.getTblId(), TableType.OLAP);
olapTable.writeLock();
try {
truncateTableInternal(olapTable, info.getPartitions(), info.isEntireTable());
Map<Long, RecyclePartitionParam> recyclePartitionParamMap = new HashMap<>();
truncateTableInternal(olapTable, info.getPartitions(), info.isEntireTable(),
recyclePartitionParamMap, isForceDrop);

// add tablet to inverted index
TabletInvertedIndex invertedIndex = Env.getCurrentInvertedIndex();
Expand Down
Loading

0 comments on commit fb2b3aa

Please sign in to comment.