]> git.openstreetmap.org Git - nominatim.git/blobdiff - nominatim/index.c
Merge remote-tracking branch 'origin/master' into osmline
[nominatim.git] / nominatim / index.c
index 07c8daba396e8a0edde1f4adb3b071aac53a1151..253b4f20667c8ff22852f4c271f779c381548196 100644 (file)
@@ -1,4 +1,6 @@
 /*
+ * triggers indexing (reparenting etc.) through setting resetting indexed_status: update placex/osmline set indexed_status = 0 where indexed_status > 0
+ * triggers placex_update and osmline_update
 */
 
 #include <stdio.h>
@@ -70,6 +72,16 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         exit(EXIT_FAILURE);
     }
     PQclear(res);
+    
+    res = PQprepare(conn, "index_sectors_osmline",
+                    "select geometry_sector,count(*) from location_property_osmline where indexed_status > 0 group by geometry_sector order by geometry_sector",
+                    0, NULL);
+    if (PQresultStatus(res) != PGRES_COMMAND_OK)
+    {
+        fprintf(stderr, "Failed preparing index_sectors: %s\n", PQerrorMessage(conn));
+        exit(EXIT_FAILURE);
+    }
+    PQclear(res);
 
     pg_prepare_params[0] = PG_OID_INT4;
     res = PQprepare(conn, "index_nosectors",
@@ -104,7 +116,28 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         exit(EXIT_FAILURE);
     }
     PQclear(res);
-
+    
+    pg_prepare_params[0] = PG_OID_INT4;
+    res = PQprepare(conn, "index_sector_places_osmline",
+                    "select place_id from location_property_osmline where geometry_sector = $1 and indexed_status > 0",
+                    1, pg_prepare_params);
+    if (PQresultStatus(res) != PGRES_COMMAND_OK)
+    {
+        fprintf(stderr, "Failed preparing index_sector_places: %s\n", PQerrorMessage(conn));
+        exit(EXIT_FAILURE);
+    }
+    PQclear(res);
+    
+    res = PQprepare(conn, "index_nosector_places_osmline",
+                    "select place_id from location_property_osmline where indexed_status > 0 order by geometry_sector",
+                    0, NULL);
+    if (PQresultStatus(res) != PGRES_COMMAND_OK)
+    {
+        fprintf(stderr, "Failed preparing index_nosector_places: %s\n", PQerrorMessage(conn));
+        exit(EXIT_FAILURE);
+    }
+    PQclear(res);
+    
     // Build the data for each thread
     thread_data = (struct index_thread_data *)malloc(sizeof(struct index_thread_data)*num_threads);
     for (i = 0; i < num_threads; i++)
@@ -116,7 +149,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             exit(EXIT_FAILURE);
         }
 
-        pg_prepare_params[0] = PG_OID_INT4;
+        pg_prepare_params[0] = PG_OID_INT8;
         res = PQprepare(thread_data[i].conn, "index_placex",
                         "update placex set indexed_status = 0 where place_id = $1",
                         1, pg_prepare_params);
@@ -126,14 +159,25 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             exit(EXIT_FAILURE);
         }
         PQclear(res);
+        
+        pg_prepare_params[0] = PG_OID_INT8;
+        res = PQprepare(thread_data[i].conn, "index_osmline",
+                        "update location_property_osmline set indexed_status = 0 where place_id = $1",
+                        1, pg_prepare_params);
+        if (PQresultStatus(res) != PGRES_COMMAND_OK)
+        {
+            fprintf(stderr, "Failed preparing index_osmline: %s\n", PQerrorMessage(conn));
+            exit(EXIT_FAILURE);
+        }
+        PQclear(res);
 
-        res = PQexec(thread_data[i].conn, "set enable_seqscan = false");
+        /*res = PQexec(thread_data[i].conn, "set enable_seqscan = false");
         if (PQresultStatus(res) != PGRES_COMMAND_OK)
         {
             fprintf(stderr, "Failed disabling sequential scan: %s\n", PQerrorMessage(conn));
             exit(EXIT_FAILURE);
         }
-        PQclear(res);
+        PQclear(res);*/
 
         nominatim_exportCreatePreparedQueries(thread_data[i].conn);
     }
