Anubis / Anubis

  Holder       : Matthieu Herrmann
  Author       : Saunders Team
  Creation     : 2014/05/16
  Last update  : 2014/08/25 15:13 by matthieu@embryo (inserted by vim).
  ======================================================================================================================


  Overview:

    SDBMS4 is an evolution of SDBMS3 which itself is an evolution of SDMBS1. SDBMS2 has been lost...
  SDBMS1 has list, SDBMS3 uses 2-4BTREE and SDBMS4 add indexes.
  On the other hand, SDBMS4 drops the "on demand" loading features of SDBMS 1 and 3, i.e. tables are loaded as soon
  as you create initialize them. It also drops the interface compatibility but if you add indexes, you have to
  refactor your code anyway!

    Also, SDBMS4 is "table-centric" instead of "database-centric", i.e. when you create/initialise a table, you get
  a handler on it and this handler is the only way to access the table. Your table is no store in a "higher"
  data-structure (the former "DB($MY_DB)" type from SDBMS 1 and 3). However, several tables still form a data-base
  and are group together (write at the same location on your disk) when sharing a same 'DBInfo' value.

    Includes you (and we) need:
read tools/2-4tree.anubis     // Mainly, BTree24Compare. See indexes.
read tools/iterators.anubis   // Results are given with iterators.

    Note: Even if the API compatibility is dropped, the functions remain roughly the same.

    Note: 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.

    Note: Unless specified, **never** assume anything about the ordering of iterators/lists returned a function !
          If you query a list of items with the list of ids [A, B, C], you may receive the list of rows
          [ (B,..), (A,..),  (C,..) ].

  *** (1) Creating databases and tables
      ==================================================================================================================

    When tables or indexes are loading, something wrong may happen
public type DBLoadingError:
  file_reading_problem(String path),  // When reading the file itself
  file_type_problem   (String path)   // When deserializing the datum
  .

    Create a basic string representation of the error. The code is self-explanatory so we keep it here.
public define String to_string(DBLoadingError e) =
  if e is
  {
    file_reading_problem(path) then "can not read file " + path
    file_type_problem(path)    then "can not deserialize data from file " + path
  }.

    Type of handlers on a 'database information'. Mainly it will contains the directory where to write the tables.
public type DBInfo:...    // Opaque !

    Type of handlers on table. The type of the rows of a table is up to you (but must be serializable). However, this
  data base 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 '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 '$Row' is the current type of the
  rows of the table, and '$HRow' ('H' like 'History') is a type containing the history of all the successive formats
  of rows in the table. What if the type $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.
                 ** Do no put several alternative to a type of rows. We don't know what could happen.**
public type DBTable($Row, $HRow):...    // Opaque !

    The first thing you have to do when using sdbms4 is defining where your data, i.e. your set of tables, are going
  to reside in your filesystem. This is done with the following function.
public define DBInfo init_db(String  data_base_directory).

    Then, you can create your table using the result of 'init_db'. At the beginning, you can will probably use the same
  type for '$Row' and '$HRow'. The 'update' and 'store' functions 'historical conversions'. At first, use 'identity'
  from 'tools/basis.anubis' This database waits 10 secondes before writing its files and uses a 50ms polling.
  See "make_AData" in tools/delegate_writer.anubis for more information.
public define Result(DBLoadingError, DBTable($Row,$HRow)) init_dbtable(
    DBInfo                the_data_base,
    String                table_name,
    $HRow -> $Row         update,
    $Row -> $HRow         store
  ).

    Same as above with custom values for twait and tpoll.
  See "make_AData" in tools/delegate_writer.anubis for more information.
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:
      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 ? The compiler will infer it from your database usage and the mistery won't last long as
  you will add row of a given type in your database. If needed, put a type annotation, e.g.
                   (Result(DBLoadingError, DBTable(Client, Client))) init_dbtable(dbi, "client, ...)
  where 'Client' and 'HClient' have been previously defined by you. Notice that a datum of type 'Client' is just an
  entire row in the 'clients' table. Hence, 'Client' should be a type with just one alternative and the components
  needed for a 'client', like 'name', 'address', etc...

    Notice that your table will be saved in 'data_base_directory'/'table_name'.{0,1}. The file '.0' is the most recent,
  the '.1' is the previous save.

  *** (2) Managing changes in table formats.
      ==================================================================================================================

    Up to here the types used as instantiations of '$Row' and '$HRow' are the same one. That's OK, and you have begun to
  distribute your program and your users have created tables whose types are your types 'Client', 'Product' etc...
  If you simply modify the definition of (say) the type '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 '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), Word32 age, to your Client

  type Client:
    client(String name, String address).

    The first thing to do is to to create the type 'HClient'. The first alternative of 'HClient' should be the first
  version of 'Client' and should never change as it represents an 'historical' client. We recommend to name it
  'version_1', and to use the name 'version_2' for the second alternative representing the new type of rows, etc..

  type HClient:
    version_1(String name, String address),
    version_2(String name, String address, Word32 age).

                                                 ** 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 'HClient' (or any other instance of '$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.
                                             ** End of important notice **

    Now, we need 'conversion' functions in both directions between 'Client' and 'HClient'.
  Presisely, we need the functions:

    (HClient -> Client)update // Transform an historival row in the current format
    (Client -> HClient)store  // The other way

    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 (version_1), we need a default value for the age. We have chosen 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 '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):

    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 'HClient' (hence the name of the function 'store'),
  and according to the above function they are stored as 'version_2(...)'. The type 'Client' is used only internally.
  This is required for being able of handling file containing different versions of the tables.


  *** (2) 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 'DBRowId($Row)' (an opaque type scheme).
public type DBRowId($Row):...

    A datum of type DBRow($Row) contains the identifier of the row, a version number and the row itself. The version
  number is 0 when the row is created (by 'add_row' or 'add_rows' above) and is incremented by 1 each time the row is
  updated. This may be used to detect the fact that the row has been modified between two operations.
public type DBRow($Row):
  dbrow(  DBRowId($Row) identifier,
          Int           version,
          $Row          row)
  .

    *** (2.1) Iterate on the whole table
        ================================================================================================================

    Given a table, you can browse it by couple (row identifier, row), by row identifiers or by rows.
                                           ** DO NOT ASSUME ANY ORDERING ! **
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).

    *** (2.2) Get rows
        ================================================================================================================

    You can get a row by its id. Return failure if the row is not found.
