Partie 15: Index secondaires

Bonjour à toutes et tous 😃

Nous avons commencé à indexer de la données via sa clef primaire, cela permet de nous passer du scan et de la désérialisation des tuples des pages. Cette indexation améliore grandement la vitesse de récupération des entrées.

Le seul petit problème c'est que une fois que l'on a défini une clef primaire, il n'est plus possible d'indexer quoique ce soit d'autre.

Enfin, pour le moment. 😅

Nous allons rajouter une commande qui aura pour rôle de crér un nouveau type d'index que nous allons désigner comme secondaires.

Il y a pas mal de boulot encore une fois, mais comme d'habitude on va tout découper et gérer les choses dans l'ordre.

A l'attaque !

Clef primaire composite

Cela semble étrange de re-parler de clef primaire, mais nous allons en avoir besoin pour quelque chose de très spécial que je ne vais pas vous spoilé, tellement innatendu que ça m'a moi-même amusé. 😍

Aujourd'hui, une clef primaire est définie de manière unique par schéma.

CREATE TABLE Users (
    id INTEGER NOT NULL PRIMARY KEY,  // 👈 ici
    nom TEXT(50) NOT NULL, 
    âge INTEGER
);

Mais imaginons que nous n'ayons pas d'id.

CREATE TABLE Identité (
    nom TEXT(50) NOT NULL, 
    prénom TEXT(50) NOT NULL, 
    âge INTEGER
);

Sauf qu'il faut une clef primaire. On peut choisir de mettre le "nom" comme index.

CREATE TABLE Identité (
    nom TEXT(50) NOT NULL PRIMARY KEY, 
    prénom TEXT(50) NOT NULL, 
    âge INTEGER
);

Mais le problème, c'est qu'il y a pas mal de Martin en france ^^'

Jean Martin
Marie Martin
Pierre Martin
Claire Martin
Jacques Martin
Sophie Martin
Paul Martin
Anne Martin
Louis Martin
Camille Martin

On peut facilement avoir des doublons. On peut minimiser cela en créant un couple (nom, prénom) comme clef primaire.

Nous allons introduire une nouvelle notation.

CREATE TABLE Identité (
    nom TEXT(50) NOT NULL, 
    prénom TEXT(50) NOT NULL, 
    âge INTEGER,
    PRIMARY KEY(nom, prénom) // 👈 ici
);

Celle-ci impose de gérer une variation dans la définition du schéma. Jusqu'à présent notre schéma était un emsemble de ColumnDefinition séparé par des virgules.

Nous devons lui rajouter une PrimaryKeyDefinition qui aura pour rôle de parser le groupe qui possédera la définition de notre clef primaire composite.

PrimaryKeyDefinition

Nous allons définir notre élément comme un tableau de colonnes.

struct PrimaryKeyDefinition {
    fields: Vec<String>,
}

Je vous fourni le code commenté du parser

impl<'a> Visitable<'a, u8> for PrimaryKeyDefinition {
    fn accept(scanner: &mut Scanner<'a, u8>) -> crate::parser::Result<Self> {
        // on nettoie de potentiels blancs
        scanner.visit::<OptionalWhitespaces>()?;

        // on reconnait le token PRIMARY
        recognize(Token::Primary, scanner)?;
        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;
        // on reconnait le token KEY
        recognize(Token::Key, scanner)?;

        // on nettoie de potentiels blancs
        scanner.visit::<OptionalWhitespaces>()?;

        // visite des colonnes
        let columns = scanner.visit::<Columns>()?;

        // on nettoie de potentiels blancs
        scanner.visit::<OptionalWhitespaces>()?;

        Ok(PrimaryKeyDefinition { fields: columns.0 })
    }
}

On peut alors reconnaître notre groupe.

#[test]
fn test_parse_primary_key() {
    let data = b"PRIMARY KEY (id)";
    let mut tokenizer = Tokenizer::new(data);
    let primary_key = tokenizer.visit::<PrimaryKeyDefinition>();
    assert_eq!(
        primary_key,
        Ok(PrimaryKeyDefinition {
            fields: vec!["id".to_string()],
        })
    );

    let data = b"PRIMARY KEY (id, name)";
    let mut tokenizer = Tokenizer::new(data);
    let primary_key = tokenizer.visit::<PrimaryKeyDefinition>();
    assert_eq!(
        primary_key,
        Ok(PrimaryKeyDefinition {
            fields: vec!["id".to_string(), "name".to_string()],
        })
    );
}