@@ -145,11 +189,159 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         writer = nominatim_exportXMLStart(structuredoutputfile);
     }
 
-    fprintf(stderr, "Starting indexing rank (%i to %i) using %i treads\n", rank_min, rank_max, num_threads);
+    fprintf(stderr, "Starting indexing rank (%i to %i) using %i threads\n", rank_min, rank_max, num_threads);
 
+    // first for the placex table
     for (rank = rank_min; rank <= rank_max; rank++)
     {
-        printf("Starting rank %d\n", rank);
+        // OSMLINE: do reindexing (=> reparenting) for interpolation lines at rank 30, but before all other objects of rank 30
+        // reason: houses (rank 30) depend on the updated interpolation line, when reparenting (see placex_update in functions.sql)
+        if (rank == 30)
+        {
+            fprintf(stderr, "Starting indexing interpolation lines (location_property_osmline)\n");
+            rankCountTuples = 0;
+            rankTotalTuples = 0;
+            resSectors = PQexecPrepared(conn, "index_sectors_osmline", 0, NULL, 0, NULL, 1);
+            if (PQresultStatus(resSectors) != PGRES_TUPLES_OK)
+            {
+                fprintf(stderr, "index_sectors_osmline: SELECT failed: %s", PQerrorMessage(conn));
+                PQclear(resSectors);
+                exit(EXIT_FAILURE);
+            }
+            if (PQftype(resSectors, 0) != PG_OID_INT4)
+            {
+                fprintf(stderr, "Sector value has unexpected type\n");
+                PQclear(resSectors);
+                exit(EXIT_FAILURE);
+            }
+            if (PQftype(resSectors, 1) != PG_OID_INT8)
+            {
+                fprintf(stderr, "Sector value has unexpected type\n");
+                PQclear(resSectors);
+                exit(EXIT_FAILURE);
+            }
+            rankStartTime = time(0);
+            for (iSector = 0; iSector < PQntuples(resSectors); iSector++)
+            {
+                rankTotalTuples += PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1)));
+            }
+            // do it only if tuples with indexed_status > 0 were found in osmline
+            int nTuples = PQntuples(resSectors);
+            if (nTuples > 0)
+            {
+                for (iSector = 0; iSector <= nTuples; iSector++)
+                {
+                    if (iSector > 0)
+                    {
+                        resPlaces = PQgetResult(conn);
+                        if (PQresultStatus(resPlaces) != PGRES_TUPLES_OK)
+                        {
+                            fprintf(stderr, "index_sector_places: SELECT failed: %s\n", PQerrorMessage(conn));
+                            PQclear(resPlaces);
+                            exit(EXIT_FAILURE);
+                        }
+                        if (PQftype(resPlaces, 0) != PG_OID_INT8)
+                        {
+                            fprintf(stderr, "Place_id value has unexpected type\n");
+                            PQclear(resPlaces);
+                            exit(EXIT_FAILURE);
+                        }
+                        resNULL = PQgetResult(conn);
+                        if (resNULL != NULL)
+                        {
+                            fprintf(stderr, "Unexpected non-null response\n");
+                            exit(EXIT_FAILURE);
+                        }
+                    }
+
+                    if (iSector < nTuples)
+                    {
+                        sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));
+            //                fprintf(stderr, "\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
+
+                        // Get all the place_id's for this sector
+                        paramSector = PGint32(sector);
+                        paramValues[0] = (char *)&paramSector;
+                        paramLengths[0] = sizeof(paramSector);
+                        paramFormats[0] = 1;
+                        if (rankTotalTuples-rankCountTuples < num_threads*1000)
+                        {
+                            // no sectors
+                            iResult = PQsendQueryPrepared(conn, "index_nosector_places_osmline", 0, NULL, 0, NULL, 1);
+                        }
+                        else
+                        {
+                            iResult = PQsendQueryPrepared(conn, "index_sector_places_osmline", 1, paramValues, paramLengths, paramFormats, 1);
+                        }
+                        if (!iResult)
+                        {
+                            fprintf(stderr, "index_sector_places_osmline: SELECT failed: %s", PQerrorMessage(conn));
+                            PQclear(resPlaces);
+                            exit(EXIT_FAILURE);
+                        }
+                    }
+                    if (iSector > 0)
+                    {
+                        count = 0;
+                        rankPerSecond = 0;
+                        tuples = PQntuples(resPlaces);
+
+                        if (tuples > 0)
+                        {
+                            // Spawn threads
+                            for (i = 0; i < num_threads; i++)
+                            {
+                                thread_data[i].res = resPlaces;
+                                thread_data[i].tuples = tuples;
+                                thread_data[i].count = &count;
+                                thread_data[i].count_mutex = &count_mutex;
+                                thread_data[i].writer = writer;
+                                thread_data[i].writer_mutex = &writer_mutex;
+                                thread_data[i].table = 0; // use osmline table
+                                pthread_create(&thread_data[i].thread, NULL, &nominatim_indexThread, (void *)&thread_data[i]);
+                            }
+                            // Monitor threads to give user feedback
+                            sleepcount = 0;
+                            while (count < tuples)
+                            {
+                                usleep(1000);
+
+                                // Aim for one update per second
+                                if (sleepcount++ > 500)
+                                {
+                                    rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);
+                                    fprintf(stderr, "  Done %i in %i @ %f per second - Interpolation Lines ETA (seconds): %f\n", (rankCountTuples + count), (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - (rankCountTuples + count)))/(float)rankPerSecond);
+                                    sleepcount = 0;
+                                }
+                            }
+
+                            // Wait for everything to finish
+                            for (i = 0; i < num_threads; i++)
+                            {
+                                pthread_join(thread_data[i].thread, NULL);
+                            }
+                            rankCountTuples += tuples;
+                        }
+                        // Finished sector
+                        rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);
+                        fprintf(stderr, "  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
+                        PQclear(resPlaces);
+                    }
+                    if (rankTotalTuples-rankCountTuples < num_threads*20 && iSector < nTuples)
+                    {
+                        iSector = nTuples - 1;
+                    }
+                }
+                PQclear(resSectors);
+            }
+            // Finished rank
+            fprintf(stderr, "\r  Done %i tuples in %i seconds- FINISHED\n", rankCountTuples,(int)(difftime(time(0), rankStartTime)));
+            if (writer)
+            {
+                nominatim_exportXMLEnd(writer);
+            }
+        }
+        fprintf(stderr, "Starting rank %d\n", rank);
         rankCountTuples = 0;
         rankPerSecond = 0;
 
