Anubis / Anubis

  Holder       : Matthieu Herrmann
  Author       : Saunders Team
  Creation     : 2014/05/16
  Last update  : 2015/07/02 14:57 by matthieu@Embryo (inserted by vim).
  ======================================================================================================================

  
  
  $begin
  $define(article)(0)()
  $input($anubisdir/library/MAML4/basis.maml)
  $input($anubisdir/library/anubis_doc.maml)
  $input($anubisdir/library/names.maml)
  $htmloptions(justify:true)
  $define(MY_DB)(0)($$MY_DB)
  $define(Row)(0)($$Row)
  $define(HRow)(0)($$HRow)
  $define(In)(0)($$In)
  $define(Out)(0)($$Out)
  $define(Key)(0)($$Key)
  $define(att)(1)($att($nolist($1)))
  
  
  $title(Simple DataBase Managment System$nl$nl ($SDBMS)$nl$nl version 4)
  
  $center($italic(Matthieu Herrmann$nl Alain Prouté))
  
  $tableofcontents
  

  $section(Overview)
  $SDBMS4 is an evolution of $SDBMS3 which itself is an evolution of $SDBMS1.
  $SDBMS2 has been lost... $SDBMS1 uses lists, $SDBMS3 uses B-trees instead and
  $SDBMS4 adds indexes. On the other hand, $SDBMS4 drops the "on demand" loading
  features of $SDBMS1, i.e. tables are loaded as soon as you initialize them. It
  also drops the interface compatibility but if you add indexes, you have to
  refactor your code anyway$~!

  Also, $SDBMS4 is $em(table centric) instead of being $em(database centric),
  i.e. when you create/initialise a table, you get a handle on it and this
  handle is the only way you can access the table. Your table is not stored into
  a ``higher'' data structure (the former $att(DB($MY_DB)) type from $SDBMS1).
  However, several tables still form a database and are grouped together
  (written at the same location on your disk) while sharing a same $att(DBInfo)
  value.
  $end
  
  Below are the Includes you (and we) need:
transmit tools/2-4tree.anubis
transmit tools/iterators.anubis

   $begin
   
   Notice that:
   $list(
     $item Even if the API compatibility is dropped, the functions remain
           roughly the same ones. 
     $item When possible, iterators are preferred to lists because they are
           lazy: filtering, mapping, etc... only occur when the iterator is
           accessed. Hence, unless you know what you are doing, do not use
           functions/conditions with side effects! Also notice that computing
           the length of an iterator is usually a bad idea because all delayed
           computation (mapping, filtering, etc...) are executed. If you want
           the number of items and needs those items later, transform the
           iterator into a list. Also, iterators can play the role of
           $em(cursors). 
     $item Unless otherwise specified, $em(never) assume anything about the
           ordering of iterators or lists returned by any of the functions
           below$~! If you query a list of items with the list of ids 
           $att([A,B,C]), you can receive the list of rows 
           $att([(B,..),(A,..),(C,..)]).
     )
           
           
           

   $section(Creating databases and tables)
   When tables or indexes are loading, something wrong can happen. Hence, the
   following type:
   $acode( 
public type DBLoadingError:
   file_reading_problem(String path),  // When reading the file itself
   file_type_problem   (String path).  // When deserializing the datum
   )
   The function below formats an error in English.
   $acode( 
public define String to_English(DBLoadingError e) =
   if e is
   {
     file_reading_problem(path) then "Cannot read file '"+path+"'", 
     file_type_problem(path)    then "Cannot unserialize data from file '"+path+"'"
   }.
   )
   Below is the type of handles on a $em(database information). Mainly, it 
   contains the directory where tables are written.
   $acode( 
public type DBInfo:...    // An $em(opaque) type
   )
   You can get the path of a database:
   $acode( 
public define String get_path(DBInfo dbi).
   )
   Below we define the type of handlers on tables. The type of the rows of a
   table is up to you (but must be serializable). However, this database
   management system includes a mecanism for handling the possible changes in
   the definition of the types of the rows of a table, in such a way that
   updating the actual tables files on the disk is automatic. For this reason,
   the type scheme $att(DBTable) representing tables has two type parameters.
   The first one represents the current type of the rows of the table, the
   second one represents the history of all successive types of the rows of the
   table. With this mecanism, and if you respects some principles explained
   below, your program will always be able to read old tables saved in old
   formats.

   Tables are represented by the following opaque type scheme, where the
   parameter $att($Row) is the current type of the rows of the table, and
   $att($HRow) ($att(H) like $em(History)) is a type containing the history of
   all the successive formats of rows in the table. What if the type $att($Row)
   has several alternatives$~? This means that your table has several sorts of
   rows. But in this case, you should perhaps better use several tables.
   Preferably, do not put several alternative to a type of rows.
   $acode( 
public type DBTable($Row, $HRow):...    // An $em(opaque) type
   )
   The first thing you have to do when using $SDBMS4 is to define where your
   data, i.e. your set of tables, are going to reside in your filesystem. This
   is done with the following function.
   $acode(  
public define DBInfo init_db(String  data_base_directory).
   )
   Then, you can create your tables using the handle returned by $att(init_db).
   At the beginning, you will probably use the same type for $att($Row) and
   $att($HRow). 
   The $att(update) and $att(store) functions are for $em(historical conversions). At first, use
   $att(identity) from $fname(tools/basis.anubis). This database waits 10 secondes before
   writing its files and uses a 50 milliseconds polling. See $att(make_AData) in
   $fname(tools/delegate_writer.anubis) for more informations.
   $acode( 
public define Result(DBLoadingError, DBTable($Row,$HRow)) init_dbtable(
    DBInfo                the_data_base,
    String                table_name,
    $HRow -> $Row         update,
    $Row -> $HRow         store
   ).
   )
   The function below is the same as above with custom values for
   $att(time_wait) and $att(time_poll). See $att(make_AData) in
   $fname(tools/delegate_writer.anubis) for more informations.
   $acode( 
public define Result(DBLoadingError, DBTable($Row,$HRow)) init_dbtable(
    DBInfo                the_data_base,
    String                table_name,
    $HRow -> $Row         update,
    $Row -> $HRow         store,
    Int                   twait,
    Word32                tpoll
   ).
   )
   For example:
   $code($_white)($att( 
      with dbi = init_db("./db"),
      if init_dbtable(dbi, "client", identity, identity) is
      {
        error(_)      then // handle error
        ok(my_table)  then // Ok, we have a table !
        .
        .
        .
      }
   ))
   But, where is my type of rows$~? The compiler will infer it from your
   database usage and the mistery won't last long as you will add rows of a given
   type in your database. If needed, put a type annotation, e.g.
   
   $center($att((Result(DBLoadingError,DBTable(Client,Client)))init_dbtable(dbi,"client",...)))
   
   where $att(Client) and $att(HClient) have been previously defined by you.
   Notice that a datum of type $att(Client) is just an entire row in the
   $att(clients) table. Hence, $att(Client) should be a type with just one
   alternative and the components needed for a $em(client), like $att(name),
   $att(address), etc...

   Notice that your table will be saved in
   $fname(data_base_directory/table_name.0) or
   $fname(data_base_directory/table_name.0). If the file $fname(.0) is the most
   recent, the $fname(.1) is the previous one (and conversely).

   
   
   $section(Managing changes in table formats)
   Up to here the types used as instantiations of $att($Row) and $att($HRow) are
   the same one. That's okay, and you have begun to distribute your program and
   your users have created tables whose types are your types $att(Client),
   $att(Product), etc... If you simply modify the definition of (say) the type
   $att(Client), your new program will not work with the tables created by your
   users, which is obviously a catastrophe. Fortunately, there is a remedy, and
   here it is.

   Assume that you want to change the definition of the type $att(Client), because
   you need more columns in the table of clients. As an example, assume that you
   want to add a new component (column of the table), $att(Word32 age), to your
   type or rows in the table of clients, which was (say) defined like this: 
   $adcode( 
$id()type Client:
   client(String name, String address).
   )
   The first thing to do is to create the type $att(HClient). The first
   alternative of $att(HClient) should be the first version of $att(Client) and
   should never change as it represents an $em(historical) client, maybe present
   in a table on your user's disks. We recommend to name it $att(version_1), and
   to use the name $att(version_2) for the second alternative representing the
   new type of rows, etc.. For example, we can have this:
   $adcode( 
$id()type HClient:
   version_1(String name, String address),
   version_2(String name, String address, Word32 age).
   )
   $em(Important notice): When a change intervenes in the type of rows of the
   table, the alternative representing the new type of rows must be added at the
   end of the type $att(HClient) (or any other instance of $att($HRow)). This is
   because in serialized data the alternative names are lost, and replaced by
   their number in the list of alternatives. Also notice that another method
   will be required if you create more than 256 versions for the type of rows of
   a table.  
   
   Now, we need $em(conversion) functions in both directions between
   $att(Client) and $att(HClient). Presisely, we need the functions:
   $adcode($att( 
   (HClient -> Client)update // Transform an historical row into the current format
   (Client -> HClient)store  // The other way
   ))
   The function $att(update) can be defined as follows: 
   $adcode( 
$id()define Client update(HClient hc) = if hc is
   {
     version_1(name,addr)      then client(name,addr,0),
     version_2(name,addr,age)  then client(name,addr,age)
   }.
   )
   Of course, in the case of a row in the old format ($att(version_1)), we need
   a default value for the age. We have chosen $att(0) in the example, but you may
   choose anything. In any traditional data base system, when you add a column
   to a table, you need to fill this column with a default value, even if this
   default value is $tt(NULL) (not filled at all).

   The other function does not require default values in principle, because it
   is mainly a conversion between the current version and the latest one (i.e.
   essentially the same one):
   $adcode( 
$id()define HClient store(Client c) = 
   if c is client(name,addr,age) then version_2(name,addr,age).
   )
   The type used for storing the data on the disk is always $att(HClient) (hence
   the name of the function $att(store)), and according to the above function
   they are stored as $att(version_2(...)). The type $att(Client) is used only
   internally. This is required for being able of handling file containing
   different versions of the tables.


   $section(Consulting the database)
   The system generates an identifier for each newly created row in a table. It
   is warranted that not two rows in the same table can have the same row
   identifier. Row identifiers are of type $att(DBRowId($Row)) (an $em(opaque)
   type scheme).
   $acode( 
public type DBRowId($Row):...
   )
   The generic function below allows to compare two IDs of the same type. Of
   course, IDs must come from the same table or this is meaningless.
   $acode( 
public define Compare compare(DBRowId($Row) l, DBRowId($Row) r).
   )
   A datum of type $att(DBRow($Row)) contains the identifier of the row, a
   version number and the row itself. The version number is $att(0) when the row
   is created (by $att(add_row) or $att(add_rows)) and is incremented by $att(1)
   each time the row is updated. This may be used to detect the fact that the
   row has been modified between two operations.
   $acode( 
public type DBRow($Row):
  dbrow(  DBRowId($Row) identifier,
          Int           version,
          $Row          row). 
   ) 

   
   
   $section(Iterating on a whole table)
   Given a table, you can browse it by $em(couple) (row identifier, row), by row
   identifiers or by rows. Again, $em(do not assume any ordering).
   $acode( 
public define Iterator( (DBRowId($Row), DBRow($Row)) )  take_id_rows
   ( DBTable($Row, $HRow)  table).
public define Iterator( DBRowId($Row) )                 take_ids
   (     DBTable($Row, $HRow)  table).
public define Iterator( DBRow($Row) )                   take_rows
   (    DBTable($Row, $HRow)  table).
   )
   
   
   
   
   $section(Getting rows from a table)
   You can get a row by its id. The function below returns $att(failure) if the
   row is not found. 
   $acode( 
public define Maybe(DBRow($Row)) get_row(
    DBTable($Row, $HRow)  table,
    DBRowId($Row)         id
   ).
   )
   You can query several rows at once, and you get an iterator on several rows,
   by ids. Notice that you can get less rows than the number of given ids if the
   table does not contains some ids.
   $acode( 
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    Iterator(DBRowId($Row)) ids
   ).
   )
   The function below is the same as above, but you give a list and you get a
   list. 
   $acode( 
public define List(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    List(DBRowId($Row))     ids
   ).
   )
   You can also get rows by specifying a condition that applies to a complete
   row... 
   $acode( 
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
   ).
   )
   or by a condition on the data only:
   $acode( 
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
   ).
   )
   
   
   
   
   $section(Counting rows)
   You can get the number of rows in a table that fulfil a condition applying
   to a complete row. 
   $acode( 
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
   ).
   )
   The same thing but the condition applies to the data in the row only. 
   $acode( 
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
   ).
   )
   You can also test if there is $em(strictly) more rows fulfilling a condition
   then a given number. The function returns as soon as possible.
   $acode( 
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which,
    Int                   n
   ).
   )
   The same as above, testing only the data.
   $acode( 
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which,
    Int                   n
   ).
   )
   
   

   $section(Manipulating the tables in the database)
   While a table is manipulated, associated indexes are updated accordingly. See
   indexes below for more informations. 

   $subsection(Adding rows)
   Adding a single row to a table.
   $acode( 
public define DBRowId($Row) add_row(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    $Row                  new_row
   ).
   )
   Adding several rows to a table.
   $acode( 
public define Iterator(DBRowId($Row)) add_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    Iterator($Row)        new_rows
   ).
   )
   The same one, but the rows are given in a list. 
   $acode( 
public define Iterator(DBRowId($Row)) add_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    List($Row)            new_rows
   ).
   )
   
   
   
   $subsection(Deleting rows)
   $subsubsection(By identifier)
   Delete a row giving its id. Update the indexes (delete, set version, save).
   $acode( 
public define Maybe($Row) delete_row(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    DBRowId($Row)         id
   ).
   )
   Delete several rows giving their ids. Update the indexes (delete, set version, save).
   $acode( 
public define Iterator($Row) delete_rows(
    DBInfo                  dbi,
    DBTable($Row, $HRow)    table,
    Iterator(DBRowId($Row)) ids
   ).
   )
   Delete several rows, shorthand for list, update the indexes (delete, set version, save).
   $acode( 
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    List(DBRowId($Row))   ids
   ).
   )
   
   
   
   $subsubsection(By condition)
   Delete rows by condition on the complete row, update the indexes (delete, set version, save).
   $acode( 
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
   ).
   )
   Delete rows by condition on the data only, update the indexes (delete, set version, save).
   $acode( 
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
   ).
   )
   
   
   $subsection(Updating rows)
   Updating rows in a table is generally the consequence of a previous reading
   of these rows. Indeed, interactive programs will first show the row to be
   updated to a human user (phase 1). The user is supposed to modify the data by
   hand. When the data are modified, they may be put in the table in order to
   replace the previous values (phase 2).

   This makes a problem because, the same data may have been modified by another
   human user (or in some automatic way) in the meantime (i.e. between phase 1
   and phase 2). Hence, phase 2 should perhaps not be performed if the data have
   been modified in the meantime. For this reason, when you want to update one
   or several rows in a table, you do not provide the new values, but a function
   (denoted $att(how) below) of type:

   $center($att(DBRow($Row) -> Maybe($Row)))

   This function is supposed to remember the version number of the row got
   during phase 1. It receives as its argument the new row as it stands in the
   database when phase 2 begins. It should compare the version number of this
   new row with the one it remembers and decide if the row must be updated or
   not. If the row must not be updated, the function $att(how) must return
   $att(failure). If on the contrary, the row must be updated, it must return
   the value $att(success(r)), where $att(r) is the new value of the row.

   Now, if no error occurs, the function $att(update_row) returns a datum of
   type: 
   $acode( 
public type DBUpdateResult($Row):
   not_updated   (DBRow($Row) the_row),
   updated       (DBRow($Row) the_row).
   )
   i.e. the result is either $att(not_updated(r)) or $att(updated(r)). It is
   $att(not_updated(r)) when the row has not been updated, i.e. when the
   function $att(how) has returned $att(failure), and it is $att(updated(r)) if
   the row has been updated. In both cases, $att(r) is the new current value of
   the row as recorded in the table.

   Again, you may just want to update the row(s) the id(s) of which you have at
   hand, or update all rows satisfying some condition.

   $subsubsection(By identifier)
   Update one row, update the indexes (update, set version, save).
   $acode( 
public define Maybe(DBUpdateResult($Row)) update_row(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    DBRowId($Row)               id,
    DBRow($Row) -> Maybe($Row)  how
   ).
   )
   Update several rows, update the indexes (update, set version, save).
   $acode( 
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    Iterator(DBRowId($Row))     ids,
    DBRow($Row) -> Maybe($Row)  how
   ).
   )
   Update several rowss, shorthand for list, update the indexes (update, set version, save).
   $acode( 
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    List(DBRowId($Row))         ids,
    DBRow($Row) -> Maybe($Row)  how
   ).
   )
   
   
   $subsubsection(By condition)
   Update rows by condition on the complete row, update the indexes (update, set version, save).
   $acode( 
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    DBRow($Row) -> Bool         which,
    DBRow($Row) -> Maybe($Row)  how
   ).
   )
   Update rows by condition on the data only, update the indexes (update, set version, save).
   $acode( 
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    $Row -> Bool                which,
    DBRow($Row) -> Maybe($Row)  how
   ).
   )
   
   $end
   
   $subsection(Multi-tables queries)
   See sdbms1.anubis. Yes, I'm lazy !
   $acode( 
public define List($Out) query(
    DBTable($Row,$HRow)         the_first_table,
    DBRow($Row) -> Bool         the_first_condition,
    (DBRow($Row),$In) -> $Out   the_first_selection,
    DBRow($Row) -> List($In)    the_subquery
   ).
   )
   
   
   $begin
   $section(Adding indexes)
   Below is the public opaque type of indexes: this is in the hand of the user,
   only to be queried by appropriate functions. 
   $acode( 
public type DBIndex($Key, $Row):...
   )
   
   
   
   $subsection(Initialisation)
   Indexes are $em(plugged) on a table and sort rows with a given custom
   function. This function is both a filter and an indexation function, i.e. not
   all rows may be registered in the index. You also need to provide an ordering
   on keys. This index waits 10 secondes before writing its files and uses a 50
   milliseconds polling. See $att(make_AData) in
   $fname(tools/delegate_writer.anubis) for more informations.
   $acode(  
public define Result(DBLoadingError, DBIndex($Key, $Row)) init_index(
    DBInfo                            the_data_base,  // The database and the table on which
    DBTable($Row,$HRow)               the_table,      // the index must be installed.
    String                            index_name,     // Name of the index
    DBRow($Row) -> Maybe($Key)        filter_add,     // Adding filter function
    ($Key, $Key) -> Compare           cmp             // Comparison function for the index keys
   ).
   )
   Same as above with custom values for $att(time_wait) and $att(time_poll).
   See $att(make_AData) in $fname(tools/delegate_writer.anubis) for more
   informations. 
   $acode( 
public define Result(DBLoadingError, DBIndex($Key, $Row)) init_index(
    DBInfo                            the_data_base,  // The database and the table on which
    DBTable($Row,$HRow)               the_table,      // the index must be installed.
    String                            index_name,     // Name of the index
    DBRow($Row) -> Maybe($Key)        filter_add,     // Adding filter function
    ($Key, $Key) -> Compare           cmp,            // Comparison function for the index keys
    Int                               twait,          // Waiting time for the delegate writer
    Word32                            tpoll           // polling time for the delegate writer
   ).
   )
   Those indexes are like the one you find in books: given a key, you get a
   $em(list of identifiers) matching that key. Also, keys are ordered. Thus, you
   can use those index in several ways: as an index, a filter, a sorting
   function...

   
   
   $subsection(Consultation)
   Iterate in order (lower to upper) over all the keys, obtaining the each time
   the list of associated identifiers. 
   $acode( 
public define Iterator( ($Key, List(DBRowId($Row))) ) take_key_ids
   (DBIndex($Key, $Row) my_idx).
   )
   Iterate in order (lower to upper) over all the keys, obtaining the each time
   the list of associated rows. 
   $acode( 
public define Iterator( ($Key, List(DBRow($Row))) ) take_key_rows
   (DBIndex($Key, $Row) my_idx).
   )
   Iterate in order (lower to upper) over all the keys, obtaining the each time
   the list of associated identifiers. 
   $acode( 
public define Iterator( $Key ) take_key(DBIndex($Key, $Row) my_idx).
   )
   Iterate in order of the keys (lower to upper) over all the identifiers.
   $acode( 
public define Iterator( DBRowId($Row) ) take_ids(DBIndex($Key, $Row) my_idx).
   )
   Iterate in order of the keys (lower to upper) over all the rows.
   $acode(  
public define Iterator( DBRow($Row) ) take_rows(DBIndex($Key, $Row) my_idx).
   )
   Get the list of identifiers associated to a key.
   $acode( 
public define Iterator(DBRowId($Row)) get_ids($Key key, DBIndex($Key, $Row) my_idx).
   )
   Get the list of rows associated to a key.
   $acode( 
public define Iterator(DBRow($Row)) get_rows($Key key, DBIndex($Key, $Row) my_idx).
   )
   
   
   
   
   $subsection(Example)
   Here is a simple example, adding string read from the command line into a
   database and indexing them by length. Note: for maximum interactivity, this
   example sets the waiting time and the poll time to 0. Set those according to
   your needs. 
   $adcode($nolist( 
$id()read tools/basis.anubis
$id()read tools/iterators.anubis
$id()read tools/2-4tree.anubis
$id()read data_base/sdbms4.anubis

$id()define Compare cmp(Int i, Int j) = 
   if i < j then before else if i = j then same else after.

$id()module One test_sdbms4(List(String) args) =
    with dbi    = init_db("./db"),
    if (Result(DBLoadingError, DBTable(String, String))) init_dbtable(dbi, "test_table", identity, identity, 0, 0) is
    {
      error(_)    then unique
      ok(table1)  then
      if init_index(dbi, table1, "idx_version", (DBRow(String) r)|-> success(length(row(r))), cmp, 0, 0) is
      {
        error(_) then unique
        ok(idx1) then

        println("\n==== Store unique items...");
        map_forget( (String s)|->
          if get_rows(table1, (String t)|-> t = s) is
          {
            nil     then forget(add_row(dbi, table1, s))
            h .. t  then forget(update_row(dbi, table1, identifier(h), (DBRow(String) r)|-> success(row(r))))
          },
          args
        );
        println(to_string(take_rows(table1), (DBRow(String) r)|-> row(r)+"  (version "+version(r)+")","\n"));

        println("\n==== Delete version > 10\n");
        forget(delete_rows(dbi, table1, (DBRow(String) r)|-> version(r) > 10));
        println(to_string(take_rows(table1), (DBRow(String) r) |-> row(r)+"  (version "+version(r)+")","\n"));

        println("\n==== Check keys of idx \n");
        println(to_string(take_key(idx1), (Int i) |-> ""+i, "\n"));

        println("\n==== Check content of idx \n");
        println(to_string(take_rows(idx1), (DBRow(String) r) |-> row(r)+"  (version "+version(r)+")","\n"));

        println("\n==== Check content of length 4\n");
        println(to_string(get_rows(4, idx1), (DBRow(String) r) |-> row(r)+"  (version "+version(r)+")","\n"))
      }
    }.
   ))
   $par$par$par$par$par
   $end

  ======================================================================================================================
  = Private part =======================================================================================================
  ======================================================================================================================

