Partie 20: Réindexation et compaction de tables

Bonjour à toutes et tous 😃

Lors de la partie précédente, nous avons supprimé des enregistrements de la base de données. Mais en agissant ainsi, nous avons laissé des trous dans nos pages de stockage.

En effet, chaque insertion incrémente un ID interne propre à chaque page, ce qui fait que même si nous supprimons un enregistrement, son emplacement n'est pas réaffecté pour l'enregistrement suivant. Au lieu de cela, une nouvelle entrée est créée dans la page ayant le plus grand ID interne. S'il n'y a pas de place, alors nous ajoutons une nouvelle page.

Nous allons considérer que les pages possèdent au maximum 3 enregistrements.

# page 0
0 => data0
1 => data1
2 => data2
# page 1
3 => data3
4 => data4

Supprimons quelques enregistrements.

# page 0 
0 => NULL
1 => data1
2 => NULL
# page 1
3 => data3
4 => NULL

Si l'on ajoute un nouvel enregistrement, il va se trouver dans la page 1. Et si l'on ajoute un autre enregistrement, il va se trouver dans la page 2 nouvellement créée.

# page 0
0 => NULL
1 => data1
2 => NULL
# page 1
3 => data3
4 => NULL
5 => data5
# page 2
6 => data6

On se retrouve avec des trous dans nos pages aux index 0, 2 et 4.

Ces emplacements vides imposent un surplus de pages. Nous voulons réduire le nombre de pages nécessaire.

# page 0
0 => data6
1 => data1
2 => data5
# page 1
3 => data3

Comme vous pouvez le voir, les ID internes 0 et 2 ont été réaffectés pour deux enregistrements déjà existants, respectivement data6 et data5.

Cependant, les index, et plus particulièrement les index primaires, ne sont pas au courant de ces changements et restent donc dans le même état qu'avant la compaction de la table.

Il va donc falloir mettre à jour ces index pour que notre table reste cohérente.

Aujourd'hui encore, cela ne sera pas si simple, mais cela promet d'être intéressant. 😎

Compactation de la table

Nous allons commencer par nous charger de la compaction de la table.

Notre algorithme de compaction sera volontairement naïf.

Nous allons passer en revue les pages une par une et dès que l'on trouve un emplacement qui a été libéré par une suppression, nous allons le remplacer par l'enregistrement ayant le plus grand ID interne.

Et l'on répète ce processus jusqu'à ce que tous les trous aient été comblés par des les enregistrements de haut ID.

Lorsque le premier trou rencontré se trouve au-delà du nombre total d'enregistrements, le processus de compaction est terminé.

Comme tout cela est un peu compliqué, je vous ai fait une animation pour vous expliquer tout cela.

La légende est la suivante:

• les blocs roses sont les enregistrements actifs de la page
• les blocs verts sont les enregistrements supprimés de la page
• les blocs blancs sont les espaces libres de la page

Les numéros des blocs représentent l'ID de données de l'enregistrement.

L'ID interne lui est représenté par la position du bloc dans chaque rectangle.

Chacun de ces rectangles représente une page de 20 enregistrements.

Aussi bien pour les pages que pour les blocs, les ID se lisent de haut en bas et de gauche à droite.

On y voit tout le processus de déplacement des données.

D'abord on recherche le premier emplacement supprimé. Il se trouve être l'ID 1. On recherche alors le dernier enregistrement actif de la base de données. Dans notre cas c'est l'enregistrement ayant l'ID 76.

On inverse les deux enregistrements 1 et 76.

Sur l'animation, je swap les deux enregistrements, mais dans la réalité je replace l'enregistrement ayant l'ID 1 par celui ayant l'ID 76.

Comme l'ID 1 est supprimé, il n'est pas nécessaire de conserver les donnees de l'enregistrement ayant l'ID 1.

On s'économise alors une copie de données par inversion.

Et on repart pour un tour en recherchant le prochain emplacement supprimé et le dernier enregistrement actif, respectivement l'ID 16 et l'ID 75.

La recherche des emplacements vides et actifs est optimisée par notre système de header de page qui nous permet de trouver rapidement les emplacements vides.

La dernière inversion est entre l'enregistrement ayant l'ID 59 et celui ayant l'ID 63.