@@ -157,10 +349,11 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         paramValues[0] = (char *)&paramRank;
         paramLengths[0] = sizeof(paramRank);
         paramFormats[0] = 1;
-        if (rank < 16)
-            resSectors = PQexecPrepared(conn, "index_nosectors", 1, paramValues, paramLengths, paramFormats, 1);
-        else
-            resSectors = PQexecPrepared(conn, "index_sectors", 1, paramValues, paramLengths, paramFormats, 1);
+//        if (rank < 16)
+//            resSectors = PQexecPrepared(conn, "index_nosectors", 1, paramValues, paramLengths, paramFormats, 1);
+//        else
+        resSectors = PQexecPrepared(conn, "index_sectors", 1, paramValues, paramLengths, paramFormats, 1);
+
         if (PQresultStatus(resSectors) != PGRES_TUPLES_OK)
         {
             fprintf(stderr, "index_sectors: SELECT failed: %s", PQerrorMessage(conn));
@@ -179,7 +372,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             PQclear(resSectors);
             exit(EXIT_FAILURE);
         }
-
+        
         rankTotalTuples = 0;
         for (iSector = 0; iSector < PQntuples(resSectors); iSector++)
         {
@@ -187,6 +380,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         }
 
         rankStartTime = time(0);
+
         for (iSector = 0; iSector <= PQntuples(resSectors); iSector++)
         {
             if (iSector > 0)
@@ -198,7 +392,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                     PQclear(resPlaces);
                     exit(EXIT_FAILURE);
                 }