read tools/basis.anubis               // Fold for lists
read tools/delegate_writer.anubis     // Writing data asynchronously
read tools/safe_save_retrieve.anubis  // Safe reading/writing with checksum

  *** [1] Type definitions
      ==================================================================================================================

public type DBInfo:
  dbinfo(String directory).

public define String get_path(DBInfo dbi) = directory(dbi).

public type DBRowId($Row):
  dbrowid(Int value )
  .

    *** [1.1] Index, on disk and in memory
        ================================================================================================================

    Type of index data when saved on the disk
type IDXDataSave($Key, $Row):
  idx_data_save(  Int                                         version,
                  TreeSerializeKV($Key, List(DBRowId($Row)))  serialized_tree)
  .

    Type of index data when in memory
type IDXData($Key, $Row):
  idx_data( Int                                 version,
            TreeKV($Key, List(DBRowId($Row)))   data_tree,
            DBRow($Row) -> Maybe($Key)          filter_add)
  .

    An index from the a table point of view: functions with side effects
type IDXTable($Row):
  idx_table(  String                            name,
              DBInfo -> One                     save,
              Int -> One                        set_version,
              DBRow($Row) -> One                add,
              DBRow($Row) -> One                delete,
              (DBRow($Row), DBRow($Row)) -> One update)
  .

    Pre-declaration for DBIndex: allows to link between an index and its related table