À partir de ce moment, le dernier enregistrement actif est l'enregistrement ayant l'ID 62 et le premier emplacement supprimé est celui ayant l'ID 63.

L'algorithme de compaction est donc terminé.

Nous nous retrouvons avec tous les enregistrements actifs de la base de données contigus et l'ensemble des enregistrements supprimés à la suite.

La dernière étape consiste à les marquer comme inutilisés en modifiant l'ID interne de la table pour qu'elle écrive le prochain enregistrement à l'ID interne 63.

Il existe un cas particulier de l'algorithme qui amène à vider entièrement une page.

Dans ce cas, la page est devenue inutile et peut être supprimée.

Supression des pages

Nous avons donc besoin d'un mécanisme pour réaliser la supression des pages.

Nous allons implémenter ce système sur le pager associé à la table.

impl Pager {
    /// Delete the last n pages
    pub fn delete_last_n_pages(&mut self, n: usize) {
        self.pages.drain(self.pages.len() - n..);
    }
}

L'API drain consomme les élément d'un itérateur et les supprime de celui-ci.

Implémentation du garbage collector

Le processus de compaction nécessite deux étapes:

• le déplacement des enregistrements supprimés à la suite des enregistrements actifs
• la suppression des pages vides

La suppression des pages a été réalisée juste au-dessus.

Il nous reste à réaliser le déplacement des enregistrements supprimés.

Pour faire cela, nous allons créer une structure que nous allons appeler garbage collector.

Et associer cette structure à la table.

struct GarbageCollector<'a> {
    table: &'a mut Table,
}

Comme la compaction est une opération complexe, nous allons créer une structure que nous appellerons VacuumReport qui nous donnera des statistiques sur la compaction.

struct VacuumReport {
    pub table_name: String,
    pub swapped_records: BTreeMap<usize, usize>,
    pub pages_deleted: usize,
    pub vacuumed_records: usize,
}

Nous pouvons désormais implementer notre garbage collector.

impl GarbageCollector<'_> {
    /// Vacuum the table
    /// Workflow:
    /// 1. Start from the first page
    /// 2. Get the header of the page
    /// 3. Get the list of deleted records in the page
    /// 4. Get the last page lower than the last available record
    /// 5. Swap the first deleted record with the last available record
    /// 6. Scan for the next last available record
    /// 7. Repeat until the last deleted record exceeds the last available record
    pub fn vacuum(&mut self) -> Result<VacuumReport, VacuumError> {
        let old_row_number = self.table.row_number;

        // on initialise le curseur de l'enregistrement
        // avec le premier de la première page
        let mut current_row = 0;

        // on initialise le curseur de la derniere page
        // avec le dernier de la derière page
        let mut last_row = self.table.row_number;

        // on initialise le rapport
        let mut report = VacuumReport {
            table_name: self.table.table_name.clone(),
            ..Default::default()
        };

        // on cherche le premier enregistrement supprimé
        while current_row < last_row {
            
            // via les headers de page
            // nous sommes capables de savoir si un enregistrement 
            // est supprimé ou non sans devoir lire les données de l'enregistrement
            if self
                .table
                .pager
                .check_if_record_deleted(current_row)
                .map_err(VacuumError::Page)?
            {
                // on cherche le dernier enregistrement actif
                while last_row > current_row {
                    // là aussi le header de page permet d'accélérer la recherche
                    if !self
                        .table
                        .pager
                        .check_if_record_deleted(last_row - 1)
                        .map_err(VacuumError::Page)?
                    {
                        // si on trouve un enregistrement actif
                        report.vacuumed_records += 1;
                        report.swapped_records.insert(current_row, last_row - 1);

                        // on va le remplacer par l'enregistrement supprimé
                        self.table
                            .pager
                            .replace_record(last_row - 1, current_row)
                            .map_err(VacuumError::Page)?;

                        // et on va le marquer comme actif
                        self.table
                            .pager
                            .set_record_undeleted(current_row)
                            .map_err(VacuumError::Page)?;
                        
                        // on décrémente le dernier enregistrement de 1
                        last_row -= 1;
                        // et on quitte la boucle
                        break;
                    }
                    // sinon on poursuit la recherche de l'enregistrement actif
                    last_row -= 1;
                }
            }
            // sinon on poursuit la recherche de l'enregistrement supprimé
            current_row += 1;
        }

        // on va mettre la table sur le dernier enregistrement actif
        self.table.row_number = last_row;

        // calcul le nombre d'enregistrements supprimés
        let delta_row_number = old_row_number - self.table.row_number + 1;

        // calcul le nombre de pages supprimables
        let pages_to_delete = delta_row_number / self.table.pager.rows_per_page;

        // on supprime les pages inutiles
        self.table.pager.delete_last_n_pages(pages_to_delete);

        // on met à jour le rapport
        report.pages_deleted = pages_to_delete;

        Ok(report)
    }
}

