Partie 19: Suppression d'enregistrements

Bonjour à toutes et tous 😃

Lors de la partie précédente, nous avons enfin atteint le stade où nous sommes en mesure de rechercher des enregistrements en nous basant sur des index.

Nous allons réutiliser ces index et le plan logique pour supprimer des enregistrements.

Mais avant cela, nous allons devoir modifier le format de stockage de notre base de données pour indiquer quels enregistrements sont supprimés.

Nous allons également devoir modifier le scanner pour renvoyer le rowid des enregistrements à supprimer.

Cet article sera plus simple que le précédent, mais il y a quelques petites choses à faire tout de même.

Trêve de palabres, c'est parti ! 😄

Page header

Théorie

Pour le moment nos pages de stockage ont le layout suivant:

null_table tuple null_table tuple null_table tuple null_table tuple
null_table tuple null_table tuple null_table tuple null_table tuple
null_table tuple null_table tuplenull_table tuple null_table tuple
null_table tuple null_table tuple null_table tuple padding

Chaque enregistrement est préfixé par un tableau de bits qui indique quels champs du tuple sont NULL. Nous l'avons appelé null_table. Puis le tuple en lui-meme.

Et tout de suite après une autre null_table du tuple suivant, puis le tuple suivant et ainsi de suite.

La page se termine par un padding permettant à nos enregistrements de rester aligné sur une seule page.

Nous allons modifier le format de stockage de notre base de données pour indiquer quels enregistrements sont supprimés.

Pour cela, nous allons rajouter un header permettant de définir les métadonnées de la page.

Le nouveau format de page devient:

header           null_table tuple null_table tuple null_table tuple
null_table tuple null_table tuple null_table tuple null_table tuple
null_table tuple null_table tuple null_table tuple null_table tuple
null_table tuple null_table tuple null_table tuple padding

Ce header lui-même contient une null_table qui indique quels offsets de la page doivent être ignorés.

Nos null_table sont alignées sur un octet, donc 8 bits au minimum. Si cela n'est pas suffisant, nous ajoutons des octets pour le reste.

Pour une page contenant 18 enregistrements, nous aurons un header de 3 octets, car 3 * 8 = 24 bits.

0000 0000 0000 0000 0000 0000

Pour supprimer le premier enregistrement, il faut modifier le premier bit du premier octet du header.

0000 0001 0000 0000 0000 0000
        ↑

Pour supprimer le quatrième enregistrement, il faut modifier le quatrième bit du premier octet du header.

0000 1000 0000 0000 0000 0000
     ↑

Pour supprimer le 18e enregistrement, il faut modifier le 2e bit du 3e octet du header.

0000 0000 0000 0000 0000 0010
                           ↑

J'explique les techniques pour lire les bits dans la partie 14.

En résumé, pour vérifier si un enregistrement est supprimé, on donne son index dans la page et on regarde si le bit correspondant est 1 ou 0.

Pour cela, on décale le bit qui nous intéresse et on le compare avec 0b1.

Si ce bit vaut 0b1, alors le ET logique renvoie 0b1 et la condition est true. Le bit est 1, donc l'enregistrement est supprimé.

Si ce bit vaut 0b0, alors le ET logique renvoie 0b0 et la condition est false. Le bit est 0, donc l'enregistrement n'est pas supprimé.

partition = index / 8
bit = index % 8
null_table[partition] >> bit & 0b1 == 0b1 

Manipulation du header de la page

Nous allons créer deux fonctions permettant de vérifier les états d'enregistrement et de modifier l'en-tête.

pub const PARTITION_SIZE: usize = 8;

/// Check if a bit in the page header is 1
pub fn is_record_deleted(header: &[u8], index: usize) -> bool {
    let partition = index / PARTITION_SIZE;
    let bit = index % PARTITION_SIZE;

    (header[partition] >> bit) & 0b1 == 0b1
}

/// Set a bit in the page header
pub fn set_record_deleted(header: &mut [u8], index: usize) {
    let partition = index / PARTITION_SIZE;
    let bit = index % PARTITION_SIZE;

    header[partition] |= 0b1 << bit;
}

Les méthodes sont utilisées comme suit:

