实验 04：位图索引与 Bloom 过滤器 —— 把 WHERE 子句变成位运算

import sys, pathlib

# Put the repo root (the directory containing `system/`) on sys.path so
# `from system...` works whether this notebook is opened from its own
# folder or from the repo root.
for _p in (pathlib.Path.cwd(), *pathlib.Path.cwd().parents):
    if (_p / "system").is_dir():
        if str(_p) not in sys.path:
            sys.path.insert(0, str(_p))
        break

from system.data_classes import Address
import statistics
import time
import itertools as it
from system.bit_sequences import IntegerBitSequence, BitListBitSequence, WAHBitSequence
from system.interfaces.bit_sequence import (
    BitSequence,
    UncompressedBitSequence,
    CompressedBitSequence,
)
from system.indexes.bloom_filters import BloomFilter
from system.indexes.bitmap_indexes import (
    SortedEqualityEncodedBitmapIndex,
    RangeEncodedBitmapIndex,
)
import pickle
from typing import Type
import matplotlib.pyplot as plt
import numpy as np

# Four possible cities
possible_cities: list[str] = ["New York", "Cuppertino", "Berlin", "Saarbrücken"]

# Five possible streets
possible_streets: list[str] = [
    "Unistreet",
    "Macstreet",
    "Longstreet",
    "Msstreet",
    "BigDataStreet",
]

# Possible house numbers. This is the size knob for the whole notebook: the
# number of addresses is len(cities) * len(streets) * len(house_numbers), and it
# drives every index build -- in particular the range-encoded index, which
# creates one bit-sequence per distinct house number. We cap it at 200 (-> 4000
# addresses) to keep CI fast (~4s instead of ~19s); the relative comparisons and
# plots stay representative. For the lecture, uncomment the range(1, 1001) line.
possible_house_numbers: list[int] = list(range(1, 201))
# possible_house_numbers: list[int] = [i for i in range(1, 1001)]

# Create all combinations (len(cities) * len(streets) * len(house_numbers)) of addresses
all_addresses: list[Address] = []
for idx, comb in enumerate(
    it.product(possible_cities, possible_streets, possible_house_numbers)
):
    all_addresses.append(Address(idx, comb[0], comb[1], comb[2]))


# Create bulkloads to be used for different indexes
def city_bulkload_generator():
    yield from ((address.city, address.id) for address in all_addresses)


def street_bulkload_generator():
    yield from ((address.street, address.id) for address in all_addresses)


def house_number_bulkload_generator():
    yield from ((address.house_number, address.id) for address in all_addresses)

# Store the results of the query above
query_1_naive_result: set[int] = set()

start = time.time()

for address in all_addresses:
    # Make String comparisons
    if address.city == "New York" and (
        address.street in ("Unistreet", "BigDataStreet")
    ):
        query_1_naive_result.add(address.id)

end = time.time()
print(f"Time taken for naive query: {end - start}")
print(f"The number of corresponding ids is: {len(query_1_naive_result)}")

Time taken for naive query: 0.0008704662322998047
The number of corresponding ids is: 2000

def create_indexes_and_execute_query(
    bit_sequence_type: Type[UncompressedBitSequence],
    compression_type: Type[CompressedBitSequence] | None = None,
    print_info=True,
):
    """
    Execute the query from above using different bitmap index implementations.
    :param bit_sequence_type: The type of bit-sequence to use.
    :param compression_type: The type of compression to use. None if no compression should be used.
    :param print_info: Print information about different metrics.
    :return: (The time to create an index, the time to execute the query, the number of bits stored in the index)
    """

    start_index = time.time()

    # Create Bitmap index for city
    city_bitmap_index: SortedEqualityEncodedBitmapIndex = (
        SortedEqualityEncodedBitmapIndex(
            bit_sequence_type=bit_sequence_type, compression_type=compression_type
        )
    )
    city_bitmap_index.bulkload(city_bulkload_generator())

    # Create Bitmap index for street
    street_bitmap_index: SortedEqualityEncodedBitmapIndex = (
        SortedEqualityEncodedBitmapIndex(
            bit_sequence_type=bit_sequence_type, compression_type=compression_type
        )
    )
    street_bitmap_index.bulkload(street_bulkload_generator())
    end_index = time.time()

    start_exec = time.time()

    # Get a bit_sequence containing all addresses in New York
    street_equal_new_york: BitSequence = city_bitmap_index.get_equal("New York")

    # Get a bit_sequence containing all addresses in Unistreet
    street_equal_unistreet: BitSequence = street_bitmap_index.get_equal("Unistreet")

    # Get a bit_sequence containing all addresses in BigDataStreet
    street_equal_bigdatastreet: BitSequence = street_bitmap_index.get_equal(
        "BigDataStreet"
    )

    # Combine results via bit operations
    result: BitSequence = street_equal_new_york & (
        street_equal_unistreet | street_equal_bigdatastreet
    )
    end_exec = time.time()
    if print_info:
        print(f"\tMeasure performance for index of type {bit_sequence_type}")
        print(f"\t\tTime taken for creating an index: {end_index - start_index}")
        print(
            f"\t\tStored bits for the index: {city_bitmap_index.get_number_of_bits() + street_bitmap_index.get_number_of_bits()}"
        )
        print(f"\t\tTime taken for indexed query: {end_exec - start_exec}")
        print(f"\t\tThe number of corresponding ids is: {len(result.as_set())}")
    return (
        end_index - start_index,
        end_exec - start_exec,
        city_bitmap_index.get_number_of_bits()
        + street_bitmap_index.get_number_of_bits(),
        result.as_set(),
    )