Schéma à clef primaire composite

Maintenant que nous sommes capables de reconnaître notre PrimaryKeyDefinition, nous pouvons le réintégrer au sein de notre schéma.

Pour cela on défini un ColumnDefinitionResult, qui va nous permettre de réaliser l'union de reconnaissance entre ColumnDefinition et PrimaryKeyDefinition

enum ColumnDefinitionResult {
    Column(ColumnDefinition),
    PrimaryKey(PrimaryKeyDefinition),
}

impl<'a> Visitable<'a, u8> for ColumnDefinitionResult {
    fn accept(scanner: &mut Scanner<'a, u8>) -> crate::parser::Result<Self> {
        Acceptor::new(scanner)
            .try_or(|column| Ok(ColumnDefinitionResult::Column(column)))?
            .try_or(|primary_key| Ok(ColumnDefinitionResult::PrimaryKey(primary_key)))?
            .finish()
            .ok_or(ParseError::UnexpectedToken)
    }
}

On peut alors modifier le parser de Schema pour prendre en compte cette union.

impl<'a> Schema {
    fn parse(scanner: &mut Scanner<'a, u8>) -> crate::parser::Result<Self> {
        
        // snip ...

        // que l'on visite comme une liste de définition de colonne séparé par des virgules
        let columns_definitions = fields_group_tokenizer
            // ⚠️ c'est  désormais un ColumnDefinitionResult
            .visit::<SeparatedList<ColumnDefinitionResult, SeparatorComma>>()?; 

        // on récupère la primary key
        let primary_key = parse_primary_key(&columns_definitions.data)?;

        // que l'on transforme en notre map de définition de champs
        let columns =
            columns_definitions
                .into_iter()
                .fold(Vec::new(), |mut fields, column_definition| {
                    // on fitre sur les définition de colonnes, la primary key
                    // ne nous intéresse pas
                    if let ColumnDefinitionResult::Column(column_definition) = column_definition {
                        fields.push((
                            column_definition.name,
                            crate::data::ColumnDefinition::new(
                                column_definition.field,
                                // avec les contraintes associées
                                column_definition.constraints,
                            ),
                        ));
                    }
                    fields
                });

        // on n'oublie pas de déplacer le curseur du scanner externe de la taille du groupe
        scanner.bump_by(fields_group.data.len());

        Ok(Schema::new(columns, primary_key))
    }
}

La méthode parse_primary_key prend également quelque changements

fn parse_primary_key(fields: &Vec<ColumnDefinitionResult>) -> crate::parser::Result<Vec<String>> {
    let mut primary_keys = Vec::new();
    for definition in fields {
        match definition {
            // si c'est une colonne
            ColumnDefinitionResult::Column(column_definition) => {
                // récupérer les contrainte
                if !column_definition
                    .constraints
                    .iter()
                    .filter(|constraint| constraint == &&Constraint::PrimaryKey)
                    .collect::<Vec<_>>()
                    .is_empty()
                {
                    // et accumuler la colonne aux résultat si c'est
                    // une primary key
                    primary_keys.push(vec![column_definition.name.clone()]);
                }
            }
            // si le groupe est une primary key, l'accumuler au résultats
            ColumnDefinitionResult::PrimaryKey(primary_key_definition) => {
                primary_keys.push(primary_key_definition.fields.clone());
            }
        }
    }

    // il faut au moins une primary key par schéma
    if primary_keys.is_empty() {
        return Err(ParseError::PrimaryKeyNotExists);
    }

    // la primary key ne peut être qu'unique
    if primary_keys.len() > 1 {
        return Err(ParseError::NonUniquePrimaryKey(primary_keys));
    }

    Ok(primary_keys.remove(0))
}

Notre clef primaire peut désormais prendre en compte deux champs.

