<?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\compressors\GzipCompressor;
use seekquarry\yioop\library\data_structures\PackedTableTools;
/**
* Loads crawlLog functions if needed
*/
require_once __DIR__ . "/../Utility.php";
/**
* A partition document bundle is a collection of partition each of which in
* turn can hold a concatenated sequence of compressed documents and which
* are managed together. It is a successor format to the earlier
* WebArchiveBundle of Yioop. The partition document bundle stores individual
* records using a record format defined via the PackedTableTools class.
* This basic format has been extended by two new types BLOB and SERIAL (a
* PHP serialized object represesnted as a blob).
* Data for columns of these types are stored in separate files from the rest
* of records. Offset into these archive files for blobs and serial's are
* stored in a record as columns representing a difference list of int's
* together with a LAST_BLOB_LEN column. Using this info, blob's and serial's
* associated with a record can be retrieved. How many documents are
* together with a collected into a partition can be tuned
* for read, write, and in-memory efficiency.
*
* @author Chris Pollett
*/
class PartitionDocumentBundle
{
/**
* Compression strategy used to compress blob and serial columns
*/
const DEFAULT_COMPRESSOR = C\NS_COMPRESSORS . "NonCompressor";
/**
* Default parameters to use when constructing a PartitionDocumentBundle
*/
const DEFAULT_PARAMETERS = ["RECORD_COMPRESSOR" => self::DEFAULT_COMPRESSOR,
"BLOB_COMPRESSOR" => self::DEFAULT_COMPRESSOR, "COUNT" => 0,
"OVERFLOW_THRESHOLD" => C\OVERFLOW_THRESHOLD,
"PARTITION_SIZE_THRESHOLD" => self::PARTITION_SIZE_THRESHOLD,
"FORMAT" => ["PRIMARY KEY" => "KEY", "VALUE" => "BLOB"],
"MAX_ITEMS_PER_FILE" => self::MAX_ITEMS_PER_FILE,
"SAVE_PARTITION" => 0, "ACTIVE_COUNT" => 0
];
/**
* Default maximum number of records to store in a partition
*/
const MAX_ITEMS_PER_FILE = 16384;
/**
* File name of file used to store the parameters of this
* PartitionDocumentBundle
*/
const PARAMETERS_FILE = "pdb_parameters.txt";
/**
* Prefix to block folders of PartitionDocumentBundle partition files
*/
const BLOCK_PREFIX = "B";
/**
* Number of partition files to store in a block folder before making
* another one
*/
const BLOCK_FACTOR = 10000;
/**
* Base filename used for the bundle's on-disk archive data
*/
const ARCHIVE_FILENAME = "archive";
/**
* Base filename for the bundle's primary-key index file
*/
const INDEX_FILENAME = "index";
/**
* Prefix to file names of PartitionDocumentBundle partition files
*/
const PARTITION_PREFIX = "P";
/**
* Maximum number of bytes a partition can have before the next partition
* is started. Notice this implies a maximum file size to store
* in BLOB columns
*/
const PARTITION_SIZE_THRESHOLD = 2147483648;
/**
* Sub-folder under the bundle root that holds overflow files
* for individual records whose blob/serial column data is too
* large to fit in a regular partition
*/
const OVERFLOW_DIR_FOLDER = "Overflow";
/**
* Used to store the file handle to, the partition number, and last add time
* for the last time an item's blob/serial columns were added to for
* the PartitionDocumentBundle
* @var array
*/
public $add_archive_cache = [null, "", -1];
/**
* Used to store the file handle to, the partition number, and last access
* time for the last time an item's blob/serial columns were accessed for
* the PartitionDocumentBundle
* @var array
*/
public $get_archive_cache = [null, "", -1];
/**
* Array of column names for the columns in a PartitionDocumentBundle which
* are of type BLOB or SERIAL
* @var array
*/
public $blob_columns;
/**
* The seekquarry\yioop\library\compressors\Compressor object used to
* compress record files.
* @var object
*/
public $record_compressor;
/**
* The seekquarry\yioop\library\compressors\Compressor object used to
* compress blob columns.
* @var object
*/
public $blob_compressor;
/**
* Folder path where the PartitionDocumentBundle is stored
* @var string
*/
public $folder;
/**
* In memory cache of partitions from the PartitionDocumentBundle
* @var array
*/
public $index_cache;
/**
* Maximum number of items the partition cache is allowed to hold
* @var int
*/
public $index_cache_size;
/**
* $log_partition_work says whether this bundle writes a line to the
* log for the work it does with partitions, such as reading one into
* memory or adding pages to one. A crawl wants those lines; a search
* does not, and a search run from the command line would otherwise
* print them among its own output.
* @var bool
*/
public $log_partition_work = false;
/**
* Used to keep track of when this instance was created, as part of managing
* file handles expiration (could be set/updated externally to reflect
* some other instance using the bundle)
* @var int
*/
public $instance_time;
/**
* Name of primary key column for records
* @var string
*/
public $key_field;
/**
* Stores the constructor parameters used to create this
* PartitionDocumentBundle
* @var array
*/
public $parameters;
/**
* Array of column names for the columns in a PartitionDocumentBundle which
* are of type SERIAL
* @var array
*/
public $serial_columns;
/**
* The PackedTableTools object used to pack and unpack records in
* partitions
* @var object
*/
public $table_tools;
/**
* Holds loaded unserialized index file data from $partition partition
* bundle
* @var mixed
*/
public $save_index;
/**
* $defer_index_save says whether put keeps the save partition's index
* in memory from one call to the next and leaves saving it to
* saveIndex. The indexer sets it while it stores a fetcher's file,
* which it does in groups of up to QueueServer::SITES_STORED_AT_ONCE
* sites, so the index is written once per file rather than once per
* group. Searches see the file's pages once saveIndex runs at the end.
* @var bool
*/
public $defer_index_save = false;
/**
* $save_index_partition is the partition whose index $save_index
* holds, or -1 when it holds none put can use again.
* @var int
*/
public $save_index_partition = -1;
/**
* How many of the keys the last call to put added were not already in
* the save partition. A page the fetchers send twice is stored once, so
* this is the count of documents the partition gained, which is what a
* visited-pages tally should count.
* @var int
*/
public $new_keys_put = 0;
/**
* $save_keys holds, as its array keys, every document key already in
* the save partition, so that put can tell a new page from one sent
* again with one array lookup. Searching the packed index string
* for each page instead read the whole index once per page, and the
* index grows to the size of a partition, so storing a file of pages
* took longer and longer as a partition filled and fell back only
* when a new one was begun.
* @var array
*/
public $save_keys = [];
/**
* $save_keys_partition says which partition $save_keys was built
* for, so that put rebuilds the set when the save partition has
* changed and reuses it otherwise.
* @var int
*/
public $save_keys_partition = -1;
/**
* Used to create a new instance of a PartitionDocumentBundle
*
* @param string $folder the path to the folder to store this
* PartitionDocumentBundle
* @param array $format the column names, keys and types for this
* PartitionDocumentBundle object
* @param int $max_items_per_file maximum number of items to store
* in a partition before making the next partition
* @param int $partition_size_threshold maximum length of a partition
* file in bytes before a new partition file should be started
* @param object $record_compressor_type
* seekquarry\yioop\library\compressors\Compressor object used to
* compress record files excluding blob columns.
* @param object $blob_compressor_type
* seekquarry\yioop\library\compressors\Compressor object used to
* compress blob columns.
* @param int $overflow_threshold maximum total bytes of in-memory
* records before the bundle rolls over to a new on-disk partition
* @return null returned (leaving the instance partially initialised)
* on a corrupt on-disk parameters file or when the resolved
* format disagrees with the constructor arguments
*/
public function __construct($folder, $format =
self::DEFAULT_PARAMETERS["FORMAT"],
$max_items_per_file = self::MAX_ITEMS_PER_FILE,
$partition_size_threshold = self::PARTITION_SIZE_THRESHOLD,
$record_compressor_type = self::DEFAULT_COMPRESSOR,
$blob_compressor_type = self::DEFAULT_COMPRESSOR,
$overflow_threshold = C\OVERFLOW_THRESHOLD)
{
$initial_parameters = self::DEFAULT_PARAMETERS;
$initial_parameters["PARTITION_SIZE_THRESHOLD"] =
$partition_size_threshold;
$initial_parameters["MAX_ITEMS_PER_FILE"] = $max_items_per_file;
$initial_parameters["RECORD_COMPRESSOR"] = $record_compressor_type;
$initial_parameters["BLOB_COMPRESSOR"] = $blob_compressor_type;
$initial_parameters["OVERFLOW_THRESHOLD"] = $overflow_threshold;
$this->record_compressor = new $record_compressor_type();
$this->blob_compressor = new $blob_compressor_type();
$initial_parameters["FORMAT"] = $format;
$this->instance_time = hrtime(true);
$this->index_cache_size = min(50, floor(L\metricToInt(
ini_get('memory_limit'))/128000000));
$this->folder = $folder;
$folder_paths = [$folder];
if ($overflow_threshold > 0) {
$folder_paths[] = $folder . "/" . self::OVERFLOW_DIR_FOLDER;
}
$changed_parameters = false;
foreach ($folder_paths as $folder_path) {
if (!file_exists($folder_path)) {
$changed_parameters = true;
if (!mkdir($folder_path)) {
chmod($folder_path, 0777);
return null;
}
}
}
$this->parameters = self::getParameterInfo($folder);
foreach (self::DEFAULT_PARAMETERS as $field => $value) {
if (!isset($this->parameters[$field])) {
$this->parameters[$field] = $initial_parameters[$field];
$changed_parameters = true;
}
}
$format = $this->parameters["FORMAT"];
$packed_table_format = [];
$this->blob_columns = [];
$this->serial_columns = [];
$this->key_field = null;
foreach ($format as $field_name => $type) {
$upper_field_name = strtoupper($field_name);
if ($upper_field_name == "PRIMARY KEY") {
$packed_table_format["PRIMARY KEY"] = $type;
$this->key_field = (is_array($type)) ? $type[0] : $type;
continue;
}
$upper_type = strtoupper($type);
if (in_array($upper_type, ["BLOB", "SERIAL"])) {
$this->blob_columns[] = $field_name;
if ($upper_type == "SERIAL") {
$this->serial_columns[] = $field_name;
}
$packed_table_format[$field_name] = "INT";
} else if (isset(PackedTableTools::TYPE_SYNONYMS[$upper_type])) {
$packed_table_format[$field_name] =
PackedTableTools::TYPE_SYNONYMS[$upper_type];
} else {
return null;
}
}
if (empty($this->key_field)) {
return null;
}
if (!empty($this->blob_columns)) {
$packed_table_format["LAST_BLOB_LEN"] = "INT";
}
$this->table_tools = new PackedTableTools($packed_table_format,
$record_compressor_type);
if ($changed_parameters) {
$this->saveParameters();
}
}
/**
* Returns $fields columns from the record associated with $key in
* the $partition partition of this PartitionDocumentBundle if exists.
* If $fields is empty all columns returned.
*
* @param string $key to look up in partition
* @param int $partition to look for record in
* @param array $fields names of fields in this PartitionDocumentBundle
* to return
* @return array|false unpacked record on success, otherwise false
*/
public function get($key, $partition, $fields = [])
{
$table_tools = $this->table_tools;
$partition_filename = $this->getPartition($partition);
$index = $this->loadPartitionIndex($partition, false,
PackedTableTools::AS_STRING_MODE);
if (is_string($index)) {
$index_data_raw = $table_tools->findRowFromKeyTableString($index,
$key);
} else {
$index_data_raw = $table_tools->find($index, $key);
}
if (empty($index_data_raw)) {
return false;
}
if (!empty($fields)) {
if (isset($fields[0]) && $fields == array_values($fields)) {
$fields = array_combine($fields, $fields);
}
}
$index_data = $table_tools->unpack($index_data_raw)[0];
if (!empty($this->blob_columns)) {
$num_blob_columns = count($this->blob_columns);
$offset = intval($index_data[$this->blob_columns[0]]);
for ($i = 0; $i < $num_blob_columns; $i++) {
$column_name = $this->blob_columns[$i];
$len = ($i + 1 < $num_blob_columns) ?
intval($index_data[$this->blob_columns[$i + 1]]) :
$index_data["LAST_BLOB_LEN"];
if (empty($fields) || isset($fields[$column_name])) {
set_error_handler(null);
$index_data[$column_name] = ($len == 0) ? "" :
@$this->getArchive($partition_filename, $offset, $len);
restore_error_handler();
}
$offset += $len;
}
unset($index_data["LAST_BLOB_LEN"]);
}
if (empty($fields)) {
foreach ($this->serial_columns as $field_name) {
$index_data[$field_name] = unserialize(
$index_data[$field_name]);
}
$index_data[$this->key_field] = $key;
$out_data = $index_data;
} else {
$out_data = [];
if (isset($fields[$this->key_field])) {
$out_data[$fields[$this->key_field]] = $key;
unset($fields[$this->key_field]);
}
set_error_handler(null);
foreach ($fields as $in_name => $out_name) {
if (isset($index_data[$in_name])) {
if (in_array($in_name, $this->serial_columns)) {
$out_data[$out_name] = @unserialize(
$index_data[$in_name]);
} else {
$out_data[$out_name] = $index_data[$in_name];
}
}
}
restore_error_handler();
}
return $out_data;
}
/**
* Cuts a blob length taken from this bundle's index down to
* something it is safe to read: no further than the end of the
* partition file, and no larger than MAX_BLOB_READ_LEN. fread()
* asks PHP for the whole length before it reads a byte, so a
* corrupted length asks for that much memory however small the file
* is, which is enough to end a long-running server. A length past
* the maximum is logged with its file and offset and the read is
* skipped.
*
* @param string $archive_filename path to the partition archive file
* the blob is being read from
* @param int $offset byte offset into that file where the blob starts
* @param int $len length of the blob as the index gives it
* @return int number of bytes it is safe to read, possibly zero
*/
public static function lengthWorthReading($archive_filename, $offset,
$len)
{
if ($len <= 0) {
return 0;
}
if ($len > C\MAX_BLOB_READ_LEN) {
L\crawlLog("PartitionDocumentBundle::getArchive: blob length " .
"$len at offset $offset in $archive_filename exceeds " .
C\MAX_BLOB_READ_LEN . "; treating the entry as corrupt " .
"and skipping the read");
return 0;
}
$size = @filesize($archive_filename);
if ($size === false) {
return $len;
}
$room = $size - $offset;
if ($room <= 0) {
return 0;
}
return min($len, $room);
}
/**
* Retrieve a BLOB string in the file $archive_filename at byte position
* $offset of length $len. It uncompresses this string using
* $compressor->uncompress and return the result.
*
* @param string $archive_filename the filename of a partition archive
* file to get a blob object from
* @param int $offset a byte position in that file
* @param int $len number of bytes from $offset to read.
* @return string the result of uncompressing the string at $offset of
* length $len
*/
public function getArchive($archive_filename, $offset, $len)
{
list($fh, $previous_archive_filename, $previous_instance_time) =
$this->get_archive_cache;
$blob_compressor = $this->blob_compressor;
if (!$fh || $previous_archive_filename != $archive_filename ||
$previous_instance_time != $this->instance_time) {
if ($fh) {
fclose($fh);
}
$fh = fopen($archive_filename , "r");
$previous_archive_filename = $archive_filename;
$previous_instance_time = $this->instance_time;
}
$value = false;
if (fseek($fh, $offset) == 0) {
$len = self::lengthWorthReading($archive_filename, $offset,
$len);
}
if ($len > 0) {
$compressed_file = fread($fh, $len);
$value = $blob_compressor->uncompress($compressed_file);
}
$this->get_archive_cache = [$fh, $previous_archive_filename,
$previous_instance_time];
return $value;
}
/**
* Get the file path in the LSMTree for the block folder the partition $i
* should be in
* @param int $i a partition number to find the block path for
* @return int file path for block folder
*/
public function getPartitionBlock($i)
{
$threshold = $this->parameters["OVERFLOW_THRESHOLD"];
$folder = ($threshold > 0 && $i > $threshold) ?
$this->folder . "/" . self::OVERFLOW_DIR_FOLDER : $this->folder;
$block = sprintf("%'.05d", floor($i / self::BLOCK_FACTOR));
return $folder . "/". self::BLOCK_PREFIX . "$block";
}
/**
* Get the file path in the LSMTree for the partition $i folder
*
* @param int $i a partition number to find the partition folder for
* @return int file path forpartition folder
*/
public function getPartitionFolder($i)
{
return $this->getPartitionBlock($i) . "/" . self::PARTITION_PREFIX .
sprintf("%'.010d", $i);
}
/**
* Returns the path to the archive file (used to store BLOB and SERIAL
* columns) for the $i partition in this PartitionDocumentBundle
*
* @param int $i partition to get the archive file name for
* @return string path of $i partition archive file
*/
public function getPartition($i)
{
return $this->getPartitionFolder($i) . "/" . self::ARCHIVE_FILENAME .
$this->blob_compressor->fileExtension();
}
/**
* Returns the path to the index file (used to store all columns
* a partition record except blob and serial columns) for the $i partition
* in this PartitionDocumentBundle
*
* @param int $i partition to get the index file name for
* @return string path of $i partition index file
*/
public function getPartitionIndex($i)
{
return $this->getPartitionFolder($i) . "/" . self::INDEX_FILENAME;
}
/**
* Returns the unserialized index file for the $partition partition of
* this PartitionDocumentBundle. If $force_load is set to true then reloads
* from disk rather than use a cached value if present.
*
* @param int $partition which partition index to read
* @param bool $force_load whether to reload the index from disk or to
* use a cached value if present
* @param int $mode PackedTableTools mode to use when reading in partition
* @return mixed either a string if $mode as AS_STRING_MODE, or
* array $key => packed records pairs where records are
* packed according to this PartitionDocumentBundle's signature
*/
public function loadPartitionIndex($partition, $force_load = false,
$mode = PackedTableTools::REPLACE_MODE)
{
$index_file_name = $this->getPartitionIndex($partition);
$time = microtime(true);
if (!empty($this->index_cache[$partition]) && !$force_load) {
$index = $this->index_cache[$partition][0];
$this->index_cache[$partition][1] = $time;
} else {
$this->index_cache ??= [];
$cache_size = count($this->index_cache);
$logging = $this->log_partition_work &&
(php_sapi_name() == 'cli') &&
!C\nsdefined("IS_OWN_WEB_SERVER");
if ($logging) {
L\crawlLog("Partition $partition not in cache, adding now...");
L\crawlLog("...Checking for space in current cache");
L\crawlLog(
"...PartitionDocumentBundle cache size: $cache_size");
}
if ($cache_size > $this->index_cache_size) {
$oldest_partition = -1;
$oldest_time = 2 * $time;
foreach ($this->index_cache as $index_partition => $cache_info){
if (empty($cache_info[1])) {
unset($this->index_cache[$index_partition]);
} else if (!empty($cache_info[1]) &&
$cache_info[1] <= $oldest_time) {
$oldest_time = $cache_info[1];
$oldest_partition = $index_partition;
}
}
if ($logging) {
L\crawlLog("...cache too full, evicting ".
"$oldest_partition partition.");
}
unset($this->index_cache[$oldest_partition]);
}
if ($force_load) {
unset($this->index_cache[$partition]);
}
$this->index_cache[$partition] = [
$this->table_tools->load($index_file_name,
$mode), $time];
$index = $this->index_cache[$partition][0];
}
if (empty($index)) {
return false;
}
return $index;
}
/**
* Used to add new records to the PartitionDocumentBundle
*
* @param array $row_or_rows either array of record with fields given
* by this PartitionDocumentBundle's signature or an array of rows.
* @return bool success or not
*/
public function put($row_or_rows)
{
$rows = (empty($row_or_rows[$this->key_field])) ?
$row_or_rows : [$row_or_rows];
$table_tools = $this->table_tools;
$num_rows = count($rows);
$i = 0;
$save_partition = $this->parameters["SAVE_PARTITION"];
// remove $save_partition from read cache
unset($this->index_cache[$save_partition]);
$save_partition_name = $this->getPartition($save_partition);
if (!file_exists($save_partition_name)) {
$save_block_name = $this->getPartitionBlock($save_partition);
if (!file_exists($save_block_name)) {
mkdir($save_block_name);
chmod($save_block_name, 0777);
}
$save_folder_name = $this->getPartitionFolder($save_partition);
if (!file_exists($save_folder_name)) {
mkdir($save_folder_name);
chmod($save_folder_name, 0777);
}
}
clearstatcache();
$save_partition_len = file_exists($save_partition_name) ?
filesize($save_partition_name) : 0;
$step_start = microtime(true);
if (!$this->defer_index_save ||
$this->save_index_partition != $save_partition ||
!isset($this->save_index)) {
$this->save_index = $this->loadPartitionIndex($save_partition,
false, PackedTableTools::AS_STRING_MODE);
}
unset($this->index_cache[$save_partition]);
$this->new_keys_put = 0;
if ($this->save_keys_partition != $save_partition) {
$this->loadSaveKeys($save_partition);
}
L\AnalyticsManager::addTiming("partition index loads", $step_start);
foreach ($rows as $row) {
if ($this->log_partition_work) {
L\crawlTimeoutLog(
"..still adding pages partition document bundle. " .
" Have added %s of %s.", $i, $num_rows);
}
$key = $row[$this->key_field] ?? false;
if ($key === false) {
L\crawlLog("PartitionDocumentBundle Put Failed A");
return false;
}
unset($row[$this->key_field]);
$value = $row;
$memory_limit = L\memoryCeiling(C\MEMORY_FILL_FACTOR);
$exceed_memory = memory_get_usage() > $memory_limit;
/* A partition holds at most MAX_ITEMS_PER_FILE items, which
readers such as WordIterator rely on, so a new partition is
started once the active one holds that many. */
if (($this->parameters['ACTIVE_COUNT'] >=
$this->parameters["MAX_ITEMS_PER_FILE"]) || (
$save_partition_len >
$this->parameters["PARTITION_SIZE_THRESHOLD"]) || (
$exceed_memory && $this->parameters['ACTIVE_COUNT'] >
0.1 * $this->parameters["MAX_ITEMS_PER_FILE"])) {
$active_count = $this->parameters['ACTIVE_COUNT'];
L\crawlLog(
"PartitionDocumentBundle advancing partition because ".
"ACTIVE_COUNT = $active_count and SAVE_PARTITION_LEN = ".
$save_partition_len . " and memory usage = " .
memory_get_usage());
$this->advanceSavePartition();
$save_partition = $this->parameters["SAVE_PARTITION"];
$save_partition_len = 0;
$this->save_index = "";
$index_dirty = false;
} else if ($exceed_memory) {
L\crawlLog("!!Too little memory to index properly!!");
L\crawlLog("!!Memory is " . memory_get_usage() . " but ".
"ACTIVE_COUNT is just " .
$this->parameters['ACTIVE_COUNT']);
}
$blob_columns = $this->blob_columns ?? [];
$serial_columns = $this->serial_columns ?? [];
$format = $this->parameters["FORMAT"];
unset($format['PRIMARY KEY']);
$format_keys = array_keys($format);
$out_value = [];
foreach ($format_keys as $format_key) {
$out_value[$format_key] = $value[$format_key] ?? null;
}
if (!empty($blob_columns)) {
$old_offset = 0;
foreach ($blob_columns as $blob_column) {
$blob = $out_value[$blob_column] ?? "";
/* A fetcher sends a page already serialized and gzip
compressed, so the indexer stores its bytes rather
than compress them again. A bundle that stores
blobs another way gets the page back first. */
$is_compressed = false;
$precompressed = (is_array($blob)) ? ($blob[
L\CrawlConstants::PRECOMPRESSED_PAGE] ?? null) :
null;
if (is_string($precompressed)) {
if ($this->blob_compressor instanceof
GzipCompressor) {
$blob = $precompressed;
$is_compressed = true;
} else {
$blob = unserialize((string)@gzdecode(
$precompressed));
}
}
if (!$is_compressed &&
in_array($blob_column, $serial_columns)) {
$blob = serialize($blob);
}
if (($add_info = $this->addArchive($blob,
$is_compressed)) === false) {
L\crawlLog("PartitionDocumentBundle Put Failed C");
return false;
}
list($offset, $len, $partition) = $add_info;
$save_partition_len += $len;
$out_value[$blob_column] = $offset - $old_offset;
$old_offset = $offset;
}
$out_value["LAST_BLOB_LEN"] = $len;
}
$out_value = $table_tools->pack($out_value);
$is_new_key = !isset($this->save_keys[$key]);
if ($table_tools->add($this->save_index, $key, $out_value,
$table_tools::ADD_MEM_TABLE_STRING)) {
$this->parameters['ACTIVE_COUNT']++;
if ($is_new_key) {
$this->new_keys_put++;
$this->save_keys[$key] = true;
}
} else {
L\crawlLog("PartitionDocumentBundle Put Failed D");
return false;
}
$i++;
}
$this->save_index_partition = $save_partition;
if (!$this->defer_index_save) {
$this->saveIndex();
}
return true;
}
/**
* saveIndex writes the save partition's index that put built, and the
* bundle's parameters, to disk, where readers such as searches find
* them. put calls it at its end unless $defer_index_save is set; the
* indexer then calls it once a fetcher's file is stored.
*/
public function saveIndex()
{
$step_start = microtime(true);
if (isset($this->save_index) && $this->save_index_partition >= 0) {
$this->table_tools->save($this->getPartitionIndex(
$this->save_index_partition), $this->save_index);
}
$this->saveParameters();
L\AnalyticsManager::addTiming(
"partition index and parameter saves", $step_start);
}
/**
* loadSaveKeys builds the set of document keys already stored in a
* partition from that partition's packed index string, and remembers
* which partition it was built for. put reads the set to tell a new
* page from one sent again, so the set is built once when a
* partition becomes the save partition and then kept in step as
* pages are added, rather than the index string being searched for
* every page.
*
* Each entry in the packed string is the key followed by the row's
* packed values, with the two bytes that could be mistaken for the
* entry marker escaped to two bytes each. A key of fixed length is
* therefore held whole in the first two-keys-worth of bytes of its
* entry, so only that much is decoded. A key of variable length
* carries its length as its first byte and is decoded whole.
*
* @param int $partition number of the partition whose keys to load,
* which is the save partition
*/
public function loadSaveKeys($partition)
{
$this->save_keys = [];
$this->save_keys_partition = $partition;
if (!is_string($this->save_index) || $this->save_index === "") {
return;
}
$key_len = $this->table_tools->key_len;
$entries = explode("\xFF", $this->save_index);
foreach ($entries as $entry) {
if ($entry === "") {
continue;
}
if ($key_len > 0) {
$key = substr(L\decode255(substr($entry, 0,
2 * $key_len)), 0, $key_len);
} else {
$decoded = L\decode255($entry);
$key = substr($decoded, 1, ord($decoded[0]));
}
$this->save_keys[$key] = true;
}
}
/**
* Saves the current save partition, adds one to the save partition number,
* and starts a new save partition.
*
* @param int $new_save_partition partition and add one to. If use default,
* then this method will use the parameters "SAVE_PARTITION"
* value.
* @return bool false if the requested partition index is not larger
* than the current SAVE_PARTITION (no advance happens); void on
* the success path
*/
public function advanceSavePartition($new_save_partition = 0)
{
$save_partition = $this->parameters["SAVE_PARTITION"];
$new_save_partition = (($new_save_partition) > 0) ?
$new_save_partition : $save_partition + 1;
if ($new_save_partition <= $save_partition) {
return false;
}
$save_index_name = $this->getPartitionIndex($save_partition);
if (isset($this->save_index)) {
$this->table_tools->save($save_index_name, $this->save_index);
}
$new_save_index_name = $this->getPartitionIndex($new_save_partition);
if (file_exists($new_save_index_name)) {
unlink($new_save_index_name);
}
$new_save_file_name = $this->getPartition($new_save_partition);
if (file_exists($new_save_file_name)) {
unlink($new_save_file_name);
}
$new_save_block_name = $this->getPartitionBlock($new_save_partition);
if (!file_exists($new_save_block_name)) {
mkdir($new_save_block_name);
chmod($new_save_block_name, 0777);
}
$new_save_folder_name = $this->getPartitionFolder($new_save_partition);
if (!file_exists($new_save_folder_name)) {
mkdir($new_save_folder_name);
chmod($new_save_folder_name, 0777);
}
$this->parameters["SAVE_PARTITION"] = $new_save_partition;
$this->parameters['COUNT'] += $this->parameters['ACTIVE_COUNT'];
$this->parameters['ACTIVE_COUNT'] = 0;
/* The new partition holds no documents yet, so the set of keys
put keeps for the save partition starts empty. */
$this->save_keys = [];
$this->save_keys_partition = $new_save_partition;
return true;
}
/**
* Creates a new counter $field to be maintained
*
* @param string $field field of info struct to add a counter for
*/
public function initCountIfNotExists($field = "COUNT")
{
$this->parameters[$field] ??= 0;
$this->saveParameters();
}
/**
* Add $num to maintained counter $field
*
* @param int $num number of items to add to current count
* @param string $field field of info struct to add to the count of
*/
public function addCount($num, $field = "COUNT")
{
$this->parameters[$field] ??= 0;
$this->parameters[$field] += $num;
$this->saveParameters();
}
/**
* saveParameters writes this bundle's parameters, such as its record
* count and the partition being written, to its parameters file. Every
* change to a count or to the partition calls it, so the file always
* holds what the next process to open the bundle should start from.
*/
public function saveParameters()
{
$parameter_path = $this->folder . "/" . self::PARAMETERS_FILE;
/* The parameters are written to a file of their own and renamed
over the old one, which replaces it in one step. Writing in place
empties the file first, and a rebuild opens the bundle in several
processes at once, so one that read the file in that moment took
the counts as missing and saved defaults over them. The new file
sits in the same folder because a rename is only a single step
within one file system. */
$new_path = $parameter_path . "." . getmypid() . "." . hrtime(true);
file_put_contents($new_path, serialize($this->parameters));
rename($new_path, $parameter_path);
}
/**
* Returns the parameters (such as its signature, max number of
* documents per partition and counts) used to configure the
* PartitionDocumentBundle stored at $folder
*
* @param string $folder file path to a stored PartitionDocumentBundle
* @return array configuration info about the PartitionDocumentBundle
*/
public static function getParameterInfo($folder)
{
$parameter_path = $folder . "/" . self::PARAMETERS_FILE;
if (file_exists($parameter_path)) {
$parameters = unserialize(file_get_contents($parameter_path));
if (!is_array($parameters)) {
$parameters = [];
}
if (!empty($parameters["COMPRESSOR"])) {
//original format didn't distinguish between compressor use
$parameters["RECORD_COMPRESSOR"] ??= $parameters["COMPRESSOR"];
$parameters["BLOB_COMPRESSOR"] ??= $parameters["COMPRESSOR"];
}
if (empty($parameters['SAVE_PARTITION']) ||
$parameters['SAVE_PARTITION'] == 0) {
$block_folders = glob("$folder/" . self::BLOCK_PREFIX . "*");
$parameters['SAVE_PARTITION'] = 0;
if (!empty($block_folders)) {
$overflow_block_folders = glob("$folder/" .
self::OVERFLOW_DIR_FOLDER . "/" .
self::BLOCK_PREFIX . "*");
if (!empty($overflow_block_folders) &&
count($overflow_block_folders) > 0) {
$block_folders = $overflow_block_folders;
}
sort($block_folders);
$last_folder = $block_folders[count($block_folders) - 1];
$partition_path = "$last_folder/" . self::PARTITION_PREFIX;
$partition_folders = glob("$partition_path*");
/* A block folder can exist before any partition has
been written into it, so the list can be empty;
the save partition then stays at zero rather than
being read from a folder name that is not there. */
if (!empty($partition_folders)) {
sort($partition_folders);
$last_index_file = $partition_folders[
count($partition_folders) - 1];
$parameters['SAVE_PARTITION'] = intval(substr(
$last_index_file, strlen($partition_path)));
}
}
}
return $parameters;
} else {
return [];
}
}
/**
* addArchive adds a blob item to the current save partition file,
* compressing it with the bundle's blob compressor unless it comes
* compressed already.
*
* @param string $value blob item to be added to file
* @param bool $is_compressed whether $value is already compressed the
* way the blob compressor would, as a fetcher sends a page
* @return array [offset into save partition, length stored,
* partition number OF current save partition]
*/
protected function addArchive($value, $is_compressed = false)
{
list($fh, $previous_partition_filename,
$previous_instance_time) = $this->add_archive_cache;
$blob_compressor = $this->blob_compressor;
$save_partition = $this->parameters["SAVE_PARTITION"];
$partition_filename = $this->getPartition($save_partition);
if (!is_resource($fh) ||
$partition_filename != $previous_partition_filename ||
$previous_instance_time != $this->instance_time) {
if (!empty($fh) && is_resource($fh)) {
fclose($fh);
}
$fh = fopen($partition_filename , "c+");
$previous_partition_filename = $partition_filename;
$previous_instance_time = $this->instance_time;
}
fseek($fh, 0, SEEK_END);
$offset = ftell($fh);
$step_start = microtime(true);
$compressed_value = ($is_compressed) ? $value :
$blob_compressor->compress($value);
$len = strlen($compressed_value);
$step_start = L\AnalyticsManager::addTiming(
"archive items compressed", $step_start);
fwrite($fh, $compressed_value, $len);
L\AnalyticsManager::addTiming("archive items written", $step_start);
$this->add_archive_cache = [$fh, $previous_partition_filename,
$previous_instance_time];
return [$offset, $len, $save_partition];
}
}
- Yioop Repo - Seekquarry