#[test]
fn test_delete_then_check() {
    // init empty header
    let mut header = [0b0000_0000, 0b0000_0000];
    // delete first record
    set_record_deleted(&mut header, 0);
    // check if first record is deleted
    assert!(is_record_deleted(&header, 0));
    // delete eleventh record
    set_record_deleted(&mut header, 11);
    // check if eleventh record is deleted
    assert!(is_record_deleted(&header, 11));
    // check final header state
    assert_eq!(header, [0b0000_0001, 0b0000_1000]);
}

La référence mutable &mut [u8] permet de modifier le header en fonction des enregistrements que l'on désire supprimer.

Il est à remarquer que les bits après le 18e (de 19 à 24) ne sont pas considérés comme des enregistrements et ne sont là que pour respecter l'alignement sur un octet.

Association du header à la page

Maintenant que nous avons notre header rudimentaire, nous allons le lier aux pages.

Pour cela, il va falloir faire un peu de gymnastique mathématique.

Nous connaissons la taille d'un enregistrement, qui est définie par le schéma de la table.

Nous connaissons la taille de la page, qui est fixe et qui vaut 4096 octets.

Nous sommes donc capables de calculer le nombre d'enregistrements par page.

$$ \text{nb of records per page} = \frac{\text{page size}}{\text{record size}} $$

Nous avons besoin d'une null_table qui accueille au moins le nombre d'enregistrement par page.

Les null_table sont des multiples de 8bits. Nous avons besoin de 1 bit par enregistrement.

$$ \text{null table size} = \left \lceil \frac{\text{nb of records per page}}{8} \right \rceil $$

Nous utilisons ici le ceil pour être certains d'avoir au moins 1 bit par enregistrement.

Si nous faisions la division entière pour 18 enregistrements, cela donnerait :

$$ \frac{\text{18}}{8} = 2.25 $$

Ce qui est effectivement ce que l'on s'aperçoit dans l'occupation de la null_table. Mais le souci c'est que la division entière nous donnerait seulement 2 octets pour 18 enregistrements.

L'utilisation de ceil nous permet de prendre le plus grand entier qui est inférieur ou égal à notre division. Ce qui nous donne 3 octets. Avec 3 * 8 = 24 bits, nous avons suffisamment de place pour 18 enregistrements.

Une fois que nous avons la taille de notre null_table, nous pouvons la retrancher à la taille de la page.

$$ \text{page usable space} = \text{page size} - \text{null table size} $$

Ensuite, nous recalculons le nombre d'enregistrements par page afin d'éviter tout désalignement des enregistrements suite à l'introduction de l'en-tête.

$$ \text{nb of records per page computed} = \frac{\text{page usable space}}{\text{record size}} $$

Et on recalcule la taille de la page utilisable pour les enregistrements.

$$ \text{real page usable space} = \text{nb of records per page computed} * \text{record size} $$

Cette opération peut nous donner un

$$\text{nb of records per page computed} < \text{nb of records per page computed}$$

et donc potentiellement un octet de moins nécessaire pour la null_table, mais c'est un sacrifice sur 4 Ko de page qui est acceptable.

Quoi qu'il en soit, la taille de la page que l'on utilisera pour les calculs de pages de stockage et d'offsets désormais sera $\text{real page usable space}$.

impl Pager {
    /// Size of a physical page
    pub const PAGE_SIZE: usize = 4096;

    pub fn new(row_size: usize) -> Self {
        // Calculate the maximum number of rows that can fit in a page
        let max_row_per_page = PAGE_SIZE / row_size;
        // Calculate the size of the page header based on partitions
        let page_header = max_row_per_page.div_ceil(PARTITION_SIZE);
        // Determine the usable space in the page by subtracting the header size from the total page size
        let usable_space = PAGE_SIZE - page_header;
        // Calculate the number of rows that can fit in the usable space
        let available_rows = usable_space / row_size;
        // Calculate the total size of the page based on the available rows and row size

        let page_size = available_rows * row_size;

        Pager {
            pages: BTreeMap::default(),
            row_size,
            page_size,
            header_size: page_header,
        }
    }
}

Nous avons désormais un emplacement réservé dans la page qui ne peut plus être occupé par un enregistrement. Nous pouvons donc l'utiliser comme header.

Récupération du header dans la page

