Skip to content

Commit

Permalink
[fix](compatibility) Version 1.2 upgraded to 2.0 compatible with mini…
Browse files Browse the repository at this point in the history
…load metadata (apache#22590)
  • Loading branch information
luozenglin authored Aug 4, 2023
1 parent 9cf6b1b commit dc06c48
Show file tree
Hide file tree
Showing 3 changed files with 107 additions and 21 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -865,6 +865,8 @@ public static LoadJob read(DataInput in) throws IOException {
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());
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@
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;
Expand Down Expand Up @@ -815,7 +814,8 @@ public void updateJobProgress(Long jobId, Long beId, TUniqueId loadId, TUniqueId
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) {
Expand All @@ -836,25 +836,9 @@ public void readFields(DataInput in) throws IOException {
}

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());
Expand Down
100 changes: 100 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/load/loadv2/MiniLoadJob.java
Original file line number Diff line number Diff line change
@@ -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());
}
}
}

0 comments on commit dc06c48

Please sign in to comment.