Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.doris.catalog.EnvFactory;
import org.apache.doris.catalog.FsBroker;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
import org.apache.doris.catalog.MaterializedIndexMeta;
Expand Down Expand Up @@ -1537,8 +1538,8 @@ protected Partition resetTabletForRestore(OlapTable localTbl, OlapTable remoteTb
for (Map.Entry<Tag, List<Long>> entry : beIds.entrySet()) {
for (Long beId : entry.getValue()) {
long newReplicaId = env.getNextId();
Replica newReplica = new Replica(newReplicaId, beId, ReplicaState.NORMAL, visibleVersion,
schemaHash);
Replica newReplica = new LocalReplica(newReplicaId, beId, ReplicaState.NORMAL,
visibleVersion, schemaHash);
newTablet.addReplica(newReplica, true /* is restore */);
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,11 +107,11 @@ public Tablet createTablet(long tabletId) {
}

public Replica createReplica() {
return new Replica();
return new LocalReplica();
}

public Replica createReplica(Replica.ReplicaContext context) {
return new Replica(context);
return new LocalReplica(context);
}

public ReplicaAllocation createDefReplicaAllocation() {
Expand Down
114 changes: 114 additions & 0 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/LocalReplica.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
// 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;

import org.apache.doris.thrift.TUniqueId;

import com.google.gson.annotations.SerializedName;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class LocalReplica extends Replica {
private static final Logger LOG = LogManager.getLogger(LocalReplica.class);

@SerializedName(value = "rds", alternate = {"remoteDataSize"})
private volatile long remoteDataSize = 0;
@SerializedName(value = "ris", alternate = {"remoteInvertedIndexSize"})
private Long remoteInvertedIndexSize = 0L;
@SerializedName(value = "rss", alternate = {"remoteSegmentSize"})
private Long remoteSegmentSize = 0L;

private TUniqueId cooldownMetaId;
private long cooldownTerm = -1;

public LocalReplica() {
super();
}

public LocalReplica(ReplicaContext context) {
super(context);
}

// for rollup
// the new replica's version is -1 and last failed version is -1
public LocalReplica(long replicaId, long backendId, int schemaHash, ReplicaState state) {
super(replicaId, backendId, schemaHash, state);
}

// for create tablet and restore
public LocalReplica(long replicaId, long backendId, ReplicaState state, long version, int schemaHash) {
super(replicaId, backendId, state, version, schemaHash);
}

public LocalReplica(long replicaId, long backendId, long version, int schemaHash, long dataSize,
long remoteDataSize, long rowCount, ReplicaState state, long lastFailedVersion, long lastSuccessVersion) {
super(replicaId, backendId, version, schemaHash, dataSize, remoteDataSize, rowCount, state, lastFailedVersion,
lastSuccessVersion);
this.remoteDataSize = remoteDataSize;
}

@Override
public long getRemoteDataSize() {
return remoteDataSize;
}

@Override
public void setRemoteDataSize(long remoteDataSize) {
this.remoteDataSize = remoteDataSize;
}

@Override
public Long getRemoteInvertedIndexSize() {
return remoteInvertedIndexSize;
}

@Override
public void setRemoteInvertedIndexSize(long remoteInvertedIndexSize) {
this.remoteInvertedIndexSize = remoteInvertedIndexSize;
}

@Override
public Long getRemoteSegmentSize() {
return remoteSegmentSize;
}

@Override
public void setRemoteSegmentSize(long remoteSegmentSize) {
this.remoteSegmentSize = remoteSegmentSize;
}

@Override
public TUniqueId getCooldownMetaId() {
return cooldownMetaId;
}

@Override
public void setCooldownMetaId(TUniqueId cooldownMetaId) {
this.cooldownMetaId = cooldownMetaId;
}

@Override
public long getCooldownTerm() {
return cooldownTerm;
}

@Override
public void setCooldownTerm(long cooldownTerm) {
this.cooldownTerm = cooldownTerm;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -969,7 +969,7 @@ && getTableProperty().getDynamicPartitionProperty().getBuckets()
for (Map.Entry<Tag, List<Long>> entry3 : tag2beIds.entrySet()) {
for (Long beId : entry3.getValue()) {
long newReplicaId = env.getNextId();
Replica replica = new Replica(newReplicaId, beId, ReplicaState.NORMAL,
Replica replica = new LocalReplica(newReplicaId, beId, ReplicaState.NORMAL,
visibleVersion, schemaHash);
newTablet.addReplica(replica, true /* is restore */);
}
Expand Down
48 changes: 28 additions & 20 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Replica.java
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,6 @@ public static class ReplicaContext {
private int schemaHash = -1;
@SerializedName(value = "ds", alternate = {"dataSize"})
private volatile long dataSize = 0;
@SerializedName(value = "rds", alternate = {"remoteDataSize"})
private volatile long remoteDataSize = 0;
@SerializedName(value = "rc", alternate = {"rowCount"})
private volatile long rowCount = 0;
@SerializedName(value = "st", alternate = {"state"})
Expand All @@ -120,14 +118,6 @@ public static class ReplicaContext {
@Getter
@SerializedName(value = "lss", alternate = {"localSegmentSize"})
private Long localSegmentSize = 0L;
@Setter
@Getter
@SerializedName(value = "ris", alternate = {"remoteInvertedIndexSize"})
private Long remoteInvertedIndexSize = 0L;
@Setter
@Getter
@SerializedName(value = "rss", alternate = {"remoteSegmentSize"})
private Long remoteSegmentSize = 0L;

private volatile long totalVersionCount = -1;
private volatile long visibleVersionCount = -1;
Expand All @@ -137,9 +127,6 @@ public static class ReplicaContext {
// bad means this Replica is unrecoverable, and we will delete it
private boolean bad = false;

private TUniqueId cooldownMetaId;
private long cooldownTerm = -1;

// A replica version should increase monotonically,
// but backend may missing some versions due to disk failure or bugs.
// FE should found these and mark the replica as missing versions.
Expand Down Expand Up @@ -217,7 +204,6 @@ public Replica(long replicaId, long backendId, long version, int schemaHash,
this.schemaHash = schemaHash;

this.dataSize = dataSize;
this.remoteDataSize = remoteDataSize;
this.rowCount = rowCount;
this.state = state;
if (this.state == null) {
Expand Down Expand Up @@ -280,11 +266,33 @@ public void setDataSize(long dataSize) {
}

public long getRemoteDataSize() {
return remoteDataSize;
return 0;
}

public void setRemoteDataSize(long remoteDataSize) {
this.remoteDataSize = remoteDataSize;
if (remoteDataSize > 0) {
throw new UnsupportedOperationException("setRemoteDataSize is not supported in Replica");
}
}
Comment on lines 272 to +276

Copilot AI Dec 29, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throwing UnsupportedOperationException in setter may break existing code paths. The setRemoteDataSize method now throws UnsupportedOperationException when value > 0, but the base class Replica has an updateWithReport method (line 414) that calls this setter. If a CloudReplica receives a tablet report with remoteDataSize > 0, this will throw an exception and break the report handling. Consider either overriding updateWithReport in subclasses or checking the replica type before calling these setters.

Copilot uses AI. Check for mistakes.

public Long getRemoteInvertedIndexSize() {
return 0L;
}

public void setRemoteInvertedIndexSize(long remoteInvertedIndexSize) {
if (remoteInvertedIndexSize > 0) {
throw new UnsupportedOperationException("setRemoteInvertedIndexSize is not supported in Replica");
}
}

public Long getRemoteSegmentSize() {
return 0L;
}

public void setRemoteSegmentSize(long remoteSegmentSize) {
if (remoteSegmentSize > 0) {
throw new UnsupportedOperationException("setRemoteSegmentSize is not supported in Replica");
}
}
Comment on lines +282 to 296

Copilot AI Dec 29, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throwing UnsupportedOperationException in setter may break existing code paths. Similar to setRemoteDataSize, these methods throw UnsupportedOperationException but may be called from existing code paths. Verify that all call sites properly handle this exception or ensure these methods are never called on base Replica instances.

Copilot uses AI. Check for mistakes.

public long getRowCount() {
Expand Down Expand Up @@ -344,19 +352,19 @@ public boolean setBad(boolean bad) {
}

public TUniqueId getCooldownMetaId() {
return cooldownMetaId;
return null;
}

public void setCooldownMetaId(TUniqueId cooldownMetaId) {
this.cooldownMetaId = cooldownMetaId;
throw new UnsupportedOperationException("setCooldownMetaId is not supported in Replica");
}

public long getCooldownTerm() {
return cooldownTerm;
return -1;
}

public void setCooldownTerm(long cooldownTerm) {
this.cooldownTerm = cooldownTerm;
throw new UnsupportedOperationException("setCooldownTerm is not supported in Replica");
}

public boolean needFurtherRepair() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
Expand Down Expand Up @@ -1017,7 +1018,7 @@ public CloneTask createCloneReplicaAndTask() throws SchedException {
|| tabletHealth.status == TabletStatus.REPLICA_RELOCATING || type == Type.BALANCE
|| tabletHealth.status == TabletStatus.COLOCATE_MISMATCH
|| tabletHealth.status == TabletStatus.REPLICA_MISSING_FOR_TAG) {
replica = new Replica(
replica = new LocalReplica(
Env.getCurrentEnv().getNextId(), destBackendId,
-1 /* version */, schemaHash,
-1 /* data size */, -1, -1 /* row count */,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.doris.catalog.JdbcTable;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.ListPartitionItem;
import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
Expand Down Expand Up @@ -1112,7 +1113,7 @@ private void unprotectAddReplica(OlapTable olapTable, ReplicaPersistInfo info) {
schemaHash = olapTable.getSchemaHashByIndexId(info.getIndexId());
}

Replica replica = new Replica(info.getReplicaId(), info.getBackendId(), info.getVersion(), schemaHash,
Replica replica = new LocalReplica(info.getReplicaId(), info.getBackendId(), info.getVersion(), schemaHash,
info.getDataSize(),
info.getRemoteDataSize(), info.getRowCount(), ReplicaState.NORMAL, info.getLastFailedVersion(),
info.getLastSuccessVersion());
Expand Down Expand Up @@ -3350,7 +3351,7 @@ public TStorageMedium createTablets(MaterializedIndex index, ReplicaState replic
for (List<Long> backendIds : chosenBackendIds.values()) {
for (long backendId : backendIds) {
long replicaId = idGeneratorBuffer.getNextId();
Replica replica = new Replica(replicaId, backendId, replicaState, version,
Replica replica = new LocalReplica(replicaId, backendId, replicaState, version,
tabletMeta.getOldSchemaHash());
tablet.addReplica(replica);
totalReplicaNum++;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.doris.catalog.DiskInfo;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Index;
import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexState;
import org.apache.doris.catalog.MaterializedIndexMeta;
Expand Down Expand Up @@ -1573,7 +1574,7 @@ private static boolean addReplica(long tabletId, TabletMeta tabletMeta, TTabletI
}

// use replicaId reported by BE to maintain replica meta consistent between FE and BE
Replica replica = new Replica(replicaId, backendId, version, schemaHash,
Replica replica = new LocalReplica(replicaId, backendId, version, schemaHash,
dataSize, remoteDataSize, rowCount, ReplicaState.NORMAL,
lastFailedVersion, version);
tablet.addReplica(replica);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@
import org.apache.doris.catalog.JdbcTable;
import org.apache.doris.catalog.ListPartitionInfo;
import org.apache.doris.catalog.ListPartitionItem;
import org.apache.doris.catalog.LocalReplica;
import org.apache.doris.catalog.LocalTablet;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.MapType;
Expand Down Expand Up @@ -522,8 +523,7 @@ public class GsonUtils {
// runtime adapter for class "CloudReplica".
private static RuntimeTypeAdapterFactory<Replica> replicaTypeAdapterFactory = RuntimeTypeAdapterFactory
.of(Replica.class, "clazz")
.registerDefaultSubtype(Replica.class)
.registerSubtype(Replica.class, Replica.class.getSimpleName())
.registerSubtype(LocalReplica.class, LocalReplica.class.getSimpleName())
.registerSubtype(CloudReplica.class, CloudReplica.class.getSimpleName());

private static RuntimeTypeAdapterFactory<Tablet> tabletTypeAdapterFactory;
Expand All @@ -536,9 +536,12 @@ public class GsonUtils {
if (Config.isNotCloudMode()) {
tabletTypeAdapterFactory.registerDefaultSubtype(LocalTablet.class);
tabletTypeAdapterFactory.registerCompatibleSubtype(LocalTablet.class, Tablet.class.getSimpleName());
replicaTypeAdapterFactory.registerDefaultSubtype(LocalReplica.class);
replicaTypeAdapterFactory.registerCompatibleSubtype(LocalReplica.class, Replica.class.getSimpleName());
} else {
// compatible with old cloud code.
tabletTypeAdapterFactory.registerDefaultSubtype(CloudTablet.class);
replicaTypeAdapterFactory.registerDefaultSubtype(CloudReplica.class);
}
}

Expand Down
Loading
Loading