<?php
/**
* SeekQuarry/Yioop --
* Open Source Pure PHP Search Engine, Crawler, and Indexer
*
* Copyright (C) 2009 - 2026 Chris Pollett chris@pollett.org
*
* LICENSE:
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
* END LICENSE
*
* @author Chris Pollett chris@pollett.org
* @license https://www.gnu.org/licenses/ GPL3
* @link https://www.seekquarry.com/
* @copyright 2009 - 2026
* @filesource
*/
namespace seekquarry\yioop\library\storage_formats;
use seekquarry\yioop\library as L;
use seekquarry\yioop\configs as C;
use seekquarry\yioop\library\CrawlConstants;
use seekquarry\yioop\library\UrlParser;
use seekquarry\yioop\library\data_structures\LSMTree;
use seekquarry\yioop\library\data_structures\PackedTableTools;
use seekquarry\yioop\library\data_structures\UndoLog;
use seekquarry\yioop\library\language_processing\PhraseParser;
/**
* Used for crawlLog, crawlHash, and garbageCollect
*/
require_once __DIR__ . '/../Utility.php';
/**
* IndexDocumentBundle holds a crawl's web pages and an inverted index of
* the terms in them, so a search can find the pages that hold a word.
* The pages are kept in a PartitionDocumentBundle; each partition has its
* own inverted index files, which are merged into a dictionary LSMTree.
*
* @author Chris Pollett
*/
class IndexDocumentBundle implements CrawlConstants
{
/**
* $temp_folder is the folder the signal totals a process works out
* are saved in, so a later process reads them, Yioop's temporary
* folder. IndexDocumentBundleTest and WordIteratorTest set it to a
* folder in the tests folder.
* @var string
*/
public static $temp_folder = C\TEMP_DIR;
/**
* REDIRECT_HOST_WORDS is the field of a page, while its partition is
* built, that holds the host words of the home pages that redirect to
* it, such as coke for https://www.coca-cola.com/. invertOneSite adds
* them to the page's own host words.
*/
const REDIRECT_HOST_WORDS = "REDIRECT_HOST_WORDS";
/**
* File name used to store within the folder of the IndexDocumentBundle
* parameter/configuration information about the bundle
*/
const ARCHIVE_INFO_FILE = "archive_info.txt";
/**
* The version of this IndexDocumentBundle. The lowest format number is
* 3.0 as prior inverted index/document stores used IndexArchiveBundle's
*/
const DEFAULT_VERSION = "4.0";
/**
* Default values for the configuration parameters of an
* IndexDocumentBundle
*/
const DEFAULT_PARAMETERS = ["DESCRIPTION" => "",
"VERSION" => self::DEFAULT_VERSION
];
/**
* Subfolder of IndexDocumentBundle to store the LSMTree with
* term => posting list information (i.e., the inverted index)
*/
const DICTIONARY_FOLDER = "Dictionary";
/**
* DocIds are made of three parts: hash of url, hash of document, hash
* of url hostname. Each of these hashes is DOCID_PART_LEN long
*/
const DOCID_PART_LEN = 8;
/**
* Length of DocIds used by this IndexDocumentBundle
*/
const DOCID_LEN = 24;
/**
* Length of the term ids used by this IndexDocumentBundle
*/
const TERMID_LEN = 16;
/**
* Length of terms' bloom filter string in bytes
*/
const TERMSFILTER_LEN = 125;
/**
* Number of terms from a doc to store in term filter (this
* would typically be the top terms according to soem metric)
*/
const NUM_TERMS_FILTER = 300;
/**
* Length of terms' bloom filter string in bits
*/
const TERMSFILTER_BITS_LEN = 1000;
/**
* Number of hash functions to use while constructing the
* terms' bloom filter string
*/
const TERMSFILTER_HASHFN_COUNT = 3;
/**
* Partition i in an IndexDocumentBundle has a subfolder i
* within self::POSITIONS_DOC_MAP_FOLDER. Within this subfolder i,
* self::DOC_MAP_FILENAME is the name of the file used to store the
* document map for the partition. The document map consists of a sequence
* of records associated with each doc_id of a document stored in the
* partition. The first record is ["POS" => $num_words,
* "SCORE" => floatval($global_score_for_document)]. The second record is:
* ["POS" => $length_of_title_of_document, "SCORE" =>
* floatval($num_description_scores)]]
* Here a description score is a score for the importance for a section
* of a document. Subsequence records, list [POS => the length of the jth
* section of the document, SCORE => its score].
*/
const DOC_MAP_FILENAME = "doc_map";
/**
* Name of the file in a partition's folder that says where in the
* document map every DOC_MAP_MARK_EVERY-th entry starts. It holds,
* as 4 byte unsigned numbers, the document map's size and number of
* entries, then the byte offset of entries 0, DOC_MAP_MARK_EVERY,
* 2 * DOC_MAP_MARK_EVERY and so on. docMapEntry uses it to read one
* entry without reading the whole document map.
*/
const DOC_MAP_MARKS_FILENAME = "doc_map_marks";
/**
* Folder used to store the partition data of this IndexDocumentBundle
* These will consists of .txt.gz files for each partition which are used
* to store summaries of documents and actual documents (web pages) and
* .ix files which are used to store doc_id and the associated offsets to
* their summary and actual document within the .txt.gz file
*/
const DOCUMENTS_FOLDER = "Documents";
/**
* Name of the last entries file used to help compute difference lists
* for doc_map_index, and position list offsets used in postings for the
* partition. This file is also used to track the total number of
* occurrences of term in a partition
*/
const LAST_ENTRIES_FILENAME = "last_entries";
/**
* The filename of a file that is used to keep track of the integer that
* says what is the next partition with documents that can be added to
* this IndexDocumentBundle's dictionary. I.e., It should be that
* next_partition <= save_partition
*/
const NEXT_PARTITION_FILE = "next_partition.txt";
/**
* BUILD_CLAIM_TIMEOUT is how many seconds a claim file that holds no
* process id may stand before claimIsAbandoned treats its maker as
* stopped. A claim that holds a process id stands while that process
* runs: a partition of about 147,000 documents took 12 to 13 minutes to
* build on a synthetic crawl, longer than this.
*/
const BUILD_CLAIM_TIMEOUT = 600;
/**
* Seconds after which a claim is abandoned even when a process with
* its holder's id is running, since after that long the id may belong
* to another process.
*/
const BUILD_CLAIM_MOST_SECONDS = 21600;
/**
* Most words whose term ids addTermPostingLists keeps; when full, the
* cache is emptied and fills again.
*/
const TERM_ID_CACHE_SIZE = 100000;
/**
* FIELD_TERM_MARK is added to the safe value of a combined key to make
* a word's field key. A page is filed under the field key of a word
* when the word is in the page's host, title or path, the parts of a
* page that earn the host, title and path bonuses. The field keys
* form the first tier of a tiered index: a query for a common word
* reads the short list of pages that name the word in those parts as
* a disjunct of its own, beside the word's full list.
*/
const FIELD_TERM_MARK = " field";
/**
* Names for the files which appear within a partition sub-folder
*/
const PARTITION_FILENAMES = [self::DOC_MAP_FILENAME,
self::LAST_ENTRIES_FILENAME, self::POSITIONS_FILENAME,
self::POSTINGS_FILENAME];
/**
* Name of the file within a partitions positions_doc_maps folder used
* to contain the partition's position list for all terms in partition.
*/
const POSITIONS_FILENAME = "positions";
/**
* Name of the file in a partition's folder that holds, as JSON, how many
* of the partition's documents carry each url signal, as signalCounts
* gives them, so a query can tell how common each signal is
*/
const SIGNALS_FILENAME = "signals";
/**
* Name of the file in a partition's folder that holds the md5 (a 32
* character digest of a file's bytes) of the partition's document
* index as it was when the partition's
* inverted index files were built, so partitionIsBuilt can tell
* whether documents have been added since
*/
const BUILT_FROM_FILENAME = "built_from";
/**
* Name of the file within a partition's positions_doc_maps folder with
* posting information for all terms in that partition. This consists of
* key value pairs term_id => posting records for all documents with that
* term.
*/
const POSTINGS_FILENAME = "postings";
/**
* How many bytes of posting to buffer before writing, when
* addPartitionPostingsDictionary
*/
const POSTINGS_BUFFER_SIZE = 1000000;
/**
* Maximum number of posting slices to cache
*/
const MAX_POSTING_CACHE_ITEMS = 100;
/**
* $max_postings_read is the largest posting string length, in bytes,
* that getPostingsString has read since the server started. The
* periodic memory log reports it so an oversized read driven by a
* corrupt dictionary entry shows up as a large value here.
* @var int
*/
public static $max_postings_read = 0;
/**
* $archive_info holds the settings of this IndexDocumentBundle, such
* as its DESCRIPTION and VERSION, as kept in ARCHIVE_INFO_FILE
* @var array
*/
public $archive_info;
/**
* $dir_name is the folder this IndexDocumentBundle is kept in
* @var string
*/
public $dir_name;
/**
* $description is meant as a short text name for this bundle. The
* constructor keeps the description it is given in $archive_info
* instead, and this class does not set this field.
* @var string
*/
public $description;
/**
* $next_partition_to_add is the number of the next partition whose
* postings are to be added to the dictionary, as read from
* NEXT_PARTITION_FILE when the bundle is opened
* @var int
*/
public $next_partition_to_add;
/**
* $dictionary is the LSMTree that holds, for each term, a row for
* each partition with the term: its document and occurrence counts
* and where its postings are in the partition's postings file
* @var LSMTree
*/
public $dictionary;
/**
* $exact_count_partitions is how many partitions' dictionary rows
* getWordInfo reads and sums exactly when asked for a word's whole
* count. Past it the count is estimated. It starts at
* EXACT_WORD_COUNT_PARTITIONS; a test sets it lower to reach the
* estimate on a small bundle.
* @var int
*/
public $exact_count_partitions = C\EXACT_WORD_COUNT_PARTITIONS;
/**
* $documents is the PartitionDocumentBundle that stores the crawled
* pages and their summaries, keyed by doc id
* @var PartitionDocumentBundle
*/
public $documents;
/**
* $doc_map is, while a partition is built, its document map as a
* packed table string of doc_id => doc_record rows, which
* addScoresDocMap adds to and which is saved as DOC_MAP_FILENAME
* @var string
*/
public $doc_map;
/**
* $doc_map_tools packs, unpacks and finds rows of document maps, both
* $doc_map and the doc_map files of built partitions
* @var PackedTableTools
*/
public $doc_map_tools;
/**
* $last_entries keeps track of the previous values of
* posting quantities so difference lists can be computed. For example,
* previous $doc_map_index, previous position list offset. It also tracks
* the total number of occurrences of a term within a partition.
* @var array
*/
public $last_entries;
/**
* $last_entries_tools packs and unpacks the rows of $last_entries
* @var PackedTableTools
*/
public $last_entries_tools;
/**
* $last_rows holds, during a build, each term's last posting as
* [LAST_INDEX, LAST_OFFSET,
* NUM_OCCURRENCES], kept as plain arrays so a posting does not pack and
* unpack a row; packLastEntries turns them into $last_entries once the
* build's documents are all added.
* @var array
*/
public $last_rows = [];
/**
* $term_id_cache holds term ids made during this process, keyed by
* word and the
* page's language and safety combinations, so addTermPostingLists does
* not hash the same word again for each page it is on.
* @var array
*/
public $term_id_cache = [];
/**
* $merge_step_id is the id of the undo step
* addPartitionPostingsDictionary begins for a
* partition's dictionary merge, finished once the count of merged
* partitions is written; "" when none is open.
* @var string
*/
public $merge_step_id = "";
/**
* $merge_step_paths lists the files and folder the open dictionary
* merge step writes, synced and checkpointed when the step finishes
* @var array
*/
public $merge_step_paths = [];
/**
* $posting_counts holds, during a build, how many entries each term's
* row in $postings holds; joinPostings writes it at the front of the
* row.
* @var array
*/
public $posting_counts = [];
/**
* $unpack_map maps each of the 256 posting codes to the unpack format
* string for its four fields, DOC_MAP_INDEX, FREQUENCY,
* POSITIONS_OFFSET and POSITIONS_LEN, each 1, 2, 4 or 8 bytes long
* @var array
*/
public $unpack_map;
/**
* $unpack_len_map gives, for each posting code, how many bytes the
* format $unpack_map holds for it reads
* @var array
*/
public $unpack_len_map;
/**
* $positions holds, while a partition is built, the encoded position
* lists of each document in turn, and within a document of each of
* its terms, saved as POSITIONS_FILENAME when the build finishes
* @var string
*/
public $positions;
/**
* $postings holds, while a partition is built, each term id's posting
* list records in the partition as a packed string, keyed by term id
* @var array
*/
public $postings;
/**
* $postings_tools packs, unpacks and finds rows of posting lists,
* both $postings and the postings files of built partitions
* @var PackedTableTools
*/
public $postings_tools;
/**
* $doc_map_counter counts the documents added so far to the partition
* being built
* @var int
*/
public $doc_map_counter;
/**
* $partition_signals holds the signal counts of the partition being
* built, summed over its documents
* as signalCounts gives them for each, written beside the partition's
* other files when the build finishes
* @var array
*/
public $partition_signals = [];
/**
* SIGNAL_SHARES_CHECK_SECONDS is how often, in seconds, signalShares
* lists the bundle's partition folders to see whether one was added.
* A search calls signalShares for every partition block it scores, and
* a large bundle on a slow disk took longer to list than a worker's
* whole time for a request.
*/
const SIGNAL_SHARES_CHECK_SECONDS = 1;
/**
* $signal_shares_cache holds, for each bundle folder signalShares has
* read, the number of partitions whose counts it has added, the totals
* of those counts, when it last looked, and the shares it worked out,
* so each partition's counts are read once
* @var array
*/
public static $signal_shares_cache = [];
/**
* $doc_map_marks_cache holds, by document map path, the marks
* docMapMarks read: [document map size, number of entries, offsets],
* so a search reads each partition's doc_map_marks file once
* @var array
*/
public static $doc_map_marks_cache = [];
/**
* Most document maps whose marks $doc_map_marks_cache keeps; when
* full, the one read longest ago is dropped
*/
const MAX_DOC_MAP_MARKS_CACHED = 1000;
/**
* $doc_map_handles holds, by document map path, an open handle on the
* map and the map's size when it was opened, so docMapEntry opens each
* map once rather than for every entry it reads
* @var array
*/
public static $doc_map_handles = [];
/**
* $doc_map_entries holds the document map entries docMapEntry has
* read, by map, entry number and map size, so the parts of a union
* query that score the same page read its entry once
* @var array
*/
public static $doc_map_entries = [];
/**
* Most entries $doc_map_entries keeps; when full, the one read
* longest ago is dropped
*/
const MAX_DOC_MAP_ENTRIES = 10000;
/**
* $extract_phrase_time adds up the seconds spent finding phrases
* (sequences of adjacent words) in page descriptions while a partition
* is built, which the build logs when it finishes
* @var float
*/
public $extract_phrase_time;
/**
* __construct opens the IndexDocumentBundle in $dir_name, or makes it
* there when it is opened for writing and does not exist yet.
*
* @param string $dir_name folder name to store this bundle
* @param bool $read_only_archive whether to open archive only for reading
* or reading and writing
* @param string $description a text name/serialized info about this
* IndexDocumentBundle
* @param int $num_docs_per_partition the number of documents to be stored
* in a single partition
* @param int $overflow_threshold maximum total bytes of postings held
* in memory before the bundle flushes them out to a new on-disk
* partition
* @return bool false when no on-disk bundle exists and the
* constructor was invoked in read-only mode; the instance is
* fully initialised on the normal path
*/
public function __construct($dir_name, $read_only_archive = true,
$description = null, $num_docs_per_partition =
C\NUM_DOCS_PER_PARTITION, $overflow_threshold = C\OVERFLOW_THRESHOLD)
{
$this->dir_name = $dir_name;
$is_dir = is_dir($this->dir_name);
if (!$is_dir && !$read_only_archive) {
mkdir($this->dir_name);
} else if (!$is_dir) {
return false;
}
/* A step that a stopped or killed process left part way is undone
before any file it changed is read. Only a process that writes
the bundle does this. A search opens it for reading, and undo
checks the length of every file in the checkpoint, which took
searches past their time limit while a crawl ran. */
if (!$read_only_archive) {
$undo_start = microtime(true);
$undo_log = new UndoLog($this->dir_name);
foreach ($undo_log->undo() as $undone_step) {
L\crawlLog("IndexDocumentBundle undid an unfinished " .
"$undone_step in {$this->dir_name}");
}
foreach ($undo_log->checkpoint_notes as $note) {
L\crawlLog("IndexDocumentBundle checkpoint: $note");
}
/* The check reads every file the checkpoint names, so its
time is logged apart from the rest of the open. */
L\crawlLog("IndexDocumentBundle checked its checkpoint in " .
L\changeInMicrotime($undo_start));
}
$archive_info_path = $this->dir_name . "/" . self::ARCHIVE_INFO_FILE;
$this->archive_info = self::DEFAULT_PARAMETERS;
if (!empty($description)) {
$this->archive_info["DESCRIPTION"] = $description;
}
$just_got_info = false;
if (file_exists($archive_info_path)) {
$this->archive_info = unserialize(file_get_contents(
$archive_info_path));
$just_got_info = true;
}
$this->archive_info['RECORD_COMPRESSOR'] ??=
C\NS_COMPRESSORS . "NonCompressor";
$record_compressor = $this->archive_info['RECORD_COMPRESSOR'];
$this->archive_info['BLOB_COMPRESSOR'] ??=
C\NS_COMPRESSORS . "GzipCompressor";
$blob_compressor = $this->archive_info['BLOB_COMPRESSOR'];
if (!$read_only_archive && !$just_got_info) {
/* A new bundle's partitions are all built by code that stores
combined terms and field terms, so queries may ask for
them. */
$this->archive_info['COMBINED_TERMS'] = true;
$this->archive_info['FIELD_TERMS'] = true;
file_put_contents($archive_info_path,
serialize($this->archive_info));
}
$next_partition_path = $this->dir_name . "/".
self::NEXT_PARTITION_FILE;
if (file_exists($next_partition_path)) {
$this->next_partition_to_add = intval(
file_get_contents($next_partition_path));
} else if (!$read_only_archive) {
$this->next_partition_to_add = 0;
file_put_contents($next_partition_path,
$this->next_partition_to_add);
}
$this->documents = new PartitionDocumentBundle($dir_name . "/" .
self::DOCUMENTS_FOLDER, ["PRIMARY KEY" => [self::DOC_ID,
self::DOCID_LEN],
self::SUMMARY => "SERIAL", self::PAGE => "SERIAL"],
$num_docs_per_partition,
PartitionDocumentBundle::PARTITION_SIZE_THRESHOLD,
$record_compressor, $blob_compressor,
$overflow_threshold);
/* Pages arrive compressed by the fetchers, so what is compressed
here is summaries, at a faster level than pages. */
if ($this->documents->blob_compressor instanceof
\seekquarry\yioop\library\compressors\GzipCompressor) {
$this->documents->blob_compressor->level =
C\SUMMARY_COMPRESSION_LEVEL;
}
if (!$read_only_archive) {
$this->documents->index_cache_size = 1;
/* Only a crawl writes lines about the work it does with
partitions. A search opens the same bundle for reading,
and from the command line those lines would land among a
reader's own output. */
$this->documents->log_partition_work = true;
}
$this->doc_map_tools = new PackedTableTools([
"PRIMARY KEY" => ["DOC_KEYS", self::DOCID_LEN], "POS" => "INT",
"SCORE" => "FLOAT"], $record_compressor);
$this->postings_tools = new PackedTableTools([
"PRIMARY KEY" => ["TERM", self::TERMID_LEN],
"DOC_MAP_INDEX" => "INT", "FREQUENCY" => "INT",
"POSITIONS_OFFSET" => "INT", "POSITIONS_LEN" => "INT"],
$record_compressor);
$unpack_codes = [0 => "C", 1 => "n", 2=> "N", 3 => "J"];
$len_codes = [0 => 1, 1 => 2, 2 => 4, 3 => 8];
for ($i = 0; $i < 4; $i++) {
for ($j = 0; $j < 4; $j++) {
for ($k = 0; $k < 4; $k++) {
for ($m = 0; $m < 4; $m++) {
$this->unpack_map[] =
$unpack_codes[$i] . "DOC_MAP_INDEX/" .
$unpack_codes[$j] . "FREQUENCY/" .
$unpack_codes[$k] . "POSITIONS_OFFSET/" .
$unpack_codes[$m] . "POSITIONS_LEN";
$this->unpack_len_map[] = $len_codes[$i] +
$len_codes[$j] + $len_codes[$k] + $len_codes[$m];
}
}
}
}
$this->last_entries_tools = new PackedTableTools([
"PRIMARY KEY" => ["TERM", 16], "LAST_INDEX" => "INT",
"LAST_OFFSET" => "INT", "NUM_OCCURRENCES" => "INT"],
$record_compressor);
if (!$read_only_archive) {
$this->documents->initCountIfNotExists("VISITED_URLS_COUNT");
}
$this->dictionary = new LSMTree($this->dir_name . "/" .
self::DICTIONARY_FOLDER, ["PRIMARY KEY" => ["TERM", 16],
"PARTITION" => "INT", "NUM_DOCS" => "INT",
"NUM_OCCURRENCES" => "INT", "POSTINGS_OFFSET" => "INT",
"POSTINGS_LEN" => "INT"], read_only: $read_only_archive);
}
/**
* addPages adds the array of $pages to the documents
* PartitionDocumentBundle and adds $visited_urls_count, at most the
* number of new pages stored, to the count of visited urls.
*
* @param array $pages data to store
* @param int $visited_urls_count number to add to the count of visited urls
* (visited urls is a smaller number than the total count of objects
* stored in the index).
* @return bool success or failure of adding the pages
*/
public function addPages($pages, $visited_urls_count)
{
L\crawlLog("Indexer adding pages to document bundle...");
$success = $this->documents->put($pages);
$this->documents->addCount(min($visited_urls_count,
$this->documents->new_keys_put), "VISITED_URLS_COUNT");
return $success;
}
/**
* claimIsAbandoned says whether a build, build slot or merge claim file
* may be taken over. A claim holds the process id of its holder, so it
* stands for as long as that process runs, however long its build
* takes; before, a claim counted as abandoned after BUILD_CLAIM_TIMEOUT,
* and a build longer than that let other fetchers build too. A claim
* with no id, or older than BUILD_CLAIM_MOST_SECONDS, falls back to its
* age. mergeNextPartition, buildClosedPartition and
* buildUnclaimedPartition call it before taking a claim.
*
* @param string $claim_file the claim file
* @return bool whether the claim may be removed and taken
*/
public static function claimIsAbandoned($claim_file)
{
clearstatcache(true, $claim_file);
if (!file_exists($claim_file)) {
return false;
}
$age = time() - filemtime($claim_file);
$holder = intval(@file_get_contents($claim_file));
if ($holder > 0 && $age < self::BUILD_CLAIM_MOST_SECONDS) {
return !L\CrawlDaemon::isProcessAlive($holder);
}
return $age > self::BUILD_CLAIM_TIMEOUT;
}
/**
* partitionIsBuilt says whether a partition's inverted index files
* hold all of its documents, so it may be merged into the dictionary
* as it is. The files must all exist, and the MD5 hash of the
* partition's document index must be the one in its built_from file,
* which the build wrote from the index it read. The Indexer builds the
* partition still being filled, by a periodic save while a crawl is
* young and when a crawl stops, and documents added after such a
* build change the document index, so the partition is built again
* before it is merged. Files built before built_from files were
* written have none and count as built, as they did before.
*
* @param int $partition number of the partition to check
* @return bool whether the partition's inverted index files are there
* and were built from its documents as they are now
*/
public function partitionIsBuilt($partition)
{
$folder = $this->getPartitionBaseFolder($partition);
foreach (self::PARTITION_FILENAMES as $filename) {
clearstatcache(true, "$folder/$filename");
if (!file_exists("$folder/$filename")) {
return false;
}
}
$built_from_file = $folder . "/" . self::BUILT_FROM_FILENAME;
$index_file = $this->documents->getPartitionIndex($partition);
clearstatcache(true, $built_from_file);
clearstatcache(true, $index_file);
if (!file_exists($built_from_file) || !file_exists($index_file)) {
return true;
}
return file_get_contents($built_from_file) === md5_file($index_file);
}
/**
* mergeNextPartition adds the next closed partition of the bundle at
* $bundle_path to its dictionary, if that partition's inverted index
* files are built and nobody else is merging. Fetchers on the queue
* server's machine call it between crawl batches, after trying to build
* a partition, so the crawl's own processes do the dictionary work and
* no DictionaryUpdater is started. Merges go one at a time and in order,
* since each adds to the dictionary's first tier and may cascade through
* the tiers above it; the merge_claim file keeps a second process out,
* and is touched while a long merge runs so it does not look abandoned.
* The Indexer calls it too, while a stopping crawl finishes its closed
* partitions and when no fetcher here keeps up.
*
* @param string $bundle_path folder of the crawl's IndexDocumentBundle
* @return int the partition merged, or -1 if none was
*/
public static function mergeNextPartition($bundle_path)
{
$next_partition = intval(@file_get_contents($bundle_path . "/" .
self::NEXT_PARTITION_FILE));
$parameters = PartitionDocumentBundle::getParameterInfo(
$bundle_path . "/" . self::DOCUMENTS_FOLDER);
$save_partition = intval($parameters["SAVE_PARTITION"] ?? 0);
if ($next_partition >= $save_partition) {
return -1;
}
$reader = new IndexDocumentBundle($bundle_path, true);
if (!$reader->partitionIsBuilt($next_partition)) {
return -1;
}
if (file_exists("$bundle_path/build_claim_$next_partition")) {
return -1;
}
$claim = "$bundle_path/merge_claim";
if (self::claimIsAbandoned($claim)) {
@unlink($claim);
}
$claim_handle = @fopen($claim, "x");
if ($claim_handle === false) {
return -1;
}
fwrite($claim_handle, (string)getmypid());
fclose($claim_handle);
$open_start = microtime(true);
$index_archive = new IndexDocumentBundle($bundle_path, false,
"", C\NUM_DOCS_PER_PARTITION);
L\crawlLog("Opened bundle to merge partition $next_partition in " .
L\changeInMicrotime($open_start));
$index_archive->updateDictionary($claim, false,
"merge_$next_partition");
unlink($claim);
return $next_partition;
}
/**
* buildClosedPartition builds the inverted index files of one closed
* partition of the bundle at $bundle_path that nobody has built or is
* building, and says which partition it built. A partition is closed
* once the crawl has moved on to write the next one, and it waits
* between the partition named in next_partition.txt, which the Indexer
* merges next, and the one being written. A caller claims a partition
* by creating its build_claim_ file, which only one process can create,
* so fetchers and the Indexer on one machine never build the same
* partition. A claim whose builder has stopped is taken over, as
* claimIsAbandoned decides. Fetchers on the queue
* server's machine call this before asking for a crawl batch, and the
* Indexer calls it when no fetcher has taken the next partition.
*
* @param string $bundle_path folder of the crawl's index bundle
* @param int $only_partition the one partition to consider, or -1 to
* take the first closed partition that needs building
* @param int $build_slots how many processes may build partitions of
* this bundle at once, or 0 for no cap. A caller takes a
* build_slot_ file first, which only one process can create, and
* returns -1 when every slot is taken.
* @return int the partition built, or -1 when none needed building or
* no build slot was free
*/
public static function buildClosedPartition($bundle_path,
$only_partition = -1, $build_slots = 0)
{
$slot = "";
for ($i = 0; $i < $build_slots && $slot === ""; $i++) {
$slot_file = "$bundle_path/build_slot_$i";
if (self::claimIsAbandoned($slot_file)) {
@unlink($slot_file);
}
$slot_handle = @fopen($slot_file, "x");
if ($slot_handle !== false) {
fwrite($slot_handle, (string)getmypid());
fclose($slot_handle);
$slot = $slot_file;
}
}
if ($build_slots > 0 && $slot === "") {
return -1;
}
$built = self::buildUnclaimedPartition($bundle_path, $only_partition);
if ($slot !== "") {
@unlink($slot);
}
return $built;
}
/**
* buildUnclaimedPartition does buildClosedPartition's work once any
* build slot is held: it claims the first closed partition that no
* process has built or is building, builds it, and says which.
*
* @param string $bundle_path folder of the crawl's index bundle
* @param int $only_partition the one partition to consider, or -1 to
* take the first closed partition that needs building
* @return int the partition built, or -1 when none needed building
*/
public static function buildUnclaimedPartition($bundle_path,
$only_partition = -1)
{
$next_partition = intval(@file_get_contents($bundle_path . "/" .
self::NEXT_PARTITION_FILE));
$parameters = PartitionDocumentBundle::getParameterInfo(
$bundle_path . "/" . self::DOCUMENTS_FOLDER);
$save_partition = intval($parameters["SAVE_PARTITION"] ?? 0);
$reader = new IndexDocumentBundle($bundle_path, true);
$first = ($only_partition >= 0) ? $only_partition : $next_partition;
$last = ($only_partition >= 0) ? $only_partition :
$save_partition - 1;
for ($partition = $first; $partition <= $last &&
$partition < $save_partition; $partition++) {
$claim = "$bundle_path/build_claim_$partition";
if ($reader->partitionIsBuilt($partition) &&
!file_exists($claim)) {
continue;
}
if (self::claimIsAbandoned($claim)) {
@unlink($claim);
}
$claim_handle = @fopen($claim, "x");
if ($claim_handle === false) {
continue;
}
fwrite($claim_handle, (string)getmypid());
fclose($claim_handle);
$index_archive = new IndexDocumentBundle($bundle_path, false,
"", C\NUM_DOCS_PER_PARTITION);
$index_archive->updateDictionary(null, false,
"rebuild_$partition");
unlink($claim);
return $partition;
}
return -1;
}
/**
* updateDictionary adds the posting list information of the partitions
* from next partition up to save partition to the dictionary LSMTree,
* building any partition's inverted index first where needed. When
* all are added, next partition and save partition are the same.
*
* @param string $taking_too_long_touch a filename of a file to touch
* so its last modified time becomes the current time. In a typical
* Yioop crawl this is done for the CrawlConstants::crawl_status_file
* file to prevent Yioop's web interface from stopping the crawl because
* it has seen no recent progress activity on a crawl.
* @param bool $till_equal is set to true will keep adding each partition
* up till the save partition if set to false, only adds one partition
* @param string $rebuild_or_remerge either the string "rebuild',
* "rebuild_some_number", "merge_some_number", or "remerge". If it
* is "merge_some_number", only that one partition is added to the
* dictionary, from its inverted index files, which are built first if
* any is missing or out of date (see partitionIsBuilt). If it
* "rebuild", it will both recompute partition inverted indexes then
* build a global dictionary from these. If it "remerge", a partition
* inverted index that partitionIsBuilt accepts is directly merged
* into a new global dictionary without recomputing it.
* "rebuild_some_number" just rebuild partition some_number's inverted
* index, but does not merge it into the global dictionary.
* @return bool true if at least one partition's inverted index was
* advanced (built or merged) during the call, false otherwise
*/
public function updateDictionary($taking_too_long_touch = null,
$till_equal = true, $rebuild_or_remerge = "rebuild")
{
$merge_one = false;
if (preg_match("/rebuild\_(\d+)/", $rebuild_or_remerge, $matches)) {
$next_partition = intval($matches[1]);
$rebuild_or_remerge = "partition_only";
L\crawlLog("Rebuilding just partition $next_partition!");
} else if (preg_match("/merge\_(\d+)/", $rebuild_or_remerge,
$matches)) {
/* A merge of one named partition adds only that partition and
does not run on past it. ArcTool starts such a merge only after
the partition's builder has finished, so no builder is writing
its files; if a file is still missing or out of date the
partition is built here before it is merged, rather than
skipped, since a skipped partition would leave its words out
of the dictionary. */
$next_partition = intval($matches[1]);
$rebuild_or_remerge = "remerge";
$till_equal = false;
$merge_one = true;
L\crawlLog("Merging just partition $next_partition!");
} else {
$next_partition = $this->next_partition_to_add;
}
$save_partition = $this->documents->parameters["SAVE_PARTITION"];
$current_num_docs = $this->documents->parameters['ACTIVE_COUNT'];
$max_items_per_partition =
$this->documents->parameters['MAX_ITEMS_PER_FILE'];
if ($rebuild_or_remerge == "rebuild") {
L\crawlLog(
"Current save partition has $current_num_docs documents.");
}
L\crawlLog("Max documents per partition: $max_items_per_partition.");
$memory_limit = L\metricToInt(ini_get("memory_limit"));
$before_usage = memory_get_usage();
L\crawlLog("Indexer Memory limit is $memory_limit. Usage is " .
$before_usage);
$advanced_partition = false;
while ($next_partition < $save_partition && $advanced_partition <=
$till_equal) {
if ($rebuild_or_remerge == "rebuild") {
L\crawlLog("Indexer adding Partition to dictionary...");
L\crawlLog("...because save partition changed");
}
$switch_time = microtime(true);
// Save current shard dictionary to main dictionary
$have_inverted_index_files = false;
if ($rebuild_or_remerge == "remerge") {
$have_inverted_index_files = $this->partitionIsBuilt(
$next_partition);
}
if ($have_inverted_index_files) {
L\crawlLog(
"...Partition $next_partition has all of its inverted".
" index files, not recomputing, just merging.");
} else if ($this->buildInvertedIndexPartition($next_partition,
$taking_too_long_touch) === false) {
break;
}
$num_freed = L\garbageCollect();
/* addPartitionPostingsDictionary returns false when the
partition's postings or last_entries file is missing, as
when another process is building the partition again. The
partition's words are then not in the dictionary, so it is
not recorded as merged. It stays the next partition to
merge, and a later pass adds it. */
if ($rebuild_or_remerge != "partition_only" &&
$this->addPartitionPostingsDictionary($next_partition,
$taking_too_long_touch) === false) {
L\crawlLog("Indexer could not add partition " .
"$next_partition to the dictionary; it stays the next " .
"partition to merge.");
break;
}
L\crawlLog(
"Indexer force running garbage collector after partition".
" advance. This freed $num_freed bytes.");
$after_usage = memory_get_usage();
L\crawlLog(
"Indexer after partition changed memory usage: $after_usage");
L\crawlLog("Switch Partition time:".
L\changeInMicrotime($switch_time));
$next_partition++;
if ($rebuild_or_remerge != "partition_only") {
$finish_start = microtime(true);
/* A build of one partition alone puts nothing into the
dictionary, so it leaves the count of partitions the
dictionary holds as it was. ArcTool runs such builds
beside a merge that is advancing that count. */
L\filePutContentsWhole($this->dir_name . "/" .
self::NEXT_PARTITION_FILE, $next_partition);
L\filePutContentsWhole($this->dir_name . "/" .
self::NEXT_PARTITION_FILE . "-advanced", $next_partition);
if (!empty($this->merge_step_id)) {
$undo_log = new UndoLog($this->dir_name);
/* The tier 0 slot the merge wrote is synced, but
only the partition count files are checkpointed:
tier files never grow, and a tier merge soon moves
them away. */
$undo_log->finish($this->merge_step_id,
$this->merge_step_paths,
array_slice($this->merge_step_paths, 1));
$this->merge_step_id = "";
}
L\crawlLog("Recorded partition " . ($next_partition - 1) .
" as merged in " . L\changeInMicrotime($finish_start));
/* Tiers merge only once the partition's merge is done;
each tier merge is an undo step of its own. A tier
merge removes the tiers it merged, which this bundle's
checkpoint may name, so the checkpoint drops them. */
$tiers_start = microtime(true);
$this->dictionary->mergeTiersSpread();
(new UndoLog($this->dir_name))->recordCheckpoint([]);
L\crawlLog("Merged dictionary tiers and recorded the " .
"checkpoint in " . L\changeInMicrotime($tiers_start));
}
$advanced_partition = true;
}
$this->next_partition_to_add = $next_partition;
return $advanced_partition;
}
/**
* addPartitionPostingsDictionary adds the inverted index already built
* for $partition to the dictionary, the inverted index of the whole
* bundle.
*
* @param int $partition which partitions inverted index to add, by
* default the current save partition
* @param string $taking_too_long_touch a filename of a file to touch
* so its last modified time becomes the current time. In a typical
* Yioop crawl this is done for the CrawlConstants::crawl_status_file
* file to prevent Yioop's web interface from stopping the crawl because
* it has seen no recent progress activity on a crawl.
* @return bool false if there is no eligible partition to add or the
* required postings file is missing; true once the partition has
* been merged into the global dictionary
*/
public function addPartitionPostingsDictionary($partition = -1,
$taking_too_long_touch = null)
{
$save_partition = $this->documents->parameters["SAVE_PARTITION"];
if ($partition < 0 ) {
if ($save_partition <= 0) {
return false;
}
$partition = $save_partition - 1;
}
$base_folder = $this->getPartitionBaseFolder($partition);
$postings_tools = $this->postings_tools;
unset($this->postings, $this->doc_map, $this->positions,
$this->last_entries);
$last_entries_tools = $this->last_entries_tools;
$dictionary = $this->dictionary;
$postings_filename = $base_folder . "/" . self::POSTINGS_FILENAME;
$last_entries_filename = $base_folder . "/" .
self::LAST_ENTRIES_FILENAME;
if (!file_exists($postings_filename)) {
L\crawlLog($postings_filename);
L\crawlLog("Postings file for partition $partition does not exist");
return false;
}
if (!file_exists($last_entries_filename)) {
L\crawlLog(
"Last entries file for partition $partition does not exist");
return false;
}
L\crawlLog("Start Adding Partition Posting Info to Dictionary");
$start_time = microtime(true);
$postings_string = $postings_tools->load($postings_filename,
PackedTableTools::AS_STRING_MODE);
$posting_files_len = strlen($postings_string);
//add a marker for the end of the file as a string
$key_len = $this->postings_tools->key_len;
/* Only each term's NUM_OCCURRENCES is needed, so it is read
straight from each row of the file. A row is the 16 byte term,
the record count, one byte of size codes for the three INT
columns, then LAST_INDEX, LAST_OFFSET and NUM_OCCURRENCES. A
row of any other shape is unpacked in full. */
$occurrences = [];
$unpack_codes = [1 => "n", 2 => "N", 3 => "J"];
foreach (explode("\xFF", $last_entries_tools->load(
$last_entries_filename, PackedTableTools::AS_STRING_MODE)) as
$encoded_row) {
$row = L\decode255($encoded_row);
$codes = ord($row[$key_len + 1] ?? "\x00");
$position = $key_len + 2 + (1 << (($codes >> 6) & 3)) +
(1 << (($codes >> 4) & 3));
$code = ($codes >> 2) & 3;
$term = substr($row, 0, $key_len);
if (($row[$key_len] ?? "") !== "\x01" ||
strlen($row) != $position + (1 << $code)) {
$occurrences[$term] = $last_entries_tools->unpack(
substr($row, $key_len))[0]["NUM_OCCURRENCES"] ?? 0;
continue;
}
$bytes = substr($row, $position);
$first = ord($bytes[0]);
$bytes[0] = chr($first & 127);
$value = ($code == 0) ? ord($bytes) :
unpack($unpack_codes[$code], $bytes)[1];
$occurrences[$term] = ($first > 127) ? -$value : $value;
}
/* A partition that holds only link documents has an empty postings
file, and counting separators in an empty string would still give
one posting, which was put into the dictionary with an empty term
as a record with no key. */
$num_postings = ($postings_string === "") ? 0 :
substr_count($postings_string, "\xFF") + 1;
$last_marker = 0;
$slot = ($dictionary->occupiedTier(0)) ? "B" : "A";
$dictionary->selectPutSlot($slot);
/* The partition's rows and the count of partitions merged change
together: a stop before both are written removes the rows and
puts the count back, so the partition is merged again whole. */
$next_partition_path = $this->dir_name . "/" .
self::NEXT_PARTITION_FILE;
$undo_log = new UndoLog($this->dir_name);
$this->merge_step_paths = [$dictionary->getTierFolder(0) . "/$slot",
$next_partition_path, $next_partition_path . "-advanced"];
$this->merge_step_id = $undo_log->begin(
"dictionary merge of partition $partition", [],
$this->merge_step_paths);
for ($i = 0; $i < $num_postings; $i++) {
$cur_marker = strpos($postings_string, "\xFF", $last_marker);
$diff = ($cur_marker === false) ? null :
$cur_marker - $last_marker;
$pre_row = substr($postings_string, $last_marker, $diff);
$postings_offset = $last_marker + $key_len;
$last_marker = $cur_marker + 1;
$term = substr($pre_row, 0, $key_len);
$encode_row = substr($pre_row, $key_len);
$postings_len = strlen($encode_row);
$row = L\decode255($encode_row);
if (L\crawlTimeoutLog("..Indexer Still processing partition %s. " .
"Have completed %s postings of %s.", $partition, $i,
$num_postings) && $taking_too_long_touch) {
if (file_exists($taking_too_long_touch)) {
touch($taking_too_long_touch);
}
}
$start = 0;
$num_docs_term = L\vByteDecode($row, $start);
$num_occurrences_term = $occurrences[$term] ?? 0;
$dictionary->put(["TERM" => $term, "PARTITION" => $partition,
"NUM_DOCS" => $num_docs_term,
"NUM_OCCURRENCES" => $num_occurrences_term,
"POSTINGS_OFFSET" => $postings_offset,
"POSTINGS_LEN" => $postings_len]);
}
$dictionary->flush();
L\crawlLog("...Finished Adding Partition Posting Info to " .
"Dictionary: " . L\changeInMicrotime($start_time));
}
/**
* getPartitionBaseFolder gives the folder that holds the inverted index
* files of the partition numbered $partition.
*
* @param int $partition desired partition index
* @return string file path to where this partitions index data is stored
* (Not the original documents which are stored in the
* PartitionDocumentBundle)
*/
public function getPartitionBaseFolder($partition)
{
return $this->documents->getPartitionFolder($partition);
}
/**
* getSummary gives the summary of the document $doc_id in $partition,
* or [] if the partition does not hold it.
*
* @param string $doc_id of document to look up
* @param int $partition to look for document in
* @return array desired summary or [] if look up failed
*/
public function getSummary($doc_id, $partition)
{
$row = $this->documents->get($doc_id, $partition, [self::SUMMARY]);
return $row[self::SUMMARY] ?? [];
}
/**
* getCachePage gives the cached page of the document $doc_id in
* $partition, or [] if the partition does not hold it.
*
* @param string $doc_id of document to look up
* @param int $partition to look for document in
* @return array desired page cache or [] if look up failed
*/
public function getCachePage($doc_id, $partition)
{
$row = $this->documents->get($doc_id, $partition, [self::PAGE]);
return $row[self::PAGE] ?? [];
}
/**
* buildInvertedIndexPartition builds the inverted index files, the
* document map, positions and postings, of one partition of the
* documents PartitionDocumentBundle.
*
* @param int $partition to build index for
* @param string $taking_too_long_touch a filename of a file to touch
* so its last modified time becomes the current time. In a typical
* Yioop crawl this is done for the CrawlConstants::crawl_status_file
* file to prevent Yioop's web interface from stopping the crawl because
* it has seen no recent progress activity on a crawl.
* @param bool $just_stats if true, skip writing the postings/positions
* files for the partition and return only the stats array described
* in the @return below; if false (default), build the inverted index
* on disk and return whether the job completed
* @return mixed whether job executed to completion (true or false) if
* !$just_stats, otherwise, an array with NUM_DOCS, NUM_LINKS,
* and TERM_STATISTICS (the latter having term frequency info)
*/
public function buildInvertedIndexPartition($partition = -1,
$taking_too_long_touch = null, $just_stats = false)
{
$start_time = microtime(true);
L\crawlLog(
" Indexer start building inverted index ... Current Memory:".
memory_get_usage());
if ($partition < 0) {
$partition = $this->documents->parameters["SAVE_PARTITION"];
}
L\crawlLog(
"Indexer Building index inverted index for partition $partition");
$base_folder = $this->getPartitionBaseFolder($partition);
/* set up $doc_map_filename, $postings_filename, $postings_filename,
$positions_filename, etc
*/
if (!$just_stats) {
foreach (self::PARTITION_FILENAMES as $filename) {
$component_filename = $base_folder . "/" . $filename;
if (file_exists($component_filename)) {
unlink($component_filename);
}
$component = $filename . "_filename";
$$component = $component_filename;
}
}
$doc_map_tools = $this->doc_map_tools;
$postings_tools = $this->postings_tools;
$last_entries_tools = $this->last_entries_tools;
$this->doc_map = "";
$this->doc_map_counter = 0;
$this->partition_signals = [];
$this->postings = [];
$this->posting_counts = [];
$this->last_entries = [];
$this->last_rows = [];
$this->positions = "";
/* The document index is hashed before it is read, so a change
made while the build reads it makes the hash differ and the
partition is built again, rather than a build that missed the
change counting as current. */
$index_file = $this->documents->getPartitionIndex($partition);
$built_from = file_exists($index_file) ? md5_file($index_file) :
"";
L\crawlLog("Indexer Preparing Index Map...");
$index_map = $this->prepareIndexMap($partition);
L\crawlLog("Done Prepare Index Map. Number of documents in mapped ".
"partition:" . count($index_map));
$stored_size = 0;
foreach ([$this->documents->getPartitionIndex($partition),
$this->documents->getPartition($partition)] as $stored_file) {
clearstatcache(true, $stored_file);
$stored_size += file_exists($stored_file) ? filesize($stored_file) :
0;
}
if (!$just_stats && empty($index_map) && $stored_size > 0) {
L\crawlLog("Indexer not building partition $partition: its " .
"documents and their index hold $stored_size bytes but no " .
"document was read from them. Save partition is " .
($this->documents->parameters["SAVE_PARTITION"] ?? "unset") .
"; the build is left for a later pass.");
return false;
}
$cnt = 0;
$num_non_aux_docs = 0;
$num_links = 0;
$num_partition = count($index_map);
$doc_field = self::DOC_ID;
$score_field = self::SCORE;
$aux_docs_field = self::AUX_DOCS;
$get_summaries_time = 0;
$aux_get_summaries_time = 0;
$safe_score_time = 0;
$safe_meta_score_time = 0;
$invert_pages_time = 0;
$invert_metas_time = 0;
$invert_links_time = 0;
$this->extract_phrase_time = 0;
/* A builder that logs nothing for 5 * LOG_TIMEOUT seconds is taken
as stuck, so every loop below that can run long reports where it
is at most once a LOG_TIMEOUT: the document loop on every
document, whatever kind it is, and the loops over the link and
grouped documents of one url, which can number in the tens of
thousands for a popular page. */
$report_progress = function ($step) use ($partition, $num_partition,
&$num_non_aux_docs, &$cnt, $taking_too_long_touch) {
if (L\crawlTimeoutLog("..Indexer Still building inverted index " .
"for partition $partition, document %s of %s, %s links or " .
"docs so far, %s. Memory: %s.", $num_non_aux_docs,
$num_partition, $num_non_aux_docs + $cnt, $step,
memory_get_usage()) && $taking_too_long_touch &&
file_exists($taking_too_long_touch)) {
touch($taking_too_long_touch);
}
};
$current_file = self::currentDocumentFile($this->dir_name,
$partition);
$skip_hash = "";
if (file_exists($current_file)) {
/* The builder before this one stopped while on this document,
most likely because reading it made PHP exit, so it is
skipped this time rather than stopping this builder too. */
[$skip_hash, $skip_url] = array_pad(explode("\t",
file_get_contents($current_file), 2), 2, "");
L\crawlLog("..Indexer skipping document $skip_url in partition " .
"$partition: the builder before this one stopped on it.");
}
/* The file is opened once and written over for each document, as
opening and closing it for each of two writes a document can take
about as long as the rest of the build on a slow file system. */
$current_handle = fopen($current_file, "c");
$mark_current = function ($text) use ($current_handle,
$current_file) {
if ($current_handle === false) {
file_put_contents($current_file, $text);
return;
}
ftruncate($current_handle, 0);
rewind($current_handle);
fwrite($current_handle, $text);
};
foreach ($index_map as $hash_url => $url_info) {
$site = [];
$num_non_aux_docs++;
if ($hash_url === $skip_hash) {
continue;
}
$mark_current($hash_url);
$report_progress("reading its summary");
if (!empty($url_info[$doc_field])) {
$start_get_summaries = microtime(true);
$site = $this->getSummary($url_info[$doc_field], $partition);
$get_summaries_time += L\changeInMicrotime(
$start_get_summaries);
if (empty($site) || !is_array($site)) {
continue;
}
}
/* if $site still empty here then current group'd urls didn't have a
document (downloaded webpage) amongst themselves
*/
$max_description_len ??= C\MAX_DESCRIPTION_LEN;
$max_description_len = (empty($site[self::DESCRIPTION])) ?
$max_description_len : max($max_description_len,
strlen($site[self::DESCRIPTION]));
$metas_only = ($url_info[$aux_docs_field] == 'metas_only');
$aux_description = "";
$tmp_description = $site[self::DESCRIPTION] ?? "";
if (isset($site[self::TYPE]) && $site[self::TYPE] == "link") {
$site_url = $site[self::TITLE];
} else {
$site_url = str_replace('|', "%7C", $site[self::URL] ?? "");
}
$mark_current("$hash_url\t$site_url");
if ($metas_only) {
$start_safe_meta_time = microtime(true);
if (PhraseParser::computeSafeSearchScore($tmp_description,
$site_url) < PhraseParser::SAFE_PHRASE_THRESHOLD) {
$site[self::IS_SAFE] = true;
$url_info[self::IS_SAFE] = true;
} else {
$site[self::IS_SAFE] = false;
$url_info[self::IS_SAFE] = false;
}
$safe_meta_score_time +=
L\changeInMicrotime($start_safe_meta_time);
$site[self::JUST_METAS] = true;
$start_invert_metas = microtime(true);
$site_url = $this->invertOneSite($site, $url_info, $num_links);
$invert_metas_time +=
L\changeInMicrotime($start_invert_metas);
continue;
}
/*
Index pages that were hashed together or links to page
before page itself.
*/
$pre_aux_docs = explode("\xFF", $url_info[$aux_docs_field]);
$aux_sites = [];
$redirect_host_words = "";
$num_aux_docs = count($pre_aux_docs);
foreach ($pre_aux_docs as $aux_index => $pre_aux_doc) {
$report_progress("reading grouped document " .
($aux_index + 1) . " of $num_aux_docs");
$aux_doc = L\decode255($pre_aux_doc);
$start_get_summaries = microtime(true);
$aux_site = $this->getSummary($aux_doc, $partition);
$aux_get_summaries_time +=
L\changeInMicrotime($start_get_summaries);
if (empty($aux_site) || !is_array($aux_site)) {
$aux_site = []; // make sure empty
continue;
}
$aux_site[self::JUST_METAS] = true;
if (!empty($aux_site[self::DESCRIPTION])) {
$is_link = (($aux_site[self::TYPE] ?? "") == "link");
$aux_words = ($is_link) ?
self::linkWords($aux_site[self::DESCRIPTION]) :
$aux_site[self::DESCRIPTION];
if ($is_link && str_starts_with(
$aux_site[self::DESCRIPTION], "location:")) {
$redirect_host_words .= UrlParser::getWordsInHostUrl(
substr($aux_site[self::DESCRIPTION],
strlen("location:")));
}
if (strlen($aux_description) + strlen($aux_words) <
$max_description_len) {
$aux_description .= " .. " . $aux_words;
}
}
$aux_sites[] = $aux_site;
}
unset($pre_aux_docs);
if (empty($site) && !empty($aux_site) && is_array($aux_site)) {
//use one left aux_site for site if site empty
unset($aux_site[self::JUST_METAS]);
$site = $aux_site;
$site[self::DESCRIPTION] = "";
}
$site[self::DESCRIPTION] ??= "";
$site[self::DESCRIPTION] .= $aux_description;
if ($redirect_host_words !== "") {
$site[self::REDIRECT_HOST_WORDS] = $redirect_host_words;
}
$start_safe_time = microtime(true);
$description = $site[self::DESCRIPTION];
[$finished, $safe_score] = L\runWithTimeLimit(
C\DOCUMENT_TIME_LIMIT, fn() =>
PhraseParser::computeSafeSearchScore($description, $site_url));
if (!$finished) {
L\crawlLog("..Indexer skipping document $site_url: scoring " .
"it for safe search took over " . C\DOCUMENT_TIME_LIMIT .
" seconds.");
continue;
}
if ($safe_score < PhraseParser::SAFE_PHRASE_THRESHOLD) {
$site[self::IS_SAFE] = true;
$url_info[self::IS_SAFE] = true;
} else {
$site[self::IS_SAFE] = false;
$url_info[self::IS_SAFE] = false;
}
$safe_score_time +=
L\changeInMicrotime($start_safe_time);
$cnt++;
$start_invert_page = microtime(true);
$site_url = $this->invertOneSite($site, $url_info, $num_links);
$invert_pages_time += L\changeInMicrotime($start_invert_page);
foreach ($aux_sites as $aux_index => $aux_site) {
$report_progress("indexing grouped document " .
($aux_index + 1) . " of " . count($aux_sites));
$cnt++;
$start_invert_links = microtime(true);
$site_url = $this->invertOneSite($aux_site, $url_info,
$num_links);
$invert_links_time += L\changeInMicrotime($start_invert_links);
}
$memory_usage = memory_get_usage();
$link_to = (isset($site[self::TYPE]) &&
$site[self::TYPE] == "link") ? "LINK TO:" : "";
$time_string = L\makeTimestamp();
if (L\crawlTimeoutLog("..Indexer Still building inverted index ".
"for partition $partition \n" .
"$time_string ....Current Indexer Memory Usage is %s.\n" .
$time_string .
" ....Indexer has processed %s of %s documents.\n" .
$time_string .
" ....Total links or docs processed by Indexer is %s.\n" .
"$time_string ....Last url Indexer processed was %s.",
$memory_usage, $num_non_aux_docs, $num_partition,
$num_non_aux_docs + $cnt, $link_to . $site_url) &&
$taking_too_long_touch) {
if (file_exists($taking_too_long_touch)) {
touch($taking_too_long_touch);
}
}
}
if ($current_handle !== false) {
fclose($current_handle);
}
if (file_exists($current_file)) {
unlink($current_file);
}
$this->joinPostings();
if ($just_stats) {
$term_stats = [];
foreach ($this->postings as $term => $postings) {
$stat_pos = 0;
$num_records = L\vByteDecode($postings, $stat_pos);
$term_stats[$term] = $num_records;
}
$statistics = [
"NUM_DOCS" => $this->doc_map_counter,
"NUM_LINKS" => $num_links,
"TERM_STATISTICS" => $term_stats
];
return $statistics;
}
$start_save_times = microtime(true);
/* The four files are the partition's inverted index only together:
a kill after some are saved is undone the next time the bundle
opens, so the partition is built again from its documents. */
/* A partition stopped before any of its documents was saved has
no folder yet, and its files cannot be written without one. */
if (!file_exists(dirname($doc_map_filename))) {
L\makePath(dirname($doc_map_filename));
}
$undo_log = new UndoLog($this->dir_name);
$built_from_filename = $base_folder . "/" .
self::BUILT_FROM_FILENAME;
$step_id = $undo_log->begin("build of partition $partition", [],
[$doc_map_filename, $postings_filename, $last_entries_filename,
$positions_filename, $built_from_filename]);
$doc_map_tools->save($doc_map_filename, $this->doc_map, false);
self::saveDocMapMarks($doc_map_filename, $this->doc_map);
ksort($this->postings);
$postings_tools->save($postings_filename, $this->postings, false);
$this->packLastEntries();
$last_entries_tools->save($last_entries_filename, $this->last_entries,
false);
L\filePutContentsWhole($positions_filename, $this->positions, false);
$signals_filename = $base_folder . "/" . self::SIGNALS_FILENAME;
L\filePutContentsWhole($signals_filename, json_encode(
$this->partition_signals + ["DOCS" => $this->doc_map_counter]),
false);
L\filePutContentsWhole($built_from_filename, $built_from, false);
/* A built partition's files are written whole and never grow, so
they are synced but not checkpointed. */
$undo_log->finish($step_id, [$doc_map_filename, $postings_filename,
$last_entries_filename, $positions_filename, $signals_filename,
$built_from_filename], []);
$final_save_time = L\changeInMicrotime($start_save_times);
$time_string = L\makeTimestamp();
L\crawlLog(" Indexer build inverted index time ".
L\changeInMicrotime($start_time) .
"\n$time_string ..Component times:" .
"\n$time_string ....Get page summaries time: $get_summaries_time" .
"\n$time_string ....Get link summaries time: " .
$aux_get_summaries_time .
"\n$time_string ....Compute Safe Page time: $safe_score_time" .
"\n$time_string ....Compute Safe Meta time: $safe_meta_score_time".
"\n$time_string ....Invert pages time: $invert_pages_time" .
"\n$time_string ....Invert meta pages time: $invert_metas_time" .
"\n$time_string ....Invert links time: $invert_links_time" .
"\n$time_string ....Final file saves time: $final_save_time" .
"\n$time_string ----" .
"\n$time_string ....Of Invert times, time in " .
"extractPhrasesInLists:". $this->extract_phrase_time);
return true;
}
/**
* chooseAmongDuplicates picks, among documents whose pages have the
* same content, the one that keeps its words in the index; the others
* keep only their meta words, so a search does not list the same page
* several times. The shortest url wins, so that a site's home page
* beats a longer url that serves the same page, as
* https://www.google.com/webhp?tab=ww serves google.com's home page.
* Among urls of one length, or when no url is known, the higher score
* wins, which is the document added first with the most links to it.
* @param array $candidates keyed by the hash of each url, holding a
* pair of that url (empty if not known) and its score
* @return string the hash of the url chosen
*/
public static function chooseAmongDuplicates($candidates)
{
$chosen = "";
foreach ($candidates as $hash_url => [$url, $score]) {
if ($chosen === "") {
$chosen = $hash_url;
continue;
}
[$chosen_url, $chosen_score] = $candidates[$chosen];
$length = ($url === "") ? PHP_INT_MAX : strlen($url);
$chosen_length = ($chosen_url === "") ? PHP_INT_MAX :
strlen($chosen_url);
if ($length < $chosen_length || ($length == $chosen_length &&
$score > $chosen_score)) {
$chosen = $hash_url;
}
}
return $chosen;
}
/**
* currentDocumentFile gives the path of the file in which the builder of
* a partition writes which document it is reading. The builder writes
* the document's hash, then its url, before it reads the document, and
* removes the file when the partition is built. So a file left behind
* names the document that the builder was on when it stopped; the next
* builder of that partition skips that document, and the rebuild names
* it when it reports a builder as stuck.
*
* @param string $dir_name folder of the index document bundle
* @param int $partition number of the partition being built
* @return string path of the file that names the current document
*/
public static function currentDocumentFile($dir_name, $partition)
{
return "$dir_name/partition_$partition.current";
}
/**
* linkWords gives the words a link record adds to the page it points
* to. For most links these are the link's own text. A redirect leaves
* a link whose text is location: followed by the address that
* redirected, such as location:https://www.coke.com/. That address
* read as text makes one long term nobody searches for, so its host
* and path words, here coke, are given in its place. A search for
* coke can then reach the page www.coke.com redirects to.
*
* @param string $link_text the text of the link
* @return string the words to index for the link
*/
public static function linkWords($link_text)
{
if (!str_starts_with($link_text, "location:")) {
return $link_text;
}
$redirect_from = substr($link_text, strlen("location:"));
return UrlParser::getWordsInHostUrl($redirect_from) . " " .
UrlParser::getWordsLastPathPartUrl($redirect_from);
}
/**
* invertOneSite adds one page's postings, positions and document map
* row to the partition being built.
*
* @param array $site site to invert
* @param array $url_info collection of url and hash's of documents which
* map to the same document
* @param int &$num_links current count of number of links discovered so far
* @return string $site_url canonical url for site
*/
public function invertOneSite($site, $url_info, &$num_links)
{
$interim_time = microtime(true);
if (!isset($site[self::HASH]) ||
(isset($site[self::ROBOT_METAS]) &&
in_array("JUSTFOLLOW", $site[self::ROBOT_METAS]))) {
return "";
}
//this case might occur on a recrawl
if (isset($site[self::TYPE]) && $site[self::TYPE] == "link") {
$is_link = true;
$num_links++;
$site_url = $site[self::TITLE];
$host = UrlParser::getHost($site_url);
$link_parts = explode('|', $site[self::HASH]);
if (isset($link_parts[5])) {
$link_origin = $link_parts[5];
} else {
$link_origin = $site_url;
}
$url_info = [];
if (!empty($site[self::LANG])) {
$url_info[self::LANG] = $site[self::LANG];
}
$meta_ids = PhraseParser::calculateLinkMetas($site_url,
$host, $site[self::DESCRIPTION], $link_origin,
$url_info);
$link_to = "LINK TO:";
} else {
$is_link = false;
$site_url = str_replace('|', "%7C", $site[self::URL]);
$meta_ids = PhraseParser::calculateMetas($site);
$link_to = "";
}
$word_lists = [];
$host_keywords_end_pos = 0;
$title_end_pos = 0;
$path_keywords_end_pos = 0;
$triplet_lists = [];
/*
self::JUST_METAS check to avoid getting sitemaps in results
for popular words
*/
$lang = null;
if (!isset($site[self::JUST_METAS])) {
$host_words = UrlParser::getWordsInHostUrl($site_url);
/* A site's home page is also known by the host words of a
home page that redirects to it, as www.coca-cola.com is by
coke from www.coke.com, and they get the host bonus. */
if ($host_words !== "") {
$host_words .= $site[self::REDIRECT_HOST_WORDS] ?? "";
}
$path_words = UrlParser::getWordsLastPathPartUrl(
$site_url);
if ($is_link) {
$phrase_string = self::linkWords($site[self::DESCRIPTION]);
} else {
if (isset($site[self::LANG])) {
if (isset($this->programming_language_extension[
$site[self::LANG]])) {
$phrase_string = $site[self::DESCRIPTION];
} else {
/* r6t was chosen as short enough not to be
changed by chargramming, but rare enough
that can be used as a useful splitter
*/
$phrase_string = $host_words . " r6t ".
$site[self::TITLE] . " r6t ". $path_words .
" r6t ". $site[self::DESCRIPTION];
}
} else {
$phrase_string = $host_words . " r6t " .
$site[self::TITLE] . " r6t " . $path_words .
" r6t ". $site[self::DESCRIPTION];
}
}
/* at this point we have already extracted meta words,
we attempt to compute the lang here as a value different
from empty or mul for the purposes of stemming chargramming.
(helps with extracting words from images because the image
itself might have had few words to guess the lnaguage
but when combined with inlinks it does)
*/
if (empty($site[self::LANG]) || $site[self::LANG] == "mul") {
$lang = L\guessLocaleFromString(
$site[self::DESCRIPTION]);
} else {
$lang = $site[self::LANG];
}
[$finished, $word_and_qa_lists] = L\runWithTimeLimit(
C\DOCUMENT_TIME_LIMIT, fn() =>
PhraseParser::extractPhrasesInLists($phrase_string, $lang));
if (!$finished) {
L\crawlLog("..Indexer skipping document " .
($site[self::URL] ?? "") . ": finding its words took " .
"over " . C\DOCUMENT_TIME_LIMIT . " seconds.");
return "";
}
if (!isset($this->extract_phrase_time)) {
$this->extract_phrase_time = 0;
}
$this->extract_phrase_time +=
$word_and_qa_lists['TIMES']['TOTAL_TIME'] ?? 0;
$word_lists = $word_and_qa_lists['WORD_LIST'];
if (!empty($word_lists["r6t"][2])) {
if ($path_keywords_end_pos < 255) {
$host_keywords_end_pos = $word_lists["r6t"][0];
$title_end_pos = $word_lists["r6t"][1];
}
$path_keywords_end_pos = $word_lists["r6t"][2];
unset($word_lists["r6t"]);
} else if (!empty($word_lists["r6t"])) {
$path_keywords_end_pos = $word_lists["r6t"][
count($word_lists["r6t"]) - 1];
unset($word_lists["r6t"]);
}
}
$description_scores =
(empty($site[self::DESCRIPTION_SCORES])) ? [] :
$site[self::DESCRIPTION_SCORES];
$user_ranks =
(empty($site[self::USER_RANKS])) ? [] :
$site[self::USER_RANKS];
$num_words = 0;
foreach ($word_lists as $word => $position_list)
{
$num_words += count($position_list);
}
$doc_id = ($url_info[self::DOC_ID] ??
($url_info[self::AUX_DOCS][0] ?? ""));
if (empty($doc_id)) {
return "";
}
$terms = array_keys($word_lists);
if (count($terms) > self::NUM_TERMS_FILTER) {
$terms = array_slice($terms, 0, self::NUM_TERMS_FILTER);
}
$terms_filter = $this->storeTerms($terms);
$this->addScoresDocMap($doc_id, $num_words,
$url_info[self::SCORE], $host_keywords_end_pos, $title_end_pos,
$path_keywords_end_pos, $description_scores,
$user_ranks, $terms_filter);
foreach (self::signalCounts($doc_id) as $signal => $count) {
$this->partition_signals[$signal] =
($this->partition_signals[$signal] ?? 0) + $count;
}
$this->addTermPostingLists(0, $word_lists, $meta_ids,
$this->doc_map_counter, $path_keywords_end_pos);
$this->doc_map_counter++;
$interim_elapse = L\changeInMicrotime($interim_time);
if ($interim_elapse > 5) {
L\crawlLog("..Indexer Inverting " . $link_to . $site_url .
"...took > 5s.");
}
return $site_url;
}
/**
* computeDocId makes the doc id of a page from its $site array: a hash
* of its CrawlConstants::URL, its CrawlConstants::HASH, a letter coding
* its type and url signals, and a hash of its host.
*
* @param array $site site to compute doc_id for
* @return string the computed doc_id
*/
public static function computeDocId($site)
{
$doc_id = false;
if (isset($site[self::TYPE]) && $site[self::TYPE] == "link") {
$doc_id = $site[self::HTTP_CODE];
} else {
$letter_code = chr(0);
$main_type = (!empty($site[self::TYPE])) ?
substr($site[self::TYPE], 0, 4) : "binary";
if (!empty($site[self::IS_VIDEO])) {
$letter_code = chr(64);
} else if ($main_type == "text" ) {
$letter_code = chr(56);
} else if ($main_type == "imag") {
$letter_code = chr(48);
}
$hash = $site[self::HASH];
if ($letter_code == chr(56) && !empty($site[self::TITLE])) {
$trim_title = trim($site[self::TITLE]);
$hash = substr(L\crawlHash($trim_title, true), 0, 4) .
substr($hash, 4);
}
$site_url = str_replace('|', "%7C", $site[self::URL]);
$host = UrlParser::getHost($site_url);
$cld = UrlParser::getCompanyLevelDomain($site_url);
if (in_array($site_url, ["https://$cld/",
"https://www.$cld/", "http://$cld/", "http://www.$cld/"])) {
$letter_code = chr(ord($letter_code) + 128);
}
if (str_contains($cld, "wikipedia")) {
$letter_code = chr(ord($letter_code) + 4);
}
$num_slashes = substr_count(substr($site[self::URL], strlen($host)),
'/');
/*
* Discount any trailing slashes in the URL.
*/
if ($num_slashes > 0 && str_ends_with($site[self::URL], '/')) {
$num_slashes--;
}
/*
* The first two bits hold the number of / values (for the
* NUM_SLASHES_BONUS). This value is mapped into buckets of
* {0-1, 2-4, 5-6, 7+}, wherein all the values in a bucket get the
* same bonus. These buckets were decided after experimentation;
* the fundamental idea is that URLs for root pages/singly-nested
* pages are usually more important than those doubly nested or
* quadruply nested, which are in turn more important than those
* quintuply nested, etc.
*/
if ($num_slashes >= 2 && $num_slashes < 5) {
$letter_code = chr(ord($letter_code) + 1);
} else if ($num_slashes >=5 && $num_slashes < 7){
$letter_code = chr(ord($letter_code) + 2);
} else if ($num_slashes >=7) {
$letter_code = chr(ord($letter_code) + 3);
}
$doc_id = L\crawlHash($site_url, true) . $hash .
$letter_code . substr(L\crawlHash($host . "/", true), 1);
}
return $doc_id;
}
/**
* addScoresDocMap adds a doc_id => doc_record row to the current
* partition's document map ($this->doc_map). A doc record records the
* number of words in the document, an overall length of the document,
* the length of its title, scores for each of the sentences included
* into the summary for the documents, and classifier scores for each
* classifier that was used by the crawl.
*
* @param string $doc_id new document id to add a record for
* @param int $num_words number of terms in the document associated with the
* doc-id
* @param float $score overall score for the important of this document
* @param int $host_keywords_end_pos end of the portion of the
* document summary containing terms coming from the hostname
* @param int $title_end_pos end of the portion of the document
* summary containing terms in the title
* @param int $path_keywords_end_pos length of the portion of the
* document summary containing terms in the url path
* @param array $description_scores pairs of the form (length of summary
* portion, score for that portion)
* @param array $user_ranks for each user defined classifier for this crawl
* the float score of the classifier on this document
* @param string $terms_filter Bloom-filter prefix produced by
* storeTerms() identifying which terms appear in this document;
* empty string is allowed and stores the doc map without a
* terms-existence filter
*/
public function addScoresDocMap($doc_id, $num_words, $score,
$host_keywords_end_pos, $title_end_pos, $path_keywords_end_pos,
$description_scores, $user_ranks, $terms_filter = "")
{
$num_description_scores = count($description_scores);
$preface_positions =
((($host_keywords_end_pos << 8) + $title_end_pos) << 8) +
$path_keywords_end_pos;
$out_rows = [["POS" => $num_words, "SCORE" => floatval($score)],
["POS" => $preface_positions, "SCORE" =>
floatval($num_description_scores)]];
foreach ($description_scores as $position => $score) {
$out_rows[] = ["POS" => $position, "SCORE" => floatval($score)];
}
foreach ($user_ranks as $user_rank) {
$out_rows[] = ["POS" => 0, "SCORE" => floatval($score)];
}
$entry = $this->doc_map_tools->pack($out_rows);
/* the doc_map entry is prepended with a string representing
* the bloom filter of terms in the document
*/
$entry = $terms_filter . $entry;
$this->doc_map_tools->add($this->doc_map, $doc_id, $entry,
PackedTableTools::ADD_MEM_TABLE_STRING);
}
/**
* storeTerms makes a bloom filter string (a bit string that says a
* term may be in a set, or is surely not) of the terms given, the
* most important terms of a document. checkTermExists later uses it
* to check if a term belongs to the document.
*
* @param array $terms terms in document
* @return string term bloom filter, prepended with 't'
* to check for backward compatibility
*/
public static function storeTerms($terms)
{
$hash_functions = self::TERMSFILTER_HASHFN_COUNT;
$size = self::TERMSFILTER_BITS_LEN;
$terms_filter = str_repeat(chr(0), self::TERMSFILTER_LEN);
foreach ($terms as $term) {
$term = L\canonicalTerm($term);
for ($i = 0; $i < $hash_functions; $i++) {
$hash = crc32($term . $i) % $size;
$byte = (int)($hash >> 3);
$bit = $hash & 7;
$ascii_char = ord($terms_filter[$byte]);
$ascii_char |= (1 << $bit);
$terms_filter[$byte] = chr($ascii_char);
}
}
return 't' . $terms_filter;
}
/**
* checkTermExists says if a term may be in the term bloom filter that
* storeTerms made for a doc_map entry.
*
* @param string $term to look up
* @param string $terms_filter term bloom filter
* @return boolean exists or not
*/
public static function checkTermExists($term, $terms_filter)
{
$hash_functions = self::TERMSFILTER_HASHFN_COUNT;
$size = self::TERMSFILTER_BITS_LEN;
for ($i = 0; $i < $hash_functions; $i++) {
$hash = crc32($term . $i) % $size;
$byte = (int)($hash >> 3);
$bit = $hash & 7;
$ascii_char = ord($terms_filter[$byte]);
if (($ascii_char & (1 << $bit)) == 0) {
return false;
}
}
return true;
}
/**
* addTermPostingLists adds a document's posting records to the posting
* lists of the partition being built.
*
* @param int $position_offset number of header bytes that might be used
* before including any position data in the file that positions will
* eventually be stored.
* @param array $word_lists term => positions within current document of
* that term for the document whose posting data we are adding
* @param array $meta_ids meta terms associated with the document we are
* adding. An example, meta term might be "media:news"
* @param int $doc_map_index which document within the partition is the one
* we are adding. I.e., 5 would mean there were 5 earlier documents whose
* postings we have already added.
* @param int $field_end_pos the position in the document where its
* host, title and path words end; a word first found before it is
* also filed under its field keys, where the bundle has FIELD_TERMS
*/
public function addTermPostingLists($position_offset, $word_lists,
$meta_ids, $doc_map_index, $field_end_pos = 0)
{
/* A bundle built with combined terms also files each word under a
key for every short language code of the page with safe:all,
and with safe:true when the page is safe, the combinations a
guessed query adds, so such a query reads one key per word. The
combined postings point at the word's own position list. */
$combinations = [];
if (!empty($this->archive_info['COMBINED_TERMS'])) {
$safe_values = in_array("safe:true", $meta_ids) ?
["all", "true"] : ["all"];
foreach ($meta_ids as $meta_id) {
if (str_starts_with($meta_id, "lang:") &&
$meta_id != "lang:all" && !str_contains($meta_id, "-")) {
foreach ($safe_values as $safe) {
$combinations[] = [substr($meta_id, 5), $safe];
}
}
}
}
foreach ($meta_ids as $meta_id) {
$word_lists[$meta_id] = [];
}
/* A word's term ids depend only on the word and the page's
language and safety, so they are made once and reused on every
later page that has the same word and combinations. */
$combination_key = "";
foreach ($combinations as [$lang, $safe]) {
$combination_key .= " $lang:$safe";
}
$with_fields = !empty($this->archive_info['FIELD_TERMS']);
$size_codes = PackedTableTools::SIZE_CODES;
$postings = &$this->postings;
$posting_counts = &$this->posting_counts;
$last_rows = &$this->last_rows;
foreach ($word_lists as $word => $position_list) {
/* Positions are listed in the order the words appear, so the
first is the earliest. */
$in_fields = $with_fields && !empty($position_list) &&
$position_list[0] < $field_end_pos;
$cache_key = empty($position_list) ? "$word\x00" :
"$word\x00$combination_key" . ($in_fields ? " field" : "");
if (!isset($this->term_id_cache[$cache_key])) {
if (count($this->term_id_cache) >= self::TERM_ID_CACHE_SIZE) {
$this->term_id_cache = [];
}
$term_id = L\canonicalTerm($word);
$made_ids = [$term_id];
if (!empty($position_list)) {
foreach ($combinations as [$lang, $safe]) {
$made_ids[] = L\combinedTerm($word, $lang, $safe,
$term_id);
if ($in_fields) {
$made_ids[] = L\combinedTerm($word, $lang,
$safe . self::FIELD_TERM_MARK, $term_id);
}
}
}
$this->term_id_cache[$cache_key] = $made_ids;
}
$term_ids = $this->term_id_cache[$cache_key];
$occurrences = count($position_list);
if ($occurrences > 0) {
$encoded_position_list = L\encodePositionList($position_list);
$offset = $position_offset + strlen($this->positions);
$len = strlen($encoded_position_list);
$this->positions .= $encoded_position_list;
} else {
$offset = 0;
$len = 0;
}
/* An entry is written as postings_tools->pack writes a record of
DOC_MAP_INDEX, FREQUENCY, POSITIONS_OFFSET and POSITIONS_LEN,
without the count pack puts first: a byte of the four two bit
size codes, then the four values. The frequency and positions
length are the same for every key of the word, so they are
written once here. */
$frequency_bytes = PackedTableTools::packIntColumn($occurrences);
$length_bytes = PackedTableTools::packIntColumn($len);
$word_size_codes = ($size_codes[strlen($frequency_bytes)] << 4) |
$size_codes[strlen($length_bytes)];
foreach ($term_ids as $term_id) {
[$last_index, $last_offset, $num_occurrences] =
$last_rows[$term_id] ?? [0, 0, 0];
$diff_doc_map_index = $doc_map_index - $last_index;
$diff_offset = ($occurrences > 0) ?
$offset - $last_offset : 0;
$index_bytes = ($diff_doc_map_index >= 0 &&
$diff_doc_map_index < 128) ? chr($diff_doc_map_index) :
PackedTableTools::packIntColumn($diff_doc_map_index);
$offset_bytes = ($diff_offset >= 0 && $diff_offset < 128) ?
chr($diff_offset) :
PackedTableTools::packIntColumn($diff_offset);
$entry = chr(($size_codes[strlen($index_bytes)] << 6) |
($size_codes[strlen($offset_bytes)] << 2) |
$word_size_codes) . $index_bytes . $frequency_bytes .
$offset_bytes . $length_bytes;
/* A term's entries are appended in place, without the
count each packed record starts with; joinPostings puts
the term's count in front once all documents are added.
Rebuilding the whole row on each entry, as merging packed
rows does, made a common term's row cost time in the
square of its length. */
if (isset($postings[$term_id])) {
$postings[$term_id] .= $entry;
$posting_counts[$term_id]++;
} else {
$postings[$term_id] = $entry;
$posting_counts[$term_id] = 1;
}
$last_rows[$term_id] = [$doc_map_index, $offset,
$num_occurrences + $occurrences];
}
}
}
/**
* joinPostings puts the count of each term's entries at the front of
* the term's row in $postings, making the packed row format:
* the vbyte-encoded count, then the entries one after another.
* addTermPostingLists appends entries without it while a build adds
* documents. buildInvertedIndexPartition, and FeedDocumentBundle's
* build, call it once all documents are added.
*/
public function joinPostings()
{
foreach ($this->posting_counts as $term_id => $count) {
$this->postings[$term_id] = L\vByteEncode($count) .
$this->postings[$term_id];
}
$this->posting_counts = [];
}
/**
* packLastEntries turns the last posting of each term, kept in
* $last_rows while a build adds documents, into the packed rows of
* $last_entries that the last_entries file holds, one row per term.
* buildInvertedIndexPartition, and FeedDocumentBundle's build, call it
* just before saving that file.
*/
public function packLastEntries()
{
$last_entries_tools = $this->last_entries_tools;
foreach ($this->last_rows as $term_id => [$last_index, $last_offset,
$num_occurrences]) {
$last_entries_tools->add($this->last_entries, $term_id,
$last_entries_tools->pack(["LAST_INDEX" => $last_index,
"LAST_OFFSET" => $last_offset,
"NUM_OCCURRENCES" => $num_occurrences]));
}
$this->last_rows = [];
}
/**
* signalCounts says which url signals a document's id carries, read from
* the same bits computeDocRank reads: HOST_URL_BONUS for the root page
* of any host, and CLD_URL_BONUS as well when that host is the
* company-level domain itself or its www name, such as yahoo.com or
* www.yahoo.com, so it ranks above a subdomain such as sports.yahoo.com,
* WIKI_BONUS for a wiki page, and NUM_SLASHES_BONUS for the url's path
* depth. Each signal counts as occurrences of a pseudo word: one for a
* signal present, and 1/(number of slashes + 1) for the path depth. The
* build sums them over a partition's documents, and computeDocRank
* scores the ones a document has.
*
* @param string $doc_key a document's id
* @return array signal name => occurrences in this document
*/
public static function signalCounts($doc_key)
{
$counts = [];
$type_byte = ord($doc_key[self::DOCID_PART_LEN << 1] ?? chr(0));
if (self::isAHostDocId($doc_key)) {
$counts["HOST_URL_BONUS"] = 1;
if (($type_byte & 128) > 0) {
$counts["CLD_URL_BONUS"] = 1;
}
}
if (($type_byte & 96) != 96) {
if (($type_byte & 4) > 0) {
$counts["WIKI_BONUS"] = 1;
}
$counts["NUM_SLASHES_BONUS"] = 1 / (($type_byte & 3) + 1);
}
return $counts;
}
/**
* signalShares gives, for each url signal, its occurrences per document
* over the bundle's partitions, from the counts each partition's build
* wrote. A partition built before those counts were kept has no file and
* is left out. It looks at most once each SIGNAL_SHARES_CHECK_SECONDS,
* reads next_partition.txt for how many partitions are built, and reads
* only the counts of partitions built since its last look, so a search
* never lists the bundle's folders. The totals are saved under
* C\TEMP_DIR, so the first look of a later process reads them and the
* counts of partitions built since, not every partition's counts.
* WordIterator hands the shares to computeDocRank.
*
* @return array signal name => occurrences per document, or nothing
* when no partition has counts
*/
public function signalShares()
{
$cached = self::$signal_shares_cache[$this->dir_name] ??
[0, [], 0, []];
$now = microtime(true);
if ($now - $cached[2] < self::SIGNAL_SHARES_CHECK_SECONDS) {
return $cached[3];
}
[$read_up_to, $totals] = $cached;
$built = intval(@file_get_contents($this->dir_name . "/" .
self::NEXT_PARTITION_FILE));
/* A process's first look starts from the totals an earlier
process saved under C\TEMP_DIR, so it reads one file rather than
one per partition. The saved totals are used only when the last
signals file they counted has the time it had then: a rebuild
writes that file again and so makes them be read anew. */
$saved_path = self::$temp_folder . "/signal_totals_" .
md5($this->dir_name) . ".txt";
$last_signals = function ($partition) {
$signals_path = $this->getPartitionBaseFolder($partition - 1) .
"/" . self::SIGNALS_FILENAME;
clearstatcache(true, $signals_path);
return @filemtime($signals_path);
};
if ($read_up_to == 0) {
$saved = json_decode((string)@file_get_contents($saved_path),
true);
$saved_up_to = $saved["PARTITIONS"] ?? 0;
if ($saved_up_to > 0 && $saved_up_to <= $built &&
is_array($saved["TOTALS"] ?? null) &&
($saved["LAST_TIME"] ?? -1) === $last_signals($saved_up_to)) {
$read_up_to = $saved_up_to;
$totals = $saved["TOTALS"];
}
}
for ($partition = $read_up_to; $partition < $built; $partition++) {
$counts = json_decode((string)@file_get_contents(
$this->getPartitionBaseFolder($partition) . "/" .
self::SIGNALS_FILENAME), true);
foreach (is_array($counts) ? $counts : [] as $signal => $count) {
$totals[$signal] = ($totals[$signal] ?? 0) + $count;
}
}
if ($built > $read_up_to && file_exists(self::$temp_folder)) {
L\filePutContentsWhole($saved_path, json_encode([
"PARTITIONS" => $built, "TOTALS" => $totals,
"LAST_TIME" => $last_signals($built)]));
}
$shares = [];
foreach ($totals as $signal => $count) {
if ($signal != "DOCS" && !empty($totals["DOCS"])) {
$shares[$signal] = $count / $totals["DOCS"];
}
}
self::$signal_shares_cache[$this->dir_name] = [max($read_up_to,
$built), $totals, $now, $shares];
return $shares;
}
/**
* isAHostDocId tells the doc_id of a host url, such as
* https://www.yahoo.com/, from that of a page under the host, such as
* https://www.yahoo.com/foo
* @param string $key to check if doc or not
* @return bool true if $key represents a host url (DOCID_LEN bytes
* with the host hash matching at the prefix and suffix positions)
*/
public static function isAHostDocId($key)
{
if (strlen($key) == self::DOCID_LEN && substr($key, 1, 7) ==
substr($key, -7)) {
return true;
}
return false;
}
/**
* isType checks if a doc_id is of a large scale type among
* external_link, internal_link, link (union of previous two),
* binary, feed, image, text, video, document (union of previous five)
*
* @param string $key to check if doc or not
* @param string|array $types if a string then a particular type from above
* list to check against; if an array then an array of types to check
* against
* @return bool true if a document
*/
public static function isType($key, $types)
{
$type_map = [
0 => "binary",
8 => "old_doc",
16 => "external_link",
24 => "feed",
32 => "internal_link",
40 => "old_link",
48 => "image",
56 => "text",
64 => "video",
];
/*
* This map is maintained for backward compatibility, i.e.,
* for the $key values using the previous letter code format.
*/
$old_type_map = [
"b" => "binary",
"d" => "old_doc",
"e" => "external_link",
"f" => "feed",
"i" => "internal_link",
"l" => "old_link",
"p" => "image",
"t" => "text",
"v" => "video",
];
if (is_string($types)) {
$types = [$types];
}
if (in_array("link", $types)) {
$types = array_merge($types, ["external_link", "internal_link",
"old_link"]);
} else if (in_array("doc", $types) || in_array("document", $types)) {
$types = array_merge($types, ["binary", "feed", "image",
"old_doc", "text", "video"]);
}
$doc_id_format = ord($key[self::DOCID_PART_LEN << 1] ?? 0) & 96;
if ($doc_id_format != 96) {
$key_type = ord($key[self::DOCID_PART_LEN << 1] ?? 0) & 120;
return in_array($type_map[$key_type] ?? "old_link", $types);
}
// $key uses the old letter code format
$key_type = chr(ord($key[self::DOCID_PART_LEN << 1] ?? 0) & 127);
return in_array($old_type_map[$key_type] ?? "old_link", $types);
}
/**
* prepareIndexMap is the step before a partition's inverted index is
* made: it groups documents and links to documents into single
* objects.
* It also does simple deduplication of documents that have the same hash.
* It then returns an array of the grouped document data.
* Grouping is done by giving a score to each document based on
* (number of doc in index - order doc added). For two entries with
* the same hash_url, a document will be chosen over a link as the
* representative; otherwise, the one with higher score will be chosen as
* the representative. The representative document is given the sum of
* the scores of its constituents. A second phase where documents are
* grouped by hash of the text body is also done. Finally, the returned
* documents are sorted by their scores. So the order of documents from
* this process is roughly in the order of importance.
*
* @param int $partition index of partition to do deduplication for
* in the case that test index is empty
* @param array $test_index is non-null only when doing testing of what
* this method does. In which case, it should consist of an array
* of $doc_id => string represent a possible record for that doc.
* As deduplication is done entirely based on component of the doc_id
* (hash_url, doc_type, hash_doc, hash_host) the string doesn't matter
* too much.
* @return array groups doc_id => records associated with that doc_id
*/
public function prepareIndexMap($partition, $test_index = [])
{
if (empty($test_index)) {
$doc_index = $this->documents->loadPartitionIndex($partition, true);
} else {
$doc_index = $test_index;
}
if (empty($doc_index)) {
return [];
}
$doc_ids = array_keys($doc_index);
$num_ids = count($doc_ids);
$grouped_urls = [];
$grouped_hashes = [];
$score = $num_ids;
$doc_key_len = self::DOCID_PART_LEN;
$doc_field = self::DOC_ID;
$score_field = self::SCORE;
$aux_docs_field = self::AUX_DOCS;
foreach ($doc_ids as $doc_id) {
list($hash_url, $hash_code, ) = str_split($doc_id, $doc_key_len);
$current_grouped_urls = $grouped_urls[$hash_url] ??
[$aux_docs_field => "", $score_field => 0];
$current_grouped_hashes = $grouped_hashes[$hash_code] ?? "";
if (!$this->isType($doc_id, "link")) {
$current_grouped_hashes .= "\xFF". L\encode255($hash_url);
$current_grouped_urls[$doc_field] = $doc_id;
} else {
$current_grouped_urls[$aux_docs_field] .= "\xFF".
L\encode255($doc_id);
}
$current_grouped_urls[$score_field] += $score;
if (!empty($current_grouped_hashes)) {
$grouped_hashes[$hash_code] = $current_grouped_hashes;
}
$grouped_urls[$hash_url] = $current_grouped_urls;
$score--;
}
foreach ($grouped_hashes as $pre_same_hash_group) {
if (strlen($pre_same_hash_group) <= 2 * $doc_key_len) {
continue;
}
$same_hash_group = explode("\xFF", $pre_same_hash_group);
$candidates = [];
foreach ($same_hash_group as $pre_hash_url) {
if (empty($pre_hash_url)) {
continue;
}
$hash_url = L\decode255($pre_hash_url);
$summary = empty($test_index) ? $this->getSummary(
$grouped_urls[$hash_url][$doc_field], $partition) : [];
$candidates[$hash_url] = [$summary[self::URL] ?? "",
$grouped_urls[$hash_url][$score_field]];
}
$max_url = self::chooseAmongDuplicates($candidates);
$max_group = $grouped_urls[$max_url];
foreach ($same_hash_group as $pre_hash_url) {
if (empty($pre_hash_url)) {
continue;
}
$hash_url = L\decode255($pre_hash_url);
if ($hash_url != $max_url) {
$hash_group = $grouped_urls[$hash_url];
$max_group[$score_field] += $hash_group[$score_field];
if ($max_group[$aux_docs_field] != "metas_only") {
$max_group[$aux_docs_field] .= "\xFF" .
L\encode255($hash_group[$doc_field]) .
$hash_group[$aux_docs_field];
}
$hash_group[$aux_docs_field] = "metas_only";
$grouped_urls[$hash_url] = $hash_group;
}
}
$grouped_urls[$max_url] = $max_group;
}
unset($grouped_hashes);
uasort($grouped_urls, function ($one, $two) use ($score_field) {
return intval($two[$score_field] - $one[$score_field]);
});
return $grouped_urls;
}
/**
* forceSave builds the inverted index files of the partition still
* being filled, so its documents can be found by a search.
*/
public function forceSave()
{
$this->buildInvertedIndexPartition();
}
/**
* stopIndexing is called when a crawl stops, and builds the last
* partition's inverted index with forceSave, so the bundle is a working
* index on its own.
*/
public function stopIndexing()
{
$this->forceSave();
}
/**
* getWordInfo gives the dictionary rows of the word id $term_id, one
* for each partition holding it, with the active partition's row
* after them, and the word's counts.
*
* @param string $term_id id of phrase or word to look up in bundle
* dictionary
* @param int $threshold after the number of results exceeds this amount
* stop looking for more dictionary entries.
* @param int $offset position among the word's dictionary rows of the
* first row to give; 0 starts at the oldest partition
* @param int $num_partitions how many partitions to walk forward from
* $offset; -1 walks all remaining partitions
* @param bool $with_remaining_total whether TOTAL_COUNT and
* TOTAL_OCCURRENCES count every partition from $offset on, when
* $num_partitions reads only a batch of them; WordIterator asks
* for this so a query reports the word's whole count. They are
* exact up to exact_count_partitions rows and estimated past it
* @return array [] if the bundle has no dictionary; otherwise ROWS,
* the rows, each with PARTITION, NUM_DOCS and NUM_OCCURRENCES,
* and POSTINGS_OFFSET and POSTINGS_LEN, or for the active
* partition POSTINGS, along with TOTAL_COUNT, TOTAL_OCCURRENCES,
* AVG_ITEMS_PER_PARTITION, TOTAL_NUM_DOCS,
* TOTAL_NUM_LINKS_AND_DOCS, MAX_ITEMS_PER_FILE,
* TOTAL_NUMBER_OF_PARTITIONS, and THESHOLD_EXCEEDED when
* $threshold cut the rows short
*/
public function getWordInfo($term_id, $threshold = -1,
$offset = 0, $num_partitions = -1, $with_remaining_total = false)
{
$dictionary = $this->dictionary ?? [];
if (!$dictionary) {
return [];
}
$result = ["ROWS" =>
$dictionary->get($term_id, $offset, $num_partitions)];
if (empty($result["ROWS"])) {
$result = [];
}
$max_found_partition = 0;
$doc_count = 0;
$occurrence_count = 0;
$num_rows = 0;
$threshold_met = false;
$save_partition = $this->documents->parameters["SAVE_PARTITION"];
if (empty($result['ROWS'])) {
$result['ROWS'] = [];
}
foreach ($result['ROWS'] as $row) {
if ($threshold > 0 && $doc_count > $threshold) {
$result['ROWS'] = array_slice($result['ROWS'], 0, $num_rows);
$threshold_met = true;
break;
}
$max_found_partition = ($max_found_partition < $row['PARTITION']) ?
$row['PARTITION'] : $max_found_partition;
$doc_count += $row['NUM_DOCS'];
$occurrence_count += $row['NUM_OCCURRENCES'];
$num_rows++;
}
$parameters = $this->documents->parameters;
$result['AVG_ITEMS_PER_PARTITION'] = $doc_count/max($num_rows, 1.0);
$result['TOTAL_NUM_DOCS'] = $parameters["VISITED_URLS_COUNT"] ?? 0;
$result['TOTAL_NUM_LINKS_AND_DOCS'] = $parameters["ACTIVE_COUNT"] +
$parameters["COUNT"];
$result['MAX_ITEMS_PER_FILE'] = $parameters["MAX_ITEMS_PER_FILE"];
$result['TOTAL_NUMBER_OF_PARTITIONS'] = $parameters["SAVE_PARTITION"]
+ 1;
if ($threshold_met) {
$fraction_seen = ($save_partition - $offset) /
($max_found_partition - $offset);
$result['TOTAL_COUNT'] = $fraction_seen * $doc_count;
$result['TOTAL_OCCURRENCES'] = $fraction_seen * $occurrence_count;
$result['THESHOLD_EXCEEDED'] = true;
return $result;
}
/* A caller reading the rows a batch at a time still wants the
word's count over every partition. Rows past the batch are added
to the totals, though not to the rows returned, up to
exact_count_partitions rows in all. Where rows may remain past
that, the totals are scaled by the crawl's completed partitions
over the partitions the rows read reach, so a common word in a
very large crawl costs a bounded read. */
if ($with_remaining_total && $num_partitions > 0 &&
$num_rows >= $num_partitions) {
$more = max($this->exact_count_partitions - $num_rows, 0);
$rows_past = ($more > 0) ? $dictionary->get($term_id,
$offset + $num_partitions, $more) : [];
$last_partition = $max_found_partition;
foreach ($rows_past as $row) {
$doc_count += $row['NUM_DOCS'];
$occurrence_count += $row['NUM_OCCURRENCES'];
$last_partition = max($last_partition, $row['PARTITION']);
}
if (count($rows_past) >= $more) {
$scale = $save_partition / ($last_partition + 1);
if ($scale > 1) {
$doc_count = (int)round($doc_count * $scale);
$occurrence_count = (int)round($occurrence_count *
$scale);
}
}
}
/* A later batch that filled up is not the word's last, so the
active partition's row is not added to it, and its totals are
not read, so that row is not looked up. */
$is_later_full_batch = $offset > 0 && $num_partitions > 0 &&
$num_rows >= $num_partitions;
$row = ($is_later_full_batch) ? [] :
$this->activePartitionRow($term_id);
if (!empty($row)) {
$doc_count += $row["NUM_DOCS"];
$occurrence_count += $row['NUM_OCCURRENCES'];
/* The active partition comes after every partition in the
dictionary, so its row belongs after the word's last
dictionary row. A batch that filled up may have rows after
it; given the row there, a walk stepped from the batch's
last partition straight to the active one and on loading
the next batch skipped every partition below it, so a word
in more partitions than one batch holds was never read past
its first batch. Its counts are still added above, since
the totals of the first batch stand for the whole word. */
if ($num_partitions <= 0 || $num_rows < $num_partitions) {
$result['ROWS'][] = $row;
}
}
$result['TOTAL_COUNT'] = $doc_count;
$result['TOTAL_OCCURRENCES'] = $occurrence_count;
return $result;
}
/**
* saveDocMapMarks writes the doc_map_marks file of a document map: the
* map's size, its number of entries, and the byte offset at which
* every DOC_MAP_MARK_EVERY-th entry starts. Entries in a document map
* are separated by the byte 255, which encode255 keeps out of the
* entries themselves. buildInvertedIndexPartition calls it when it
* saves a partition's document map, and docMapMarks calls it for a
* partition built before such files were written. A failed write is
* left alone: docMapMarks then works from the marks it made.
*
* @param string $doc_map_filename path of the document map
* @param string $doc_map the document map's contents
* @return array [document map size, number of entries, offsets]
*/
public static function saveDocMapMarks($doc_map_filename, $doc_map)
{
$offsets = [];
$num_entries = 0;
$size = strlen($doc_map);
if ($size > 0) {
$num_entries = 1;
$offsets[] = 0;
$at = strpos($doc_map, "\xFF");
while ($at !== false) {
if ($num_entries % C\DOC_MAP_MARK_EVERY == 0) {
$offsets[] = $at + 1;
}
$num_entries++;
$at = strpos($doc_map, "\xFF", $at + 1);
}
}
$marks_filename = dirname($doc_map_filename) . "/" .
self::DOC_MAP_MARKS_FILENAME;
$writing = "$marks_filename.writing-" . getmypid();
if (@file_put_contents($writing, pack("N*", $size, $num_entries,
...$offsets)) !== false) {
@rename($writing, $marks_filename);
}
return [$size, $num_entries, $offsets];
}
/**
* docMapMarks gives the marks of a partition's document map: its
* size, its number of entries, and where every DOC_MAP_MARK_EVERY-th
* entry starts. They are read from the doc_map_marks file, or made
* from the document map and saved when that file is missing or was
* made for a document map of another size, as a partition built
* again by older code leaves. They are kept in $doc_map_marks_cache.
*
* @param string $doc_map_filename path of the document map
* @return array [document map size, number of entries, offsets], or
* null when the document map is missing
*/
public function docMapMarks($doc_map_filename)
{
clearstatcache(true, $doc_map_filename);
$size = @filesize($doc_map_filename);
if ($size === false) {
return null;
}
$marks = self::$doc_map_marks_cache[$doc_map_filename] ?? null;
if ($marks !== null && $marks[0] == $size) {
return $marks;
}
$saved = @file_get_contents(dirname($doc_map_filename) . "/" .
self::DOC_MAP_MARKS_FILENAME);
$numbers = (strlen($saved ?: "") >= 8) ?
array_values(unpack("N*", $saved)) : [];
if (count($numbers) >= 2 && $numbers[0] == $size) {
$marks = [$size, $numbers[1], array_slice($numbers, 2)];
} else {
$marks = self::saveDocMapMarks($doc_map_filename,
(string)@file_get_contents($doc_map_filename));
}
if (count(self::$doc_map_marks_cache) >=
self::MAX_DOC_MAP_MARKS_CACHED) {
unset(self::$doc_map_marks_cache[
array_key_first(self::$doc_map_marks_cache)]);
}
self::$doc_map_marks_cache[$doc_map_filename] = $marks;
return $marks;
}
/**
* docMapEntry gives entry number $doc_map_index of a partition's
* document map, the same entry PackedTableTools::
* findEntryAtIndexTableName gives, without reading the whole map. It
* seeks to the mark before the entry, from docMapMarks, and reads
* DOC_MAP_READ bytes at a time until it has passed the entries in
* between and read the one asked for.
*
* @param int $partition the partition whose document map to read
* @param int $doc_map_index number of the entry, counting from 0
* @return string the entry, its doc id followed by its packed
* records, or "" if the map has no such entry
*/
public function docMapEntry($partition, $doc_map_index)
{
$doc_map_filename = $this->getPartitionBaseFolder($partition) .
"/" . self::DOC_MAP_FILENAME;
$marks = $this->docMapMarks($doc_map_filename);
/* Each part of a union query scores the same pages again, so an
entry read is kept for the rest of the query, under its map's
size so a map saved again is read anew. */
$entry_key = "$doc_map_filename $doc_map_index " . ($marks[0] ?? -1);
if (isset(self::$doc_map_entries[$entry_key])) {
return self::$doc_map_entries[$entry_key];
}
if (empty($marks) || $doc_map_index < 0 ||
$doc_map_index >= $marks[1]) {
return "";
}
$mark = intdiv($doc_map_index, C\DOC_MAP_MARK_EVERY);
$to_skip = $doc_map_index % C\DOC_MAP_MARK_EVERY;
/* A search reads many entries from each partition it scores, so
a document map is kept open between reads. At most
DOC_MAP_HANDLES_KEPT maps stay open, the one used longest ago
closed first, so a process never runs out of the files it may
open. A map saved again since has another size, so its old
handle is closed and the map opened again. */
[$file, $size] = self::$doc_map_handles[$doc_map_filename] ??
[null, -1];
unset(self::$doc_map_handles[$doc_map_filename]);
if ($file && $size != $marks[0]) {
fclose($file);
$file = null;
}
if (!$file) {
$file = @fopen($doc_map_filename, "r");
}
while (count(self::$doc_map_handles) >= C\DOC_MAP_HANDLES_KEPT) {
$oldest = array_key_first(self::$doc_map_handles);
if (self::$doc_map_handles[$oldest][0]) {
fclose(self::$doc_map_handles[$oldest][0]);
}
unset(self::$doc_map_handles[$oldest]);
}
if ($file) {
self::$doc_map_handles[$doc_map_filename] = [$file, $marks[0]];
}
if (!$file || fseek($file, $marks[2][$mark]) < 0) {
return "";
}
$read = "";
$start = 0;
while (true) {
while ($to_skip > 0 && ($at = strpos($read, "\xFF", $start)) !==
false) {
$start = $at + 1;
$to_skip--;
}
$end = ($to_skip == 0) ? strpos($read, "\xFF", $start) : false;
if ($end !== false) {
break;
}
$more = fread($file, C\DOC_MAP_READ);
if ($more === false || $more === "") {
$end = strlen($read);
break;
}
$read .= $more;
}
$entry = ($to_skip == 0) ?
L\decode255(substr($read, $start, $end - $start)) : "";
if (count(self::$doc_map_entries) >= self::MAX_DOC_MAP_ENTRIES) {
unset(self::$doc_map_entries[
array_key_first(self::$doc_map_entries)]);
}
self::$doc_map_entries[$entry_key] = $entry;
return $entry;
}
/**
* activePartitionRow gives a term's row in the active partition, the
* newest one, whose postings are not in the dictionary yet. Its row
* holds the postings themselves, where a dictionary row holds only
* where they are. getWordInfo adds it after the term's dictionary
* rows, and newestRow asks it before the dictionary.
*
* @param string $term_id id of the term to look up
* @return array the row, with PARTITION, NUM_DOCS, NUM_OCCURRENCES and
* POSTINGS, or an empty array if the active partition does not
* hold the term
*/
public function activePartitionRow($term_id)
{
$save_partition = $this->documents->parameters["SAVE_PARTITION"];
$postings_filename = $this->getPartitionBaseFolder($save_partition) .
"/" . self::POSTINGS_FILENAME;
if (!file_exists($postings_filename)) {
return [];
}
$postings_tool = $this->postings_tools;
$active_dictionary = $postings_tool->load($postings_filename,
$postings_tool::AS_STRING_MODE, true);
/* buildInvertedIndexPartition sorts the postings by term before
saving them, so a binary search finds a term's row. A query
looks a term up here for each page it scores, to find the
page's latest copy, and a search of the whole string would
read every row each time. */
$active_postings_entry = $postings_tool->findRowInSortedTableString(
$active_dictionary, $term_id);
$active_postings = (empty($active_postings_entry)) ? [] :
$postings_tool->unpack($active_postings_entry);
if (empty($active_postings)) {
return [];
}
/* The postings are stored as differences from the one before, and
this puts back their own values as it sums their frequencies. */
$num_occurrences =
$this->deDeltaPostingsSumFrequencies($active_postings);
return ["PARTITION" => $save_partition,
"NUM_DOCS" => count($active_postings),
"POSTINGS" => $active_postings,
"NUM_OCCURRENCES" => $num_occurrences];
}
/**
* newestRow gives a term's row in the newest partition holding it,
* with that partition's postings of the term. The active partition is
* asked first, as it is newer than every partition in the dictionary.
* Otherwise the dictionary is walked newest tier first and stops at
* the first row it finds, whose postings are then read from its
* partition's postings file. IndexManager::lookupLatestVersionPage
* uses it to find the newest copy of a page from its info: key.
*
* @param string $term_id id of the term to look up
* @return array the row, with PARTITION and POSTINGS, or an empty
* array if no partition holds the term
*/
public function newestRow($term_id)
{
$row = $this->activePartitionRow($term_id);
if (!empty($row) || empty($this->dictionary)) {
return $row;
}
$row = $this->dictionary->get($term_id, 0, 1, true)[0] ?? [];
if (empty($row['POSTINGS_LEN'])) {
return [];
}
$postings_info = $this->unpackPostings($this->getPostingsString(
$row['PARTITION'], $row['POSTINGS_OFFSET'],
$row['POSTINGS_LEN'], $term_id));
$row['POSTINGS'] = $postings_info[0] ?? [];
return $row;
}
/**
* dropOldestPostingHandle takes the handle that has gone longest
* without a read out of the list of open postings files.
* getPostingsString calls it when the process is near its memory
* limit or holds too many handles.
*
* The list is keyed by the postings file each handle reads, and
* array_shift renumbers the keys it leaves behind, so after one
* shift a lookup by partition handed back the handle of another
* partition's postings file. A read then took a row's offset and
* length into the wrong file, which decodes as nonsense or as
* nothing at all.
*
* @param array &$file_handles open postings files, keyed by the
* path of the file each reads, least recently read first
*/
public static function dropOldestPostingHandle(&$file_handles)
{
$oldest_partition = array_key_first($file_handles);
if ($oldest_partition === null) {
return;
}
unset($file_handles[$oldest_partition]);
}
/**
* getPostingsString reads the postings stored in a partition's
* postings file from $offset to $offset + $len, with the 255 encoding
* removed. The amount read
* is capped at the bytes the partition file actually holds, so a
* corrupt dictionary entry that asks for a wildly long posting list
* cannot make the server allocate a huge buffer and run out of
* memory; an over-long request is logged and clamped to the file.
*
* @param int $partition partition to retrieve posting from
* @param int $offset byte offset int partition/postings file to look for
* them
* @param int $len length of the posting list to retrieve.
* @param string $term_id the dictionary key the length came from,
* written into the log when the length does not fit the partition,
* so the term whose row looks corrupt can be looked up; the empty
* string leaves the term out of the log
* @return string encoded posting list data -- vbyte encoded number of
* postings, followed by the posting data in PacktableTools format
*/
public function getPostingsString($partition, $offset, $len,
$term_id = "")
{
/* A dictionary row that names a length the partition cannot
hold is the fault to chase, and the term it belongs to is
what names it, so the two log lines below say which term
asked. */
$term_said = ($term_id == "") ? "" : " for term " .
rawurlencode($term_id);
static $file_handles = [];
static $memory_limit = 0;
if (!$memory_limit) {
$memory_limit =
L\metricToInt(C\INDEX_FILE_MEMORY_LIMIT) * C\MEMORY_FILL_FACTOR;
}
if (memory_get_usage() > $memory_limit ||
count($file_handles) > self::MAX_POSTING_CACHE_ITEMS) {
self::dropOldestPostingHandle($file_handles);
}
/* The handles are kept for the whole process, which may read
more than one index, as a crawl mix does, so a handle is kept
under the file it reads and not under its partition number,
which every index uses. */
$postings_filename = $this->getPartitionBaseFolder($partition) .
"/" . IndexDocumentBundle::POSTINGS_FILENAME;
if (empty($file_handles[$postings_filename])) {
if (file_exists($postings_filename)) {
$fh = fopen($postings_filename, "r");
} else {
return "";
}
$file_handles[$postings_filename] = $fh;
} else {
$fh = $file_handles[$postings_filename];
unset($file_handles[$postings_filename]); // move to front
$file_handles[$postings_filename] = $fh;
}
if ($fh && fseek($fh, $offset) == 0 && $len > 0) {
$stats = fstat($fh);
$file_size = $stats['size'] ?? 0;
$available = $file_size - $offset;
if ($available <= 0) {
return "";
}
if ($len > $available) {
L\crawlLog("getPostingsString: postings length " . $len .
" exceeds the " . $available . " available bytes " .
"at offset " . $offset . " in partition " .
$partition . $term_said .
"; dictionary entry looks corrupt, " .
"clamping to the file size");
$len = $available;
}
if ($len > C\MAX_POSTING_READ_LEN) {
L\crawlLog("getPostingsString: postings length " . $len .
" exceeds the maximum readable posting length " .
C\MAX_POSTING_READ_LEN . " at offset " . $offset .
" in partition " . $partition . $term_said .
"; treating the " .
"dictionary entry as corrupt and skipping the read");
return "";
}
if ($len > self::$max_postings_read) {
self::$max_postings_read = $len;
}
$out = L\decode255(fread($fh, $len) ?? "");
return $out;
}
return "";
}
/**
* unpackPostings turns a term's postings string from one partition
* into an array of postings, doing de-delta of doc_map_indices and
* de-delta of positions. Each posting represents occurrence of a term
* in a documents, so the frequency component is the number of occurrences
* of the term in the document. This method also computes the sum of these
* frequencies over all postings in partition.
*
* @param string $postings_string compress string representation of a
* set of postings for a term
* @return array a pair [array of unpacked postings, sum of frequencies
* of all the postings]
*/
public function unpackPostings($postings_string)
{
if (empty($postings_string)) {
return [];
}
$unpack_map = $this->unpack_map;
$unpack_len_map = $this->unpack_len_map;
$current_pos = 0;
$num_items = L\vByteDecode($postings_string, $current_pos);
$len_posting_strings = strlen($postings_string);
/* A corrupt posting string can say it holds more items than its
bytes could possibly encode. Each item takes at least two
bytes, one naming the format and one of payload, so a count
past half the string's length is not believable and the list
is treated as corrupt rather than walked.
*/
if ($num_items > intdiv($len_posting_strings, 2)) {
L\crawlLogCounter("Posting decode error - item count beyond" .
" string");
return [[], 0];
}
$items = [];
$sum_frequencies = 0;
$doc_map_index = 0;
$positions_offset = 0;
for ($i = 0; $i < $num_items; $i++) {
if (!isset($postings_string[$current_pos])) {
L\crawlLogCounter(
"Posting decode error - Start beyond posting");
return [$items, $sum_frequencies]; // sanity check 1
}
$int_info = ord($postings_string[$current_pos]);
$current_pos++;
$len_unpack_info = $unpack_len_map[$int_info];
if ($current_pos + $len_unpack_info > $len_posting_strings) {
L\crawlLogCounter("Posting decode error -".
" Decode length longer than string");
return [$items, $sum_frequencies]; // sanity check 2
}
$pre_item = unpack($unpack_map[$int_info], $postings_string,
$current_pos);
if ($pre_item["FREQUENCY"] > C\MAX_DESCRIPTION_LEN) {
L\crawlLogCounter("Posting decode error! Frequency too large");
return [$items, $sum_frequencies]; // sanity check 3
}
$item = $pre_item;
$item["DOC_MAP_INDEX"] += $doc_map_index;
$item["POSITIONS_OFFSET"] += $positions_offset;
$doc_map_index += $pre_item["DOC_MAP_INDEX"];
$positions_offset += $pre_item["POSITIONS_OFFSET"];
$sum_frequencies += $pre_item["FREQUENCY"];
$current_pos += $len_unpack_info;
$items[] = $item;
}
return [$items, $sum_frequencies];
}
/**
* deDeltaPostingsSumFrequencies puts back the DOC_MAP_INDEX and
* POSITIONS_OFFSET values of postings, which are stored as differences
* from the posting before, and sums the postings' frequencies. Only
* activePartitionRow uses it, for the active partition, whose terms
* have not been added to the dictionary LSMTree yet.
*
* @param array &$postings a reference to an array of posting lists for a
* term (this will be changed by this method)
* @return int sum of the frequencies of term occurrences as given by the
* above postings
*/
public function deDeltaPostingsSumFrequencies(&$postings)
{
if (empty($postings) || !is_array($postings)) {
return 0;
}
list($doc_map_index, $sum_frequencies, $positions_offset) =
array_values($postings[0]);
$num_postings = count($postings);
for ($i = 1; $i < $num_postings; $i++) {
$posting = & $postings[$i];
list($doc_map_delta, $frequency, $positions_delta) =
array_values($posting);
$sum_frequencies += $frequency;
$doc_map_index += $doc_map_delta;
$positions_offset += $positions_delta;
$posting["DOC_MAP_INDEX"] = $doc_map_index;
$posting["POSITIONS_OFFSET"] = $positions_offset;
}
return $sum_frequencies;
}
/**
* getArchiveInfo gives the description, count of documents, and number
* of partitions of the bundle in the supplied directory. If the file
* arc_description.txt exists, this is viewed as a dummy index archive for
* the sole purpose of allowing conversions of downloaded data such as arc
* files into Yioop! format.
*
* @param string $dir_name path to a directory containing a documents
* IndexDocumentBundle
* @return array summary of the given archive
*/
public static function getArchiveInfo($dir_name)
{
if (file_exists($dir_name . "/arc_description.txt")) {
$crawl = [];
$info = [];
$crawl['DESCRIPTION'] = substr(
file_get_contents($dir_name . "/arc_description.txt"), 0, 256);
$crawl['ARCFILE'] = true;
$info['VISITED_URLS_COUNT'] = 0;
$info['COUNT'] = 0;
$info['NUM_DOCS_PER_PARTITION'] = 0;
$info['WRITE_PARTITION'] = 0;
$info["VERSION"] = self::DEFAULT_VERSION;
$info['DESCRIPTION'] = serialize($crawl);
return $info;
}
$info_path = $dir_name . "/" . self::ARCHIVE_INFO_FILE;
if (!file_exists($info_path)) {
$info = [];
$info['DESCRIPTION'] =
"Archive does not exist OR Archive description file not found";
$info['COUNT'] = 0;
$info['NUM_DOCS_PER_PARTITION'] = -1;
$info["VERSION"] = self::DEFAULT_VERSION;
return $info;
}
$info = unserialize(file_get_contents($info_path));
if (!is_array($info)) {
$info = [];
}
$table_info = PartitionDocumentBundle::getParameterInfo($dir_name . "/".
self::DOCUMENTS_FOLDER);
if (!is_array($table_info)) {
$table_info = [];
}
$info = array_diff_key($info, $table_info);
$info = array_merge($table_info, $info);
return $info;
}
/**
* setArchiveInfo sets the archive info struct of the bundle in
* $dir_name. This struct has fields like: DESCRIPTION
* (serialized store of global parameters of the crawl like seed sites,
* timestamp, etc).
*
* @param string $dir_name folder with archive bundle
* @param array $update_info struct with above fields
*/
public static function setArchiveInfo($dir_name, $update_info)
{
$archive_info_path = $dir_name . "/" . self::ARCHIVE_INFO_FILE;
if (file_exists($archive_info_path)) {
$info = self::getArchiveInfo($dir_name);
}
if (empty($info) || !is_array($info)) {
$info = [];
}
$pdb_info = [];
$got_pdb_info = false;
if (!empty($info)) {
$doc_folder = $dir_name. "/" . self::DOCUMENTS_FOLDER;
if (file_exists($doc_folder)) {
$pdb_info = PartitionDocumentBundle::getParameterInfo(
$doc_folder);
if (!empty($pdb_info)) {
$got_pdb_info = true;
}
// avoid getting same data (_COUNTS) stored in two locations
if (!empty($info) && !empty($pdb_info)) {
$info = array_diff_key($info, $pdb_info);
}
}
}
$pdb_change = false;
foreach ($update_info as $field => $value) {
if (isset($pdb_info[$field])) {
$pdb_info[$field] = $value;
$pdb_change = true;
}
$info[$field] = $value;
}
if (empty($info["VERSION"])) {
$info["VERSION"] = self::DEFAULT_VERSION;
}
L\filePutContentsWhole($archive_info_path, serialize($info));
if ($got_pdb_info && $pdb_change) {
$parameter_path = $doc_folder . "/" .
PartitionDocumentBundle::PARAMETERS_FILE;
/* Written new and renamed into place, as saveParameters does,
so a process reading the file never finds it empty. */
$new_path = $parameter_path . "." . getmypid();
file_put_contents($new_path, serialize($pdb_info));
rename($new_path, $parameter_path);
}
}
/**
* getParamModifiedTime gives the last time the archive info of the
* bundle was modified.
*
* @param string $dir_name folder with archive bundle
* @return mixed Unix timestamp of the bundle's status file if it
* exists, or false if the file is missing
*/
public static function getParamModifiedTime($dir_name)
{
$doc_param_path = $dir_name . "/" . self::DOCUMENTS_FOLDER . "/" .
PartitionDocumentBundle::PARAMETERS_FILE;
if (file_exists($doc_param_path)) {
clearstatcache();
return filemtime($doc_param_path);
}
return false;
}
}
- Yioop Repo - Seekquarry