#[test]
fn test_primary_key_composite() {
    let data = b"(id INTEGER NOT NULL, name TEXT(50) NOT NULL, data INTEGER, PRIMARY KEY (id, name))";
    let mut tokenizer = Tokenizer::new(data);
    let schema = tokenizer.visit::<Schema>();
    assert_eq!(
        schema,
        Ok(Schema::new(
            vec![
                (
                    "id".to_string(),
                    ColumnDefinition::new(ColumnType::Integer, vec![Constraint::NotNull,])
                ),
                (
                    "name".to_string(),
                    ColumnDefinition::new(ColumnType::Text(50), vec![Constraint::NotNull,])
                ),
                (
                    "data".to_string(),
                    ColumnDefinition::new(ColumnType::Integer, vec![])
                ),
            ],
            vec!["id".to_string(), "name".to_string()]
        ))
    )
}

Fin du crochet sur les clefs primaires composites. 😎

Les index dans une base de données

Vous allez voir c'est très semblables dans sa construction aux clefs primaires composites.

Il s'agit à nouveau de "marquer" un groupe de colonne pour qu'il ait un sens particulier.

On pourrait également utiliser le schéma pour le faire, mais la création d'index peut survenir après la création de la table et le début de l'insertion de tuples.

Cela obligerait à recréer une nouvelle table avec ces index. Impliquant de copier tous les tuples de la table 1 vers la table 2.

Ce n'est pas pratique ni efficace.

A la place nous allons créer une nouvelle commande qui aura pour rôle d'ajouter un ou plusieurs index sur une table déjà créer.

Qu'est-ce qu'un index?

Un index est un raccourci sur de la donnée.

On associe un tuple à un row_id.

Prenons la table défini par

CREATE TABLE Identité (
    nom TEXT(50) NOT NULL, 
    prénom TEXT(50) NOT NULL, 
    ville TEXT(50),
    PRIMARY KEY(nom, prénom)
);

Avec les données suivantes

row_id | nom    | prénom  | ville      
-------|--------|---------|-----------
1      | Martin | Jean    | Paris      
2      | Martin | Marie   | Lyon       
3      | Martin | Pierre  | Marseille  
4      | Martin | Claire  | Bordeaux   
5      | Martin | Jacques | Nantes     
6      | Martin | Sophie  | Toulouse   
7      | Martin | Paul    | Lille      
8      | Martin | Anne    | Strasbourg 
9      | Martin | Louis   | Nice       
10     | Martin | Camille | Rennes     

Le row_id est invisible à l'utilisateur.

Lorsque l'on a créé nos index de clef primaire. On a simuler l'index au travers d'une map.

BTreeMap<Vec<Value>, usize>

Cet table produit alors cet index primaire.

(Martin, Jean)     =>  1
(Martin, Marie)    =>  2
(Martin, Pierre)   =>  3
(Martin, Claire)   =>  4
(Martin, Jacques)  =>  5
(Martin, Sophie)   =>  6
(Martin, Paul)     =>  7
(Martin, Anne)     =>  8
(Martin, Louis)    =>  9
(Martin, Camille)  =>  10

Mais on peut également avoir cette typologie.

CREATE TABLE Identité (
    id INTEGER NOT NULL PRIMARY KEY,
    nom TEXT(50), 
    prénom TEXT(50), 
    ville TEXT(50),
);

Avec les données.

row_id | id     | nom    | prénom  | ville      
-------|--------|--------|---------|-----------
1      | 1      | Martin | Jean    | Paris      
2      | 2      | Martin | Marie   | Lyon       
3      | 3      | Martin | Pierre  | Nantes  
4      | 4      | Martin | Claire  | Lyon   
5      | 5      | Martin | Jacques | Nantes     
6      | 6      | Martin | Sophie  | Paris   
7      | 7      | Martin | Paul    | Lille      
8      | 8      | Martin | Anne    | Nantes 
9      | 9      | Martin | Louis   | Nice       
10     | 10     | Martin | Camille | Paris     

Ce qui provoque l'index suivante.

(1,)   =>  1
(2,)   =>  2
(3,)   =>  3
(4,)   =>  4
(5,)   =>  5
(6,)   =>  6
(7,)   =>  7
(8,)   =>  8
(9,)   =>  9
(10,)  =>  10

Nous allons définir deux types d'index différents:

• unique : la table ne peut avoir qu'une valeur identique
• non unique : la table peut exister à de multiples occasions

Un index unique serait (nom, prénom).