type DBTableData($Row):...

    Public opaque type of indexes: this is in the hand of the user, only to be queried by appropriate functions.
public type DBIndex($Key, $Row):
  index(  String                    name,
          Var(IDXData($Key, $Row))  data,
          Var(DBTableData($Row))    table)
  .


    *** [1.2] Table, on disk and in memory
        ================================================================================================================

type DBTableDataSave($HRow):
  dbtable_data_save(  Int                                             counter,
                      Int                                             version,
                      TreeSerializeKV(DBRowId($HRow), DBRow($HRow))   serialized_tree)
  .

type DBTableData($Row):
  dbtable_data(   Int                                 counter,
                  Int                                 version,
                  TreeKV(DBRowId($Row), DBRow($Row))  data_tree,
                  List(IDXTable($Row))                indexes)
  .

type DBTable($Row, $HRow):
  dbtable(  String                  name,
            DBInfo -> One           save,
            Var(DBTableData($Row))  data)
  .



  *** [2] Base tools
      ==================================================================================================================

    We need a comparison function for the row identifiers. This function works for all identifiers types as the type
  parameter is not used.
public define Compare compare(DBRowId($Row) l, DBRowId($Row) r) =
  with li = value(l), with ri = value(r),
  if li < ri then before else if li = ri then same else after.



  *** [3] Indexes management
      ==================================================================================================================

    *** [3.1] "DB side" functions
        ================================================================================================================

      *** [3.1.1] Storer
          ==============================================================================================================

define (DBInfo -> One) create_storer(DBIndex($Key, $Row) my_idx, Int twait, Word32 tpoll) =
  // Storing an index requires conversion from the "in memory" format to the "on disk" format
  with converter  = (IDXData($Key, $Row) idxdata)|-> idx_data_save(version(idxdata), to_serialize(data_tree(idxdata))),
  // We also need an asynchronous writer (using default timings), configured for that index (using its name)
  with writing_fun = (IDXDataSave($Key, $Row) idx, String path) |-> safe_save(idx, path),
  with asaver = make_AData("",name(my_idx), writing_fun, twait, tpoll),
  // When writing, we set the directory on the fly
  (DBInfo dbi)|->
    set_prefix(directory(dbi), asaver);
    save(converter(*data(my_idx)), asaver)
  .

      *** [3.1.2] Insertion
          ==============================================================================================================

