You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by yu...@apache.org on 2026/08/26 13:23:41 UTC
(fluss) branch main updated: [server] Support primary-key writes to historical partitions in server (#4001)
This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 9a915d9f7 [server] Support primary-key writes to historical partitions in server (#4001)
9a915d9f7 is described below
commit 9a915d9f73714d75d3e1b89b0027feddc6989ea4
Author: yuxia Luo <lu...@alumni.sjtu.edu.cn>
AuthorDate: Wed Aug 26 21:23:35 2026 +0800
[server] Support primary-key writes to historical partitions in server (#4001)
---
.../fluss/client/table/getter/PartitionGetter.java | 47 +-
.../org/apache/fluss/row/RowPartitionGetter.java | 16 +-
.../fluss/rpc/entity/PutKvResultForBucket.java | 45 +-
.../fluss/rpc/netty/server/NettyServerHandler.java | 4 +
.../fluss/rpc/netty/server/RequestsMetrics.java | 5 +-
.../org/apache/fluss/rpc/protocol/ApiKeys.java | 3 +-
.../fluss/rpc/util/CommonRpcMessageUtils.java | 12 +
fluss-rpc/src/main/proto/FlussApi.proto | 4 +
fluss-rust/crates/fluss/proto/FlussApi.proto | 4 +
fluss-rust/crates/fluss/src/client/write/sender.rs | 1 +
fluss-rust/crates/fluss/src/proto/fluss.rs | 6 +
fluss-rust/crates/fluss/src/rpc/api_key.rs | 5 +-
fluss-rust/crates/fluss/src/rpc/message/put_kv.rs | 1 +
.../crates/fluss/src/rpc/server_connection.rs | 16 +-
.../fluss/server/entity/PutKvDataForBucket.java | 59 ++
.../java/org/apache/fluss/server/kv/KvManager.java | 2 -
.../apache/fluss/server/kv/KvRecoverHelper.java | 33 +-
.../apache/fluss/server/kv/KvStateAccessor.java | 121 +++
.../fluss/server/kv/KvStateLookupResult.java | 131 +++
.../java/org/apache/fluss/server/kv/KvTablet.java | 500 +++--------
.../apache/fluss/server/kv/KvWriteProcessor.java | 564 ++++++++++++
.../kv/historical/HistoricalKvKeyEncoder.java | 79 ++
.../kv/historical/HistoricalValueLookup.java | 37 +
.../apache/fluss/server/kv/rocksdb/RocksDBKv.java | 34 +-
.../server/metrics/group/TableMetricGroup.java | 35 +-
.../org/apache/fluss/server/replica/Replica.java | 176 +++-
.../fluss/server/replica/ReplicaManager.java | 186 +++-
.../HistoricalLakeLookupManager.java | 305 ++-----
.../historical/HistoricalPartitionManager.java | 353 ++++++++
.../HistoricalPartitionTaskExecutor.java | 210 +++++
.../apache/fluss/server/tablet/TabletService.java | 36 +-
.../fluss/server/utils/ServerRpcMessageUtils.java | 45 +-
.../kv/historical/HistoricalKvKeyEncoderTest.java | 80 ++
.../fluss/server/kv/rocksdb/RocksDBKvTest.java | 12 +
.../HistoricalLakeLookupManagerTest.java | 391 ++------
.../historical/HistoricalPartitionManagerTest.java | 997 +++++++++++++++++++++
.../HistoricalPartitionTaskExecutorTest.java | 232 +++++
.../server/utils/ServerRpcMessageUtilsTest.java | 83 ++
.../maintenance/observability/monitor-metrics.md | 15 +-
39 files changed, 3858 insertions(+), 1027 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java b/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java
index ef0766ed4..6d05c887e 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java
@@ -19,59 +19,28 @@ package org.apache.fluss.client.table.getter;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.types.DataType;
+import org.apache.fluss.row.RowPartitionGetter;
import org.apache.fluss.types.RowType;
-import org.apache.fluss.utils.PartitionUtils;
-import java.util.ArrayList;
import java.util.List;
-import static org.apache.fluss.utils.Preconditions.checkArgument;
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
-
/** A getter to get partition name from a row. */
public class PartitionGetter {
- private final List<String> partitionKeys;
- private final List<InternalRow.FieldGetter> partitionFieldGetters;
- private final List<DataType> partitionTypes;
+ private final RowPartitionGetter delegate;
+ /** Creates a partition getter for the given row type and partition keys. */
public PartitionGetter(RowType rowType, List<String> partitionKeys) {
- // check the partition column
- List<String> fieldNames = rowType.getFieldNames();
- this.partitionKeys = partitionKeys;
- partitionFieldGetters = new ArrayList<>();
- partitionTypes = new ArrayList<>();
- for (String partitionKey : partitionKeys) {
- int partitionColumnIndex = fieldNames.indexOf(partitionKey);
- checkArgument(
- partitionColumnIndex >= 0,
- "The partition column %s is not in the row %s.",
- partitionKey,
- rowType);
-
- // check the data type of the partition column
- DataType partitionColumnDataType = rowType.getTypeAt(partitionColumnIndex);
- partitionTypes.add(partitionColumnDataType);
- partitionFieldGetters.add(
- InternalRow.createFieldGetter(partitionColumnDataType, partitionColumnIndex));
- }
+ this.delegate = new RowPartitionGetter(rowType, partitionKeys);
}
+ /** Returns the partition name extracted from the given row. */
public String getPartition(InternalRow row) {
- return getResolvedPartitionSpec(row).getPartitionName();
+ return delegate.getPartition(row);
}
+ /** Returns the resolved partition spec extracted from the given row. */
public ResolvedPartitionSpec getResolvedPartitionSpec(InternalRow row) {
- List<String> partitionValues = new ArrayList<>();
- for (int i = 0; i < partitionFieldGetters.size(); i++) {
- InternalRow.FieldGetter partitionFieldGetter = partitionFieldGetters.get(i);
- DataType dataType = partitionTypes.get(i);
- Object partitionValue = partitionFieldGetter.getFieldOrNull(row);
- checkNotNull(partitionValue, "Partition value shouldn't be null.");
- partitionValues.add(
- PartitionUtils.convertValueOfType(partitionValue, dataType.getTypeRoot()));
- }
- return new ResolvedPartitionSpec(partitionKeys, partitionValues);
+ return delegate.getResolvedPartitionSpec(row);
}
}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java b/fluss-common/src/main/java/org/apache/fluss/row/RowPartitionGetter.java
similarity index 86%
copy from fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java
copy to fluss-common/src/main/java/org/apache/fluss/row/RowPartitionGetter.java
index ef0766ed4..6d4f4728f 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/getter/PartitionGetter.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/RowPartitionGetter.java
@@ -15,10 +15,10 @@
* limitations under the License.
*/
-package org.apache.fluss.client.table.getter;
+package org.apache.fluss.row;
+import org.apache.fluss.annotation.Internal;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
-import org.apache.fluss.row.InternalRow;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.PartitionUtils;
@@ -29,15 +29,16 @@ import java.util.List;
import static org.apache.fluss.utils.Preconditions.checkArgument;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
-/** A getter to get partition name from a row. */
-public class PartitionGetter {
+/** Extracts a partition from an {@link InternalRow}. */
+@Internal
+public class RowPartitionGetter {
private final List<String> partitionKeys;
private final List<InternalRow.FieldGetter> partitionFieldGetters;
private final List<DataType> partitionTypes;
- public PartitionGetter(RowType rowType, List<String> partitionKeys) {
- // check the partition column
+ /** Creates a partition getter for the given row type and partition keys. */
+ public RowPartitionGetter(RowType rowType, List<String> partitionKeys) {
List<String> fieldNames = rowType.getFieldNames();
this.partitionKeys = partitionKeys;
partitionFieldGetters = new ArrayList<>();
@@ -50,7 +51,6 @@ public class PartitionGetter {
partitionKey,
rowType);
- // check the data type of the partition column
DataType partitionColumnDataType = rowType.getTypeAt(partitionColumnIndex);
partitionTypes.add(partitionColumnDataType);
partitionFieldGetters.add(
@@ -58,10 +58,12 @@ public class PartitionGetter {
}
}
+ /** Returns the partition name extracted from the given row. */
public String getPartition(InternalRow row) {
return getResolvedPartitionSpec(row).getPartitionName();
}
+ /** Returns the resolved partition spec extracted from the given row. */
public ResolvedPartitionSpec getResolvedPartitionSpec(InternalRow row) {
List<String> partitionValues = new ArrayList<>();
for (int i = 0; i < partitionFieldGetters.size(); i++) {
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PutKvResultForBucket.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PutKvResultForBucket.java
index eb8de8677..3d6d343c7 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PutKvResultForBucket.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/PutKvResultForBucket.java
@@ -23,6 +23,8 @@ import org.apache.fluss.rpc.messages.PutKvRequest;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.Errors;
+import javax.annotation.Nullable;
+
/** Result of {@link PutKvRequest} for each table bucket. */
@Internal
public class PutKvResultForBucket extends WriteResultForBucket {
@@ -30,27 +32,58 @@ public class PutKvResultForBucket extends WriteResultForBucket {
/** Backpressure pressure value: 0=normal, (0,1)=DELAYED zone. */
private final float pressure;
+ private final @Nullable String originalPartitionName;
+
public PutKvResultForBucket(TableBucket tableBucket, long changeLogEndOffset) {
- this(tableBucket, changeLogEndOffset, 0f);
+ this(tableBucket, changeLogEndOffset, ApiError.NONE, 0f, null);
}
public PutKvResultForBucket(TableBucket tableBucket, long changeLogEndOffset, float pressure) {
- super(tableBucket, changeLogEndOffset, ApiError.NONE);
- this.pressure = pressure;
+ this(tableBucket, changeLogEndOffset, ApiError.NONE, pressure, null);
}
public PutKvResultForBucket(TableBucket tableBucket, ApiError error) {
- super(tableBucket, -1L, error);
- this.pressure = 0f;
+ this(tableBucket, -1L, error, 0f, null);
+ }
+
+ public static PutKvResultForBucket historicalSuccess(
+ TableBucket tableBucket,
+ long changeLogEndOffset,
+ @Nullable String originalPartitionName) {
+ return new PutKvResultForBucket(
+ tableBucket, changeLogEndOffset, ApiError.NONE, 0f, originalPartitionName);
+ }
+
+ public static PutKvResultForBucket historicalFailure(
+ TableBucket tableBucket, ApiError error, @Nullable String originalPartitionName) {
+ return new PutKvResultForBucket(tableBucket, -1L, error, 0f, originalPartitionName);
+ }
+
+ private PutKvResultForBucket(
+ TableBucket tableBucket,
+ long changeLogEndOffset,
+ ApiError error,
+ float pressure,
+ @Nullable String originalPartitionName) {
+ super(tableBucket, changeLogEndOffset, error);
+ this.pressure = pressure;
+ this.originalPartitionName = originalPartitionName;
}
public float getPressure() {
return pressure;
}
+ /** Returns the original partition name for a historical write, or null for a normal write. */
+ public @Nullable String getOriginalPartitionName() {
+ return originalPartitionName;
+ }
+
@Override
public <T extends WriteResultForBucket> T copy(Errors newError) {
//noinspection unchecked
- return (T) new PutKvResultForBucket(tableBucket, newError.toApiError());
+ return (T)
+ new PutKvResultForBucket(
+ tableBucket, -1L, newError.toApiError(), 0f, originalPartitionName);
}
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java
index 25328fe44..b04722d5c 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServerHandler.java
@@ -27,6 +27,7 @@ import org.apache.fluss.rpc.messages.AuthenticateRequest;
import org.apache.fluss.rpc.messages.AuthenticateResponse;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.LookupRequest;
+import org.apache.fluss.rpc.messages.PutKvRequest;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.ApiKeys;
import org.apache.fluss.rpc.protocol.ApiManager;
@@ -57,6 +58,7 @@ import static org.apache.fluss.rpc.protocol.MessageCodec.encodeErrorResponse;
import static org.apache.fluss.rpc.protocol.MessageCodec.encodeServerFailure;
import static org.apache.fluss.rpc.protocol.MessageCodec.encodeSuccessResponse;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup;
+import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalPut;
/** Implementation of the channel handler to process inbound requests for RPC server. */
public final class NettyServerHandler extends ChannelInboundHandlerAdapter {
@@ -308,6 +310,8 @@ public final class NettyServerHandler extends ChannelInboundHandlerAdapter {
isFromFollower = fetchLogRequest.getFollowerServerId() >= 0;
} else if (request.getApiKey() == ApiKeys.LOOKUP.id) {
isHistorical = hasHistoricalLookup((LookupRequest) requestMessage);
+ } else if (request.getApiKey() == ApiKeys.PUT_KV.id) {
+ isHistorical = hasHistoricalPut((PutKvRequest) requestMessage);
}
return requestsMetrics.getMetrics(request.getApiKey(), isFromFollower, isHistorical);
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
index 3fff71f70..d3aa345ca 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/RequestsMetrics.java
@@ -59,6 +59,9 @@ public class RequestsMetrics {
if (apiKey == ApiKeys.LOOKUP) {
addMetrics(serverMetricsGroup, toRequestName(apiKey, false, true));
}
+ if (apiKey == ApiKeys.PUT_KV) {
+ addMetrics(serverMetricsGroup, toRequestName(apiKey, false, true));
+ }
}
this.requestMetricGroup = serverMetricsGroup.addGroup("request");
}
@@ -103,7 +106,7 @@ public class RequestsMetrics {
case PRODUCE_LOG:
return "produceLog";
case PUT_KV:
- return "putKv";
+ return isHistorical ? "historicalPutKv" : "putKv";
case LOOKUP:
return isHistorical ? "historicalLookup" : "lookup";
case PREFIX_LOOKUP:
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index e9f18b3d6..979674c25 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -51,7 +51,8 @@ public enum ApiKeys {
// Version 2: Understands the STORAGE_BACKPRESSURE_EXCEPTION error code (72) returned when the
// KV storage engine rejects a write under pressure; older versions receive the
// retriable KV_STORAGE_EXCEPTION instead.
- PUT_KV(1016, 0, 2, PUBLIC),
+ // Version 3: Supports original_partition_name in requests and responses for historical writes.
+ PUT_KV(1016, 0, 3, PUBLIC),
// Version 0: Uses lake's encoder for primary key encoding (legacy behavior).
// Version 1: Uses CompactedKeyEncoder for primary key encoding when bucket key differs from
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
index a5aa22cb3..219b51087 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
@@ -34,6 +34,7 @@ import org.apache.fluss.rpc.messages.PbKeyValue;
import org.apache.fluss.rpc.messages.PbPartitionSpec;
import org.apache.fluss.rpc.messages.PbRemoteLogFetchInfo;
import org.apache.fluss.rpc.messages.PbRemoteLogSegment;
+import org.apache.fluss.rpc.messages.PutKvRequest;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.security.acl.AccessControlEntry;
import org.apache.fluss.security.acl.AccessControlEntryFilter;
@@ -71,6 +72,17 @@ public class CommonRpcMessageUtils {
&& lookupRequest.getBucketsReqAt(0).hasOriginalPartitionName();
}
+ /**
+ * Returns whether the put-KV request is for historical partition writes.
+ *
+ * <p>Normal and historical write buckets cannot be mixed in the same request, so the first
+ * bucket determines the request type.
+ */
+ public static boolean hasHistoricalPut(PutKvRequest putKvRequest) {
+ return putKvRequest.getBucketsReqsCount() > 0
+ && putKvRequest.getBucketsReqAt(0).hasOriginalPartitionName();
+ }
+
public static List<PbAclInfo> toPbAclInfos(Collection<AclBinding> aclBindings) {
return aclBindings.stream()
.map(CommonRpcMessageUtils::toPbAclInfo)
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index d03acb92f..9d5e05fbd 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -939,6 +939,8 @@ message PbPutKvReqForBucket {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
required bytes records = 3;
+ // The original partition name for historical PK writes. It is unset for normal writes.
+ optional string original_partition_name = 4;
}
message PbPutKvRespForBucket {
@@ -953,6 +955,8 @@ message PbPutKvRespForBucket {
// 0 = normal (no pressure), (0,1) = DELAYED zone (normalized L0 ratio)
// introduced in v1.0
optional float pressure = 6;
+ // The original partition name for historical PK writes. It is unset for normal writes.
+ optional string original_partition_name = 7;
}
message PbLookupReqForBucket {
diff --git a/fluss-rust/crates/fluss/proto/FlussApi.proto b/fluss-rust/crates/fluss/proto/FlussApi.proto
index d03acb92f..9d5e05fbd 100644
--- a/fluss-rust/crates/fluss/proto/FlussApi.proto
+++ b/fluss-rust/crates/fluss/proto/FlussApi.proto
@@ -939,6 +939,8 @@ message PbPutKvReqForBucket {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
required bytes records = 3;
+ // The original partition name for historical PK writes. It is unset for normal writes.
+ optional string original_partition_name = 4;
}
message PbPutKvRespForBucket {
@@ -953,6 +955,8 @@ message PbPutKvRespForBucket {
// 0 = normal (no pressure), (0,1) = DELAYED zone (normalized L0 ratio)
// introduced in v1.0
optional float pressure = 6;
+ // The original partition name for historical PK writes. It is unset for normal writes.
+ optional string original_partition_name = 7;
}
message PbLookupReqForBucket {
diff --git a/fluss-rust/crates/fluss/src/client/write/sender.rs b/fluss-rust/crates/fluss/src/client/write/sender.rs
index f264fac0e..d5f25e8dd 100644
--- a/fluss-rust/crates/fluss/src/client/write/sender.rs
+++ b/fluss-rust/crates/fluss/src/client/write/sender.rs
@@ -1209,6 +1209,7 @@ mod tests {
error_message: None,
log_end_offset: None,
pressure: Some(0.5),
+ original_partition_name: None,
}],
};
sender
diff --git a/fluss-rust/crates/fluss/src/proto/fluss.rs b/fluss-rust/crates/fluss/src/proto/fluss.rs
index 6fe822a1a..95e7045c5 100644
--- a/fluss-rust/crates/fluss/src/proto/fluss.rs
+++ b/fluss-rust/crates/fluss/src/proto/fluss.rs
@@ -1249,6 +1249,9 @@ pub struct PbPutKvReqForBucket {
pub bucket_id: i32,
#[prost(bytes = "bytes", required, tag = "3")]
pub records: ::prost::bytes::Bytes,
+ /// The original partition name for historical PK writes. It is unset for normal writes.
+ #[prost(string, optional, tag = "4")]
+ pub original_partition_name: ::core::option::Option<::prost::alloc::string::String>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PbPutKvRespForBucket {
@@ -1269,6 +1272,9 @@ pub struct PbPutKvRespForBucket {
/// introduced in v1.0
#[prost(float, optional, tag = "6")]
pub pressure: ::core::option::Option<f32>,
+ /// The original partition name for historical PK writes. It is unset for normal writes.
+ #[prost(string, optional, tag = "7")]
+ pub original_partition_name: ::core::option::Option<::prost::alloc::string::String>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct PbLookupReqForBucket {
diff --git a/fluss-rust/crates/fluss/src/rpc/api_key.rs b/fluss-rust/crates/fluss/src/rpc/api_key.rs
index 239b1d448..e9720a4e2 100644
--- a/fluss-rust/crates/fluss/src/rpc/api_key.rs
+++ b/fluss-rust/crates/fluss/src/rpc/api_key.rs
@@ -129,8 +129,9 @@ impl ApiKey {
| ApiKey::GetClusterHealth
| ApiKey::ListRemoteLogManifests
| ApiKey::ListKvSnapshots => Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(0))),
- // PutKv v2 adds the storage backpressure error code.
- ApiKey::PutKv => Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(2))),
+ // PutKv v2 adds the storage backpressure error code; v3 adds historical partition
+ // context to requests and responses.
+ ApiKey::PutKv => Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(3))),
// Lookup / PrefixLookup support v0 (legacy key encoding) and v1
// (Paimon BinaryRow key encoding for kv_format_version=2
// non-default bucket keys). The Rust client encodes both.
diff --git a/fluss-rust/crates/fluss/src/rpc/message/put_kv.rs b/fluss-rust/crates/fluss/src/rpc/message/put_kv.rs
index 8ffdaeefe..c6ce9eb5c 100644
--- a/fluss-rust/crates/fluss/src/rpc/message/put_kv.rs
+++ b/fluss-rust/crates/fluss/src/rpc/message/put_kv.rs
@@ -51,6 +51,7 @@ impl PutKvRequest {
partition_id: ready_batch.table_bucket.partition_id(),
bucket_id: ready_batch.table_bucket.bucket_id(),
records: ready_batch.write_batch.build()?,
+ original_partition_name: None,
})
}
diff --git a/fluss-rust/crates/fluss/src/rpc/server_connection.rs b/fluss-rust/crates/fluss/src/rpc/server_connection.rs
index 885dcbbf6..a8b36cbec 100644
--- a/fluss-rust/crates/fluss/src/rpc/server_connection.rs
+++ b/fluss-rust/crates/fluss/src/rpc/server_connection.rs
@@ -1186,11 +1186,11 @@ mod tests {
assert_eq!(
resolve_api_version_for(None, ApiKey::PutKv).unwrap(),
- ApiVersion(2)
+ ApiVersion(3)
);
let server_versions = vec![
- // PutKv: server v0..v3, client v0..v2 → negotiated v2
+ // PutKv: server v0..v3, client v0..v3 → negotiated v3
PbApiVersion {
api_key: 1016,
min_version: 0,
@@ -1220,6 +1220,18 @@ mod tests {
// Successful negotiation cases
assert_eq!(
negotiated.highest_available_version(ApiKey::PutKv).unwrap(),
+ ApiVersion(3)
+ );
+
+ let old_server_versions = ServerApiVersions::new(&[PbApiVersion {
+ api_key: 1016,
+ min_version: 0,
+ max_version: 2,
+ }]);
+ assert_eq!(
+ old_server_versions
+ .highest_available_version(ApiKey::PutKv)
+ .unwrap(),
ApiVersion(2)
);
assert_eq!(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/entity/PutKvDataForBucket.java b/fluss-server/src/main/java/org/apache/fluss/server/entity/PutKvDataForBucket.java
new file mode 100644
index 000000000..af164d521
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/entity/PutKvDataForBucket.java
@@ -0,0 +1,59 @@
+/*
+ * 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.fluss.server.entity;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.KvRecordBatch;
+
+import javax.annotation.Nullable;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Put KV request data and historical partition context for one table bucket. */
+public final class PutKvDataForBucket {
+
+ private final TableBucket tableBucket;
+ private final KvRecordBatch records;
+ private final @Nullable String originalPartitionName;
+
+ /** Creates decoded put-KV data for one table bucket. */
+ public PutKvDataForBucket(
+ TableBucket tableBucket,
+ KvRecordBatch records,
+ @Nullable String originalPartitionName) {
+ this.tableBucket = checkNotNull(tableBucket, "tableBucket must not be null.");
+ this.records = checkNotNull(records, "records must not be null.");
+ this.originalPartitionName = originalPartitionName;
+ }
+
+ /** Returns the physical table bucket targeted by this request data. */
+ public TableBucket tableBucket() {
+ return tableBucket;
+ }
+
+ /** Returns the encoded KV records for this table bucket. */
+ public KvRecordBatch records() {
+ return records;
+ }
+
+ /** Returns the original partition name for a historical write, or null. */
+ public @Nullable String originalPartitionName() {
+ return originalPartitionName;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
index c1ee5f867..84273ab38 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvManager.java
@@ -386,8 +386,6 @@ public final class KvManager extends TabletManagerBase implements ServerReconfig
dropKvTablet.getKvTabletDir().getAbsolutePath()),
e);
}
- } else {
- LOG.warn("Fail to delete kv bucket {}.", tableBucket.getBucket());
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
index 29c6cae8c..780533c68 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvRecoverHelper.java
@@ -31,11 +31,13 @@ import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.row.BinaryRow;
import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.RowPartitionGetter;
import org.apache.fluss.row.encode.KeyEncoder;
import org.apache.fluss.row.encode.RowEncoder;
import org.apache.fluss.row.encode.ValueEncoder;
import org.apache.fluss.row.indexed.IndexedRow;
import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
+import org.apache.fluss.server.kv.historical.HistoricalKvKeyEncoder;
import org.apache.fluss.server.log.FetchIsolation;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -53,6 +55,9 @@ import javax.annotation.Nullable;
import java.util.List;
import static org.apache.fluss.server.TabletManagerBase.getTableInfo;
+import static org.apache.fluss.server.kv.KvStateAccessor.HISTORICAL_TOMBSTONE;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
/** A helper for recovering Kv from log. */
public class KvRecoverHelper {
@@ -67,6 +72,7 @@ public class KvRecoverHelper {
private final KvFormat kvFormat;
private final LogFormat logFormat;
private final RemoteLogFetcher remoteLogFetcher;
+ private final boolean historicalPartition;
// will be initialized when first encounter a log record during recovering from log
private Integer currentSchemaId;
@@ -77,6 +83,7 @@ public class KvRecoverHelper {
private final SchemaGetter schemaGetter;
private InternalRow.FieldGetter[] currentFieldGetters;
+ private @Nullable RowPartitionGetter historicalPartitionGetter;
public KvRecoverHelper(
KvTablet kvTablet,
@@ -88,7 +95,8 @@ public class KvRecoverHelper {
KvFormat kvFormat,
LogFormat logFormat,
SchemaGetter schemaGetter,
- RemoteLogFetcher remoteLogFetcher) {
+ RemoteLogFetcher remoteLogFetcher,
+ boolean historicalPartition) {
this.kvTablet = kvTablet;
this.logTablet = logTablet;
this.recoverPointOffset = recoverPointOffset;
@@ -99,6 +107,7 @@ public class KvRecoverHelper {
this.logFormat = logFormat;
this.schemaGetter = schemaGetter;
this.remoteLogFetcher = remoteLogFetcher;
+ this.historicalPartition = historicalPartition;
}
public void recover() throws Exception {
@@ -134,7 +143,11 @@ public class KvRecoverHelper {
ThrowingConsumer<KeyValueAndLogOffset, Exception> resumeRecordApplier =
(resumeRecord) -> {
if (resumeRecord.value == null) {
- kvBatchWriter.delete(resumeRecord.key);
+ if (historicalPartition) {
+ kvBatchWriter.put(resumeRecord.key, HISTORICAL_TOMBSTONE);
+ } else {
+ kvBatchWriter.delete(resumeRecord.key);
+ }
} else {
kvBatchWriter.put(resumeRecord.key, resumeRecord.value);
}
@@ -263,6 +276,13 @@ public class KvRecoverHelper {
if (changeType != ChangeType.UPDATE_BEFORE) {
InternalRow logRow = logRecord.getRow();
byte[] key = keyEncoder.encodeKey(logRow);
+ if (historicalPartition) {
+ key =
+ HistoricalKvKeyEncoder.encode(
+ checkNotNull(historicalPartitionGetter)
+ .getPartition(logRow),
+ key);
+ }
byte[] value = null;
if (changeType != ChangeType.DELETE) {
// the log row format may not compatible with kv row format,
@@ -331,6 +351,15 @@ public class KvRecoverHelper {
tableInfo.getPhysicalPrimaryKeys(),
tableInfo.getTableConfig(),
tableInfo.isDefaultBucketKey());
+ if (historicalPartition) {
+ List<String> partitionKeys = tableInfo.getPartitionKeys();
+ checkArgument(
+ !partitionKeys.isEmpty(),
+ "Historical KV recovery requires at least one partition key.");
+ historicalPartitionGetter = new RowPartitionGetter(currentRowType, partitionKeys);
+ } else {
+ historicalPartitionGetter = null;
+ }
rowEncoder = RowEncoder.create(kvFormat, dataTypes);
currentFieldGetters = new InternalRow.FieldGetter[currentRowType.getFieldCount()];
for (int i = 0; i < currentRowType.getFieldCount(); i++) {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java
new file mode 100644
index 000000000..1e0cd3996
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateAccessor.java
@@ -0,0 +1,121 @@
+/*
+ * 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.fluss.server.kv;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.server.kv.historical.HistoricalKvKeyEncoder;
+import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer;
+import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.Key;
+import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.TruncateReason;
+import org.apache.fluss.server.kv.rocksdb.RocksDBKv;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Accesses a KV tablet's local prewrite buffer and RocksDB state. */
+@Internal
+public final class KvStateAccessor {
+
+ /** Encoded RocksDB value marking a deleted key in historical KV state. */
+ static final byte[] HISTORICAL_TOMBSTONE = new byte[0];
+
+ private final KvPreWriteBuffer preWriteBuffer;
+ private final RocksDBKv rocksDBKv;
+ private final boolean historicalPartition;
+
+ KvStateAccessor(
+ KvPreWriteBuffer preWriteBuffer, RocksDBKv rocksDBKv, boolean historicalPartition) {
+ this.preWriteBuffer = preWriteBuffer;
+ this.rocksDBKv = rocksDBKv;
+ this.historicalPartition = historicalPartition;
+ }
+
+ /**
+ * Encodes the logical primary key into the physical key used by this state.
+ *
+ * <p>Normal KV state keeps the primary key unchanged. Historical KV state also encodes the
+ * original partition context because multiple original partitions share one historical KV
+ * tablet. {@code originalPartitionName} is null for normal state and must be present for
+ * historical state.
+ */
+ public Key encodeKey(byte[] primaryKey, @Nullable String originalPartitionName) {
+ if (!historicalPartition) {
+ checkArgument(
+ originalPartitionName == null,
+ "A normal KV state key must not carry an original partition name");
+ return Key.of(primaryKey);
+ }
+
+ String partitionName =
+ checkNotNull(originalPartitionName, "originalPartitionName must not be null");
+ checkArgument(!partitionName.isEmpty(), "originalPartitionName must not be empty");
+ return Key.of(HistoricalKvKeyEncoder.encode(partitionName, primaryKey));
+ }
+
+ /** Looks up an encoded key from the local prewrite buffer and RocksDB state. */
+ public KvStateLookupResult lookup(Key key) throws IOException {
+ KvPreWriteBuffer.Value bufferedValue = preWriteBuffer.get(key);
+ if (bufferedValue != null) {
+ byte[] value = bufferedValue.get();
+ return value == null
+ ? KvStateLookupResult.deleted()
+ : KvStateLookupResult.present(value);
+ }
+
+ byte[] value = rocksDBKv.get(key.get());
+ if (value == null) {
+ return KvStateLookupResult.notFound();
+ }
+ // Historical KV tablets persist deletes as empty values so that a local miss does not
+ // expose a stale value from lake storage after the buffered delete has been flushed.
+ return value.length == 0
+ ? KvStateLookupResult.deleted()
+ : KvStateLookupResult.present(value);
+ }
+
+ /** Adds an insert mutation to the prewrite buffer. */
+ public void insert(Key key, byte[] value, long logOffset) {
+ if (historicalPartition) {
+ checkArgument(value.length > 0, "Historical KV insert value must not be empty");
+ }
+ preWriteBuffer.insert(key, value, logOffset);
+ }
+
+ /** Adds an update mutation to the prewrite buffer. */
+ public void update(Key key, @Nullable byte[] value, long logOffset) {
+ if (historicalPartition) {
+ checkNotNull(value, "Historical KV update value must not be null");
+ checkArgument(value.length > 0, "Historical KV update value must not be empty");
+ }
+ preWriteBuffer.update(key, value, logOffset);
+ }
+
+ /** Adds a delete mutation to the prewrite buffer. */
+ public void delete(Key key, long logOffset) {
+ preWriteBuffer.delete(key, logOffset);
+ }
+
+ /** Truncates pending mutations to the given log offset. */
+ public void truncateTo(long logOffset, TruncateReason reason) {
+ preWriteBuffer.truncateTo(logOffset, reason);
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateLookupResult.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateLookupResult.java
new file mode 100644
index 000000000..fd21586a6
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvStateLookupResult.java
@@ -0,0 +1,131 @@
+/*
+ * 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.fluss.server.kv;
+
+import org.apache.fluss.annotation.Internal;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Result of looking up a key from a KV state source.
+ *
+ * <p>{@link Status#NOT_FOUND} means that the queried source has no state for the key, so the caller
+ * may consult another source. {@link Status#DELETED} confirms that the key has been deleted and
+ * must stop fallback from exposing an older value.
+ */
+@Internal
+public final class KvStateLookupResult {
+
+ /** Status of a KV state lookup. */
+ public enum Status {
+ /** The key was not found in the queried state source. */
+ NOT_FOUND,
+
+ /** The queried state source contains a non-empty encoded value. */
+ PRESENT,
+
+ /** The key is known to have been deleted. */
+ DELETED
+ }
+
+ private static final KvStateLookupResult NOT_FOUND =
+ new KvStateLookupResult(Status.NOT_FOUND, null);
+ private static final KvStateLookupResult DELETED =
+ new KvStateLookupResult(Status.DELETED, null);
+
+ private final Status status;
+ private final @Nullable byte[] value;
+
+ private KvStateLookupResult(Status status, @Nullable byte[] value) {
+ this.status = status;
+ this.value = value;
+ }
+
+ /** Returns a result indicating that the key was not found in the queried state source. */
+ public static KvStateLookupResult notFound() {
+ return NOT_FOUND;
+ }
+
+ /** Returns a result containing a non-empty encoded value. */
+ public static KvStateLookupResult present(byte[] value) {
+ checkNotNull(value, "value must not be null");
+ checkArgument(value.length > 0, "value must not be empty");
+ return new KvStateLookupResult(Status.PRESENT, value);
+ }
+
+ /** Returns a result indicating that the key is known to have been deleted. */
+ public static KvStateLookupResult deleted() {
+ return DELETED;
+ }
+
+ /** Returns the lookup status. */
+ public Status status() {
+ return status;
+ }
+
+ /** Returns whether this result contains an encoded value. */
+ public boolean isPresent() {
+ return status == Status.PRESENT;
+ }
+
+ /** Returns whether the key is known to have been deleted. */
+ public boolean isDeleted() {
+ return status == Status.DELETED;
+ }
+
+ /**
+ * Returns the encoded value, or null when the key is absent or deleted.
+ *
+ * <p>The returned byte array is owned by the underlying state and must not be modified.
+ */
+ public @Nullable byte[] value() {
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ KvStateLookupResult that = (KvStateLookupResult) o;
+ return status == that.status && Arrays.equals(value, that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * status.hashCode() + Arrays.hashCode(value);
+ }
+
+ @Override
+ public String toString() {
+ return "KvStateLookupResult{"
+ + "status="
+ + status
+ + ", value="
+ + Arrays.toString(value)
+ + '}';
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
index b43bd93e0..8fae46559 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
@@ -21,44 +21,29 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.compression.ArrowCompressionInfo;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.exception.DeletionDisabledException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.KvStorageException;
-import org.apache.fluss.exception.SchemaNotExistException;
import org.apache.fluss.exception.StorageBackpressureException;
import org.apache.fluss.memory.MemorySegmentPool;
import org.apache.fluss.metadata.ChangelogImage;
-import org.apache.fluss.metadata.DeleteBehavior;
import org.apache.fluss.metadata.KvFormat;
-import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.PhysicalTablePath;
-import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
-import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.record.BinaryValue;
import org.apache.fluss.record.ChangeType;
-import org.apache.fluss.record.KvRecord;
import org.apache.fluss.record.KvRecordBatch;
-import org.apache.fluss.record.KvRecordReadContext;
-import org.apache.fluss.row.BinaryRow;
-import org.apache.fluss.row.PaddingRow;
import org.apache.fluss.row.arrow.ArrowWriterPool;
-import org.apache.fluss.row.arrow.ArrowWriterProvider;
-import org.apache.fluss.row.encode.ValueDecoder;
import org.apache.fluss.rpc.protocol.MergeMode;
import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
import org.apache.fluss.server.kv.autoinc.AutoIncrementManager;
-import org.apache.fluss.server.kv.autoinc.AutoIncrementUpdater;
+import org.apache.fluss.server.kv.historical.HistoricalValueLookup;
import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer;
import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.PreparedFlush;
-import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.TruncateReason;
import org.apache.fluss.server.kv.rocksdb.RocksDBKv;
import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder;
import org.apache.fluss.server.kv.rocksdb.RocksDBResourceContainer;
import org.apache.fluss.server.kv.rocksdb.RocksDBStatistics;
-import org.apache.fluss.server.kv.rowmerger.DefaultRowMerger;
import org.apache.fluss.server.kv.rowmerger.RowMerger;
import org.apache.fluss.server.kv.scan.OpenScanResult;
import org.apache.fluss.server.kv.scan.ScannerContext;
@@ -66,18 +51,12 @@ import org.apache.fluss.server.kv.snapshot.KvFileHandleAndLocalPath;
import org.apache.fluss.server.kv.snapshot.KvSnapshotDataUploader;
import org.apache.fluss.server.kv.snapshot.RocksIncrementalSnapshot;
import org.apache.fluss.server.kv.snapshot.TabletState;
-import org.apache.fluss.server.kv.wal.ArrowWalBuilder;
-import org.apache.fluss.server.kv.wal.CompactedWalBuilder;
-import org.apache.fluss.server.kv.wal.IndexWalBuilder;
-import org.apache.fluss.server.kv.wal.WalBuilder;
import org.apache.fluss.server.log.LogAppendInfo;
import org.apache.fluss.server.log.LogTablet;
import org.apache.fluss.server.metrics.group.TabletServerMetricGroup;
import org.apache.fluss.server.utils.FatalErrorHandler;
import org.apache.fluss.server.utils.ResourceGuard;
import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.BufferAllocator;
-import org.apache.fluss.types.RowType;
-import org.apache.fluss.utils.BytesUtils;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.IOUtils;
@@ -102,6 +81,9 @@ import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import static org.apache.fluss.server.kv.KvStateAccessor.HISTORICAL_TOMBSTONE;
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
import static org.apache.fluss.utils.Preconditions.checkState;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
@@ -135,36 +117,24 @@ public final class KvTablet {
private final PhysicalTablePath physicalPath;
private final TableBucket tableBucket;
+ private final boolean historicalPartition;
private final LogTablet logTablet;
- private final ArrowWriterProvider arrowWriterProvider;
- private final MemorySegmentPool memorySegmentPool;
private final File kvTabletDir;
private final long writeBatchSize;
private final RocksDBKv rocksDBKv;
private final KvPreWriteBuffer kvPreWriteBuffer;
+ private final KvStateAccessor kvStateAccessor;
+ private final KvWriteProcessor kvWriteProcessor;
private final TabletServerMetricGroup serverMetricGroup;
private final KvFlushScheduler kvFlushScheduler;
private final boolean closeFlushScheduler;
// A lock that guards all modifications to the kv.
private final ReadWriteLock kvLock = new ReentrantReadWriteLock();
- private final LogFormat logFormat;
- private final KvFormat kvFormat;
- // defines how to merge rows on the same primary key
- private final RowMerger rowMerger;
- // Pre-created DefaultRowMerger for OVERWRITE mode (undo recovery scenarios)
- // This avoids creating a new instance on every putAsLeader call
- private final RowMerger overwriteRowMerger;
- private final ArrowCompressionInfo arrowCompressionInfo;
private final AutoIncrementManager autoIncrementManager;
- private final SchemaGetter schemaGetter;
-
- // the changelog image mode for this tablet
- private final ChangelogImage changelogImage;
-
// RocksDB statistics accessor for this tablet
@Nullable private final RocksDBStatistics rocksDBStatistics;
@@ -201,7 +171,6 @@ public final class KvTablet {
TabletServerMetricGroup serverMetricGroup,
RocksDBKv rocksDBKv,
long writeBatchSize,
- LogFormat logFormat,
BufferAllocator arrowBufferAllocator,
MemorySegmentPool memorySegmentPool,
KvFormat kvFormat,
@@ -216,6 +185,8 @@ public final class KvTablet {
AutoIncrementManager autoIncrementManager) {
this.physicalPath = physicalPath;
this.tableBucket = tableBucket;
+ this.historicalPartition =
+ HISTORICAL_PARTITION_VALUE.equals(physicalPath.getPartitionName());
this.logTablet = logTablet;
this.kvTabletDir = kvTabletDir;
this.rocksDBKv = rocksDBKv;
@@ -224,22 +195,30 @@ public final class KvTablet {
this.kvFlushScheduler = kvFlushScheduler;
this.closeFlushScheduler = closeFlushScheduler;
this.kvPreWriteBuffer = new KvPreWriteBuffer(serverMetricGroup);
- this.logFormat = logFormat;
- this.arrowWriterProvider = new ArrowWriterPool(arrowBufferAllocator);
- this.memorySegmentPool = memorySegmentPool;
- this.kvFormat = kvFormat;
- this.rowMerger = rowMerger;
- // Pre-create DefaultRowMerger for OVERWRITE mode to avoid creating new instances
- // on every putAsLeader call. Used for undo recovery scenarios.
- this.overwriteRowMerger = new DefaultRowMerger(kvFormat, DeleteBehavior.ALLOW);
- this.arrowCompressionInfo = arrowCompressionInfo;
- this.schemaGetter = schemaGetter;
- this.changelogImage = changelogImage;
+ this.kvStateAccessor =
+ new KvStateAccessor(kvPreWriteBuffer, rocksDBKv, historicalPartition);
+ this.kvWriteProcessor =
+ new KvWriteProcessor(
+ tableBucket,
+ logTablet,
+ new ArrowWriterPool(arrowBufferAllocator),
+ memorySegmentPool,
+ kvFormat,
+ rowMerger,
+ arrowCompressionInfo,
+ schemaGetter,
+ changelogImage,
+ autoIncrementManager);
this.rocksDBStatistics = rocksDBStatistics;
this.autoIncrementManager = autoIncrementManager;
this.flushCompleteListener = flushCompleteListener;
- // disable row count for WAL image mode.
- this.rowCount = changelogImage == ChangelogImage.WAL ? ROW_COUNT_DISABLED : 0L;
+ // TODO: Support row count for historical partitions.
+ // Historical state only contains the WAL tail that has not been tiered to lake, so it
+ // cannot maintain a table-level row count.
+ this.rowCount =
+ historicalPartition || changelogImage == ChangelogImage.WAL
+ ? ROW_COUNT_DISABLED
+ : 0L;
}
/**
@@ -371,7 +350,6 @@ public final class KvTablet {
serverMetricGroup,
kv,
serverConf.get(ConfigOptions.KV_WRITE_BATCH_SIZE).getBytes(),
- logTablet.getLogFormat(),
arrowBufferAllocator,
memorySegmentPool,
kvFormat,
@@ -428,6 +406,11 @@ public final class KvTablet {
return kvTabletDir;
}
+ /** Returns the total size in bytes of the live RocksDB SST files. */
+ public long liveSstFilesSize() {
+ return rocksDBKv.liveSstFilesSize();
+ }
+
/**
* Get RocksDB statistics accessor for this tablet.
*
@@ -514,6 +497,72 @@ public final class KvTablet {
public LogAppendInfo putAsLeader(
KvRecordBatch kvRecords, @Nullable int[] targetColumns, MergeMode mergeMode)
throws Exception {
+ checkState(
+ !historicalPartition,
+ "putAsLeader is not supported for historical KV tablet %s",
+ tableBucket);
+ return putAsLeader(kvRecords, targetColumns, mergeMode, null, null);
+ }
+
+ /**
+ * Puts records for one original partition into this historical KV tablet.
+ *
+ * <p>The original partition name namespaces the physical primary keys because one historical
+ * bucket can contain records from multiple original partitions. The supplied fallback may only
+ * read lake results already resolved for this request; it must not perform lake I/O while the
+ * tablet lock is held.
+ */
+ public LogAppendInfo putHistoricalAsLeader(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ String originalPartitionName,
+ HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ checkState(historicalPartition, "%s is not a historical KV tablet", tableBucket);
+ return putAsLeader(
+ kvRecords,
+ targetColumns,
+ mergeMode,
+ checkNotNull(originalPartitionName, "originalPartitionName must not be null"),
+ checkNotNull(memoizedLakeLookup, "memoizedLakeLookup must not be null"));
+ }
+
+ /**
+ * Finds keys whose historical write requires an old value that is absent from local state.
+ *
+ * <p>This method only reads KV entries and uses the tablet read lock. Lake I/O must be
+ * performed by the caller after this method releases the tablet lock.
+ */
+ public List<byte[]> findKeysRequiringLakeLookup(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ String originalPartitionName)
+ throws Exception {
+ checkState(historicalPartition, "%s is not a historical KV tablet", tableBucket);
+ return inReadLock(
+ kvLock,
+ () -> {
+ rocksDBKv.checkIfRocksDBClosed();
+ return kvWriteProcessor.findKeysRequiringLakeLookup(
+ kvRecords,
+ targetColumns,
+ mergeMode,
+ kvStateAccessor,
+ checkNotNull(
+ originalPartitionName,
+ "originalPartitionName must not be null"));
+ });
+ }
+
+ private LogAppendInfo putAsLeader(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ @Nullable String originalPartitionName,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
return inWriteLock(
kvLock,
() -> {
@@ -535,297 +584,16 @@ public final class KvTablet {
tableBucket));
}
- SchemaInfo schemaInfo = schemaGetter.getLatestSchemaInfo();
- Schema latestSchema = schemaInfo.getSchema();
- short latestSchemaId = (short) schemaInfo.getSchemaId();
- validateSchemaId(kvRecords.schemaId(), latestSchemaId);
-
- AutoIncrementUpdater currentAutoIncrementUpdater =
- autoIncrementManager.getUpdaterForSchema(kvFormat, latestSchemaId);
-
- // Validate targetColumns doesn't contain auto-increment column
- currentAutoIncrementUpdater.validateTargetColumns(targetColumns);
-
- // Determine the row merger based on mergeMode:
- // - DEFAULT: Use the configured merge engine (rowMerger)
- // - OVERWRITE: Bypass merge engine, use pre-created overwriteRowMerger
- // to directly replace values (for undo recovery scenarios)
- // We only support ADD COLUMN, so targetColumns is fine to be used directly.
- RowMerger currentMerger =
- (mergeMode == MergeMode.OVERWRITE)
- ? overwriteRowMerger.configureTargetColumns(
- targetColumns, latestSchemaId, latestSchema)
- : rowMerger.configureTargetColumns(
- targetColumns, latestSchemaId, latestSchema);
-
- RowType latestRowType = latestSchema.getRowType();
- WalBuilder walBuilder = createWalBuilder(latestSchemaId, latestRowType);
- walBuilder.setWriterState(kvRecords.writerId(), kvRecords.batchSequence());
- // we only support ADD COLUMN LAST, so the BinaryRow after RowMerger is
- // only has fewer ending columns than latest schema, so we pad nulls to
- // the end of the BinaryRow to get the latest schema row.
- PaddingRow latestSchemaRow = new PaddingRow(latestRowType.getFieldCount());
- // get offset to track the offset corresponded to the kv record
- long logEndOffsetOfPrevBatch = logTablet.localLogEndOffset();
-
- try {
- processKvRecords(
- kvRecords,
- kvRecords.schemaId(),
- currentMerger,
- currentAutoIncrementUpdater,
- walBuilder,
- latestSchemaRow,
- logEndOffsetOfPrevBatch);
-
- // There will be a situation that these batches of kvRecordBatch have not
- // generated any CDC logs, for example, when client attempts to delete
- // some non-existent keys or MergeEngineType set to FIRST_ROW. In this case,
- // we cannot simply return, as doing so would cause a
- // OutOfOrderSequenceException problem. Therefore, here we will build an
- // empty batch with lastLogOffset to 0L as the baseLogOffset is 0L. As doing
- // that, the logOffsetDelta in logRecordBatch will be set to 0L. So, we will
- // put a batch into file with recordCount 0 and offset plus 1L, it will
- // update the batchSequence corresponding to the writerId and also increment
- // the CDC log offset by 1.
- LogAppendInfo logAppendInfo = logTablet.appendAsLeader(walBuilder.build());
-
- // if the batch is duplicated, we should truncate the kvPreWriteBuffer
- // already written.
- if (logAppendInfo.duplicated()) {
- kvPreWriteBuffer.truncateTo(
- logEndOffsetOfPrevBatch, TruncateReason.DUPLICATED);
- }
- return logAppendInfo;
- } catch (Throwable t) {
- // While encounter error here, the CDC logs may fail writing to disk,
- // and the client probably will resend the batch. If we do not remove the
- // values generated by the erroneous batch from the kvPreWriteBuffer, the
- // retry-send batch will produce incorrect CDC logs.
- // TODO for some errors, the cdc logs may already be written to disk, for
- // those errors, we should not truncate the kvPreWriteBuffer.
- kvPreWriteBuffer.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.ERROR);
- throw t;
- } finally {
- // deallocate the memory and arrow writer used by the wal builder
- walBuilder.deallocate();
- }
+ return kvWriteProcessor.putAsLeader(
+ kvRecords,
+ targetColumns,
+ mergeMode,
+ kvStateAccessor,
+ originalPartitionName,
+ memoizedLakeLookup);
});
}
- private void validateSchemaId(short schemaIdOfNewData, short latestSchemaId) {
- if (schemaIdOfNewData > latestSchemaId || schemaIdOfNewData < 0) {
- throw new SchemaNotExistException(
- "Invalid schema id: "
- + schemaIdOfNewData
- + ", latest schema id: "
- + latestSchemaId);
- }
- }
-
- private void processKvRecords(
- KvRecordBatch kvRecords,
- short schemaIdOfNewData,
- RowMerger currentMerger,
- AutoIncrementUpdater autoIncrementUpdater,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long startLogOffset)
- throws Exception {
- long logOffset = startLogOffset;
-
- // TODO: reuse the read context and decoder
- KvRecordBatch.ReadContext readContext =
- KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
- ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, kvFormat);
-
- for (KvRecord kvRecord : kvRecords.records(readContext)) {
- byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
- KvPreWriteBuffer.Key key = KvPreWriteBuffer.Key.of(keyBytes);
- BinaryRow row = kvRecord.getRow();
- BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
-
- if (currentValue == null) {
- logOffset =
- processDeletion(
- key,
- currentMerger,
- valueDecoder,
- walBuilder,
- latestSchemaRow,
- logOffset);
- } else {
- logOffset =
- processUpsert(
- key,
- currentValue,
- currentMerger,
- autoIncrementUpdater,
- valueDecoder,
- walBuilder,
- latestSchemaRow,
- logOffset);
- }
- }
- }
-
- private long processDeletion(
- KvPreWriteBuffer.Key key,
- RowMerger currentMerger,
- ValueDecoder valueDecoder,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long logOffset)
- throws Exception {
- DeleteBehavior deleteBehavior = currentMerger.deleteBehavior();
- if (deleteBehavior == DeleteBehavior.IGNORE) {
- // skip delete rows if the merger doesn't support yet
- return logOffset;
- } else if (deleteBehavior == DeleteBehavior.DISABLE) {
- throw new DeletionDisabledException(
- "Delete operations are disabled for this table. "
- + "The table.delete.behavior is set to 'disable'.");
- }
-
- byte[] oldValueBytes = getFromBufferOrKv(key);
- if (oldValueBytes == null) {
- LOG.debug(
- "The specific key can't be found in kv tablet although the kv record is for deletion, "
- + "ignore it directly as it doesn't exist in the kv tablet yet.");
- return logOffset;
- }
-
- BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
- BinaryValue newValue = currentMerger.delete(oldValue);
-
- // if newValue is null, it means the row should be deleted
- if (newValue == null) {
- return applyDelete(key, oldValue, walBuilder, latestSchemaRow, logOffset);
- } else {
- return applyUpdate(key, oldValue, newValue, walBuilder, latestSchemaRow, logOffset);
- }
- }
-
- private long processUpsert(
- KvPreWriteBuffer.Key key,
- BinaryValue currentValue,
- RowMerger currentMerger,
- AutoIncrementUpdater autoIncrementUpdater,
- ValueDecoder valueDecoder,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long logOffset)
- throws Exception {
- // Optimization: IN WAL mode，when using DefaultRowMerger (full update, not partial update)
- // and there is no auto-increment column, we can skip fetching old value for better
- // performance since the result always reflects the new value. In this case, both INSERT and
- // UPDATE will produce UPDATE_AFTER.
- if (changelogImage == ChangelogImage.WAL
- && !autoIncrementUpdater.hasAutoIncrement()
- && currentMerger instanceof DefaultRowMerger) {
- return applyUpdate(key, null, currentValue, walBuilder, latestSchemaRow, logOffset);
- }
-
- byte[] oldValueBytes = getFromBufferOrKv(key);
- if (oldValueBytes == null) {
- BinaryValue valueToInsert = currentMerger.merge(null, currentValue);
- return applyInsert(
- key,
- valueToInsert,
- walBuilder,
- latestSchemaRow,
- logOffset,
- autoIncrementUpdater);
- }
-
- BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
- BinaryValue newValue = currentMerger.merge(oldValue, currentValue);
-
- if (newValue == oldValue) {
- // no actual change, skip this record
- return logOffset;
- }
-
- return applyUpdate(key, oldValue, newValue, walBuilder, latestSchemaRow, logOffset);
- }
-
- private long applyDelete(
- KvPreWriteBuffer.Key key,
- BinaryValue oldValue,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long logOffset)
- throws Exception {
- walBuilder.append(ChangeType.DELETE, latestSchemaRow.replaceRow(oldValue.row));
- kvPreWriteBuffer.delete(key, logOffset);
- return logOffset + 1;
- }
-
- private long applyInsert(
- KvPreWriteBuffer.Key key,
- BinaryValue currentValue,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long logOffset,
- AutoIncrementUpdater autoIncrementUpdater)
- throws Exception {
- BinaryValue newValue = autoIncrementUpdater.updateAutoIncrementColumns(currentValue);
- walBuilder.append(ChangeType.INSERT, latestSchemaRow.replaceRow(newValue.row));
- kvPreWriteBuffer.insert(key, newValue.encodeValue(), logOffset);
- return logOffset + 1;
- }
-
- private long applyUpdate(
- KvPreWriteBuffer.Key key,
- BinaryValue oldValue,
- BinaryValue newValue,
- WalBuilder walBuilder,
- PaddingRow latestSchemaRow,
- long logOffset)
- throws Exception {
- if (changelogImage == ChangelogImage.WAL) {
- walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
- kvPreWriteBuffer.update(key, newValue.encodeValue(), logOffset);
- return logOffset + 1;
- } else {
- walBuilder.append(ChangeType.UPDATE_BEFORE, latestSchemaRow.replaceRow(oldValue.row));
- walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
- kvPreWriteBuffer.update(key, newValue.encodeValue(), logOffset + 1);
- return logOffset + 2;
- }
- }
-
- private WalBuilder createWalBuilder(int schemaId, RowType rowType) throws Exception {
- switch (logFormat) {
- case INDEXED:
- if (kvFormat == KvFormat.COMPACTED) {
- // convert from compacted row to indexed row is time cost, and gain
- // less benefits, currently we won't support compacted as kv format and
- // indexed as cdc log format.
- // so in here we throw exception directly
- throw new IllegalArgumentException(
- "Primary Key Table with COMPACTED kv format doesn't support INDEXED cdc log format.");
- }
- return new IndexWalBuilder(schemaId, memorySegmentPool);
- case COMPACTED:
- return new CompactedWalBuilder(schemaId, rowType, memorySegmentPool);
- case ARROW:
- return new ArrowWalBuilder(
- schemaId,
- arrowWriterProvider.getOrCreateWriter(
- tableBucket.getTableId(),
- schemaId,
- // we don't limit size of the arrow batch, because all the
- // changelogs should be in a single batch
- Integer.MAX_VALUE,
- rowType,
- arrowCompressionInfo),
- memorySegmentPool);
- default:
- throw new IllegalArgumentException("Unsupported log format: " + logFormat);
- }
- }
-
@VisibleForTesting
long localLogEndOffset() {
return logTablet.localLogEndOffset();
@@ -988,7 +756,13 @@ public final class KvTablet {
for (KvPreWriteBuffer.KvEntry entry : segment.entries()) {
KvPreWriteBuffer.Value value = entry.getValue();
if (value.get() == null) {
- kvBatchWriter.delete(entry.getKey().get());
+ if (historicalPartition) {
+ // A physical delete would turn a local miss into a lake lookup and
+ // could expose the stale value that this mutation deleted.
+ kvBatchWriter.put(entry.getKey().get(), HISTORICAL_TOMBSTONE);
+ } else {
+ kvBatchWriter.delete(entry.getKey().get());
+ }
} else {
kvBatchWriter.put(entry.getKey().get(), value.get());
}
@@ -1126,11 +900,11 @@ public final class KvTablet {
ChangeType changeType, byte[] key, @Nullable byte[] value, long logOffset) {
KvPreWriteBuffer.Key wrapKey = KvPreWriteBuffer.Key.of(key);
if (changeType == ChangeType.DELETE && value == null) {
- kvPreWriteBuffer.delete(wrapKey, logOffset);
+ kvStateAccessor.delete(wrapKey, logOffset);
} else if (changeType == ChangeType.INSERT) {
- kvPreWriteBuffer.insert(wrapKey, value, logOffset);
+ kvStateAccessor.insert(wrapKey, value, logOffset);
} else if (changeType == ChangeType.UPDATE_AFTER) {
- kvPreWriteBuffer.update(wrapKey, value, logOffset);
+ kvStateAccessor.update(wrapKey, value, logOffset);
} else {
throw new IllegalArgumentException(
"Unsupported change type for putToPreWriteBuffer: " + changeType);
@@ -1148,15 +922,6 @@ public final class KvTablet {
return runnable -> inWriteLock(kvLock, runnable::run);
}
- // get from kv pre-write buffer first, if can't find, get from rocksdb
- private byte[] getFromBufferOrKv(KvPreWriteBuffer.Key key) throws IOException {
- KvPreWriteBuffer.Value value = kvPreWriteBuffer.get(key);
- if (value == null) {
- return rocksDBKv.get(key.get());
- }
- return value.get();
- }
-
public List<byte[]> multiGet(List<byte[]> keys) throws IOException {
return inReadLock(
kvLock,
@@ -1175,18 +940,45 @@ public final class KvTablet {
* #multiGet} so that clients only observe flushed data.
*/
public List<byte[]> multiGetFromBufferOrKv(List<byte[]> keys) throws IOException {
+ checkState(
+ !historicalPartition,
+ "multiGetFromBufferOrKv is not supported for historical KV tablet %s",
+ tableBucket);
return inReadLock(
kvLock,
() -> {
rocksDBKv.checkIfRocksDBClosed();
List<byte[]> values = new ArrayList<>(keys.size());
for (byte[] key : keys) {
- values.add(getFromBufferOrKv(KvPreWriteBuffer.Key.of(key)));
+ values.add(
+ kvStateAccessor
+ .lookup(kvStateAccessor.encodeKey(key, null))
+ .value());
}
return values;
});
}
+ /** Looks up one key from the flushed local state for an original historical partition. */
+ public KvStateLookupResult lookupHistoricalLocal(String originalPartitionName, byte[] key)
+ throws IOException {
+ checkState(historicalPartition, "%s is not a historical KV tablet", tableBucket);
+ return inReadLock(
+ kvLock,
+ () -> {
+ rocksDBKv.checkIfRocksDBClosed();
+ byte[] value =
+ rocksDBKv.get(
+ kvStateAccessor.encodeKey(key, originalPartitionName).get());
+ if (value == null) {
+ return KvStateLookupResult.notFound();
+ }
+ return value.length == 0
+ ? KvStateLookupResult.deleted()
+ : KvStateLookupResult.present(value);
+ });
+ }
+
public List<byte[]> prefixLookup(byte[] prefixKey) throws IOException {
return inReadLock(
kvLock,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java
new file mode 100644
index 000000000..76fc94e5c
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java
@@ -0,0 +1,564 @@
+/*
+ * 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.fluss.server.kv;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.compression.ArrowCompressionInfo;
+import org.apache.fluss.exception.DeletionDisabledException;
+import org.apache.fluss.exception.SchemaNotExistException;
+import org.apache.fluss.memory.MemorySegmentPool;
+import org.apache.fluss.metadata.ChangelogImage;
+import org.apache.fluss.metadata.DeleteBehavior;
+import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.metadata.LogFormat;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.BinaryValue;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.KvRecord;
+import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.record.KvRecordReadContext;
+import org.apache.fluss.row.BinaryRow;
+import org.apache.fluss.row.PaddingRow;
+import org.apache.fluss.row.arrow.ArrowWriterProvider;
+import org.apache.fluss.row.encode.ValueDecoder;
+import org.apache.fluss.rpc.protocol.MergeMode;
+import org.apache.fluss.server.kv.autoinc.AutoIncrementManager;
+import org.apache.fluss.server.kv.autoinc.AutoIncrementUpdater;
+import org.apache.fluss.server.kv.historical.HistoricalValueLookup;
+import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer;
+import org.apache.fluss.server.kv.prewrite.KvPreWriteBuffer.TruncateReason;
+import org.apache.fluss.server.kv.rowmerger.DefaultRowMerger;
+import org.apache.fluss.server.kv.rowmerger.RowMerger;
+import org.apache.fluss.server.kv.wal.ArrowWalBuilder;
+import org.apache.fluss.server.kv.wal.CompactedWalBuilder;
+import org.apache.fluss.server.kv.wal.IndexWalBuilder;
+import org.apache.fluss.server.kv.wal.WalBuilder;
+import org.apache.fluss.server.log.LogAppendInfo;
+import org.apache.fluss.server.log.LogTablet;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.ByteArrayWrapper;
+import org.apache.fluss.utils.BytesUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Processes a KV record batch into local state mutations and the corresponding WAL records.
+ *
+ * <p>For each input record, this processor reads the current value through {@link KvStateAccessor},
+ * applies the configured row-merge semantics, stages the resulting mutation in the state pre-write
+ * buffer, and appends the generated changelog to the {@link LogTablet}. Staged mutations are
+ * truncated when the append fails or is detected as a duplicate.
+ *
+ * <p>The supplied {@link KvStateAccessor} defines how keys and state are accessed. Normal writes
+ * use the original primary key and local state, while historical writes use partition-scoped keys.
+ * On a historical local miss, the processor can consult a lake result already memoized for the
+ * current request. Resolving that result from lake storage remains the caller's responsibility and
+ * must happen outside the tablet lock. The merge and WAL generation path is shared by both write
+ * kinds.
+ */
+@Internal
+@NotThreadSafe
+public final class KvWriteProcessor {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KvWriteProcessor.class);
+
+ private final TableBucket tableBucket;
+ private final LogTablet logTablet;
+ private final ArrowWriterProvider arrowWriterProvider;
+ private final MemorySegmentPool memorySegmentPool;
+ private final LogFormat logFormat;
+ private final KvFormat kvFormat;
+ // defines how to merge rows on the same primary key
+ private final RowMerger rowMerger;
+ // Pre-created DefaultRowMerger for OVERWRITE mode (undo recovery scenarios)
+ // This avoids creating a new instance on every putAsLeader call
+ private final RowMerger overwriteRowMerger;
+ private final ArrowCompressionInfo arrowCompressionInfo;
+ private final SchemaGetter schemaGetter;
+ // the changelog image mode for this tablet
+ private final ChangelogImage changelogImage;
+ private final AutoIncrementManager autoIncrementManager;
+
+ /** Creates a KV write processor. */
+ public KvWriteProcessor(
+ TableBucket tableBucket,
+ LogTablet logTablet,
+ ArrowWriterProvider arrowWriterProvider,
+ MemorySegmentPool memorySegmentPool,
+ KvFormat kvFormat,
+ RowMerger rowMerger,
+ ArrowCompressionInfo arrowCompressionInfo,
+ SchemaGetter schemaGetter,
+ ChangelogImage changelogImage,
+ AutoIncrementManager autoIncrementManager) {
+ this.tableBucket = tableBucket;
+ this.logTablet = logTablet;
+ this.arrowWriterProvider = arrowWriterProvider;
+ this.memorySegmentPool = memorySegmentPool;
+ this.logFormat = logTablet.getLogFormat();
+ this.kvFormat = kvFormat;
+ this.rowMerger = rowMerger;
+ // Pre-create DefaultRowMerger for OVERWRITE mode to avoid creating new instances
+ // on every putAsLeader call. Used for undo recovery scenarios.
+ this.overwriteRowMerger = new DefaultRowMerger(kvFormat, DeleteBehavior.ALLOW);
+ this.arrowCompressionInfo = arrowCompressionInfo;
+ this.schemaGetter = schemaGetter;
+ this.changelogImage = changelogImage;
+ this.autoIncrementManager = autoIncrementManager;
+ }
+
+ /** Processes a KV batch against the supplied state and appends its WAL. */
+ public LogAppendInfo putAsLeader(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ KvStateAccessor stateAccessor,
+ @Nullable String originalPartitionName,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ WriteContext writeContext = createWriteContext(kvRecords, targetColumns, mergeMode);
+ RowType latestRowType = writeContext.latestSchema.getRowType();
+ WalBuilder walBuilder = createWalBuilder(writeContext.latestSchemaId, latestRowType);
+ walBuilder.setWriterState(kvRecords.writerId(), kvRecords.batchSequence());
+ // we only support ADD COLUMN LAST, so the BinaryRow after RowMerger is
+ // only has fewer ending columns than latest schema, so we pad nulls to
+ // the end of the BinaryRow to get the latest schema row.
+ PaddingRow latestSchemaRow = new PaddingRow(latestRowType.getFieldCount());
+ // get offset to track the offset corresponded to the kv record
+ long logEndOffsetOfPrevBatch = logTablet.localLogEndOffset();
+
+ try {
+ processKvRecords(
+ kvRecords,
+ kvRecords.schemaId(),
+ writeContext.rowMerger,
+ writeContext.autoIncrementUpdater,
+ walBuilder,
+ latestSchemaRow,
+ logEndOffsetOfPrevBatch,
+ stateAccessor,
+ originalPartitionName,
+ memoizedLakeLookup);
+
+ // There will be a situation that these batches of kvRecordBatch have not
+ // generated any CDC logs, for example, when client attempts to delete
+ // some non-existent keys or MergeEngineType set to FIRST_ROW. In this case,
+ // we cannot simply return, as doing so would cause a
+ // OutOfOrderSequenceException problem. Therefore, here we will build an
+ // empty batch with lastLogOffset to 0L as the baseLogOffset is 0L. As doing
+ // that, the logOffsetDelta in logRecordBatch will be set to 0L. So, we will
+ // put a batch into file with recordCount 0 and offset plus 1L, it will
+ // update the batchSequence corresponding to the writerId and also increment
+ // the CDC log offset by 1.
+ LogAppendInfo logAppendInfo = logTablet.appendAsLeader(walBuilder.build());
+
+ // if the batch is duplicated, we should truncate the state pre-write
+ // buffer already written.
+ if (logAppendInfo.duplicated()) {
+ stateAccessor.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.DUPLICATED);
+ }
+ return logAppendInfo;
+ } catch (Throwable t) {
+ // While encounter error here, the CDC logs may fail writing to disk,
+ // and the client probably will resend the batch. If we do not remove the
+ // values generated by the erroneous batch from the state pre-write buffer,
+ // the retry-send batch will produce incorrect CDC logs.
+ // TODO for some errors, the cdc logs may already be written to disk, for
+ // those errors, we should not truncate the state pre-write buffer.
+ stateAccessor.truncateTo(logEndOffsetOfPrevBatch, TruncateReason.ERROR);
+ throw t;
+ } finally {
+ // deallocate the memory and arrow writer used by the wal builder
+ walBuilder.deallocate();
+ }
+ }
+
+ /**
+ * Finds the original primary keys whose previous values must be loaded from lake storage before
+ * applying a historical write batch.
+ *
+ * <p>A key is returned only when its previous value is required and its partition-scoped key is
+ * absent from local state. Records that can establish their result without a previous value are
+ * skipped, and each key is returned at most once.
+ */
+ List<byte[]> findKeysRequiringLakeLookup(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ KvStateAccessor stateAccessor,
+ String originalPartitionName)
+ throws Exception {
+ WriteContext writeContext = createWriteContext(kvRecords, targetColumns, mergeMode);
+
+ List<byte[]> keysRequiringLakeLookup = new ArrayList<>();
+ // Track keys whose lake-lookup requirement has already been evaluated.
+ Set<ByteArrayWrapper> keysEvaluatedForLakeLookup = new HashSet<>();
+ KvRecordBatch.ReadContext readContext =
+ KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ for (KvRecord kvRecord : kvRecords.records(readContext)) {
+ byte[] primaryKey = BytesUtils.toArray(kvRecord.getKey());
+ ByteArrayWrapper wrappedKey = new ByteArrayWrapper(primaryKey);
+ if (kvRecord.getRow() == null) {
+ if (shouldIgnoreDeletion(writeContext.rowMerger)) {
+ continue;
+ }
+ } else if (canSkipOldValueLookup(
+ writeContext.rowMerger, writeContext.autoIncrementUpdater)) {
+ // A full-row WAL upsert establishes the state without reading its previous value.
+ keysEvaluatedForLakeLookup.add(wrappedKey);
+ continue;
+ }
+
+ // Probe local state only for the first record that needs a previous value. A true local
+ // miss schedules one lake lookup shared by all records for this key in the batch.
+ if (keysEvaluatedForLakeLookup.add(wrappedKey)
+ && stateAccessor
+ .lookup(
+ stateAccessor.encodeKey(
+ primaryKey, originalPartitionName))
+ .status()
+ == KvStateLookupResult.Status.NOT_FOUND) {
+ keysRequiringLakeLookup.add(primaryKey);
+ }
+ }
+ return keysRequiringLakeLookup;
+ }
+
+ private WriteContext createWriteContext(
+ KvRecordBatch kvRecords, @Nullable int[] targetColumns, MergeMode mergeMode) {
+ SchemaInfo schemaInfo = schemaGetter.getLatestSchemaInfo();
+ Schema latestSchema = schemaInfo.getSchema();
+ short latestSchemaId = (short) schemaInfo.getSchemaId();
+ validateSchemaId(kvRecords.schemaId(), latestSchemaId);
+
+ AutoIncrementUpdater autoIncrementUpdater =
+ autoIncrementManager.getUpdaterForSchema(kvFormat, latestSchemaId);
+ autoIncrementUpdater.validateTargetColumns(targetColumns);
+
+ // OVERWRITE bypasses the configured merge engine for undo recovery.
+ RowMerger configuredRowMerger =
+ (mergeMode == MergeMode.OVERWRITE)
+ ? overwriteRowMerger.configureTargetColumns(
+ targetColumns, latestSchemaId, latestSchema)
+ : rowMerger.configureTargetColumns(
+ targetColumns, latestSchemaId, latestSchema);
+ return new WriteContext(
+ latestSchema, latestSchemaId, autoIncrementUpdater, configuredRowMerger);
+ }
+
+ private void validateSchemaId(short schemaIdOfNewData, short latestSchemaId) {
+ if (schemaIdOfNewData > latestSchemaId || schemaIdOfNewData < 0) {
+ throw new SchemaNotExistException(
+ "Invalid schema id: "
+ + schemaIdOfNewData
+ + ", latest schema id: "
+ + latestSchemaId);
+ }
+ }
+
+ private void processKvRecords(
+ KvRecordBatch kvRecords,
+ short schemaIdOfNewData,
+ RowMerger currentMerger,
+ AutoIncrementUpdater autoIncrementUpdater,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long startLogOffset,
+ KvStateAccessor stateAccessor,
+ @Nullable String originalPartitionName,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ long logOffset = startLogOffset;
+
+ // TODO: reuse the read context and decoder
+ KvRecordBatch.ReadContext readContext =
+ KvRecordReadContext.createReadContext(kvFormat, schemaGetter);
+ ValueDecoder valueDecoder = new ValueDecoder(schemaGetter, kvFormat);
+
+ for (KvRecord kvRecord : kvRecords.records(readContext)) {
+ byte[] keyBytes = BytesUtils.toArray(kvRecord.getKey());
+ KvPreWriteBuffer.Key key = stateAccessor.encodeKey(keyBytes, originalPartitionName);
+ BinaryRow row = kvRecord.getRow();
+ BinaryValue currentValue = row == null ? null : new BinaryValue(schemaIdOfNewData, row);
+
+ if (currentValue == null) {
+ logOffset =
+ processDeletion(
+ key,
+ currentMerger,
+ valueDecoder,
+ walBuilder,
+ latestSchemaRow,
+ logOffset,
+ stateAccessor,
+ keyBytes,
+ memoizedLakeLookup);
+ } else {
+ logOffset =
+ processUpsert(
+ key,
+ currentValue,
+ currentMerger,
+ autoIncrementUpdater,
+ valueDecoder,
+ walBuilder,
+ latestSchemaRow,
+ logOffset,
+ stateAccessor,
+ keyBytes,
+ memoizedLakeLookup);
+ }
+ }
+ }
+
+ private long processDeletion(
+ KvPreWriteBuffer.Key key,
+ RowMerger currentMerger,
+ ValueDecoder valueDecoder,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long logOffset,
+ KvStateAccessor stateAccessor,
+ byte[] primaryKey,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ if (shouldIgnoreDeletion(currentMerger)) {
+ return logOffset;
+ }
+
+ byte[] oldValueBytes = getFromState(key, primaryKey, stateAccessor, memoizedLakeLookup);
+ if (oldValueBytes == null) {
+ LOG.debug(
+ "The specific key can't be found in kv tablet although the kv record is for deletion, "
+ + "ignore it directly as it doesn't exist in the kv tablet yet.");
+ return logOffset;
+ }
+
+ BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
+ BinaryValue newValue = currentMerger.delete(oldValue);
+
+ // if newValue is null, it means the row should be deleted
+ if (newValue == null) {
+ return applyDelete(
+ key, oldValue, walBuilder, latestSchemaRow, logOffset, stateAccessor);
+ } else {
+ return applyUpdate(
+ key, oldValue, newValue, walBuilder, latestSchemaRow, logOffset, stateAccessor);
+ }
+ }
+
+ private long processUpsert(
+ KvPreWriteBuffer.Key key,
+ BinaryValue currentValue,
+ RowMerger currentMerger,
+ AutoIncrementUpdater autoIncrementUpdater,
+ ValueDecoder valueDecoder,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long logOffset,
+ KvStateAccessor stateAccessor,
+ byte[] primaryKey,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ if (canSkipOldValueLookup(currentMerger, autoIncrementUpdater)) {
+ return applyUpdate(
+ key, null, currentValue, walBuilder, latestSchemaRow, logOffset, stateAccessor);
+ }
+
+ byte[] oldValueBytes = getFromState(key, primaryKey, stateAccessor, memoizedLakeLookup);
+ if (oldValueBytes == null) {
+ BinaryValue valueToInsert = currentMerger.merge(null, currentValue);
+ return applyInsert(
+ key,
+ valueToInsert,
+ walBuilder,
+ latestSchemaRow,
+ logOffset,
+ autoIncrementUpdater,
+ stateAccessor);
+ }
+
+ BinaryValue oldValue = valueDecoder.decodeValue(oldValueBytes);
+ BinaryValue newValue = currentMerger.merge(oldValue, currentValue);
+
+ if (newValue == oldValue) {
+ // no actual change, skip this record
+ return logOffset;
+ }
+
+ return applyUpdate(
+ key, oldValue, newValue, walBuilder, latestSchemaRow, logOffset, stateAccessor);
+ }
+
+ private long applyDelete(
+ KvPreWriteBuffer.Key key,
+ BinaryValue oldValue,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long logOffset,
+ KvStateAccessor stateAccessor)
+ throws Exception {
+ walBuilder.append(ChangeType.DELETE, latestSchemaRow.replaceRow(oldValue.row));
+ stateAccessor.delete(key, logOffset);
+ return logOffset + 1;
+ }
+
+ private long applyInsert(
+ KvPreWriteBuffer.Key key,
+ BinaryValue currentValue,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long logOffset,
+ AutoIncrementUpdater autoIncrementUpdater,
+ KvStateAccessor stateAccessor)
+ throws Exception {
+ BinaryValue newValue = autoIncrementUpdater.updateAutoIncrementColumns(currentValue);
+ walBuilder.append(ChangeType.INSERT, latestSchemaRow.replaceRow(newValue.row));
+ stateAccessor.insert(key, newValue.encodeValue(), logOffset);
+ return logOffset + 1;
+ }
+
+ private long applyUpdate(
+ KvPreWriteBuffer.Key key,
+ @Nullable BinaryValue oldValue,
+ BinaryValue newValue,
+ WalBuilder walBuilder,
+ PaddingRow latestSchemaRow,
+ long logOffset,
+ KvStateAccessor stateAccessor)
+ throws Exception {
+ if (changelogImage == ChangelogImage.WAL) {
+ walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
+ stateAccessor.update(key, newValue.encodeValue(), logOffset);
+ return logOffset + 1;
+ } else {
+ walBuilder.append(ChangeType.UPDATE_BEFORE, latestSchemaRow.replaceRow(oldValue.row));
+ walBuilder.append(ChangeType.UPDATE_AFTER, latestSchemaRow.replaceRow(newValue.row));
+ stateAccessor.update(key, newValue.encodeValue(), logOffset + 1);
+ return logOffset + 2;
+ }
+ }
+
+ /**
+ * Returns the previous value from local state, falling back to the memoized lake result only on
+ * a genuine local miss.
+ *
+ * @param localStateKey the key used by the local prewrite buffer and RocksDB; it wraps {@code
+ * primaryKey} for a normal partition and adds the original partition namespace for a
+ * historical partition
+ * @param primaryKey the encoded bytes of the logical primary key, without the historical
+ * partition namespace; used by the lake lookup
+ * @return the previous value, or null if the key is absent or locally marked as deleted
+ */
+ private byte[] getFromState(
+ KvPreWriteBuffer.Key localStateKey,
+ byte[] primaryKey,
+ KvStateAccessor stateAccessor,
+ @Nullable HistoricalValueLookup memoizedLakeLookup)
+ throws Exception {
+ KvStateLookupResult localResult = stateAccessor.lookup(localStateKey);
+ if (localResult.status() != KvStateLookupResult.Status.NOT_FOUND
+ || memoizedLakeLookup == null) {
+ return localResult.value();
+ }
+ return memoizedLakeLookup.lookup(primaryKey);
+ }
+
+ private boolean canSkipOldValueLookup(
+ RowMerger currentMerger, AutoIncrementUpdater autoIncrementUpdater) {
+ // A full-row WAL upsert does not need the old value: its result is always the incoming row,
+ // and WAL mode emits only UPDATE_AFTER. Partial updates and auto-increment writes still
+ // need the old value to construct the final row or distinguish an insert from an update.
+ return changelogImage == ChangelogImage.WAL
+ && !autoIncrementUpdater.hasAutoIncrement()
+ && currentMerger instanceof DefaultRowMerger;
+ }
+
+ private static boolean shouldIgnoreDeletion(RowMerger currentMerger) {
+ DeleteBehavior deleteBehavior = currentMerger.deleteBehavior();
+ if (deleteBehavior == DeleteBehavior.IGNORE) {
+ return true;
+ } else if (deleteBehavior == DeleteBehavior.DISABLE) {
+ throw new DeletionDisabledException(
+ "Delete operations are disabled for this table. "
+ + "The table.delete.behavior is set to 'disable'.");
+ }
+ return false;
+ }
+
+ private WalBuilder createWalBuilder(int schemaId, RowType rowType) throws Exception {
+ switch (logFormat) {
+ case INDEXED:
+ if (kvFormat == KvFormat.COMPACTED) {
+ // convert from compacted row to indexed row is time cost, and gain
+ // less benefits, currently we won't support compacted as kv format and
+ // indexed as cdc log format.
+ // so in here we throw exception directly
+ throw new IllegalArgumentException(
+ "Primary Key Table with COMPACTED kv format doesn't support INDEXED cdc log format.");
+ }
+ return new IndexWalBuilder(schemaId, memorySegmentPool);
+ case COMPACTED:
+ return new CompactedWalBuilder(schemaId, rowType, memorySegmentPool);
+ case ARROW:
+ return new ArrowWalBuilder(
+ schemaId,
+ arrowWriterProvider.getOrCreateWriter(
+ tableBucket.getTableId(),
+ schemaId,
+ // we don't limit size of the arrow batch, because all the
+ // changelogs should be in a single batch
+ Integer.MAX_VALUE,
+ rowType,
+ arrowCompressionInfo),
+ memorySegmentPool);
+ default:
+ throw new IllegalArgumentException("Unsupported log format: " + logFormat);
+ }
+ }
+
+ private static final class WriteContext {
+ private final Schema latestSchema;
+ private final short latestSchemaId;
+ private final AutoIncrementUpdater autoIncrementUpdater;
+ private final RowMerger rowMerger;
+
+ private WriteContext(
+ Schema latestSchema,
+ short latestSchemaId,
+ AutoIncrementUpdater autoIncrementUpdater,
+ RowMerger rowMerger) {
+ this.latestSchema = latestSchema;
+ this.latestSchemaId = latestSchemaId;
+ this.autoIncrementUpdater = autoIncrementUpdater;
+ this.rowMerger = rowMerger;
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoder.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoder.java
new file mode 100644
index 000000000..b52e93588
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoder.java
@@ -0,0 +1,79 @@
+/*
+ * 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.fluss.server.kv.historical;
+
+import org.apache.fluss.annotation.Internal;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Encoder for keys stored in historical KV state.
+ *
+ * <p>A historical bucket can store local state for multiple original partitions. Prefixing each
+ * physical primary key with its original partition name preserves the partition identity in the
+ * shared RocksDB key space.
+ */
+@Internal
+public final class HistoricalKvKeyEncoder {
+
+ private HistoricalKvKeyEncoder() {}
+
+ /**
+ * Encodes an original partition name and primary key into one unambiguous storage key.
+ *
+ * <p>The key layout is a four-byte big-endian UTF-8 partition-name length, followed by the
+ * partition-name bytes and original primary-key bytes.
+ */
+ public static byte[] encode(String originalPartitionName, byte[] originalPrimaryKey) {
+ checkNotNull(originalPartitionName, "originalPartitionName must not be null");
+ checkArgument(!originalPartitionName.isEmpty(), "originalPartitionName must not be empty");
+ checkNotNull(originalPrimaryKey, "originalPrimaryKey must not be null");
+
+ byte[] partitionNameBytes = originalPartitionName.getBytes(StandardCharsets.UTF_8);
+ long encodedLength =
+ Integer.BYTES + (long) partitionNameBytes.length + originalPrimaryKey.length;
+ checkArgument(encodedLength <= Integer.MAX_VALUE, "The encoded historical key is too long");
+
+ return ByteBuffer.allocate((int) encodedLength)
+ .putInt(partitionNameBytes.length)
+ .put(partitionNameBytes)
+ .put(originalPrimaryKey)
+ .array();
+ }
+
+ /** Returns a copy of the original primary key from an encoded historical KV key. */
+ public static byte[] extractOriginalPrimaryKey(byte[] encodedPrimaryKey) {
+ checkNotNull(encodedPrimaryKey, "encodedPrimaryKey must not be null");
+ checkArgument(
+ encodedPrimaryKey.length >= Integer.BYTES,
+ "The encoded historical key is shorter than its partition-name length prefix");
+
+ int partitionNameLength = ByteBuffer.wrap(encodedPrimaryKey).getInt();
+ checkArgument(
+ partitionNameLength >= 0
+ && partitionNameLength <= encodedPrimaryKey.length - Integer.BYTES,
+ "The encoded historical key contains an invalid partition-name length");
+ int primaryKeyOffset = Integer.BYTES + partitionNameLength;
+ return Arrays.copyOfRange(encodedPrimaryKey, primaryKeyOffset, encodedPrimaryKey.length);
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalValueLookup.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalValueLookup.java
new file mode 100644
index 000000000..bdc723df9
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/historical/HistoricalValueLookup.java
@@ -0,0 +1,37 @@
+/*
+ * 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.fluss.server.kv.historical;
+
+import org.apache.fluss.annotation.Internal;
+
+import javax.annotation.Nullable;
+
+/** Resolves a lake value already memoized for the current historical write request. */
+@Internal
+@FunctionalInterface
+public interface HistoricalValueLookup {
+
+ /**
+ * Returns the encoded value for the primary key, or null when it does not exist.
+ *
+ * <p>This method is invoked while the KV write lock is held and must not perform lake or file
+ * I/O.
+ */
+ @Nullable
+ byte[] lookup(byte[] primaryKey);
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKv.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKv.java
index f672240b9..ce8ef903c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKv.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBKv.java
@@ -57,6 +57,9 @@ public class RocksDBKv implements AutoCloseable {
/** RocksDB property name for the number of SST files at level 0 (column-family scoped). */
private static final String NUM_FILES_AT_LEVEL0 = "rocksdb.num-files-at-level0";
+ /** RocksDB property name for the total size of SST files in the current LSM tree. */
+ private static final String LIVE_SST_FILES_SIZE = "rocksdb.live-sst-files-size";
+
private static final float DELAYED_WRITE_PRESSURE_FLOOR = 0.8f;
private static final long L0_REFRESH_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(100);
@@ -270,6 +273,11 @@ public class RocksDBKv implements AutoCloseable {
}
}
+ /** Returns the total size in bytes of the live SST files. */
+ public long liveSstFilesSize() {
+ return readLongProperty(LIVE_SST_FILES_SIZE);
+ }
+
@Override
public void close() throws Exception {
close(KvCloseMode.PRESERVE_LOCAL_STATE);
@@ -406,21 +414,7 @@ public class RocksDBKv implements AutoCloseable {
}
private long sampleL0FileCount() {
- if (closed) {
- return 0L;
- }
- ResourceGuard.Lease lease = null;
- try {
- lease = rocksDBResourceGuard.acquireResource();
- return readLongProperty(NUM_FILES_AT_LEVEL0);
- } catch (IOException acquireFailed) {
- // RocksDB is already closed (or being closed); treat as no pressure.
- return 0L;
- } finally {
- if (lease != null) {
- lease.close();
- }
- }
+ return readLongProperty(NUM_FILES_AT_LEVEL0);
}
/**
@@ -443,14 +437,20 @@ public class RocksDBKv implements AutoCloseable {
}
private long readLongProperty(String propertyName) {
- try {
+ if (closed) {
+ return 0L;
+ }
+ try (ResourceGuard.Lease ignored = rocksDBResourceGuard.acquireResource()) {
String value = db.getProperty(defaultColumnFamilyHandle, propertyName);
if (value == null || value.isEmpty()) {
return 0L;
}
return Long.parseLong(value);
+ } catch (IOException ignored) {
+ // RocksDB is already closed or being closed.
+ return 0L;
} catch (RocksDBException | NumberFormatException e) {
- LOG.warn("Failed to query RocksDB property {} for backpressure", propertyName, e);
+ LOG.warn("Failed to query RocksDB property {}", propertyName, e);
return 0L;
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java
index 303ed5805..02be54ed9 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/metrics/group/TableMetricGroup.java
@@ -221,6 +221,24 @@ public class TableMetricGroup extends AbstractMetricGroup {
}
}
+ /** Returns the counter for historical put-KV requests received by this table. */
+ public Counter totalHistoricalPutKvRequests() {
+ if (kvMetrics == null) {
+ return NoOpCounter.INSTANCE;
+ } else {
+ return kvMetrics.totalHistoricalPutKvRequests;
+ }
+ }
+
+ /** Returns the counter for failed historical put-KV requests for this table. */
+ public Counter failedHistoricalPutKvRequests() {
+ if (kvMetrics == null) {
+ return NoOpCounter.INSTANCE;
+ } else {
+ return kvMetrics.failedHistoricalPutKvRequests;
+ }
+ }
+
/**
* Records a historical lake table point lookup.
*
@@ -578,6 +596,8 @@ public class TableMetricGroup extends AbstractMetricGroup {
private final Counter failedLookupRequests;
private final Counter totalHistoricalLookupRequests;
private final Counter failedHistoricalLookupRequests;
+ private final Counter totalHistoricalPutKvRequests;
+ private final Counter failedHistoricalPutKvRequests;
private final LookupFileDownloadedMetricGroup downloadedHistoricalLookupMetrics;
private final LookupFileDownloadedMetricGroup nonDownloadedHistoricalLookupMetrics;
private final Counter totalPutKvRequests;
@@ -596,15 +616,24 @@ public class TableMetricGroup extends AbstractMetricGroup {
failedLookupRequests = new ThreadSafeSimpleCounter();
meter(MetricNames.FAILED_LOOKUP_REQUESTS_RATE, new MeterView(failedLookupRequests));
// for historical lookup request
- MetricGroup historicalLookupMetrics = addGroup("historical");
+ MetricGroup historicalMetrics = addGroup("historical");
totalHistoricalLookupRequests = new ThreadSafeSimpleCounter();
- historicalLookupMetrics.meter(
+ historicalMetrics.meter(
MetricNames.TOTAL_LOOKUP_REQUESTS_RATE,
new MeterView(totalHistoricalLookupRequests));
failedHistoricalLookupRequests = new ThreadSafeSimpleCounter();
- historicalLookupMetrics.meter(
+ historicalMetrics.meter(
MetricNames.FAILED_LOOKUP_REQUESTS_RATE,
new MeterView(failedHistoricalLookupRequests));
+ // for historical put kv request
+ totalHistoricalPutKvRequests = new ThreadSafeSimpleCounter();
+ historicalMetrics.meter(
+ MetricNames.TOTAL_PUT_KV_REQUESTS_RATE,
+ new MeterView(totalHistoricalPutKvRequests));
+ failedHistoricalPutKvRequests = new ThreadSafeSimpleCounter();
+ historicalMetrics.meter(
+ MetricNames.FAILED_PUT_KV_REQUESTS_RATE,
+ new MeterView(failedHistoricalPutKvRequests));
// Separate groups expose the same metric names with different downloaded-file labels
// without adding the label key to the logical metric scope.
downloadedHistoricalLookupMetrics =
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index 0d30b8381..5e6036849 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -24,6 +24,7 @@ import org.apache.fluss.config.ConfigurationUtils;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.FencedLeaderEpochException;
import org.apache.fluss.exception.InvalidColumnProjectionException;
+import org.apache.fluss.exception.InvalidPartitionException;
import org.apache.fluss.exception.InvalidTableException;
import org.apache.fluss.exception.InvalidTimestampException;
import org.apache.fluss.exception.InvalidUpdateVersionException;
@@ -39,6 +40,7 @@ import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
@@ -59,9 +61,11 @@ import org.apache.fluss.server.coordinator.CoordinatorContext;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.kv.KvManager;
import org.apache.fluss.server.kv.KvRecoverHelper;
+import org.apache.fluss.server.kv.KvStateLookupResult;
import org.apache.fluss.server.kv.KvTablet;
import org.apache.fluss.server.kv.RemoteLogFetcher;
import org.apache.fluss.server.kv.autoinc.AutoIncIDRange;
+import org.apache.fluss.server.kv.historical.HistoricalValueLookup;
import org.apache.fluss.server.kv.rocksdb.RocksDBKvBuilder;
import org.apache.fluss.server.kv.scan.OpenScanResult;
import org.apache.fluss.server.kv.scan.ScannerContext;
@@ -141,7 +145,9 @@ import java.util.function.IntSupplier;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
+import static org.apache.fluss.utils.Preconditions.checkState;
import static org.apache.fluss.utils.concurrent.LockUtils.inReadLock;
import static org.apache.fluss.utils.concurrent.LockUtils.inWriteLock;
@@ -186,6 +192,7 @@ public final class Replica {
private final SchemaGetter schemaGetter;
private volatile TableInfo tableInfo;
+ private final boolean historicalPartition;
// logFormat and arrowCompressionInfo are immutable and used in hot-path, so cache them here.
private final LogFormat logFormat;
private final ArrowCompressionInfo arrowCompressionInfo;
@@ -272,6 +279,8 @@ public final class Replica {
tableInfo.getSchema());
this.tableInfo = tableInfo;
TableConfig tableConfig = tableInfo.getTableConfig();
+ String partitionName = physicalPath.getPartitionName();
+ this.historicalPartition = HISTORICAL_PARTITION_VALUE.equals(partitionName);
this.logFormat = tableConfig.getLogFormat();
this.arrowCompressionInfo = tableConfig.getArrowCompressionInfo();
this.snapshotContext = snapshotContext;
@@ -316,6 +325,12 @@ public final class Replica {
public long logicalStorageKvSize() {
if (isLeader() && isKvTable()) {
+ if (isHistoricalPartition()) {
+ // Historical KV tablets do not create snapshots, so account for the local overlay
+ // using live SST files instead.
+ KvTablet currentKvTablet = kvTablet;
+ return currentKvTablet == null ? 0L : currentKvTablet.liveSstFilesSize();
+ }
checkNotNull(kvSnapshotManager, "kvSnapshotManager is null");
return kvSnapshotManager.getSnapshotSize();
} else {
@@ -396,6 +411,19 @@ public final class Replica {
return kvTablet;
}
+ SchemaGetter schemaGetter() {
+ return schemaGetter;
+ }
+
+ /** Returns the latest schema used by historical partition operations. */
+ public SchemaInfo getLatestSchemaInfo() {
+ return schemaGetter.getLatestSchemaInfo();
+ }
+
+ boolean isHistoricalPartition() {
+ return historicalPartition;
+ }
+
public TablePath getTablePath() {
return physicalPath.getTablePath();
}
@@ -723,8 +751,14 @@ public final class Replica {
e);
}
}
- // start periodic kv snapshot
- startPeriodicKvSnapshot(snapshotUsed.orElse(null));
+ // A historical KV tablet is a disposable overlay over the lake snapshot. It is recovered
+ // by replaying WAL from the lake log end offset and does not create its own KV snapshots.
+ if (isHistoricalPartition()) {
+ // TODO: Clean up historical KV state after the corresponding WAL is fully tiered to
+ // lake storage.
+ } else {
+ startPeriodicKvSnapshot(snapshotUsed.orElse(null));
+ }
}
private void dropKv() {
@@ -797,8 +831,11 @@ public final class Replica {
// when replica become follower, we'll always delete the kv files.
// get the offset from which, we should restore from. default is 0
- long restoreStartOffset = 0;
- Optional<CompletedSnapshot> optCompletedSnapshot = getLatestSnapshot(tableBucket);
+ long restoreStartOffset = isHistoricalPartition() ? historicalRecoveryStartOffset() : 0;
+ // The lake snapshot is the durable base for a historical overlay. Historical replicas
+ // therefore never restore a normal KV snapshot, even if one exists from older code.
+ Optional<CompletedSnapshot> optCompletedSnapshot =
+ isHistoricalPartition() ? Optional.empty() : getLatestSnapshot(tableBucket);
try {
Long rowCount;
AutoIncIDRange autoIncIDRange;
@@ -844,7 +881,11 @@ public final class Replica {
this::onKvFlushComplete);
// we don't support rowCount
- rowCount = tableConfig.getChangelogImage() == ChangelogImage.WAL ? null : 0L;
+ rowCount =
+ isHistoricalPartition()
+ || tableConfig.getChangelogImage() == ChangelogImage.WAL
+ ? null
+ : 0L;
// TODO: it is possible that this is a recovered kv tablet without kv snapshot but
// with changelogs, in this case, the kv tablet should also have the
// autoIncIDRange, we may need to get it from the changelog in the future.
@@ -977,7 +1018,8 @@ public final class Replica {
tableConfig.getKvFormat(),
tableConfig.getLogFormat(),
schemaGetter,
- remoteLogFetcher);
+ remoteLogFetcher,
+ isHistoricalPartition());
kvRecoverHelper.recover();
} finally {
remoteLogFetcher.close();
@@ -998,6 +1040,19 @@ public final class Replica {
end - start);
}
+ private long historicalRecoveryStartOffset() {
+ long lakeLogEndOffset = logTablet.getLakeLogEndOffset();
+ long logStartOffset = logTablet.logStartOffset();
+ long recoveryStartOffset = lakeLogEndOffset >= 0 ? lakeLogEndOffset : 0L;
+ checkState(
+ recoveryStartOffset >= logStartOffset,
+ "Cannot recover historical KV state: recovery start offset %s is before the "
+ + "available log start offset %s.",
+ recoveryStartOffset,
+ logStartOffset);
+ return recoveryStartOffset;
+ }
+
private void startPeriodicKvSnapshot(@Nullable CompletedSnapshot completedSnapshot) {
checkNotNull(kvTablet);
KvTabletSnapshotTarget kvTabletSnapshotTarget;
@@ -1104,6 +1159,10 @@ public final class Replica {
"Leader not local for bucket %s on tabletServer %d",
tableBucket, localTabletServerId));
}
+ if (isHistoricalPartition()) {
+ throw new InvalidPartitionException(
+ "Normal write request must not target a historical partition.");
+ }
validateInSyncReplicaSize(requiredAcks);
@@ -1164,6 +1223,10 @@ public final class Replica {
"Leader not local for bucket %s on tabletServer %d",
tableBucket, localTabletServerId));
}
+ if (isHistoricalPartition()) {
+ throw new InvalidPartitionException(
+ "Normal write request must not target a historical partition.");
+ }
validateInSyncReplicaSize(requiredAcks);
KvTablet kv = this.kvTablet;
@@ -1184,6 +1247,107 @@ public final class Replica {
});
}
+ /**
+ * Finds historical write keys that require lake fallback without mutating local KV state.
+ *
+ * <p>The caller must keep historical writes for this table bucket ordered until the subsequent
+ * {@link #putHistoricalRecordsToLeader} call completes.
+ */
+ public List<byte[]> findKeysRequiringLakeLookup(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ String originalPartitionName,
+ int expectedLeaderEpoch,
+ int requiredAcks)
+ throws Exception {
+ return inReadLock(
+ leaderIsrUpdateLock,
+ () -> {
+ validateHistoricalWrite(expectedLeaderEpoch, requiredAcks);
+ KvTablet kv = this.kvTablet;
+ checkNotNull(kv, "KvTablet for the historical replica shouldn't be null.");
+ return kv.findKeysRequiringLakeLookup(
+ kvRecords, targetColumns, mergeMode, originalPartitionName);
+ });
+ }
+
+ /** Writes records to the local historical KV overlay of the leader replica. */
+ public LogAppendInfo putHistoricalRecordsToLeader(
+ KvRecordBatch kvRecords,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ String originalPartitionName,
+ HistoricalValueLookup memoizedLakeLookup,
+ int expectedLeaderEpoch,
+ int requiredAcks)
+ throws Exception {
+ return inReadLock(
+ leaderIsrUpdateLock,
+ () -> {
+ validateHistoricalWrite(expectedLeaderEpoch, requiredAcks);
+ KvTablet kv = this.kvTablet;
+ checkNotNull(kv, "KvTablet for the historical replica shouldn't be null.");
+ LogAppendInfo appendInfo =
+ kv.putHistoricalAsLeader(
+ kvRecords,
+ targetColumns,
+ mergeMode,
+ originalPartitionName,
+ memoizedLakeLookup);
+ maybeIncrementLeaderHW(logTablet, clock.milliseconds());
+ return appendInfo;
+ });
+ }
+
+ private void validateHistoricalWrite(int expectedLeaderEpoch, int requiredAcks) {
+ if (!isLeader()) {
+ throw new NotLeaderOrFollowerException(
+ String.format(
+ "Leader not local for bucket %s on tabletServer %d",
+ tableBucket, localTabletServerId));
+ }
+ if (!isHistoricalPartition()) {
+ throw new InvalidPartitionException(
+ "Historical write request must target a historical partition.");
+ }
+ if (leaderEpoch != expectedLeaderEpoch) {
+ throw new FencedLeaderEpochException(
+ String.format(
+ "Historical write for %s was prepared at leader epoch %s, "
+ + "but the current leader epoch is %s.",
+ tableBucket, expectedLeaderEpoch, leaderEpoch));
+ }
+ validateInSyncReplicaSize(requiredAcks);
+ }
+
+ /** Looks up keys from the local historical KV overlay of the leader replica. */
+ public List<KvStateLookupResult> lookupHistoricalLocal(
+ String originalPartitionName, List<byte[]> keys) throws Exception {
+ return inReadLock(
+ leaderIsrUpdateLock,
+ () -> {
+ if (!isLeader()) {
+ throw new NotLeaderOrFollowerException(
+ String.format(
+ "Leader not local for bucket %s on tabletServer %d",
+ tableBucket, localTabletServerId));
+ }
+ if (!isHistoricalPartition()) {
+ throw new InvalidPartitionException(
+ "Historical lookup request must target a historical partition.");
+ }
+
+ KvTablet kv = this.kvTablet;
+ checkNotNull(kv, "KvTablet for the historical replica shouldn't be null.");
+ List<KvStateLookupResult> results = new ArrayList<>(keys.size());
+ for (byte[] key : keys) {
+ results.add(kv.lookupHistoricalLocal(originalPartitionName, key));
+ }
+ return results;
+ });
+ }
+
public LogReadInfo fetchRecords(FetchParams fetchParams) throws IOException {
if (fetchParams.projection() != null && logFormat != LogFormat.ARROW) {
throw new InvalidColumnProjectionException(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index db8b2e8ed..ef3b16b2c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -42,13 +42,13 @@ import org.apache.fluss.fs.FsPath;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.Schema;
-import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.metrics.MetricNames;
import org.apache.fluss.metrics.groups.MetricGroup;
import org.apache.fluss.plugin.PluginManager;
+import org.apache.fluss.record.DefaultKvRecordBatch;
import org.apache.fluss.record.KeyRecordBatch;
import org.apache.fluss.record.KvRecordBatch;
import org.apache.fluss.record.MemoryLogRecords;
@@ -82,6 +82,7 @@ import org.apache.fluss.server.entity.NotifyLakeTableOffsetData;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrResultForBucket;
import org.apache.fluss.server.entity.NotifyRemoteLogOffsetsData;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
import org.apache.fluss.server.entity.StopReplicaData;
import org.apache.fluss.server.entity.StopReplicaResultForBucket;
import org.apache.fluss.server.entity.UserContext;
@@ -114,6 +115,7 @@ import org.apache.fluss.server.replica.delay.DelayedTableBucketKey;
import org.apache.fluss.server.replica.delay.DelayedWrite;
import org.apache.fluss.server.replica.fetcher.InitialFetchStatus;
import org.apache.fluss.server.replica.fetcher.ReplicaFetcherManager;
+import org.apache.fluss.server.replica.historical.HistoricalPartitionManager;
import org.apache.fluss.server.storage.DiskUsageMonitor;
import org.apache.fluss.server.storage.LocalDiskManager;
import org.apache.fluss.server.utils.FatalErrorHandler;
@@ -138,6 +140,7 @@ import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -158,7 +161,6 @@ import java.util.stream.Stream;
import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2;
import static org.apache.fluss.server.TabletManagerBase.getTableInfo;
-import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
import static org.apache.fluss.utils.Preconditions.checkArgument;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
import static org.apache.fluss.utils.Preconditions.checkState;
@@ -237,7 +239,7 @@ public class ReplicaManager implements ServerReconfigurable {
private final ScannerManager scannerManager;
- private final HistoricalLakeLookupManager historicalLakeLookupManager;
+ private final HistoricalPartitionManager historicalPartitionManager;
public ReplicaManager(
Configuration conf,
@@ -363,8 +365,8 @@ public class ReplicaManager implements ServerReconfigurable {
// Historical lookup cache capacity currently uses only the first data volume.
File dataDir = localDiskManager.dataDirs().get(0);
long dataDirVolumeBytes = Files.getFileStore(dataDir.toPath()).getTotalSpace();
- this.historicalLakeLookupManager =
- new HistoricalLakeLookupManager(
+ this.historicalPartitionManager =
+ new HistoricalPartitionManager(
conf,
pluginManager,
localDiskManager,
@@ -376,7 +378,7 @@ public class ReplicaManager implements ServerReconfigurable {
}
public void startup() {
- historicalLakeLookupManager.startup(scheduler);
+ historicalPartitionManager.startup(scheduler);
// start up ISR expiration thread.
// A follower can log behind leader for up tp configOptions#LOG_REPLICA_MAX_LAG_TIME x 1.5
@@ -427,7 +429,7 @@ public class ReplicaManager implements ServerReconfigurable {
@Override
public void reconfigure(Configuration newConfig) {
- historicalLakeLookupManager.reconfigure(newConfig);
+ historicalPartitionManager.reconfigure(newConfig);
int newMinInSyncReplicas =
newConfig.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER);
if (newMinInSyncReplicas == minInSyncReplicas) {
@@ -448,16 +450,16 @@ public class ReplicaManager implements ServerReconfigurable {
MetricGroup historicalMetrics = serverMetricGroup.addGroup("historical");
historicalMetrics.gauge(
MetricNames.HISTORICAL_INFLIGHT_REQUESTS,
- historicalLakeLookupManager::numInflightRequests);
+ historicalPartitionManager::numInflightRequests);
historicalMetrics.gauge(
MetricNames.HISTORICAL_LOOKUP_CACHE_DISK_SIZE,
- historicalLakeLookupManager::lookupCacheDiskSize);
+ historicalPartitionManager::lookupCacheDiskSize);
historicalMetrics.gauge(
MetricNames.HISTORICAL_LOOKUP_CACHE_TABLE_COUNT,
- historicalLakeLookupManager::cachedTableCount);
+ historicalPartitionManager::cachedTableCount);
historicalMetrics.counter(
MetricNames.HISTORICAL_LOOKUP_CACHE_CAPACITY_EVICTIONS,
- historicalLakeLookupManager.capacityEvictions());
+ historicalPartitionManager.capacityEvictions());
serverMetricGroup.gauge(
MetricNames.REPLICA_LEADER_COUNT,
@@ -617,7 +619,7 @@ public class ReplicaManager implements ServerReconfigurable {
ioExecutor.execute(
() ->
deletedTableIds.forEach(
- historicalLakeLookupManager
+ historicalPartitionManager
::invalidateTableLookuper));
}
});
@@ -742,6 +744,118 @@ public class ReplicaManager implements ServerReconfigurable {
timeoutMs, requiredAcks, entriesPerBucket.size(), kvPutResult, responseCallback);
}
+ /** Puts records to historical partition leaders. */
+ public void putHistoricalRecordsToKv(
+ int timeoutMs,
+ int requiredAcks,
+ Collection<PutKvDataForBucket> entriesPerBucket,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ short apiVersion,
+ Consumer<List<PutKvResultForBucket>> responseCallback) {
+ if (isRequiredAcksInvalid(requiredAcks)) {
+ throw new InvalidRequiredAcksException("Invalid required acks: " + requiredAcks);
+ }
+ localDiskManager.ensureWritable();
+
+ if (entriesPerBucket.isEmpty()) {
+ responseCallback.accept(Collections.emptyList());
+ return;
+ }
+
+ List<PutKvResultForBucket> results = Collections.synchronizedList(new ArrayList<>());
+ AtomicInteger remaining = new AtomicInteger(entriesPerBucket.size());
+ entriesPerBucket.forEach(
+ putData ->
+ historicalPutKv(putData, targetColumns, mergeMode, requiredAcks, apiVersion)
+ .whenComplete(
+ (result, error) -> {
+ PutKvResultForBucket completedResult =
+ error == null
+ ? result
+ : PutKvResultForBucket
+ .historicalFailure(
+ putData.tableBucket(),
+ ApiError.fromThrowable(
+ error),
+ putData
+ .originalPartitionName());
+ maybeAddDelayedWrite(
+ timeoutMs,
+ requiredAcks,
+ 1,
+ Collections.singletonMap(
+ putData.tableBucket(), completedResult),
+ delayedResult -> {
+ results.add(delayedResult.get(0));
+ if (remaining.decrementAndGet() == 0) {
+ responseCallback.accept(
+ new ArrayList<>(results));
+ }
+ });
+ }));
+ }
+
+ private CompletableFuture<PutKvResultForBucket> historicalPutKv(
+ PutKvDataForBucket putData,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ int requiredAcks,
+ short apiVersion) {
+ TableMetricGroup tableMetrics = null;
+ try {
+ Replica replica = getReplicaOrException(putData.tableBucket());
+ tableMetrics = replica.tableMetrics();
+ tableMetrics.totalHistoricalPutKvRequests().inc();
+ if (!replica.isKvTable()) {
+ throw new NonPrimaryKeyTableException(
+ "Historical writes are only supported for primary key tables.");
+ }
+ if (!replica.isHistoricalPartition()) {
+ throw new InvalidPartitionException(
+ "Historical write request must target a historical partition.");
+ }
+ validateClientVersionForPkTable(apiVersion, replica.getTableInfo());
+
+ KvRecordBatch records = putData.records();
+ checkArgument(
+ records instanceof DefaultKvRecordBatch,
+ "Historical RPC write requires DefaultKvRecordBatch, but found %s.",
+ records.getClass().getName());
+ PutKvDataForBucket copiedData =
+ new PutKvDataForBucket(
+ putData.tableBucket(),
+ copyToHeap((DefaultKvRecordBatch) records),
+ putData.originalPartitionName());
+ TableMetricGroup historicalPutMetrics = tableMetrics;
+ return historicalPartitionManager
+ .putKv(replica, copiedData, targetColumns, mergeMode, requiredAcks)
+ .thenApply(
+ result -> {
+ if (result.failed()
+ && isUnexpectedHistoricalException(
+ result.getError().exception())) {
+ historicalPutMetrics.failedHistoricalPutKvRequests().inc();
+ }
+ return result;
+ });
+ } catch (Throwable t) {
+ ApiError error = ApiError.fromThrowable(t);
+ if (tableMetrics != null && isUnexpectedHistoricalException(error.exception())) {
+ tableMetrics.failedHistoricalPutKvRequests().inc();
+ }
+ return CompletableFuture.completedFuture(
+ PutKvResultForBucket.historicalFailure(
+ putData.tableBucket(), error, putData.originalPartitionName()));
+ }
+ }
+
+ private static DefaultKvRecordBatch copyToHeap(DefaultKvRecordBatch records) {
+ byte[] bytes = new byte[records.sizeInBytes()];
+ records.getMemorySegment().get(records.getPosition(), bytes);
+ return DefaultKvRecordBatch.pointToBytes(bytes);
+ }
+
/** Context for tracking missing keys that need to be inserted. */
public static class MissingKeysContext {
final List<Integer> missingIndexes;
@@ -840,17 +954,13 @@ public class ReplicaManager implements ServerReconfigurable {
+ replica.getTablePath()
+ " is not a primary key table.");
}
- if (!isHistoricalPartitionReplica(replica)) {
+ if (!replica.isHistoricalPartition()) {
throw new InvalidPartitionException(
"Historical lookup request must target a historical partition.");
}
- SchemaInfo latestSchemaInfo = replica.getSchemaGetter().getLatestSchemaInfo();
lookupFuture =
- historicalLakeLookupManager.lookup(
- data,
- replica.getTableInfo(),
- latestSchemaInfo,
- replica.tableMetrics()::recordHistoricalLakeLookup);
+ historicalPartitionManager.lookup(
+ replica, data, replica.tableMetrics()::recordHistoricalLakeLookup);
} catch (Exception e) {
result.add(
new LookupResultForBucket(
@@ -874,7 +984,7 @@ public class ReplicaManager implements ServerReconfigurable {
data.originalPartitionName(),
ApiError.fromThrowable(error));
if (completedResult.failed()
- && isUnexpectedHistoricalLookupException(
+ && isUnexpectedHistoricalException(
completedResult.getError().exception())) {
replica.tableMetrics().failedHistoricalLookupRequests().inc();
}
@@ -975,12 +1085,6 @@ public class ReplicaManager implements ServerReconfigurable {
LOG.debug("Lookup from local kv in {}ms", System.currentTimeMillis() - startTime);
}
- private boolean isHistoricalPartitionReplica(Replica replica) {
- String partitionName = replica.getPhysicalTablePath().getPartitionName();
- return replica.getTableInfo().getPartitionKeys().size() == 1
- && HISTORICAL_PARTITION_VALUE.equals(partitionName);
- }
-
/**
* Re-lookup missing keys after insert and merge results back into the original lookup result
* map.
@@ -1149,7 +1253,7 @@ public class ReplicaManager implements ServerReconfigurable {
try {
boolean deletingHistoricalPartition =
data.isDeleteRemote()
- && isHistoricalPartitionReplica(replica);
+ && replica.isHistoricalPartition();
result.add(
stopReplica(
tb,
@@ -1181,7 +1285,7 @@ public class ReplicaManager implements ServerReconfigurable {
});
deletedHistoricalPartitionTableIds.forEach(
- historicalLakeLookupManager::invalidateTableLookuper);
+ historicalPartitionManager::invalidateTableLookuper);
responseCallback.accept(result);
}
@@ -1291,10 +1395,13 @@ public class ReplicaManager implements ServerReconfigurable {
// register replica to remote log manager first.
remoteLogManager.registerReplica(replica);
- replica.makeLeader(data);
+ // Load the latest lake progress before leader activation. Historical KV recovery
+ // requires its lake log end offset, while failures remain best effort for normal
+ // replicas.
if (replica.isDataLakeEnabled()) {
updateWithLakeTableSnapshot(replica);
}
+ replica.makeLeader(data);
// start the remote log tiering tasks for leaders
remoteLogManager.startLogTiering(replica);
@@ -1308,7 +1415,7 @@ public class ReplicaManager implements ServerReconfigurable {
}
// NOTE: This method can be removed when fetchFromLake is deprecated
- private void updateWithLakeTableSnapshot(Replica replica) {
+ private void updateWithLakeTableSnapshot(Replica replica) throws Exception {
TableBucket tb = replica.getTableBucket();
try {
Optional<LakeTableSnapshot> optLakeTableSnapshot =
@@ -1317,14 +1424,27 @@ public class ReplicaManager implements ServerReconfigurable {
LakeTableSnapshot lakeTableSnapshot = optLakeTableSnapshot.get();
long snapshotId = optLakeTableSnapshot.get().getSnapshotId();
replica.getLogTablet().updateLakeTableSnapshotId(snapshotId);
+ if (replica.isHistoricalPartition()) {
+ // The historical overlay will be rebuilt from this snapshot's lake offset.
+ // Refresh a cached lookuper before it becomes the fallback for data omitted
+ // from the rebuilt overlay.
+ historicalPartitionManager.requireLakeSnapshot(
+ replica.getTableBucket().getTableId(), snapshotId);
+ }
lakeTableSnapshot
.getLogEndOffset(tb)
.ifPresent(replica.getLogTablet()::updateLakeLogEndOffset);
}
} catch (Exception e) {
+ if (replica.isHistoricalPartition()) {
+ // Historical recovery uses the lake offset as its durable base and replays the
+ // retained WAL from that offset. Reject leader activation if the latest lake
+ // progress cannot be loaded, instead of rebuilding the overlay from stale state.
+ throw e;
+ }
// Lake commit cleanup can race with this best-effort refresh and remove the
// referenced offsets file. A failure only leaves the snapshot/offset state stale
- // until the next synchronization, so it must not fail the leader transition.
+ // until the next synchronization, so it must not fail a normal leader transition.
LOG.warn("Failed to update replica {} with lake table snapshot.", tb, e);
}
}
@@ -1838,7 +1958,7 @@ public class ReplicaManager implements ServerReconfigurable {
|| e instanceof StorageBackpressureException);
}
- private boolean isUnexpectedHistoricalLookupException(Exception e) {
+ private boolean isUnexpectedHistoricalException(Exception e) {
return isUnexpectedException(e)
&& !(e instanceof HistoricalPartitionThrottledException
|| e instanceof InvalidPartitionException
@@ -2430,7 +2550,7 @@ public class ReplicaManager implements ServerReconfigurable {
public void shutdown() throws InterruptedException {
// Close the resources for snapshot kv
kvSnapshotResource.close();
- historicalLakeLookupManager.close();
+ historicalPartitionManager.close();
replicaFetcherManager.shutdown();
delayedWriteManager.shutdown();
delayedFetchLogManager.shutdown();
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManager.java
similarity index 67%
rename from fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java
rename to fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManager.java
index 977b60840..af86581e4 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManager.java
@@ -15,15 +15,13 @@
* limitations under the License.
*/
-package org.apache.fluss.server.replica;
+package org.apache.fluss.server.replica.historical;
import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.TableConfig;
import org.apache.fluss.exception.FlussRuntimeException;
-import org.apache.fluss.exception.HistoricalPartitionThrottledException;
-import org.apache.fluss.exception.InvalidPartitionException;
import org.apache.fluss.exception.LakeStorageNotConfiguredException;
import org.apache.fluss.lake.lakestorage.LakeStorage;
import org.apache.fluss.lake.lakestorage.LakeStoragePlugin;
@@ -38,15 +36,11 @@ import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.metrics.Counter;
import org.apache.fluss.metrics.ThreadSafeSimpleCounter;
import org.apache.fluss.plugin.PluginManager;
-import org.apache.fluss.rpc.entity.LookupResultForBucket;
-import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.server.entity.LookupDataForBucket;
import org.apache.fluss.server.storage.LocalDiskManager;
-import org.apache.fluss.utils.ExecutorUtils;
import org.apache.fluss.utils.FileUtils;
import org.apache.fluss.utils.FlussPaths;
import org.apache.fluss.utils.IOUtils;
-import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
import org.apache.fluss.utils.concurrent.Scheduler;
import com.github.benmanes.caffeine.cache.Cache;
@@ -68,13 +62,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@@ -86,10 +75,6 @@ import static org.apache.fluss.utils.Preconditions.checkState;
/**
* Handles server-side point lookup for historical partitions stored in lake storage.
*
- * <p>Accepted requests run on a dedicated executor whose threads are started lazily and released
- * when idle. A semaphore bounds the total number of accepted historical lookup tasks so slow lake
- * storage cannot create an unbounded request backlog.
- *
* <p>Creating a lake table lookuper may initialize catalog, table, and query state and allocate
* local lookup files, so lookupers are cached and reused. The cache is keyed by table ID rather
* than table path to prevent a deleted and recreated table from reusing the old table's lookuper. A
@@ -118,10 +103,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
private static final String LOOKUP_CACHE_DISK_SIZE_TASK_NAME =
"historical-lookup-cache-disk-size";
private static final Duration LOOKUP_CACHE_DISK_SIZE_CHECK_INTERVAL = Duration.ofMinutes(3);
- private static final Duration HISTORICAL_PARTITION_THREAD_KEEP_ALIVE = Duration.ofMinutes(10);
- private static final Duration HISTORICAL_PARTITION_EXECUTOR_SHUTDOWN_TIMEOUT =
- Duration.ofSeconds(10);
- private static final String HISTORICAL_PARTITION_THREAD_NAME_PREFIX = "historical-partition-io";
// TODO: Share one Paimon IOManager disk budget across all table lookupers and evict cached
// entries by data file instead of reserving fixed per-table capacity. See
// https://github.com/apache/fluss/issues/3955.
@@ -131,12 +112,8 @@ class HistoricalLakeLookupManager implements AutoCloseable {
private volatile long lakeConfigVersion;
private final @Nullable PluginManager pluginManager;
private final Counter capacityEvictions;
- private final int maxQueuedHistoricalRequests;
- private final Semaphore lookupPermits;
- // Accepted lookup futures tracked so close() can cancel tasks left after executor shutdown.
- private final Set<CompletableFuture<LookupResultForBucket>> pendingLookups;
private final Cache<Long, CachedLakeTableLookuper> lakeTableLookupers;
- private final ExecutorService historicalPartitionExecutor;
+ private final ConcurrentMap<Long, Long> requiredLakeSnapshotIds = new ConcurrentHashMap<>();
private final File historicalLookupCacheRootDir;
private final long dataDirVolumeBytes;
// TODO: Introduce a minimum lookup cache disk ratio (default 0.01). When disk usage is high,
@@ -160,7 +137,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
this(
conf,
pluginManager,
- null,
dataDir,
dataDirVolumeBytes,
Ticker.systemTicker(),
@@ -173,7 +149,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
HistoricalLakeLookupManager(
Configuration conf,
@Nullable PluginManager pluginManager,
- @Nullable ExecutorService historicalPartitionExecutor,
File dataDir,
long dataDirVolumeBytes,
Ticker ticker,
@@ -193,22 +168,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
ConfigOptions
.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO));
this.capacityEvictions = new ThreadSafeSimpleCounter();
- this.maxQueuedHistoricalRequests =
- conf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
- checkArgument(
- maxQueuedHistoricalRequests > 0,
- "%s must be greater than 0.",
- ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
- int maxThreadPoolSize =
- conf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE);
- checkArgument(
- maxThreadPoolSize > 0,
- "%s must be greater than 0.",
- ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
- this.historicalPartitionExecutor =
- historicalPartitionExecutor == null
- ? createHistoricalPartitionExecutor(maxThreadPoolSize)
- : historicalPartitionExecutor;
this.lakeTableLookupers =
Caffeine.newBuilder()
.maximumSize(MAX_CACHED_TABLES)
@@ -221,8 +180,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
.executor(Runnable::run)
.removalListener(this::onLookuperRemoved)
.build();
- this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests);
- this.pendingLookups = ConcurrentHashMap.newKeySet();
}
private static com.github.benmanes.caffeine.cache.Scheduler createCacheScheduler(
@@ -273,92 +230,53 @@ class HistoricalLakeLookupManager implements AutoCloseable {
}
/** Looks up a batch of keys from one historical lake partition. */
- CompletableFuture<LookupResultForBucket> lookup(
+ List<byte[]> lookup(
LookupDataForBucket lookupData,
TableInfo tableInfo,
SchemaInfo schemaInfo,
- LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
+ ResolvedPartitionSpec originalPartitionSpec,
+ LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder)
+ throws Exception {
+ LakeTableLookuper.LookupMetricRecorder checkedMetricRecorder =
+ checkNotNull(lookupMetricRecorder, "lookupMetricRecorder must not be null.");
checkState(started, "Historical lake lookup manager has not been started.");
- TableBucket tableBucket = lookupData.tableBucket();
- if (!lookupPermits.tryAcquire()) {
- return CompletableFuture.completedFuture(
- new LookupResultForBucket(
- tableBucket,
- null,
- lookupData.originalPartitionName(),
- ApiError.fromThrowable(
- new HistoricalPartitionThrottledException(
- "Historical lookup is throttled for "
- + tableBucket
- + "."))));
- }
-
- CompletableFuture<LookupResultForBucket> future;
+ LookupContext context =
+ createLookupContext(
+ lookupData,
+ tableInfo,
+ schemaInfo,
+ originalPartitionSpec,
+ checkedMetricRecorder);
+ CachedLakeTableLookuper cachedLookuper = acquireLookuper(context, tableInfo);
try {
- future =
- submitLookup(
- lookupData,
- tableInfo,
- schemaInfo,
- checkNotNull(
- lookupMetricRecorder,
- "lookupMetricRecorder must not be null."));
- } catch (RuntimeException e) {
- lookupPermits.release();
- throw e;
+ List<byte[]> values = new ArrayList<>(lookupData.keys().size());
+ for (byte[] key : lookupData.keys()) {
+ values.add(cachedLookuper.lookuper.lookup(key, context.lookupContext));
+ }
+ return values;
+ } finally {
+ cachedLookuper.release();
}
- future.whenComplete(
- (ignored, error) -> {
- pendingLookups.remove(future);
- lookupPermits.release();
- });
- return future;
}
@Override
public void close() {
- ExecutorUtils.gracefulShutdown(
- HISTORICAL_PARTITION_EXECUTOR_SHUTDOWN_TIMEOUT.toMillis(),
- TimeUnit.MILLISECONDS,
- historicalPartitionExecutor);
- pendingLookups.forEach(future -> future.cancel(true));
lakeTableLookupers.invalidateAll();
lakeTableLookupers.cleanUp();
- }
-
- private CompletableFuture<LookupResultForBucket> submitLookup(
- LookupDataForBucket lookupData,
- TableInfo tableInfo,
- SchemaInfo schemaInfo,
- LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
- CompletableFuture<LookupResultForBucket> future =
- CompletableFuture.supplyAsync(
- () ->
- lookupInternal(
- lookupData, tableInfo, schemaInfo, lookupMetricRecorder),
- historicalPartitionExecutor);
- pendingLookups.add(future);
- return future;
- }
-
- private ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) {
- ThreadPoolExecutor executor =
- new ThreadPoolExecutor(
- maxThreadPoolSize,
- maxThreadPoolSize,
- HISTORICAL_PARTITION_THREAD_KEEP_ALIVE.toMillis(),
- TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<>(),
- new ExecutorThreadFactory(HISTORICAL_PARTITION_THREAD_NAME_PREFIX));
- executor.allowCoreThreadTimeOut(true);
- return executor;
+ requiredLakeSnapshotIds.clear();
}
/** Invalidates the cached lake lookuper for the given table. */
void invalidateTableLookuper(long tableId) {
+ requiredLakeSnapshotIds.remove(tableId);
lakeTableLookupers.invalidate(tableId);
}
+ /** Records the required opaque lake snapshot ID, which is compared only by equality. */
+ void requireLakeSnapshot(long tableId, long snapshotId) {
+ requiredLakeSnapshotIds.put(tableId, snapshotId);
+ }
+
/** Returns the number of table lookupers currently cached. */
int cachedTableCount() {
return lakeTableLookupers.asMap().size();
@@ -369,11 +287,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
return capacityEvictions;
}
- /** Returns the number of accepted historical lookup requests that have not completed. */
- int numInflightRequests() {
- return maxQueuedHistoricalRequests - lookupPermits.availablePermits();
- }
-
/** Applies dynamic historical lookup configuration changes. */
void reconfigure(Configuration newConf) {
checkNotNull(newConf, "newConf must not be null.");
@@ -424,83 +337,6 @@ class HistoricalLakeLookupManager implements AutoCloseable {
}
}
- private LookupResultForBucket lookupInternal(
- LookupDataForBucket lookupData,
- TableInfo tableInfo,
- SchemaInfo schemaInfo,
- LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
- TableBucket tableBucket = lookupData.tableBucket();
- CachedLakeTableLookuper cachedLookuper = null;
- try {
- LookupContext context =
- createLookupContext(lookupData, tableInfo, schemaInfo, lookupMetricRecorder);
- long currentLakeConfigVersion = lakeConfigVersion;
- Configuration currentConf = conf;
- long cacheSizeBytes = lookupCacheMaxDiskBytesPerTable;
- cachedLookuper =
- lakeTableLookupers
- .asMap()
- .compute(
- context.tableId,
- (ignored, currentLookuper) -> {
- CachedLakeTableLookuper selectedLookuper = currentLookuper;
- // Create the lookuper lazily, and recreate it after schema,
- // lake configuration, or server cache size changes so it
- // reloads lake table/query state and uses the current
- // settings.
- if (selectedLookuper == null
- || selectedLookuper.schemaId != context.schemaId
- || selectedLookuper.lakeConfigVersion
- != currentLakeConfigVersion
- || selectedLookuper.cacheSizeBytes
- != cacheSizeBytes) {
- File tableLookupDir =
- FlussPaths.historicalLookupTableDir(
- historicalLookupCacheRootDir,
- context.tablePath,
- context.tableId);
- LakeTableLookuper lookuper =
- createLakeTableLookuper(
- context.tablePath,
- tableLookupDir.getAbsolutePath(),
- tableInfo.getTableConfig(),
- cacheSizeBytes,
- currentConf);
- selectedLookuper =
- new CachedLakeTableLookuper(
- context.tableId,
- context.tablePath,
- context.schemaId,
- currentLakeConfigVersion,
- cacheSizeBytes,
- tableLookupDir,
- lookuper);
- }
- // Pin the lookuper before leaving the atomic cache update.
- // Eviction or invalidation can then defer closing it until
- // this lookup releases it.
- selectedLookuper.acquire();
- return selectedLookuper;
- });
- List<byte[]> values = new ArrayList<>(lookupData.keys().size());
- for (byte[] key : lookupData.keys()) {
- values.add(cachedLookuper.lookuper.lookup(key, context.lookupContext));
- }
- return new LookupResultForBucket(
- tableBucket, values, lookupData.originalPartitionName(), ApiError.NONE);
- } catch (Exception e) {
- return new LookupResultForBucket(
- tableBucket,
- null,
- lookupData.originalPartitionName(),
- ApiError.fromThrowable(e));
- } finally {
- if (cachedLookuper != null) {
- cachedLookuper.release();
- }
- }
- }
-
private void onLookuperRemoved(
Long ignored, @Nullable CachedLakeTableLookuper cachedLookuper, RemovalCause cause) {
if (cachedLookuper == null) {
@@ -521,28 +357,10 @@ class HistoricalLakeLookupManager implements AutoCloseable {
LookupDataForBucket lookupData,
TableInfo tableInfo,
SchemaInfo schemaInfo,
+ ResolvedPartitionSpec originalPartitionSpec,
LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
TableBucket tableBucket = lookupData.tableBucket();
- String originalPartitionName = lookupData.originalPartitionName();
- if (originalPartitionName == null) {
- throw new InvalidPartitionException(
- "Historical lookup request must carry the original partition name.");
- }
-
TablePath tablePath = tableInfo.getTablePath();
-
- ResolvedPartitionSpec originalPartitionSpec;
- try {
- originalPartitionSpec =
- ResolvedPartitionSpec.fromPartitionName(
- tableInfo.getPartitionKeys(), originalPartitionName);
- } catch (RuntimeException e) {
- throw new InvalidPartitionException(
- String.format(
- "Invalid original partition name %s for historical lookup on table %s.",
- originalPartitionName, tablePath));
- }
-
LakeTableLookuper.LookupContext lookupContext =
new LakeTableLookuper.LookupContext(
originalPartitionSpec,
@@ -659,6 +477,60 @@ class HistoricalLakeLookupManager implements AutoCloseable {
}
}
+ private CachedLakeTableLookuper acquireLookuper(LookupContext context, TableInfo tableInfo) {
+ long currentLakeConfigVersion = lakeConfigVersion;
+ Configuration currentConf = conf;
+ long cacheSizeBytes = lookupCacheMaxDiskBytesPerTable;
+ Long requiredLakeSnapshotId = requiredLakeSnapshotIds.get(context.tableId);
+ return lakeTableLookupers
+ .asMap()
+ .compute(
+ context.tableId,
+ (ignored, currentLookuper) -> {
+ CachedLakeTableLookuper selectedLookuper = currentLookuper;
+ // Create the lookuper lazily, and recreate it after schema,
+ // lake configuration, or server cache size changes so it
+ // reloads lake table/query state and uses the current
+ // settings.
+ if (selectedLookuper == null
+ || selectedLookuper.schemaId != context.schemaId
+ || selectedLookuper.lakeConfigVersion
+ != currentLakeConfigVersion
+ || selectedLookuper.cacheSizeBytes != cacheSizeBytes
+ || !Objects.equals(
+ selectedLookuper.lakeSnapshotId,
+ requiredLakeSnapshotId)) {
+ File tableLookupDir =
+ FlussPaths.historicalLookupTableDir(
+ historicalLookupCacheRootDir,
+ context.tablePath,
+ context.tableId);
+ LakeTableLookuper lookuper =
+ createLakeTableLookuper(
+ context.tablePath,
+ tableLookupDir.getAbsolutePath(),
+ tableInfo.getTableConfig(),
+ cacheSizeBytes,
+ currentConf);
+ selectedLookuper =
+ new CachedLakeTableLookuper(
+ context.tableId,
+ context.tablePath,
+ context.schemaId,
+ currentLakeConfigVersion,
+ cacheSizeBytes,
+ requiredLakeSnapshotId,
+ tableLookupDir,
+ lookuper);
+ }
+ // Pin the lookuper before leaving the atomic cache update.
+ // Eviction or invalidation can then defer closing it until
+ // this lookup releases it.
+ selectedLookuper.acquire();
+ return selectedLookuper;
+ });
+ }
+
private static final class LookupContext {
private final long tableId;
private final int schemaId;
@@ -683,6 +555,9 @@ class HistoricalLakeLookupManager implements AutoCloseable {
private final int schemaId;
private final long lakeConfigVersion;
private final long cacheSizeBytes;
+ /** The required opaque lake snapshot ID when this lookuper was created, or null if none. */
+ private final @Nullable Long lakeSnapshotId;
+
private final File tableLookupDir;
private final LakeTableLookuper lookuper;
private int activeLookups;
@@ -695,6 +570,7 @@ class HistoricalLakeLookupManager implements AutoCloseable {
int schemaId,
long lakeConfigVersion,
long cacheSizeBytes,
+ @Nullable Long lakeSnapshotId,
File tableLookupDir,
LakeTableLookuper lookuper) {
this.tableId = tableId;
@@ -702,6 +578,7 @@ class HistoricalLakeLookupManager implements AutoCloseable {
this.schemaId = schemaId;
this.lakeConfigVersion = lakeConfigVersion;
this.cacheSizeBytes = cacheSizeBytes;
+ this.lakeSnapshotId = lakeSnapshotId;
this.tableLookupDir = tableLookupDir;
this.lookuper = lookuper;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java
new file mode 100644
index 000000000..f54aef0c0
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManager.java
@@ -0,0 +1,353 @@
+/*
+ * 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.fluss.server.replica.historical;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.HistoricalPartitionThrottledException;
+import org.apache.fluss.exception.InvalidPartitionException;
+import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metrics.Counter;
+import org.apache.fluss.plugin.PluginManager;
+import org.apache.fluss.rpc.entity.LookupResultForBucket;
+import org.apache.fluss.rpc.entity.PutKvResultForBucket;
+import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.rpc.protocol.MergeMode;
+import org.apache.fluss.server.entity.LookupDataForBucket;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
+import org.apache.fluss.server.kv.KvStateLookupResult;
+import org.apache.fluss.server.kv.KvStateLookupResult.Status;
+import org.apache.fluss.server.kv.historical.HistoricalValueLookup;
+import org.apache.fluss.server.log.LogAppendInfo;
+import org.apache.fluss.server.replica.Replica;
+import org.apache.fluss.server.storage.LocalDiskManager;
+import org.apache.fluss.utils.ByteArrayWrapper;
+import org.apache.fluss.utils.concurrent.Scheduler;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Coordinates lookup, write, and lifecycle operations for historical partitions. */
+@Internal
+public final class HistoricalPartitionManager implements AutoCloseable {
+
+ private final HistoricalPartitionTaskExecutor taskExecutor;
+ private final HistoricalLakeLookupManager lakeLookupManager;
+
+ /** Creates a historical partition manager from the tablet server dependencies. */
+ public HistoricalPartitionManager(
+ Configuration conf,
+ @Nullable PluginManager pluginManager,
+ LocalDiskManager localDiskManager,
+ File dataDir,
+ long dataDirVolumeBytes,
+ Scheduler scheduler) {
+ this(
+ new HistoricalPartitionTaskExecutor(conf),
+ new HistoricalLakeLookupManager(
+ conf,
+ pluginManager,
+ localDiskManager,
+ dataDir,
+ dataDirVolumeBytes,
+ scheduler));
+ }
+
+ @VisibleForTesting
+ HistoricalPartitionManager(
+ HistoricalPartitionTaskExecutor taskExecutor,
+ HistoricalLakeLookupManager lakeLookupManager) {
+ this.taskExecutor = checkNotNull(taskExecutor, "taskExecutor must not be null");
+ this.lakeLookupManager =
+ checkNotNull(lakeLookupManager, "lakeLookupManager must not be null");
+ }
+
+ /** Starts the resources used by historical partition operations. */
+ public void startup(Scheduler scheduler) {
+ lakeLookupManager.startup(scheduler);
+ }
+
+ /** Looks up historical keys from the local overlay and then lake storage. */
+ public CompletableFuture<LookupResultForBucket> lookup(
+ Replica replica,
+ LookupDataForBucket lookupData,
+ LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
+ TableBucket tableBucket = lookupData.tableBucket();
+ try {
+ LakeTableLookuper.LookupMetricRecorder checkedMetricRecorder =
+ checkNotNull(lookupMetricRecorder, "lookupMetricRecorder must not be null.");
+ return taskExecutor.submit(
+ () -> lookupInternal(replica, lookupData, checkedMetricRecorder),
+ () ->
+ new LookupResultForBucket(
+ tableBucket,
+ null,
+ lookupData.originalPartitionName(),
+ ApiError.fromThrowable(
+ new HistoricalPartitionThrottledException(
+ "Historical lookup is throttled for "
+ + tableBucket
+ + " (original partition "
+ + lookupData.originalPartitionName()
+ + ")."))));
+ } catch (RuntimeException e) {
+ return CompletableFuture.completedFuture(
+ new LookupResultForBucket(
+ tableBucket,
+ null,
+ lookupData.originalPartitionName(),
+ ApiError.fromThrowable(e)));
+ }
+ }
+
+ /** Writes records to the local overlay of a historical partition. */
+ public CompletableFuture<PutKvResultForBucket> putKv(
+ Replica replica,
+ PutKvDataForBucket putData,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ int requiredAcks) {
+ try {
+ String originalPartitionName =
+ checkNotNull(
+ putData.originalPartitionName(),
+ "originalPartitionName must not be null");
+ return taskExecutor.submitOrdered(
+ putData.tableBucket(),
+ () -> {
+ try {
+ LogAppendInfo appendInfo =
+ processPut(
+ replica,
+ putData,
+ targetColumns,
+ mergeMode,
+ requiredAcks);
+ return PutKvResultForBucket.historicalSuccess(
+ putData.tableBucket(),
+ appendInfo.lastOffset() + 1,
+ originalPartitionName);
+ } catch (Throwable t) {
+ return PutKvResultForBucket.historicalFailure(
+ putData.tableBucket(),
+ ApiError.fromThrowable(t),
+ originalPartitionName);
+ }
+ },
+ () ->
+ PutKvResultForBucket.historicalFailure(
+ putData.tableBucket(),
+ ApiError.fromThrowable(
+ new HistoricalPartitionThrottledException(
+ "Historical write is throttled for "
+ + putData.tableBucket()
+ + " (original partition "
+ + originalPartitionName
+ + ").")),
+ originalPartitionName));
+ } catch (RuntimeException e) {
+ return CompletableFuture.completedFuture(
+ PutKvResultForBucket.historicalFailure(
+ putData.tableBucket(),
+ ApiError.fromThrowable(e),
+ putData.originalPartitionName()));
+ }
+ }
+
+ /** Applies dynamic historical lookup configuration changes. */
+ public void reconfigure(Configuration newConf) {
+ lakeLookupManager.reconfigure(newConf);
+ }
+
+ /** Invalidates the cached lake lookuper for the given table. */
+ public void invalidateTableLookuper(long tableId) {
+ lakeLookupManager.invalidateTableLookuper(tableId);
+ }
+
+ /** Requires future fallback lookups to reload after the given lake snapshot notification. */
+ public void requireLakeSnapshot(long tableId, long lakeSnapshotId) {
+ lakeLookupManager.requireLakeSnapshot(tableId, lakeSnapshotId);
+ }
+
+ /** Returns the number of accepted historical operations that have not completed. */
+ public int numInflightRequests() {
+ return taskExecutor.numInflightRequests();
+ }
+
+ /** Returns the current disk usage of the historical lake lookup cache. */
+ public long lookupCacheDiskSize() {
+ return lakeLookupManager.lookupCacheDiskSize();
+ }
+
+ /** Returns the number of cached historical lake table lookupers. */
+ public int cachedTableCount() {
+ return lakeLookupManager.cachedTableCount();
+ }
+
+ /** Returns the counter for lookuper evictions caused by the table cache capacity. */
+ public Counter capacityEvictions() {
+ return lakeLookupManager.capacityEvictions();
+ }
+
+ @VisibleForTesting
+ LogAppendInfo processPut(
+ Replica replica,
+ PutKvDataForBucket putData,
+ @Nullable int[] targetColumns,
+ MergeMode mergeMode,
+ int requiredAcks)
+ throws Exception {
+ TableInfo tableInfo = replica.getTableInfo();
+ String originalPartitionName =
+ checkNotNull(
+ putData.originalPartitionName(), "originalPartitionName must not be null");
+ ResolvedPartitionSpec originalPartitionSpec =
+ ResolvedPartitionSpec.fromPartitionName(
+ tableInfo.getPartitionKeys(), originalPartitionName);
+ // The public put path holds the TableBucket ordering slot until processPut returns, so
+ // local state cannot be changed by a later historical write between resolve and apply.
+ int expectedLeaderEpoch = replica.getLeaderEpoch();
+ List<byte[]> keysRequiringLakeLookup =
+ replica.findKeysRequiringLakeLookup(
+ putData.records(),
+ targetColumns,
+ mergeMode,
+ originalPartitionName,
+ expectedLeaderEpoch,
+ requiredAcks);
+
+ Map<ByteArrayWrapper, KvStateLookupResult> lakeResults = new HashMap<>();
+ if (!keysRequiringLakeLookup.isEmpty()) {
+ List<byte[]> lakeValues =
+ lakeLookupManager.lookup(
+ new LookupDataForBucket(
+ putData.tableBucket(),
+ keysRequiringLakeLookup,
+ originalPartitionName),
+ tableInfo,
+ replica.getLatestSchemaInfo(),
+ originalPartitionSpec,
+ replica.tableMetrics()::recordHistoricalLakeLookup);
+ for (int i = 0; i < keysRequiringLakeLookup.size(); i++) {
+ byte[] lakeValue = lakeValues.get(i);
+ lakeResults.put(
+ new ByteArrayWrapper(keysRequiringLakeLookup.get(i)),
+ lakeValue == null
+ ? KvStateLookupResult.notFound()
+ : KvStateLookupResult.present(lakeValue));
+ }
+ }
+
+ HistoricalValueLookup memoizedLakeLookup =
+ primaryKey -> {
+ KvStateLookupResult result =
+ checkNotNull(
+ lakeResults.get(new ByteArrayWrapper(primaryKey)),
+ "No resolved lake value for a historical write key");
+ return result.value();
+ };
+
+ return replica.putHistoricalRecordsToLeader(
+ putData.records(),
+ targetColumns,
+ mergeMode,
+ originalPartitionName,
+ memoizedLakeLookup,
+ expectedLeaderEpoch,
+ requiredAcks);
+ }
+
+ @Override
+ public void close() {
+ taskExecutor.close();
+ lakeLookupManager.close();
+ }
+
+ private LookupResultForBucket lookupInternal(
+ Replica replica,
+ LookupDataForBucket lookupData,
+ LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
+ TableBucket tableBucket = lookupData.tableBucket();
+ String originalPartitionName = lookupData.originalPartitionName();
+ try {
+ TableInfo tableInfo = replica.getTableInfo();
+ if (originalPartitionName == null) {
+ throw new InvalidPartitionException(
+ "Historical lookup request must carry the original partition name.");
+ }
+ ResolvedPartitionSpec originalPartitionSpec =
+ ResolvedPartitionSpec.fromPartitionName(
+ tableInfo.getPartitionKeys(), originalPartitionName);
+
+ List<KvStateLookupResult> localResults =
+ replica.lookupHistoricalLocal(originalPartitionName, lookupData.keys());
+ List<byte[]> missingKeys = new ArrayList<>();
+ for (int i = 0; i < localResults.size(); i++) {
+ KvStateLookupResult localResult = localResults.get(i);
+ // Only a true local miss falls back to lake. A local value or tombstone is
+ // authoritative and must not be overwritten by an older lake value.
+ if (localResult.status() == Status.NOT_FOUND) {
+ missingKeys.add(lookupData.keys().get(i));
+ }
+ }
+
+ List<byte[]> lakeValues = Collections.emptyList();
+ if (!missingKeys.isEmpty()) {
+ // Look up all local misses together. Results preserve the order of missingKeys.
+ lakeValues =
+ lakeLookupManager.lookup(
+ new LookupDataForBucket(
+ tableBucket, missingKeys, originalPartitionName),
+ tableInfo,
+ replica.getLatestSchemaInfo(),
+ originalPartitionSpec,
+ lookupMetricRecorder);
+ }
+
+ Iterator<byte[]> lakeValueIterator = lakeValues.iterator();
+ List<byte[]> values = new ArrayList<>(localResults.size());
+ for (KvStateLookupResult localResult : localResults) {
+ // Consume one lake value for each NOT_FOUND result; local values and tombstones
+ // keep their original positions without advancing the lake iterator.
+ values.add(
+ localResult.status() == Status.NOT_FOUND
+ ? lakeValueIterator.next()
+ : localResult.value());
+ }
+ return new LookupResultForBucket(
+ tableBucket, values, originalPartitionName, ApiError.NONE);
+ } catch (Exception e) {
+ return new LookupResultForBucket(
+ tableBucket, null, originalPartitionName, ApiError.fromThrowable(e));
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java
new file mode 100644
index 000000000..9bf66848f
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutor.java
@@ -0,0 +1,210 @@
+/*
+ * 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.fluss.server.replica.historical;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.utils.ExecutorUtils;
+import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.GuardedBy;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Executes lookup and write tasks for historical partitions.
+ *
+ * <p>Lookups and writes share one executor and one request limit. A permit is held from task
+ * acceptance until completion, so the limit covers both queued and running work.
+ *
+ * <p>Normal submissions may run concurrently. Ordered submissions with the same key are chained in
+ * acceptance order, while submissions with different keys can still run in parallel.
+ */
+@Internal
+public final class HistoricalPartitionTaskExecutor implements AutoCloseable {
+
+ private static final Duration THREAD_KEEP_ALIVE = Duration.ofMinutes(10);
+ private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(10);
+ private static final String THREAD_NAME_PREFIX = "historical-partition-io";
+
+ private final int maxQueuedHistoricalRequests;
+ // Shared by lookup and write tasks, including tasks waiting in the executor queue.
+ private final Semaphore requestPermits;
+ // Keep accepted requests so close() can cancel work that remains after executor shutdown.
+ private final Set<CompletableFuture<?>> pendingRequests;
+ private final ExecutorService executor;
+ private final Object orderedTasksLock = new Object();
+
+ // The latest accepted task for each ordering key. A completed tail is removed only when it is
+ // still the current tail, so an older completion cannot remove a newer task from the chain.
+ @GuardedBy("orderedTasksLock")
+ private final Map<Object, CompletableFuture<Void>> orderedTaskTails;
+
+ /** Creates a historical-partition task executor from the server configuration. */
+ public HistoricalPartitionTaskExecutor(Configuration conf) {
+ this(conf, null);
+ }
+
+ /** Creates a historical-partition task executor backed by the supplied executor. */
+ @VisibleForTesting
+ public HistoricalPartitionTaskExecutor(
+ Configuration conf, @Nullable ExecutorService historicalPartitionExecutor) {
+ checkNotNull(conf, "conf must not be null.");
+ this.maxQueuedHistoricalRequests =
+ conf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS);
+ checkArgument(
+ maxQueuedHistoricalRequests > 0,
+ "%s must be greater than 0.",
+ ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
+ int maxThreadPoolSize =
+ conf.get(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE);
+ checkArgument(
+ maxThreadPoolSize > 0,
+ "%s must be greater than 0.",
+ ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
+ this.executor =
+ historicalPartitionExecutor == null
+ ? createHistoricalPartitionExecutor(maxThreadPoolSize)
+ : historicalPartitionExecutor;
+ this.requestPermits = new Semaphore(maxQueuedHistoricalRequests);
+ this.pendingRequests = ConcurrentHashMap.newKeySet();
+ this.orderedTaskTails = new HashMap<>();
+ }
+
+ /**
+ * Submits a task or returns the supplied throttled result when the shared request limit is
+ * full.
+ *
+ * <p>The throttled result is completed by the caller thread without entering the executor.
+ */
+ public <T> CompletableFuture<T> submit(Supplier<T> task, Supplier<T> throttledResult) {
+ checkNotNull(task, "task must not be null.");
+ checkNotNull(throttledResult, "throttledResult must not be null.");
+ if (!requestPermits.tryAcquire()) {
+ return CompletableFuture.completedFuture(throttledResult.get());
+ }
+
+ CompletableFuture<T> future;
+ try {
+ future = CompletableFuture.supplyAsync(task, executor);
+ } catch (RuntimeException e) {
+ requestPermits.release();
+ throw e;
+ }
+ return trackAcceptedRequest(future);
+ }
+
+ /**
+ * Submits a task after all previously accepted tasks with the same ordering key have finished.
+ *
+ * <p>A failed task does not break the chain: its completion still releases the next task for
+ * that key.
+ */
+ public <T> CompletableFuture<T> submitOrdered(
+ Object orderingKey, Supplier<T> task, Supplier<T> throttledResult) {
+ checkNotNull(orderingKey, "orderingKey must not be null.");
+ checkNotNull(task, "task must not be null.");
+ checkNotNull(throttledResult, "throttledResult must not be null.");
+ if (!requestPermits.tryAcquire()) {
+ return CompletableFuture.completedFuture(throttledResult.get());
+ }
+
+ CompletableFuture<T> future;
+ CompletableFuture<Void> tail;
+ try {
+ synchronized (orderedTasksLock) {
+ CompletableFuture<Void> previousTail = orderedTaskTails.get(orderingKey);
+ if (previousTail == null) {
+ future = CompletableFuture.supplyAsync(task, executor);
+ } else {
+ future = previousTail.thenApplyAsync(ignored -> task.get(), executor);
+ }
+ // Convert success or failure into a normal completion used only for sequencing.
+ tail = future.handle((ignored, error) -> null);
+ orderedTaskTails.put(orderingKey, tail);
+ }
+ } catch (RuntimeException e) {
+ requestPermits.release();
+ throw e;
+ }
+
+ CompletableFuture<Void> currentTail = tail;
+ tail.whenComplete(
+ (ignored, error) -> {
+ synchronized (orderedTasksLock) {
+ orderedTaskTails.remove(orderingKey, currentTail);
+ }
+ });
+ return trackAcceptedRequest(future);
+ }
+
+ private <T> CompletableFuture<T> trackAcceptedRequest(CompletableFuture<T> future) {
+ pendingRequests.add(future);
+ future.whenComplete(
+ (ignored, error) -> {
+ // Release the permit exactly once when the accepted task reaches a terminal
+ // state, including exceptional completion and cancellation.
+ pendingRequests.remove(future);
+ requestPermits.release();
+ });
+ return future;
+ }
+
+ /** Returns the number of accepted historical requests that have not completed. */
+ @VisibleForTesting
+ public int numInflightRequests() {
+ return maxQueuedHistoricalRequests - requestPermits.availablePermits();
+ }
+
+ @Override
+ public void close() {
+ ExecutorUtils.gracefulShutdown(
+ SHUTDOWN_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS, executor);
+ pendingRequests.forEach(future -> future.cancel(true));
+ }
+
+ private static ExecutorService createHistoricalPartitionExecutor(int maxThreadPoolSize) {
+ ThreadPoolExecutor executor =
+ new ThreadPoolExecutor(
+ maxThreadPoolSize,
+ maxThreadPoolSize,
+ THREAD_KEEP_ALIVE.toMillis(),
+ TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>(),
+ new ExecutorThreadFactory(THREAD_NAME_PREFIX));
+ executor.allowCoreThreadTimeOut(true);
+ return executor;
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
index f7a004de3..ba31a143d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
@@ -91,6 +91,7 @@ import org.apache.fluss.server.entity.NotifyKvSnapshotOffsetData;
import org.apache.fluss.server.entity.NotifyLakeTableOffsetData;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.entity.NotifyRemoteLogOffsetsData;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
import org.apache.fluss.server.entity.StopReplicaData;
import org.apache.fluss.server.entity.UserContext;
import org.apache.fluss.server.kv.scan.OpenScanResult;
@@ -123,6 +124,7 @@ import java.util.function.Consumer;
import java.util.stream.Collectors;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalLookup;
+import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.hasHistoricalPut;
import static org.apache.fluss.security.acl.OperationType.DESCRIBE;
import static org.apache.fluss.security.acl.OperationType.READ;
import static org.apache.fluss.security.acl.OperationType.WRITE;
@@ -135,7 +137,6 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifyLeade
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifyRemoteLogOffsetsData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getNotifySnapshotOffsetData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getProduceLogData;
-import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getPutKvData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getStopReplicaData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getTableFilterInfoMap;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getTableStatsRequestData;
@@ -155,6 +156,7 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeStopReplic
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toHistoricalLookupData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toLookupData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toPrefixLookupData;
+import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toPutKvDataForBuckets;
/** An RPC Gateway service for tablet server. */
public final class TabletService extends RpcServiceBase implements TabletServerGateway {
@@ -283,21 +285,35 @@ public final class TabletService extends RpcServiceBase implements TabletServerG
public CompletableFuture<PutKvResponse> putKv(PutKvRequest request) {
authorizeTable(WRITE, request.getTableId());
- Map<TableBucket, KvRecordBatch> putKvData = getPutKvData(request);
+ List<PutKvDataForBucket> putKvData = toPutKvDataForBuckets(request);
// Get mergeMode from request, default to DEFAULT if not set
MergeMode mergeMode =
request.hasAggMode()
? MergeMode.fromValue(request.getAggMode())
: MergeMode.DEFAULT;
CompletableFuture<PutKvResponse> response = new CompletableFuture<>();
- replicaManager.putRecordsToKv(
- request.getTimeoutMs(),
- request.getAcks(),
- putKvData,
- getTargetColumns(request),
- mergeMode,
- currentSession().getApiVersion(),
- bucketResponse -> response.complete(makePutKvResponse(bucketResponse)));
+ if (hasHistoricalPut(request)) {
+ replicaManager.putHistoricalRecordsToKv(
+ request.getTimeoutMs(),
+ request.getAcks(),
+ putKvData,
+ getTargetColumns(request),
+ mergeMode,
+ currentSession().getApiVersion(),
+ bucketResponse -> response.complete(makePutKvResponse(bucketResponse)));
+ } else {
+ Map<TableBucket, KvRecordBatch> recordsByBucket = new HashMap<>();
+ putKvData.forEach(
+ putData -> recordsByBucket.put(putData.tableBucket(), putData.records()));
+ replicaManager.putRecordsToKv(
+ request.getTimeoutMs(),
+ request.getAcks(),
+ recordsByBucket,
+ getTargetColumns(request),
+ mergeMode,
+ currentSession().getApiVersion(),
+ bucketResponse -> response.complete(makePutKvResponse(bucketResponse)));
+ }
return response;
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 89ae6f713..81131f1bc 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -50,7 +50,6 @@ import org.apache.fluss.record.DefaultKvRecordBatch;
import org.apache.fluss.record.DefaultValueRecordBatch;
import org.apache.fluss.record.FileChannelChunk;
import org.apache.fluss.record.FileLogRecords;
-import org.apache.fluss.record.KvRecordBatch;
import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.remote.RemoteLogFetchInfo;
@@ -188,6 +187,7 @@ import org.apache.fluss.server.entity.NotifyLakeTableOffsetData;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
import org.apache.fluss.server.entity.NotifyLeaderAndIsrResultForBucket;
import org.apache.fluss.server.entity.NotifyRemoteLogOffsetsData;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
import org.apache.fluss.server.entity.StopReplicaData;
import org.apache.fluss.server.entity.StopReplicaResultForBucket;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
@@ -1101,22 +1101,52 @@ public class ServerRpcMessageUtils {
return fetchLogResponse;
}
- public static Map<TableBucket, KvRecordBatch> getPutKvData(PutKvRequest putKvRequest) {
+ /**
+ * Converts put-KV bucket requests while preserving their historical partition context.
+ *
+ * <p>The returned original partition name is null for normal writes. Normal and historical
+ * writes cannot be mixed in one request. Historical target and partition eligibility are
+ * validated by the historical write path.
+ */
+ public static List<PutKvDataForBucket> toPutKvDataForBuckets(PutKvRequest putKvRequest) {
long tableId = putKvRequest.getTableId();
- Map<TableBucket, KvRecordBatch> produceEntryData = new HashMap<>();
+ List<PutKvDataForBucket> putKvData = new ArrayList<>(putKvRequest.getBucketsReqsCount());
+ Map<TableBucket, Set<String>> originalPartitionsByBucket = new HashMap<>();
+ boolean historicalWriteRequest =
+ putKvRequest.getBucketsReqsCount() > 0
+ && putKvRequest.getBucketsReqAt(0).hasOriginalPartitionName();
for (PbPutKvReqForBucket putKvReqForBucket : putKvRequest.getBucketsReqsList()) {
+ if (putKvReqForBucket.hasOriginalPartitionName() != historicalWriteRequest) {
+ throw new IllegalArgumentException(
+ "Normal and historical writes cannot be mixed in the same request.");
+ }
ByteBuffer recordsBuffer = toByteBuffer(putKvReqForBucket.getRecordsSlice());
DefaultKvRecordBatch kvRecords = DefaultKvRecordBatch.pointToByteBuffer(recordsBuffer);
- TableBucket tb =
+ TableBucket tableBucket =
new TableBucket(
tableId,
putKvReqForBucket.hasPartitionId()
? putKvReqForBucket.getPartitionId()
: null,
putKvReqForBucket.getBucketId());
- produceEntryData.put(tb, kvRecords);
+ String originalPartitionName =
+ putKvReqForBucket.hasOriginalPartitionName()
+ ? putKvReqForBucket.getOriginalPartitionName()
+ : null;
+ Set<String> originalPartitions =
+ originalPartitionsByBucket.computeIfAbsent(
+ tableBucket, ignored -> new HashSet<>());
+ if (!originalPartitions.add(originalPartitionName)) {
+ throw new IllegalArgumentException(
+ "A PutKv request contains duplicate table bucket "
+ + tableBucket
+ + " and original partition "
+ + originalPartitionName
+ + '.');
+ }
+ putKvData.add(new PutKvDataForBucket(tableBucket, kvRecords, originalPartitionName));
}
- return produceEntryData;
+ return putKvData;
}
public static Map<TableBucket, List<byte[]>> toLookupData(LookupRequest lookupRequest) {
@@ -1214,6 +1244,9 @@ public class ServerRpcMessageUtils {
if (tableBucket.getPartitionId() != null) {
putKvBucket.setPartitionId(tableBucket.getPartitionId());
}
+ if (bucketResult.getOriginalPartitionName() != null) {
+ putKvBucket.setOriginalPartitionName(bucketResult.getOriginalPartitionName());
+ }
if (bucketResult.failed()) {
putKvBucket.setError(bucketResult.getErrorCode(), bucketResult.getErrorMessage());
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoderTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoderTest.java
new file mode 100644
index 000000000..557863f94
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/historical/HistoricalKvKeyEncoderTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.fluss.server.kv.historical;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link HistoricalKvKeyEncoder}. */
+class HistoricalKvKeyEncoderTest {
+
+ @Test
+ void testEncodeKey() {
+ assertEncodedKey("dt=2026-07-12", new byte[] {1, 2, 3});
+ assertEncodedKey("地区=杭州", new byte[] {0, 1, -1});
+ assertEncodedKey("dt=2026-07-12/region=cn", new byte[0]);
+ }
+
+ @Test
+ void testEncodingHasUnambiguousPartitionBoundary() {
+ byte[] first = HistoricalKvKeyEncoder.encode("ab", "c".getBytes(StandardCharsets.UTF_8));
+ byte[] second = HistoricalKvKeyEncoder.encode("a", "bc".getBytes(StandardCharsets.UTF_8));
+
+ assertThat(first).isNotEqualTo(second);
+ assertThat(HistoricalKvKeyEncoder.encode("p1", new byte[] {1}))
+ .isNotEqualTo(HistoricalKvKeyEncoder.encode("p2", new byte[] {1}));
+ assertThat(HistoricalKvKeyEncoder.encode("p1", new byte[] {1}))
+ .isEqualTo(HistoricalKvKeyEncoder.encode("p1", new byte[] {1}));
+ }
+
+ @Test
+ void testRejectInvalidInput() {
+ assertThatThrownBy(() -> HistoricalKvKeyEncoder.encode(null, new byte[0]))
+ .isInstanceOf(NullPointerException.class);
+ assertThatThrownBy(() -> HistoricalKvKeyEncoder.encode("", new byte[0]))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> HistoricalKvKeyEncoder.encode("p", null))
+ .isInstanceOf(NullPointerException.class);
+ }
+
+ private static void assertEncodedKey(String partitionName, byte[] primaryKey) {
+ byte[] partitionNameBytes = partitionName.getBytes(StandardCharsets.UTF_8);
+ byte[] encoded = HistoricalKvKeyEncoder.encode(partitionName, primaryKey);
+ ByteBuffer buffer = ByteBuffer.wrap(encoded);
+
+ int partitionNameLength = buffer.getInt();
+ byte[] actualPartitionName = new byte[partitionNameLength];
+ buffer.get(actualPartitionName);
+ byte[] actualPrimaryKey = new byte[buffer.remaining()];
+ buffer.get(actualPrimaryKey);
+
+ assertThat(partitionNameLength).isEqualTo(partitionNameBytes.length);
+ assertThat(actualPartitionName).isEqualTo(partitionNameBytes);
+ assertThat(actualPrimaryKey).isEqualTo(primaryKey);
+ assertThat(HistoricalKvKeyEncoder.extractOriginalPrimaryKey(encoded)).isEqualTo(primaryKey);
+ assertThat(encoded).hasSize(Integer.BYTES + partitionNameBytes.length + primaryKey.length);
+ assertThat(Arrays.copyOfRange(encoded, Integer.BYTES, encoded.length))
+ .startsWith(partitionNameBytes);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvTest.java
index c25dd8e35..7add36047 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rocksdb/RocksDBKvTest.java
@@ -68,6 +68,18 @@ class RocksDBKvTest {
}
}
+ @Test
+ void testLiveSstFilesSize(@TempDir Path tempDir) throws Exception {
+ try (RocksDBKv rocksDBKv = buildRocksDBKv(tempDir.toFile())) {
+ rocksDBKv.put(new byte[] {1}, new byte[] {2});
+ try (FlushOptions flushOptions = new FlushOptions().setWaitForFlush(true)) {
+ rocksDBKv.db.flush(flushOptions);
+ }
+
+ assertThat(rocksDBKv.liveSstFilesSize()).isPositive();
+ }
+ }
+
@Test
void testClosePreservesUnflushedStateByDefault(@TempDir Path tempDir) throws Exception {
byte[] key = new byte[] {1};
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManagerTest.java
similarity index 55%
rename from fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java
rename to fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManagerTest.java
index afa53e551..364ec46be 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalLakeLookupManagerTest.java
@@ -15,24 +15,22 @@
* limitations under the License.
*/
-package org.apache.fluss.server.replica;
+package org.apache.fluss.server.replica.historical;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.MemorySize;
import org.apache.fluss.config.TableConfig;
-import org.apache.fluss.exception.HistoricalPartitionThrottledException;
import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.KvFormat;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaInfo;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.rpc.entity.LookupResultForBucket;
-import org.apache.fluss.rpc.protocol.Errors;
import org.apache.fluss.server.entity.LookupDataForBucket;
import org.apache.fluss.types.DataTypes;
import org.apache.fluss.utils.FlussPaths;
@@ -41,8 +39,6 @@ import com.github.benmanes.caffeine.cache.Scheduler;
import com.github.benmanes.caffeine.cache.Ticker;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.io.RandomAccessFile;
@@ -50,27 +46,19 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.concurrent.AbstractExecutorService;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.FutureTask;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.fluss.record.TestData.PARTITION_TABLE_ID;
import static org.apache.fluss.record.TestData.PARTITION_TABLE_INFO;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link HistoricalLakeLookupManager}. */
class HistoricalLakeLookupManagerTest {
private static final long DATA_DIR_VOLUME_BYTES = MemorySize.parse("800gb").getBytes();
- private static final TableBucket HISTORICAL_BUCKET = new TableBucket(PARTITION_TABLE_ID, 1L, 0);
private static final LakeTableLookuper.LookupMetricRecorder NO_OP_LOOKUP_METRIC_RECORDER =
(lookupTimeNanos, lookupFileDownloaded) -> {};
private static final Runnable NO_OP_DISK_WRITE_GUARD = () -> {};
@@ -79,141 +67,6 @@ class HistoricalLakeLookupManagerTest {
@TempDir private File ioTmpDir;
- @Test
- void testHistoricalLookupThrottledWhenPermitsExhausted() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- HistoricalLakeLookupManager manager = createManager(1, executor);
- assertThat(manager.numInflightRequests()).isZero();
-
- CompletableFuture<LookupResultForBucket> first =
- manager.lookup(
- lookupData(HISTORICAL_BUCKET),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
- assertThat(first).isNotDone();
- assertThat(executor.numQueuedTasks()).isEqualTo(1);
- assertThat(manager.numInflightRequests()).isOne();
-
- TableBucket secondBucket = new TableBucket(PARTITION_TABLE_ID, 2L, 0);
- LookupResultForBucket second =
- manager.lookup(
- lookupData(secondBucket),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER)
- .get(1, TimeUnit.SECONDS);
-
- assertThat(second.failed()).isTrue();
- assertThat(second.getError().error()).isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED);
- assertThat(second.getError().exception())
- .isInstanceOf(HistoricalPartitionThrottledException.class);
- assertThat(executor.numQueuedTasks()).isEqualTo(1);
- assertThat(manager.numInflightRequests()).isOne();
- }
-
- @Test
- void testHistoricalLookupReleasesPermitOnFailure() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- HistoricalLakeLookupManager manager = createManager(1, executor);
-
- CompletableFuture<LookupResultForBucket> first =
- manager.lookup(
- lookupData(HISTORICAL_BUCKET),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
- executor.runNext();
- LookupResultForBucket firstResult = first.get(1, TimeUnit.SECONDS);
- assertThat(firstResult.failed()).isTrue();
- assertThat(firstResult.getError().error())
- .isNotEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED);
- assertThat(manager.numInflightRequests()).isZero();
-
- CompletableFuture<LookupResultForBucket> second =
- manager.lookup(
- lookupData(HISTORICAL_BUCKET),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
- assertThat(second).isNotDone();
- assertThat(executor.numQueuedTasks()).isEqualTo(1);
- }
-
- @Test
- void testHistoricalLookupMaxQueuedRequestsUsesExplicitConfig() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- HistoricalLakeLookupManager manager = createManager(2, executor);
-
- CompletableFuture<LookupResultForBucket> first =
- manager.lookup(
- lookupData(new TableBucket(PARTITION_TABLE_ID, 1L, 0)),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
- CompletableFuture<LookupResultForBucket> second =
- manager.lookup(
- lookupData(new TableBucket(PARTITION_TABLE_ID, 2L, 0)),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
- LookupResultForBucket third =
- manager.lookup(
- lookupData(new TableBucket(PARTITION_TABLE_ID, 3L, 0)),
- PARTITION_TABLE_INFO,
- PARTITION_TABLE_INFO.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER)
- .get(1, TimeUnit.SECONDS);
-
- assertThat(first).isNotDone();
- assertThat(second).isNotDone();
- assertThat(executor.numQueuedTasks()).isEqualTo(2);
- assertThat(third.getError().error()).isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED);
- }
-
- @Test
- void testRejectNonPositiveHistoricalLookupMaxQueuedRequests() {
- Configuration conf = conf(0);
- ManualExecutor executor = new ManualExecutor();
-
- assertThatThrownBy(
- () ->
- new HistoricalLakeLookupManager(
- conf,
- null,
- executor,
- ioTmpDir,
- DATA_DIR_VOLUME_BYTES,
- Ticker.systemTicker(),
- Scheduler.disabledScheduler(),
- NO_OP_DISK_WRITE_GUARD))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining(
- ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
- }
-
- @ParameterizedTest
- @ValueSource(ints = {0, -1})
- void testRejectNonPositiveHistoricalPartitionThreadPoolMaxSize(int maxThreadPoolSize) {
- Configuration conf = conf(1);
- conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE, maxThreadPoolSize);
-
- assertThatThrownBy(
- () ->
- new HistoricalLakeLookupManager(
- conf,
- null,
- null,
- ioTmpDir,
- DATA_DIR_VOLUME_BYTES,
- Ticker.systemTicker(),
- Scheduler.disabledScheduler(),
- NO_OP_DISK_WRITE_GUARD))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessageContaining(
- ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
- }
-
@Test
void testCleansAndCreatesLookupCacheDirectoryOnStartup() throws Exception {
File serverLookupDir = FlussPaths.historicalLookupRootDir(ioTmpDir);
@@ -221,15 +74,13 @@ class HistoricalLakeLookupManagerTest {
File staleLookupFile = new File(serverLookupDir, "stale-lookup-file");
assertThat(staleLookupFile.createNewFile()).isTrue();
- ManualExecutor executor = new ManualExecutor();
- TestingHistoricalLakeLookupManager manager =
- new TestingHistoricalLakeLookupManager(conf(1), executor);
+ TestingHistoricalLakeLookupManager manager = new TestingHistoricalLakeLookupManager(conf());
assertThat(staleLookupFile).exists();
manager.startup(NO_OP_SCHEDULER);
assertThat(staleLookupFile).doesNotExist();
assertThat(serverLookupDir).isDirectory();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
assertThat(manager.createdIoTmpDirs.get(0)).startsWith(serverLookupDir.getAbsolutePath());
File liveLookupFile = new File(serverLookupDir, "live-lookup-file");
@@ -240,8 +91,7 @@ class HistoricalLakeLookupManagerTest {
@Test
void testCreatesLookuperWithTableKvConfig() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- TestingHistoricalLakeLookupManager manager = createTestingManager(executor);
+ TestingHistoricalLakeLookupManager manager = createTestingManager();
TableDescriptor indexedDescriptor =
TableDescriptor.builder(PARTITION_TABLE_INFO.toTableDescriptor())
.kvFormat(KvFormat.INDEXED)
@@ -259,7 +109,7 @@ class HistoricalLakeLookupManagerTest {
PARTITION_TABLE_INFO.getCreatedTime(),
PARTITION_TABLE_INFO.getModifiedTime());
- lookupAndRun(manager, executor, indexedTableInfo);
+ lookup(manager, indexedTableInfo);
assertThat(manager.createdTableConfigs).hasSize(1);
TableConfig createdTableConfig = manager.createdTableConfigs.get(0);
@@ -270,23 +120,21 @@ class HistoricalLakeLookupManagerTest {
@Test
void testDoesNotReuseLookuperForRecreatedTable() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- TestingHistoricalLakeLookupManager manager = createTestingManager(executor);
+ TestingHistoricalLakeLookupManager manager = createTestingManager();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
TableInfo recreatedTableInfo =
tableInfo(PARTITION_TABLE_ID + 1, PARTITION_TABLE_INFO.getSchemaId());
- lookupAndRun(manager, executor, recreatedTableInfo);
+ lookup(manager, recreatedTableInfo);
assertThat(manager.createdLookupers).hasSize(2);
}
@Test
void testInvalidatesLookuperOnSchemaAndLifecycleChanges() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- TestingHistoricalLakeLookupManager manager = createTestingManager(executor);
+ TestingHistoricalLakeLookupManager manager = createTestingManager();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0);
Schema evolvedSchema =
@@ -296,7 +144,7 @@ class HistoricalLakeLookupManagerTest {
.build();
SchemaInfo evolvedSchemaInfo =
new SchemaInfo(evolvedSchema, PARTITION_TABLE_INFO.getSchemaId() + 1);
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO, evolvedSchemaInfo);
+ lookup(manager, PARTITION_TABLE_INFO, evolvedSchemaInfo);
assertThat(initialLookuper.closed).isTrue();
assertThat(manager.createdLookupers).hasSize(2);
@@ -309,16 +157,41 @@ class HistoricalLakeLookupManagerTest {
manager.invalidateTableLookuper(PARTITION_TABLE_ID);
assertThat(evolvedLookuper.closed).isTrue();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO, evolvedSchemaInfo);
+ lookup(manager, PARTITION_TABLE_INFO, evolvedSchemaInfo);
+ assertThat(manager.createdLookupers).hasSize(3);
+ }
+
+ @Test
+ void testRefreshesLookuperWhenLakeSnapshotChanges() throws Exception {
+ TestingHistoricalLakeLookupManager manager = createTestingManager();
+
+ lookup(manager, PARTITION_TABLE_INFO);
+ TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0);
+
+ manager.requireLakeSnapshot(PARTITION_TABLE_ID, 10L);
+ lookup(manager, PARTITION_TABLE_INFO);
+ assertThat(initialLookuper.closed).isTrue();
+ assertThat(manager.createdLookupers).hasSize(2);
+
+ TestingLakeTableLookuper snapshotTenLookuper = manager.createdLookupers.get(1);
+ // Snapshot IDs are opaque; a numerically smaller ID may identify a newer snapshot.
+ manager.requireLakeSnapshot(PARTITION_TABLE_ID, 9L);
+ lookup(manager, PARTITION_TABLE_INFO);
+ assertThat(snapshotTenLookuper.closed).isTrue();
+ assertThat(manager.createdLookupers).hasSize(3);
+
+ TestingLakeTableLookuper snapshotNineLookuper = manager.createdLookupers.get(2);
+ manager.requireLakeSnapshot(PARTITION_TABLE_ID, 9L);
+ lookup(manager, PARTITION_TABLE_INFO);
+ assertThat(snapshotNineLookuper.closed).isFalse();
assertThat(manager.createdLookupers).hasSize(3);
}
@Test
void testDoesNotReplaceLookuperForUnrelatedTableConfigChange() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- TestingHistoricalLakeLookupManager manager = createTestingManager(executor);
+ TestingHistoricalLakeLookupManager manager = createTestingManager();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0);
TableDescriptor changedDescriptor =
@@ -336,7 +209,7 @@ class HistoricalLakeLookupManagerTest {
PARTITION_TABLE_INFO.getRemoteDataDir(),
PARTITION_TABLE_INFO.getCreatedTime(),
PARTITION_TABLE_INFO.getModifiedTime());
- lookupAndRun(manager, executor, changedTableInfo);
+ lookup(manager, changedTableInfo);
assertThat(manager.createdLookupers).hasSize(1);
assertThat(initialLookuper.closed).isFalse();
@@ -344,7 +217,6 @@ class HistoricalLakeLookupManagerTest {
@Test
void testDynamicallyUpdatesExpirationAndExpiresIdleLookuper() throws Exception {
- ManualExecutor executor = new ManualExecutor();
AtomicLong tickerNanos = new AtomicLong();
AtomicReference<FutureTask<Void>> expirationTask = new AtomicReference<>();
Scheduler cacheScheduler =
@@ -360,13 +232,10 @@ class HistoricalLakeLookupManagerTest {
};
TestingHistoricalLakeLookupManager manager =
new TestingHistoricalLakeLookupManager(
- confWithExpiration(Duration.ofHours(1)),
- executor,
- tickerNanos::get,
- cacheScheduler);
+ confWithExpiration(Duration.ofHours(1)), tickerNanos::get, cacheScheduler);
manager.startup(NO_OP_SCHEDULER);
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
TestingLakeTableLookuper expiredLookuper = manager.createdLookupers.get(0);
manager.reconfigure(confWithExpiration(Duration.ofMinutes(30)));
@@ -375,30 +244,21 @@ class HistoricalLakeLookupManagerTest {
expirationTask.get().run();
assertThat(expiredLookuper.closed).isTrue();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
assertThat(manager.createdLookupers).hasSize(2);
}
@Test
void testEvictsLookuperWhenCachedTableLimitIsExceeded() throws Exception {
- ManualExecutor executor = new ManualExecutor();
- Configuration conf = conf(1);
+ Configuration conf = conf();
conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.20);
TestingHistoricalLakeLookupManager manager =
new TestingHistoricalLakeLookupManager(
- conf,
- executor,
- Ticker.systemTicker(),
- Scheduler.disabledScheduler(),
- 100,
- 0);
+ conf, Ticker.systemTicker(), Scheduler.disabledScheduler(), 100, 0);
manager.startup(NO_OP_SCHEDULER);
for (int i = 0; i < 11; i++) {
- lookupAndRun(
- manager,
- executor,
- tableInfo(PARTITION_TABLE_ID + i, PARTITION_TABLE_INFO.getSchemaId()));
+ lookup(manager, tableInfo(PARTITION_TABLE_ID + i, PARTITION_TABLE_INFO.getSchemaId()));
}
assertThat(manager.createdLookupers).hasSize(11);
@@ -410,15 +270,14 @@ class HistoricalLakeLookupManagerTest {
@Test
void testReconfiguresLakePropertiesAndInvalidatesLookuper() throws Exception {
- Configuration initialConf = conf(1);
+ Configuration initialConf = conf();
initialConf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON);
initialConf.setString("datalake.paimon.warehouse", "old-warehouse");
- ManualExecutor executor = new ManualExecutor();
TestingHistoricalLakeLookupManager manager =
- new TestingHistoricalLakeLookupManager(initialConf, executor);
+ new TestingHistoricalLakeLookupManager(initialConf);
manager.startup(NO_OP_SCHEDULER);
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
TestingLakeTableLookuper initialLookuper = manager.createdLookupers.get(0);
Configuration newConf = new Configuration(initialConf);
@@ -427,46 +286,26 @@ class HistoricalLakeLookupManagerTest {
assertThat(initialLookuper.closed).isTrue();
assertThat(manager.cachedTableCount()).isZero();
- lookupAndRun(manager, executor, PARTITION_TABLE_INFO);
+ lookup(manager, PARTITION_TABLE_INFO);
assertThat(manager.createdLookupers).hasSize(2);
assertThat(manager.createdClusterConfigs.get(1).toMap())
.containsEntry("datalake.paimon.warehouse", "new-warehouse");
}
- private HistoricalLakeLookupManager createManager(
- int maxQueuedHistoricalRequests, ManualExecutor executor) {
- HistoricalLakeLookupManager manager =
- new HistoricalLakeLookupManager(
- conf(maxQueuedHistoricalRequests),
- null,
- executor,
- ioTmpDir,
- DATA_DIR_VOLUME_BYTES,
- Ticker.systemTicker(),
- Scheduler.disabledScheduler(),
- NO_OP_DISK_WRITE_GUARD);
+ private TestingHistoricalLakeLookupManager createTestingManager() {
+ TestingHistoricalLakeLookupManager manager = new TestingHistoricalLakeLookupManager(conf());
manager.startup(NO_OP_SCHEDULER);
return manager;
}
- private TestingHistoricalLakeLookupManager createTestingManager(ManualExecutor executor) {
- TestingHistoricalLakeLookupManager manager =
- new TestingHistoricalLakeLookupManager(conf(1), executor);
- manager.startup(NO_OP_SCHEDULER);
- return manager;
- }
-
- private Configuration conf(int maxQueuedHistoricalRequests) {
+ private Configuration conf() {
Configuration conf = new Configuration();
- conf.set(
- ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS,
- maxQueuedHistoricalRequests);
conf.set(ConfigOptions.DATA_DIR, ioTmpDir.getAbsolutePath());
return conf;
}
private Configuration confWithExpiration(Duration expiration) {
- Configuration conf = conf(1);
+ Configuration conf = conf();
conf.set(
ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS,
expiration);
@@ -478,13 +317,9 @@ class HistoricalLakeLookupManagerTest {
tableBucket, Collections.singletonList(new byte[] {1}), "2024");
}
- private static CompletableFuture<LookupResultForBucket> lookup(
- HistoricalLakeLookupManager manager, TableInfo tableInfo) {
- return manager.lookup(
- lookupData(new TableBucket(tableInfo.getTableId(), 1L, 0)),
- tableInfo,
- tableInfo.getSchemaInfo(),
- NO_OP_LOOKUP_METRIC_RECORDER);
+ private static void lookup(HistoricalLakeLookupManager manager, TableInfo tableInfo)
+ throws Exception {
+ lookup(manager, tableInfo, tableInfo.getSchemaInfo());
}
private static TableInfo tableInfo(long tableId, int schemaId) {
@@ -498,44 +333,18 @@ class HistoricalLakeLookupManagerTest {
PARTITION_TABLE_INFO.getModifiedTime());
}
- private static void lookupAndRun(
- HistoricalLakeLookupManager manager, ManualExecutor executor, TableInfo tableInfo)
- throws Exception {
- lookupAndRun(manager, executor, tableInfo, tableInfo.getSchemaInfo());
- }
-
- private static void lookupAndRun(
- HistoricalLakeLookupManager manager,
- ManualExecutor executor,
- TableInfo tableInfo,
- SchemaInfo schemaInfo)
- throws Exception {
- LookupResultForBucket result = lookupResultAndRun(manager, executor, tableInfo, schemaInfo);
- assertThat(result.failed()).isFalse();
- assertThat(result.originalPartitionName()).isEqualTo("2024");
- }
-
- private static LookupResultForBucket lookupResultAndRun(
- HistoricalLakeLookupManager manager, ManualExecutor executor, TableInfo tableInfo)
- throws Exception {
- return lookupResultAndRun(manager, executor, tableInfo, tableInfo.getSchemaInfo());
- }
-
- private static LookupResultForBucket lookupResultAndRun(
- HistoricalLakeLookupManager manager,
- ManualExecutor executor,
- TableInfo tableInfo,
- SchemaInfo schemaInfo)
+ private static void lookup(
+ HistoricalLakeLookupManager manager, TableInfo tableInfo, SchemaInfo schemaInfo)
throws Exception {
TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), 1L, 0);
- CompletableFuture<LookupResultForBucket> future =
- manager.lookup(
- lookupData(tableBucket),
- tableInfo,
- schemaInfo,
- NO_OP_LOOKUP_METRIC_RECORDER);
- executor.runNext();
- return future.get(1, TimeUnit.SECONDS);
+ LookupDataForBucket lookupData = lookupData(tableBucket);
+ manager.lookup(
+ lookupData,
+ tableInfo,
+ schemaInfo,
+ ResolvedPartitionSpec.fromPartitionName(
+ tableInfo.getPartitionKeys(), lookupData.originalPartitionName()),
+ NO_OP_LOOKUP_METRIC_RECORDER);
}
private static final class TestingHistoricalLakeLookupManager
@@ -547,11 +356,10 @@ class HistoricalLakeLookupManagerTest {
private final List<Configuration> createdClusterConfigs = new ArrayList<>();
private final long lookupCacheFileBytes;
- private TestingHistoricalLakeLookupManager(Configuration conf, ManualExecutor executor) {
+ private TestingHistoricalLakeLookupManager(Configuration conf) {
super(
conf,
null,
- executor,
new File(conf.get(ConfigOptions.DATA_DIR)),
DATA_DIR_VOLUME_BYTES,
Ticker.systemTicker(),
@@ -561,14 +369,10 @@ class HistoricalLakeLookupManagerTest {
}
private TestingHistoricalLakeLookupManager(
- Configuration conf,
- ManualExecutor executor,
- Ticker ticker,
- Scheduler cacheScheduler) {
+ Configuration conf, Ticker ticker, Scheduler cacheScheduler) {
super(
conf,
null,
- executor,
new File(conf.get(ConfigOptions.DATA_DIR)),
DATA_DIR_VOLUME_BYTES,
ticker,
@@ -579,7 +383,6 @@ class HistoricalLakeLookupManagerTest {
private TestingHistoricalLakeLookupManager(
Configuration conf,
- ManualExecutor executor,
Ticker ticker,
Scheduler cacheScheduler,
long dataDirVolumeBytes,
@@ -587,7 +390,6 @@ class HistoricalLakeLookupManagerTest {
super(
conf,
null,
- executor,
new File(conf.get(ConfigOptions.DATA_DIR)),
dataDirVolumeBytes,
ticker,
@@ -671,55 +473,4 @@ class HistoricalLakeLookupManagerTest {
return null;
}
}
-
- private static final class ManualExecutor extends AbstractExecutorService {
- private final BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<>();
- private volatile boolean shutdown;
-
- @Override
- public void shutdown() {
- shutdown = true;
- }
-
- @Override
- public List<Runnable> shutdownNow() {
- shutdown = true;
- List<Runnable> remainingTasks = new ArrayList<>();
- tasks.drainTo(remainingTasks);
- return remainingTasks;
- }
-
- @Override
- public boolean isShutdown() {
- return shutdown;
- }
-
- @Override
- public boolean isTerminated() {
- return shutdown && tasks.isEmpty();
- }
-
- @Override
- public boolean awaitTermination(long timeout, TimeUnit unit) {
- return isTerminated();
- }
-
- @Override
- public void execute(Runnable command) {
- if (shutdown) {
- throw new RejectedExecutionException();
- }
- tasks.add(command);
- }
-
- private void runNext() throws Exception {
- Runnable task = tasks.poll(1, TimeUnit.SECONDS);
- assertThat(task).isNotNull();
- task.run();
- }
-
- private int numQueuedTasks() {
- return tasks.size();
- }
- }
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java
new file mode 100644
index 000000000..9c69edabe
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionManagerTest.java
@@ -0,0 +1,997 @@
+/*
+ * 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.fluss.server.replica.historical;
+
+import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.AutoPartitionTimeUnit;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.HistoricalPartitionThrottledException;
+import org.apache.fluss.exception.InvalidPartitionException;
+import org.apache.fluss.lake.lakestorage.LakeTableLookuper;
+import org.apache.fluss.metadata.ChangelogImage;
+import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.SchemaGetter;
+import org.apache.fluss.metadata.SchemaInfo;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.BinaryValue;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.record.LogRecords;
+import org.apache.fluss.record.TestingSchemaGetter;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.encode.CompactedKeyEncoder;
+import org.apache.fluss.row.encode.ValueDecoder;
+import org.apache.fluss.row.encode.ValueEncoder;
+import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
+import org.apache.fluss.rpc.entity.LookupResultForBucket;
+import org.apache.fluss.rpc.entity.PutKvResultForBucket;
+import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.rpc.protocol.Errors;
+import org.apache.fluss.rpc.protocol.MergeMode;
+import org.apache.fluss.server.entity.FetchReqInfo;
+import org.apache.fluss.server.entity.LookupDataForBucket;
+import org.apache.fluss.server.entity.NotifyLeaderAndIsrData;
+import org.apache.fluss.server.entity.NotifyLeaderAndIsrResultForBucket;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
+import org.apache.fluss.server.kv.KvStateLookupResult;
+import org.apache.fluss.server.kv.KvTablet;
+import org.apache.fluss.server.log.FetchParams;
+import org.apache.fluss.server.log.LogAppendInfo;
+import org.apache.fluss.server.metadata.BucketMetadata;
+import org.apache.fluss.server.metadata.ClusterMetadata;
+import org.apache.fluss.server.metadata.PartitionMetadata;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.metadata.TableMetadata;
+import org.apache.fluss.server.replica.Replica;
+import org.apache.fluss.server.replica.ReplicaTestBase;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+import org.apache.fluss.server.zk.data.TableRegistration;
+import org.apache.fluss.server.zk.data.lake.LakeTableHelper;
+import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
+import org.apache.fluss.testutils.common.ManuallyTriggeredScheduledExecutorService;
+import org.apache.fluss.types.DataField;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.types.Tuple2;
+
+import com.github.benmanes.caffeine.cache.Scheduler;
+import com.github.benmanes.caffeine.cache.Ticker;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
+import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
+import static org.apache.fluss.server.kv.KvTabletTestUtils.flushAndWait;
+import static org.apache.fluss.server.zk.data.LeaderAndIsr.INITIAL_BUCKET_EPOCH;
+import static org.apache.fluss.server.zk.data.LeaderAndIsr.INITIAL_LEADER_EPOCH;
+import static org.apache.fluss.testutils.DataTestUtils.assertLogRecordsEqualsWithRowKind;
+import static org.apache.fluss.testutils.DataTestUtils.compactedRow;
+import static org.apache.fluss.testutils.DataTestUtils.genKvRecordBatch;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.InternalRowAssert.assertThatRow;
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for historical primary-key writes coordinated by {@link HistoricalPartitionManager}. */
+class HistoricalPartitionManagerTest extends ReplicaTestBase {
+
+ private static final long TABLE_ID = 987654L;
+ private static final long PARTITION_ID = 123L;
+ private static final TablePath TABLE_PATH =
+ TablePath.of("historical_write_db", "historical_write_table");
+ private static final String ORIGINAL_PARTITION = "20240107";
+ private static final String ANOTHER_ORIGINAL_PARTITION = "20240108";
+ private static final String HISTORICAL_PARTITION = HISTORICAL_PARTITION_VALUE;
+ private static final TableBucket TABLE_BUCKET = new TableBucket(TABLE_ID, PARTITION_ID, 0);
+
+ @Test
+ void testResolvesMultipleLakeMissesWithoutPrewriteRollback() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ KvTablet kvTablet = replica.getKvTablet();
+ assertThat(kvTablet).isNotNull();
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ CompactedKeyEncoder keyEncoder = new CompactedKeyEncoder(keyType);
+ byte[] firstKey = keyEncoder.encodeKey(row(1, "us"));
+ byte[] secondKey = keyEncoder.encodeKey(row(2, "eu"));
+ KvRecordBatch insertBatch =
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", ORIGINAL_PARTITION, "v1"}),
+ // The second record reuses the first record's staged state and must not
+ // trigger another lake lookup for the same key.
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", ORIGINAL_PARTITION, "v1-updated"}),
+ Tuple2.of(
+ new Object[] {2, "eu"},
+ new Object[] {2, "eu", ORIGINAL_PARTITION, "v2"}));
+ long truncateCount =
+ replicaManager.getServerMetricGroup().kvTruncateAsErrorCount().getCount();
+
+ try {
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(TABLE_BUCKET, insertBatch, ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+
+ assertThat(lakeLookupManager.lookupCount).hasValue(2);
+ assertThat(lakeLookupManager.lookupBatchCount).hasValue(1);
+ assertThat(replicaManager.getServerMetricGroup().kvTruncateAsErrorCount().getCount())
+ .isEqualTo(truncateCount);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ firstKey,
+ tableInfo,
+ row(1, "us", ORIGINAL_PARTITION, "v1-updated"));
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ secondKey,
+ tableInfo,
+ row(2, "eu", ORIGINAL_PARTITION, "v2"));
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testWalFullRowUpsertDoesNotLookupLake() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader(ChangelogImage.WAL);
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ KvTablet kvTablet = replica.getKvTablet();
+ assertThat(kvTablet).isNotNull();
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ byte[] primaryKey = new CompactedKeyEncoder(keyType).encodeKey(row(1, "us"));
+ KvRecordBatch insertBatch =
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", ORIGINAL_PARTITION, "v1"}));
+
+ try {
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(TABLE_BUCKET, insertBatch, ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+
+ assertThat(lakeLookupManager.lookupCount).hasValue(0);
+ assertThat(lakeLookupManager.lookupBatchCount).hasValue(0);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", ORIGINAL_PARTITION, "v1"));
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testHistoricalInsertUpdateAndDelete() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ KvTablet kvTablet = replica.getKvTablet();
+ assertThat(kvTablet).isNotNull();
+ assertThat(kvManager.getKv(TABLE_BUCKET)).contains(kvTablet);
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ RowType rowType = tableInfo.getRowType();
+ byte[] primaryKey = new CompactedKeyEncoder(keyType).encodeKey(row(1, "us"));
+
+ try {
+ // The first write misses both local state and lake, so it creates a local overlay.
+ KvRecordBatch insertBatch =
+ batch(
+ keyType,
+ rowType,
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", "20240107", "v1"}));
+ assertThat(
+ historicalPartitionManager
+ .processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET, insertBatch, ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1)
+ .lastOffset())
+ .isZero();
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240107", "v1"));
+ assertThat(lakeLookupManager.lookupCount).hasValue(1);
+
+ // The same primary key in another original partition must use a separate state entry.
+ KvRecordBatch anotherPartitionBatch =
+ batch(
+ keyType,
+ rowType,
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", "20240108", "another"}));
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET, anotherPartitionBatch, ANOTHER_ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240107", "v1"));
+ assertHistoricalValue(
+ kvTablet,
+ ANOTHER_ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240108", "another"));
+ assertThat(lakeLookupManager.lookupCount).hasValue(2);
+
+ // Exercise the ReplicaManager entry point; the update should reuse the local overlay.
+ KvRecordBatch updateBatch =
+ batch(
+ keyType,
+ rowType,
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", "20240107", "v2"}));
+ CompletableFuture<List<PutKvResultForBucket>> updateResponse =
+ new CompletableFuture<>();
+ assertThat(replica.tableMetrics().totalHistoricalPutKvRequests().getCount()).isZero();
+ assertThat(replica.tableMetrics().failedHistoricalPutKvRequests().getCount()).isZero();
+ replicaManager.putHistoricalRecordsToKv(
+ 10_000,
+ 1,
+ Collections.singletonList(
+ new PutKvDataForBucket(TABLE_BUCKET, updateBatch, ORIGINAL_PARTITION)),
+ null,
+ MergeMode.DEFAULT,
+ ApiKeys.PUT_KV.highestSupportedVersion,
+ updateResponse::complete);
+ Map<TableBucket, PutKvResultForBucket> updateResults =
+ updateResponse.get(10, TimeUnit.SECONDS).stream()
+ .collect(
+ java.util.stream.Collectors.toMap(
+ PutKvResultForBucket::getTableBucket,
+ result -> result));
+ assertThat(updateResults.get(TABLE_BUCKET).failed()).isFalse();
+ assertThat(replica.tableMetrics().totalHistoricalPutKvRequests().getCount()).isOne();
+ assertThat(replica.tableMetrics().failedHistoricalPutKvRequests().getCount()).isZero();
+ assertHistoricalValue(
+ kvTablet,
+ ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240107", "v2"));
+ assertHistoricalValue(
+ kvTablet,
+ ANOTHER_ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240108", "another"));
+ assertThat(lakeLookupManager.lookupCount).hasValue(2);
+
+ // Historical lookup should observe the updated value from the local overlay.
+ CompletableFuture<List<LookupResultForBucket>> lookupResponse =
+ new CompletableFuture<>();
+ replicaManager.historicalLookups(
+ Collections.singletonList(
+ new LookupDataForBucket(
+ TABLE_BUCKET,
+ Collections.singletonList(primaryKey),
+ ORIGINAL_PARTITION)),
+ lookupResponse::complete);
+ LookupResultForBucket lookupResult = lookupResponse.get(10, TimeUnit.SECONDS).get(0);
+ assertThat(lookupResult.failed()).isFalse();
+ BinaryValue lookedUpValue =
+ new ValueDecoder(
+ schemaGetter(tableInfo),
+ tableInfo.getTableConfig().getKvFormat())
+ .decodeValue(lookupResult.lookupValues().get(0));
+ assertThat(lookedUpValue.row.getString(3)).isEqualTo(BinaryString.fromString("v2"));
+
+ // Keep a tombstone locally so a later lookup cannot resurrect the value from lake.
+ KvRecordBatch deleteBatch =
+ batch(keyType, rowType, Tuple2.of(new Object[] {1, "us"}, null));
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(TABLE_BUCKET, deleteBatch, ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ assertThat(kvTablet.lookupHistoricalLocal(ORIGINAL_PARTITION, primaryKey))
+ .isEqualTo(KvStateLookupResult.deleted());
+ assertHistoricalValue(
+ kvTablet,
+ ANOTHER_ORIGINAL_PARTITION,
+ primaryKey,
+ tableInfo,
+ row(1, "us", "20240108", "another"));
+ assertThat(lakeLookupManager.lookupCount).hasValue(2);
+
+ CompletableFuture<List<LookupResultForBucket>> deletedLookupResponse =
+ new CompletableFuture<>();
+ replicaManager.historicalLookups(
+ Collections.singletonList(
+ new LookupDataForBucket(
+ TABLE_BUCKET,
+ Collections.singletonList(primaryKey),
+ ORIGINAL_PARTITION)),
+ deletedLookupResponse::complete);
+ LookupResultForBucket deletedLookup =
+ deletedLookupResponse.get(10, TimeUnit.SECONDS).get(0);
+ assertThat(deletedLookup.failed()).isFalse();
+ assertThat(deletedLookup.lookupValues()).containsExactly((byte[]) null);
+
+ // Historical replicas must reject the normal KV write path.
+ assertThatThrownBy(
+ () ->
+ replica.putRecordsToLeader(
+ insertBatch, null, MergeMode.DEFAULT, 1))
+ .isInstanceOf(InvalidPartitionException.class);
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testUpdateAndDeleteFromLakeFallback() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ KvTablet kvTablet = replica.getKvTablet();
+ assertThat(kvTablet).isNotNull();
+
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ RowType rowType = tableInfo.getRowType();
+ String updatePartition = "20240109";
+ String deletePartition = "20240110";
+ byte[] updateKey = new CompactedKeyEncoder(keyType).encodeKey(row(1, "us"));
+ byte[] deleteKey = new CompactedKeyEncoder(keyType).encodeKey(row(2, "eu"));
+ short schemaId = (short) tableInfo.getSchemaId();
+ // Seed lake-only values so the first local operations must use lake fallback.
+ lakeLookupManager.putLakeValue(
+ updatePartition,
+ ValueEncoder.encodeValue(
+ schemaId,
+ compactedRow(rowType, new Object[] {1, "us", updatePartition, "lake-v1"})));
+ lakeLookupManager.putLakeValue(
+ deletePartition,
+ ValueEncoder.encodeValue(
+ schemaId,
+ compactedRow(rowType, new Object[] {2, "eu", deletePartition, "lake-v1"})));
+
+ try {
+ // Updating a lake-only value emits the before and after images.
+ KvRecordBatch updateBatch =
+ batch(
+ keyType,
+ rowType,
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", updatePartition, "lake-v2"}));
+ assertThat(
+ historicalPartitionManager
+ .processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET, updateBatch, updatePartition),
+ null,
+ MergeMode.DEFAULT,
+ 1)
+ .numMessages())
+ .isEqualTo(2);
+
+ // Deleting a lake-only value emits a delete and leaves a local tombstone.
+ KvRecordBatch deleteBatch =
+ batch(keyType, rowType, Tuple2.of(new Object[] {2, "eu"}, null));
+ assertThat(
+ historicalPartitionManager
+ .processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET, deleteBatch, deletePartition),
+ null,
+ MergeMode.DEFAULT,
+ 1)
+ .numMessages())
+ .isOne();
+
+ // Verify both lake fallbacks are materialized into the shared historical tablet.
+ flushAndWait(kvTablet, Long.MAX_VALUE);
+ assertHistoricalValue(
+ kvTablet,
+ updatePartition,
+ updateKey,
+ tableInfo,
+ row(1, "us", updatePartition, "lake-v2"));
+ assertThat(kvTablet.lookupHistoricalLocal(deletePartition, deleteKey))
+ .isEqualTo(KvStateLookupResult.deleted());
+ assertThat(lakeLookupManager.lookupCount).hasValue(2);
+
+ assertLogRecordsEqualsWithRowKind(
+ tableInfo.getSchemaId(),
+ rowType,
+ fetchLog(0L),
+ Arrays.asList(
+ Tuple2.of(
+ ChangeType.UPDATE_BEFORE,
+ new Object[] {1, "us", updatePartition, "lake-v1"}),
+ Tuple2.of(
+ ChangeType.UPDATE_AFTER,
+ new Object[] {1, "us", updatePartition, "lake-v2"}),
+ Tuple2.of(
+ ChangeType.DELETE,
+ new Object[] {2, "eu", deletePartition, "lake-v1"})),
+ schemaGetter(tableInfo));
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testLeaderChangeDuringLakeLookupFencesHistoricalWrite() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+ CountDownLatch lakeLookupStarted = new CountDownLatch(1);
+ CountDownLatch finishLakeLookup = new CountDownLatch(1);
+ lakeLookupManager.setLookupHook(
+ () -> {
+ lakeLookupStarted.countDown();
+ await(finishLakeLookup);
+ });
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ KvRecordBatch insertBatch =
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {1, "us", ORIGINAL_PARTITION, "v1"}));
+ long logEndOffsetBeforeWrite = replica.getLocalLogEndOffset();
+
+ try {
+ CompletableFuture<PutKvResultForBucket> writeFuture =
+ historicalPartitionManager.putKv(
+ replica,
+ new PutKvDataForBucket(TABLE_BUCKET, insertBatch, ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ assertThat(lakeLookupStarted.await(10, TimeUnit.SECONDS)).isTrue();
+
+ // Both replica locks must be available while the lake lookup is blocked. Move the
+ // replica away and back so the retry observes a different leader epoch.
+ assertThat(replica.makeFollower(followerState())).isTrue();
+ replica.makeLeader(leaderStateAfterFollower());
+ finishLakeLookup.countDown();
+
+ PutKvResultForBucket result = writeFuture.get(10, TimeUnit.SECONDS);
+ assertThat(result.failed()).isTrue();
+ assertThat(result.getError().error()).isEqualTo(Errors.FENCED_LEADER_EPOCH_EXCEPTION);
+ assertThat(replica.getLocalLogEndOffset()).isEqualTo(logEndOffsetBeforeWrite);
+ } finally {
+ finishLakeLookup.countDown();
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testRecoversHistoricalOverlayFromLakeCommitOffset() throws Exception {
+ TableInfo tableInfo = registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ TestingHistoricalLakeLookupManager lakeLookupManager =
+ new TestingHistoricalLakeLookupManager(lookupConfiguration());
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration()),
+ lakeLookupManager);
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ CompactedKeyEncoder keyEncoder = new CompactedKeyEncoder(keyType);
+ byte[] tieredPrimaryKey = keyEncoder.encodeKey(row(1, "us"));
+ byte[] deletedPrimaryKey = keyEncoder.encodeKey(row(2, "eu"));
+
+ try {
+ LogAppendInfo firstAppend =
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET,
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {
+ 1, "us", ORIGINAL_PARTITION, "v1"
+ })),
+ ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET,
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {1, "us"},
+ new Object[] {
+ 1, "us", ANOTHER_ORIGINAL_PARTITION, "another"
+ })),
+ ANOTHER_ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET,
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(
+ new Object[] {2, "eu"},
+ new Object[] {
+ 2, "eu", ORIGINAL_PARTITION, "delete-me"
+ })),
+ ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ historicalPartitionManager.processPut(
+ replica,
+ new PutKvDataForBucket(
+ TABLE_BUCKET,
+ batch(
+ keyType,
+ tableInfo.getRowType(),
+ Tuple2.of(new Object[] {2, "eu"}, null)),
+ ORIGINAL_PARTITION),
+ null,
+ MergeMode.DEFAULT,
+ 1);
+ KvTablet kvTabletBeforeFollower = replica.getKvTablet();
+ assertThat(kvTabletBeforeFollower).isNotNull();
+ flushAndWait(kvTabletBeforeFollower, Long.MAX_VALUE);
+ assertThat(replica.getLogHighWatermark()).isEqualTo(replica.getLocalLogEndOffset());
+
+ // Persist the exclusive end offset of the first write as the lake recovery point. The
+ // replica has not received this offset locally, so becoming leader must load it before
+ // creating the historical overlay.
+ long lakeCommitOffset = firstAppend.lastOffset() + 1;
+ new LakeTableHelper(zkClient, DEFAULT_REMOTE_DATA_DIR)
+ .registerLakeTableSnapshotV1(
+ TABLE_ID,
+ new LakeTableSnapshot(
+ 1L, Collections.singletonMap(TABLE_BUCKET, lakeCommitOffset)));
+ assertThat(replica.getLakeLogEndOffset()).isEqualTo(-1L);
+
+ // Dropping and recreating the leader KV tablet forces the overlay to be rebuilt only
+ // from WAL after the lake commit offset. The recovered tombstone must remain
+ // authoritative over lake fallback.
+ assertThat(replica.makeFollower(followerState())).isTrue();
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> leaderFuture =
+ new CompletableFuture<>();
+ replicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Collections.singletonList(leaderStateAfterFollower()),
+ leaderFuture::complete);
+ assertThat(leaderFuture.get(10, TimeUnit.SECONDS))
+ .containsOnly(new NotifyLeaderAndIsrResultForBucket(TABLE_BUCKET));
+
+ KvTablet recoveredKvTablet = replica.getKvTablet();
+ assertThat(recoveredKvTablet).isNotNull();
+ assertThat(replica.getLakeLogEndOffset()).isEqualTo(lakeCommitOffset);
+ assertThat(replica.getKvSnapshotManager()).isNull();
+ assertThat(recoveredKvTablet.getFlushedLogOffset())
+ .isEqualTo(replica.getLogHighWatermark());
+ assertThat(recoveredKvTablet.getRocksDBKv().limitScan(10)).hasSize(2);
+ // The first record is covered by the lake commit offset and is not replayed locally.
+ assertThat(
+ recoveredKvTablet.lookupHistoricalLocal(
+ ORIGINAL_PARTITION, tieredPrimaryKey))
+ .isEqualTo(KvStateLookupResult.notFound());
+ assertThat(
+ recoveredKvTablet.lookupHistoricalLocal(
+ ORIGINAL_PARTITION, deletedPrimaryKey))
+ .isEqualTo(KvStateLookupResult.deleted());
+ assertHistoricalValue(
+ recoveredKvTablet,
+ ANOTHER_ORIGINAL_PARTITION,
+ tieredPrimaryKey,
+ tableInfo,
+ row(1, "us", ANOTHER_ORIGINAL_PARTITION, "another"));
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ @Test
+ void testHistoricalLookupThrottledWhenPermitsExhausted() throws Exception {
+ registerHistoricalTableAndBecomeLeader();
+ Replica replica = replicaManager.getReplicaOrException(TABLE_BUCKET);
+ ManuallyTriggeredScheduledExecutorService executor =
+ new ManuallyTriggeredScheduledExecutorService();
+ HistoricalPartitionManager historicalPartitionManager =
+ new HistoricalPartitionManager(
+ new HistoricalPartitionTaskExecutor(lookupConfiguration(), executor),
+ new TestingHistoricalLakeLookupManager(lookupConfiguration()));
+
+ RowType keyType =
+ DataTypes.ROW(
+ new DataField("id", DataTypes.INT()),
+ new DataField("region", DataTypes.STRING()));
+ byte[] primaryKey = new CompactedKeyEncoder(keyType).encodeKey(row(1, "us"));
+ LookupDataForBucket lookupData =
+ new LookupDataForBucket(
+ TABLE_BUCKET, Collections.singletonList(primaryKey), ORIGINAL_PARTITION);
+
+ try {
+ // Keep the first lookup queued so it retains the only available request permit.
+ CompletableFuture<LookupResultForBucket> first =
+ historicalPartitionManager.lookup(
+ replica, lookupData, (lookupTimeNanos, lookupFileDownloaded) -> {});
+ assertThat(first).isNotDone();
+ assertThat(executor.numQueuedRunnables()).isOne();
+ assertThat(historicalPartitionManager.numInflightRequests()).isOne();
+
+ LookupResultForBucket throttled =
+ historicalPartitionManager
+ .lookup(
+ replica,
+ lookupData,
+ (lookupTimeNanos, lookupFileDownloaded) -> {})
+ .get(10, TimeUnit.SECONDS);
+ assertThat(throttled.failed()).isTrue();
+ assertThat(throttled.getError().error())
+ .isEqualTo(Errors.HISTORICAL_PARTITION_THROTTLED);
+ assertThat(throttled.getError().exception())
+ .isInstanceOf(HistoricalPartitionThrottledException.class);
+ assertThat(executor.numQueuedRunnables()).isOne();
+ assertThat(historicalPartitionManager.numInflightRequests()).isOne();
+ } finally {
+ historicalPartitionManager.close();
+ }
+ }
+
+ private LogRecords fetchLog(long fetchOffset) throws Exception {
+ CompletableFuture<Map<TableBucket, FetchLogResultForBucket>> future =
+ new CompletableFuture<>();
+ replicaManager.fetchLogRecords(
+ new FetchParams(-1, Integer.MAX_VALUE),
+ Collections.singletonMap(
+ TABLE_BUCKET, new FetchReqInfo(TABLE_ID, fetchOffset, Integer.MAX_VALUE)),
+ null,
+ future::complete);
+ FetchLogResultForBucket result = future.get(10, TimeUnit.SECONDS).get(TABLE_BUCKET);
+ assertThat(result.failed()).isFalse();
+ return result.records();
+ }
+
+ private TableInfo registerHistoricalTableAndBecomeLeader() throws Exception {
+ return registerHistoricalTableAndBecomeLeader(ChangelogImage.FULL);
+ }
+
+ private TableInfo registerHistoricalTableAndBecomeLeader(ChangelogImage changelogImage)
+ throws Exception {
+ replicaManager.getDiskUsageMonitor().update(0.10);
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("region", DataTypes.STRING())
+ .column("dt", DataTypes.STRING())
+ .column("value", DataTypes.STRING())
+ .primaryKey("id", "region", "dt")
+ .build();
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(1, "id")
+ .partitionedBy("dt")
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_ENABLED, true)
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_KEY, "dt")
+ .property(
+ ConfigOptions.TABLE_AUTO_PARTITION_TIME_UNIT,
+ AutoPartitionTimeUnit.DAY)
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_NUM_RETENTION, 2)
+ .property(ConfigOptions.TABLE_AUTO_PARTITION_TIMEZONE, "UTC")
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .property(ConfigOptions.TABLE_DATALAKE_FORMAT, DataLakeFormat.PAIMON)
+ .property(ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED, true)
+ .property(ConfigOptions.TABLE_CHANGELOG_IMAGE, changelogImage)
+ .property(
+ ConfigOptions.TABLE_KV_FORMAT_VERSION,
+ ConfigOptions.KV_FORMAT_VERSION_2)
+ .build();
+ TableInfo tableInfo =
+ TableInfo.of(TABLE_PATH, TABLE_ID, 1, descriptor, DEFAULT_REMOTE_DATA_DIR, 1L, 1L);
+ zkClient.registerTable(
+ TABLE_PATH,
+ TableRegistration.newTable(TABLE_ID, DEFAULT_REMOTE_DATA_DIR, descriptor));
+ zkClient.registerFirstSchema(TABLE_PATH, schema);
+
+ BucketMetadata bucketMetadata =
+ new BucketMetadata(
+ TABLE_BUCKET.getBucket(),
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Collections.singletonList(TABLET_SERVER_ID));
+ ServerInfo tabletServer =
+ new ServerInfo(
+ TABLET_SERVER_ID,
+ "rack1",
+ Endpoint.fromListenersString("CLIENT://localhost:90"),
+ ServerType.TABLET_SERVER);
+ serverMetadataCache.updateClusterMetadata(
+ new ClusterMetadata(
+ null,
+ Collections.singleton(tabletServer),
+ Collections.singletonList(
+ new TableMetadata(tableInfo, Collections.emptyList())),
+ Collections.singletonList(
+ new PartitionMetadata(
+ TABLE_ID,
+ HISTORICAL_PARTITION,
+ PARTITION_ID,
+ Collections.singletonList(bucketMetadata)))));
+
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> leaderFuture =
+ new CompletableFuture<>();
+ replicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Collections.singletonList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(TABLE_PATH, HISTORICAL_PARTITION),
+ TABLE_BUCKET,
+ Collections.singletonList(TABLET_SERVER_ID),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Collections.singletonList(TABLET_SERVER_ID),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ leaderFuture::complete);
+ assertThat(leaderFuture.get(10, TimeUnit.SECONDS))
+ .containsOnly(new NotifyLeaderAndIsrResultForBucket(TABLE_BUCKET));
+ return tableInfo;
+ }
+
+ private static NotifyLeaderAndIsrData followerState() {
+ int newLeaderId = TABLET_SERVER_ID + 1;
+ List<Integer> replicas = Arrays.asList(TABLET_SERVER_ID, newLeaderId);
+ return new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(TABLE_PATH, HISTORICAL_PARTITION),
+ TABLE_BUCKET,
+ replicas,
+ new LeaderAndIsr(
+ newLeaderId,
+ INITIAL_LEADER_EPOCH + 1,
+ replicas,
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH + 1));
+ }
+
+ private static NotifyLeaderAndIsrData leaderStateAfterFollower() {
+ List<Integer> replicas = Collections.singletonList(TABLET_SERVER_ID);
+ return new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(TABLE_PATH, HISTORICAL_PARTITION),
+ TABLE_BUCKET,
+ replicas,
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH + 2,
+ replicas,
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH + 2));
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ @SafeVarargs
+ private static KvRecordBatch batch(
+ RowType keyType, RowType rowType, Tuple2<Object[], Object[]>... keyAndValues)
+ throws Exception {
+ List<Tuple2<Object[], Object[]>> records = Arrays.asList(keyAndValues);
+ return genKvRecordBatch(keyType, rowType, records);
+ }
+
+ private static void assertHistoricalValue(
+ KvTablet kvTablet,
+ String originalPartition,
+ byte[] primaryKey,
+ TableInfo tableInfo,
+ InternalRow expectedRow)
+ throws Exception {
+ KvStateLookupResult result = kvTablet.lookupHistoricalLocal(originalPartition, primaryKey);
+ assertThat(result.isPresent()).isTrue();
+ BinaryValue value =
+ new ValueDecoder(schemaGetter(tableInfo), tableInfo.getTableConfig().getKvFormat())
+ .decodeValue(result.value());
+ assertThatRow(value.row).withSchema(tableInfo.getRowType()).isEqualTo(expectedRow);
+ }
+
+ private static SchemaGetter schemaGetter(TableInfo tableInfo) {
+ return new TestingSchemaGetter(
+ new SchemaInfo(tableInfo.getSchema(), tableInfo.getSchemaId()));
+ }
+
+ private final class TestingHistoricalLakeLookupManager extends HistoricalLakeLookupManager {
+ private final AtomicInteger lookupCount = new AtomicInteger();
+ private final AtomicInteger lookupBatchCount = new AtomicInteger();
+ private final Map<String, byte[]> lakeValuesByPartition = new HashMap<>();
+ private volatile @Nullable Runnable lookupHook;
+
+ private TestingHistoricalLakeLookupManager(Configuration configuration) {
+ super(
+ configuration,
+ null,
+ new java.io.File(tempDir, "historical-lookup"),
+ 1L,
+ Ticker.systemTicker(),
+ Scheduler.disabledScheduler(),
+ () -> {});
+ }
+
+ private void putLakeValue(String partitionName, byte[] value) {
+ lakeValuesByPartition.put(partitionName, value);
+ }
+
+ private void setLookupHook(Runnable lookupHook) {
+ this.lookupHook = lookupHook;
+ }
+
+ @Override
+ List<byte[]> lookup(
+ LookupDataForBucket lookupData,
+ TableInfo tableInfo,
+ SchemaInfo schemaInfo,
+ ResolvedPartitionSpec originalPartitionSpec,
+ LakeTableLookuper.LookupMetricRecorder lookupMetricRecorder) {
+ lookupBatchCount.incrementAndGet();
+ lookupCount.addAndGet(lookupData.keys().size());
+ Runnable hook = lookupHook;
+ if (hook != null) {
+ hook.run();
+ }
+ List<byte[]> values = new ArrayList<>(lookupData.keys().size());
+ for (int i = 0; i < lookupData.keys().size(); i++) {
+ values.add(lakeValuesByPartition.get(originalPartitionSpec.getPartitionName()));
+ }
+ return values;
+ }
+ }
+
+ private Configuration lookupConfiguration() {
+ Configuration configuration = new Configuration();
+ configuration.set(ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS, 1);
+ return configuration;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutorTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutorTest.java
new file mode 100644
index 000000000..98574e7d9
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/historical/HistoricalPartitionTaskExecutorTest.java
@@ -0,0 +1,232 @@
+/*
+ * 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.fluss.server.replica.historical;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link HistoricalPartitionTaskExecutor}. */
+class HistoricalPartitionTaskExecutorTest {
+
+ @Test
+ void testSharedRequestLimit() throws Exception {
+ ManualExecutor executor = new ManualExecutor();
+ HistoricalPartitionTaskExecutor taskExecutor =
+ new HistoricalPartitionTaskExecutor(configuration(1), executor);
+
+ CompletableFuture<String> first = taskExecutor.submit(() -> "accepted", () -> "throttled");
+ CompletableFuture<String> second = taskExecutor.submit(() -> "accepted", () -> "throttled");
+
+ assertThat(first).isNotDone();
+ assertThat(second).isCompletedWithValue("throttled");
+ assertThat(taskExecutor.numInflightRequests()).isOne();
+
+ executor.runNext();
+ assertThat(first).isCompletedWithValue("accepted");
+ assertThat(taskExecutor.numInflightRequests()).isZero();
+ }
+
+ @Test
+ void testReleasesPermitAfterFailure() throws Exception {
+ ManualExecutor executor = new ManualExecutor();
+ HistoricalPartitionTaskExecutor taskExecutor =
+ new HistoricalPartitionTaskExecutor(configuration(1), executor);
+
+ CompletableFuture<String> failed =
+ taskExecutor.submit(
+ () -> {
+ throw new IllegalStateException("expected");
+ },
+ () -> "throttled");
+ executor.runNext();
+
+ assertThatThrownBy(failed::join).hasCauseInstanceOf(IllegalStateException.class);
+ assertThat(taskExecutor.numInflightRequests()).isZero();
+ assertThat(taskExecutor.submit(() -> "accepted", () -> "throttled")).isNotDone();
+ }
+
+ @Test
+ void testSerializesSameKeyAndAllowsDifferentKeysToRunConcurrently() throws Exception {
+ ManualExecutor executor = new ManualExecutor();
+ HistoricalPartitionTaskExecutor taskExecutor =
+ new HistoricalPartitionTaskExecutor(configuration(3), executor);
+ List<String> executionOrder = new ArrayList<>();
+
+ CompletableFuture<String> first =
+ taskExecutor.submitOrdered(
+ "partition-1",
+ () -> {
+ executionOrder.add("partition-1-first");
+ return "first";
+ },
+ () -> "throttled");
+ CompletableFuture<String> second =
+ taskExecutor.submitOrdered(
+ "partition-1",
+ () -> {
+ executionOrder.add("partition-1-second");
+ return "second";
+ },
+ () -> "throttled");
+ CompletableFuture<String> otherPartition =
+ taskExecutor.submitOrdered(
+ "partition-2",
+ () -> {
+ executionOrder.add("partition-2");
+ return "other";
+ },
+ () -> "throttled");
+
+ assertThat(executor.numQueuedTasks()).isEqualTo(2);
+ executor.runNext();
+ assertThat(first).isCompletedWithValue("first");
+ assertThat(second).isNotDone();
+ assertThat(executor.numQueuedTasks()).isEqualTo(2);
+
+ executor.runNext();
+ executor.runNext();
+ assertThat(otherPartition).isCompletedWithValue("other");
+ assertThat(second).isCompletedWithValue("second");
+ assertThat(executionOrder)
+ .containsExactly("partition-1-first", "partition-2", "partition-1-second");
+ }
+
+ @Test
+ void testOrderedTaskContinuesAfterPreviousFailure() throws Exception {
+ ManualExecutor executor = new ManualExecutor();
+ HistoricalPartitionTaskExecutor taskExecutor =
+ new HistoricalPartitionTaskExecutor(configuration(2), executor);
+
+ CompletableFuture<String> failed =
+ taskExecutor.submitOrdered(
+ "partition",
+ () -> {
+ throw new IllegalStateException("expected");
+ },
+ () -> "throttled");
+ CompletableFuture<String> next =
+ taskExecutor.submitOrdered("partition", () -> "accepted", () -> "throttled");
+
+ assertThat(executor.numQueuedTasks()).isOne();
+ executor.runNext();
+ assertThatThrownBy(failed::join).hasCauseInstanceOf(IllegalStateException.class);
+ assertThat(executor.numQueuedTasks()).isOne();
+
+ executor.runNext();
+ assertThat(next).isCompletedWithValue("accepted");
+ assertThat(taskExecutor.numInflightRequests()).isZero();
+ }
+
+ @Test
+ void testRejectNonPositiveRequestLimit() {
+ assertThatThrownBy(
+ () ->
+ new HistoricalPartitionTaskExecutor(
+ configuration(0), new ManualExecutor()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1})
+ void testRejectNonPositiveThreadPoolSize(int maxThreadPoolSize) {
+ Configuration conf = configuration(1);
+ conf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE, maxThreadPoolSize);
+
+ assertThatThrownBy(() -> new HistoricalPartitionTaskExecutor(conf))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ ConfigOptions.SERVER_HISTORICAL_PARTITION_THREAD_POOL_MAX_SIZE.key());
+ }
+
+ private static Configuration configuration(int maxQueuedHistoricalRequests) {
+ Configuration conf = new Configuration();
+ conf.set(
+ ConfigOptions.NETTY_SERVER_MAX_QUEUED_HISTORICAL_REQUESTS,
+ maxQueuedHistoricalRequests);
+ return conf;
+ }
+
+ private static final class ManualExecutor extends AbstractExecutorService {
+ private final BlockingQueue<Runnable> tasks = new LinkedBlockingQueue<>();
+ private volatile boolean shutdown;
+
+ @Override
+ public void shutdown() {
+ shutdown = true;
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ shutdown = true;
+ List<Runnable> remainingTasks = new ArrayList<>();
+ tasks.drainTo(remainingTasks);
+ return remainingTasks;
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return shutdown;
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return shutdown && tasks.isEmpty();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) {
+ return isTerminated();
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ if (shutdown) {
+ throw new RejectedExecutionException();
+ }
+ tasks.add(command);
+ }
+
+ private void runNext() throws Exception {
+ Runnable task = tasks.poll(1, TimeUnit.SECONDS);
+ assertThat(task).isNotNull();
+ task.run();
+ }
+
+ private int numQueuedTasks() {
+ return tasks.size();
+ }
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
new file mode 100644
index 000000000..c9e6f2304
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.fluss.server.utils;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.rpc.entity.PutKvResultForBucket;
+import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
+import org.apache.fluss.rpc.messages.PbPutKvRespForBucket;
+import org.apache.fluss.rpc.messages.PutKvRequest;
+import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.server.entity.PutKvDataForBucket;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.fluss.record.TestData.DATA_1_WITH_KEY_AND_VALUE;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest;
+import static org.apache.fluss.testutils.DataTestUtils.genKvRecordBatch;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link ServerRpcMessageUtils}. */
+class ServerRpcMessageUtilsTest {
+
+ @Test
+ void testHistoricalPutKvRequestAndResponsePreserveOriginalPartitions() throws Exception {
+ long tableId = 1L;
+ long partitionId = 2L;
+ KvRecordBatch records = genKvRecordBatch(DATA_1_WITH_KEY_AND_VALUE);
+ PutKvRequest request = newPutKvRequest(tableId, 0, 1, records);
+ PbPutKvReqForBucket bucketRequest = request.getBucketsReqsList().get(0);
+ bucketRequest.setPartitionId(partitionId).setOriginalPartitionName("dt=2025-01-01");
+ request.addAllBucketsReqs(
+ Collections.singletonList(
+ new PbPutKvReqForBucket()
+ .copyFrom(bucketRequest)
+ .setOriginalPartitionName("dt=2025-01-02")));
+
+ TableBucket tableBucket = new TableBucket(tableId, partitionId, 0);
+ List<PutKvDataForBucket> decoded = ServerRpcMessageUtils.toPutKvDataForBuckets(request);
+ assertThat(decoded).extracting(PutKvDataForBucket::tableBucket).containsOnly(tableBucket);
+ assertThat(decoded)
+ .extracting(PutKvDataForBucket::originalPartitionName)
+ .containsExactly("dt=2025-01-01", "dt=2025-01-02");
+ assertThat(decoded).extracting(PutKvDataForBucket::records).containsOnly(records);
+
+ PutKvResponse response =
+ ServerRpcMessageUtils.makePutKvResponse(
+ Arrays.asList(
+ PutKvResultForBucket.historicalSuccess(
+ tableBucket, 1L, "dt=2025-01-01"),
+ PutKvResultForBucket.historicalSuccess(
+ tableBucket, 2L, "dt=2025-01-02")));
+ assertThat(response.getBucketsRespsList())
+ .extracting(PbPutKvRespForBucket::getOriginalPartitionName)
+ .containsExactly("dt=2025-01-01", "dt=2025-01-02");
+
+ request.addAllBucketsReqs(
+ Collections.singletonList(new PbPutKvReqForBucket().copyFrom(bucketRequest)));
+ assertThatThrownBy(() -> ServerRpcMessageUtils.toPutKvDataForBuckets(request))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("duplicate table bucket");
+ }
+}
diff --git a/website/docs/maintenance/observability/monitor-metrics.md b/website/docs/maintenance/observability/monitor-metrics.md
index 4bb313e61..dd7f0dbf6 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -699,6 +699,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
<td rowspan="7">
request_produceLog
request_putKv
+ request_historicalPutKv
request_lookup
request_historicalLookup
request_prefixLookup
@@ -789,7 +790,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
</thead>
<tbody>
<tr>
- <th rowspan="37"><strong>tabletserver</strong></th>
+ <th rowspan="39"><strong>tabletserver</strong></th>
<td rowspan="20">table</td>
<td>messagesInPerSecond</td>
<td>The number of messages written per second to this table.</td>
@@ -891,7 +892,17 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM
<td>Meter</td>
</tr>
<tr>
- <td rowspan="4">table_historical</td>
+ <td rowspan="6">table_historical</td>
+ <td>totalPutKvRequestsPerSecond</td>
+ <td>The number of historical put kv requests to this table per second.</td>
+ <td>Meter</td>
+ </tr>
+ <tr>
+ <td>failedPutKvRequestsPerSecond</td>
+ <td>The number of historical put kv requests that failed unexpectedly for this table per second.</td>
+ <td>Meter</td>
+ </tr>
+ <tr>
<td>totalLookupRequestsPerSecond</td>
<td>The number of historical lookup requests to this table per second.</td>
<td>Meter</td>