Maintenant que nous avons un emplacement pour notre header, il va s'agir de le positionner dans la page.

J'ai décidé de le mettre en haut de la page. Ce qui signifie en octet 0. Nous connaissons sa taille, nous pouvons donc récupérer la slice d'octets correspondants.

Pour cela, nous allons modifier les fonctions get et get_mut qui précédemment ne permettaient que de récupérer les octets après un certain offset.

Or, nous nous devons de récupérer le header compris entre les octets 0 et header_size.

Pour cela nous allons utiliser le trait SliceIndex qui permet de prendre un sous ensemble de slice.

impl Page {
    pub fn get_mut<S: SliceIndex<[u8], Output = [u8]>>(&mut self, range: S) -> Option<&mut [u8]> {
        self.inner.get_mut(range)
    }

    pub fn get<S: SliceIndex<[u8], Output = [u8]>>(&self, range: S) -> Option<&[u8]> {
        self.inner.get(range)
    }
}

Une fois cela fait, on peut définir les méthodes pour récupérer le header.

Le header sera récupéré par row ID, car nous voulons déterminer si un enregistrement est supprimé ou non ou le supprimer.

Pour cela, nous réutilisons la méthode get_position qui renvoie une Position que nous avons définie précédemment.

Celui-ci nous donne alors la page concernée par l'enregistrement.

On récupère ensuite la page et on extrait le header.

Comme notre header est de taille header_size et qu'il débute en octet 0, la slice est donc ..self.header_size.

Nous venons de virtuellement positionner notre header en octet 0.

impl Pager {
    /// Gets the header of the record at the given a position
    fn get_header(&self, position: &Position) -> Result<&[u8], PageError> {
        let Position { page_number, .. } = position;

        let page = self
            .pages
            .get(page_number)
            .ok_or(PageError::PageNotFound(*page_number))?;
        page.get(..self.header_size)
            .ok_or(PageError::PageSliceRangeTo(..self.header_size))
    }

    /// Gets the mutable header of the record at the given position
    fn get_mut_header(&mut self, position: &Position) -> Result<&mut [u8], PageError> {
        let Position { page_number, .. } = position;

        let page = self
            .pages
            .get_mut(page_number)
            .ok_or(PageError::PageNotFound(*page_number))?;

        page.get_mut(..self.header_size)
            .ok_or(PageError::PageSliceRangeTo(..self.header_size))
    }
}

Mais du coup, tous les calculs de position et d'offsets des enregistrements doivent prendre en compte la place que prend le header et donc décaler d'autant.

Et c'est pour cela que nous rajoutons un self.header_size aux offsets.

impl Pager {
    fn get_position(&self, row_number: usize) -> Position {
        // calcul de l'index global
        let global_index = row_number * self.row_size;
        // calcul la page contenant le record
        let page_number = global_index / self.page_size;
        // calcul l'offset de la valeur dans cette page
        let offset = (global_index - page_number * self.page_size) + self.header_size;
        Position {
            page_number,
            offset,
        }
    }
}

Désormais, le header est devenu invisible pour les calculs de pages et d'offsets qui conditionnent l'accès aux enregistrements.

Nous avons bien notre layout de page souhaité.

header
record
record
record
...
padding

On en profite pour définir deux méthodes, l'une pour supprimer un enregistrement et l'autre pour savoir si un enregistrement est supprimé.

impl Pager {
    /// Check if the record at the given row number is deleted.
    /// returns None if the record does not exist
    /// returns Some(true) if the record is deleted
    /// returns Some(false) if the record is not deleted
    pub fn check_if_record_deleted(&self, row_number: usize) -> Result<bool, PageError> {
        let positon = self.get_position(row_number);
        let index = positon.offset / self.row_size;
        let header = self.get_header(&positon)?;
        Ok(data::page_header::is_record_deleted(header, index))
    }

    /// Set a bit in the page header
    /// flags the record as deleted
    pub fn set_record_deleted(&mut self, row_number: usize) -> Result<(), PageError> {
        let position = self.get_position(row_number);
        let index = position.offset / self.row_size;
        let header = self.get_mut_header(&position)?;
        data::page_header::set_record_deleted(header, index);
        Ok(())
    }
}

Pas mal, pas mal, on avance ! 🚀