-                if (PQftype(resPlaces, 0) != PG_OID_INT4)
+                if (PQftype(resPlaces, 0) != PG_OID_INT8)
                 {
                     fprintf(stderr, "Place_id value has unexpected type\n");
                     PQclear(resPlaces);
@@ -215,7 +409,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             if (iSector < PQntuples(resSectors))
             {
                 sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));
-                //printf("\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
+//                fprintf(stderr, "\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
 
                 // Get all the place_id's for this sector
                 paramRank = PGint32(rank);
@@ -226,10 +420,14 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                 paramValues[1] = (char *)&paramSector;
                 paramLengths[1] = sizeof(paramSector);
                 paramFormats[1] = 1;
-                if (rank < 16)
+                if (rankTotalTuples-rankCountTuples < num_threads*1000)
+                {
                     iResult = PQsendQueryPrepared(conn, "index_nosector_places", 1, paramValues, paramLengths, paramFormats, 1);
+                }
                 else
+                {
                     iResult = PQsendQueryPrepared(conn, "index_sector_places", 2, paramValues, paramLengths, paramFormats, 1);
+                }
                 if (!iResult)
                 {
                     fprintf(stderr, "index_sector_places: SELECT failed: %s", PQerrorMessage(conn));
@@ -255,6 +453,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                         thread_data[i].count_mutex = &count_mutex;
                         thread_data[i].writer = writer;
                         thread_data[i].writer_mutex = &writer_mutex;
+                        thread_data[i].table = 1;  // use placex table
                         pthread_create(&thread_data[i].thread, NULL, &nominatim_indexThread, (void *)&thread_data[i]);
                     }
 
@@ -268,7 +467,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                         if (sleepcount++ > 500)
                         {
                             rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);
-                            printf("  Done %i in %i @ %f per second - Rank %i ETA (seconds): %f\n", (rankCountTuples + count), (int)(difftime(time(0), rankStartTime)), rankPerSecond, rank, ((float)(rankTotalTuples - (rankCountTuples + count)))/rankPerSecond);
+                            fprintf(stderr, "  Done %i in %i @ %f per second - Rank %i ETA (seconds): %f\n", (rankCountTuples + count), (int)(difftime(time(0), rankStartTime)), rankPerSecond, rank, ((float)(rankTotalTuples - (rankCountTuples + count)))/rankPerSecond);
                             sleepcount = 0;
                         }
                     }
@@ -284,35 +483,49 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
 
                 // Finished sector
                 rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);
-                printf("  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
+                fprintf(stderr, "  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
 
                 PQclear(resPlaces);
             }
+            if (rankTotalTuples-rankCountTuples < num_threads*20 && iSector < PQntuples(resSectors))
+            {
+                iSector = PQntuples(resSectors) - 1;
+            }
         }
         // Finished rank
