Fix performance issue, affecting projinfo EPSG:7842

Fixes #1913

AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates() issued
a complex SQL query that pushes the SQLite3 query plan optimizer to its limits.
Was working reasonably with sqlite 3.11, but not with later versions.
So put less constraints in the main query and do post-processing checks and
auxiliary requests to avoid such issues.

For some unknown reason, this slightly slows down a bit execution time of the
whole test_cpp_api binary (~ 10%), but couldn't come with something better,
despite trying many variations of the main SQL query. It seems that in the
general case the non-filter LEFT JOIN on the supersession table helped,
except on this EPSG:7842 case.

3 files changed, 228 insertions, 105 deletions

diff --git a/data/sql/proj_db_table_defs.sql b/data/sql/proj_db_table_defs.sql
index b820b207..f34e1799 100644
--- a/data/sql/proj_db_table_defs.sql
+++ b/data/sql/proj_db_table_defs.sql
@@ -1367,6 +1367,8 @@ CREATE TABLE supersession(
     source TEXT
 );
 
+CREATE INDEX idx_supersession ON supersession(superseded_table_name, superseded_auth_name, superseded_code);
+
 CREATE TRIGGER supersession_insert_trigger
 BEFORE INSERT ON supersession
 FOR EACH ROW BEGIN
diff --git a/src/iso19111/factory.cpp b/src/iso19111/factory.cpp
index 54aa993f..39d48386 100644
--- a/src/iso19111/factory.cpp
+++ b/src/iso19111/factory.cpp
@@ -4280,56 +4280,29 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
 
     // For some reason, filtering on v1.deprecated and v2.deprecated kills
     // performance