(Martin, Jean)     =>  1
(Martin, Marie)    =>  2
(Martin, Pierre)   =>  3
(Martin, Claire)   =>  4
(Martin, Jacques)  =>  5
(Martin, Sophie)   =>  6
(Martin, Paul)     =>  7
(Martin, Anne)     =>  8
(Martin, Louis)    =>  9
(Martin, Camille)  =>  10

Et un index non-unique (ville,).

(Paris,)   => [ 1, 6, 10 ]
(Lyon,)    => [ 2, 4 ]
(Nantes,)  => [ 3, 5, 8 ]
(Lille,)   => [ 7 ]
(Nice,)    => [ 9 ]

Commande CREATE INDEX

On introduit deux nouvelles commandes.

Une pour les index unique.

CREATE UNIQUE INDEX idx_ville ON Identité (nom, prénom);

Le idx_ville est le nom de cet index, il se doit d'être unique par table.

On peut également créé sur cette table un index non-unique.

CREATE INDEX idx_ville ON Identité (ville);

Parser CREATE INDEX

Il nous faut 3 nouveaux tokens

enum Token {
    // snip ...
    /// INDEX token
    Index,
    /// UNIQUE token
    Unique,
    /// ON token
    On,
}

On défini une énumération sur le type d'index: "unique" ou "non-unique".

enum Uniqueness {
    Unique,
    NonUnique,
}

On matérialise la commande.

struct CreateIndexCommand {
    // nom de l'index
    pub name: String,
    // table associé à l'index
    pub table: String,
    // colonnes associé à l'index
    pub columns: Vec<String>,
    pub uniqueness: Uniqueness,
}

Que l'on peut visiter

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

        // reconnaissance du token UNIQUE
        let uniqueness = Token::Unique
            .recognize(scanner)?
            .map(|_| Uniqueness::Unique)
            .unwrap_or(Uniqueness::NonUnique);

        // on nettoie de potentiel blanc
        scanner.visit::<OptionalWhitespaces>()?;

        // on reconnait le token Index
        recognize(Token::Index, scanner)?;

        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;

        // nom de l'index
        let index_name_tokens = forecast(UntilToken(Token::Whitespace), scanner)?;
        let index_name =
            String::from_utf8(index_name_tokens.data.to_vec()).map_err(ParseError::Utf8Error)?;
        scanner.bump_by(index_name_tokens.data.len());

        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;

        // ON token
        recognize(Token::On, scanner)?;

        // on reconnait au moins 1 blanc
        scanner.visit::<Whitespaces>()?;

        // table name
        let table_name_tokens = Forecaster::new(scanner)
            .add_forecastable(UntilToken(Token::Whitespace))
            .add_forecastable(UntilToken(Token::OpenParen))
            .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 de potentiel blanc
        scanner.visit::<OptionalWhitespaces>()?;

        // columns
        let columns_tokens = forecast(GroupKind::Parenthesis, scanner)?;
        let mut columns_tokenizer =
            Scanner::new(&columns_tokens.data[1..columns_tokens.data.len() - 1]);
        // on visite une liste de colonnes
        let columns_list = columns_tokenizer.visit::<SeparatedList<Column, SeparatorComma>>()?;
        let columns = columns_list
            .into_iter()
            .map(|x| x.0)
            .collect::<Vec<String>>();
        scanner.bump_by(columns_tokens.data.len());

        // on nettoie de potentiel blanc
        scanner.visit::<OptionalWhitespaces>()?;

        // doit terminer par un point-virgule
        recognize(Token::Semicolon, scanner)?;

        Ok(CreateIndexCommand {
            name: index_name,
            table: table_name,
            columns,
            uniqueness,
        })
    }
}

Ce qui permet de parser ces différentes commandes

#[test]
fn parse_create_index_non_unique_multiple_columns() {
    let data = b"CREATE UNIQUE INDEX idx_identity ON Users (name, lastname);";
    let mut tokenizer = Scanner::new(data);
    let command = CreateIndexCommand::accept(&mut tokenizer);
    assert_eq!(
        command,
        Ok(CreateIndexCommand {
            name: "idx_identity".to_string(),
            table: "Users".to_string(),
            columns: vec!["name".to_string(), "lastname".to_string()],
            uniqueness: Uniqueness::Unique
        })
    );
}