-        printf("\r  Done %i in %i @ %f per second - FINISHED                      \n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);
+        fprintf(stderr, "\r  Done %i in %i @ %f per second - FINISHED                      \n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);
 
         PQclear(resSectors);
     }
+    
 
-    if (writer)
+    if (rank_max == 30)
     {
-        nominatim_exportXMLEnd(writer);
+        // Close all connections
+        for (i = 0; i < num_threads; i++)
+        {
+            PQfinish(thread_data[i].conn);
+        }
+        PQfinish(conn);
     }
 }
 
 void *nominatim_indexThread(void * thread_data_in)
 {
     struct index_thread_data * thread_data = (struct index_thread_data * )thread_data_in;
+    struct export_data querySet;
 
     PGresult   *res;
 
-    const char *paramValues[1];
+    const char  *paramValues[1];
     int         paramLengths[1];
     int         paramFormats[1];
-    uint32_t    paramPlaceID;
-    uint32_t   place_id;
-    time_t             updateStartTime;
+    uint64_t    paramPlaceID;
+    uint64_t    place_id;
+    time_t      updateStartTime;
+    uint        table;
+    
+    table = (uint)(thread_data->table);
 
     while (1)
     {
@@ -323,31 +536,74 @@ void *nominatim_indexThread(void * thread_data_in)
             break;
         }
 
-        place_id = PGint32(*((uint32_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));
+        place_id = PGint64(*((uint64_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));
         (*thread_data->count)++;
 
         pthread_mutex_unlock( thread_data->count_mutex );
 
-        if (verbose) printf("  Processing place_id %d\n", place_id);
+        if (verbose) fprintf(stderr, "  Processing place_id %ld\n", place_id);
 
         updateStartTime = time(0);
-        paramPlaceID = PGint32(place_id);
-        paramValues[0] = (char *)&paramPlaceID;
-        paramLengths[0] = sizeof(paramPlaceID);
-        paramFormats[0] = 1;
-        res = PQexecPrepared(thread_data->conn, "index_placex", 1, paramValues, paramLengths, paramFormats, 1);
-        if (PQresultStatus(res) != PGRES_COMMAND_OK)
+        int done = 0;
+
+        if (thread_data->writer)
         {
-            fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
-            PQclear(res);
-            exit(EXIT_FAILURE);
+             nominatim_exportPlaceQueries(place_id, thread_data->conn, &querySet);
+        }
+
+        while(!done)
+        {
+            paramPlaceID = PGint64(place_id);
+            paramValues[0] = (char *)&paramPlaceID;
+            paramLengths[0] = sizeof(paramPlaceID);
+            paramFormats[0] = 1;
+            if (table == 1) // table=1 for placex
+            {
+                res = PQexecPrepared(thread_data->conn, "index_placex", 1, paramValues, paramLengths, paramFormats, 1);
+            }
+            else // table=0 for osmline
+            {
+                res = PQexecPrepared(thread_data->conn, "index_osmline", 1, paramValues, paramLengths, paramFormats, 1);
+            }
+            if (PQresultStatus(res) == PGRES_COMMAND_OK)
+                done = 1;
+            else
+            {
+                if (!strncmp(PQerrorMessage(thread_data->conn), "ERROR:  deadlock detected", 25))
+                {
+                    if (table == 1)
+                    {
+                        fprintf(stderr, "index_placex: UPDATE failed - deadlock, retrying (%ld)\n", place_id);
+                    }
+                    else
+                    {
+                        fprintf(stderr, "index_osmline: UPDATE failed - deadlock, retrying (%ld)\n", place_id);
+                    }
+                    PQclear(res);
+                    sleep(rand() % 10);
+                }
+                else
+                {
+                    if (table == 1)
+                    {
+                        fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
+                    }
+                    else
+                    {
+                        fprintf(stderr, "index_osmline: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
+                    }
+                    PQclear(res);
+                    exit(EXIT_FAILURE);
+                }
+            }
         }
         PQclear(res);
-        if (difftime(time(0), updateStartTime) > 1) printf("  Slow place_id %d\n", place_id);
+        if (difftime(time(0), updateStartTime) > 1) fprintf(stderr, "  Slow place_id %ld\n", place_id);
 
         if (thread_data->writer)
         {
-            nominatim_exportPlace(place_id, thread_data->conn, thread_data->writer, thread_data->writer_mutex);
+            nominatim_exportPlace(place_id, thread_data->conn, thread_data->writer, thread_data->writer_mutex, &querySet);
+            nominatim_exportFreeQueries(&querySet);
         }
     }