cacti/lib/dbparallel.php

<?php
/*
 +-------------------------------------------------------------------------+
 | Copyright (C) 2004-2026 The Cacti Group                                 |
 |                                                                         |
 | 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 2          |
 | 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.                            |
 +-------------------------------------------------------------------------+
 | Cacti: The Complete RRDtool-based Graphing Solution                     |
 +-------------------------------------------------------------------------+
 | This code is designed, written, and maintained by the Cacti Group. See  |
 | about.php and/or the AUTHORS file for specific developer information.   |
 +-------------------------------------------------------------------------+
 | http://www.cacti.net/                                                   |
 +-------------------------------------------------------------------------+
*/

/**
 * db_fetch_cell_parallel - run a 'select' sql query and return the first column of the
 * first row found
 *
 * @param string $sql      The SQL query to execute
 * @param string $col_name Use this column name instead of the first one
 * @param int    $threads  The number of parallel threads to use
 * @param bool   $log      Whether to log error messages, defaults to true
 * @param mixed  $db_conn  The connection to use or false to use the default
 *
 * @return mixed The output of the sql query as a single variable
 */
function db_fetch_cell_parallel(string $sql, string $col_name = '', int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_cell($sql, $col_name = \'' . $col_name . '\', $log = true, $db_conn = false)' . "\n");
	}

	return db_fetch_cell_prepared($sql, [], $col_name, $log, $db_conn);
}

/**
 * db_fetch_cell_parallel_prepared - run a 'select' sql query and return the first column of the
 * first row found
 *
 * @param string $sql      The SQL query to execute
 * @param array  $params   An array of values to be prepared into the SQL
 * @param string $col_name Use this column name instead of the first one
 * @param int    $threads  The number of parallel threads to use
 * @param bool   $log      Whether to log error messages, defaults to true
 * @param mixed  $db_conn  The connection to use or false to use the default
 *
 * @return mixed The output of the sql query as a single variable
 */
function db_fetch_cell_parallel_prepared(string $sql, array $params = [], string $col_name = '', int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_cell_prepared($sql, $params = ' . clean_up_lines(var_export($params, true)) . ', $col_name = \'' . $col_name . '\', $log = true, $db_conn = false)' . "\n");
	}

	return db_execute_prepared($sql, $params, $log, $db_conn, 'Cell', false, 'db_fetch_cell_return', $col_name);
}

/**
 * db_fetch_row_parallel - run a 'select' sql query and return the first row found
 *
 * @param string $sql     The SQL query to execute
 * @param int    $threads The number of parallel threads to use
 * @param bool   $log     Whether to log error messages, defaults to true
 * @param mixed  $db_conn The connection to use or false to use the default
 *
 * @return mixed The first row of the result or false if failed
 */
function db_fetch_row_parallel(string $sql, int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_row(\'' . clean_up_lines($sql) . '\', $log = ' . $log . ', $db_conn = ' . ($db_conn ? 'true' : 'false') . ')' . "\n");
	}

	return db_fetch_row_prepared($sql, [], $log, $db_conn);
}

/**
 * db_fetch_row_parallel_prepared - run a 'select' sql query and return the first row found
 *
 * @param string $sql     The SQL query to execute
 * @param array  $params  An array of values to be prepared into the SQL
 * @param int    $threads The number of parallel threads to use
 * @param bool   $log     Whether to log error messages, defaults to true
 * @param mixed  $db_conn The connection to use or false to use the default
 *
 * @return mixed The first row of the result or false if failed
 */
function db_fetch_row_parallel_prepared(string $sql, array $params = [], int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_row_prepared(\'' . clean_up_lines($sql) . '\', $params = (\'' . implode('\', \'', $params) . '\'), $log = ' . $log . ', $db_conn = ' . ($db_conn ? 'true' : 'false') . ')' . "\n");
	}

	return db_execute_prepared($sql, $params, $log, $db_conn, 'Row', false, 'db_fetch_row_return');
}

/**
 * db_fetch_assoc_parallel - run a 'select' sql query and return all rows found
 *
 * @param string $sql     The SQL query to execute
 * @param int    $threads The number of parallel threads to use
 * @param bool   $log     Whether to log error messages, defaults to true
 * @param mixed  $db_conn The connection to use or false to use the default
 *
 * @return mixed The entire result set or false on error
 */