#[test]
fn parse_create_index_unique() {
    let data = b"CREATE INDEX idx_name ON Users (name);";
    let mut tokenizer = Scanner::new(data);
    let command = CreateIndexCommand::accept(&mut tokenizer);
    assert_eq!(
        command,
        Ok(CreateIndexCommand {
            name: "idx_name".to_string(),
            table: "Users".to_string(),
            columns: vec!["name".to_string()],
            uniqueness: Uniqueness::NonUnique
        })
    );
}

Index de table

Nous allons généraliser les index primaires et secondaire au sein d'un même container Index.

IndexError

On défini une erreur spécialisé dans la gestion des index.

enum IndexError {
    DuplicateIndexValue {
        index_name: String,
        table_name: String,
        values: Vec<Value>,
    },
    DuplicateIndex {
        index_name: String,
        table_name: String,
    },
    ColumnNotExist {
        column: String,
        table_name: String,
        index_name: String,
    },
}

Indexable

Comme nous voulons gérer des index uniques et non uniques. Nous devons trouver un moyen de réaliser son union.

Généralement on passerait par une énumération, mais on va essayer une nouvelle manière au travers des traits.

pub trait Indexable: DerefMut<Target = BTreeMap<Vec<Value>, Self::Value>> {
    type Value;
    /// Push a new value to the index
    fn push(&mut self, row_id: usize, value: Vec<Value>);
    /// Check value is compatible to the index
    fn check(&self, columns: &[String], row: &HashMap<String, Value>) -> Result<(), IndexError>;
}

Transformer un tuple en clef d'index

Il est nécessaire de transformer le tuple de données à insérer sous la forme d'un autre tuple uniquement composé des colonnes de l'index.

/// Convert the tuple row as a tuple index
/// with an index like (lastname, firstname)
/// on a schema (lastname, firstname, city)
/// a row tuple (Doe, John, Paris)
/// the index tuple becomes (Doe, John)
fn as_index_tuple(index_columns: &[String], row: &HashMap<String, Value>) -> Vec<Value> {
    index_columns.iter().fold(vec![], |mut acc, column| {
        if let Some(v) = row.get(column) {
            acc.push(v.clone())
        }
        acc
    })
}

Index

Un index est un ensemble est une map de tuple et de T.

Ce T va permettre de matérialiser l'union des index uniques et non-uniques.

struct Index<T: Sized> {
    name: String,
    table_name: String,
    inner: BTreeMap<Vec<Value>, T>,
}

impl<T> Index<T> {
    pub fn new(name: String, table_name: String) -> Self {
        Self {
            name,
            table_name,
            inner: BTreeMap::new(),
        }
    }
}

On implémente DerefMut sur l'Index pour exposer la map.

impl<T> Deref for Index<T> {
    type Target = BTreeMap<Vec<Value>, T>;

    fn deref(&self) -> &Self::Target {
        &self.inner
    }
}

impl<T> DerefMut for Index<T> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.inner
    }
}

Index Unique et NonUnique

On matérialise alors nos index.

pub type UniqueIndex = Index<usize>;
pub type NonUniqueIndex = Index<Vec<usize>>;

Et en implémenté le trait Indexable.

Les index unique force une unicité des valeurs.

impl Indexable for UniqueIndex {
    type Value = usize;
    fn push(&mut self, row_id: usize, value: Vec<Value>) {
        println!(
            "Insert {:?} into unique index {} => row_id {row_id}",
            value, self.name
        );
        self.insert(value, row_id);
    }

    fn check(&self, columns: &[String], row: &HashMap<String, Value>) -> Result<(), IndexError> {
        let key = as_index_tuple(columns, row);
        // on vérifie que le tuple n'est pas déjà présent
        if self.contains_key(&key) {
            // sinon, on lève une exception
            return Err(IndexError::DuplicateIndexValue {
                index_name: self.name.to_string(),
                table_name: self.table_name.to_string(),
                values: key,
            });
        }

        Ok(())
    }
}

Contrairement à un index non-unique, qui permet d'accumuler autant de valeurs identiques que l'on veut.

impl Indexable for NonUniqueIndex {
    type Value = Vec<usize>;
    fn push(&mut self, row_id: usize, value: Vec<Value>) {
        println!(
            "Insert {:?} into non unique index {} => row_id {row_id}",
            value, self.name
        );
        self.entry(value).or_default().push(row_id);
    }