-    std::string sqlProlog("SELECT v1.table_name as table1, "
-                          "v1.auth_name AS auth_name1, v1.code AS code1, "
-                          "v1.deprecated AS deprecated1, "
-                          "v2.table_name as table2, "
-                          "v2.auth_name AS auth_name2, v2.code AS code2, "
-                          "v2.deprecated AS deprecated2, "
-                          "a1.south_lat AS south_lat1, "
-                          "a1.west_lon AS west_lon1, "
-                          "a1.north_lat AS north_lat1, "
-                          "a1.east_lon AS east_lon1, "
-                          "a2.south_lat AS south_lat2, "
-                          "a2.west_lon AS west_lon2, "
-                          "a2.north_lat AS north_lat2, "
-                          "a2.east_lon AS east_lon2 ");
-    if (discardSuperseded) {
-        sqlProlog += ", ss1.replacement_auth_name AS replacement_auth_name1, "
-                     "ss1.replacement_code AS replacement_code1, "
-                     "ss2.replacement_auth_name AS replacement_auth_name2, "
-                     "ss2.replacement_code AS replacement_code2 ";
-    }
-    sqlProlog += "FROM coordinate_operation_view v1 "
-                 "JOIN coordinate_operation_view v2 "
-                 "JOIN geodetic_crs g_source "
-                 "JOIN geodetic_crs g_v1s "
-                 "JOIN geodetic_crs g_v1t "
-                 "JOIN geodetic_crs g_v2s "
-                 "JOIN geodetic_crs g_v2t "
-                 "JOIN geodetic_crs g_target "
-                 "ON g_v1s.auth_name = v1.source_crs_auth_name "
-                 "AND g_v1s.code = v1.source_crs_code "
-                 "AND g_v1t.auth_name = v1.target_crs_auth_name "
-                 "AND g_v1t.code = v1.target_crs_code "
-                 "AND g_v2s.auth_name = v2.source_crs_auth_name "
-                 "AND g_v2s.code = v2.source_crs_code "
-                 "AND g_v2t.auth_name = v2.target_crs_auth_name "
-                 "AND g_v2t.code = v2.target_crs_code ";
-
-    const std::string joinSupersession(
-        "LEFT JOIN supersession ss1 ON "
-        "ss1.superseded_table_name = v1.table_name AND "
-        "ss1.superseded_auth_name = v1.auth_name AND "
-        "ss1.superseded_code = v1.code AND "
-        "ss1.superseded_table_name = ss1.replacement_table_name "
-        "LEFT JOIN supersession ss2 ON "
-        "ss2.superseded_table_name = v2.table_name AND "
-        "ss2.superseded_auth_name = v2.auth_name AND "
-        "ss2.superseded_code = v2.code AND "
-        "ss2.superseded_table_name = ss2.replacement_table_name ");
+    const std::string sqlProlog("SELECT v1.table_name as table1, "
+                                "v1.auth_name AS auth_name1, v1.code AS code1, "
+                                "v1.deprecated AS deprecated1, "
+                                "v2.table_name as table2, "
+                                "v2.auth_name AS auth_name2, v2.code AS code2, "
+                                "v2.deprecated AS deprecated2 "
+                                "FROM coordinate_operation_view v1 "
+                                "JOIN coordinate_operation_view v2 "
+                                "JOIN geodetic_crs g_source "
+                                "JOIN geodetic_crs g_v1s "
+                                "JOIN geodetic_crs g_v1t "
+                                "JOIN geodetic_crs g_v2s "
+                                "JOIN geodetic_crs g_v2t "
+                                "JOIN geodetic_crs g_target "
+                                "ON g_v1s.auth_name = v1.source_crs_auth_name "
+                                "AND g_v1s.code = v1.source_crs_code "
+                                "AND g_v1t.auth_name = v1.target_crs_auth_name "
+                                "AND g_v1t.code = v1.target_crs_code "
+                                "AND g_v2s.auth_name = v2.source_crs_auth_name "
+                                "AND g_v2s.code = v2.source_crs_code "
+                                "AND g_v2t.auth_name = v2.target_crs_auth_name "
+                                "AND g_v2t.code = v2.target_crs_code ");
     const std::string joinArea(
-        (discardSuperseded ? joinSupersession : std::string()) +
         "JOIN area a1 ON v1.area_of_use_auth_name = a1.auth_name "
         "AND v1.area_of_use_code = a1.code "
         "JOIN area a2 ON v2.area_of_use_auth_name = a2.auth_name "
@@ -4342,8 +4315,13 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
     additionalWhere +=
         "WHERE g_source.auth_name = ? AND g_source.code = ? "
         "AND g_target.auth_name = ? AND g_target.code = ? "
-        "AND intersects_bbox(south_lat1, west_lon1, north_lat1, east_lon1, "
-        "south_lat2, west_lon2, north_lat2, east_lon2) = 1 ";
+        "AND intersects_bbox("
+        "a1.south_lat, a1.west_lon, a1.north_lat, a1.east_lon, "
+        "a2.south_lat, a2.west_lon, a2.north_lat, a2.east_lon) = 1 ";
+
+#if 0
+    // While those additonal constraints are correct, they are found to
+    // kill performance. So enforce them as post-processing
 
     if (!allowedAuthorities.empty()) {
         additionalWhere += "AND v1.auth_name IN (";
@@ -4358,7 +4336,7 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
                 additionalWhere += ',';
             additionalWhere += '?';
         }
-        additionalWhere += ')';
+        additionalWhere += ") ";
         for (const auto &allowedAuthority : allowedAuthorities) {
             params.emplace_back(allowedAuthority);
         }
@@ -4371,6 +4349,8 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
         params.emplace_back(d->authority());
         params.emplace_back(d->authority());
     }