public define Maybe(DBRow($Row)) get_row(
    DBTable($Row, $HRow)  table,
    DBRowId($Row)         id
  ).

    You can query several row at once: get an iterator on several rows, by ids.
  Note: You can have less rows than ids if the table does not contains some ids.
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    Iterator(DBRowId($Row)) ids
  ).

    Same as above, but if you give a list, you get a list !
public define List(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)    table,
    List(DBRowId($Row))     ids
  ).

    You can also get rows by condition on a complete row.
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
  ).

    Or by condition on the data only
public define Iterator(DBRow($Row)) get_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
  ).

    *** (2.3) Count rows
        ================================================================================================================

    Count rows fulfilling a condition on the complete row.
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which
  ).

    Count rows fulfilling a condition on the data only.
public define Int get_number_of_rows(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which
  ).

    Test if their is **strictly** more rows fulfilling a condition then a given number. Stop as soon as possible.
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    DBRow($Row) -> Bool   which,
    Int                   n
  ).

    Same as above, testing only the data.
public define Bool has_more_rows_than(
    DBTable($Row, $HRow)  table,
    $Row -> Bool          which,
    Int                   n
  ).


  *** (3) Manipulating the database
      ==================================================================================================================
    While a table is manipulated, associated index are updated accordingly. See indexes for more informations.

    *** (3.1) Adding 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
  ).

    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
  ).

    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
  ).

    *** (3.2) Deleting rows
        ================================================================================================================

      *** (3.2.1) By identifier
          ==============================================================================================================

    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
  ).

    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
  ).

    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
  ).

      *** (3.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
  ).

    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
  ).

    *** (3.3) 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 'how' below) of type:

                                               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 data base when phase 2 begins. It should compare the version number with the one it
  remembers and decide if the row must be updated or not. If the row must not be updated, the function 'how' must return
  'failure'. If on the contrary, the row must be updated, it must return the value 'success(r)', where 'r' is the new
  value of the row.

    Now, if no error occurs, the 'update_row' function returns a datum of type:

public type DBUpdateResult($Row):
   not_updated   (DBRow($Row) the_row),
   updated       (DBRow($Row) the_row).

    i.e. the result is either 'not_updated(r)' or 'updated(r)'. It is 'not_updated(r)' when the row has not been
  updated, i.e. when the 'how' function has returned 'failure', and it is 'updated(r)' if the row has been updated. In
  both cases, '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.

      *** (3.3.1) By identifier
          ==============================================================================================================

    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
  ).

    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
  ).

    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
  ).

      *** (3.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
  ).

    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
  ).

    *** (3.4) Multi-tables queries.
        ================================================================================================================

  See sdbms1.anubis. Yes, I'm lazy !

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
  ).

  *** (4) Adding indexes
      ==================================================================================================================

    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):...

    *** (4.1) Initialisation
        ================================================================================================================

    Indexes are 'pluged' 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 50ms polling.
  See "make_AData" in tools/delegate_writer.anubis for more information.
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) -> BTree24Compare    cmp             // Comparison function for the index keys
  ).

    Same as above with custom values for twait and tpoll.
  See "make_AData" in tools/delegate_writer.anubis for more information.
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) -> BTree24Compare    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 find a **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...

    *** (4.2) Consultation
        ================================================================================================================

    Iterate in order (lower to upper) over all the keys, obtaining the each time the list of associated identifiers.
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 identifiers.
public define Iterator( $Key ) take_key(DBIndex($Key, $Row) my_idx).

    Iterate in order of the keys (lower to upper) over all the identifiers.
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.
public define Iterator( DBRow($Row) ) take_rows(DBIndex($Key, $Row) my_idx).

    Get the list of identifiers associated to a key.
public define Iterator(DBRowId($Row)) get_ids($Key key, DBIndex($Key, $Row) my_idx).

    Get the list of rows associated to a key.
public define Iterator(DBRow($Row)) get_rows($Key key, DBIndex($Key, $Row) my_idx).

    *** (4.3) Example
        ================================================================================================================
    Here is a simple example, adding string read from the command line into a database and indexing them by length.
  Note: for maximum interativity, this example sets the waiting time and the poll time to 0. Set those according to
  your needs !!

  read tools/basis.anubis
  read tools/iterators.anubis
  read tools/2-4tree.anubis
  read data_base/sdbms4.anubis

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

  global define 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"))
      }
    }.



  ======================================================================================================================
  = 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 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.
define BTree24Compare 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) -> BTree24Compare    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) -> BTree24Compare    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) -> BTree24Compare    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 ) 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
    )
  )
  .