database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[fix](compatibility) Version 1.2 upgraded to 2.0 compatible with miniload metadata (#22590)

Browse Source
This commit is contained in:
luozenglin
2023-08-04 16:52:51 +08:00
committed by GitHub
parent 9cf6b1b4cf
commit dc06c486e8
3 changed files with 107 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadJob.java
View File

@ -865,6 +865,8 @@ public abstract class LoadJob extends AbstractTxnStateChangeCallback implements
job = new SparkLoadJob();
} else if (type == EtlJobType.INSERT) {
job = new InsertLoadJob();
} else if (type == EtlJobType.MINI) {
job = new MiniLoadJob();
} else {
throw new IOException("Unknown load type: " + type.name());
}

26
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
View File

@ -51,7 +51,6 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.OriginStatement;
import org.apache.doris.thrift.TUniqueId;
import org.apache.doris.transaction.DatabaseTransactionMgr;
import org.apache.doris.transaction.TransactionState;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
@ -815,7 +814,8 @@ public class LoadManager implements Writable {
public void write(DataOutput out) throws IOException {
long currentTimeMs = System.currentTimeMillis();
List<LoadJob> loadJobs =
idToLoadJob.values().stream().filter(t -> !t.isExpired(currentTimeMs)).collect(Collectors.toList());
idToLoadJob.values().stream().filter(t -> !t.isExpired(currentTimeMs))
.filter(t -> !(t instanceof MiniLoadJob)).collect(Collectors.toList());
out.writeInt(loadJobs.size());
for (LoadJob loadJob : loadJobs) {
@ -836,25 +836,9 @@ public class LoadManager implements Writable {
}
if (loadJob.getJobType() == EtlJobType.MINI) {
// This is a bug fix. the mini load job should not with state LOADING.
if (loadJob.getState() == JobState.LOADING) {
LOG.warn("skip mini load job {} in db {} with LOADING state", loadJob.getId(), loadJob.getDbId());
continue;
}
if (loadJob.getState() == JobState.PENDING) {
// bad case. When a mini load job is created and then FE restart.
// the job will be in PENDING state forever.
// This is a temp solution to remove these jobs.
// And the mini load job should be deprecated in Doris v1.1
TransactionState state = Env.getCurrentEnv().getGlobalTransactionMgr()
.getTransactionState(loadJob.getDbId(), loadJob.getTransactionId());
if (state == null) {
LOG.warn("skip mini load job {} in db {} with PENDING state and with txn: {}", loadJob.getId(),
loadJob.getDbId(), loadJob.getTransactionId());
continue;
}
}
LOG.warn("skip mini load job {} in db {} as it is no longer supported", loadJob.getId(),
loadJob.getDbId());
continue;
}
idToLoadJob.put(loadJob.getId(), loadJob);
Map<String, List<LoadJob>> map = dbIdToLabelToLoadJobs.get(loadJob.getDbId());

100
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MiniLoadJob.java Normal file
View File

@ -0,0 +1,100 @@
// 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.load.loadv2;
import org.apache.doris.catalog.AuthorizationInfo;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.io.Text;
import org.apache.doris.load.EtlJobType;
import org.apache.doris.transaction.TransactionState;
import com.google.common.collect.Sets;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Set;
@Deprecated
public class MiniLoadJob extends LoadJob {
private static final Logger LOG = LogManager.getLogger(MiniLoadJob.class);
private String tableName;
private long tableId;
public MiniLoadJob() {
super(EtlJobType.MINI);
}
@Override
public Set<String> getTableNamesForShow() {
return Sets.newHashSet(tableName);
}
@Override
public Set<String> getTableNames() throws MetaNotFoundException {
return Sets.newHashSet(tableName);
}
@Override
public void beginTxn() {
}
@Override
protected void replayTxnAttachment(TransactionState txnState) {
updateLoadingStatue(txnState);
}
@Override
public void write(DataOutput out) throws IOException {
super.write(out);
Text.writeString(out, tableName);
}
@Override
public void readFields(DataInput in) throws IOException {
super.readFields(in);
tableName = Text.readString(in);
}
public AuthorizationInfo gatherAuthInfo() throws MetaNotFoundException {
Database database = Env.getCurrentInternalCatalog().getDbOrMetaException(dbId);
return new AuthorizationInfo(database.getFullName(), getTableNames());
}
private void updateLoadingStatue(TransactionState txnState) {
MiniLoadTxnCommitAttachment miniLoadTxnCommitAttachment =
(MiniLoadTxnCommitAttachment) txnState.getTxnCommitAttachment();
if (miniLoadTxnCommitAttachment == null) {
// aborted txn may not has attachment
LOG.info("no miniLoadTxnCommitAttachment, txn id: {} status: {}", txnState.getTransactionId(),
txnState.getTransactionStatus());
return;
}
loadingStatus.replaceCounter(DPP_ABNORMAL_ALL, String.valueOf(miniLoadTxnCommitAttachment.getFilteredRows()));
loadingStatus.replaceCounter(DPP_NORMAL_ALL, String.valueOf(miniLoadTxnCommitAttachment.getLoadedRows()));
if (miniLoadTxnCommitAttachment.getErrorLogUrl() != null) {
loadingStatus.setTrackingUrl(miniLoadTxnCommitAttachment.getErrorLogUrl());
}
}
}