Changeset 34919

Timestamp:

Jan 12, 2013, 2:24:56 PM (14 years ago)

Author:

eugene

Message:

add buffered inserts with multiple rows per insert

Location:

branches/eam_branches/ipp-20121219/Ohana/src/dvopsps

Files:

• include/dvopsps.h (modified) (1 diff)
• src/insert_detections_dvopsps.c (modified) (1 diff)
• src/insert_detections_dvopsps_catalog.c (modified) (5 diffs)
• test/dbadmin.sh (modified) (1 diff)

branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/include/dvopsps.h

@@ -43 +43 @@
 int    insert_detections_dvopsps          PROTO(());
 int    insert_detections_dvopsps_parallel PROTO((SkyTable *sky));
-int    insert_detections_mysql             PROTO((Average *average, Measure *measure, MYSQL *mysql));
-int    insert_detections_dvopsps_catalog   PROTO((Catalog *catalog, MYSQL *mysql));
+int    insert_detections_dvopsps_catalog  PROTO((Catalog *catalog, MYSQL *mysql));
       
 MYSQL *mysql_dvopsps_connect              PROTO((MYSQL *mysqlBase));
+
+int    insert_detections_mysql            PROTO((Average *average, Measure *measure, MYSQL *mysql));
+
+int    insert_detections_mysql_commit     PROTO((IOBuffer *buffer, MYSQL *mysql));
+int    insert_detections_mysql_value      PROTO((IOBuffer *buffer, Average *average, Measure *measure));
+int    insert_detections_mysql_init       PROTO((IOBuffer *buffer));

branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps.c

@@ -1 +1 @@
 # include "dvopsps.h"
-# define USE_MYSQL 0
+# define USE_MYSQL 1
 
 // determine the relevant catalogs, launch parallel clients if desired

branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/src/insert_detections_dvopsps_catalog.c

@@ -1 +1 @@
 # include "dvopsps.h"
+
+# define MARKTIME(MSG,...) {                    \
+    float dtime;                                \
+    gettimeofday (&stop, (void *) NULL);        \
+    dtime = DTIME (stop, start);                \
+    fprintf (stderr, MSG, __VA_ARGS__); }
 
 int insert_detections_dvopsps_catalog (Catalog *catalog, MYSQL *mysql) {
@@ -5 +11 @@
   off_t i, j;
   int missingID = 0;
+  struct timeval start, stop;
+
+  IOBuffer buffer;
+  buffer.Nalloc = 0;
 
   Average *average = catalog->average;
@@ -11 +21 @@
   off_t found = 0;
 
+  gettimeofday (&start, (void *) NULL);
+
+  insert_detections_mysql_init (&buffer);
+
+  if (!mysql) {
+    catalog[0].Naverage = 5;
+  }
   for (i = 0; i < catalog[0].Naverage; i++) {
 
@@ -21 +38 @@
 
       // XXX check return status
-      insert_detections_mysql (&average[i], &measure[m+j], mysql);
+      insert_detections_mysql_value (&buffer, &average[i], &measure[m+j]);
+
+      if (buffer.Nalloc > 0x100000) {
+        insert_detections_mysql_commit (&buffer, mysql);
+        if (0) {
+          FlushIOBuffer (&buffer);
+        } else {
+          buffer.Nbuffer = 0;
+          bzero (buffer.buffer, buffer.Nalloc);
+        }
+        insert_detections_mysql_init (&buffer);
+      }
 
       found ++;
     }
   }
+  insert_detections_mysql_commit (&buffer, mysql);
 
-  fprintf (stderr, "inserted "OFF_T_FMT" rows, skipped %d without IDs\n", found, missingID);
+  MARKTIME("-- inserted "OFF_T_FMT" rows in %f sec, skipped %d without IDs\n", found, dtime, missingID);
   return (TRUE);
 }
 
-int insert_detections_mysql (Average *average, Measure *measure, MYSQL *mysql) {
+int insert_detections_mysql_init (IOBuffer *buffer) {
 
-  // XXX length OK?
-  char sql[1024];
+  if (buffer->Nalloc == 0) {
+    InitIOBuffer (buffer, 1024);
+  }
 
+  PrintIOBuffer (buffer, "INSERT INTO dvoDetection (imageID, ippDetectID, detectID, ippObjID, objID, flags, zp, zpErr, airMass, expTime, ra, dec_, raErr, decErr) VALUES \n");
+
+  return TRUE;
+}
+
+int insert_detections_mysql_value (IOBuffer *buffer, Average *average, Measure *measure) {
+
+  // XXX I am changing the def of ippObjID if I use the code below
   PhotCode *code = GetPhotcodebyCode(measure->photcode);
-  sprintf(sql, 
-          "INSERT INTO dvoDetection (imageID, ippDetectID, detectID, ippObjID, objID, flags, zp, zpErr, airMass, expTime, ra, dec_, raErr, decErr) VALUES (%d, %u, %lu, %lu, %lu, %u, %f, %f, %f, %f, %lf, %lf, %f, %f)",
+  PrintIOBuffer (buffer, "    (%d, %u, %lu, %lu, %lu, %u, %f, %f, %f, %f, %lf, %lf, %f, %f),\n",
           measure->imageID,    // imageID
           measure->detID,   // ippDetectID
           measure->extID,   // detectID
-          (uint64_t)average->catID*1000000000 + (uint64_t)average->objID, // ippObjID
+          ((uint64_t)average->catID << 32) + (uint64_t)average->objID, // ippObjID
           average->extID,   // objID
           measure->dbFlags,  // flags
@@ -54 +91 @@
           measure->dYccd * 0.01 * fabs(measure->pltscale) // estimate of decErr
     );
-  
+  return TRUE;
+}
+
+int insert_detections_mysql_commit (IOBuffer *buffer, MYSQL *mysql) {
+
+  // check that the last two chars are ,\n and replace with ;\n
+  if (!strcmp(&buffer->buffer[buffer->Nbuffer-2], ",\n")) {
+    fprintf (stderr, "valid sql\n");
+    buffer->buffer[buffer->Nbuffer-2] = ';';
+  } else {
+    fprintf (stderr, "invalid sql?\n");
+    return FALSE;
+  }
+
   // XXX check return status
   if (mysql) {
-    mysql_query(mysql, sql); 
+    mysql_query(mysql, buffer->buffer); 
   } else {
-    fprintf (stderr, "%s\n", sql);
+    fprintf (stderr, "%s\n", buffer->buffer);
   }
 

branches/eam_branches/ipp-20121219/Ohana/src/dvopsps/test/dbadmin.sh

@@ -47 +47 @@
 endif
 
+if ("$1" == "delete") then
+  if ($#argv != 4) goto usage;
+  set dbhost = $2
+  set dbname = $3
+  set dbuser = $4
+
+  mysql -h $dbhost -u $dbuser -p <<EOF > /dev/null
+   USE $dbname;
+   describe dvoDetection;
+EOF
+
+  if ($status) then
+    echo "database does not contain dvoDetections, not deleting"
+    exit 1;
+  endif
+
+  echo "database is valid, deleting"
+  mysql -h $dbhost -u $dbuser -p <<EOF
+   drop database $dbname;
+EOF
+
+   exit 0;
+endif
+
 usage:
   echo "USAGE: dbadmin.sh (options)"