blob: 5c00348273fb581000b9f9dd9706f08172704d82 [file]
/*
* 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.
*/
/*
* Copyright (c) Facebook, Inc. and its affiliates.
*
* 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.
*/
#include "compute/delta/DeltaDeletionVectorReader.h"
#include <cstring>
#include "velox/common/base/BitUtil.h"
#include "velox/common/base/Exceptions.h"
namespace gluten::delta {
namespace {
constexpr uint64_t kDeltaBitmapArrayMagicBytes = 4;
constexpr uint64_t kDeltaNativeBitmapArrayLengthBytes = 4;
constexpr uint64_t kDeltaStoredPayloadLengthBytes = 4;
constexpr uint32_t kDeltaPortableBitmapArrayMagicNumber = 1681511377;
constexpr uint32_t kDeltaNativeBitmapArrayMagicNumber = 1681511376;
uint32_t readUint32LittleEndian(const char* data) {
const auto* bytes = reinterpret_cast<const uint8_t*>(data);
return static_cast<uint32_t>(bytes[0]) | (static_cast<uint32_t>(bytes[1]) << 8) |
(static_cast<uint32_t>(bytes[2]) << 16) | (static_cast<uint32_t>(bytes[3]) << 24);
}
roaring::Roaring64Map deserializeDeltaBitmapArray(std::string_view serializedPayload, const std::string& dvPath) {
VELOX_USER_CHECK_GE(
serializedPayload.size(),
kDeltaBitmapArrayMagicBytes,
"Deletion vector payload is too small for Delta bitmap array: {}",
dvPath);
const auto magic = readUint32LittleEndian(serializedPayload.data());
if (magic == kDeltaPortableBitmapArrayMagicNumber) {
const auto portablePayload = serializedPayload.substr(kDeltaBitmapArrayMagicBytes);
return roaring::Roaring64Map::readSafe(portablePayload.data(), portablePayload.size());
}
if (magic == kDeltaNativeBitmapArrayMagicNumber) {
VELOX_USER_CHECK_GE(
serializedPayload.size(),
kDeltaBitmapArrayMagicBytes + kDeltaNativeBitmapArrayLengthBytes,
"Deletion vector payload is too small for Delta native bitmap array: {}",
dvPath);
const auto bitmapCount = readUint32LittleEndian(serializedPayload.data() + kDeltaBitmapArrayMagicBytes);
size_t offset = kDeltaBitmapArrayMagicBytes + kDeltaNativeBitmapArrayLengthBytes;
roaring::Roaring64Map result;
for (uint64_t bitmapIndex = 0; bitmapIndex < bitmapCount; ++bitmapIndex) {
VELOX_USER_CHECK_LE(
offset + kDeltaStoredPayloadLengthBytes,
serializedPayload.size(),
"Deletion vector payload ended before bitmap {} size for {}",
bitmapIndex,
dvPath);
const auto bitmapSize = readUint32LittleEndian(serializedPayload.data() + offset);
offset += kDeltaStoredPayloadLengthBytes;
VELOX_USER_CHECK_LE(
offset + bitmapSize,
serializedPayload.size(),
"Deletion vector bitmap {} range exceeds payload size for {}",
bitmapIndex,
dvPath);
auto bitmap = roaring::Roaring::readSafe(serializedPayload.data() + offset, bitmapSize);
VELOX_USER_CHECK_EQ(
bitmap.getSizeInBytes(true),
bitmapSize,
"Deletion vector bitmap {} size mismatch for {}: expected {}, got {}",
bitmapIndex,
dvPath,
bitmapSize,
bitmap.getSizeInBytes(true));
const uint64_t rowBase = bitmapIndex << 32;
for (auto it = bitmap.begin(); it != bitmap.end(); ++it) {
result.add(rowBase | static_cast<uint64_t>(*it));
}
offset += bitmapSize;
}
VELOX_USER_CHECK_EQ(
offset,
serializedPayload.size(),
"Deletion vector payload has {} unexpected trailing bytes for {}",
serializedPayload.size() - offset,
dvPath);
return result;
}
VELOX_USER_FAIL("Unexpected Delta bitmap array magic number {} for {}", magic, dvPath);
}
} // namespace
void DeltaDeletionVectorReader::loadSerializedDeletionVectorInternal(
std::string_view serializedPayload,
const std::string& debugName,
std::optional<uint64_t> expectedCardinality) {
VELOX_USER_CHECK_GT(serializedPayload.size(), 0, "Serialized deletion vector is empty: {}", debugName);
deletionBitmap_ = deserializeDeltaBitmapArray(serializedPayload, debugName);
if (expectedCardinality.has_value()) {
const auto actualCardinality = deletionBitmap_->cardinality();
VELOX_USER_CHECK_EQ(
actualCardinality,
expectedCardinality.value(),
"Deletion vector cardinality mismatch for {}: expected {}, got {}",
debugName,
expectedCardinality.value(),
actualCardinality);
}
}
void DeltaDeletionVectorReader::loadSerializedDeletionVector(
std::string_view serializedPayload,
std::optional<uint64_t> expectedCardinality) {
try {
loadSerializedDeletionVectorInternal(serializedPayload, "serialized deletion vector", expectedCardinality);
} catch (const std::exception& e) {
VELOX_USER_FAIL("Failed to load serialized deletion vector: {}", e.what());
}
}
bool DeltaDeletionVectorReader::isRowDeleted(uint64_t rowPosition) {
if (!deletionBitmap_.has_value()) {
return false;
}
return deletionBitmap_->contains(rowPosition);
}
void DeltaDeletionVectorReader::applyDeletionFilter(uint64_t baseReadOffset, uint64_t size, BufferPtr deleteBitmap) {
VELOX_CHECK_NOT_NULL(deleteBitmap, "Delete bitmap buffer is required");
if (!deletionBitmap_.has_value()) {
std::memset(deleteBitmap->asMutable<uint8_t>(), 0, bits::nbytes(size));
deleteBitmap->setSize(0);
return;
}
auto* rawBitmap = deleteBitmap->asMutable<uint64_t>();
std::memset(rawBitmap, 0, bits::nbytes(size));
// Use an iterator-based approach instead of per-row contains() lookups.
// This is O(deletions_in_range) rather than O(batch_size), which is
// significantly faster when deletions are sparse relative to batch size.
// Guard against uint64_t overflow when baseReadOffset is near UINT64_MAX.
const uint64_t rangeEnd = (size <= UINT64_MAX - baseReadOffset) ? baseReadOffset + size : UINT64_MAX;
auto it = deletionBitmap_->begin();
if (!it.move_equalorlarger(baseReadOffset)) {
// No deleted rows at or after baseReadOffset — nothing to mark.
deleteBitmap->setSize(0);
return;
}
bool hasDeletedRows = false;
uint64_t highestDeletedIndex = 0;
while (it != deletionBitmap_->end()) {
const uint64_t absoluteRowPos = *it;
if (absoluteRowPos >= rangeEnd) {
break;
}
const uint64_t relativeIndex = absoluteRowPos - baseReadOffset;
bits::setBit(rawBitmap, relativeIndex);
hasDeletedRows = true;
highestDeletedIndex = relativeIndex;
++it;
}
deleteBitmap->setSize(hasDeletedRows ? bits::nbytes(highestDeletedIndex + 1) : 0);
}
uint64_t DeltaDeletionVectorReader::estimatedDeletedRowCount() const {
if (!deletionBitmap_.has_value()) {
return 0;
}
// Return actual cardinality instead of estimated size
return deletionBitmap_->cardinality();
}
} // namespace gluten::delta