* Copyright (c) Huawei Technologies Co., Ltd. 2026. All rights reserved.
*
* Licensed 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.
*/
* Description: Slot compact artifact builder.
*/
#include "datasystem/common/l2cache/slot_client/slot_compactor.h"
#include <algorithm>
#include <unordered_map>
#include <unordered_set>
#include <fcntl.h>
#include <unistd.h>
#include "datasystem/common/l2cache/slot_client/slot_file_util.h"
#include "datasystem/common/l2cache/slot_client/slot_index_codec.h"
#include "datasystem/common/log/log.h"
#include "datasystem/common/util/file_util.h"
#include "datasystem/common/util/raii.h"
#include "datasystem/common/util/status_helper.h"
namespace datasystem {
namespace {
constexpr size_t SLOT_IO_CHUNK_BYTES = 4UL * 1024UL * 1024UL;
std::string BaseName(const std::string &path)
{
auto pos = path.find_last_of('/');
return pos == std::string::npos ? path : path.substr(pos + 1);
}
}
SlotCompactor::SlotCompactor(std::string slotPath, uint64_t maxDataFileBytes)
: slotPath_(std::move(slotPath)), maxDataFileBytes_(maxDataFileBytes)
{
}
Status SlotCompactor::CopyRecordToTargetFile(int sourceFd, uint64_t sourceOffset, uint64_t size, int targetFd,
uint64_t targetOffset) const
{
CHECK_FAIL_RETURN_STATUS(sourceFd >= 0 && targetFd >= 0, StatusCode::K_INVALID, "invalid compact file fd");
std::vector<char> buffer(std::min<uint64_t>(size, SLOT_IO_CHUNK_BYTES));
uint64_t readOffset = sourceOffset;
uint64_t writeOffset = targetOffset;
uint64_t remaining = size;
while (remaining > 0) {
auto bytesToCopy = static_cast<size_t>(std::min<uint64_t>(remaining, buffer.size()));
RETURN_IF_NOT_OK(ReadFile(sourceFd, buffer.data(), bytesToCopy, static_cast<off_t>(readOffset)));
RETURN_IF_NOT_OK(WriteFile(targetFd, buffer.data(), bytesToCopy, static_cast<off_t>(writeOffset)));
readOffset += bytesToCopy;
writeOffset += bytesToCopy;
remaining -= bytesToCopy;
}
return Status::OK();
}
Status SlotCompactor::BuildArtifacts(const SlotManifestData &manifest, const SlotSnapshot &snapshot,
uint64_t compactEpochMs, SlotCompactBuildResult &result)
{
VLOG(1) << "Building slot compact artifacts, slotPath=" << slotPath_ << ", compactEpochMs=" << compactEpochMs
<< ", activeIndex=" << manifest.activeIndex << ", activeDataCount=" << manifest.activeData.size();
result = SlotCompactBuildResult{};
SlotCompactBuildResult buildResult;
bool keepArtifacts = false;
Raii cleanupOnFailure([this, &buildResult, &keepArtifacts]() {
if (!keepArtifacts) {
(void)CleanupArtifacts(buildResult);
}
});
RETURN_IF_NOT_OK(AllocateCompactIndexFile(manifest, compactEpochMs, buildResult.indexFile));
auto indexPath = JoinPath(slotPath_, buildResult.indexFile);
RETURN_IF_NOT_OK(SlotIndexCodec::EnsureIndexFile(indexPath));
int indexFd = -1;
RETURN_IF_NOT_OK(OpenFile(indexPath, O_RDWR, &indexFd));
Raii closeIndexFd([&indexFd]() {
if (indexFd >= 0) {
close(indexFd);
indexFd = -1;
}
});
uint64_t indexOffset = static_cast<uint64_t>(FdFileSize(indexFd));
std::vector<SlotPutRecord> visiblePuts;
RETURN_IF_NOT_OK(snapshot.CollectVisiblePuts(visiblePuts));
VLOG(1) << "Collected visible puts for slot compact, slotPath=" << slotPath_
<< ", visiblePutCount=" << visiblePuts.size();
std::sort(visiblePuts.begin(), visiblePuts.end(), [](const SlotPutRecord &lhs, const SlotPutRecord &rhs) {
if (lhs.fileId != rhs.fileId) {
return lhs.fileId < rhs.fileId;
}
if (lhs.offset != rhs.offset) {
return lhs.offset < rhs.offset;
}
if (lhs.key != rhs.key) {
return lhs.key < rhs.key;
}
return lhs.version < rhs.version;
});
uint32_t nextFileId = 0;
RETURN_IF_NOT_OK(GetNextDataFileId(manifest, nextFileId));
auto openTargetDataFile = [this, &buildResult](uint32_t fileId, int &fd, uint64_t &fileSize) -> Status {
if (fd >= 0) {
RETURN_IF_NOT_OK(FsyncFd(fd));
close(fd);
fd = -1;
}
auto filename = FormatDataFileName(fileId);
buildResult.dataFiles.emplace_back(filename);
auto dataPath = JoinPath(slotPath_, filename);
RETURN_IF_NOT_OK(EnsureFile(dataPath));
RETURN_IF_NOT_OK(OpenFile(dataPath, O_RDWR, &fd));
fileSize = static_cast<uint64_t>(FdFileSize(fd));
return Status::OK();
};
int currentTargetDataFd = -1;
Raii closeTargetDataFd([¤tTargetDataFd]() {
if (currentTargetDataFd >= 0) {
close(currentTargetDataFd);
currentTargetDataFd = -1;
}
});
uint64_t currentTargetFileSize = 0;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
if (visiblePuts.empty()) {
RETURN_IF_NOT_OK(FsyncFd(indexFd));
RETURN_IF_NOT_OK(FsyncFd(currentTargetDataFd));
RETURN_IF_NOT_OK(FsyncDir(slotPath_));
keepArtifacts = true;
result = buildResult;
VLOG(1) << "Built empty slot compact artifacts, slotPath=" << slotPath_
<< ", pendingIndex=" << buildResult.indexFile
<< ", pendingDataCount=" << buildResult.dataFiles.size();
return Status::OK();
}
std::unordered_map<uint32_t, int> sourceDataFds;
Raii closeSourceFds([&sourceDataFds]() {
for (auto &entry : sourceDataFds) {
if (entry.second >= 0) {
close(entry.second);
}
}
});
std::string indexBuffer;
indexBuffer.reserve(1024 * 1024);
for (const auto &record : visiblePuts) {
auto fdIt = sourceDataFds.find(record.fileId);
if (fdIt == sourceDataFds.end()) {
int sourceFd = -1;
RETURN_IF_NOT_OK(OpenFile(JoinPath(slotPath_, FormatDataFileName(record.fileId)), O_RDONLY, &sourceFd));
fdIt = sourceDataFds.emplace(record.fileId, sourceFd).first;
}
bool dedicateFile = record.size >= maxDataFileBytes_;
if (!dedicateFile && currentTargetFileSize + record.size > maxDataFileBytes_ && currentTargetFileSize > 0) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
auto offset = currentTargetFileSize;
uint32_t targetFileId = nextFileId;
if (dedicateFile && currentTargetFileSize > 0) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
offset = 0;
targetFileId = nextFileId;
}
RETURN_IF_NOT_OK(CopyRecordToTargetFile(fdIt->second, record.offset, record.size, currentTargetDataFd, offset));
SlotPutRecord compacted = record;
compacted.fileId = targetFileId;
compacted.offset = offset;
std::string encoded;
RETURN_IF_NOT_OK(SlotIndexCodec::EncodePut(compacted, encoded));
indexBuffer.append(encoded);
if (indexBuffer.size() >= 1024 * 1024) {
RETURN_IF_NOT_OK(SlotIndexCodec::AppendEncodedRecords(indexFd, indexOffset, indexBuffer));
indexBuffer.clear();
}
currentTargetFileSize = offset + record.size;
if (dedicateFile) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
}
RETURN_IF_NOT_OK(SlotIndexCodec::AppendEncodedRecords(indexFd, indexOffset, indexBuffer));
RETURN_IF_NOT_OK(FsyncFd(indexFd));
RETURN_IF_NOT_OK(FsyncFd(currentTargetDataFd));
RETURN_IF_NOT_OK(FsyncDir(slotPath_));
keepArtifacts = true;
result = buildResult;
VLOG(1) << "Built slot compact artifacts successfully, slotPath=" << slotPath_
<< ", pendingIndex=" << buildResult.indexFile
<< ", pendingDataCount=" << buildResult.dataFiles.size();
return Status::OK();
}
Status SlotCompactor::ApplyDeltaRecords(const std::vector<SlotRecord> &records, SlotCompactBuildResult &result)
{
VLOG(1) << "Applying slot compact delta records, slotPath=" << slotPath_ << ", recordCount=" << records.size()
<< ", pendingIndex=" << result.indexFile << ", pendingDataCount=" << result.dataFiles.size();
CHECK_FAIL_RETURN_STATUS(!result.indexFile.empty(), StatusCode::K_INVALID,
"Compact build result missing pending index file");
if (records.empty()) {
return Status::OK();
}
auto indexPath = JoinPath(slotPath_, result.indexFile);
RETURN_IF_NOT_OK(SlotIndexCodec::EnsureIndexFile(indexPath));
int indexFd = -1;
RETURN_IF_NOT_OK(OpenFile(indexPath, O_RDWR, &indexFd));
Raii closeIndexFd([&indexFd]() {
if (indexFd >= 0) {
close(indexFd);
indexFd = -1;
}
});
uint64_t indexOffset = static_cast<uint64_t>(FdFileSize(indexFd));
uint32_t nextFileId = 0;
if (!result.dataFiles.empty()) {
RETURN_IF_NOT_OK(ParseDataFileId(result.dataFiles.back(), nextFileId));
}
auto openTargetDataFile = [this, &result](uint32_t fileId, int &fd, uint64_t &fileSize) -> Status {
if (fd >= 0) {
RETURN_IF_NOT_OK(FsyncFd(fd));
close(fd);
fd = -1;
}
auto filename = FormatDataFileName(fileId);
if (result.dataFiles.empty() || result.dataFiles.back() != filename) {
result.dataFiles.emplace_back(filename);
}
auto dataPath = JoinPath(slotPath_, filename);
RETURN_IF_NOT_OK(EnsureFile(dataPath));
RETURN_IF_NOT_OK(OpenFile(dataPath, O_RDWR, &fd));
fileSize = static_cast<uint64_t>(FdFileSize(fd));
return Status::OK();
};
int currentTargetDataFd = -1;
Raii closeTargetDataFd([¤tTargetDataFd]() {
if (currentTargetDataFd >= 0) {
close(currentTargetDataFd);
currentTargetDataFd = -1;
}
});
uint64_t currentTargetFileSize = 0;
if (!result.dataFiles.empty()) {
RETURN_IF_NOT_OK(ParseDataFileId(result.dataFiles.back(), nextFileId));
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
std::unordered_map<uint32_t, int> sourceDataFds;
Raii closeSourceFds([&sourceDataFds]() {
for (auto &entry : sourceDataFds) {
if (entry.second >= 0) {
close(entry.second);
}
}
});
std::string indexBuffer;
indexBuffer.reserve(1024 * 1024);
for (const auto &record : records) {
switch (record.type) {
case SlotRecordType::PUT: {
auto fdIt = sourceDataFds.find(record.put.fileId);
if (fdIt == sourceDataFds.end()) {
int sourceFd = -1;
RETURN_IF_NOT_OK(
OpenFile(JoinPath(slotPath_, FormatDataFileName(record.put.fileId)), O_RDONLY, &sourceFd));
fdIt = sourceDataFds.emplace(record.put.fileId, sourceFd).first;
}
if (currentTargetDataFd < 0) {
nextFileId = nextFileId == 0 ? 1 : nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
bool dedicateFile = record.put.size >= maxDataFileBytes_;
if (!dedicateFile && currentTargetFileSize + record.put.size > maxDataFileBytes_
&& currentTargetFileSize > 0) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
auto offset = currentTargetFileSize;
uint32_t targetFileId = nextFileId;
if (dedicateFile && currentTargetFileSize > 0) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
offset = 0;
targetFileId = nextFileId;
}
RETURN_IF_NOT_OK(CopyRecordToTargetFile(fdIt->second, record.put.offset, record.put.size,
currentTargetDataFd, offset));
SlotPutRecord compacted = record.put;
compacted.fileId = targetFileId;
compacted.offset = offset;
std::string encoded;
RETURN_IF_NOT_OK(SlotIndexCodec::EncodePut(compacted, encoded));
indexBuffer.append(encoded);
currentTargetFileSize = offset + record.put.size;
if (dedicateFile) {
++nextFileId;
RETURN_IF_NOT_OK(openTargetDataFile(nextFileId, currentTargetDataFd, currentTargetFileSize));
}
break;
}
case SlotRecordType::DELETE: {
std::string encoded;
RETURN_IF_NOT_OK(SlotIndexCodec::EncodeDelete(record.del, encoded));
indexBuffer.append(encoded);
break;
}
case SlotRecordType::IMPORT_BEGIN:
case SlotRecordType::IMPORT_END:
RETURN_STATUS(StatusCode::K_TRY_AGAIN, "Compact delta stream hit concurrent import records");
}
if (indexBuffer.size() >= 1024 * 1024) {
RETURN_IF_NOT_OK(SlotIndexCodec::AppendEncodedRecords(indexFd, indexOffset, indexBuffer));
indexBuffer.clear();
}
}
RETURN_IF_NOT_OK(SlotIndexCodec::AppendEncodedRecords(indexFd, indexOffset, indexBuffer));
RETURN_IF_NOT_OK(FsyncFd(indexFd));
if (currentTargetDataFd >= 0) {
RETURN_IF_NOT_OK(FsyncFd(currentTargetDataFd));
}
RETURN_IF_NOT_OK(FsyncDir(slotPath_));
VLOG(1) << "Applied slot compact delta records successfully, slotPath=" << slotPath_
<< ", recordCount=" << records.size() << ", pendingDataCount=" << result.dataFiles.size();
return Status::OK();
}
Status SlotCompactor::CleanupArtifacts(const SlotCompactBuildResult &result)
{
VLOG(1) << "Cleaning slot compact artifacts, slotPath=" << slotPath_ << ", indexFile=" << result.indexFile
<< ", dataFileCount=" << result.dataFiles.size();
if (!result.indexFile.empty() && FileExist(JoinPath(slotPath_, result.indexFile))) {
(void)DeleteFile(JoinPath(slotPath_, result.indexFile));
}
for (const auto &dataFile : result.dataFiles) {
if (FileExist(JoinPath(slotPath_, dataFile))) {
(void)DeleteFile(JoinPath(slotPath_, dataFile));
}
}
return Status::OK();
}
Status SlotCompactor::AllocateCompactIndexFile(const SlotManifestData &manifest, uint64_t compactEpochMs,
std::string &indexFile) const
{
std::unordered_set<std::string> reservedFiles;
reservedFiles.insert(manifest.activeIndex);
if (!manifest.pendingIndex.empty()) {
reservedFiles.insert(manifest.pendingIndex);
}
if (!manifest.obsoleteIndex.empty()) {
reservedFiles.insert(manifest.obsoleteIndex);
}
uint64_t candidateEpochMs = compactEpochMs;
for (size_t attempt = 0; attempt < 1024; ++attempt, ++candidateEpochMs) {
auto candidate = FormatCompactIndexFileName(candidateEpochMs);
if (reservedFiles.count(candidate) > 0 || FileExist(JoinPath(slotPath_, candidate))) {
continue;
}
indexFile = candidate;
return Status::OK();
}
RETURN_STATUS(StatusCode::K_RUNTIME_ERROR, "No available compact index file for build");
}
Status SlotCompactor::GetNextDataFileId(const SlotManifestData &manifest, uint32_t &nextFileId) const
{
std::vector<std::string> dataPaths;
RETURN_IF_NOT_OK(Glob(JoinPath(slotPath_, "data_*.bin"), dataPaths));
uint32_t maxFileId = 0;
auto updateMaxId = [&maxFileId](const std::string &relativeName) {
uint32_t fileId = 0;
auto rc = ParseDataFileId(relativeName, fileId);
if (rc.IsOk()) {
maxFileId = std::max(maxFileId, fileId);
}
};
for (const auto &dataPath : dataPaths) {
updateMaxId(BaseName(dataPath));
}
for (const auto &dataFile : manifest.activeData) {
updateMaxId(dataFile);
}
for (const auto &dataFile : manifest.pendingData) {
updateMaxId(dataFile);
}
for (const auto &dataFile : manifest.obsoleteData) {
updateMaxId(dataFile);
}
nextFileId = maxFileId == 0 ? 1 : maxFileId + 1;
return Status::OK();
}
}