Renvoyer les row ID lors du scan

Pour le moment, nos méthodes de scans ne renvoient que le tuple du record sans son row ID associé.

Or, pour récupérer le header de la page concernant l'enregistrement à supprimer, nous avons besoin de son row ID.

Nous allons donc modifier le retour de nos itérateurs de scans.

Nous définissons un type alias pour le tuple de row ID et de valeurs.

type Row = (RowId, Vec<Value>);

Puis nous modifions nos itérateurs.

impl Iterator for IndexIterator<'_> {
    type Item = Result<Row, SelectError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.row_ids.next() {
                None => break None,
                Some(row_id) => {
                    let row = self.query_engine.get_row(*row_id);
                    match row {
                        Ok(Some(row)) => break Some(Ok((*row_id, row))),
                        Ok(None) => continue,
                        Err(err) => break Some(Err(err)),
                    }
                }
            }
        }
    }
}

impl Iterator for Scanner<'_> {
    type Item = Result<Row, SelectError>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            // on va jusqu'au dernier tuple de la table
            if self.current_row >= self.query_engine.get_table().row_number {
                return None;
            }

            let row = self.query_engine.get_row(self.current_row);

            self.current_row += 1;
            match row {
                Ok(Some(row)) => break Some(Ok((self.current_row - 1, row))),
                // on ignore les tuples supprimés
                Ok(None) => continue,
                Err(err) => break Some(Err(err)),
            }
        }
    }
}

Désormais nos itérateurs produiront des couples RowId et Vec<Value>.

(0, [1, 2, 3])
(1, [4, 5, 6])
(2, [7, 8, 9])

Le premier tuple correspond au row ID 0, le deuxieme au 1 et ainsi de suite.

De fait, nos itérateurs de plan se voient également modifiés.

type PlanExecInput<'a> = Box<dyn Iterator<Item = Result<Row, SelectError>> + 'a>;
/// The output of the step of the plan executed
pub type PlanExecOutput<'a> =
    Result<Box<dyn Iterator<Item = Result<Row, SelectError>> + 'a>, SelectError>;

Un SELECT produira désormais ceci:

SELECT * FROM Client;
0 => [Integer(1), Text("Smith"), Text("John"), Text("M"), Text("Paris")]
1 => [Integer(2), Text("Martin"), Text("Marie"), Text("F"), Text("New York")]
2 => [Integer(3), Text("Haddad"), Text("Karim (كريم)"), Text("M"), Text("Tokyo")]
3 => [Integer(4), Text("Dubois"), Text("Sophie"), Text("F"), Text("Beyrouth")]
4 => [Integer(5), Text("Tanaka"), Text("Hiroshi (ひろし)"), Text("M"), Text("Beyrouth")]
5 => [Integer(6), Text("Yamamoto"), Text("Sakura (さくら)"), Text("F"), Text("Paris")]
6 => [Integer(7), Text("Smith"), Text("Emily"), Text("F"), Text("Osaka")]
7 => [Integer(8), Text("Martin"), Text("Jean"), Text("M"), Text("Lyon")]
8 => [Integer(9), Text("Haddad"), Text("Layla (ليلى)"), Text("F"), Text("New York")]
9 => [Integer(10), Text("Dubois"), Text("Paul"), Text("M"), Text("Tokyo")]

Command DELETE FROM

Maintenant que le stockage pour la suppression est réglé, nous pouvons ajouter la commande DELETE FROM.

DELETE FROM table_name [WHERE condition];

Sa grammaire est très similaire à celle du SELECT, excepté qu'il n'y a pas de projection.

On modélise la commande comme suit:

#[derive(Debug, PartialEq)]
pub struct DeleteFromCommand {
    pub table_name: String,
    pub where_clause: Option<WhereClause>,
}

Que l'on parse de cette maniere:

impl<'a> Visitable<'a, u8> for DeleteFromCommand {
    fn accept(scanner: &mut Tokenizer<'a>) -> parser::Result<Self> {
        // on nettoie les potentiels espaces
        scanner.visit::<OptionalWhitespaces>()?;
        recognize(Token::Delete, scanner)?;
        // on reconnait au moins un espace
        scanner.visit::<Whitespaces>()?;
        // on reconnait le token FROM
        recognize(Token::From, scanner)?;
        // on reconnait au moins un espace
        scanner.visit::<Whitespaces>()?;

        // on reconnait le nom de la table, celui-ci peut soit se terminer par un blanc
        // alors un nouveau groupe commence un WHERE par exemple
        // soit se terminer par un point-virgule, la fin de la commande
        let table_name_tokens = Forecaster::new(scanner)
            .add_forecastable(UntilToken(Token::Whitespace))
            .add_forecastable(UntilToken(Token::Semicolon))
            .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 espaces
        scanner.visit::<OptionalWhitespaces>()?;

        let where_clause = scanner.visit::<Optional<WhereClause>>()?.0;

        // on reconnait le point virgule terminal
        recognize(Token::Semicolon, scanner)?;

        Ok(DeleteFromCommand {
            table_name,
            where_clause,
        })
    }
}

On peut ensuite implémenter la commande au set de commandes connues.

impl<'a> Visitable<'a, u8> for Command {
    fn accept(scanner: &mut Scanner<'a, u8>) -> crate::parser::Result<Self> {
        Acceptor::new(scanner)
            .try_or(|command| Ok(Command::Select(command)))?
            .try_or(|command| Ok(Command::CreateTable(command)))?
            .try_or(|command| Ok(Command::InsertInto(command)))?
            .try_or(|command| Ok(Command::CreateIndex(command)))?
            .try_or(|command| Ok(Command::DeleteFrom(command)))? // <-- ici
            .finish()
            .ok_or(ParseError::UnexpectedToken)
    }
}

Et finalement ajouter la commande elle-meme.

pub enum Command {
    CreateTable(create_table::CreateTableCommand),
    Select(select::SelectCommand),
    InsertInto(insert_into::InsertIntoCommand),
    CreateIndex(create_index::CreateIndexCommand),
    DeleteFrom(delete_from::DeleteFromCommand), // <-- ici
}

Une fois cela fait on peut définir son exécution.

impl Execute for DeleteFromCommand {
    fn execute(
        self,
        database: &mut Database,
        explain: bool,
    ) -> Result<ExecuteResult, ExecutionError> {
        let DeleteFromCommand {
            table_name,
            where_clause,
        } = self;

        database
            .delete(&table_name, where_clause, explain)
            .map_err(ExecutionError::Delete)
    }
}

impl Execute for Command {
    fn execute(
        self,
        database: &mut Database,
        explain: bool,
    ) -> Result<ExecuteResult, ExecutionError> {
        match self {
            Command::CreateTable(command) => command.execute(database, explain),
            Command::Select(command) => command.execute(database, explain),
            Command::InsertInto(command) => command.execute(database, explain),
            Command::CreateIndex(command) => command.execute(database, explain),
            Command::DeleteFrom(command) => command.execute(database, explain), // <-- ici
        }
    }
}

Implémentation de la commande DELETE

Maintenant que nous sommes capables de reconnaître la commande, il faut qu'elle interagisse réellement avec la base de données.

Pour implémenter la commande DELETE, nous allons réutiliser la fonction select de la base de données.

En effet, celle-ci gère toute la mécanique de gestion du plan de recherche optimisé par index.

En première intention, je souhaitais utiliser l'itérateur du scan sans le collecter, et supprimer au fil de l'eau les enregistrements.

Mais le problème est que l'itérateur est une référence non mutable sur la base de données. Et que supprimer un enregistrement nécessite de modifier la base de données.

Or en Rust, il n'est pas possible d'avoir une référence mutable et non mutable en même temps sur la même variable.

Donc je me rabats sur une solution moins optimale, mais qui fonctionne. Scanner quelques enregistrements consomme l'itérateur et le collecte sous forme de vecteur de row ID.

Ainsi, je relâche ma référence non mutable sur la base de données, ce qui me permets d'en déclarer une mutable sur laquelle je peux modifier la base de données.

// maximum number of rows to scan
const MAX_SCAN_LENGTH: usize = 1000;

impl Database {
    