    fn check(&self, _columns: &[String], _row: &HashMap<String, Value>) -> Result<(), IndexError> {
        Ok(())
    }
}

IndexRegistry

Chaque table peut posséder plusieurs index à la fois unique ou non unique.

Mais il faut que le nom de l'index reste unique, il ne peut pas y avoir plus d'un index avec le même nom.

On a alors set de noms qui gère l'unicité des noms et deux map d'index: unique et non-unique.

struct IndexRegistry {
    indexes: BTreeMap<Vec<String>, NonUniqueIndex>,
    unique_indexes: BTreeMap<Vec<String>, UniqueIndex>,
    names: BTreeSet<String>,
}

On implémente l'ajout d'index dans la registry.

impl IndexRegistry {
    pub fn add_index(
        &mut self,
        columns: Vec<String>,
        name: &str,
        table_name: &str,
        unique: Uniqueness,
    ) -> Result<(), IndexError> {
        // vérification de l'unicité du nom d'index
        if self.names.contains(name) {
            return Err(IndexError::DuplicateIndex {
                index_name: name.to_string(),
                table_name: table_name.to_string(),
            });
        }

        // création et insertion de l'index
        match unique {
            Uniqueness::Unique => {
                let index = UniqueIndex::new(name.to_string(), table_name.to_string());
                self.unique_indexes.insert(columns, index);
            }
            Uniqueness::NonUnique => {
                let index = NonUniqueIndex::new(name.to_string(), table_name.to_string());
                self.indexes.insert(columns, index);
            }
        }

        // insertion de son nom
        self.names.insert(name.to_string());

        Ok(())
    }
}

Pour tous les index, on vient pousser une nouvelle entré.

fn insert_into_index<I: Indexable>(
    index: &mut BTreeMap<Vec<String>, I>,
    row: &HashMap<String, Value>,
    row_id: usize,
) -> Result<(), IndexError> {
    // pour tous les index
    for (index_key, index) in index.iter_mut() {
        // pour toutes les valeurs de la row
        let key = as_index_tuple(index_key, row);
        // on pousse le tuple dans l'index
        index.push(row_id, key)
    }
    Ok(())
}

On peut alors l'utiliser dans notre IndexRegistry

impl IndexRegistry {
    pub fn insert_into_index(
        &mut self,
        value: &HashMap<String, Value>,
        row_id: usize,
    ) -> Result<(), IndexError> {
        // sur les index unique
        insert_into_index(&mut self.unique_indexes, value, row_id)?;
        // non unique
        insert_into_index(&mut self.indexes, value, row_id)?;

        Ok(())
    }
}

Nous avons désormais un moyen de définir des index secondaire.

Création des index de table et indexation des données

Nous allons rattacher notre IndexRegistry à la Table

struct Table {
    table_name: String,
    pub schema: Schema,
    pub row_number: usize,
    // 👇 l'index primaire devient un Index
    primary_indexes: Index<usize>,
    pub pager: Pager,
    // 👇 on rajoute les index secondaires
    indexes: IndexRegistry,
}

Pour permettre l'ajout des index secondaire, on expose une nouvelle méthode au sein de la table.

impl Table {
    fn create_index(
        &mut self,
        name: String,
        columns: Vec<String>,
        uniqueness: Uniqueness,
    ) -> Result<(), CreateIndexError> {
        // avant de permettre l'ajout de l'index
        // on vérifie que la définition de l'index est conforme 
        // au schéma
        for column in columns.iter() {
            // si une colonne n'existe pas on lève une erreur
            if !self.schema.column_indexes.contains_key(column) {
                return Err(CreateIndexError::IndexError(IndexError::ColumnNotExist {
                    column: column.to_string(),
                    table_name: self.table_name.clone(),
                    index_name: name.clone(),
                }));
            }
        }

        // sinon on rajoute un index, si le nom n'est pas déjà utilisé sur la table
        self.indexes
            .add_index(columns, &name, &self.table_name, uniqueness)
            .map_err(CreateIndexError::IndexError)?;

        Ok(())
    }
}

Lors de l'insertion de nouvelle de données, nous avons désormais la possibilité de les indexer.