+#endif
+
     for (const auto &extent : {intersectingExtent1, intersectingExtent2}) {
         if (extent) {
             const auto &geogExtent = extent->geographicElements();
@@ -4386,10 +4366,10 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
                     if (south_lat != -90.0 || west_lon != -180.0 ||
                         north_lat != 90.0 || east_lon != 180.0) {
                         additionalWhere +=
-                            "AND intersects_bbox(south_lat1, "
-                            "west_lon1, north_lat1, east_lon1, ?, ?, ?, ?) AND "
-                            "intersects_bbox(south_lat2, west_lon2, "
-                            "north_lat2, east_lon2, ?, ?, ?, ?) ";
+                            "AND intersects_bbox(a1.south_lat, a1.west_lon, "
+                            "a1.north_lat, a1.east_lon, ?, ?, ?, ?) AND "
+                            "intersects_bbox(a2.south_lat, a2.west_lon, "
+                            "a2.north_lat, a2.east_lon, ?, ?, ?, ?)  ";
                         params.emplace_back(south_lat);
                         params.emplace_back(west_lon);
                         params.emplace_back(north_lat);
@@ -4427,53 +4407,192 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
     auto res = d->run(sql + additionalWhere, params);
     // fprintf(stderr, "after\n");
 
-    const auto filterOutSuperseded = [](SQLResultSet &&resultSet) {
-        std::set<std::pair<std::string, std::string>> setTransf1;
-        std::set<std::pair<std::string, std::string>> setTransf2;
+    const auto filterDeprecatedAndNotMatchingAuth =
+        [&](SQLResultSet &&resultSet) {
+
+            SQLResultSet filteredResultSet;
+            for (const auto &row : resultSet) {
+                const auto &deprecated1 = row[3];
+                const auto &deprecated2 = row[7];
+                if (deprecated1 == "1" || deprecated2 == "1") {
+                    continue;
+                }
+                const auto &auth_name1 = row[1];
+                const auto &auth_name2 = row[5];
+                if (d->hasAuthorityRestriction()) {
+                    if (auth_name1 != d->authority() ||
+                        auth_name2 != d->authority()) {
+                        continue;
+                    }
+                }
+                if (!allowedAuthorities.empty()) {
+                    {
+                        bool found = false;
+                        for (const auto &auth : allowedAuthorities) {
+                            if (auth_name1 == auth) {
+                                found = true;
+                                break;
+                            }
+                        }
+                        if (!found) {
+                            continue;
+                        }
+                    }
+                    {
+                        bool found = false;
+                        for (const auto &auth : allowedAuthorities) {
+                            if (auth_name2 == auth) {
+                                found = true;
+                                break;
+                            }
+                        }
+                        if (!found) {
+                            continue;
+                        }
+                    }
+                }
+
+                filteredResultSet.emplace_back(row);
+            }
+            return filteredResultSet;
+        };
+
+    const auto filterOutSuperseded = [&](SQLResultSet &&resultSet) {
+        std::set<std::pair<std::string, std::string>> setTransf;
+        std::string findSupersededSql("SELECT superseded_table_name, "
+                                      "superseded_auth_name, superseded_code, "
+                                      "replacement_auth_name, replacement_code "
+                                      "FROM supersession WHERE ");
+        bool findSupersededFirstWhere = true;
+        ListOfParams findSupersededParams;
+
+        std::set<std::string> setAlreadyAsked;
+        const auto keyMapSupersession = [](
+            const std::string &table_name, const std::string &auth_name,
+            const std::string &code) { return table_name + auth_name + code; };
+
         for (const auto &row : resultSet) {
-            // table1
+            const auto &table1 = row[0];
             const auto &auth_name1 = row[1];
             const auto &code1 = row[2];
-            const auto &deprecated1 = row[3];
-            // table2
+            const auto key1 = keyMapSupersession(table1, auth_name1, code1);
+            if (setAlreadyAsked.find(key1) == setAlreadyAsked.end()) {
+                setAlreadyAsked.insert(key1);
+                if (!findSupersededFirstWhere)
+                    findSupersededSql += " OR ";
+                findSupersededFirstWhere = false;
+                findSupersededSql +=
+                    "(superseded_table_name = ? AND replacement_table_name = "
+                    "superseded_table_name AND superseded_auth_name = ? AND "
+                    "superseded_code = ?)";
+                findSupersededParams.push_back(table1);
+                findSupersededParams.push_back(auth_name1);
+                findSupersededParams.push_back(code1);
+            }
+
+            const auto &table2 = row[4];
             const auto &auth_name2 = row[5];
             const auto &code2 = row[6];
-            const auto &deprecated2 = row[7];
-            if (deprecated1 == "1" || deprecated2 == "1") {
-                continue;
+            const auto key2 = keyMapSupersession(table2, auth_name2, code2);
+            if (setAlreadyAsked.find(key2) == setAlreadyAsked.end()) {
+                setAlreadyAsked.insert(key2);
+                if (!findSupersededFirstWhere)
+                    findSupersededSql += " OR ";
+                findSupersededFirstWhere = false;
+                findSupersededSql +=
+                    "(superseded_table_name = ? AND replacement_table_name = "
+                    "superseded_table_name AND superseded_auth_name = ? AND "
+                    "superseded_code = ?)";
+                findSupersededParams.push_back(table2);
+                findSupersededParams.push_back(auth_name2);
+                findSupersededParams.push_back(code2);
             }
-            setTransf1.insert(
+
+            setTransf.insert(
                 std::pair<std::string, std::string>(auth_name1, code1));
-            setTransf2.insert(
+            setTransf.insert(
                 std::pair<std::string, std::string>(auth_name2, code2));
         }
+
+        std::map<std::string, std::vector<std::pair<std::string, std::string>>>
+            mapSupersession;
+
+        if (!findSupersededParams.empty()) {
+            const auto resSuperseded =
+                d->run(findSupersededSql, findSupersededParams);
+            for (const auto &row : resSuperseded) {
+                const auto &superseded_table_name = row[0];
+                const auto &superseded_auth_name = row[1];
+                const auto &superseded_code = row[2];
+                const auto &replacement_auth_name = row[3];
+                const auto &replacement_code = row[4];
+                mapSupersession[keyMapSupersession(superseded_table_name,
+                                                   superseded_auth_name,
+                                                   superseded_code)]
+                    .push_back(std::pair<std::string, std::string>(
+                        replacement_auth_name, replacement_code));
+            }
+        }
+
         SQLResultSet filteredResultSet;
         for (const auto &row : resultSet) {
-            const auto &replacement_auth_name1 = row[16];
-            const auto &replacement_code1 = row[17];
-            const auto &replacement_auth_name2 = row[18];
-            const auto &replacement_code2 = row[19];
-            if (!replacement_auth_name1.empty() &&
-                setTransf1.find(std::pair<std::string, std::string>(
-                    replacement_auth_name1, replacement_code1)) !=
-                    setTransf1.end()) {
-                // Skip transformations that are superseded by others that got
-                // returned in the result set.
-                continue;
+            const auto &table1 = row[0];
+            const auto &auth_name1 = row[1];
+            const auto &code1 = row[2];
+            const auto &table2 = row[4];
+            const auto &auth_name2 = row[5];
+            const auto &code2 = row[6];
+
+            auto iter1 = mapSupersession.find(
+                keyMapSupersession(table1, auth_name1, code1));
+            if (iter1 != mapSupersession.end()) {
+                bool foundReplacement = false;
+                for (const auto &replacement : iter1->second) {
+                    const auto &replacement_auth_name = replacement.first;
+                    const auto &replacement_code = replacement.second;
+                    if (setTransf.find(std::pair<std::string, std::string>(
+                            replacement_auth_name, replacement_code)) !=
+                        setTransf.end()) {
+                        // Skip transformations that are superseded by others
+                        // that got
+                        // returned in the result set.
+                        foundReplacement = true;
+                        break;
+                    }
+                }
+                if (foundReplacement) {
+                    continue;
+                }
             }
-            if (!replacement_auth_name2.empty() &&
-                setTransf2.find(std::pair<std::string, std::string>(
-                    replacement_auth_name2, replacement_code2)) !=
-                    setTransf2.end()) {
-                // Skip transformations that are superseded by others that got
-                // returned in the result set.
-                continue;
+
+            auto iter2 = mapSupersession.find(
+                keyMapSupersession(table2, auth_name2, code2));
+            if (iter2 != mapSupersession.end()) {
+                bool foundReplacement = false;
+                for (const auto &replacement : iter2->second) {
+                    const auto &replacement_auth_name = replacement.first;
+                    const auto &replacement_code = replacement.second;
+                    if (setTransf.find(std::pair<std::string, std::string>(
+                            replacement_auth_name, replacement_code)) !=
+                        setTransf.end()) {
+                        // Skip transformations that are superseded by others
+                        // that got
+                        // returned in the result set.
+                        foundReplacement = true;
+                        break;
+                    }
+                }
+                if (foundReplacement) {
+                    continue;
+                }
             }
+
             filteredResultSet.emplace_back(row);
         }
         return filteredResultSet;
     };
 
+    res = filterDeprecatedAndNotMatchingAuth(std::move(res));
     if (discardSuperseded) {
         res = filterOutSuperseded(std::move(res));
     }
@@ -4484,14 +4603,10 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
         const auto &table1 = row[0];
         const auto &auth_name1 = row[1];
         const auto &code1 = row[2];
-        const auto &deprecated1 = row[3];
         const auto &table2 = row[4];
         const auto &auth_name2 = row[5];
         const auto &code2 = row[6];
-        const auto &deprecated2 = row[7];
-        if (deprecated1 == "1" || deprecated2 == "1") {
-            continue;
-        }
+
         auto op1 = d->createFactory(auth_name1)
                        ->createCoordinateOperation(
                            code1, true, usePROJAlternativeGridNames, table1);
@@ -4572,6 +4687,8 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
     // fprintf(stderr, "before %s\n", (sql + additionalWhere).c_str());
     res = d->run(sql + additionalWhere, params);
     // fprintf(stderr, "after\n");
+
+    res = filterDeprecatedAndNotMatchingAuth(std::move(res));
     if (discardSuperseded) {
         res = filterOutSuperseded(std::move(res));
     }
@@ -4579,14 +4696,10 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
         const auto &table1 = row[0];
         const auto &auth_name1 = row[1];
         const auto &code1 = row[2];
-        const auto &deprecated1 = row[3];
         const auto &table2 = row[4];
         const auto &auth_name2 = row[5];
         const auto &code2 = row[6];
-        const auto &deprecated2 = row[7];
-        if (deprecated1 == "1" || deprecated2 == "1") {
-            continue;
-        }
+
         auto op1 = d->createFactory(auth_name1)
                        ->createCoordinateOperation(
                            code1, true, usePROJAlternativeGridNames, table1);
@@ -4667,6 +4780,8 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
     // fprintf(stderr, "before %s\n", (sql + additionalWhere).c_str());
     res = d->run(sql + additionalWhere, params);
     // fprintf(stderr, "after\n");
+
+    res = filterDeprecatedAndNotMatchingAuth(std::move(res));
     if (discardSuperseded) {
         res = filterOutSuperseded(std::move(res));
     }
@@ -4674,14 +4789,10 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
         const auto &table1 = row[0];
         const auto &auth_name1 = row[1];
         const auto &code1 = row[2];
-        const auto &deprecated1 = row[3];
         const auto &table2 = row[4];
         const auto &auth_name2 = row[5];
         const auto &code2 = row[6];
-        const auto &deprecated2 = row[7];
-        if (deprecated1 == "1" || deprecated2 == "1") {
-            continue;
-        }
+
         auto op1 = d->createFactory(auth_name1)
                        ->createCoordinateOperation(
                            code1, true, usePROJAlternativeGridNames, table1);
@@ -4762,6 +4873,8 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
     // fprintf(stderr, "before %s\n", (sql + additionalWhere).c_str());
     res = d->run(sql + additionalWhere, params);
     // fprintf(stderr, "after\n");
+
+    res = filterDeprecatedAndNotMatchingAuth(std::move(res));
     if (discardSuperseded) {
         res = filterOutSuperseded(std::move(res));
     }
@@ -4769,14 +4882,10 @@ AuthorityFactory::createBetweenGeodeticCRSWithDatumBasedIntermediates(
         const auto &table1 = row[0];
         const auto &auth_name1 = row[1];
         const auto &code1 = row[2];
-        const auto &deprecated1 = row[3];
         const auto &table2 = row[4];
         const auto &auth_name2 = row[5];
         const auto &code2 = row[6];
-        const auto &deprecated2 = row[7];
-        if (deprecated1 == "1" || deprecated2 == "1") {
-            continue;
-        }
+
         auto op1 = d->createFactory(auth_name1)
                        ->createCoordinateOperation(
                            code1, true, usePROJAlternativeGridNames, table1);
diff --git a/test/unit/test_crs.cpp b/test/unit/test_crs.cpp
index 87efc59d..66bfbfcd 100644
--- a/test/unit/test_crs.cpp
+++ b/test/unit/test_crs.cpp
@@ -5197,6 +5197,18 @@ TEST(crs, crs_createBoundCRSToWGS84IfPossible) {
             dbContext, CoordinateOperationContext::IntermediateCRSUse::NEVER);
         EXPECT_EQ(bound, crs);
     }
+    {
+        // GDA2020 geocentric
+        auto crs = factory->createCoordinateReferenceSystem("7842");
+        const auto time_before =
+            ::testing::UnitTest::GetInstance()->elapsed_time();
+        crs->createBoundCRSToWGS84IfPossible(
+            dbContext, CoordinateOperationContext::IntermediateCRSUse::
+                           IF_NO_DIRECT_TRANSFORMATION);
+        const auto time_after =
+            ::testing::UnitTest::GetInstance()->elapsed_time();
+        EXPECT_LE(time_after - time_before, 500);
+    }
 }
 
 // ---------------------------------------------------------------------------