    fn collect_row_ids(
        &self,
        table_name: &str,
        where_clause: &Option<WhereClause>,
        limit: Option<usize>,
    ) -> Result<Vec<usize>, SelectError> {
        let rows = self.select(table_name, where_clause, false)?;
        let ExecuteResult::Tuples(rows) = rows else {
            unreachable!("Explain mode is disabled");
        };
        
        // limit the number of rows to scan 
        // by default, we scan at most 1000 rows
        // if the limit is not set, we scan at most 1000 rows
        let limit = max(limit.unwrap_or(MAX_SCAN_LENGTH), MAX_SCAN_LENGTH);

        let rows_id = rows
            .map(|row| row.map(|row| row.0))
            .take(limit)
            .collect::<Result<Vec<usize>, SelectError>>()?;
        Ok(rows_id)
    }
}

Pour supprimer tous les enregistrements concernés par la commande DELETE, nous allons appeler la fonction collect_row_ids en boucle jusqu'à ce qu'il n'y ait plus de résultats dans l'itérateur.

Le retour de notre appel de suppression sera le nombre de lignes supprimées.

On ajoute dans notre énumération ExecuteResult une variante pour le nombre de lignes affectées.

pub enum ExecuteResult<'a> {
    Nil,
    Tuples(Box<dyn Iterator<Item = Result<Row, SelectError>> + 'a>),
    Explain(Plan),
    AffectedRows(usize),
}

On en profite également pour modifier la variante Tuples pour lui faire renvoyer les row ID associés aux tuples.

// maximum number of rows to scan by default
const DEFAULT_SCAN_LENGTH: usize = 100;

impl Database {
    pub fn delete(
        &mut self,
        table_name: &String,
        where_clause: Option<WhereClause>,
        explain: bool,
    ) -> Result<ExecuteResult, DeletionError> {
        // si on est en mode explain
        if explain {
            return self
                .select(table_name, &where_clause, true)
                .map_err(DeletionError::SelectionError);
        }
        // on définit un compteur de lignes supprimées
        let mut total_deleted = 0;

        // tant qu'il reste des lignes à supprimer
        loop {
            // on collecte les ids des lignes à supprimer
            let rows_id = self
                .collect_row_ids(table_name, &where_clause, Some(DEFAULT_SCAN_LENGTH))
                .map_err(DeletionError::SelectionError)?;

            // si il n'y a plus de lignes à supprimer
            if rows_id.is_empty() {
                break;
            }

            // on incrémente le compteur
            total_deleted += rows_id.len();

            // on récupère une reference mutable sur la table
            let table = self
                .tables
                .get_mut(table_name)
                .ok_or(DeletionError::TableNotExist(table_name.to_string()))?;

            // on supprime les lignes
            for row_id in rows_id {
                table.delete(row_id)?;
            }
        }

        Ok(ExecuteResult::AffectedRows(total_deleted))
    }
    }

La suppression en tant sur telle dans la table est un appel à la fonction set_record_deleted du pager de la table.

impl Table {
    
    pub fn delete(&mut self, row_number: usize) -> Result<(), DeletionError> {
        self.pager
            .set_record_deleted(row_number)
            .map_err(DeletionError::Page)?;
        Ok(())
    }
}

Et voilà c'est terminé ! 🎉

On peut tester

CREATE TABLE Client (
    id INTEGER PRIMARY KEY,
    nom TEXT(50),
    prénom Text(50),
    genre TEXT(2),
    ville Text(100)
);

CREATE UNIQUE INDEX idx_identité ON Client(nom, prénom);
CREATE INDEX idx_ville ON Client(ville);

INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (1, 'Smith', 'John', 'M', 'Paris');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (2, 'Martin', 'Marie', 'F', 'New York');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (3, 'Haddad', 'Karim (كريم)', 'M', 'Tokyo');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (4, 'Dubois', 'Sophie', 'F', 'Beyrouth');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (5, 'Tanaka', 'Hiroshi (ひろし)', 'M', 'Beyrouth');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (6, 'Yamamoto', 'Sakura (さくら)', 'F', 'Paris');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (7, 'Smith', 'Emily', 'F', 'Osaka');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (8, 'Martin', 'Jean', 'M', 'Lyon');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (9, 'Haddad', 'Layla (ليلى)', 'F', 'New York');
INSERT INTO Client (id, nom, prénom, genre, ville) VALUES (10, 'Dubois', 'Paul', 'M', 'Tokyo');

Si on liste nos clients, nous avons 10 lignes. Avec le row ID correspondant.

