IPP Software Navigation Tools IPP Links Communication Pan-STARRS Links

Changeset 38967


Ignore:
Timestamp:
Oct 27, 2015, 10:21:00 AM (11 years ago)
Author:
eugene
Message:

adding fits_to_mysql

Location:
branches/eam_branches/ipp-20150625/Ohana/src/tools
Files:
2 edited

Legend:

Unmodified
Added
Removed

  • branches/eam_branches/ipp-20150625/Ohana/src/tools/Makefile

    r38942 r38967  
    1717PROGRAMS = gconfig fhead pspsconvert ftable fields list_astro glockfile \
    1818radec mktemp precess csystem fits_insert \
    19 medianfilter mefhead ckfits roc random
     19medianfilter mefhead ckfits roc random fits_to_mysql
    2020
    2121all tools: $(PROGRAMS)

  • branches/eam_branches/ipp-20150625/Ohana/src/tools/src/fits_to_mysql.c

    r38966 r38967  
    22# include <gfitsio.h>
    33# include "inttypes.h"
    4 
    5 # define NMAX    0x100
    6 # define NBUFFER 0x100000
    7 # define NROWS 10000
    8 
    9 # define TO_BUFFER 1
    10 
    11 char        *DATABASE_HOST;
    12 char        *DATABASE_USER;
    13 char        *DATABASE_PASS;
    14 char        *DATABASE_NAME;
    15 
    16 int args (int *argc, char **argv);
    17 void usage();
    18 
    19 void print_data (FTable *table);
    20 void *gfits_get_bintable_column_data_by_Ncol (FTable *table, int Nfield, char *type, off_t *Nrow, int *Ncol, char nativeOrder);
    21 void FFPRINT (char *buffer, int *nbuffer, double value);
    22 
    23 int main (int argc, char **argv) {
    24 
    25   args (&argc, argv);
    26   if (argc != 2) usage ();
    27   char *input = argv[1];
    28 
    29   FTable table;
    30   Header header;
    31   table.header = &header;
    32   table.buffer = NULL;
    33 
    34   FILE *f = fopen (input, "r");
    35   if (f == (FILE *) NULL) {
    36     fprintf (stderr, "can't open file %s\n", input);
    37     exit (1);
    38   }
    39 
    40   int Nsection = 0;
    41   while (gfits_fread_header (f, &header)) {
    42 
    43     /* extract the EXTNAME for this component (set to PHU for 0th component) */
    44     char extname[80];
    45     int status = gfits_scan (&header, "EXTNAME", "%s", 1, extname);
    46     if (!status) {
    47       if (Nsection == 0) {
    48         strcpy (extname, "PHU");
    49       } else {
    50         strcpy (extname, "UNKNOWN");
    51       }
    52     }
    53     fprintf (stdout, "%-30s ", extname);
    54 
    55     /* extract the datatype for this component (IMAGE for 0th component) */
    56     char exttype[80];
    57     if (Nsection == 0) {
    58       strcpy (exttype, "IMAGE");
    59     } else {
    60       status = gfits_scan (&header, "XTENSION", "%s", 1, exttype);
    61       if (!status) {
    62         strcpy (exttype, "UNKNOWN");
    63       }
    64     }
    65     fprintf (stdout, "%-15s ", exttype);
    66 
    67     fprintf (stdout, " %4d", header.Naxes);
    68 
    69     /* extract the individual axes */
    70     int i;
    71     for (i = 0; i < header.Naxes; i++) {
    72       fprintf (stdout, " "OFF_T_FMT, header.Naxis[i]);
    73     }
    74     fprintf (stdout, "\n");
    75 
    76     if (strcmp(exttype, "BINTABLE")) {
    77       off_t Nbytes = gfits_data_size (&header);
    78       gfits_free_header (&header);
    79       fseeko (f, Nbytes, SEEK_CUR);
    80       Nsection ++;
    81       continue;
    82     }
    83 
    84     IOBuffer insert;
    85     InitIOBuffer (&insert, 1024);
    86 
    87     create_table (&header. extname, &insert);
    88 
    89     // need to reset this on each pass?
    90     myAssert (!table.buffer, "oops");
    91 
    92     // total number of bytes in the data section (including pad at the end)
    93     off_t Nbytes = gfits_data_size (&header);
    94    
    95     off_t Ntotal = header.Naxis[1];
    96     int Nrows = NROWS;
    97     int start = 0;
    98     while (start < Ntotal) {
    99       if (!gfits_fread_ftable_range (f, FALSE, TRUE, &table, start, Nrows)) {
    100         fprintf (stderr, "error reading table for extension %d\n", Nsection);
    101         exit (1);
    102       }
    103       fprintf (stderr, "read "OFF_T_FMT" rows\n", header.Naxis[1]);
    104 
    105       write_data (&table, &insert);
    106 
    107       start += Nrows;
    108       Nbytes -= header.Naxis[0]*header.Naxis[1]*abs(header.bitpix / 8);
    109       header.Naxis[1] = Ntotal;
    110     }
    111 
    112     gfits_free_table (&table);
    113     gfits_free_header (&header);
    114 
    115     // move the pointer to the end of this data segment
    116     fseeko (f, Nbytes, SEEK_CUR);
    117     Nsection ++;
    118 
    119     FreeIOBuffer (&insert);
    120   }
    121   gfits_free_header (&header);
    122 
    123   fclose (f);
    124 
    125   ohana_memdump (TRUE);
    126 
    127   exit (0);
    128 }
    129 
    130 // the 'insert' buffer will be used to generate the insert lines downstream (call InitIOBuffer on it first)
    131 int create_tables (Header *header, char *extname, IOBuffer *insert) {
    132 
    133   int i, j, Nf;
    134 
    135   // print out the columns (need to swap first)
    136 
    137   int Nfields;
    138   if (!gfits_scan (header, "TFIELDS", "%d", 1, &Nfields)) return FALSE;
    139 
    140   IOBuffer buffer;
    141   InitIOBuffer (&buffer, 1024);
    142 
    143   // drop the table if it exists
    144   PrintIOBuffer (&buffer, "DROP TABLE if exists %s\n", extname);
    145   status = mysql_query(mysql, buffer.buffer);
    146   if (status) {
    147     fprintf (stderr, "failed to drop table:\n");
    148     fprintf (stderr, "%s\n", mysql_error(mysql));
    149     return FALSE;
    150   }
    151   buffer.Nbuffer = 0;
    152   bzero (buffer.buffer, buffer.Nalloc);
    153 
    154   // Only send the necessary fields (eg, do not sent parallax and pm)
    155   PrintIOBuffer (&buffer, "CREATE TABLE %s (\n", extname);
    156   PrintIOBuffer (&buffer, "INSERT INTO %s (\n", extname);
    157 
    158   // first, extract the table data:
    159   for (i = 0; i < Nfields; i++) {
    160 
    161     snprintf (field, 256, "TTYPE%d", i);
    162     gfits_scan (header, field, "%s", 1, tlabel);
    163 
    164     snprintf (field, 256, "TFORM%d", N);
    165     gfits_scan (header, field, "%s", 1, format);
    166 
    167     char type[16];
    168     if (!gfits_bintable_format (format, type, &Nval, &Nbytes)) return (NULL);
    169 
    170     if (!strcmp (type, "byte"))    {
    171       PrintIOBuffer (&buffer, "%s TINYINT", tlabel);
    172     }
    173     if (!strcmp (type, "short"))   {
    174       PrintIOBuffer (&buffer, "%s SMALLINT", tlabel);
    175     }
    176     if (!strcmp (type, "int"))     {
    177       PrintIOBuffer (&buffer, "%s INT", tlabel);
    178     }
    179     if (!strcmp (type, "int64_t")) {
    180       PrintIOBuffer (&buffer, "%s BITINT", tlabel);
    181     }
    182     if (!strcmp (type, "float"))   {
    183       PrintIOBuffer (&buffer, "%s FLOAT", tlabel);
    184     }
    185     if (!strcmp (type, "double"))  {
    186       PrintIOBuffer (&buffer, "%s DOUBLE", tlabel);
    187     }
    188     if (!strcmp (type, "char"))    {
    189       PrintIOBuffer (&buffer, "%s VARCHAR(%d)", tlabel, Nval);
    190     }
    191 
    192     PrintIOBuffer (insert, "%s", tlabel);
    193     if (i == Nfields - 1)  {
    194       PrintIOBuffer (&buffer, ")\n");
    195       PrintIOBuffer (insert, ") VALUES \n");
    196     } else {
    197       PrintIOBuffer (&buffer, ",\n");
    198       PrintIOBuffer (insert, ",\n");
    199     }
    200   }
    201 
    202   if (DEBUG) fprintf (stderr, "%s\n", buffer.buffer);
    203   status = mysql_query(mysql, buffer.buffer);
    204   if (status) {
    205     fprintf (stderr, "failed to create table:\n");
    206     fprintf (stderr, "%s\n", mysql_error(mysql));
    207     fprintf (stderr, "Nbuffer: %d\n", buffer.Nbuffer);
    208   }
    209 
    210   FreeIOBuffer (&buffer);
    211   return TRUE;
    212 }
    213 
    214 typedef enum {
    215   TYPE_NONE, 
    216   TYPE_CHAR, 
    217   TYPE_BYTE, 
    218   TYPE_SHORT,
    219   TYPE_INT,   
    220   TYPE_INT64,
    221   TYPE_FLOAT,
    222   TYPE_DOUBLE
    223 } TypeInt;
    224 
    225 void write_data (FTable *table, IOBuffer *insert) {
    226 
    227   int i, j, Nf;
    228 
    229   // copy the 'insert' IOBuffer to the output buffer
    230   IOBuffer output;
    231   InitIOBuffer (&output, 1024);
    232   IOBufferCopy (&output, &insert);
    233 
    234   // print out the columns (need to swap first)
    235 
    236   Header *header = table->header;
    237   int Ny = table->header->Naxis[1];
    238 
    239   int Nfields;
    240   if (!gfits_scan (header, "TFIELDS", "%d", 1, &Nfields)) return;
    241 
    242   ALLOCATE_PTR (Nrow, off_t,  Nfields);
    243   ALLOCATE_PTR (Ncol, int,    Nfields);
    244   ALLOCATE_PTR (data, void *, Nfields);
    245   ALLOCATE_PTR (type, char *, Nfields);
    246   ALLOCATE_PTR (Type, int,    Nfields);
    247 
    248   // first, extract the table data: (probably should save this structure and re-use in create_table)
    249   for (i = 0; i < Nfields; i++) {
    250     ALLOCATE (type[i], char, 16);
    251     data[i] = gfits_get_bintable_column_data_by_Ncol (table, i + 1, type[i], &Nrow[i], &Ncol[i], FALSE);
    252     myAssert (Nrow[i] == Ny, "oops");
    253     Type[i] = TYPE_NONE;
    254     if (!strcmp (type[i], "char"))    Type[i] = TYPE_CHAR;
    255     if (!strcmp (type[i], "byte"))    Type[i] = TYPE_BYTE;
    256     if (!strcmp (type[i], "short"))   Type[i] = TYPE_SHORT;
    257     if (!strcmp (type[i], "int"))     Type[i] = TYPE_INT;
    258     if (!strcmp (type[i], "int64_t")) Type[i] = TYPE_INT64;
    259     if (!strcmp (type[i], "float"))   Type[i] = TYPE_FLOAT;
    260     if (!strcmp (type[i], "double"))  Type[i] = TYPE_DOUBLE;
    261     myAssert (Type[i] != TYPE_NONE, "oops");
    262   }
    263 
    264   // convert the type characters to an enum for speed here
    265   // could also generate pointers of all types to the data;
    266 
    267   for (i = 0; i < Ny; i++) {
    268     PrintIOBuffer (output, "( ");
    269     for (Nf = 0; Nf < Nfields; Nf++) {
    270       switch (Type[Nf]) {
    271         case TYPE_CHAR: {
    272           char line[1024];
    273           char *value = data[Nf];
    274           memset (line, 0, 128);
    275           memcpy (line, &value[i*Ncol[Nf]], Ncol[Nf]);
    276           PrintIOBuffer (output, "%s", line);
    277           break;
    278         }
    279         case TYPE_BYTE: {
    280           char *value = data[Nf];
    281           for (j = 0; j < Ncol[Nf]; j++) {
    282             PrintIOBuffer (output, "%hhd", value[i*Ncol[Nf] + j]);
    283           }
    284           break;
    285         }
    286         case TYPE_SHORT: {
    287           short *value = data[Nf];
    288           for (j = 0; j < Ncol[Nf]; j++) {
    289             PrintIOBuffer (output, "%hd", value[i*Ncol[Nf] + j]);
    290           }
    291           break;
    292         }
    293         case TYPE_INT: {
    294           int *value = data[Nf];
    295           for (j = 0; j < Ncol[Nf]; j++) {
    296             PrintIOBuffer (output, "%d", value[i*Ncol[Nf] + j]);
    297           }
    298           break;
    299         }
    300         case TYPE_INT64: {
    301           int64_t *value = data[Nf];
    302           for (j = 0; j < Ncol[Nf]; j++) {
    303             PrintIOBuffer (output, "%lld", (long long int) value[i*Ncol[Nf] + j]);
    304           }
    305           break;
    306         }
    307         case TYPE_FLOAT: {
    308           float *value = data[Nf];
    309           for (j = 0; j < Ncol[Nf]; j++) {
    310             PrintIOBuffer (output, "%e", value[i*Ncol[Nf] + j]);
    311           }
    312           break;
    313         }
    314         case TYPE_DOUBLE: {
    315           double *value = data[Nf];
    316           for (j = 0; j < Ncol[Nf]; j++) {
    317             PrintIOBuffer (output, "%e", value[i*Ncol[Nf] + j]);
    318           }
    319           break;
    320         }
    321         default:
    322           myAbort ("impossible");
    323       }
    324       if (Nf < Nfields - 1) {
    325         PrintIOBuffer (output, ",\n");
    326       } else {
    327         PrintIOBuffer (output, ")\n");
    328       }
    329       if (output->Nbuffer > MAX_BUFFER) {
    330       }
    331     }
    332   }
    333 
    334   for (i = 0 ; i < Nfields; i++) {
    335     free (type[i]);
    336     free (data[i]);
    337   }
    338   free (Nrow);
    339   free (Ncol);
    340   free (data);
    341   free (type);
    342   free (Type);
    343 
    344   return;
    345 }
     4# include "mysql.h"
    3465
    3476# define SWAP_BYTE {                                    \
    … …  
    35918    tmp = Pin[3]; Pin[3] = Pin[4]; Pin[4] = tmp; }
    36019
     20# define DEBUG 0
     21# define NMAX    0x100
     22# define NBUFFER 0x100000
     23# define NROWS 10000
     24# define MAX_BUFFER 0x080000
     25
     26# define TO_BUFFER 1
     27
     28typedef enum {
     29  TYPE_NONE, 
     30  TYPE_CHAR, 
     31  TYPE_BYTE, 
     32  TYPE_SHORT,
     33  TYPE_INT,   
     34  TYPE_INT64,
     35  TYPE_FLOAT,
     36  TYPE_DOUBLE
     37} TypeInt;
     38
     39// generate a structure to describe the table:
     40typedef struct {
     41  char *rawname; // name of the column in the fits header
     42  char *colname; // name of the column to be used in the db
     43  char *format;
     44  TypeInt type;
     45  int Nval;
     46  int Nbytes;
     47} ColumnDef;
     48
     49char        *DATABASE_HOST;
     50char        *DATABASE_USER;
     51char        *DATABASE_PASS;
     52char        *DATABASE_NAME;
     53
     54int args (int *argc, char **argv);
     55void usage();
     56
     57void FreeColumnDef (ColumnDef *columns, int Ncolumns);
     58ColumnDef *parse_table (Header *header, int *ncolumn);
     59int create_table (ColumnDef *columns, int Ncolumns, char *extname, IOBuffer *insert, MYSQL *mysql);
     60void write_data (ColumnDef *columns, int Ncolumns, FTable *table, IOBuffer *insert, MYSQL *mysql);
     61
     62void *gfits_get_bintable_column_data_by_Ncol (FTable *table, int Nfield, char *type, off_t *Nrow, int *Ncol, char nativeOrder);
     63int CopyIOBuffer (IOBuffer *output, IOBuffer *input);
     64
     65MYSQL *mysql_connect (MYSQL *mysqlBase);
     66int mysql_commit_buffer (IOBuffer *buffer, MYSQL *mysql);
     67
     68int main (int argc, char **argv) {
     69
     70  args (&argc, argv);
     71  if (argc != 2) usage ();
     72  char *input = argv[1];
     73
     74  MYSQL  mysqlBase;
     75  MYSQL *mysqlReal = mysql_connect (&mysqlBase);
     76  if (!mysqlReal) {
     77    fprintf (stderr, "failed to connect to mysql\n");
     78    exit (1);
     79  }
     80
     81  FTable table;
     82  Header header;
     83  table.header = &header;
     84  table.buffer = NULL;
     85
     86  FILE *f = fopen (input, "r");
     87  if (f == (FILE *) NULL) {
     88    fprintf (stderr, "can't open file %s\n", input);
     89    exit (1);
     90  }
     91
     92  int Nsection = 0;
     93  while (gfits_fread_header (f, &header)) {
     94
     95    /* extract the EXTNAME for this component (set to PHU for 0th component) */
     96    char extname[80];
     97    int status = gfits_scan (&header, "EXTNAME", "%s", 1, extname);
     98    if (!status) {
     99      if (Nsection == 0) {
     100        strcpy (extname, "PHU");
     101      } else {
     102        strcpy (extname, "UNKNOWN");
     103      }
     104    }
     105    fprintf (stdout, "%-30s ", extname);
     106
     107    /* extract the datatype for this component (IMAGE for 0th component) */
     108    char exttype[80];
     109    if (Nsection == 0) {
     110      strcpy (exttype, "IMAGE");
     111    } else {
     112      status = gfits_scan (&header, "XTENSION", "%s", 1, exttype);
     113      if (!status) {
     114        strcpy (exttype, "UNKNOWN");
     115      }
     116    }
     117    fprintf (stdout, "%-15s ", exttype);
     118
     119    fprintf (stdout, " %4d", header.Naxes);
     120
     121    /* extract the individual axes */
     122    int i;
     123    for (i = 0; i < header.Naxes; i++) {
     124      fprintf (stdout, " "OFF_T_FMT, header.Naxis[i]);
     125    }
     126    fprintf (stdout, "\n");
     127
     128    if (strcmp(exttype, "BINTABLE")) {
     129      off_t Nbytes = gfits_data_size (&header);
     130      gfits_free_header (&header);
     131      fseeko (f, Nbytes, SEEK_CUR);
     132      Nsection ++;
     133      continue;
     134    }
     135
     136    int Ncolumns;
     137    ColumnDef *columns = parse_table (&header, &Ncolumns);
     138
     139    IOBuffer insert;
     140    InitIOBuffer (&insert, 1024);
     141
     142    create_table (columns, Ncolumns, extname, &insert, mysqlReal);
     143
     144    // need to reset this on each pass?
     145    myAssert (!table.buffer, "oops");
     146
     147    // total number of bytes in the data section (including pad at the end)
     148    off_t Nbytes = gfits_data_size (&header);
     149   
     150    off_t Ntotal = header.Naxis[1];
     151    int Nrows = NROWS;
     152    int start = 0;
     153    while (start < Ntotal) {
     154      if (!gfits_fread_ftable_range (f, FALSE, TRUE, &table, start, Nrows)) {
     155        fprintf (stderr, "error reading table for extension %d\n", Nsection);
     156        exit (1);
     157      }
     158      fprintf (stderr, "read "OFF_T_FMT" rows\n", header.Naxis[1]);
     159
     160      write_data (columns, Ncolumns, &table, &insert, mysqlReal);
     161
     162      start += Nrows;
     163      Nbytes -= header.Naxis[0]*header.Naxis[1]*abs(header.bitpix / 8);
     164      header.Naxis[1] = Ntotal;
     165    }
     166
     167    gfits_free_table (&table);
     168    gfits_free_header (&header);
     169
     170    // move the pointer to the end of this data segment
     171    fseeko (f, Nbytes, SEEK_CUR);
     172    Nsection ++;
     173
     174    FreeColumnDef (columns, Ncolumns);
     175    FreeIOBuffer (&insert);
     176  }
     177  gfits_free_header (&header);
     178
     179  fclose (f);
     180
     181  FREE (DATABASE_HOST);
     182  FREE (DATABASE_USER);
     183  FREE (DATABASE_PASS);
     184  FREE (DATABASE_NAME);
     185
     186  ohana_memdump (TRUE);
     187
     188  exit (0);
     189}
     190
     191void FreeColumnDef (ColumnDef *columns, int Ncolumns) {
     192
     193  int i;
     194
     195  if (!columns) return;
     196  for (i = 0; i < Ncolumns; i++) {
     197    FREE (columns[i].rawname);
     198    FREE (columns[i].colname);
     199    FREE (columns[i].format);
     200  }
     201  FREE (columns);
     202  return;
     203}
     204
     205ColumnDef *parse_table (Header *header, int *ncolumn) {
     206
     207  int i, j;
     208
     209  int Ncolumn = 0;
     210  int NCOLUMN = 50;
     211  ColumnDef *column = NULL;
     212  ALLOCATE (column, ColumnDef, NCOLUMN);
     213
     214  // print out the columns (need to swap first)
     215
     216  int Nfields;
     217  if (!gfits_scan (header, "TFIELDS", "%d", 1, &Nfields)) return NULL;
     218
     219  // first, extract the table data:
     220  for (i = 0; i < Nfields; i++) {
     221
     222    char field[256], tlabel[256], format[256];
     223    snprintf (field, 256, "TTYPE%d", i+1);
     224    gfits_scan (header, field, "%s", 1, tlabel);
     225
     226    snprintf (field, 256, "TFORM%d", i+1);
     227    gfits_scan (header, field, "%s", 1, format);
     228
     229    // check for duplicates
     230    int Ndup = 0;
     231    for (j = 0; j < Ncolumn; j++) {
     232      if (strcmp (column[j].rawname, tlabel)) continue;
     233      Ndup ++;
     234    }
     235    if (Ndup) {
     236      char tmpname[256];
     237      snprintf (tmpname, 256, "%s_%d", tlabel, Ndup);
     238      column[Ncolumn].colname = strcreate (tmpname);
     239    } else {
     240      column[Ncolumn].colname = strcreate (tlabel);
     241    }
     242    column[Ncolumn].rawname = strcreate (tlabel);
     243    column[Ncolumn].format = strcreate (format);
     244
     245    char type[16];
     246    int Nval, Nbytes;
     247    if (!gfits_bintable_format (format, type, &Nval, &Nbytes)) return NULL;
     248    column[Ncolumn].Nval = Nval;
     249    column[Ncolumn].Nbytes = Nbytes;
     250
     251    // save the type as well
     252    if (!strcmp (type, "byte"))    column[Ncolumn].type = TYPE_BYTE;
     253    if (!strcmp (type, "short"))   column[Ncolumn].type = TYPE_SHORT;
     254    if (!strcmp (type, "int"))     column[Ncolumn].type = TYPE_INT;
     255    if (!strcmp (type, "int64_t")) column[Ncolumn].type = TYPE_INT64;
     256    if (!strcmp (type, "float"))   column[Ncolumn].type = TYPE_FLOAT;
     257    if (!strcmp (type, "double"))  column[Ncolumn].type = TYPE_DOUBLE;
     258    if (!strcmp (type, "char"))    column[Ncolumn].type = TYPE_CHAR;
     259
     260    Ncolumn ++;
     261    if (Ncolumn == NCOLUMN) {
     262      NCOLUMN += 100;
     263      REALLOCATE (column, ColumnDef, NCOLUMN);
     264    }
     265  }
     266  *ncolumn = Ncolumn;
     267  return column;
     268}
     269
     270// the 'insert' buffer will be used to generate the insert lines downstream (call InitIOBuffer on it first)
     271int create_table (ColumnDef *columns, int Ncolumns, char *extname, IOBuffer *insert, MYSQL *mysql) {
     272
     273  int i;
     274
     275  IOBuffer buffer;
     276  InitIOBuffer (&buffer, 1024);
     277
     278  char *tablename = strcreate (extname);
     279
     280  // convert "." in tablename to _
     281  char *p = tablename;
     282  while (p && *p) {
     283    if (*p == '.') *p = '_';
     284    p++;
     285  }
     286
     287  // drop the table if it exists
     288  PrintIOBuffer (&buffer, "DROP TABLE if exists %s\n", tablename);
     289  int status = mysql_query(mysql, buffer.buffer);
     290  if (status) {
     291    fprintf (stderr, "failed to drop table:\n");
     292    fprintf (stderr, "%s\n", mysql_error(mysql));
     293    return FALSE;
     294  }
     295  buffer.Nbuffer = 0;
     296  bzero (buffer.buffer, buffer.Nalloc);
     297
     298  // Only send the necessary fields (eg, do not sent parallax and pm)
     299  PrintIOBuffer (&buffer, "CREATE TABLE %s (\n", tablename);
     300  PrintIOBuffer (insert, "INSERT INTO %s (\n", tablename);
     301
     302  // first, extract the table data:
     303  for (i = 0; i < Ncolumns; i++) {
     304
     305    // watch out for Nval != 1
     306    switch (columns[i].type) {
     307      case TYPE_BYTE:
     308        PrintIOBuffer (&buffer, "%s TINYINT", columns[i].colname);
     309        break;
     310      case TYPE_SHORT:
     311        PrintIOBuffer (&buffer, "%s SMALLINT", columns[i].colname);
     312        break;
     313      case TYPE_INT:
     314        PrintIOBuffer (&buffer, "%s INT", columns[i].colname);
     315        break;
     316      case TYPE_INT64:
     317        PrintIOBuffer (&buffer, "%s BIGINT", columns[i].colname);
     318        break;
     319      case TYPE_FLOAT:
     320        PrintIOBuffer (&buffer, "%s FLOAT", columns[i].colname);
     321        break;
     322      case TYPE_DOUBLE:
     323        PrintIOBuffer (&buffer, "%s DOUBLE", columns[i].colname);
     324        break;
     325      case TYPE_CHAR:
     326        PrintIOBuffer (&buffer, "%s VARCHAR(%d)", columns[i].colname, columns[i].Nval);
     327        break;
     328      default:
     329        myAbort ("oops");
     330    }
     331
     332    PrintIOBuffer (insert, "%s", columns[i].colname);
     333    if (i == Ncolumns - 1)  {
     334      PrintIOBuffer (&buffer, ")\n");
     335      PrintIOBuffer (insert, ") VALUES \n");
     336    } else {
     337      PrintIOBuffer (&buffer, ",\n");
     338      PrintIOBuffer (insert, ",\n");
     339    }
     340  }
     341
     342  if (DEBUG) fprintf (stderr, "%s\n", buffer.buffer);
     343  status = mysql_query(mysql, buffer.buffer);
     344  if (status) {
     345    fprintf (stderr, "failed to create table:\n");
     346    fprintf (stderr, "%s\n", mysql_error(mysql));
     347    fprintf (stderr, "Nbuffer: %d\n", buffer.Nbuffer);
     348  }
     349
     350  FreeIOBuffer (&buffer);
     351  free (tablename);
     352
     353  return TRUE;
     354}
     355
     356void write_data (ColumnDef *columns, int Ncolumns, FTable *table, IOBuffer *insert, MYSQL *mysql) {
     357
     358  int i, j, Nf;
     359
     360  // copy the 'insert' IOBuffer to the output buffer
     361  IOBuffer output;
     362  InitIOBuffer (&output, 1024);
     363  CopyIOBuffer (&output, insert);
     364
     365  int Ny = table->header->Naxis[1];
     366
     367  ALLOCATE_PTR (data, void *, Ncolumns);
     368
     369  // first, extract the table data: (probably should save this structure and re-use in create_table)
     370  for (i = 0; i < Ncolumns; i++) {
     371    off_t Nrow;
     372    int Ncol;
     373    char type[16];
     374    data[i] = gfits_get_bintable_column_data_by_Ncol (table, i + 1, type, &Nrow, &Ncol, FALSE);
     375    myAssert (Nrow == Ny, "oops");
     376  }
     377
     378  // convert the type characters to an enum for speed here
     379  // could also generate pointers of all types to the data;
     380
     381  for (i = 0; i < Ny; i++) {
     382    PrintIOBuffer (&output, "( ");
     383    for (Nf = 0; Nf < Ncolumns; Nf++) {
     384      switch (columns[Nf].type) {
     385        case TYPE_CHAR: {
     386          char line[1024];
     387          char *value = data[Nf];
     388          memset (line, 0, 128);
     389          memcpy (line, &value[i*columns[Nf].Nval], columns[Nf].Nval);
     390          PrintIOBuffer (&output, "%s", line);
     391          break;
     392        }
     393        case TYPE_BYTE: {
     394          char *value = data[Nf];
     395          for (j = 0; j < columns[Nf].Nval; j++) {
     396            PrintIOBuffer (&output, "%hhd", value[i*columns[Nf].Nval + j]);
     397          }
     398          break;
     399        }
     400        case TYPE_SHORT: {
     401          short *value = data[Nf];
     402          for (j = 0; j < columns[Nf].Nval; j++) {
     403            PrintIOBuffer (&output, "%hd", value[i*columns[Nf].Nval + j]);
     404          }
     405          break;
     406        }
     407        case TYPE_INT: {
     408          int *value = data[Nf];
     409          for (j = 0; j < columns[Nf].Nval; j++) {
     410            PrintIOBuffer (&output, "%d", value[i*columns[Nf].Nval + j]);
     411          }
     412          break;
     413        }
     414        case TYPE_INT64: {
     415          int64_t *value = data[Nf];
     416          for (j = 0; j < columns[Nf].Nval; j++) {
     417            PrintIOBuffer (&output, "%lld", (long long int) value[i*columns[Nf].Nval + j]);
     418          }
     419          break;
     420        }
     421        case TYPE_FLOAT: {
     422          float *value = data[Nf];
     423          for (j = 0; j < columns[Nf].Nval; j++) {
     424            PrintIOBuffer (&output, "%e", value[i*columns[Nf].Nval + j]);
     425          }
     426          break;
     427        }
     428        case TYPE_DOUBLE: {
     429          double *value = data[Nf];
     430          for (j = 0; j < columns[Nf].Nval; j++) {
     431            PrintIOBuffer (&output, "%e", value[i*columns[Nf].Nval + j]);
     432          }
     433          break;
     434        }
     435        default:
     436          myAbort ("impossible");
     437      }
     438      if (Nf < Ncolumns - 1) {
     439        PrintIOBuffer (&output, ",\n");
     440      } else {
     441        PrintIOBuffer (&output, "),\n");
     442      }
     443    }
     444    if (output.Nbuffer > MAX_BUFFER) {
     445      if (DEBUG) fprintf (stderr, "%s\n", output.buffer);
     446      mysql_commit_buffer (&output, mysql);
     447      CopyIOBuffer (&output, insert);
     448    }
     449  }
     450
     451  if (DEBUG) fprintf (stderr, "%s\n", output.buffer);
     452  mysql_commit_buffer (&output, mysql);
     453  FreeIOBuffer (&output);
     454 
     455  for (i = 0 ; i < Ncolumns; i++) {
     456    free (data[i]);
     457  }
     458  free (data);
     459  return;
     460}
     461
    361462void *gfits_get_bintable_column_data_by_Ncol (FTable *table, int Nfield, char *type, off_t *Nrow, int *Ncol, char nativeOrder) {
    362463
    … …  
    523624  } else usage();
    524625
     626  return TRUE;
    525627}
    526628
    … …  
    534636  exit (2);
    535637}
     638
     639int dump_result (MYSQL *connection) {
     640
     641  MYSQL_RES *result = mysql_store_result (connection);
     642  if (!result) return FALSE;
     643  int Nrows = mysql_num_rows(result);
     644  int Ncols = mysql_num_fields(result);
     645
     646  int i, j;
     647  for (j = 0; j < Nrows; j++) {
     648    MYSQL_ROW row = mysql_fetch_row(result);
     649    for (i = 0; i < Ncols; i++) {
     650      fprintf (stderr, "%s ", row[i]);
     651    }
     652    fprintf (stderr, "\n");
     653  }
     654  mysql_free_result (result);
     655  return (TRUE);
     656}
     657
     658// DATABASE_* are supplied on the command line and are global variables in dvopsps.h
     659MYSQL *mysql_connect (MYSQL *mysqlBase) {
     660
     661  char query[256];
     662
     663  mysql_init (mysqlBase);
     664  MYSQL *connection = mysql_real_connect (mysqlBase, DATABASE_HOST, DATABASE_USER, DATABASE_PASS, DATABASE_NAME, 0, 0, 0);
     665
     666  if (connection == NULL) {
     667    fprintf (stderr, "failed to connect to database\n");
     668    fprintf (stderr, "%s\n", mysql_error (mysqlBase));
     669    return NULL;
     670  }
     671   
     672  sprintf (query, "set @@interactive_timeout = 30000;");
     673  if (mysql_query (connection, query)) {
     674    fprintf (stderr, "failed to set interactive timout\n");
     675    fprintf (stderr, "%s\n", mysql_error (connection));
     676    return NULL;
     677  }
     678  dump_result (connection);
     679   
     680  sprintf (query, "set @@wait_timeout = 30000;");
     681  if (mysql_query (connection, query)) {
     682    fprintf (stderr, "failed to set wait timout\n");
     683    fprintf (stderr, "%s\n", mysql_error (connection));
     684    return NULL;
     685  }
     686  dump_result (connection);
     687   
     688  // sprintf (query, "set max_allowed_packet = %d;", 2*MAX_BUFFER);
     689  // if (mysql_query (connection, query)) {
     690  //   fprintf (stderr, "failed to set max_allowed_packet\n");
     691  //   fprintf (stderr, "%s\n", mysql_error (connection));
     692  //   return NULL;
     693  // }
     694  // dump_result (connection);
     695   
     696  // sprintf (query, "show variables like 'max_allowed_packet';");
     697  // if (mysql_query (connection, query)) {
     698  //   fprintf (stderr, "failed to set max_allowed_packet\n");
     699  //   fprintf (stderr, "%s\n", mysql_error (connection));
     700  //   return NULL;
     701  // }
     702  // dump_result (connection);
     703   
     704  if (0) {
     705    sprintf (query, "set autocommit=0;");
     706    if (mysql_query (connection, query)) {
     707      fprintf (stderr, "failed to turn off autocommit\n");
     708      fprintf (stderr, "%s\n", mysql_error (connection));
     709      return NULL;
     710    }
     711    dump_result (connection);
     712  }
     713   
     714  return connection;
     715}
     716
     717int mysql_commit_buffer (IOBuffer *buffer, MYSQL *mysql) {
     718
     719  MYSQL_RES *result;
     720
     721  // check that the last two chars are ,\n and replace with ;\n
     722  if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) {
     723    buffer->buffer[buffer->Nbuffer-2] = ';';
     724  } else {
     725    fprintf (stderr, "invalid sql?\n");
     726    return FALSE;
     727  }
     728
     729  // XXX check return status
     730  if (mysql) {
     731    int status;
     732    if (DEBUG) fprintf (stderr, "%s\n", buffer->buffer);
     733    status = mysql_query(mysql, buffer->buffer);
     734    if (status) {
     735      fprintf (stderr, "error with insert:\n");
     736      fprintf (stderr, "%s\n", mysql_error(mysql));
     737      fprintf (stderr, "Nbuffer: %d\n", buffer->Nbuffer);
     738    }
     739    result = mysql_store_result (mysql);
     740    if (result) mysql_free_result (result);
     741  }
     742
     743  return TRUE;
     744}
Note: See TracChangeset for help on using the changeset viewer.