Commande VACUUM TABLE

Contrairement à ce que laisse penser son nom de "garbage collector", celui-ci n'est ni une tâche de fond, ni une opération autonome.

Elle sera déclenchée par la commande VACUUM TABLE.

VACUUM TABLE table_name;

Notre commande sera matérialisée par la structure suivante:

pub struct VacuumCommand {
    table_name: String,
}

Et son parser:

impl<'a> Visitable<'a, u8> for VacuumCommand {
    fn accept(scanner: &mut Scanner<'a, u8>) -> parser::Result<Self> {
        // on nettoie de potentiel blanc
        scanner.visit::<OptionalWhitespaces>()?;
        // on reconnait le token VACUUM
        recognize(Token::Vacuum, scanner)?;
        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;
        // on reconnait le token TABLE
        recognize(Token::Table, scanner)?;
        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;

        // on reconnait une literal string représentant le nom de table qui se termine
        // soit un par un blanc, soit par un point-virgule
        let table_name_tokens = Forecaster::new(scanner)
            .add_forecastable(UntilToken(Token::Semicolon))
            .add_forecastable(UntilToken(Token::Whitespace))
            .forecast()?
            .ok_or(ParseError::UnexpectedToken)?;

        let table_name =
            String::from_utf8(table_name_tokens.data.to_vec()).map_err(ParseError::Utf8Error)?;
        scanner.bump_by(table_name_tokens.data.len());

        // on nettoie les potentiels blancs
        scanner.visit::<OptionalWhitespaces>()?;
        // on reconnaît le point-virgule final
        recognize(Token::Semicolon, scanner)?;
        Ok(VacuumCommand { table_name })
    }
}

Réindexation de la table

Comme je vous l'ai dit en introduction, nous avons besoin de réindexer notre table après compaction pour éviter une décohérence des index par rapport au contenu de la table.

Mais fort heureusement, la réindexation n'est pas un processus si complexe que cela, au vu de l'avancé actuelle de notre projet.

L'algorithme de réindexation est le suivant:

• Pour tous les index de la table, vider les entrées
• Scanner tous les enregistrements de la table
• Pour chaque enregistrement, réindexer les données

On rajoute sur le trait Indexable une méthode clear qui a pour responsabilité de vider un index.

pub trait Indexable {
    fn clear(&mut self) -> Result<(), IndexError>;
}

Nous avons deux implémentations de Indexable :

• ValueIndex
• PrimaryIndex

impl Indexable for PrimaryIndex {
    fn clear(&mut self) -> Result<(), IndexError> {
        self.inner.clear();
        Ok(())
    }
}

impl Indexable for ValueIndex {
    fn clear(&mut self) -> Result<(), IndexError> {
        self.inner.clear();
        Ok(())
    }
}

On implémente alors une méthode sur l'IndexRegistry qui permet de vider tous les index de la table.

impl IndexRegistry {
        /// Clear all indexes
    pub fn clear_all_indexes(&mut self) -> Result<(), IndexError> {
        for indexes in self.indexes.values_mut() {
            for index in indexes.values_mut() {
                index.clear()?;
            }
        }
        Ok(())
    }
}

On implémente ensuite la réindexation de la table.

Tout comme lorsque l'on a réalisé la suppression dans la partie 19, nous avons besoin de lire les enregistrements de la table et de les reindexer.

Mais là encore, il n'est pas possible de le faire lors du scan de la table, on doit d'abord lire un certain nombre d'enregistrements, que l'on puisse ensuite reindexer.

On défini alors une "continuation" qui correspond au dernier enregistrement lu.

Ce qui permet de relancer le scan en commençant à partir de cette continuation.

Et l'on recommence ainsi jusqu'a ce que l'on ait lu tous les enregistrements de la table.