db > SELECT * FROM Client;
0 => [Integer(1), Text("Smith"), Text("John"), Text("M"), Text("Paris")]
1 => [Integer(2), Text("Martin"), Text("Marie"), Text("F"), Text("New York")]
2 => [Integer(3), Text("Haddad"), Text("Karim (كريم)"), Text("M"), Text("Tokyo")]
3 => [Integer(4), Text("Dubois"), Text("Sophie"), Text("F"), Text("Beyrouth")]
4 => [Integer(5), Text("Tanaka"), Text("Hiroshi (ひろし)"), Text("M"), Text("Beyrouth")]
5 => [Integer(6), Text("Yamamoto"), Text("Sakura (さくら)"), Text("F"), Text("Paris")]
6 => [Integer(7), Text("Smith"), Text("Emily"), Text("F"), Text("Osaka")]
7 => [Integer(8), Text("Martin"), Text("Jean"), Text("M"), Text("Lyon")]
8 => [Integer(9), Text("Haddad"), Text("Layla (ليلى)"), Text("F"), Text("New York")]
9 => [Integer(10), Text("Dubois"), Text("Paul"), Text("M"), Text("Tokyo")]

Supprimons donc tous les clients Homme de la table. Nous avons à supprimer 5 lignes.

db > DELETE FROM Client WHERE genre = 'M';
5 rows affected

Le système nous renvoie que 5 lignes ont été affectées.

On full scan la table.

db > SELECT * FROM Client;
1 => [Integer(2), Text("Martin"), Text("Marie"), Text("F"), Text("New York")]
3 => [Integer(4), Text("Dubois"), Text("Sophie"), Text("F"), Text("Beyrouth")]
5 => [Integer(6), Text("Yamamoto"), Text("Sakura (さくら)"), Text("F"), Text("Paris")]
6 => [Integer(7), Text("Smith"), Text("Emily"), Text("F"), Text("Osaka")]
8 => [Integer(9), Text("Haddad"), Text("Layla (ليلى)"), Text("F"), Text("New York")]

Il n'y effectivement plus que des femmes.

Si on essaie de supprimer l'ID 8, le système nous dit qu'aucune ligne n'a été affectée.

db > DELETE FROM Client WHERE id = 8;
0 rows affected

Par contre si on supprime l'ID 9, le système nous dit que 1 ligne a été affectée.

db > DELETE FROM Client WHERE id = 9;
1 rows affected

Un dernière SELECT nous assure que le row ID 9 n'a plus de ligne.

db > SELECT * FROM Client;
1 => [Integer(2), Text("Martin"), Text("Marie"), Text("F"), Text("New York")]
3 => [Integer(4), Text("Dubois"), Text("Sophie"), Text("F"), Text("Beyrouth")]
5 => [Integer(6), Text("Yamamoto"), Text("Sakura (さくら)"), Text("F"), Text("Paris")]
6 => [Integer(7), Text("Smith"), Text("Emily"), Text("F"), Text("Osaka")]

Et bien évidemment tout ces requêtes sont débuggables.

db > EXPLAIN DELETE FROM Client WHERE genre = 'M';
Full scan Client
Filter: genre = 'M'

db > EXPLAIN DELETE FROM Client WHERE id = 9;
Scan index PRIMARY : id = 9 for table Client
Filter: id = 9

db > EXPLAIN DELETE FROM Client WHERE ville = "Paris";  
Scan index idx_ville : ville = 'Paris' for table Client
Filter: ville = 'Paris'

Nous avons réutiliser les index pour supprimer des lignes.

Notre base de données commence à avoir une certaine tenue. ^^

Conclusion

En réutilisant le plan logique de la partie 17 et en nous appuyant sur le travail sur les index de la partie 18, nous avons pu efficacement supprimer nos enregistrements de la base de données.

En réutilisant le principe de la null table, nous avons réalisé un raccourci sur les données de chaque page, évitant ainsi de désérialiser des enregistrements avant de nous rendre compte que ceux-ci sont supprimés.

Petit à petit, notre base de données utilise ses propres primitives pour construire des comportements plus complexes.

Dans la prochaine partie nous allons définir le processus permettant de nettoyer nos pages et nos index.

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.