define TreeKV($Key, List(DBRowId($Row))) idx_insert($Key k, DBRowId($Row) v, TreeKV($Key, List(DBRowId($Row))) tkv) =
  update(k, tkv, (Maybe(List(DBRowId($Row))) mblist)|->
    if mblist is success(list)
    then success([v . list])  // If the entry already exists, add the identifier to the list ...
    else success([v])         // ... else create a new one
  ).

define (DBRow($Row) -> One) create_inserter(DBIndex($Key, $Row) my_idx) =
  with data_v = data(my_idx),
  // We only add items that pass the filter
  (DBRow($Row) r)|->
    with d  = *data_v,
    with fa = filter_add(d),
    if fa(r) is success(key)
    then data_v <- idx_data(0, idx_insert(key, identifier(r), data_tree(d)), fa)
    else unique
  .

      *** [3.1.3] Deletion
          ==============================================================================================================

define TreeKV($Key, List(DBRowId($Row))) idx_delete($Key k, DBRowId($Row) v, TreeKV($Key, List(DBRowId($Row))) tkv) =
  if get_remove(k, tkv) is success(tuple)
  then (
    if tuple is(rm_dic, rm_l)
    then if remove_element(rm_l, (DBRowId($Row) e)|-> e = v) is
    {
      [ ]     then rm_dic,
      [h . t] then insert(k, [h . t], rm_dic)
    }
  )
  else tkv.

define (DBRow($Row) -> One) create_deleter(DBIndex($Key, $Row) my_idx) =
  with data_v = data(my_idx),
  // We only remove items that pass the filter
  (DBRow($Row) r)|->
    with d  = *data_v,
    with fa = filter_add(d),
    if fa(r) is success(key)
    then data_v <- idx_data(0, idx_delete(key, identifier(r), data_tree(d)), fa)
    else unique
  .

      *** [3.1.4] Update
          ==============================================================================================================

define ( (DBRow($Row),DBRow($Row)) -> One) create_updater(DBIndex($Key, $Row) my_idx) =
  with data_v = data(my_idx),
  (DBRow($Row) old, DBRow($Row) new)|->
    with fa = filter_add(*data_v),
    ( if fa(old) is success(key)
      then data_v <- idx_data(0, idx_delete(key, identifier(old), data_tree(*data_v)), fa)
      else unique );
    ( if fa(new) is success(key)
      then data_v <- idx_data(0, idx_insert(key, identifier(new), data_tree(*data_v)), fa)
      else unique )
  .

      *** [3.1.5] Version setter
          ==============================================================================================================

define (Int -> One) create_version_setter(DBIndex($Key, $Row) my_idx) =
  with data_v = data(my_idx),
  (Int v)|->
    with d  = *data_v,
    data_v <- idx_data(v, data_tree(d), filter_add(d))
  .

    *** [3.2] Index loading and initialisation
        ================================================================================================================

      *** [3.2.1] Index loading and conversion
          ==============================================================================================================

    Try to read an index from a file, returning an empty one if the file is not found.
define Result(DBLoadingError, IDXDataSave($Key, $Row)) read_index(
    String  directory,
    String  name
  ) =
  with path = directory + "/" + name,
  if (RetrieveResult(IDXDataSave($Key, $Row)))safe_retrieve(path) is
  {
    cannot_find_file  then ok(idx_data_save(0, empty_serialize_kv))
    read_error        then error(file_reading_problem(path))
    type_error        then error(file_type_problem(path))
    ok(t)             then ok(t)
  }.

    Convert from the IDXDataSave format to the IDXData format, checking that the version number matches the version
  number of the target table. If it does not match, the index is recreated from the table's data.
define IDXData($Key, $Row) check_convert_rebuilt(
    IDXDataSave($Key, $Row)           save_data,   // The data to convert or rebuilt
    DBRow($Row) -> Maybe($Key)        filter_add,  // Adding filter function
    ($Key, $Key) -> Compare           cmp,         // Comparison function for the index keys
    DBTableData($Row)                 tdata        // Allow to check the version number and rebuilt the index if needed.
  ) =
  if version(tdata) = version(save_data)
  // Version numbers match: convert the index.
  then idx_data(version(save_data), from_serialize(serialized_tree(save_data), cmp), filter_add)
  // Version numbers mismatch: rebuilt the index.
  else println("Rebuilding index...");
    with rebuilt_data =
      fold_left(take_left_value(data_tree(tdata)), new_tree(cmp), (TreeKV($Key, List(DBRowId($Row))) tkv_idx , DBRow($Row) r)|->
        if filter_add(r) is success(key)
          then update(key, tkv_idx, (Maybe(List(DBRowId($Row))) mblist)|->
            if mblist is success(list)
            then success([identifier(r) . list])  // If the entry already exists, add the identifier to the list ...
            else success([identifier(r)])         // ... else create a new one
          )
        else tkv_idx // Skip: do not add anything
      ),
    idx_data(version(tdata), rebuilt_data, filter_add)
  .



      *** [3.2.2] Index initialisation
          ==============================================================================================================

public define Result(DBLoadingError, DBIndex($Key, $Row)) init_index(
    DBInfo                            the_data_base,  // The database and the table on which
    DBTable($Row,$HRow)               the_table,      // the index must be installed.
    String                            index_name,     // Name of the index
    DBRow($Row) -> Maybe($Key)        filter_add,     // Adding filter function
    ($Key, $Key) -> Compare           cmp,            // Comparison function for the index keys
    Int                               twait,          // Waiting time for the delegate writer
    Word32                            tpoll           // polling time for the delegate writer
  ) =
  // 1: try to read the index from a file or create a new empty index if the file does not exist
  if read_index(directory(the_data_base), index_name) is
  {
    error(e)        then error(e),
    ok(saved_data)  then
      // 2: Check if the loaded data match the table's version number and convert or rebuilt the index
      with tdata_v          = data(the_table),
      with tdata            = *tdata_v,
      with loaded_idx_data  = check_convert_rebuilt(saved_data, filter_add, cmp, tdata),
      // 3: Create the "user" side of the index, holding the "$Key" typed data
      with my_idx_data      = var(loaded_idx_data),                     // ** Dynamic variable created here **
      with my_idx           = index(index_name, my_idx_data, tdata_v),  // Keep a link on the table
      // 4: Create the "table" side of the index, with a set of action
      with store_fun  = create_storer(my_idx, twait, tpoll),
      with set_fun    = create_version_setter(my_idx),
      with add_fun    = create_inserter(my_idx),
      with delete_fun = create_deleter(my_idx),
      with update_fun = create_updater(my_idx),
      with idxt       = idx_table(index_name, store_fun, set_fun, add_fun, delete_fun, update_fun),
      // 5: Add the index to the table's list of indexes
      tdata_v <- dbtable_data(counter(tdata), version(tdata), data_tree(tdata), [ idxt . indexes(tdata) ]);
      // 6: Ok, everything is set up, return the index
      ok(my_idx)
  }.