_, _, query_1_boolean_stored_bits, query_1_boolean_result_bits = (
    create_indexes_and_execute_query(bit_sequence_type=BitListBitSequence)
)

	Measure performance for index of type <class 'system.bit_sequences.BitListBitSequence'>
		Time taken for creating an index: 0.039728641510009766
		Stored bits for the index: 180000
		Time taken for indexed query: 0.0064814090728759766
		The number of corresponding ids is: 2000

_, _, query_1_integer_stored_bits, query_1_integer_result_bits = (
    create_indexes_and_execute_query(bit_sequence_type=IntegerBitSequence)
)

	Measure performance for index of type <class 'system.bit_sequences.IntegerBitSequence'>
		Time taken for creating an index: 0.03580427169799805
		Stored bits for the index: 180000
		Time taken for indexed query: 2.3365020751953125e-05
		The number of corresponding ids is: 2000

start_index = time.time()

# Create BitSequence for city
house_number_bitmap_index: SortedEqualityEncodedBitmapIndex = (
    SortedEqualityEncodedBitmapIndex(bit_sequence_type=IntegerBitSequence)
)
house_number_bitmap_index.bulkload(house_number_bulkload_generator())

end_index = time.time()
print(f"Time taken for creating an index: {end_index - start_index}")

start_exec = time.time()
smaller_than_500_equality: BitSequence = house_number_bitmap_index.get_smaller(500)

end_exec = time.time()
print(f"Time taken for indexed query: {end_exec - start_exec}")
print(f"The number of corresponding ids is: {len(smaller_than_500_equality.as_set())}")

Time taken for creating an index: 0.026168107986450195
Time taken for indexed query: 0.0007965564727783203
The number of corresponding ids is: 9980

start_index = time.time()

# Create BitSequence for house number
house_number_bitmap_index: RangeEncodedBitmapIndex = RangeEncodedBitmapIndex(
    bit_sequence_type=IntegerBitSequence
)
house_number_bitmap_index.bulkload(house_number_bulkload_generator())
end_index = time.time()
print(f"Time taken for creating an index: {end_index - start_index}")

start_exec = time.time()

# Only one access is required to compute the result.
smaller_than_500_range: BitSequence = house_number_bitmap_index.get_smaller(500)

end_exec = time.time()
print(f"Time taken for indexed query: {end_exec - start_exec}")
print(f"The number of corresponding ids is: {len(smaller_than_500_range.as_set())}")

Time taken for creating an index: 4.523980379104614
Time taken for indexed query: 5.817413330078125e-05
The number of corresponding ids is: 9980

start_exec = time.time()

# Compute individual results
house_number_greater_equal_300: BitSequence = (
    house_number_bitmap_index.get_greater_or_equal(300)
)
house_number_equal_30: BitSequence = house_number_bitmap_index.get_equal(30)

# Combine results
ge_300_e_30_result: BitSequence = house_number_greater_equal_300 | house_number_equal_30

end_exec = time.time()
print(f"Time taken for indexed query: {end_exec - start_exec}")
print(f"The number of corresponding ids is: {len(ge_300_e_30_result.as_set())}")

Time taken for indexed query: 9.012222290039062e-05
The number of corresponding ids is: 14040

start_index = time.time()

# Create equality-encoded indexes for house numbers and cities
house_number_equality_bitmap_index: SortedEqualityEncodedBitmapIndex = (
    SortedEqualityEncodedBitmapIndex(
        bit_sequence_type=IntegerBitSequence,
    )
)
house_number_equality_bitmap_index.bulkload(house_number_bulkload_generator())
city_equality_bitmap_index: SortedEqualityEncodedBitmapIndex = (
    SortedEqualityEncodedBitmapIndex(
        bit_sequence_type=IntegerBitSequence,
    )
)
city_equality_bitmap_index.bulkload(city_bulkload_generator())