function db_fetch_assoc_parallel(string $sql, int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_assoc($sql, $log = true, $db_conn = false)' . "\n");
	}

	return db_fetch_assoc_prepared($sql, [], $log, $db_conn);
}

/**
 * db_fetch_assoc_parallel_prepared - run a 'select' sql query and return all rows found
 *
 * @param string $sql     The sql query to execute
 * @param array  $params  An array of values to be prepared into the SQL
 * @param int    $threads The number of parallel threads to use
 * @param bool   $log     Whether to log error messages, defaults to true
 * @param mixed  $db_conn The connection to use or false to use the default
 *
 * @return mixed The entire result or false on error
 */
function db_fetch_assoc_parallel_prepared(string $sql, array $params = [], int $threads = 2, bool $log = true, mixed $db_conn = false) : mixed {
	global $config;

	if (!empty($config['DEBUG_SQL_FLOW'])) {
		db_echo_sql('db_fetch_assoc_prepared($sql, $params = array(), $log = true, $db_conn = false)' . "\n");
	}

	return db_execute_prepared($sql, $params, $log, $db_conn, 'Row', [], 'db_fetch_assoc_return');
}

/**
 * db_query_parallelize - break a query into pieces to parallelize and return
 * components to the parent.  It will only work with very simple level of
 * parallelism around both UNION and PARTITION queries including only
 * certain aggregation functions.
 *
 * Cases that are not currently supported are:
 * - Queries with aggregation other than MIN, MAX, AVG, SUM, and COUNT
 *
 * More specifically, it will not handle the following aggregators:
 * - BIT_OR, BIT_AND, or BIT_XOR
 * - GROUP_CONCAT
 * - COUNT DISTINCT
 * - JSON*
 * - STD, STDDEV, STDDEV_POP, STDDEV_SAMP
 * - VARIANCE, VAR_POP, VAR_SAMP
 *
 * At some point, additional support may be added, but at a later date
 *
 * The components that are returned are as follows:
 *
 * $response = array(
 *    'error'              => boolean true or false
 *    'error_message'      => a message to the caller about the error
 *    'temp_table_name'    => string of the temp table to be removed after query complete
 *    'temp_create_syntax' => '',
 *    'reduce_query'       => string of the final reduce query
 *    'map_queries'        => an array of map queries decomposed from the original query
 * );
 *
 * @param string $sql     The query to execute
 * @param bool   $log     Whether to log error messages, defaults to true
 * @param mixed  $db_conn The connection to use or false to use the default
 *
 * @return array The query pieces, the map and reduce components with
 *               and error and error_message
 */