public define Result(DBLoadingError, DBIndex($Key, $Row)) init_index(
    DBInfo                            the_data_base,  // The database and the table on which
    DBTable($Row,$HRow)               the_table,      // the index must be installed.
    String                            index_name,     // Name of the index
    DBRow($Row) -> Maybe($Key)        filter_add,     // Adding filter function
    ($Key, $Key) -> Compare           cmp             // Comparison function for the index keys
  ) = init_index(the_data_base, the_table, index_name, filter_add, cmp, 10000, 50).



    *** [3.3] Index consultation
        ================================================================================================================

public define Iterator( ($Key, List(DBRowId($Row))) ) take_key_ids( DBIndex($Key, $Row) my_idx) =
  take_left(data_tree(*data(my_idx))).

public define Iterator( ($Key, List(DBRow($Row))) ) take_key_rows(DBIndex($Key, $Row) my_idx) =
  with table_data = data_tree(*table(my_idx)),
  map(  ( ($Key, List(DBRowId($Row))) tuple ) |->
          if tuple is (k, l) then
          with new_l = map_select( (DBRowId($Row) id) |-> get(id, table_data), l),
          (k, new_l),
  take_key_ids(my_idx)
  ).

public define Iterator( $Key ) take_key( DBIndex($Key, $Row) my_idx) =
  take_left_key(data_tree(*data(my_idx))).

public define Iterator( DBRowId($Row) ) take_ids( DBIndex($Key, $Row) my_idx) =
  fold_right( take_left_value(data_tree(*data(my_idx))), nil,
    (List(DBRowId($Row)) l, Iterator(DBRowId($Row)) it)|-> append(take_left(l), it)
  ).

public define Iterator( DBRow($Row) ) take_rows( DBIndex($Key, $Row) my_idx) =
  with table_data = data_tree(*table(my_idx)),
  with idx_data   = data_tree(*data(my_idx)),
  filter(
    (DBRowId($Row) id) |-> get(id, table_data),
    fold_left( take_left_value(idx_data), nil,
      (Iterator(DBRowId($Row)) it, List(DBRowId($Row)) l)|->
      append(it, take_left(l))
    )
  ).

public define Iterator(DBRowId($Row)) get_ids($Key key, DBIndex($Key, $Row) my_idx) =
  if get(key, data_tree(*data(my_idx))) is
  {
    failure     then nil
    success(l)  then take_left(l)
  }.

public define Iterator(DBRow($Row)) get_rows($Key key, DBIndex($Key, $Row) my_idx) =
  with table_data = data_tree(*table(my_idx)),
  filter( (DBRowId($Row) id) |-> get(id, table_data), get_ids(key, my_idx) ).

  *** [4] Database initialisation
      ==================================================================================================================

    *** [4.1] The database itself
        ================================================================================================================

    Create the directory if it does not exist.
public define DBInfo init_db (
    String  data_base_directory
  ) =
  forget(make_directory(data_base_directory, default_directory_mode));
  dbinfo(data_base_directory).

    *** [4.2] The tables
        ================================================================================================================

    Try to read table from a file, returning an empty one if the file is not found.
define Result(DBLoadingError, DBTableDataSave($HRow)) read_table(
    String  directory,
    String  name
  ) =
  with path = directory/name,
  if (RetrieveResult(DBTableDataSave($HRow)))safe_retrieve(path) is
  {
    cannot_find_file  then ok(dbtable_data_save(0, 0, empty_serialize_kv))
    read_error        then error(file_reading_problem(path))
    type_error        then error(file_type_problem(path))
    ok(t)             then ok(t)
  }.

    Convert from "On-disk serialized $HRow" to "In-memory $Row"
define TreeKV(DBRowId($Row), DBRow($Row)) convert_read_update(
    TreeSerializeKV(DBRowId($HRow), DBRow($HRow)) tkv,
    $HRow -> $Row                                 update
  ) =
  // Convert the tree from its serialized format to its in-memory "$HRow" format, using our "compare" function for RowId.
  with tree = from_serialize(tkv, compare),
  // Map the tree to "$Row format" with the "update" function, using our "compare" function for RowId.
  map( // Map the TreeKV with the "update" function
        ( DBRowId($HRow) id ) |-> dbrowid(value(id)),
        ( DBRow($HRow)   r  ) |-> dbrow(dbrowid(value(identifier(r))), version(r), update(row(r))),
        tree, compare
      )
  .

    Convert from "In-memory $Row" to "On-disk serialized $HRow"
define TreeSerializeKV(DBRowId($HRow), DBRow($HRow)) convert_write_store(
    TreeKV(DBRowId($Row), DBRow($Row))  tkv,
    $Row -> $HRow                       store
  )=
    to_serialize(
      map( // Map the TreeKV with the "store" function
        ( DBRowId($Row) id ) |-> dbrowid(value(id)),
        ( DBRow($Row)   r  ) |-> dbrow(dbrowid(value(identifier(r))), version(r), store(row(r))),
        tkv,
        compare
      )
    ).

public define Result(DBLoadingError, DBTable($Row,$HRow)) init_dbtable(
    DBInfo                the_data_base,  // The database on which the table is installed
    String                table_name,
    $HRow -> $Row         update,
    $Row -> $HRow         store,
    Int                   twait,          // Waiting time for the delegate writer
    Word32                tpoll           // polling time for the delegate writer
  ) =
  // 1: try to read the table from a file or create a new empty one if the file is not found
  if read_table(directory(the_data_base), table_name) is
  {
    error(e)        then error(e)
    ok(saved_data)  then if saved_data is dbtable_data_save(c, v, st) then
      // 2: Convert the data from "On-disk" format to "In-memory" format
      with tdata = dbtable_data(c, v, convert_read_update(st, update), [ ]), // By default, no index
      // 3: Create the variable holding the data
      with tdata_v = var(tdata),  //  ** Dynamic variable created here **
      // 4: Create the saving function (delegate writer + conversion)
      with writing_fun  = (DBTableDataSave($HRow) idx, String path) |-> safe_save(idx, path),
      with asaver       = make_AData("", table_name, writing_fun, twait, tpoll),
      with saver        = (DBInfo dbi)|->
        set_prefix(directory(dbi), asaver);
        if *tdata_v is dbtable_data(c_, v_, st_, _) then
        save( dbtable_data_save(c_, v_, convert_write_store(st_, store)) , asaver),
      // 5: Ok, everything is set up, return the table
      ok(dbtable(table_name, saver, tdata_v))
  }.

public define Result(DBLoadingError, DBTable($Row,$HRow)) init_dbtable(
    DBInfo                the_data_base,  // The database on which the table is installed
    String                table_name,
    $HRow -> $Row         update,
    $Row -> $HRow         store
  ) = init_dbtable(the_data_base, table_name, update, store, 10000, 50).


  *** [5] Database consultation
      ==================================================================================================================

    *** [5.1] All records
        ================================================================================================================