# Create range-encoded index for house numbers
house_number_range_bitmap_index: RangeEncodedBitmapIndex = RangeEncodedBitmapIndex(
    bit_sequence_type=IntegerBitSequence,
)
house_number_range_bitmap_index.bulkload(house_number_bulkload_generator())
end_index = time.time()
print(f"Time taken for creating an index: {end_index - start_index}")

start_exec = time.time()

# Compute individual result sets
house_number_equal_23 = house_number_equality_bitmap_index.get_equal(23)
city_equal_new_york: BitSequence = city_equality_bitmap_index.get_equal("New York")
house_number_smaller_40: BitSequence = house_number_range_bitmap_index.get_smaller(40)
end_exec = time.time()

combined_result: BitSequence = house_number_equal_23 | (
    city_equal_new_york & house_number_smaller_40
)
print(f"Time taken for indexed query: {end_exec - start_exec}")
print(f"The number of corresponding ids is: {len(combined_result.as_set())}")

Time taken for creating an index: 4.189512729644775
Time taken for indexed query: 9.34600830078125e-05
The number of corresponding ids is: 210

# Write data
house_numbers_equal_1: list[Address] = []

# Get all addresses from New York
for address in all_addresses:
    if address.house_number < 10:
        house_numbers_equal_1.append(address)

# Write addresses to file
with open("addresses.pkl", "wb") as f:
    pickle.dump([vars(address) for address in house_numbers_equal_1], f)

def id_is_present_on_disk(id_: int) -> bool:
    with open("addresses.pkl", "rb") as f:
        # Iterate over each stored address to find out, whether the id is present.
        for address in pickle.load(f):
            if address["id"] == id_:
                return True
        return False

contained_ids_without_bloom_filter: set[int] = set()
start_exec = time.time()
for address in all_addresses:
    if id_is_present_on_disk(address.id):
        contained_ids_without_bloom_filter.add(address.id)
end_exec = time.time()
no_filter: float = end_exec - start_exec
print(f"Time taken for evaluating without any bloom filter: {no_filter} seconds")

Time taken for evaluating without any bloom filter: 0.7118265628814697 seconds

# Define method to go through all addresses in order to identify those that are contained in set
required_time: list[float] = []
contained_ids_with_bloom_filter: list[set[int]] = []


def check_all_addresses(bloom_filter: BloomFilter):
    contained_ids: set[int] = set()
    start_exec = time.time()
    for address in all_addresses:
        if bloom_filter.contains_key(address.id):
            if id_is_present_on_disk(address.id):
                contained_ids.add(address.id)
    end_exec = time.time()
    required_time.append(end_exec - start_exec)
    contained_ids_with_bloom_filter.append(contained_ids)


# Create bloom filters with different bit sizes
numbers_of_available_bits: list[int] = [10, 20, 50, 100, 200, 500, 1000, 2000, 5000]
for number in numbers_of_available_bits:
    # Create Bloom Filter
    bloom_filter: BloomFilter = BloomFilter(
        data_objects=house_numbers_equal_1,
        key_attribute_name="id",
        number_of_available_bits=number,
        bit_sequence_type=BitListBitSequence,
    )
    check_all_addresses(bloom_filter)

x = np.arange(len(numbers_of_available_bits))  # the label locations
fig, ax = plt.subplots(
    layout="constrained",
    figsize=(5, 4),
)

ax.plot(x, required_time, label="Bloom Filter")
plt.axhline(y=no_filter, color="r", linestyle="--", label="No Bloom Filter")
plt.legend()
ax.set_ylabel("Required Time to answer the query in s")

ax.set_xlabel("Number of bits stored")
ax.set_xticks(x, numbers_of_available_bits)
plt.show()

