db/migrate/006_remove_segments_helper.cc

   1 #include <mysql.h>
   2 #include <stdint.h>
   3 #include <stdio.h>
   4 #include <stdlib.h>
   5 #include <string.h>
   6 #include <vector>
   7 #include <list>
   8 #include <sstream>
   9
  10 using namespace std;
  11
  12 template <typename T>
  13 static T parse(const char *str) {
  14   istringstream in(str);
  15   T t;
  16   in >> t;
  17   return t;
  18 }
  19
  20 static void exit_mysql_err(MYSQL *mysql) {
  21   const char *err = mysql_error(mysql);
  22   if (err) {
  23     fprintf(stderr, "005_remove_segments_helper: MySQL error: %s\n", err);
  24   } else {
  25     fprintf(stderr, "005_remove_segments_helper: MySQL error\n");
  26   }
  27   abort();
  28   exit(EXIT_FAILURE);
  29 }
  30
  31 static void exit_stmt_err(MYSQL_STMT *stmt) {
  32   const char *err = mysql_stmt_error(stmt);
  33   if (err) {
  34     fprintf(stderr, "005_remove_segments_helper: MySQL stmt error: %s\n", err);
  35   } else {
  36     fprintf(stderr, "005_remove_segments_helper: MySQL stmt error\n");
  37   }
  38   abort();
  39   exit(EXIT_FAILURE);
  40 }
  41
  42 struct segment {
  43   uint32_t from, to;
  44 };
  45
  46 struct data {
  47   MYSQL *mysql, *mysql2;
  48
  49   uint64_t seg_maxid, way_maxid;
  50   uint64_t new_way_id;
  51   uint64_t new_relation_id;
  52
  53   size_t segs_len;
  54   struct segment *segs;
  55   unsigned char *rem_segs;
  56
  57   FILE *ways, *way_nodes, *way_tags,
  58     *relations, *relation_members, *relation_tags;
  59 };
  60
  61 static uint64_t select_u64(MYSQL *mysql, const char *q) {
  62   MYSQL_RES *res;
  63   MYSQL_ROW row;
  64   uint64_t ret;
  65
  66   if (mysql_query(mysql, q))
  67     exit_mysql_err(mysql);
  68
  69   res = mysql_store_result(mysql);
  70   if (!res) exit_mysql_err(mysql);
  71
  72   row = mysql_fetch_row(res);
  73   if (!row) exit_mysql_err(mysql);
  74
  75   if (row[0]) {
  76     ret = parse<uint64_t>(row[0]);
  77   } else {
  78     ret = 0;
  79   }
  80
  81   mysql_free_result(res);
  82
  83   return ret;
  84 }
  85
  86 static void find_maxids(struct data *d) {
  87   d->seg_maxid = select_u64(d->mysql, "SELECT max(id) FROM current_segments");
  88   d->segs_len = d->seg_maxid + 1;
  89   d->way_maxid = select_u64(d->mysql, "SELECT max(id) FROM current_ways");
  90   d->new_way_id = d->way_maxid + 1;
  91   d->new_relation_id = select_u64(d->mysql, "SELECT max(id) FROM current_relations") + 1;
  92 }
  93
  94 static void populate_segs(struct data *d) {
  95   MYSQL_RES *res;
  96   MYSQL_ROW row;
  97   size_t id;
  98
  99   d->segs = (segment *) malloc(sizeof(struct segment) * d->segs_len);
 100   memset(d->segs, 0, sizeof(struct segment) * d->segs_len);
 101
 102   d->rem_segs = (unsigned char *) malloc(d->segs_len);
 103   memset(d->rem_segs, 0, d->segs_len);
 104
 105   if (mysql_query(d->mysql, "SELECT id, node_a, node_b "
 106       "FROM current_segments WHERE visible"))
 107     exit_mysql_err(d->mysql);
 108
 109   res = mysql_use_result(d->mysql);
 110   if (!res) exit_mysql_err(d->mysql);
 111
 112   while (row = mysql_fetch_row(res)) {
 113     id = parse<size_t>(row[0]);
 114     if (id >= d->segs_len) continue;
 115     d->segs[id].from = parse<uint32_t>(row[1]);
 116     d->segs[id].to   = parse<uint32_t>(row[2]);
 117     d->rem_segs[id] = 1;
 118   }
 119   if (mysql_errno(d->mysql)) exit_mysql_err(d->mysql);
 120
 121   mysql_free_result(res);
 122 }
 123
 124 static void write_csv_col(FILE *f, const char *str, char end) {
 125   char *out = (char *) malloc(2 * strlen(str) + 4);
 126   char *o = out;
 127   size_t len;
 128
 129   *(o++) = '\"';
 130   for (; *str; str++) {
 131     if (*str == '\0') {
 132       break;
 133     } else if (*str == '\"') {
 134       *(o++) = '\"';
 135       *(o++) = '\"';
 136     } else {
 137       *(o++) = *str;
 138     }
 139   }
 140   *(o++) = '\"';
 141   *(o++) = end;
 142   *(o++) = '\0';
 143
 144   len = strlen(out);
 145   if (fwrite(out, len, 1, f) != 1) {
 146     perror("fwrite");
 147     exit(EXIT_FAILURE);
 148   }
 149
 150   free(out);
 151 }
 152
 153 static void convert_ways(struct data *d) {
 154   MYSQL_RES *res;
 155   MYSQL_ROW row;
 156   MYSQL_STMT *load_segs, *load_tags;
 157   const char
 158     load_segs_stmt[] = "SELECT segment_id FROM current_way_segments "
 159       "WHERE id = ? ORDER BY sequence_id",
 160     load_tags_stmt[] = "SELECT k, v FROM current_way_tags WHERE id = ?";
 161   char *k, *v;
 162   const size_t max_tag_len = 1 << 16;
 163   long long mysql_id, mysql_seg_id;
 164   unsigned long res_len;
 165   my_bool res_error;
 166   MYSQL_BIND bind[1], seg_bind[1], tag_bind[2];
 167
 168   /* F***ing libmysql only support fixed size buffers for string results of
 169    * prepared statements.  So allocate 65k for the tag key and the tag value
 170    * and hope it'll suffice. */
 171   k = (char *) malloc(max_tag_len);
 172   v = (char *) malloc(max_tag_len);
 173
 174   load_segs = mysql_stmt_init(d->mysql2);
 175   if (!load_segs) exit_mysql_err(d->mysql2);
 176   if (mysql_stmt_prepare(load_segs, load_segs_stmt, sizeof(load_segs_stmt)))
 177     exit_stmt_err(load_segs);
 178
 179   memset(bind, 0, sizeof(bind));
 180   bind[0].buffer_type = MYSQL_TYPE_LONGLONG;
 181   bind[0].buffer = (char *) &mysql_id;
 182   bind[0].is_null = 0;
 183   bind[0].length = 0;
 184   if (mysql_stmt_bind_param(load_segs, bind))
 185     exit_stmt_err(load_segs);
 186
 187   memset(bind, 0, sizeof(seg_bind));
 188   seg_bind[0].buffer_type = MYSQL_TYPE_LONGLONG;
 189   seg_bind[0].buffer = (char *) &mysql_seg_id;
 190   seg_bind[0].is_null = 0;
 191   seg_bind[0].length = 0;
 192   seg_bind[0].error = &res_error;
 193   if (mysql_stmt_bind_result(load_segs, seg_bind))
 194     exit_stmt_err(load_segs);
 195
 196   load_tags = mysql_stmt_init(d->mysql2);
 197   if (!load_tags) exit_mysql_err(d->mysql2);
 198   if (mysql_stmt_prepare(load_tags, load_tags_stmt, sizeof(load_tags_stmt)))
 199     exit_stmt_err(load_tags);
 200
 201   memset(bind, 0, sizeof(bind));
 202   bind[0].buffer_type = MYSQL_TYPE_LONGLONG;
 203   bind[0].buffer = (char *) &mysql_id;
 204   bind[0].is_null = 0;
 205   bind[0].length = 0;
 206
 207   if (mysql_stmt_bind_param(load_tags, bind))
 208     exit_stmt_err(load_tags);
 209
 210   memset(bind, 0, sizeof(tag_bind));
 211   tag_bind[0].buffer_type = MYSQL_TYPE_STRING;
 212   tag_bind[0].buffer = k;
 213   tag_bind[0].is_null = 0;
 214   tag_bind[0].length = &res_len;
 215   tag_bind[0].error = &res_error;
 216   tag_bind[0].buffer_length = max_tag_len;
 217   tag_bind[1].buffer_type = MYSQL_TYPE_STRING;
 218   tag_bind[1].buffer = v;
 219   tag_bind[1].is_null = 0;
 220   tag_bind[1].length = &res_len;
 221   tag_bind[1].error = &res_error;
 222   tag_bind[1].buffer_length = max_tag_len;
 223   if (mysql_stmt_bind_result(load_tags, tag_bind))
 224     exit_stmt_err(load_tags);
 225
 226   if (mysql_query(d->mysql, "SELECT id, user_id, timestamp "
 227       "FROM current_ways WHERE visible"))
 228     exit_mysql_err(d->mysql);
 229
 230   res = mysql_use_result(d->mysql);
 231   if (!res) exit_mysql_err(d->mysql);
 232
 233   while (row = mysql_fetch_row(res)) {
 234     uint64_t id;
 235     const char *user_id, *timestamp;
 236
 237     id = parse<uint64_t>(row[0]);
 238     user_id = row[1];
 239     timestamp = row[2];
 240
 241     mysql_id = (long long) id;
 242
 243     if (mysql_stmt_execute(load_segs))
 244       exit_stmt_err(load_segs);
 245
 246     if (mysql_stmt_store_result(load_segs))
 247       exit_stmt_err(load_segs);
 248
 249     list<segment> segs;
 250     while (!mysql_stmt_fetch(load_segs)) {
 251       if (mysql_seg_id >= d->segs_len) continue;
 252       segs.push_back(d->segs[mysql_seg_id]);
 253       d->rem_segs[mysql_seg_id] = 0;
 254     }
 255
 256     list<list<uint32_t> > node_lists;
 257     while (segs.size()) {
 258       list<uint32_t> node_list;
 259       node_list.push_back(segs.front().from);
 260       node_list.push_back(segs.front().to);
 261       segs.pop_front();
 262       while (true) {
 263         bool found = false;
 264         for (list<segment>::iterator it = segs.begin();
 265             it != segs.end(); ) {
 266           if (it->from == node_list.back()) {
 267             node_list.push_back(it->to);
 268             segs.erase(it++);
 269             found = true;
 270           } else if (it->to == node_list.front()) {
 271             node_list.insert(node_list.begin(), it->from);
 272             segs.erase(it++);
 273             found = true;
 274           } else {
 275             ++it;
 276           }
 277         }
 278         if (!found) break;
 279       }
 280       node_lists.push_back(node_list);
 281     }
 282
 283     vector<uint64_t> ids; ids.reserve(node_lists.size());
 284     bool orig_id_used = false;
 285     for (list<list<uint32_t> >::iterator it = node_lists.begin();
 286         it != node_lists.end(); ++it) {
 287       uint64_t way_id;
 288       int sid;
 289       if (orig_id_used) {
 290         way_id = d->new_way_id++;
 291       } else {
 292         way_id = id;
 293         orig_id_used = true;
 294       }
 295       ids.push_back(way_id);
 296
 297       fprintf(d->ways, "\"%Lu\",", way_id);
 298       write_csv_col(d->ways, user_id, ',');
 299       write_csv_col(d->ways, timestamp, '\n');
 300
 301       sid = 1;
 302       for (list<uint32_t>::iterator nit = it->begin();
 303           nit != it->end(); ++nit) {
 304         fprintf(d->way_nodes, "\"%Lu\",\"%lu\",\"%i\"\n", way_id, *nit, sid++);
 305       }
 306     }
 307
 308     if (mysql_stmt_execute(load_tags))
 309       exit_stmt_err(load_tags);
 310
 311     if (mysql_stmt_store_result(load_tags))
 312       exit_stmt_err(load_tags);
 313
 314     bool multiple_parts = ids.size() > 1,
 315       create_multipolygon = false;
 316
 317     while (!mysql_stmt_fetch(load_tags)) {
 318       if (multiple_parts && !create_multipolygon) {
 319         if (!strcmp(k, "natural")) {
 320           if (strcmp(v, "coastline")) {
 321             create_multipolygon = true;
 322           }
 323         } else if (!strcmp(k, "waterway")) {
 324           if (!strcmp(v, "riverbank")) {
 325             create_multipolygon = true;
 326           }
 327         } else if (!strcmp(k, "leisure") || !strcmp(k, "landuse")
 328             || !strcmp(k, "sport") || !strcmp(k, "amenity")
 329             || !strcmp(k, "tourism") || !strcmp(k, "building")) {
 330           create_multipolygon = true;
 331         }
 332       }
 333
 334       for (vector<uint64_t>::iterator it = ids.begin();
 335           it != ids.end(); ++it) {
 336         fprintf(d->way_tags, "\"%Lu\",", *it);
 337         write_csv_col(d->way_tags, k, ',');
 338         write_csv_col(d->way_tags, v, '\n');
 339       }
 340     }
 341
 342     if (multiple_parts && create_multipolygon) {
 343       uint64_t ent_id = d->new_relation_id++;
 344
 345       fprintf(d->relations, "\"%Lu\",", ent_id);
 346       write_csv_col(d->relations, user_id, ',');
 347       write_csv_col(d->relations, timestamp, '\n');
 348
 349       fprintf(d->relation_tags,
 350         "\"%Lu\",\"type\",\"multipolygon\"\n", ent_id);
 351
 352       for (vector<uint64_t>::iterator it = ids.begin();
 353           it != ids.end(); ++it) {
 354         fprintf(d->relation_members,
 355           "\"%Lu\",\"way\",\"%Lu\",\"\"\n", ent_id, *it);
 356       }
 357     }
 358   }
 359   if (mysql_errno(d->mysql)) exit_stmt_err(load_tags);
 360
 361   mysql_stmt_close(load_segs);
 362   mysql_stmt_close(load_tags);
 363
 364   mysql_free_result(res);
 365   free(k);
 366   free(v);
 367 }
 368
 369 static int read_seg_tags(char **tags, char **k, char **v) {
 370   if (!**tags) return 0;
 371   char *i = strchr(*tags, ';');
 372   if (!i) i = *tags + strlen(*tags);
 373   char *j = strchr(*tags, '=');
 374   *k = *tags;
 375   if (j && j < i) {
 376     *v = j + 1;
 377   } else {
 378     *v = i;
 379   }
 380   *tags = *i ? i + 1 : i;
 381   *i = '\0';
 382   if (j) *j = '\0';
 383   return 1;
 384 }
 385
 386 static void mark_tagged_segs(struct data *d) {
 387   MYSQL_RES *res;
 388   MYSQL_ROW row;
 389
 390   if (mysql_query(d->mysql, "SELECT id, tags FROM current_segments "
 391       "WHERE visible && tags != '' && tags != 'created_by=JOSM'"))
 392     exit_mysql_err(d->mysql);
 393
 394   res = mysql_use_result(d->mysql);
 395   if (!res) exit_mysql_err(d->mysql);
 396
 397   while (row = mysql_fetch_row(res)) {
 398     size_t id = parse<size_t>(row[0]);
 399     if (d->rem_segs[id]) continue;
 400     char *tags_it = row[1], *k, *v;
 401     while (read_seg_tags(&tags_it, &k, &v)) {
 402       if (!strcmp(k, "created_by")) {
 403         d->rem_segs[id] = 1;
 404         break;
 405       }
 406     }
 407   }
 408
 409   mysql_free_result(res);
 410 }
 411
 412 static void convert_remaining_segs(struct data *d) {
 413   MYSQL_STMT *load_seg;
 414   MYSQL_BIND args[1], res[3];
 415   const size_t max_tag_len = 1 << 16;
 416   char *tags, timestamp[100];
 417   char *k, *v;
 418   int user_id;
 419   long long mysql_id;
 420   unsigned long res_len;
 421   my_bool res_error;
 422   const char load_seg_stmt[] =
 423     "SELECT user_id, tags, CAST(timestamp AS CHAR) FROM current_segments "
 424     "WHERE visible && id = ?";
 425
 426   tags = (char *) malloc(max_tag_len);
 427
 428   load_seg = mysql_stmt_init(d->mysql);
 429   if (!load_seg) exit_mysql_err(d->mysql);
 430   if (mysql_stmt_prepare(load_seg, load_seg_stmt, sizeof(load_seg_stmt)))
 431     exit_stmt_err(load_seg);
 432
 433   memset(args, 0, sizeof(args));
 434   args[0].buffer_type = MYSQL_TYPE_LONGLONG;
 435   args[0].buffer = (char *) &mysql_id;
 436   args[0].is_null = 0;
 437   args[0].length = 0;
 438   if (mysql_stmt_bind_param(load_seg, args))
 439     exit_stmt_err(load_seg);
 440
 441   memset(res, 0, sizeof(res));
 442   res[0].buffer_type = MYSQL_TYPE_LONG;
 443   res[0].buffer = (char *) &user_id;
 444   res[0].is_null = 0;
 445   res[0].length = 0;
 446   res[0].error = &res_error;
 447   res[1].buffer_type = MYSQL_TYPE_STRING;
 448   res[1].buffer = tags;
 449   res[1].is_null = 0;
 450   res[1].length = &res_len;
 451   res[1].error = &res_error;
 452   res[1].buffer_length = max_tag_len;
 453   res[2].buffer_type = MYSQL_TYPE_STRING;
 454   res[2].buffer = timestamp;
 455   res[2].is_null = 0;
 456   res[2].length = &res_len;
 457   res[2].error = &res_error;
 458   res[2].buffer_length = sizeof(timestamp);
 459   if (mysql_stmt_bind_result(load_seg, res))
 460     exit_stmt_err(load_seg);
 461
 462   for (size_t seg_id = 0; seg_id < d->segs_len; seg_id++) {
 463     if (!d->rem_segs[seg_id]) continue;
 464     segment seg = d->segs[seg_id];
 465
 466     mysql_id = seg_id;
 467     if (mysql_stmt_execute(load_seg)) exit_stmt_err(load_seg);
 468     if (mysql_stmt_store_result(load_seg)) exit_stmt_err(load_seg);
 469
 470     while (!mysql_stmt_fetch(load_seg)) {
 471       uint64_t way_id = d->new_way_id++;
 472
 473       fprintf(d->ways, "\"%Lu\",\"%i\",", way_id, user_id);
 474       write_csv_col(d->ways, timestamp, '\n');
 475
 476       fprintf(d->way_nodes, "\"%Lu\",\"%lu\",\"%i\"\n", way_id, seg.from, 1);
 477       fprintf(d->way_nodes, "\"%Lu\",\"%lu\",\"%i\"\n", way_id, seg.to, 2);
 478
 479       char *tags_it = tags;
 480       while (read_seg_tags(&tags_it, &k, &v)) {
 481         fprintf(d->way_tags, "\"%Lu\",", way_id);
 482         write_csv_col(d->way_tags, k, ',');
 483         write_csv_col(d->way_tags, v, '\n');
 484       }
 485     }
 486   }
 487
 488   mysql_stmt_close(load_seg);
 489
 490   free(tags);
 491 }
 492
 493 static MYSQL *connect_to_mysql(char **argv) {
 494   MYSQL *mysql = mysql_init(NULL);
 495   if (!mysql) exit_mysql_err(mysql);
 496
 497   if (!mysql_real_connect(mysql, argv[1], argv[2], argv[3], argv[4],
 498       argv[5][0] ? atoi(argv[5]) : 0, argv[6][0] ? argv[6] : NULL, 0))
 499     exit_mysql_err(mysql);
 500
 501   if (mysql_set_character_set(mysql, "utf8"))
 502     exit_mysql_err(mysql);
 503
 504   return mysql;
 505 }
 506
 507 static void open_file(FILE **f, char *fn) {
 508   *f = fopen(fn, "w+");
 509   if (!*f) {
 510     perror("fopen");
 511     exit(EXIT_FAILURE);
 512   }
 513 }
 514
 515 int main(int argc, char **argv) {
 516   struct data data;
 517   struct data *d = &data;
 518   size_t prefix_len;
 519   char *tempfn;
 520
 521   if (argc != 8) {
 522     printf("Usage: 006_remove_segments_helper host user passwd database port socket prefix\n");
 523     exit(EXIT_FAILURE);
 524   }
 525
 526   d->mysql = connect_to_mysql(argv);
 527   d->mysql2 = connect_to_mysql(argv);
 528
 529   prefix_len = strlen(argv[7]);
 530   tempfn = (char *) malloc(prefix_len + 15);
 531   strcpy(tempfn, argv[7]);
 532
 533   strcpy(tempfn + prefix_len, "ways");
 534   open_file(&d->ways, tempfn);
 535
 536   strcpy(tempfn + prefix_len, "way_nodes");
 537   open_file(&d->way_nodes, tempfn);
 538
 539   strcpy(tempfn + prefix_len, "way_tags");
 540   open_file(&d->way_tags, tempfn);
 541
 542   strcpy(tempfn + prefix_len, "relations");
 543   open_file(&d->relations, tempfn);
 544
 545   strcpy(tempfn + prefix_len, "relation_members");
 546   open_file(&d->relation_members, tempfn);
 547
 548   strcpy(tempfn + prefix_len, "relation_tags");
 549   open_file(&d->relation_tags, tempfn);
 550
 551   free(tempfn);
 552
 553   find_maxids(d);
 554   populate_segs(d);
 555   convert_ways(d);
 556   mark_tagged_segs(d);
 557   convert_remaining_segs(d);
 558
 559   mysql_close(d->mysql);
 560   mysql_close(d->mysql2);
 561
 562   fclose(d->ways);
 563   fclose(d->way_nodes);
 564   fclose(d->way_tags);
 565
 566   fclose(d->relations);
 567   fclose(d->relation_members);
 568   fclose(d->relation_tags);
 569
 570   free(d->segs);
 571   free(d->rem_segs);
 572
 573   exit(EXIT_SUCCESS);
 574 }