public define Iterator( (DBRowId($Row), DBRow($Row)) ) take_id_rows(
    DBTable($Row, $HRow)  table
  ) = take_left(data_tree(*data(table))).

public define Iterator( DBRowId($Row) ) take_ids(
    DBTable($Row, $HRow)  table
  ) = take_left_key(data_tree(*data(table))).

public define Iterator( DBRow($Row) ) take_rows(
    DBTable($Row, $HRow)  table
  ) = take_left_value(data_tree(*data(table))).

    *** [5.2] By id
        ================================================================================================================

    Get a row by its id. It can fail.
public define Maybe(DBRow($Row)) get_row(
    DBTable($Row, $HRow)  table,
    DBRowId($Row)         id
  ) = get(id, data_tree(*data(table))).

    Get an iterator on several rows, by ids. Note that you can have less row than ids. If you want to check that,
  convert the iterator into a list an check its length.
  Note: **do not use "get_row"** in this definition: we are lazy so we need to "capture" the current state of our
  table. "get_row" take a fresh look in the variable each time it is called so do not do that !
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    Iterator(DBRowId($Row)) ids
  ) =
    with dt = data_tree(*data(table)),
    filter( (DBRowId($Row) id)|-> get(id, dt), ids).

public define List(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    List(DBRowId($Row))     ids
  ) =
    with dt = data_tree(*data(table)),
    map_select( (DBRowId($Row) id)|-> get(id, dt), ids).

    *** [5.3] By condition
        ================================================================================================================
    A condition is simply a (lazy) filter on all records. Consider using indexes...

    By condition on the complete row
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
  ) = filter(which, take_rows(table)).

    By condition on the data only
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
  ) = get_rows(table, (DBRow($Row) r)  |-> which(row(r))).

    *** [5.4] Couting row
        ================================================================================================================

    Test the complete row
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
  ) = length(get_rows(table, which)).

    Test only on the data
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
  ) = get_number_of_rows(table, (DBRow($Row) r)  |-> which(row(r))).

    Test the complete row
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which,
    Int                   n
  ) =
  fold_while_left(take_rows(table), 0,
    (Int accu, DBRow($Row) row) |-> if which(row) then accu+1 else accu,
    (Int z)|->z =< n
  ) > n. // In the API : 'true' if the table has strictly more than 'n' rows satisfying else 'ok(false)'

    Test only on the data
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which,
    Int                   n
  ) = has_more_rows_than(table, (DBRow($Row) r)  |-> which(row(r)), n).



  *** [6] Database manipulation
      ==================================================================================================================

    *** [6.1] Add rows
        ================================================================================================================

    Add a single row, update the indexes (add, set version, save).
public define DBRowId($Row) add_row(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    $Row                  new_row
  ) =
  if *data(table) is dbtable_data(counter, version, data_tree, indexes) then
  with dbr = dbrow(dbrowid(counter), 0, new_row),
  with id  = identifier(dbr),
  with nversion = version +1,
  map_forget((IDXTable($Row) idx)|-> add(idx)(dbr); set_version(idx)(nversion); save(idx)(dbi), indexes);
  data(table) <- dbtable_data(counter+1, nversion, insert(id, dbr, data_tree), indexes);
  save(table)(dbi);                                               // Save the table asynchronously
  id.

    Add several rows, update the indexes (add, set version, save).
public define Iterator(DBRowId($Row)) add_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    Iterator($Row)        new_rows
  ) =
  if *data(table) is dbtable_data(counter, version, data_tree, indexes) then
  if fold_left(new_rows, (data_tree, counter, nil), ((TreeKV(DBRowId($Row), DBRow($Row)), Int, Iterator(DBRowId($Row))) tuple, $Row nr)|->
      if tuple is (tr, cnt, it) then
        with dbr = dbrow(dbrowid(counter), 0, nr),
        with id  = identifier(dbr),
        map_forget((IDXTable($Row) idx)|-> add(idx)(dbr), indexes);
        (insert(id, dbr, tr), counter+1, id .. it)
  ) is (ntree, ncounter, it) then
  with nversion = version+1,
  map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes);
  data(table) <- dbtable_data(ncounter, nversion, ntree, indexes);
  save(table)(dbi);                                               // Save the table asynchronously
  it.

    Add several rows, shorthand for list, update the indexes (add, set version, save).
public define Iterator(DBRowId($Row)) add_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    List($Row)            new_rows
  ) = add_rows(dbi, table, take_left(new_rows)).

    *** [6.2] Deleting rows
        ================================================================================================================

      *** [6.2.1] By ids
          ==============================================================================================================

    Delete a row giving its id. Update the indexes (delete, set version, save).
public define Maybe($Row) delete_row(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    DBRowId($Row)         id
  ) =
  if *data(table) is dbtable_data(counter, version, data_tree, indexes) then
    if get_remove(id, data_tree) is success(s)
    then
    ( if s is (ntree, deleted) then
      with nversion = version+1,
      map_forget((IDXTable($Row) idx)|-> delete(idx)(deleted); set_version(idx)(nversion); save(idx)(dbi), indexes);
      data(table) <- dbtable_data(counter, nversion, ntree, indexes);
      save(table)(dbi);                                               // Save the table asynchronously
      success(row(deleted))
    )
    else failure.

    Delete several rows giving their ids. Update the indexes (delete, set version, save).
public define Iterator($Row) delete_rows(
    DBInfo                  dbi,
    DBTable($Row, $HRow)    table,
    Iterator(DBRowId($Row)) ids
  ) =
  if *data(table) is dbtable_data(counter, version, data_tree, indexes) then
    if fold_left(ids, (data_tree, nil), ( (TreeKV(DBRowId($Row), DBRow($Row)), Iterator($Row)) tuple, DBRowId($Row) id)|->
      if tuple is (tkv, accu) then
        if get_remove(id, tkv) is
        {
          failure     then tuple // Do nothing...
          success(s)  then if s is (tree, deleted) then
            map_forget((IDXTable($Row) idx)|-> delete(idx)(deleted), indexes); // Update the indexes.
            (tree, row(deleted) .. accu) // Return the new tuple with the updated tree and the removed row.
        }
    ) is (ntree, it) then
    with nversion = version+1,   // Update version, save the indexes.
    map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes);
    data(table) <- dbtable_data(counter, nversion, ntree, indexes);
    save(table)(dbi);                   // Save the table asynchronously
    it.

    Delete several rows, shorthand for list, update the indexes (delete, set version, save).
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    List(DBRowId($Row))   ids
  ) = delete_rows(dbi, table, take_left(ids)).

      *** [6.2.2] By condition
          ==============================================================================================================

    Delete rows by condition on the complete row, update the indexes (delete, set version, save).
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
  ) =
  if *data(table) is dbtable_data(counter, version, data_tree, indexes) then
    with values = filter(which, take_left_value(data_tree)), // Iterate and filter items to be removed on the fly.
    if fold_left(values, (data_tree, nil), ( (TreeKV(DBRowId($Row), DBRow($Row)), Iterator($Row)) tuple, DBRow($Row) r)|->
      if tuple is (tkv, accu) then
        map_forget((IDXTable($Row) idx)|-> delete(idx)(r), indexes);  // Update the indexes.
        (remove(identifier(r), tkv), row(r) .. accu)                  // Remove the item from the tree, put it in the iterator
    ) is (ntree, it) then
    with nversion = version+1,   // Update version, save the indexes.
    map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes);
    data(table) <- dbtable_data(counter, nversion, ntree, indexes);
    save(table)(dbi);                   // Save the table asynchronously
    it.

    Delete rows by condition on the data only, update the indexes (delete, set version, save).
