diff --git a/.github/workflows/ci-musl-test-workflow.yml b/.github/workflows/ci-musl-test-workflow.yml index fe30ec393..d1ee0a604 100644 --- a/.github/workflows/ci-musl-test-workflow.yml +++ b/.github/workflows/ci-musl-test-workflow.yml @@ -205,12 +205,14 @@ jobs: CMAKE_CXX_COMPILER_LAUNCHER: ccache run: su-exec "$RUN_AS_USER:$RUN_AS_GROUP" env HOME="$RUN_AS_HOME" sh -lc 'make relwithdebinfo' - - name: Generate datasets - run: su-exec "$RUN_AS_USER:$RUN_AS_GROUP" env HOME="$RUN_AS_HOME" sh -lc 'bash scripts/generate_binary_demo.sh --lbug-shell-mode relwithdebinfo' - - name: Install uv run: pip3 install uv + - name: Generate datasets + run: | + su-exec "$RUN_AS_USER:$RUN_AS_GROUP" env HOME="$RUN_AS_HOME" sh -lc 'bash scripts/generate_binary_demo.sh --lbug-shell-mode relwithdebinfo' + su-exec "$RUN_AS_USER:$RUN_AS_GROUP" env HOME="$RUN_AS_HOME" sh -lc 'uv run --with pyarrow python3 scripts/generate-dictionary-bug-fixture.py' + - name: Test env: CMAKE_C_COMPILER_LAUNCHER: ccache diff --git a/.github/workflows/ci-workflow.yml b/.github/workflows/ci-workflow.yml index 58729192a..94b0b7811 100644 --- a/.github/workflows/ci-workflow.yml +++ b/.github/workflows/ci-workflow.yml @@ -256,7 +256,9 @@ jobs: - name: Generate datasets if: matrix.platform != 'windows' - run: bash scripts/generate_binary_demo.sh --lbug-shell-mode relwithdebinfo + run: | + bash scripts/generate_binary_demo.sh --lbug-shell-mode relwithdebinfo + uv run --with pyarrow python3 scripts/generate-dictionary-bug-fixture.py - name: Generate datasets (Windows) if: matrix.platform == 'windows' @@ -265,6 +267,7 @@ jobs: LBUG_SHELL="$(find build -path '*/src/lbug_shell.exe' -print -quit)" test -n "$LBUG_SHELL" bash scripts/generate_binary_demo.sh --lbug-shell "$LBUG_SHELL" + uv run --with pyarrow python3 scripts/generate-dictionary-bug-fixture.py - name: Test shell: bash diff --git a/.gitignore b/.gitignore index 0181ba504..cc2d4b906 100644 --- a/.gitignore +++ b/.gitignore @@ -90,3 +90,6 @@ scripts/generate-cpp-docs/c/docs dataset/binary-demo dataset/databases/tinysnb dataset/ldbc-1/csv + +# Generated dictionary-bug e2e parquet fixtures +test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/*.parquet diff --git a/scripts/generate-dictionary-bug-fixture.py b/scripts/generate-dictionary-bug-fixture.py new file mode 100644 index 000000000..33d696fcd --- /dev/null +++ b/scripts/generate-dictionary-bug-fixture.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Generate parquet fixtures for the relationship property corruption e2e. + +The e2e in +``test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated.test`` +needs a specific import layout. Sequential ``id = 1..N`` node tables do not +reproduce the pre-fix failure; the minimized node insertion order and the +exact 953-row relationship table do. + +This script reconstructs that layout from compact zlib payloads next to the +generated parquet files under +``test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/`` +and writes: + +- ``A.parquet`` / ``B.parquet``: 34,747 nodes with the original insertion order +- ``A_TO_B.parquet``: 953 relationships with the original endpoints and types + +Generated parquet files should not be committed (see ``.gitignore``). + +Usage: + uv run --with pyarrow python3 scripts/generate-dictionary-bug-fixture.py +""" + +from __future__ import annotations + +import argparse +import hashlib +import os +import struct +import sys +import zlib + +import pyarrow as pa +import pyarrow.parquet as pq + +SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) +PROJECT_ROOT = os.path.dirname(SCRIPT_DIR) +DEFAULT_FIXTURE_DIR = os.path.join( + PROJECT_ROOT, + "test", + "test_files", + "dictionary_bug", + "orb383_relationship_projection_obfuscated_fixture", +) + +NODE_IDS_PAYLOAD = os.path.join(DEFAULT_FIXTURE_DIR, "node_ids.u32le.zlib") +RELS_PAYLOAD = os.path.join(DEFAULT_FIXTURE_DIR, "rels.u32u32u8.zlib") + +NUM_NODES = 34747 +NUM_RELS = 953 +TYPE_BY_CODE = {0: "REL_A", 1: "REL_B", 2: "REL_C"} +EXPECTED_TYPE_COUNTS = {"REL_A": 739, "REL_B": 202, "REL_C": 12} +EXPECTED_NODE_IDS_SHA256 = ( + "9563bacdbfe3afc53a7c9c0a377e8a9cdb617794171c56ffc1bcd205e65f9ed8" +) +EXPECTED_RELS_SHA256 = ( + "43f42222288bd8c1097396fb787855dba0aa6e75fa35e48b46a62c52ff191b2f" +) + + +def _sha256(data: bytes) -> str: + """Return the hex SHA-256 digest of ``data``.""" + return hashlib.sha256(data).hexdigest() + + +def _read_zlib(path: str) -> bytes: + """Read and decompress a zlib payload file. + + Args: + path: Absolute path to a zlib-compressed file. + + Returns: + The decompressed payload bytes. + """ + with open(path, "rb") as handle: + return zlib.decompress(handle.read()) + + +def load_node_ids(path: str = NODE_IDS_PAYLOAD) -> list[int]: + """Load the minimized node insertion-order IDs. + + Args: + path: Path to the little-endian uint32 zlib payload. + + Returns: + Exactly ``NUM_NODES`` unique IDs forming the set ``1..NUM_NODES``. + """ + raw = _read_zlib(path) + if _sha256(raw) != EXPECTED_NODE_IDS_SHA256: + raise ValueError(f"node id payload checksum mismatch for {path}") + if len(raw) != NUM_NODES * 4: + raise ValueError(f"unexpected node id payload size: {len(raw)}") + ids = list(struct.unpack(f"<{NUM_NODES}I", raw)) + if set(ids) != set(range(1, NUM_NODES + 1)): + raise ValueError("node id payload is not a permutation of 1..NUM_NODES") + return ids + + +def load_relationships( + path: str = RELS_PAYLOAD, +) -> tuple[list[int], list[int], list[str]]: + """Load the minimized relationship endpoints and type labels. + + Args: + path: Path to the packed ``(a_id, b_id, type_code)`` zlib payload. + + Returns: + Parallel lists of source IDs, destination IDs, and type strings. + """ + raw = _read_zlib(path) + if _sha256(raw) != EXPECTED_RELS_SHA256: + raise ValueError(f"relationship payload checksum mismatch for {path}") + record_size = 9 + if len(raw) != NUM_RELS * record_size: + raise ValueError(f"unexpected relationship payload size: {len(raw)}") + + src_ids: list[int] = [] + dst_ids: list[int] = [] + types: list[str] = [] + for offset in range(0, len(raw), record_size): + src, dst, type_code = struct.unpack_from(" None: + """Write a node parquet table with ``id`` / ``key`` columns. + + Args: + output_path: Destination parquet path. + column_prefix: Column name prefix (``a`` or ``b``). + ids: Node IDs in insertion order. + """ + keys = [f"node_{node_id}" for node_id in ids] + table = pa.table( + { + f"{column_prefix}.id": pa.array(ids, type=pa.int64()), + f"{column_prefix}.key": pa.array(keys, type=pa.utf8()), + } + ) + pq.write_table(table, output_path, compression="snappy", use_dictionary=False) + size_kb = os.path.getsize(output_path) // 1024 + print(f"Generated {output_path} ({len(ids)} rows, {size_kb} KB)") + + +def write_relationship_table( + output_path: str, + src_ids: list[int], + dst_ids: list[int], + types: list[str], +) -> None: + """Write the relationship parquet table with dictionary-encoded types. + + Args: + output_path: Destination parquet path. + src_ids: Source node IDs. + dst_ids: Destination node IDs. + types: Relationship ``type`` property values. + """ + table = pa.table( + { + "a.id": pa.array(src_ids, type=pa.int64()), + "b.id": pa.array(dst_ids, type=pa.int64()), + "r.type": pa.array(types, type=pa.utf8()), + } + ) + pq.write_table( + table, + output_path, + compression="snappy", + use_dictionary=["r.type"], + ) + size_kb = os.path.getsize(output_path) // 1024 + print(f"Generated {output_path} ({len(src_ids)} rows, {size_kb} KB)") + + +def generate_fixtures(fixture_dir: str) -> None: + """Generate all three parquet fixtures into ``fixture_dir``. + + Args: + fixture_dir: Directory that receives ``A.parquet``, ``B.parquet``, and + ``A_TO_B.parquet``. + """ + os.makedirs(fixture_dir, exist_ok=True) + node_ids = load_node_ids() + src_ids, dst_ids, types = load_relationships() + + write_node_table(os.path.join(fixture_dir, "A.parquet"), "a", node_ids) + write_node_table(os.path.join(fixture_dir, "B.parquet"), "b", node_ids) + write_relationship_table( + os.path.join(fixture_dir, "A_TO_B.parquet"), + src_ids, + dst_ids, + types, + ) + + +def parse_args(argv: list[str]) -> argparse.Namespace: + """Parse CLI arguments. + + Args: + argv: Argument vector excluding the program name. + + Returns: + Parsed argparse namespace. + """ + parser = argparse.ArgumentParser( + description="Generate dictionary-bug relationship projection parquet fixtures." + ) + parser.add_argument( + "--fixture-dir", + default=DEFAULT_FIXTURE_DIR, + help="Output directory for A.parquet, B.parquet, and A_TO_B.parquet.", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + """Entry point for fixture generation. + + Args: + argv: Optional argument vector excluding the program name. + + Returns: + Process exit code (``0`` on success). + """ + args = parse_args(sys.argv[1:] if argv is None else argv) + generate_fixtures(args.fixture_dir) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/include/storage/table/dictionary_column.h b/src/include/storage/table/dictionary_column.h index ad2f8110e..06a29c846 100644 --- a/src/include/storage/table/dictionary_column.h +++ b/src/include/storage/table/dictionary_column.h @@ -1,5 +1,8 @@ #pragma once +#include +#include + #include "dictionary_chunk.h" #include "storage/table/column.h" #include "storage/table/column_chunk_data.h" @@ -8,19 +11,14 @@ namespace lbug { namespace storage { +class StringColumn; + class DictionaryColumn { public: DictionaryColumn(const std::string& name, FileHandle* dataFH, MemoryManager* mm, ShadowFile* shadowFile, bool enableCompression); void scan(const SegmentState& state, DictionaryChunk& dictChunk) const; - // Offsets to scan should be a sorted list of pairs mapping the index of the entry in the string - // dictionary (as read from the index column) to the output index in the result vector to store - // the string. - template - void scan(const SegmentState& offsetState, const SegmentState& dataState, - std::vector>& offsetsToScan, - Result* result, const ColumnChunkMetadata& indexMeta) const; DictionaryChunk::string_index_t append(const DictionaryChunk& dictChunk, SegmentState& state, std::string_view val) const; @@ -32,12 +30,31 @@ class DictionaryColumn { Column* getOffsetColumn() const { return offsetColumn.get(); } private: + friend class StringColumn; + + // Offsets to scan should be a sorted list of pairs mapping the index of the entry in the string + // dictionary (as read from the index column) to the output index in the result vector to store + // the string. + void scan(const SegmentState& offsetState, const SegmentState& dataState, + std::vector>& offsetsToScan, + common::ValueVector* result, const ColumnChunkMetadata& indexMeta) const; + + // Materializes unique source dictionary indexes into a StringChunkData dictionary. + // The input indexes identify entries in the persistent source dictionary. The function may + // reorder them for locality, appends the corresponding string bytes and offsets to the result + // dictionary, and returns the actual old-index to new-index mapping. It does not write row + // indexes; callers that know result row positions must update the StringChunkData index chunk. + std::vector> + materializeToStringChunkDictionary(const SegmentState& offsetState, + const SegmentState& dataState, std::vector& indexesToScan, + StringChunkData& result, const ColumnChunkMetadata& indexMeta) const; + void scanOffsets(const SegmentState& state, DictionaryChunk::string_offset_t* offsets, uint64_t index, uint64_t numValues, uint64_t dataSize) const; - void scanValue(const SegmentState& dataState, uint64_t startOffset, uint64_t endOffset, - StringChunkData* result, uint64_t offsetInVector) const; void scanValue(const SegmentState& dataState, uint64_t startOffset, uint64_t endOffset, common::ValueVector* resultVector, uint64_t offsetInVector) const; + DictionaryChunk::string_index_t appendScannedValueToDictionary(const SegmentState& dataState, + uint64_t startOffset, uint64_t length, StringChunkData& result) const; static bool canDataCommitInPlace(const SegmentState& dataState, uint64_t totalStringLengthToAdd); diff --git a/src/storage/table/dictionary_column.cpp b/src/storage/table/dictionary_column.cpp index f1213e546..98efd3b32 100644 --- a/src/storage/table/dictionary_column.cpp +++ b/src/storage/table/dictionary_column.cpp @@ -13,7 +13,6 @@ #include "storage/table/string_chunk_data.h" #include "storage/table/string_column.h" #include -#include using namespace lbug::common; using namespace lbug::transaction; @@ -68,16 +67,15 @@ void DictionaryColumn::scan(const SegmentState& state, DictionaryChunk& dictChun } } -template void DictionaryColumn::scan(const SegmentState& offsetState, const SegmentState& dataState, - std::vector>& offsetsToScan, Result* result, + std::vector>& offsetsToScan, ValueVector* result, const ColumnChunkMetadata& indexMeta) const { string_index_t firstOffsetToScan = 0, lastOffsetToScan = 0; auto comp = [](auto pair1, auto pair2) { return pair1.first < pair2.first; }; auto duplicationFactor = (double)offsetState.metadata.numValues / indexMeta.numValues; if (duplicationFactor <= 0.5) { // If at least 50% of strings are duplicated, sort the offsets so we can re-use scanned - // strings + // strings. std::sort(offsetsToScan.begin(), offsetsToScan.end(), comp); firstOffsetToScan = offsetsToScan.front().first; lastOffsetToScan = offsetsToScan.back().first; @@ -100,13 +98,6 @@ void DictionaryColumn::scan(const SegmentState& offsetState, const SegmentState& scanOffsets(offsetState, offsets.data(), firstOffsetToScan, numOffsetsToScan, dataState.metadata.numValues); - if constexpr (std::same_as) { - auto& offsetChunk = *result->getDictionaryChunk()->getOffsetChunk(); - if (offsetChunk.getNumValues() + offsetsToScan.size() > offsetChunk.getCapacity()) { - offsetChunk.resize(std::bit_ceil(offsetChunk.getNumValues() + offsetsToScan.size())); - } - } - for (auto pos = 0u; pos < offsetsToScan.size(); pos++) { auto startOffset = offsets[offsetsToScan[pos].first - firstOffsetToScan]; auto endOffset = offsets[offsetsToScan[pos].first - firstOffsetToScan + 1]; @@ -114,36 +105,56 @@ void DictionaryColumn::scan(const SegmentState& offsetState, const SegmentState& DASSERT(endOffset >= startOffset); scanValue(dataState, startOffset, lengthToScan, result, offsetsToScan[pos].second); // For each string which has the same index in the dictionary as the one we scanned, - // copy the scanned string to its position in the result vector - if constexpr (std::same_as) { - auto& scannedString = result->template getValue(offsetsToScan[pos].second); - while (pos + 1 < offsetsToScan.size() && - offsetsToScan[pos + 1].first == offsetsToScan[pos].first) { - pos++; - result->template setValue(offsetsToScan[pos].second, scannedString); - } - } else { - // When scanning to chunks de-duplication should be done prior to this function such - // that you can have multiple positions in the string index chunk pointing to one string - // in this dictionary chunk. - // The offset chunk cannot have multiple offsets pointing to the same data, even if - // consecutive, since that would break the mechanism for calculating the size of a - // string. - DASSERT(pos == offsetsToScan.size() - 1 || - offsetsToScan[pos].first != offsetsToScan[pos + 1].first); + // copy the scanned string to its position in the result vector. + auto& scannedString = result->getValue(offsetsToScan[pos].second); + while (pos + 1 < offsetsToScan.size() && + offsetsToScan[pos + 1].first == offsetsToScan[pos].first) { + pos++; + result->setValue(offsetsToScan[pos].second, scannedString); } } } -template void DictionaryColumn::scan(const SegmentState& offsetState, - const SegmentState& dataState, - std::vector>& offsetsToScan, - common::ValueVector* result, const ColumnChunkMetadata& indexMeta) const; +std::vector> +DictionaryColumn::materializeToStringChunkDictionary(const SegmentState& offsetState, + const SegmentState& dataState, std::vector& indexesToScan, + StringChunkData& result, const ColumnChunkMetadata& indexMeta) const { + if (indexesToScan.empty()) { + return {}; + } -template void DictionaryColumn::scan(const SegmentState& offsetState, - const SegmentState& dataState, - std::vector>& offsetsToScan, - StringChunkData* result, const ColumnChunkMetadata& indexMeta) const; + string_index_t firstOffsetToScan = 0, lastOffsetToScan = 0; + auto comp = [](auto index1, auto index2) { return index1 < index2; }; + auto duplicationFactor = (double)offsetState.metadata.numValues / indexMeta.numValues; + if (duplicationFactor <= 0.5) { + std::sort(indexesToScan.begin(), indexesToScan.end(), comp); + firstOffsetToScan = indexesToScan.front(); + lastOffsetToScan = indexesToScan.back(); + } else { + const auto& [min, max] = + std::minmax_element(indexesToScan.begin(), indexesToScan.end(), comp); + firstOffsetToScan = *min; + lastOffsetToScan = *max; + } + + auto numOffsetsToScan = lastOffsetToScan - firstOffsetToScan + 1; + std::vector offsets(numOffsetsToScan + 1); + scanOffsets(offsetState, offsets.data(), firstOffsetToScan, numOffsetsToScan, + dataState.metadata.numValues); + + std::vector> mapping; + mapping.reserve(indexesToScan.size()); + for (const auto indexToScan : indexesToScan) { + auto startOffset = offsets[indexToScan - firstOffsetToScan]; + auto endOffset = offsets[indexToScan - firstOffsetToScan + 1]; + auto lengthToScan = endOffset - startOffset; + DASSERT(endOffset >= startOffset); + auto newIndex = + appendScannedValueToDictionary(dataState, startOffset, lengthToScan, result); + mapping.emplace_back(indexToScan, newIndex); + } + return mapping; +} string_index_t DictionaryColumn::append(const DictionaryChunk& dictChunk, SegmentState& state, std::string_view val) const { @@ -179,26 +190,24 @@ void DictionaryColumn::scanValue(const SegmentState& dataState, uint64_t startOf } } -void DictionaryColumn::scanValue(const SegmentState& dataState, uint64_t startOffset, - uint64_t length, StringChunkData* result, uint64_t offsetInResult) const { - auto& stringDataChunk = *result->getDictionaryChunk().getStringDataChunk(); - auto& offsetChunk = *result->getDictionaryChunk().getOffsetChunk(); - auto& indexChunk = *result->getIndexColumnChunk(); +string_index_t DictionaryColumn::appendScannedValueToDictionary(const SegmentState& dataState, + uint64_t startOffset, uint64_t length, StringChunkData& result) const { + auto& stringDataChunk = *result.getDictionaryChunk().getStringDataChunk(); + auto& offsetChunk = *result.getDictionaryChunk().getOffsetChunk(); if (stringDataChunk.getCapacity() < stringDataChunk.getNumValues() + length) { stringDataChunk.resize(std::bit_ceil(stringDataChunk.getNumValues() + length)); } if (offsetChunk.getNumValues() == offsetChunk.getCapacity()) { offsetChunk.resize(std::bit_ceil(offsetChunk.getNumValues() + 1)); } - if (offsetInResult >= indexChunk.getCapacity()) { - indexChunk.resize(std::bit_ceil(offsetInResult + 1)); + const auto newIndex = static_cast(offsetChunk.getNumValues()); + if (length > 0) { + dataColumn->scanSegment(dataState, startOffset, length, + stringDataChunk.getData() + stringDataChunk.getNumValues()); } - dataColumn->scanSegment(dataState, startOffset, length, - stringDataChunk.getData() + stringDataChunk.getNumValues()); - indexChunk.setValue(offsetChunk.getNumValues(), offsetInResult); - offsetChunk.setValue(stringDataChunk.getNumValues(), - offsetChunk.getNumValues()); + offsetChunk.setValue(stringDataChunk.getNumValues(), newIndex); stringDataChunk.setNumValues(stringDataChunk.getNumValues() + length); + return newIndex; } bool DictionaryColumn::canCommitInPlace(const SegmentState& state, uint64_t numNewStrings, diff --git a/src/storage/table/string_column.cpp b/src/storage/table/string_column.cpp index d561b8628..43b5a97d9 100644 --- a/src/storage/table/string_column.cpp +++ b/src/storage/table/string_column.cpp @@ -166,32 +166,46 @@ void StringColumn::scanSegment(const SegmentState& state, ColumnChunkData* resul } dictionary.scan(state, stringResultChunk->getDictionaryChunk()); } else { - // Any strings which are duplicated only need to be scanned once, so we track duplicate - // indices - std::unordered_map indexMap; - std::vector> offsetsToScan; + // Partial chunk scans first collect old dictionary indexes per result row. The dictionary + // materialization step may scan those old indexes in a different order for locality, so row + // index values are written only after the actual old-to-new dictionary mapping is known. + std::unordered_map oldToNewIndex; + std::vector oldIndexesByRow; + std::vector indexesToScan; + oldIndexesByRow.resize(numValuesToScan); for (auto i = 0u; i < numValuesToScan; i++) { - if (!resultChunk->isNull(startOffsetInResult + i)) { - auto index = indexChunk->getValue(startOffsetInResult + i); - auto element = indexMap.find(index); - if (element == indexMap.end()) { - indexMap.insert(std::make_pair(index, initialDictSize + offsetsToScan.size())); - indexChunk->setValue(initialDictSize + offsetsToScan.size(), - startOffsetInResult + i); - offsetsToScan.emplace_back(index, initialDictSize + offsetsToScan.size()); - } else { - indexChunk->setValue(element->second, startOffsetInResult + i); - } + auto resultPos = startOffsetInResult + i; + if (resultChunk->isNull(resultPos)) { + continue; + } + auto index = indexChunk->getValue(resultPos); + oldIndexesByRow[i] = index; + if (oldToNewIndex.find(index) == oldToNewIndex.end()) { + oldToNewIndex.insert(std::make_pair(index, 0)); + indexesToScan.push_back(index); } } - if (offsetsToScan.size() == 0) { + if (indexesToScan.empty()) { // All scanned values are null return; } - dictionary.scan(getChildState(state, ChildStateIndex::OFFSET), - getChildState(state, ChildStateIndex::DATA), offsetsToScan, stringResultChunk, + auto materializedIndexes = dictionary.materializeToStringChunkDictionary( + getChildState(state, ChildStateIndex::OFFSET), + getChildState(state, ChildStateIndex::DATA), indexesToScan, *stringResultChunk, getChildState(state, ChildStateIndex::INDEX).metadata); + for (const auto& [oldIndex, newIndex] : materializedIndexes) { + oldToNewIndex[oldIndex] = newIndex; + } + for (auto i = 0u; i < numValuesToScan; i++) { + auto resultPos = startOffsetInResult + i; + if (resultChunk->isNull(resultPos)) { + continue; + } + auto element = oldToNewIndex.find(oldIndexesByRow[i]); + DASSERT(element != oldToNewIndex.end()); + indexChunk->setValue(element->second, resultPos); + } } DASSERT(resultChunk->getNumValues() == startOffsetInResult + numValuesToScan && stringResultChunk->getIndexColumnChunk()->getNumValues() == diff --git a/test/storage/CMakeLists.txt b/test/storage/CMakeLists.txt index 83efdeabb..b1b9d9a67 100644 --- a/test/storage/CMakeLists.txt +++ b/test/storage/CMakeLists.txt @@ -7,6 +7,7 @@ add_lbug_test(rel_tests rel_scan_test.cpp rel_delete_test.cpp) add_lbug_test(node_update_test node_update_test.cpp) add_lbug_test(detach_delete_test detach_delete_test.cpp) add_lbug_test(storage_utils_test storage_utils_test.cpp) +add_lbug_test(string_chunk_scan_test string_chunk_scan_test.cpp) add_lbug_test(ice_disk_utils_test ice_disk_utils_test.cpp) target_include_directories(compression_test PRIVATE ${PROJECT_SOURCE_DIR}/third_party/alp/include) diff --git a/test/storage/string_chunk_scan_test.cpp b/test/storage/string_chunk_scan_test.cpp new file mode 100644 index 000000000..285316e80 --- /dev/null +++ b/test/storage/string_chunk_scan_test.cpp @@ -0,0 +1,378 @@ +#include +#include +#include +#include +#include +#include + +#include "common/vector/value_vector.h" +#include "graph_test/private_graph_test.h" +#include "gtest/gtest.h" +#include "storage/page_manager.h" +#include "storage/storage_manager.h" +#include "storage/table/column_chunk.h" +#include "storage/table/column_chunk_data.h" +#include "storage/table/string_chunk_data.h" +#include "storage/table/string_column.h" +#include + +using namespace lbug::common; +using namespace lbug::storage; + +namespace lbug { +namespace testing { +namespace { + +using MaybeString = std::optional; +using string_index_t = DictionaryChunk::string_index_t; + +constexpr std::string_view NULL_SENTINEL = ""; + +class StringChunkScanTest : public DBTest { +public: + void SetUp() override { + BaseGraphTest::SetUp(); // NOLINT + createDBAndConn(); + } + + std::string getInputDir() override { return TestHelper::appendLbugRootPath("dataset/empty/"); } +}; + +MaybeString value(std::string_view str) { + return MaybeString{str}; +} + +MaybeString nullValue() { + return std::nullopt; +} + +std::string printableValue(const MaybeString& maybeValue) { + if (!maybeValue.has_value()) { + return std::string{NULL_SENTINEL}; + } + return std::string{*maybeValue}; +} + +std::vector printableValues(std::span values) { + std::vector result; + result.reserve(values.size()); + for (const auto& maybeValue : values) { + result.push_back(printableValue(maybeValue)); + } + return result; +} + +std::vector expectedRange(std::span values, offset_t startRow, + offset_t numRows) { + std::vector result; + result.reserve(numRows); + for (auto i = 0u; i < numRows; i++) { + result.push_back(printableValue(values[startRow + i])); + } + return result; +} + +std::vector expectedPrefixedRange(std::span prefix, + std::span values, offset_t startRow, offset_t numRows) { + auto result = printableValues(prefix); + auto range = expectedRange(values, startRow, numRows); + result.insert(result.end(), range.begin(), range.end()); + return result; +} + +LogicalType makeStringLikeType(const StringColumn& column) { + if (column.getDataType().getLogicalTypeID() == LogicalTypeID::JSON) { + return LogicalType::JSON(); + } + return LogicalType::STRING(); +} + +void writeMaybeStringValue(StringChunkData& chunk, ValueVector& vector, offset_t row, + const MaybeString& maybeValue) { + vector.setNull(0, !maybeValue.has_value()); + if (maybeValue.has_value()) { + StringVector::addString(&vector, 0, *maybeValue); + } + chunk.write(&vector, 0, row); +} + +struct PersistedStringChunk { + std::unique_ptr chunk; + ChunkState state; +}; + +PersistedStringChunk buildPersistedStringChunk(MemoryManager& memoryManager, + PageManager& pageManager, StringColumn& column, std::span values) { + auto chunk = ColumnChunkFactory::createColumnChunkData(memoryManager, + makeStringLikeType(column), true /*enableCompression*/, + std::max(values.size(), 1), ResidencyState::IN_MEMORY); + auto& stringChunk = chunk->cast(); + ValueVector valueVector{makeStringLikeType(column), &memoryManager}; + for (auto row = 0u; row < values.size(); row++) { + writeMaybeStringValue(stringChunk, valueVector, row, values[row]); + } + chunk->flush(pageManager); + + ChunkState state; + state.column = &column; + state.segmentStates.resize(1); + chunk->initializeScanState(state.segmentStates[0], &column); + return {std::move(chunk), std::move(state)}; +} + +struct MaterializedStringChunk { + std::vector values; + std::vector> indexes; + uint64_t dictionarySize; +}; + +MaterializedStringChunk materializeStringChunk(const StringChunkData& chunk, offset_t numRows) { + MaterializedStringChunk result; + result.values.reserve(numRows); + result.indexes.reserve(numRows); + result.dictionarySize = chunk.getDictionaryChunk().getOffsetChunk()->getNumValues(); + const auto* indexChunk = chunk.getIndexColumnChunk(); + for (auto row = 0u; row < numRows; row++) { + if (chunk.isNull(row)) { + result.values.push_back(std::string{NULL_SENTINEL}); + result.indexes.push_back(std::nullopt); + continue; + } + result.values.push_back(chunk.getValue(row)); + result.indexes.push_back(indexChunk->getValue(row)); + } + return result; +} + +void expectIndexesInBounds(const MaterializedStringChunk& chunk) { + for (const auto& maybeIndex : chunk.indexes) { + if (maybeIndex.has_value()) { + EXPECT_LT(*maybeIndex, chunk.dictionarySize); + } + } +} + +std::vector materializeValueVector(const ValueVector& vector, offset_t numRows) { + std::vector values; + values.reserve(numRows); + for (auto row = 0u; row < numRows; row++) { + if (vector.isNull(row)) { + values.push_back(std::string{NULL_SENTINEL}); + continue; + } + values.push_back(vector.getValue(row).getAsString()); + } + return values; +} + +MaterializedStringChunk scanToStringChunk(MemoryManager& memoryManager, StringColumn& column, + const ChunkState& state, offset_t startRow, offset_t numRows, + std::span prefix = {}) { + auto outputCapacity = std::max(prefix.size() + numRows, 1); + auto outputChunk = + ColumnChunkFactory::createColumnChunkData(memoryManager, makeStringLikeType(column), + true /*enableCompression*/, outputCapacity, ResidencyState::IN_MEMORY); + auto& outputStringChunk = outputChunk->cast(); + ValueVector valueVector{makeStringLikeType(column), &memoryManager}; + for (auto row = 0u; row < prefix.size(); row++) { + writeMaybeStringValue(outputStringChunk, valueVector, row, prefix[row]); + } + + if (prefix.empty()) { + column.scan(state, outputChunk.get(), startRow, numRows); + } else { + static_cast(column).scanSegment(state.segmentStates[0], outputChunk.get(), + startRow, numRows); + } + + return materializeStringChunk(outputStringChunk, prefix.size() + numRows); +} + +std::vector scanToValueVector(MemoryManager& memoryManager, StringColumn& column, + const ChunkState& state, offset_t startRow, offset_t numRows) { + ValueVector outputVector{makeStringLikeType(column), &memoryManager}; + column.scan(state, startRow, numRows, &outputVector, 0 /*offsetInVector*/); + return materializeValueVector(outputVector, numRows); +} + +struct ScanCase { + std::string name; + std::vector values; + offset_t startRow; + offset_t numRows; + std::vector prefix; +}; + +} // namespace + +TEST_F(StringChunkScanTest, PartialAndFullStringChunkScansMatchValueVectorScans) { + auto* memoryManager = getMemoryManager(*database); + auto* storageManager = getStorageManager(*database); + PageManager pageManager{storageManager->getDataFH()}; + StringColumn column{"scan_value", LogicalType::STRING(), storageManager->getDataFH(), + memoryManager, &storageManager->getShadowFile(), true /*enableCompression*/}; + + const std::vector cases = { + { + "duplicate first occurrence before partial range", + {value("REL_A"), value("REL_B"), value("REL_B"), value("REL_A")}, + 1, + 3, + {}, + }, + { + "dictionary order differs from row encounter order", + {value("ALPHA"), value("BETA"), value("ALPHA"), value("GAMMA"), value("BETA")}, + 1, + 4, + {}, + }, + { + "unique values partial range", + {value("ALPHA"), value("BETA"), value("GAMMA"), value("DELTA")}, + 1, + 3, + {}, + }, + { + "all values duplicate", + {value("ALPHA"), value("ALPHA"), value("ALPHA"), value("ALPHA")}, + 1, + 3, + {}, + }, + { + "duplicate values separated by null", + {value("ALPHA"), nullValue(), value("ALPHA"), value("BETA"), value("ALPHA")}, + 0, + 5, + {}, + }, + { + "all-null partial range", + {value("ROOT"), nullValue(), nullValue(), nullValue(), value("TAIL")}, + 1, + 3, + {}, + }, + { + "empty strings and duplicates", + {value(""), value("ALPHA"), value(""), nullValue(), value("BETA"), value("")}, + 0, + 6, + {}, + }, + { + "full segment uses entire dictionary fast path", + {value("REL_A"), value("REL_B"), value("REL_B"), value("REL_A")}, + 0, + 4, + {}, + }, + { + "partial scan appends after existing dictionary rows", + {value("REL_A"), value("REL_B"), value("REL_B"), value("REL_A")}, + 1, + 3, + {value("PREFIX_ONE"), value("PREFIX_TWO")}, + }, + { + "high duplicate pressure partial range", + {value("ROOT"), value("HOT"), value("HOT"), value("HOT"), value("COLD"), value("HOT"), + value("COLD"), value("HOT")}, + 1, + 7, + {}, + }, + { + "low duplicate pressure partial range", + {value("ROOT"), value("A"), value("B"), value("C"), value("D"), value("E"), value("F"), + value("A")}, + 1, + 7, + {}, + }, + }; + + for (const auto& scanCase : cases) { + SCOPED_TRACE(scanCase.name); + auto persisted = + buildPersistedStringChunk(*memoryManager, pageManager, column, scanCase.values); + + const auto valueVectorValues = scanToValueVector(*memoryManager, column, persisted.state, + scanCase.startRow, scanCase.numRows); + const auto stringChunkValues = scanToStringChunk(*memoryManager, column, persisted.state, + scanCase.startRow, scanCase.numRows, scanCase.prefix); + + EXPECT_EQ(valueVectorValues, + expectedRange(scanCase.values, scanCase.startRow, scanCase.numRows)); + EXPECT_EQ(stringChunkValues.values, expectedPrefixedRange(scanCase.prefix, scanCase.values, + scanCase.startRow, scanCase.numRows)); + expectIndexesInBounds(stringChunkValues); + } +} + +TEST_F(StringChunkScanTest, PartialStringChunkScansPreserveIndexInvariantsWhenAppending) { + auto* memoryManager = getMemoryManager(*database); + auto* storageManager = getStorageManager(*database); + PageManager pageManager{storageManager->getDataFH()}; + StringColumn column{"scan_value", LogicalType::STRING(), storageManager->getDataFH(), + memoryManager, &storageManager->getShadowFile(), true /*enableCompression*/}; + + const std::vector values = {value("REL_A"), value("REL_B"), value("REL_B"), + value("REL_A")}; + const std::vector prefix = {value("PREFIX_ONE"), value("PREFIX_TWO")}; + auto persisted = buildPersistedStringChunk(*memoryManager, pageManager, column, values); + + const auto scanned = scanToStringChunk(*memoryManager, column, persisted.state, 1, 3, prefix); + + EXPECT_EQ(scanned.values, expectedPrefixedRange(prefix, values, 1, 3)); + expectIndexesInBounds(scanned); + ASSERT_EQ(scanned.indexes.size(), 5u); + ASSERT_TRUE(scanned.indexes[2].has_value()); + ASSERT_TRUE(scanned.indexes[3].has_value()); + ASSERT_TRUE(scanned.indexes[4].has_value()); + EXPECT_EQ(scanned.indexes[2], scanned.indexes[3]); + EXPECT_NE(scanned.indexes[2], scanned.indexes[4]); + EXPECT_GE(*scanned.indexes[2], static_cast(prefix.size())); + EXPECT_GE(*scanned.indexes[4], static_cast(prefix.size())); +} + +TEST_F(StringChunkScanTest, JsonStringChunkScansMatchValueVectorScans) { + auto* memoryManager = getMemoryManager(*database); + auto* storageManager = getStorageManager(*database); + PageManager pageManager{storageManager->getDataFH()}; + StringColumn column{"scan_json", LogicalType::JSON(), storageManager->getDataFH(), + memoryManager, &storageManager->getShadowFile(), true /*enableCompression*/}; + + const std::vector values = {value("{\"kind\":\"root\"}"), + value("{\"kind\":\"edge\"}"), value("{\"kind\":\"edge\"}"), nullValue(), + value("{\"kind\":\"leaf\"}")}; + auto persisted = buildPersistedStringChunk(*memoryManager, pageManager, column, values); + + const auto valueVectorValues = scanToValueVector(*memoryManager, column, persisted.state, 1, 4); + const auto stringChunkValues = scanToStringChunk(*memoryManager, column, persisted.state, 1, 4); + + EXPECT_EQ(valueVectorValues, expectedRange(values, 1, 4)); + EXPECT_EQ(stringChunkValues.values, expectedRange(values, 1, 4)); + expectIndexesInBounds(stringChunkValues); +} + +TEST_F(StringChunkScanTest, ValueVectorPartialScansRemainCorrectWhenDictionaryOffsetsCanBeSorted) { + auto* memoryManager = getMemoryManager(*database); + auto* storageManager = getStorageManager(*database); + PageManager pageManager{storageManager->getDataFH()}; + StringColumn column{"scan_value", LogicalType::STRING(), storageManager->getDataFH(), + memoryManager, &storageManager->getShadowFile(), true /*enableCompression*/}; + + const std::vector values = {value("REL_A"), value("REL_B"), value("REL_B"), + value("REL_A")}; + auto persisted = buildPersistedStringChunk(*memoryManager, pageManager, column, values); + + const auto scanned = scanToValueVector(*memoryManager, column, persisted.state, 1, 3); + + EXPECT_EQ(scanned, expectedRange(values, 1, 3)); +} + +} // namespace testing +} // namespace lbug diff --git a/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated.test b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated.test new file mode 100644 index 000000000..749515a82 --- /dev/null +++ b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated.test @@ -0,0 +1,80 @@ +-DATASET PARQUET empty +-SKIP_IN_MEM + +-- + +-CASE AnonymousParquetDeleteReload +-STATEMENT CREATE NODE TABLE `A` (`id` INT64, `key` STRING, PRIMARY KEY(`id`)); +---- ok + +-STATEMENT CREATE NODE TABLE `B` (`id` INT64, `key` STRING, PRIMARY KEY(`id`)); +---- ok + +-STATEMENT CREATE REL TABLE `A_TO_B` (FROM `A` TO `B`, `type` STRING, MANY_MANY); +---- ok + +-STATEMENT COPY `A` (`id`, `key`) FROM "${LBUG_ROOT_DIRECTORY}/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/A.parquet"; +---- ok + +-STATEMENT COPY `B` (`id`, `key`) FROM "${LBUG_ROOT_DIRECTORY}/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/B.parquet"; +---- ok + +-STATEMENT COPY `A_TO_B` (`type`) FROM "${LBUG_ROOT_DIRECTORY}/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/A_TO_B.parquet"; +---- ok + +-STATEMENT MATCH ()-[r:A_TO_B]->() RETURN count(r); +---- 1 +953 + +-STATEMENT MATCH (a:A)-[r:A_TO_B]->(b:B) + WHERE offset(ID(r)) = 952 + RETURN a.key, r.type, b.key; +---- 1 +node_10049|REL_A|node_10050 + +-STATEMENT MATCH (a:A {key: 'node_3379'})-[r:A_TO_B]->(b:B {key: 'node_3382'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A + +-STATEMENT MATCH (b:B {key: 'node_3382'})<-[r:A_TO_B]-(a:A {key: 'node_3379'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A + +-STATEMENT MATCH ()-[r:A_TO_B]->() + WHERE offset(ID(r)) = 952 + DELETE r; +---- ok + +-STATEMENT MATCH ()-[r:A_TO_B]->() + WHERE offset(ID(r)) = 952 + RETURN count(r); +---- 1 +0 + +-STATEMENT MATCH (a:A {key: 'node_3379'})-[r:A_TO_B]->(b:B {key: 'node_3382'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A + +-STATEMENT MATCH (b:B {key: 'node_3382'})<-[r:A_TO_B]-(a:A {key: 'node_3379'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A + +-RELOADDB + +-STATEMENT MATCH ()-[r:A_TO_B]->() RETURN count(r); +---- 1 +952 + +-STATEMENT MATCH (a:A {key: 'node_3379'})-[r:A_TO_B]->(b:B {key: 'node_3382'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A + +-STATEMENT MATCH (b:B {key: 'node_3382'})<-[r:A_TO_B]-(a:A {key: 'node_3379'}) + RETURN offset(ID(r)), r.type; +---- 1 +0|REL_A diff --git a/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/node_ids.u32le.zlib b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/node_ids.u32le.zlib new file mode 100644 index 000000000..f6ead4f2f Binary files /dev/null and b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/node_ids.u32le.zlib differ diff --git a/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/rels.u32u32u8.zlib b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/rels.u32u32u8.zlib new file mode 100644 index 000000000..8dbba4355 Binary files /dev/null and b/test/test_files/dictionary_bug/orb383_relationship_projection_obfuscated_fixture/rels.u32u32u8.zlib differ