result_dict: dict[str, tuple[list[float], list[float], list[int]]] = {
    index_type: ([], [], [])
    for index_type in [
        "BooleanListBitSequence",
        "IntegerBitSequence",
        "CompressedIntegerBitSequence",
        "CompressedBooleanListBitSequence",
    ]
}
bitset_types: dict[
    str, tuple[Type[UncompressedBitSequence], Type[CompressedBitSequence] | None]
] = {
    "IntegerBitSequence": (IntegerBitSequence, None),
    "BooleanListBitSequence": (BitListBitSequence, None),
    "CompressedIntegerBitSequence": (IntegerBitSequence, WAHBitSequence),
    "CompressedBooleanListBitSequence": (BitListBitSequence, WAHBitSequence),
}
marker_dict: dict[str, str] = {
    "IntegerBitSequence": "*",
    "BooleanListBitSequence": "D",
    "CompressedIntegerBitSequence": "s",
    "CompressedBooleanListBitSequence": "^",
}
word_lengths: list[int] = [4, 8, 16, 32, 64, 128, 256]
benchmark_results: list[BitSequence] = []
for bitset_type, result_list in result_dict.items():
    for i in word_lengths:
        WAHBitSequence.WORD_LENGTH = i

        # Use median out of multiple operations
        index_times = []
        exec_times = []
        bits_stored = None
        for _ in range(2):
            index_time, exec_time, bits_stored, result_bits = (
                create_indexes_and_execute_query(
                    bit_sequence_type=bitset_types[bitset_type][0],
                    compression_type=bitset_types[bitset_type][1],
                    print_info=False,
                )
            )
            index_times.append(index_time)
            exec_times.append(exec_time)
            benchmark_results.append(result_bits)

        result_list[0].append(statistics.median(index_times))
        result_list[1].append(statistics.median(exec_times))
        result_list[2].append(bits_stored)

x = np.arange(len(word_lengths))  # the label locations
fig, (ax_index_time, ax_exec_time, ax_bits_stored) = plt.subplots(
    layout="constrained", figsize=(12, 4), ncols=3
)
for bitset_type, bitset_type_tuple in result_dict.items():
    ax_index_time.plot(
        x, bitset_type_tuple[0], label=bitset_type, marker=marker_dict[bitset_type]
    )
    ax_exec_time.plot(
        x, bitset_type_tuple[1], label=bitset_type, marker=marker_dict[bitset_type]
    )
    ax_bits_stored.plot(
        x, bitset_type_tuple[2], label=bitset_type, marker=marker_dict[bitset_type]
    )

ax_index_time.set_ylabel("Required Time to create an index in s")
ax_exec_time.set_ylabel("Required Time to execute the query in s")
ax_exec_time.legend(loc="center", bbox_to_anchor=(0.5, 1.2))
ax_bits_stored.set_ylabel("Number of bits stored")

ax_exec_time.set_yscale("log")
ax_bits_stored.set_yscale("log")

for ax in [ax_index_time, ax_exec_time, ax_bits_stored]:
    ax.set_xlabel("Bit length of a word")
    ax.set_xticks(x, word_lengths)

import unittest


class BitmapTests(unittest.TestCase):
    def test_uncompressed_bitmap_index(self):

        # Equality-encoded city + street indexes store one bit per address per
        # distinct value, i.e. len(addresses) * (num cities + num streets) bits.
        # Derived from the data so the check holds for any house-number range.
        expected_stored_bits: int = len(all_addresses) * (
            len(possible_cities) + len(possible_streets)
        )
        for bits_stored in [query_1_integer_stored_bits, query_1_boolean_stored_bits]:
            self.assertEqual(bits_stored, expected_stored_bits)
        for bitmap_result in [query_1_integer_result_bits, query_1_boolean_result_bits]:
            self.assertEqual(bitmap_result, query_1_naive_result)

        # Check house number smaller than 500 result
        naive_smaller_than_500: set[int] = set()
        for address in all_addresses:
            if address.house_number < 500:
                naive_smaller_than_500.add(address.id)

        for bitmap_result in [smaller_than_500_equality, smaller_than_500_range]:
            self.assertEqual(naive_smaller_than_500, bitmap_result.as_set())

        # House number greater/equal 300 or equal 30
        naive_ge_300_e_30: set[int] = set()
        for address in all_addresses:
            if address.house_number >= 300 or address.house_number == 30:
                naive_ge_300_e_30.add(address.id)
        self.assertEqual(naive_ge_300_e_30, ge_300_e_30_result.as_set())

        # Combined result as described above
        naive_combined_result: set[int] = set()
        for address in all_addresses:
            if address.house_number == 23 or (
                address.city == "New York" and address.house_number < 40
            ):
                naive_combined_result.add(address.id)
        self.assertEqual(naive_combined_result, combined_result.as_set())

    def test_bloom_filters(self):

        # Bloom Filter should produce the same result
        for contained_ids in contained_ids_with_bloom_filter:
            self.assertEqual(contained_ids, contained_ids_without_bloom_filter)

    def test_benchmark(self):

        # Verify that all benchmark produce the correct results
        for result in benchmark_results:
            self.assertEqual(query_1_naive_result, result)


unittest.main(
    argv=["ignored", "-v"], defaultTest="BitmapTests", verbosity=2, exit=False
)