impl Table {
    /// Reindex a part of the table
    fn reindex_part(&mut self, rows: Vec<Row>) {
        // on parcourt tous les enregistrements du paquet
        for row in rows {
            // on construit un HashMap avec les colonnes et leurs valeurs
            let row_data = self
                .schema
                .columns
                .iter()
                .cloned()
                .zip(row.1.into_iter())
                .collect();

            println!("Reindexing row {}", row.0);

            // on reindex l'enregistrement
            self.indexes.insert_into_index(&row_data, row.0).unwrap();
        }
    }

    /// Reindex the table
    pub fn reindex(&mut self) -> Result<(), SelectError> {
        self.indexes
            .clear_all_indexes()
            .map_err(SelectError::IndexError)?;

        let mut continuation = 0;

        // on lit par paquets de 20 enregistrements
        loop {
            let query_engine = QueryEngine::new(self);
            let scanner = Scanner::new(query_engine);
            let rows = scanner
                .skip(continuation)
                .take(20)
                .collect::<Result<Vec<Row>, _>>()?;

            if rows.is_empty() {
                break;
            }

            continuation += rows.len();

            // que l'on reindex
            self.reindex_part(rows);
        }

        Ok(())
    }
    }

Implémentation de la commande VACUUM

Maintenant que nous avons tous les éléments nécessaires, nous pouvons implémenter la commande VACUUM.

On implémente la commande sur la base de données:

impl Database {
    pub fn vacuum(&mut self, table_name: String) -> Result<VacuumReport, VacuumError> {
        match self.tables.get_mut(&table_name) {
            Some(table) => table.vacuum(),
            None => Err(VacuumError::TableNotExist(table_name))?,
        }
    }
}

Puis sur la table:

On y construit le garbage collector que l'on exécute, ce qui produit un rapport de compaction.

On réindex alors la table.

Et on renvoie le rapport.

impl Table {
    /// Vacuum the table and reindex it
    pub fn vacuum(&mut self) -> Result<VacuumReport, VacuumError> {
        let mut gc = garbage_collector::GarbageCollector::new(self);
        let report = gc.vacuum()?;
        self.reindex().map_err(VacuumError::SelectionError)?;
        Ok(report)
    }
}

Afin de rendre plus intelligible le rapport de compaction, on définit le trait Display:

impl Display for VacuumReport {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut buffer = format!("\nVacuum Report for table: {}\nPages deleted: {}\nVacuumed records: {}\nSwap records:\n", self.table_name, self.pages_deleted, self.vacuumed_records);
        for (i, j) in self.swapped_records.iter() {
            buffer = format!("{buffer}\tSwap #{i} -> #{j}\n");
        }
        write!(f, "{buffer}")
    }
}

Ce qui permet de l'afficher en retour de commande dans la méthode run du projet:

pub fn run() -> Result<(), Box<dyn Error>> {
    let mut database = database::Database::new();

    // snip ...

    loop {
        
        match database.run(command) {
            // snip ...
            Ok(ExecuteResult::Vaccum(report)) => {
                println!("{}", report);
            }
            Ok(ExecuteResult::Nil) => {}
            Err(err) => {
                println!("{}", err);
            }
        }
    }
}

On test !

On va maintenant tester notre commande VACUUM.

CREATE TABLE Users (id INTEGER PRIMARY KEY, value TEXT(180));
CREATE INDEX idx_value ON Users (value);

Puis on ajoute quelques enregistrements:

-- INSERT for 1 to 77
INSERT INTO Users (id, value) VALUES (0, 'test_0');
INSERT INTO Users (id, value) VALUES (1, 'test_1');
INSERT INTO Users (id, value) VALUES (2, 'test_2');
INSERT INTO Users (id, value) VALUES (3, 'test_3');
-- ...
INSERT INTO Users (id, value) VALUES (74, 'test_74');
INSERT INTO Users (id, value) VALUES (75, 'test_75');
INSERT INTO Users (id, value) VALUES (76, 'test_76');

La liste complète est ici.

Puis on supprime quelques enregistrements pour créer des trous dans nos pages:

-- DELETE 1, 8, 16, 19, 28, 33, 35, 37, 39, 49, 52, 55, 56, 59, 64, 65, 66, 69, 71, 72,
DELETE FROM Users WHERE id = 1;
DELETE FROM Users WHERE id = 8;
DELETE FROM Users WHERE id = 16;
DELETE FROM Users WHERE id = 19;
DELETE FROM Users WHERE id = 28;
DELETE FROM Users WHERE id = 33;
DELETE FROM Users WHERE id = 35;
DELETE FROM Users WHERE id = 37;
DELETE FROM Users WHERE id = 39;
DELETE FROM Users WHERE id = 49;
DELETE FROM Users WHERE id = 52;
DELETE FROM Users WHERE id = 55;
DELETE FROM Users WHERE id = 56;
DELETE FROM Users WHERE id = 59;
DELETE FROM Users WHERE id = 64;
DELETE FROM Users WHERE id = 65;
DELETE FROM Users WHERE id = 66;
DELETE FROM Users WHERE id = 69;
DELETE FROM Users WHERE id = 71;
DELETE FROM Users WHERE id = 72;

On ajoute un enregistrement:

INSERT INTO Users (id, value) VALUES (666, 'satan');

Si on liste les enregistrements de la table:

SELECT * FROM Users;

On obtient:

On remarque que:

• les enregistrements sont bien supprimés.
• le dernier enregistrement est ajouté à la fin.
• les ID internes et de l'enregistrement sont synchronisés.

On va maintenant lancer le processus de compaction.

VACUUM TABLE Users;

On obtient:

Vacuuming table Users
Clearing index PRIMARY
Clearing index idx_value
Reindexing row 0
Insert [Text("test_0")] into non unique index idx_value => row_id 0
Reindexing row 1
Insert [Text("satan")] into non unique index idx_value => row_id 1
Reindexing row 2
Insert [Text("test_2")] into non unique index idx_value => row_id 2
Reindexing row 3
// ...
Reindexing row 55
Insert [Text("test_60")] into non unique index idx_value => row_id 55
Reindexing row 56
Insert [Text("test_58")] into non unique index idx_value => row_id 56
Reindexing row 57
Insert [Text("test_57")] into non unique index idx_value => row_id 57

Vacuum Report for table: Users
Pages deleted: 1
Vacuumed records: 13
Swap records:
        Swap #1 -> #77
        Swap #8 -> #76
        Swap #16 -> #75
        Swap #19 -> #74
        Swap #28 -> #73
        Swap #33 -> #70
        Swap #35 -> #68
        Swap #37 -> #67
        Swap #39 -> #63
        Swap #49 -> #62
        Swap #52 -> #61
        Swap #55 -> #60
        Swap #56 -> #58

On y voit la purge des index puis la réindaxation des enregistrements.

Et finallement le rapport de compaction.

On y remarque les opérations d'inversement de blocs dans les pages.

Le nombre de pages supprimées ainsi que les rows supprimées du stockage.

On recommence à lire la table:

SELECT * FROM Users;

Cette fois-ci les ID internes et les ID ne sont plus synchronisés.

Par contre, il n'y a plus de saut sur les ID internes.

De même l'ID interne maximale qui était de 77 est passé à 57.

Dans l'opération, nous venons de supprimer du stockage sans perte de données ! 😄

Parlant de perte de données, vérifions que nos index sont toujours cohérents:

EXPLAIN SELECT * FROM Users WHERE value = "test_76";

On obtient le plan suivant:

Scan index idx_value : value = 'test_76' for table Users
Filter: value = 'test_76'

Donc le query planner ira chercher sa données depuis l'index idx_value.

Donc si on a foiré la réindexation de l'index idx_value, il n'y aura perte de données.

Vérifions cela:

SELECT * FROM Users WHERE value = "test_76";

Nickel, pas de perte de données ! 🎉

Conclusion

Clairement, nous sommes dans l'optimisation, mais cela me semblait intéressant de poser les bases de la réindexation de données.

Et puis, je me suis bien amusé à réaliser les petites animations. 😄

Dans la prochaine partie nous allons implementer la commande UPDATE table SET value = x WHERE id = y.

Merci de votre lecture ❤️

Vous pouvez trouver le code la partie ici et le diff là.

Auteur: Akanoa

Je découvre, j'apprends, je comprends et j'explique ce que j'ai compris dans ce blog.

Ce travail est sous licence CC BY-NC-SA 4.0.