* GetRequestFromIndexQueue gets the exactly one request corresponding to the collectionId to either for CREATE or REINDEX depending on cmdType. */
| 1566 | * GetRequestFromIndexQueue gets the exactly one request corresponding to the collectionId to either for CREATE or REINDEX depending on cmdType. |
| 1567 | */ |
| 1568 | IndexCmdRequest * |
| 1569 | GetRequestFromIndexQueue(uint64 collectionId, MemoryContext mcxt) |
| 1570 | { |
| 1571 | bool readOnly = false; |
| 1572 | int numValues = 8; |
| 1573 | bool isNull[8] = { 0 }; |
| 1574 | Datum results[8] = { 0 }; |
| 1575 | Oid userOid = InvalidOid; |
| 1576 | |
| 1577 | /** |
| 1578 | * If because of failure scenario, we end up with a index request in "Inprogress" |
| 1579 | * but there is no backend job really executing the request due to failure. |
| 1580 | * We should consider such requests as well to be picked. For such requests, |
| 1581 | * the status will be "Inprogress" but corresponding global_pid will not exist in pg_stat_activity. |
| 1582 | * The order by clause makes sure that we pick IndexCmdStatus_Queued requests first over other (ascending order). |
| 1583 | * |
| 1584 | * SELECT index_cmd, index_id, index_cmd_status, |
| 1585 | * COALESCE(attempt, 0) AS attempt, comment, update_time, user_oid, cmd_type |
| 1586 | * FROM ApiCatalogSchemaName.{ExtensionObjectPrefix}_index_queue iq |
| 1587 | * WHERE cmd_type = '%c' |
| 1588 | * AND iq.collection_id = collectionId |
| 1589 | * AND (index_cmd_status != IndexCmdStatus_Inprogress |
| 1590 | * OR (index_cmd_status = IndexCmdStatus_Inprogress |
| 1591 | * AND iq.global_pid IS NOT NULL |
| 1592 | * AND <distributed_hook_for_pid> NOT IN (SELECT distinct pid FROM pg_stat_activity WHERE pid IS NOT NULL) |
| 1593 | * ) |
| 1594 | * ) |
| 1595 | * ORDER BY cmd_type asc, index_cmd_status ASC LIMIT 1 |
| 1596 | */ |
| 1597 | StringInfo cmdStr = makeStringInfo(); |
| 1598 | appendStringInfo(cmdStr, |
| 1599 | "SELECT index_cmd, index_id, index_cmd_status, COALESCE(attempt, 0) AS attempt, comment, update_time, user_oid, cmd_type "); |
| 1600 | appendStringInfo(cmdStr, |
| 1601 | " FROM %s iq ", GetIndexQueueName()); |
| 1602 | appendStringInfo(cmdStr, " WHERE iq.collection_id = " UINT64_FORMAT, collectionId); |
| 1603 | appendStringInfo(cmdStr, " AND (index_cmd_status NOT IN (%d, %d)", |
| 1604 | IndexCmdStatus_Inprogress, IndexCmdStatus_Skippable); |
| 1605 | appendStringInfo(cmdStr, " OR (index_cmd_status = %d", IndexCmdStatus_Inprogress); |
| 1606 | appendStringInfo(cmdStr, |
| 1607 | " AND iq.global_pid IS NOT NULL AND"); |
| 1608 | |
| 1609 | const char *queryQualForInProgressBuilds = GetPidForIndexBuild(); |
| 1610 | if (queryQualForInProgressBuilds == NULL) |
| 1611 | { |
| 1612 | appendStringInfo(cmdStr, " iq.global_pid"); |
| 1613 | } |
| 1614 | else |
| 1615 | { |
| 1616 | appendStringInfo(cmdStr, "%s", queryQualForInProgressBuilds); |
| 1617 | } |
| 1618 | appendStringInfo(cmdStr, |
| 1619 | " NOT IN (SELECT distinct pid FROM pg_stat_activity WHERE pid IS NOT NULL)"); |
| 1620 | appendStringInfo(cmdStr, " )) "); |
| 1621 | appendStringInfo(cmdStr, |
| 1622 | " ORDER BY cmd_type ASC, index_cmd_status ASC LIMIT 1"); |
| 1623 | |
| 1624 | ExtensionExecuteMultiValueQueryViaSPI(cmdStr->data, readOnly, SPI_OK_SELECT, |
| 1625 | results, |
no test coverage detected