function db_query_parallelize(string $sql, bool $log = true, mixed $db_conn = false) : array {
	global $debug;

	/**
	 * For simplicity, In our tokenization of the query, we will ignore
	 * any query that includes these custom keyword options.
	 *
	 */
	$ignores = [
		'DISTINCTROW',
		'HIGH_PRIORITY',
		'SQL_SMALL_RESULT',
		'SQL_BIG_RESULT',
		'SQL_BUFFER_RESULT',
		'SQL_CACHE',
		'SQL_NO_CACHE',
		'SQL_CALC_FOUND_ROWS',
		'FOR UPDATE',
		'LOCK IN SHARE MODE'
	];

	$replaces = ['', '', '', '', '', '', '', '', ''];

	$sql = str_replace($ignores, $replaces, $sql);

	while (substr_count($sql, '  ')) {
		$sql = str_replace('  ', ' ', $sql);
	}

	/**
	 * Additionally, we run a quick and dirty Query cleanup
	 * to remove some extra spaces, line breaks and tabs that
	 * may compromise the further tokenization from
	 * being handled properly.
	 *
	 * I know this will break some exotic queries,
	 * and should in general be avoided, but I'm writing
	 * this and not you ;)
	 *
	 * This cleanup has two phases.
	 * - Remove some odd formatting, line breaks and tabs
	 * - Remove extra spaces
	 */
	$sql = str_replace(
		["\n", "\t", "\r", ';'],
		[' ', ' ', ' ', ''],
		$sql
	);

	while (substr_count($sql, '  ')) {
		$sql = str_replace('  ', ' ', $sql);
	}

	if ($debug) {
		print PHP_EOL . 'Pre-processed Query: ' . $sql . PHP_EOL . PHP_EOL;
	}

	/**
	 * In this prototype, we will reject a number of allowed SQL forms
	 * to simply keep it simple for the moment.  In the future, we may support
	 * more elaborate forms of SQL based upon a valid feature request.
	 */
	$disallowed =
		// Import and export
		'/(INFILE ' . // Import Option
		'|OUTFILE ' . // Export Option
		'|INTO DUMPFILE ' . // Export Option
		'|INTO ' . // Export Option

		// Aggregation Functions
		'|VARIANCE' . // Aggregation Function
		'|VAR_' . // Aggregation Function
		'|STD' . // Aggregation Function
		'|STDDEV' . // Aggregation Function
		'|COUNT\(DISTINCT' . // Aggregation Function

		// Oddball Query Options
		'|LOCK IN SHARE MODE ' . // Query Option
		'|FOR UPDATE ' . // Update Option
		'|FOR SYSTEM TIME ' . // System Table Support
		'|PROCEDURE ' . // Stored Procedures
		'|PARTITION ' . // PARTITION specific query

		// JSON Functions
		'|JSON_' . // Any JSON Functions

		// Index Options
		'|FORCE ' . // FORCE Index
		'|IGNORE ' . // IGNORE Index
		'| USE )/i';             // USE Index

	/**
	 * Track UNION queries as these queries must be handled using
	 * extra logic to properly move the WHERE clause into the
	 * map queries to improve performance of those map queries.
	 */
	$is_union = false;

	/**
	 * Our response object which can then be used by the caller
	 * to create the table, perform queries, etc.
	 */
	$response = [
		'error'              => false,
		'error_message'      => '',
		'temp_table_name'    => 'par_' . generate_hash() . '_res', // Limited to 40 characters due to process table length limit
		'temp_create_syntax' => '',
		'reduce_query'       => '',
		'map_queries'        => []
	];

	// Check for disallowed tokens
	if (preg_match($disallowed, $sql)) {
		$response['error']         = true;
		$response['error_message'] = 'The SQL Query included one of the disallowed SQL options, functions, or modifiers.';

		return $response;
	}

	// Create a split on the FROM clause
	$from_split   = explode('FROM ', $sql, 2);

	/**
	 * We need to track the partitions for operating on partitioned
	 * tables for the MAP phase.
	 */
	$partitions = [];

	/**
	 * Perform some pre-checks, UNION queries, by their nature are parallelizable.
	 * However, queries without UNIONS must come from a partitioned tables.
	 */
	if (stripos($sql, 'UNION') === false && stripos($sql, 'UNION ALL') === false) {
		// Find the table name if not a union query
		if (count($from_split) > 1) {
			$bits       = explode(' ', trim($from_split[1]));
			$table_name = trim($bits[0], '`');

			if (!db_table_partitioned($table_name, $log, $db_conn, $partitions)) {
				$response['error']         = true;
				$response['error_message'] = 'The SQL query main table is not partitioned.';

				return $response;
			}
		} else {
			$response['error']         = true;
			$response['error_message'] = 'The SQL query lacks a FROM clause.';

			return $response;
		}
	} else {
		$is_union     = true;
		$is_union_all = (stripos($sql, 'UNION ALL') !== false) ? true : false;

		/**
		 * Here we check if the UNION is properly nested where the
		 * GROUP BY, HAVING, LIMIT, etc. are processed inside or outside
		 * the table select/from logic. If outside or simple, then we need
		 * to skip the checks below GROUP BY, LIMIT, ORDER, etc. below as
		 * they are each contained within the UNION's themselves.
		 */
		if (str_starts_with(trim($from_split[1]), '(')) {
			$simple_union = false;
		} else {
			$simple_union = true;
		}
	}

	/**
	 * Query clauses must come in a specific order.  That order is always the same, though
	 * with sub-queries, it can get a bit more complicated.  Here is the general
	 * order. We are not supporting Sub-Queries with the exception of UNION queries.
	 * So, we will make broad assumptions about the structure of the queries and
	 * reject them otherwise.
	 *
	 * SELECT
	 * FROM
	 * JOIN(S)
	 * WHERE
	 * GROUP BY [ WITH ROLLUP ]
	 * HAVING
	 * ORDER BY
	 * LIMIT
	 *
	 * Knowing this information, we can tokenize the SQL
	 * statement and prepare for parallelization.  The MySQL/MariaDB language
	 * syntax has a very many modifiers that we hope not to have to cover
	 * in this exercise.  They include, but are not limited to the items included
	 * in the following two links:
	 *
	 * Select Syntax:
	 * https://mariadb.com/kb/en/select/
	 *
	 * Table References and Join Options:
	 * https://mariadb.com/kb/en/join-syntax/
	 *
	 */

	$tokens = [
		'raw_select'   => '',
		'outer_select' => '',
		'inner_select' => '',
		'from'         => '',
		'where'        => '',
		'groupby'      => '',
		'having'       => '',
		'orderby'      => '',
		'limit'        => ''
	];

	/**
	 * Tokenizing the SQL query can be handled in many ways.  I'm taking the
	 * approach of handing UNION queries separately from PARTITIONED queries
	 * and I'll be working in reverse breaking the query into the various
	 * tokens.
	 *
	 * I know that this may be inneficient, but it the way I will handle
	 * for now.  At some point it can be improved/optimized.
	 */
	if (!$is_union || ($is_union && !$simple_union)) {
		$selectstr = $from_split[0];
		$tokenstr  = $from_split[1];

		if (stripos($tokenstr, 'LIMIT') !== false) {
			$tempstr = str_ireplace('limit', 'LIMIT', $tokenstr);
			$parts   = explode('LIMIT', $tempstr);

			$elements        = cacti_sizeof($parts);
			$tokens['limit'] = 'LIMIT ' . trim($parts[$elements - 1]);

			unset($parts[$elements - 1]);

			$tokenstr = trim(implode('LIMIT', $parts));
		}

		if (stripos($tokenstr, 'ORDER BY') !== false) {
			$tempstr = str_ireplace('order by', 'ORDER BY', $tokenstr);
			$parts   = explode('ORDER BY', $tempstr);

			$elements          = cacti_sizeof($parts);
			$tokens['orderby'] = 'ORDER BY ' . trim($parts[$elements - 1]);

			unset($parts[$elements - 1]);

			$tokenstr = trim(implode('ORDER BY', $parts));
		}

		if (stripos($tokenstr, 'HAVING') !== false) {
			$tempstr = str_ireplace('having', 'HAVING', $tokenstr);
			$parts   = explode('HAVING', $tempstr);

			$elements          = cacti_sizeof($parts);
			$tokens['having']  = 'HAVING ' . trim($parts[$elements - 1]);

			unset($parts[$elements - 1]);

			$tokenstr = trim(implode('HAVING', $parts));
		}

		if (stripos($tokenstr, 'GROUP BY') !== false) {
			$tempstr = str_ireplace('group by', 'GROUP BY', $tokenstr);
			$parts   = explode('GROUP BY', $tempstr);

			$elements          = cacti_sizeof($parts);
			$tokens['groupby'] = 'GROUP BY ' . trim($parts[$elements - 1]);

			unset($parts[$elements - 1]);

			$tokenstr = trim(implode('GROUP BY', $parts));
		}

		if (stripos($tokenstr, 'WHERE') !== false) {
			$tempstr = str_ireplace('where', 'WHERE', $tokenstr);
			$parts   = explode('WHERE', $tempstr);

			$elements          = cacti_sizeof($parts);
			$tokens['where']   = 'WHERE ' . trim($parts[$elements - 1]);

			unset($parts[$elements - 1]);

			$tokenstr = trim(implode('WHERE', $parts));
		}

		$tokens['from'] = 'FROM ' . trim($tokenstr);

		// Format the raw inner and outer selectors
		if (stripos($selectstr, 'SELECT') !== false) {
			$tempstr              = str_ireplace('select', '', $selectstr);
			$tokens['raw_select'] = trim($tempstr);
		}

		// Modify the outer selector to remove aliases
		$outer = explode(',', $tokens['raw_select']);
		$inner = '';

		foreach ($outer as $index => $o) {
			$aggregate = db_get_aggregate($o);

			if ($aggregate != '' && $debug) {
				print 'Aggregate: ' . $aggregate . PHP_EOL;
			}

			if ($aggregate == '') {
				$parts = explode('.', $o);

				if (cacti_sizeof($parts) > 1) {
					if (stripos($parts[1], 'AS') !== false) {
						$parts[1]      = str_ireplace('as', 'AS', $parts[1]);
						$bits          = explode('AS', $parts[1]);
						$outer[$index] = trim($bits[1]);
					} else {
						$outer[$index] = trim($parts[1]);
					}
				} elseif (stripos($o, 'AS') !== false) {
					$o             = str_ireplace('as', 'AS', $o);
					$bits          = explode('AS', $o);
					$outer[$index] = trim($bits[1]);
				} else {
					$outer[$index] = trim($parts[0]);
				}
			} else {
				$alias = $index;

				if (stripos($o, 'AS') !== false) {
					$o     = str_ireplace('as', 'AS', $o);
					$bits  = explode('AS', $o);
					$alias = trim($bits[1]);
				}

				if ($aggregate == 'AVG') {
					// Handle AVG Aggregator
				} elseif ($aggregate == 'COUNT') {
					// Handle COUNT Aggregator
				} else {
					// Handle Aggregators MIN, MAX, SUM
				}
			}

			$inner .= ($inner != '' ? ', ' : '') . $outer[$index];
		}

		// Handle UNIONS into Partitions

		// Determine CREATE Table Syntax from UNION or PARTITION columns

		$tokens['outer_select'] = implode(',', $outer);
	} else {
		$sql = str_ireplace(['union all', 'union'], ['UNION ALL', 'UNION'], $sql);

		if ($is_union_all) {
			$partitions = explode('UNION ALL', $sql);
		} else {
			$partitions = explode('UNION', $sql);
		}

		if (cacti_sizeof($partitions)) {
			foreach ($partitions as $index => $p) {
				$partitions[$index] = trim($p);
			}
		}

		$tokens['raw_select']   = '*';
		$tokens['outer_select'] = '*';

		// Determine CREATE Table Syntax from Partition logic
	}

	if ($debug) {
		print 'SQL Tokens:';
		print PHP_EOL;
		print_r($tokens);

		print PHP_EOL;
		print 'Partitions:';
		print PHP_EOL;
		print_r($partitions);
	}

	return ['tokens' => $tokens, 'partitions' => $partitions];
}