impl Table {
    pub fn insert(&mut self, row: HashMap<String, Value>) -> Result<(), InsertionError> {
        
        //snip ...

        // on vérifie que les données à indexer sont cohérentes
        self.indexes
            .check(&row)
            .map_err(InsertionError::IndexError)?;

        // insertion des données
        self.schema
            .serialize(&mut writer, &row)
            .map_err(InsertionError::Serialization)?;

        // insertion des index secondaires
        self.indexes
            .insert_into_index(&row, self.row_number)
            .map_err(InsertionError::IndexError)?;

        // insertion dans l'index primaire de notre tuple
        // et de l'ID courant d'insertion
        self.primary_indexes.insert(pk, self.row_number);

        // incrémentation de l'ID interne
        self.row_number += 1;

        Ok(())
    }
}

Notre table est désormais capable d'indexer de la données autre que des clefs primaires. 🤩

Table Master

Il y a plusieurs centaines de lignes, je vous avais dit que les clef primaires composites allaient servir.

C'est le moment ! 😃

Nous allons définir une table qui va nous permettre de stocker nos index.

Plus précisément la commande sql CREATE INDEX qui a conduit à la création de l'index.

CREATE TABLE db_master (
    type TEXT(10),      /* type d'entité ici "index" */
    name TEXT(50),      /* nom de l'entité, ici le nom de l'index */
    tbl_name TEXT(50),  /* nom de la table associé à l'entité */
    sql TEXT(300),      /* la commande sql de création de l'entité */
    PRIMARY KEY (type, name, tbl_name)
);

On s'assure qu'au boot de la base de données la clef "db_master" existe bien.

pub const MASTER_TABLE_NAME: &str = "db_master";

impl Database {
    pub fn run(&mut self, command: &str) -> Result<ExecuteResult, ExecutionError> {
        // si la table master n'existe pas, la créer
        if !self.tables.contains_key(MASTER_TABLE_NAME) {
            let schema = "(type TEXT(10), name TEXT(50), tbl_name TEXT(50), sql TEXT(300), PRIMARY KEY (type, name, tbl_name));";
            // récupère le schéma
            let schema = schema_from_str(schema).map_err(ExecutionError::Parse)?;
            // création de la table master
            let master_table = Table::new(schema, MASTER_TABLE_NAME.to_string());
            // insertion de la table master à la base de données
            self.tables
                .insert(MASTER_TABLE_NAME.to_string(), master_table);
        }

        parse(command)
            .map_err(ExecutionError::Command)?
            .execute(self)
    }
}

Comme la table "db_master" prend en valeur le "sql" de création de l'index, nous allons définir le Display de la commande.

impl Display for CreateIndexCommand {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self.uniqueness {
            Uniqueness::Unique => {
                write!(
                    f,
                    "CREATE UNIQUE INDEX {} ON {} ({});",
                    self.name,
                    self.table,
                    self.columns.join(", ")
                )
            }
            Uniqueness::NonUnique => {
                write!(
                    f,
                    "CREATE INDEX {} ON {} ({});",
                    self.name,
                    self.table,
                    self.columns.join(", ")
                )
            }
        }
    }
}

Implémentation de la commande CREATE INDEX

Allez! Un petit effort, c'est la dernière ligne droite! 😇

impl Execute for CreateIndexCommand {
    fn execute(self, database: &mut Database) -> Result<ExecuteResult, ExecutionError> {
        let sql = format!("{}", &self);
        let CreateIndexCommand {
            name,
            table,
            columns,
            uniqueness,
        } = self;
        
        // création de la commande d'insertion de la commande de création de l'index
        // dans la table master
        let master_insert = format!(
            "INSERT INTO {MASTER_TABLE_NAME}(type, name, tbl_name, sql) VALUES('index', '{name}', '{table}', '{sql}');"
        );
        // création de l'index
        database
            .create_index(table, name, columns, uniqueness)
            .map_err(ExecutionError::CreateIndex)
            .map(|_| ExecuteResult::Nil)?;
        // insertion de l'index dans la table master
        database.run(&master_insert)
    }
}

Que l'on propage vers la base de données.

impl Database {
    pub fn create_index(
        &mut self,
        table_name: String,
        name: String,
        fields: Vec<String>,
        uniqueness: Uniqueness,
    ) -> Result<(), CreateIndexError> {
        match self.tables.get_mut(&table_name) {
            Some(table) => table.create_index(name, fields, uniqueness),
            None => Err(CreateIndexError::TableNotExist(table_name))?,
        }
    }
}

