obitools3/src/obidmscolumn.c

/****************************************************************************
 * OBIDMS_column functions                                                  *
 ****************************************************************************/

/**
 * @file obidmscolumn.c
 * @author Celine Mercier
 * @date 22 May 2015
 * @brief Functions for the shared elements of all the OBIColumn structures.
 */


#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <dirent.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdbool.h>
#include <math.h>
#include <sys/mman.h>	/* mmap() is defined in this header */

#include "obidmscolumn.h"
#include "obidmscolumndir.h"
#include "obidms.h"
#include "obitypes.h"
#include "obierrno.h"
#include "obidebug.h"
#include "obilittlebigman.h"


#define DEBUG_LEVEL 0


/**************************************************************************
 *
 * D E C L A R A T I O N   O F   T H E   P R I V A T E   F U N C T I O N S
 *
 **************************************************************************/


/**
 * @brief Internal function building the file name for a column.
 *
 * The function builds the file name corresponding to a column of an OBIDMS.
 *
 * @warning The returned pointer has to be freed by the caller.
 *
 * @param column_name the name of the OBIDMS column.
 *
 * @return a pointer to the column file name
 * @retvalue NULL if an error occurs
 *
 * ###Error values
 * 		- OBIDMS_MEMORY_ERROR    : something wrong occurred during memory allocation.
 *
 * @since May 2015
 * @author Eric Coissac (eric.coissac@metabarcoding.org)
 */
static char *build_column_file_name(const char *column_name, obiversion_t version_number);


/**
 * @brief Internal function building the file name for a column version file.
 *
 * The column version file indicates the latest version number for a column.
 * This function returns the name of the file storing this information.
 *
 * @warning The returned pointer has to be freed by the caller.
 *
 * @param column_name the name of the OBIDMS column.
 *
 * @return a pointer to the version file name
 * @retvalue NULL if an error occurs
 *
 * ###Error values
 * 		- OBIDMS_MEMORY_ERROR    : something wrong occurs during memory allocation.
 *
 * @since May 2015
 * @author Eric Coissac (eric.coissac@metabarcoding.org)
 */
static char *build_version_file_name(const char *column_name);


/**
 * @brief Internal function returning a new column version number
 *        in the `dms` database
 *
 * @param dms a pointer as returned by obi_create_dms() or obi_open_dms()
 * @param column_name the name of the column
 * @param block is the call is blocking or not
 *                 - `true` the call is blocking
 *                 - `false` the call is not blocking
 *
 * @return the bigger version number used for this column
 * @retvalue -1 if the column does not exist
 *
 * @since May 2015
 * @author Eric Coissac (eric.coissac@metabarcoding.org)
 */
static obiversion_t obi_get_new_version_number(OBIDMS_column_directory_p column_directory, bool block);


/**
 * @brief Internal function creating a new column version file
 *        in the `dms` database
 *
 * The new file is initialized with the minimum version number `0`.
 *
 * @param dms a pointer as returned by obi_create_dms() or obi_open_dms()
 * @param column_name the name of the column
 *
 * @return the next usable version number for this column : `0`
 * @retvalue -1 if the column does not exist
 *
 * @since May 2015
 * @author Eric Coissac (eric.coissac@metabarcoding.org)
 */
static int create_version_file(OBIDMS_column_directory_p column_directory);


/**
 * @brief Internal function setting the elements names of the lines of a
 * column in the header of the OBIDMS column structure.
 *
 * @param column a pointer as returned by obi_create_column()
 * @param elements_names the names of the elements with ';' as separator
 *
 * @return 0 if the operation was successfully completed
 * @retvalue -1 if an error occurred
 *
 * @since July 2015
 * @author Celine Mercier (celine.mercier@metabarcoding.org)
 */
int obi_column_set_elements_names(OBIDMS_column_p column, const char* elements_names);


/**
 * @brief Internal function computing how many lines of an OBIDMS column fill in a memory page.
 *
 * @param data_type the data OBIType
 * @param nb_elements_per_line the number of elements per line
 *
 * @return the line count for one memory page
 *
 * @since September 2015
 * @author Celine Mercier (celine.mercier@metabarcoding.org)
 */
size_t get_line_count_per_page(OBIType_t data_type, size_t nb_elements_per_line);


/************************************************************************
 *
 * D E F I N I T I O N   O F   T H E   P R I V A T E   F U N C T I O N S
 *
 ************************************************************************/