/**
 * db_get_aggregate - Return the aggregate function from a SQL select item
 *
 * @param string $colsql The SQL column string
 *
 * @return string The column aggregate function
 */
function db_get_aggregate(string $colsql) : string {
	// Decompose and prepare the query for parallelization
	$aggregators = [
		'SUM(',
		'COUNT(',
		'MIN(',
		'MAX(',
		'AVG('
	];

	foreach ($aggregators as $a) {
		if (stripos($colsql, $a) !== false) {
			return trim($a, '(');
		}
	}

	return '';
}

/**
 * db_table_partitioned - checks whether a table is partitioned and returns partition numbers
 *
 * @param string $table      The name of the table
 * @param bool   $log        Whether to log error messages, defaults to true
 * @param mixed  $db_conn    The connection to use or false to use the default
 * @param array  $partitions The connection to use or false to use the default
 *
 * @return bool The output of the sql query as a single variable
 */
function db_table_partitioned(string $table, bool $log = true, mixed $db_conn = false, array &$partitions = []) : bool {
	static $results;

	if (isset($results[$table]) && !defined('IN_CACTI_INSTALL') && !defined('IN_PLUGIN_INSTALL')) {
		$partitions = $results[$table]['partitions'];

		return $results[$table]['result'];
	}

	// Separate the database from the table and remove backticks
	preg_match("/([`]{0,1}(?<database>[\w_]+)[`]{0,1}\.){0,1}[`]{0,1}(?<table>[\w_]+)[`]{0,1}/", $table, $matches);

	if (cacti_sizeof($matches) && db_table_exists($table, false, $db_conn) && array_key_exists('table', $matches)) {
		$sql = 'SHOW CREATE TABLE ' . $matches['table'];

		$create = db_fetch_row($sql, $log, $db_conn);

		if (cacti_sizeof($create)) {
			$create = $create['Create Table'];
		}

		if (stripos($create, 'PARTITION ') !== false) {
			$results[$table]['result'] = true;

			$parts = explode("\n", $create);

			foreach ($parts as $l) {
				if (stripos($l, 'PARTITION ') !== false && stripos($l, 'PARTITION BY') === false) {
					$comp         = explode(' ', trim($l));
					$partitions[] = trim($comp[1], '`');
				}
			}

			$results[$table]['partitions'] = $partitions;
		} else {
			$results[$table]['partitions'] = [];
			$results[$table]['result']     = false;
		}

		return $results[$table]['result'];
	}

	return false;
}