A partir de maintenant la commande d'ajout d'index secondaire est entièrement fonctionnelle. 😍

Testons!

Créations des tables

On se créé deux tables.

CREATE TABLE identité (nom TEXT(50) NOT NULL, prénom TEXT(50) NOT NULL, data TEXT(50), PRIMARY KEY(nom, prénom));
CREATE TABLE identité2 (nom TEXT(50) NOT NULL, prénom TEXT(50) NOT NULL, data TEXT(50), PRIMARY KEY(nom, prénom));

Vérification de la cohérence des index

On essaie de définir un index sur une colonne "name" qui n'existe pas dans le schéma.

CREATE INDEX idx_identité ON identité(name);

Ce qui provoque bel et bien une erreur.

CreateIndex(IndexError(ColumnNotExist { column: "name", table_name: "identité", index_name: "idx_identité" }))

On créé deux index sur la même table.

CREATE INDEX idx_identité ON identité(nom);
CREATE INDEX idx_identité ON identité(prénom);
CREATE INDEX idx_identité ON identité2(prénom);

Le premier index est bien créé, par contre le second est refusé car le nom de l'index est le même.

A contrario le troisième index est accepté même si le nom est identique, la table associé n'est pas la même.

CreateIndex(IndexError(DuplicateIndex { index_name: "idx_identité", table_name: "identité" }))

Insertion de données

On peut alors commencé à insérer de la donnée.

INSERT INTO identité (nom, prénom) VALUES ('Dupond', 'Gilbert');
INSERT INTO identité (nom, prénom) VALUES ('Doe', 'Jhon');

On voit que le système réagit correctement et vient rajouter les entrés dans l'index idx_identité.

Insert [Text("Dupond")] into non unique index idx_identité => row_id 0
Insert [Text("Doe")] into non unique index idx_identité => row_id 1

Index unique

On peut également se créer un index unique et on insère de la donnée.

CREATE UNIQUE INDEX idx_identité_2 ON identité(nom, prénom);
INSERT INTO identité (nom, prénom) VALUES ('Doe', 'Jane');

On voit que la donnée se fait doublement indexée, à la fois sur l'index unique et non-unique.

Insert [Text("Doe"), Text("Jane")] into unique index idx_identité_2 => row_id 2
Insert [Text("Doe")] into non unique index idx_identité => row_id 2

Par contre, comme l'index est unique, il va interdire l'insertion de donnée dupliquée.

INSERT INTO identité (nom, prénom, data) VALUES ('Doe', 'Jane', 'ma super data de la mort qui tue');

On se fait bien rejeter pour duplication de données.

Insertion(IndexError(DuplicateIndexValue { index_name: "idx_identité_2", table_name: "identité", values: [Text("Doe"), Text("Jane")] }))

J'ai pris la décision arbitraire de refuser l'insertion, j'aurai pu choisir d'uniquement ignorer l'insertion ou de l'écraser.

Mais je préfère conserver l'index synchronisé à la donnée.

Scan des données

On vérifie que les données sont bien en base.

SELECT * FROM identité;

Ce qui est bien le cas, les donnée qui ont été refusées, ne sont réellement pas là!

[Text("Dupond"), Text("Gilbert"), Null]
[Text("Doe"), Text("Jhon"), Null]
[Text("Doe"), Text("Jane"), Null]

Table de méta-data

On peut alors vérifier l'existence des index dans la table master.

SELECT * FROM db_master;

Et oui nous avons bien 2 index sur la première table et 1 sur la seconde.

[Text("index"), Text("idx_identité"), Text("identité"), Text("CREATE INDEX idx_identité ON identité (nom);")]
[Text("index"), Text("idx_identité"), Text("identité2"), Text("CREATE INDEX idx_identité ON identité2 (prénom);")]
[Text("index"), Text("idx_identité_2"), Text("identité"), Text("CREATE UNIQUE INDEX idx_identité_2 ON identité (nom, prénom);")]

Conclusion

Notre pokédex de fonctionnalité se remplit petit à petit. 😎

Dans la prochaine partie nous verront comment supprimer de la données.

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.