public define Iterator($Row) delete_rows(
    DBInfo                dbi,
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
  ) = delete_rows(dbi, table, (DBRow($Row) r)  |-> which(row(r))).


    *** [6.3] Updating rows
        ================================================================================================================

      *** [6.3.1] By ids
          ==============================================================================================================

    Update one element, update the indexes (update, set version, save).
public define Maybe(DBUpdateResult($Row)) update_row(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    DBRowId($Row)               id,
    DBRow($Row) -> Maybe($Row)  how
  ) =
  if *data(table) is dbtable_data(c, v, data_tree, indexes) then
  if get_update(id, data_tree, (Maybe(DBRow($Row)) mb)|->
    if mb is success(row)
    then if how(row) is success(new_data)         // The row exists, try to update it.
      then                                        // Update ! ** WARNING: next version ** this is row's version
        with new_row  = dbrow(id, version(row)+1, new_data),
        map_forget((IDXTable($Row) idx)|-> update(idx)(row, new_row), indexes); // Update the indexes
        (success(new_row), success(updated(row)))
      else (failure, success(not_updated(row)))   // Do not update, return the row
    else (failure, failure)                       // The row does not exist, do nothing
  ) is (ntree, tuple) then if tuple is (_, res) then
  with nversion = v+1,                            // this is table's version
  map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes); // Update indexes' version and save.
  data(table) <- dbtable_data(c, nversion, ntree, indexes);
  save(table)(dbi);                               // Save the table asynchronously
  res
  .

    Update several elements, update the indexes (update, set version, save).
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    Iterator(DBRowId($Row))     ids,
    DBRow($Row) -> Maybe($Row)  how
  ) =
  if *data(table) is dbtable_data(c, v, data_tree, indexes) then
  if fold_left(ids, (data_tree, nil),
    ( (TreeKV(DBRowId($Row), DBRow($Row)), Iterator(DBUpdateResult($Row))) tuple, DBRowId($Row) id)|->
      if tuple is (tkv, accu) then
        if get_update(id, tkv, (Maybe(DBRow($Row)) mb)|->
          if mb is success(row)
          then if how(row) is success(new_data)                     // The row exists, try to update it.
            then
              with new_row  = dbrow(id, version(row)+1, new_data),  // Update ! ** WARNING: next version ** this is row's version
              map_forget((IDXTable($Row) idx)|-> update(idx)(row, new_row), indexes); // Update the indexes
              (success(new_row), success(updated(row)))
            else (failure, success(not_updated(row)))               // Do not update, return the row
          else (failure, failure)                                   // The row does not exist, do nothing
        )
        is (tree, uptuple) then
        if uptuple is (_, updated) then
        if updated is success(s)
        then (tree, s .. accu)
        else (tree, accu)
  ) is (ntree, it) then
  with nversion = v+1,                                              // Update version, save the indexes.
  map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes);
  data(table) <- dbtable_data(c, nversion, ntree, indexes);
  save(table)(dbi);                                                 // Save the table asynchronously
  it.

    Update several elements, shorthand for list, update the indexes (update, set version, save).
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    List(DBRowId($Row))         ids,
    DBRow($Row) -> Maybe($Row)  how
  ) = update_rows(dbi, table, take_left(ids), how).

      *** [6.3.2] By condition
          ==============================================================================================================

    Update rows by condition on the complete row, update the indexes (update, set version, save).
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    DBRow($Row) -> Bool         which,
    DBRow($Row) -> Maybe($Row)  how
  ) =
  if *data(table) is dbtable_data(c, v, data_tree, indexes) then
  with upvalues = map(                              // Filter and compute updated rows
    (DBRow($Row) r)|-> if how(r) is success(ndata)  // ** WARNING: next version ** this is row's version
      then updated(dbrow(identifier(r), version(r)+1, ndata)) else not_updated(r),
      filter(which, take_left_value(data_tree))
  ),
  if fold_left(upvalues, (data_tree, nil),
    ( (TreeKV(DBRowId($Row), DBRow($Row)), Iterator(DBUpdateResult($Row))) tuple, DBUpdateResult($Row) ur)|->
      if tuple is (tree, accu) then
        if ur is
        {
          not_updated(_)    then (tree, accu)
          updated(new_row)  then
            with t = update(identifier(new_row), tree, (Maybe(DBRow($Row)) mb)|->
              if mb is success(row)
              then map_forget((IDXTable($Row) idx)|-> update(idx)(row, new_row), indexes); success(new_row)
              else should_not_happen(failure)
            ),
            (t, ur .. accu )
        }
  ) is (ntree, it) then
  with nversion = v+1,                                              // Update version, save the indexes.
  map_forget((IDXTable($Row) idx)|-> set_version(idx)(nversion); save(idx)(dbi), indexes);
  data(table) <- dbtable_data(c, nversion, ntree, indexes);
  save(table)(dbi);                                                 // Save the table asynchronously
  it.

    Update rows by condition on the data only, update the indexes (update, set version, save).
public define Iterator(DBUpdateResult($Row)) update_rows(
    DBInfo                      dbi,
    DBTable($Row, $HRow)        table,
    $Row -> Bool                which,
    DBRow($Row) -> Maybe($Row)  how
  ) = update_rows(dbi, table, (DBRow($Row) r)  |-> which(row(r)), how).



  *** [7] Multi-tables queries.
      ==================================================================================================================

define List($Out) flatten_remove_repetitions ( Iterator(List($Out)) itl) =
  fold_left(itl, [ ], (List($Out) accu, List($Out) items)|->
    fold_left(items, accu, (List($Out) accu2, $Out item)|->
      if member(accu2, item) then accu2 else [item . accu2]
    )
  ).

public define List($Out) query(
    DBTable($Row,$HRow)         the_first_table,
    DBRow($Row) -> Bool         the_first_condition,
    (DBRow($Row),$In) -> $Out   the_first_selection,
    DBRow($Row) -> List($In)    the_subquery
  ) =
  with rows_it = get_rows(the_first_table, the_first_condition),
  flatten_remove_repetitions(
    map( (DBRow($Row) row) |->
      map( ($In i) |-> the_first_selection(row, i), the_subquery(row)),
      rows_it
    )
  )
  .