static char *build_column_file_name(const char *column_name, obiversion_t version_number)
{
	char *filename;

	// Build the database directory name
	if (asprintf(&filename,"%s@%d.odc", column_name, version_number) < 0)
	{
		obi_set_errno(OBICOL_MEMORY_ERROR);
		obidebug(1, "\nError building a column file name");
		return NULL;
	}

	return filename;
}


static char *build_version_file_name(const char *column_name)
{
	char *filename;

	// Build the database directory name
	if (asprintf(&filename,"%s.odv", column_name) < 0)
	{
		obi_set_errno(OBICOL_MEMORY_ERROR);
		obidebug(1, "\nError building a version file name");
		return NULL;
	}

	return filename;
}


static obiversion_t obi_get_new_version_number(OBIDMS_column_directory_p column_directory, bool block)
{
	off_t 			loc_size;
	obiversion_t 	new_version_number;
	char* 			version_file_name;
	int    			column_dir_file_descriptor;
	int    			version_file_descriptor;
	bool   			little_endian;
	int 			lock_mode;

	new_version_number = 0;
	loc_size = sizeof(bool) + sizeof(obiversion_t);

	// Select the correct lockf operation according to the blocking mode
	if (block)
		lock_mode=F_LOCK;
	else
		lock_mode=F_TLOCK;

	// Build the version file name
	version_file_name = build_version_file_name(column_directory->column_name);
	if (version_file_name == NULL)
		return -1;

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column file directory");
		free(version_file_name);
		return -1;
	}

	// Open the version file
	version_file_descriptor = openat(column_dir_file_descriptor, version_file_name, O_RDWR);
	if (version_file_descriptor < 0)
	{
		if (errno == ENOENT)
			return create_version_file(column_directory);
		else
		{
			obi_set_errno(OBICOL_UNKNOWN_ERROR);
			return -1;
		}
		obidebug(1, "\nError opening a version file");
		free(version_file_name);
	}

	// Test if the version file size is ok
	if (lseek(version_file_descriptor, 0, SEEK_END) < loc_size)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError testing if a version file size is ok");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	// Prepare the file for locking
	if (lseek(version_file_descriptor, 0, SEEK_SET) != 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError preparing a version file for locking");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	// Lock the file
	if (lockf(version_file_descriptor, lock_mode, loc_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError locking a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	// Read the endianness of the file
    if (read(version_file_descriptor, &little_endian, sizeof(bool)) < sizeof(bool))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError reading the endianness of a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

    // Check if endianness is correct
    if (little_endian != obi_is_little_endian())
	{
		obi_set_errno(OBICOL_BAD_ENDIAN_ERROR);
		obidebug(1, "\nEndianness of a version file is incorrect");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

    // Read the current version number
    if (read(version_file_descriptor, &new_version_number, sizeof(obiversion_t)) < sizeof(obiversion_t))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError reading a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

    new_version_number++;

    // Write the new version number
	if (lseek(version_file_descriptor, sizeof(bool), SEEK_SET) != sizeof(bool))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError writing a new version number in a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	if (write(version_file_descriptor, &new_version_number, sizeof(obiversion_t)) < sizeof(obiversion_t))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError writing a new version number in a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	// Prepare for unlocking
	if (lseek(version_file_descriptor, 0, SEEK_SET) != 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError preparing the unlocking of a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	// Unlock the file
	if (lockf(version_file_descriptor, F_ULOCK, loc_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError unlocking a version file");
		close(version_file_descriptor);
		free(version_file_name);
		return -1;
	}

	close(version_file_descriptor);
	free(version_file_name);

	return new_version_number;
}


static int create_version_file(OBIDMS_column_directory_p column_directory)
{
	off_t 			loc_size;
	obiversion_t 	version_number;
	char* 			version_file_name;
	int    			column_dir_file_descriptor;
	int    			version_file_descriptor;
	bool   			little_endian;

	loc_size = sizeof(bool) + sizeof(obiversion_t);
	version_number = 0;

	version_file_name = build_version_file_name(column_directory->column_name);
	if (version_file_name == NULL)
		return -1;

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		free(version_file_name);
		return -1;
	}

	// Get the file descriptor associated to the version file
	version_file_descriptor = openat(column_dir_file_descriptor, version_file_name, O_RDWR | O_CREAT);
	if (version_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError opening a version file");
		free(version_file_name);
		return -1;
	}

	// Lock the file
	if (lockf(version_file_descriptor, F_LOCK, loc_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError locking a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Truncate the version file to the right size
	if (ftruncate(version_file_descriptor, loc_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError truncating a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Position offset to 0 to prepare for writing
	if (lseek(version_file_descriptor, 0, SEEK_SET) != 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError changing offset of a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	little_endian = obi_is_little_endian();

	// Write endianness
	if (write(version_file_descriptor, &little_endian, sizeof(bool)) < sizeof(bool))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError writing endianness of a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Write version number
	if (write(version_file_descriptor, &version_number, sizeof(obiversion_t)) < sizeof(obiversion_t))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError writing version number in a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Prepare for unlocking
	if (lseek(version_file_descriptor, 0, SEEK_SET) != 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError preparing a version file for unlocking");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Unlock the file
	if (lockf(version_file_descriptor, F_ULOCK, loc_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError unlocking a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	close(version_file_descriptor);
	free(version_file_name);

	return version_number;
}


int obi_column_set_elements_names(OBIDMS_column_p column, const char* elements_names)
{
	strcpy((column->header)->elements_names, elements_names);
	return 0;
}

size_t get_line_count_per_page(OBIType_t data_type, size_t nb_elements_per_line)
{
	return getpagesize() / (obi_sizeof(data_type) * nb_elements_per_line);
}


/**********************************************************************
 *
 * D E F I N I T I O N   O F   T H E   P U B L I C   F U N C T I O N S
 *
 **********************************************************************/

obiversion_t obi_get_latest_version_number(OBIDMS_column_directory_p column_directory)
{
	off_t 			loc_size;
	obiversion_t 	latest_version_number;
	char * 			version_file_name;
	int    			column_dir_file_descriptor;
	int    			version_file_descriptor;
	bool   			little_endian;

	loc_size = sizeof(bool) + sizeof(obiversion_t);
	latest_version_number = 0;

	version_file_name = build_version_file_name(column_directory->column_name);
	if (version_file_name==NULL)
		return -1;

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		free(version_file_name);
		return -1;
	}

	// Get the file descriptor associated to the version file
	version_file_descriptor = openat(column_dir_file_descriptor, version_file_name, O_RDONLY);
	if (version_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError opening a version file");
		free(version_file_name);
		return -1;
	}

	// Check that the version file size is ok
	if (lseek(version_file_descriptor, 0, SEEK_END) < loc_size)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError testing if a version file size is ok");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Set the offset to 0 in the version file
	if (lseek(version_file_descriptor, 0, SEEK_SET) != 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError setting the offset of a version file to 0");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	// Read the endianness
    if (read(version_file_descriptor, &little_endian, sizeof(bool)) < sizeof(bool))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError reading the endianness of a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

    // Verify the endianness
    if (little_endian != obi_is_little_endian())
	{
		obi_set_errno(OBICOL_BAD_ENDIAN_ERROR);
		obidebug(1, "\nEndianness of a version file is incorrect");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

    // Read the latest version number
    if (read(version_file_descriptor, &latest_version_number, sizeof(obiversion_t)) < sizeof(obiversion_t))
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError reading the latest version number in a version file");
		free(version_file_name);
		close(version_file_descriptor);
		return -1;
	}

	free(version_file_name);
	close(version_file_descriptor);

	return latest_version_number;
}


obiversion_t obi_column_get_latest_version_from_name(OBIDMS_p dms, const char* column_name)
{
	OBIDMS_column_directory_p	column_directory;
	obiversion_t	   			latest_version;

	// Get the column directory structure associated to the column
	column_directory = obi_open_column_directory(dms, column_name);
	if (column_directory == NULL)
	{
		obidebug(1, "\nProblem opening a column directory structure");
		return -1;
	}

	// Get the latest version number
	latest_version = obi_get_latest_version_number(column_directory);
	if (latest_version < 0)
	{
		obidebug(1, "\nProblem getting the latest version number in a column directory");
		return -1;
	}

	return latest_version;
}


size_t obi_get_platform_header_size()
{
	return getpagesize() * 1;
}


OBIDMS_column_p obi_create_column(OBIDMS_p dms,
		                          const char *column_name,
								  OBIType_t type,
								  size_t nb_lines,
								  size_t nb_elements_per_line,
								  const char*  elements_names)
{
	OBIDMS_column_p 		new_column;
	OBIDMS_column_directory_p	column_directory;
	OBIDMS_column_header_p 	header;
	size_t 					file_size;
	obiversion_t 			version_number;
	char* 					column_file_name;
	int 					column_file_descriptor;
	int						column_dir_file_descriptor;
	size_t 					header_size;
	size_t 					data_size;
	size_t					minimum_line_count;

	new_column = NULL;

	// Check that the informations given are not NULL/invalid/greater than the allowed sizes
	if (dms == NULL)
	{
		obidebug(1, "\nCan't create column because of invalid DMS");
		return NULL;
	}
	if (column_name == NULL)
	{
		obidebug(1, "\nCan't create column because of empty column name");
		return NULL;
	}
	if ((type < 1) || (type > 4))
	{
		obidebug(1, "\nCan't create column because of invalid data type");
		return NULL;
	}

	// The initial line count should be between the minimum (corresponding to the page size) and the maximum allowed
	minimum_line_count = get_line_count_per_page(type, nb_elements_per_line);
	if (nb_lines > MAXIMUM_LINE_COUNT)
	{
		obidebug(1, "\nCan't create column because of line count greater than the maximum allowed (%lld)", MAXIMUM_LINE_COUNT);
		return NULL;
	}
	else if (nb_lines < minimum_line_count)
		nb_lines = minimum_line_count;

	// The number of elements names should be equal to the number of elements per line

	if ((elements_names == NULL) && (nb_elements_per_line > 1))
	{
		obidebug(1, "\nCan't create column because no elements names were given for a number of elements per line greater than 1");
		return NULL;
	}
	else if ((elements_names != NULL) && (nb_elements_per_line > 1))
	{
		char* token;
		size_t n = 0;
		token = strdup(elements_names);
		token = strtok(token, ";");
		while (token != NULL)
		{
			token = strtok(NULL, ";");
			n++;
		}
		if (n != nb_elements_per_line)
		{
			obidebug(1, "\nCan't create column because the number of elements names given is not equal to the number of elements per line");
			return NULL;
		}
	}
	else if ((nb_elements_per_line == 1) && (strcmp(elements_names, column_name) != 0))
	{
		obidebug(1, "\nCan't create column because the element name does not match the column name");
		return NULL;
	}

	// Get the column directory structure associated to the column
	column_directory = obi_column_directory(dms, column_name);
	if (column_directory == NULL)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError opening a column directory structure");
		return NULL;
	}

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		return NULL;
	}

	// Calculate the size needed
	header_size = obi_get_platform_header_size();
	data_size = obi_array_sizeof(type, nb_lines, nb_elements_per_line);
	file_size = header_size + data_size;

	// Get the latest version number
	version_number = obi_get_new_version_number(column_directory, true);
	if (version_number < 0)
	{
		return NULL;
	}

	// Get the column file name
	column_file_name = build_column_file_name(column_name, version_number);
	if (column_file_name == NULL)
	{
		return NULL;
	}

	// Open the column file
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDWR | O_CREAT);
	if (column_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		free(column_file_name);
		return NULL;
	}

	// Truncate the column file to the right size
	if (ftruncate(column_file_descriptor, file_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError truncating a column file to the right size");
		close(column_file_descriptor);
		free(column_file_name);
		return NULL;
	}

	// Allocate the memory for the column structure
	new_column = (OBIDMS_column_p) malloc(sizeof(OBIDMS_column_t));
	if (new_column == NULL)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError allocating the memory for the column structure");
		close(column_file_descriptor);
		free(column_file_name);
		return NULL;
	}

	// Fill the column structure
	new_column->dms    			 = dms;
	new_column->column_directory = column_directory;
	new_column->header 			 = mmap(NULL,
			                  	  	    header_size,
			                  	  	    PROT_READ | PROT_WRITE,
			                  	  	    MAP_SHARED,
			                  	  	    column_file_descriptor,
			                  	  	    0
			                 	 	   );

	if (new_column->header == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the header of a column");
		close(column_file_descriptor);
		free(column_file_name);
		free(new_column);
		return NULL;
	}

	new_column->data   			 = mmap(NULL,
			                 	 	 	data_size,
			                 	 	 	PROT_READ | PROT_WRITE,
			                 	 	 	MAP_SHARED,
			                 	 	 	column_file_descriptor,
			                 	 	 	header_size
			                			);

	if (new_column->data == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the data of a column");
		munmap(new_column->header, header_size);
		close(column_file_descriptor);
		free(column_file_name);
		free(new_column);
		return NULL;
	}

	new_column->writable  = true;

	header 				  		 = new_column->header;
	header->little_endian 		 = obi_is_little_endian();
	header->header_size   		 = header_size;
	header->line_count   	 	 = nb_lines;
	header->lines_used    		 = 0;
	header->nb_elements_per_line = nb_elements_per_line;
	header->data_type     		 = type;
	header->creation_date 		 = time(NULL);
	header->version       		 = version_number;
	header->cloned_from    		 = -1;
	header->comments[0]   		 = 0x0;

	obi_column_set_elements_names(new_column, elements_names);

	strncpy(header->name, column_name, OBIDMS_MAX_COLNAME);

	// Fill the data with NA values
	obi_ini_to_NA_values(new_column, 0, nb_lines);

	free(column_file_name);
	close(column_file_descriptor);

	return new_column;
}


OBIDMS_column_p obi_open_column(OBIDMS_p dms, const char* column_name, obiversion_t version_number)
{
	OBIDMS_column_p 		    column;
	OBIDMS_column_directory_p	column_directory;
	char* 						column_file_name;
	int 						column_file_descriptor;
	int							column_dir_file_descriptor;
	size_t 						header_size;
	size_t 						data_size;

	column = NULL;

	// Get the column directory structure associated to the column
	column_directory = obi_open_column_directory(dms, column_name);
	if (column_directory == NULL)
	{
		obidebug(1, "\nError opening a column directory structure");
		return NULL;
	}

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		return NULL;
	}

	// Calculate the header size
	header_size = obi_get_platform_header_size();

	// Get the latest version number if it has the value -1 (not given by user)
	if (version_number == -1)
	{
		version_number = obi_get_latest_version_number(column_directory);
		if (version_number < 0)
		{
			obidebug(1, "\nError getting the latest version number in a column directory");
			return NULL;
		}
	}

	// Get the column file name
	column_file_name = build_column_file_name(column_name, version_number);
	if (column_file_name == NULL)
	{
		return NULL;
	}

	// Open the column file, ALWAYS READ-ONLY
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDONLY);
	if (column_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError opening column file");
		free(column_file_name);
		return NULL;
	}

	// Allocate the memory for the column structure
	column = (OBIDMS_column_p) malloc(sizeof(OBIDMS_column_t));
	if (column == NULL)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError allocating the memory for a column structure");
		close(column_file_descriptor);
		free(column_file_name);
		return NULL;
	}

	// Fill the column structure
	column->dms    			 = dms;
	column->column_directory = column_directory;
	column->header 			 = mmap(NULL,
			                  	  	header_size,
			                  	  	PROT_READ,
			                  	  	MAP_PRIVATE,
			                  	  	column_file_descriptor,
			                  	  	0
			                 	 	);

	if (column->header == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the header of a column");
		close(column_file_descriptor);
		free(column_file_name);
		free(column);
		return NULL;
	}

	// Check endianness?

	// Compute data size from the informations in the header
	data_size = ((column->header)->line_count) * sizeof((column->header)->data_type);

	column->data   			 = mmap(NULL,
			                 	 	data_size,
			                 	 	PROT_READ,
									MAP_PRIVATE,
			                 	 	column_file_descriptor,
			                 	 	header_size
			                	    );

	if (column->data == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the data of a column");
		munmap(column->header, header_size);
		close(column_file_descriptor);
		free(column_file_name);
		free(column);
		return NULL;
	}

	column->writable  = false;

	free(column_file_name);
	close(column_file_descriptor);

	return column;
}


OBIDMS_column_p obi_clone_column(OBIDMS_p dms, const char* column_name, obiversion_t version_number, bool clone_data)
{
	OBIDMS_column_p   column_to_clone;
	OBIDMS_column_p   new_column;
	size_t 			  nb_lines;
	size_t			  nb_elements_per_line;
	OBIType_t		  data_type;

	column_to_clone = obi_open_column(dms, column_name, version_number);
	if (column_to_clone == NULL)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError opening the column to clone");
		return NULL;
	}

	data_type = (column_to_clone->header)->data_type;

	nb_elements_per_line = (column_to_clone->header)->nb_elements_per_line;

	if (clone_data)
		nb_lines = (column_to_clone->header)->line_count;
	else
		nb_lines = get_line_count_per_page(data_type, nb_elements_per_line);	// minimum line count corresponding to one memory page

	new_column = obi_create_column(dms,
								   column_name,
								   data_type,
								   nb_lines,
								   nb_elements_per_line,
								   (column_to_clone->header)->elements_names);

	if (new_column == NULL)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError creating the new column when cloning a column");
		// The new file is deleted
		const char* column_file_name = build_column_file_name(column_name, version_number);
		if (remove(column_file_name) < 0)
			obidebug(1, "\nError deleting a bad cloned file");
	}

	(new_column->header)->cloned_from = version_number;
	// TODO copy header->comments?

	if (clone_data)
	{
		memcpy(new_column->data, column_to_clone->data, nb_lines*nb_elements_per_line*sizeof(data_type));
		(new_column->header)->lines_used = nb_lines;
	}

	// close column_to_clone
	if (obi_close_column(column_to_clone) < 0)
	{
		obidebug(1, "\nError closing a column that has been cloned");
		// TODO return NULL or not?
	}

	return new_column;
}


int obi_close_column(OBIDMS_column_p column)
{
	size_t data_size;

	// Munmap data
	data_size = (column->header)->line_count * (column->header)->nb_elements_per_line * sizeof((column->header)->data_type);
	if (munmap(column->data, data_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError munmapping column data");
		return -1;
	}

	// Munmap header
	if (munmap(column->header, (column->header)->header_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError munmapping a column header");
		return -1;
	}

	free(column);

	return 0;
}


int obi_truncate_column_to_lines_used(OBIDMS_column_p column)
{
	size_t file_size;
	size_t data_size;
	size_t new_line_count;
	double multiple;
	int column_dir_file_descriptor;
	int column_file_descriptor;
	char* column_file_name;

	// Compute the new line count = the number of lines used rounded to the nearest multiple of page size
	multiple = ceil((double) ((column->header)->lines_used * (column->header)->nb_elements_per_line * obi_sizeof((column->header)->data_type)) / (double) getpagesize());
	new_line_count = (int) multiple * getpagesize();

	// Check that it is actually greater than the current number of lines allocated in the file, otherwise no need to truncate
	if ((column->header)->line_count == new_line_count)
		return 0;

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd((column->column_directory)->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		return -1;
	}

	// Get the column file name
	column_file_name = build_column_file_name((column->header)->name, (column->header)->version);
	if (column_file_name == NULL)
	{
		return -1;
	}

	// Open the column file
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDWR);
	if (column_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor of a column file");
		free(column_file_name);
		return -1;
	}

	// Unmap the data before truncating the file
	data_size = obi_array_sizeof((column->header)->data_type, (column->header)->line_count, (column->header)->nb_elements_per_line);
	if (munmap(column->data, data_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError munmapping the data of a column before truncating");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}

	// Truncate the column file
	data_size = obi_array_sizeof((column->header)->data_type, new_line_count, (column->header)->nb_elements_per_line);
	file_size = (column->header)->header_size + data_size;
	if (ftruncate(column_file_descriptor, file_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError truncating a column file at the number of lines used");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}

	// Remap the data
	column->data   			 = mmap(NULL,
			                 	 	data_size,
			                 	 	PROT_READ | PROT_WRITE,
			                 	 	MAP_SHARED,
			                 	 	column_file_descriptor,
									(column->header)->header_size
			                		);

	if (column->data == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError re-mmapping the data of a column after truncating");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}

	// Set line_count to the new line count
	(column->header)->line_count = new_line_count;

	free(column_file_name);
	close(column_file_descriptor);

	return 0;
}


int obi_enlarge_column(OBIDMS_column_p column)
{
	size_t file_size;
	size_t old_data_size;
	size_t new_data_size;
	size_t header_size;
	size_t old_line_count;
	size_t new_line_count;
	int column_dir_file_descriptor;
	int column_file_descriptor;
	char* column_file_name;
	void* new_data;

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd((column->column_directory)->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor for a column directory");
		return -1;
	}

	// Get the column file name
	column_file_name = build_column_file_name((column->header)->name, (column->header)->version);
	if (column_file_name == NULL)
	{
		return -1;
	}

	// Open the column file
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDWR | O_CREAT);
	if (column_file_descriptor < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor of a column file");
		free(column_file_name);
		return -1;
	}

	// Calculate the new file size
	old_line_count = (column->header)->line_count;
	new_line_count = old_line_count * GROWTH_FACTOR;

	if (new_line_count > MAXIMUM_LINE_COUNT)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError enlarging a column file: new line count greater than the maximum allowed");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}
	old_data_size = obi_array_sizeof((column->header)->data_type, old_line_count, (column->header)->nb_elements_per_line);
	new_data_size = old_data_size * GROWTH_FACTOR;
	header_size = (column->header)->header_size;
	file_size = header_size + new_data_size;

	// Enlarge the file // TODO isn't it possible that this makes the file "move"?
	if (ftruncate(column_file_descriptor, file_size) < 0)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError enlarging a column file");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}

	// Remap the data: try enlarging mapped region (this actually never works on my mac without the MAP_FIXED flag which overwrites everything)
	//obidebug(2, "\ntry enlarging mapped region: old size = %ld, new size = %ld, size = %ld", old_data_size, new_data_size, new_data_size - old_data_size);
	new_data	   			 = mmap(column->data,
			                 	 	new_data_size - old_data_size,
			                 	 	PROT_READ | PROT_WRITE,
			                 	 	MAP_SHARED,
			                 	 	column_file_descriptor,
									old_data_size
			                		);

	if (new_data == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError re-mmapping the data of a column after enlarging the file");
		free(column_file_name);
		close(column_file_descriptor);
		return -1;
	}

	// If remap failed: Unmap and map the data again
	if (new_data != (column->data)) // TODO check that this works without exception
	{
		//obidebug(2, "\nEnlarging mapped region failed: Unmap and map the data again, %x != %x", column->data, new_data);
		if (munmap(column->data, old_data_size) < 0)
		{
			obi_set_errno(OBICOL_UNKNOWN_ERROR);
			obidebug(1, "\nError munmapping the data of a column before enlarging");
			free(column_file_name);
			close(column_file_descriptor);
			return -1;
		}

		column->data   			 = mmap(NULL,
										new_data_size,
										PROT_READ | PROT_WRITE,
										MAP_SHARED,
										column_file_descriptor,
										header_size
										);

		if (column->data == MAP_FAILED)
		{
			obi_set_errno(OBICOL_UNKNOWN_ERROR);
			obidebug(1, "\nError re-mmapping the data of a column after enlarging the file");
			free(column_file_name);
			close(column_file_descriptor);
			return -1;
		}
	}

	// Set new line count
	(column->header)->line_count = new_line_count;

	// Initialize new data lines to NA
	obi_ini_to_NA_values(column, old_line_count, new_line_count - old_line_count);

	free(column_file_name);
	close(column_file_descriptor);

	return 0;
}


int obi_truncate_and_close_column(OBIDMS_column_p column)
{
	if (obi_truncate_column_to_lines_used(column) < 0)
		return -1;
	if (obi_close_column(column) < 0)
		return -1;

	return 0;
}


void obi_ini_to_NA_values(OBIDMS_column_p column, size_t start, size_t nb_lines)
{
	size_t i, end, nb_elements;

	nb_elements = nb_lines*((column->header)->nb_elements_per_line);
	end = start + nb_elements;

	switch ((column->header)->data_type) {
	case OBI_VOID:  // TODO;
					break;

	case OBI_INT:   for (i=start;i<end;i++)
					{
						*(((obiint_t*) (column->data)) + i) = OBIInt_NA;
					}
					break;

	case OBI_FLOAT: for (i=start;i<end;i++)
					{
						*(((obifloat_t*) (column->data)) + i) = OBIFloat_NA;
					}
					break;

	case OBI_BOOL:  for (i=start;i<end;i++)
					{
						*(((obibool_t*) (column->data)) + i) = OBIBool_NA;
					}
					break;

	case OBI_CHAR:  for (i=start;i<end;i++)
					{
						*(((obichar_t*) (column->data)) + i) = OBIChar_NA;
					}
					break;

	case OBI_IDX:   for (i=start;i<end;i++)
					{
						*(((obiidx_t*) (column->data)) + i) = OBIIdx_NA;
					}
					break;
	}
}


void obi_column_make_unwritable(OBIDMS_column_p column)
{
	column->writable = false;
}


size_t obi_column_get_line_count(OBIDMS_column_p column)
{
	return (column->header)->line_count;
}


size_t obi_column_get_nb_lines_used(OBIDMS_column_p column)
{
	return (column->header)->lines_used;
}


OBIType_t obi_column_get_data_type(OBIDMS_column_p column)
{
	return (column->header)->data_type;
}


OBIType_t obi_column_get_data_type_from_name(OBIDMS_p dms, const char* column_name)
{
	OBIDMS_column_header_p 		header;
	OBIDMS_column_directory_p	column_directory;
	char* 						column_file_name;
	int 						column_file_descriptor;
	int							column_dir_file_descriptor;
	size_t 						header_size;
	OBIType_t					data_type;
	obiversion_t	   			version_number;

	// Get the column directory structure associated to the column
	column_directory = obi_open_column_directory(dms, column_name);
	if (column_directory == NULL)
	{
		obidebug(1, "\nError opening a column directory structure");
		return -1;
	}

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor of a column directory");
		obi_close_column_directory(column_directory);
		return -1;
	}

	// Calculate the header size
	header_size = obi_get_platform_header_size();

	// Get the latest version number
	version_number = obi_get_latest_version_number(column_directory);
	if (version_number < 0)
	{
		obidebug(1, "\nError getting the latest version number in a column directory");
		return -1;
	}

	// Get the column file name
	column_file_name = build_column_file_name(column_name, version_number);
	if (column_file_name == NULL)
	{
		return -1;
	}

	// Open the column file (READ-ONLY)
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDONLY);
	if (column_file_descriptor < 0)
	{
		obidebug(1, "\nError opening a column file");
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		free(column_file_name);
		return -1;
	}

	// Fill the header structure
	header = mmap(NULL,
			 header_size,
			 PROT_READ,
			 MAP_SHARED,
			 column_file_descriptor,
			 0
			 );

	if (header == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the header of a column");
		close(column_file_descriptor);
		free(column_file_name);
		return -1;
	}

	// Check endianness?

	data_type = header->data_type;

	free(column_file_name);
	close(column_file_descriptor);
	munmap(header, header_size);
	return data_type;
}


size_t obi_column_get_line_count_from_name(OBIDMS_p dms, const char* column_name)
{
	OBIDMS_column_header_p 		header;
	OBIDMS_column_directory_p	column_directory;
	char* 						column_file_name;
	int 						column_file_descriptor;
	int							column_dir_file_descriptor;
	size_t 						header_size;
	size_t						line_count;
	obiversion_t	   			version_number;

	// Get the column directory structure associated to the column
	column_directory = obi_open_column_directory(dms, column_name);
	if (column_directory == NULL)
	{
		obidebug(1, "\nError opening a column directory structure");
		return -1;
	}

	// Get the file descriptor associated to the column directory
	column_dir_file_descriptor = dirfd(column_directory->directory);
	if (column_dir_file_descriptor < 0)
	{
		obi_set_errno(OBICOLDIR_UNKNOWN_ERROR);
		obidebug(1, "\nError getting the file descriptor of a column directory");
		obi_close_column_directory(column_directory);
		return -1;
	}

	// Calculate the header size
	header_size = obi_get_platform_header_size();

	// Get the latest version number
	version_number = obi_get_latest_version_number(column_directory);
	if (version_number < 0)
	{
		obidebug(1, "\nError getting the latest version number in a column directory");
		return -1;
	}

	// Get the column file name
	column_file_name = build_column_file_name(column_name, version_number);
	if (column_file_name == NULL)
	{
		return -1;
	}

	// Open the column file (READ-ONLY)
	column_file_descriptor = openat(column_dir_file_descriptor, column_file_name, O_RDONLY);
	if (column_file_descriptor < 0)
	{
		obidebug(1, "\nError opening a column file");
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		free(column_file_name);
		return -1;
	}

	// Fill the header structure
	header = mmap(NULL,
			 header_size,
			 PROT_READ,
			 MAP_SHARED,
			 column_file_descriptor,
			 0
			 );

	if (header == MAP_FAILED)
	{
		obi_set_errno(OBICOL_UNKNOWN_ERROR);
		obidebug(1, "\nError mmapping the header of a column");
		close(column_file_descriptor);
		free(column_file_name);
		return -1;
	}

	// Check endianness?

	line_count = header->line_count;

	free(column_file_name);
	close(column_file_descriptor);
	munmap(header, header_size);
	return line_count;
}


const char* obi_column_get_elements_names(OBIDMS_column_p column)
{
	return (column->header)->elements_names;
}


// TODO to be rewritten in an optimized and safe way
size_t obi_column_get_element_index_from_name(OBIDMS_column_p column, const char* element_name)
{
	char* elements_names;
	char* name;
	size_t element_index;

	elements_names = strdup((column->header)->elements_names);
	if (elements_names == NULL)
	{
		obidebug(1, "\nError strdup-ing the elements names");
		return -1;
	}

	element_index = 0;

	name = strtok (elements_names, ";");	// TODO not thread safe, see strtok_r maybe
	if (strcmp(element_name, name) == 0)
	{
		free(elements_names);
		return element_index;
	}
	element_index++;

	while (name != NULL)
	{
		name = strtok (NULL, ";");			// TODO not thread safe, see strtok_r maybe
		if (strcmp(element_name, name) == 0)
		{
			free(elements_names);
			return element_index;
		}
		element_index++;
	}

	obidebug(1, "\nCan't find an element name");
	free(elements_names);
	return -1;
}


size_t obi_column_get_nb_elements_per_line(OBIDMS_column_p column)
{
	return (column->header)->nb_elements_per_line;
}