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