1<?php 2 3namespace dokuwiki\plugin\struct\meta; 4 5use dokuwiki\plugin\struct\types\DateTime; 6use dokuwiki\plugin\struct\types\Decimal; 7use dokuwiki\plugin\struct\types\Page; 8use dokuwiki\plugin\struct\types\AutoSummary; 9use dokuwiki\plugin\struct\types\Text; 10use dokuwiki\plugin\struct\types\User; 11 12class Search { 13 /** 14 * This separator will be used to concat multi values to flatten them in the result set 15 */ 16 const CONCAT_SEPARATOR = "\n!_-_-_-_-_!\n"; 17 18 /** 19 * The list of known and allowed comparators 20 * (order matters) 21 */ 22 static public $COMPARATORS = array( 23 '<=', '>=', '=*', '=', '<', '>', '!=', '!~', '~', 'IN' 24 ); 25 26 /** @var \helper_plugin_sqlite */ 27 protected $sqlite; 28 29 /** @var Schema[] list of schemas to query */ 30 protected $schemas = array(); 31 32 /** @var Column[] list of columns to select */ 33 protected $columns = array(); 34 35 /** @var array the sorting of the result */ 36 protected $sortby = array(); 37 38 /** @var array the filters */ 39 protected $filter = array(); 40 41 /** @var array list of aliases tables can be referenced by */ 42 protected $aliases = array(); 43 44 /** @var int begin results from here */ 45 protected $range_begin = 0; 46 47 /** @var int end results here */ 48 protected $range_end = 0; 49 50 /** @var int the number of results */ 51 protected $count = -1; 52 /** @var string[] the PIDs of the result rows */ 53 protected $result_pids = null; 54 /** @var array the row ids of the result rows */ 55 protected $result_rids = []; 56 /** @var array the revisions of the result rows */ 57 protected $result_revs = []; 58 59 /** 60 * Search constructor. 61 */ 62 public function __construct() { 63 /** @var \helper_plugin_struct_db $plugin */ 64 $plugin = plugin_load('helper', 'struct_db'); 65 $this->sqlite = $plugin->getDB(); 66 } 67 68 /** 69 * Add a schema to be searched 70 * 71 * Call multiple times for multiple schemas. 72 * 73 * @param string $table 74 * @param string $alias 75 */ 76 public function addSchema($table, $alias = '') { 77 $schema = new Schema($table); 78 if(!$schema->getId()) { 79 throw new StructException('schema missing', $table); 80 } 81 82 // FIXME is the mixing still relevant? 83// if($this->schemas && 84// ( 85// $schema->isLookup() || 86// reset($this->schemas)->isLookup() 87// ) 88// ) { 89// throw new StructException('nolookupmix'); 90// } 91 92 $this->schemas[$schema->getTable()] = $schema; 93 if($alias) $this->aliases[$alias] = $schema->getTable(); 94 } 95 96 /** 97 * Add a column to be returned by the search 98 * 99 * Call multiple times for multiple columns. Be sure the referenced tables have been 100 * added before 101 * 102 * @param string $colname may contain an alias 103 */ 104 public function addColumn($colname) { 105 if($this->processWildcard($colname)) return; // wildcard? 106 $col = $this->findColumn($colname); 107 if(!$col) return; //FIXME do we really want to ignore missing columns? 108 $this->columns[] = $col; 109 } 110 111 /** 112 * Add sorting options 113 * 114 * Call multiple times for multiple columns. Be sure the referenced tables have been 115 * added before 116 * 117 * @param string $colname may contain an alias 118 * @param bool $asc sort direction (ASC = true, DESC = false) 119 * @param bool $nc set true for caseinsensitivity 120 */ 121 public function addSort($colname, $asc = true, $nc = true) { 122 $col = $this->findColumn($colname); 123 if(!$col) return; //FIXME do we really want to ignore missing columns? 124 125 $this->sortby[$col->getFullQualifiedLabel()] = array($col, $asc, $nc); 126 } 127 128 /** 129 * Returns all set sort columns 130 * 131 * @return array 132 */ 133 public function getSorts() { 134 return $this->sortby; 135 } 136 137 /** 138 * Adds a filter 139 * 140 * @param string $colname may contain an alias 141 * @param string|string[] $value 142 * @param string $comp @see self::COMPARATORS 143 * @param string $op either 'OR' or 'AND' 144 */ 145 public function addFilter($colname, $value, $comp, $op = 'OR') { 146 /* Convert certain filters into others 147 * this reduces the number of supported filters to implement in types */ 148 if($comp == '*~') { 149 $value = $this->filterWrapAsterisks($value); 150 $comp = '~'; 151 } elseif($comp == '<>') { 152 $comp = '!='; 153 } 154 155 if(!in_array($comp, self::$COMPARATORS)) throw new StructException("Bad comperator. Use " . join(',', self::$COMPARATORS)); 156 if($op != 'OR' && $op != 'AND') throw new StructException('Bad filter type . Only AND or OR allowed'); 157 158 $col = $this->findColumn($colname); 159 if(!$col) return; // ignore missing columns, filter might have been for different schema 160 161 // map filter operators to SQL syntax 162 switch($comp) { 163 case '~': 164 $comp = 'LIKE'; 165 break; 166 case '!~': 167 $comp = 'NOT LIKE'; 168 break; 169 case '=*': 170 $comp = 'REGEXP'; 171 break; 172 } 173 174 // we use asterisks, but SQL wants percents 175 if($comp == 'LIKE' || $comp == 'NOT LIKE') { 176 $value = $this->filterChangeToLike($value); 177 } 178 179 if ($comp == 'IN' && !is_array($value)) { 180 $value = $this->parseFilterValueList($value); 181 //col IN ('a', 'b', 'c') is equal to col = 'a' OR 'col = 'b' OR col = 'c' 182 $comp = '='; 183 } 184 185 // add the filter 186 $this->filter[] = array($col, $value, $comp, $op); 187 } 188 189 /** 190 * Parse SQLite row value into array 191 * 192 * @param string $value 193 * @return string[] 194 */ 195 protected function parseFilterValueList($value) { 196 $Handler = new FilterValueListHandler(); 197 $Lexer = new \Doku_Lexer($Handler, 'base', true); 198 199 $Lexer->addEntryPattern('\(','base','row'); 200 $Lexer->addPattern('\s*,\s*','row'); 201 $Lexer->addExitPattern('\)','row'); 202 203 $Lexer->addEntryPattern('"', 'row', 'double_quote_string'); 204 $Lexer->addSpecialPattern('\\\\"','double_quote_string','escape_sequence'); 205 $Lexer->addExitPattern('"', 'double_quote_string'); 206 207 $Lexer->addEntryPattern("'", 'row', 'single_quote_string'); 208 $Lexer->addSpecialPattern("\\\\'",'single_quote_string','escape_sequence'); 209 $Lexer->addExitPattern("'", 'single_quote_string'); 210 211 $Lexer->mapHandler('double_quote_string','single_quote_string'); 212 213 $Lexer->addSpecialPattern('[-+]?[0-9]*\.?[0-9]+(?:[eE][-+]?[0-9]+)?','row','number'); 214 215 $res = $Lexer->parse($value); 216 217 if (!$res || $Lexer->_mode->getCurrent() != 'base') { 218 throw new StructException('invalid row value syntax'); 219 } 220 221 return $Handler->get_row(); 222 } 223 224 /** 225 * Wrap given value in asterisks 226 * 227 * @param string|string[] $value 228 * @return string|string[] 229 */ 230 protected function filterWrapAsterisks($value) { 231 $map = function ($input) { 232 return "*$input*"; 233 }; 234 235 if(is_array($value)) { 236 $value = array_map($map, $value); 237 } else { 238 $value = $map($value); 239 } 240 return $value; 241 } 242 243 /** 244 * Change given string to use % instead of * 245 * 246 * @param string|string[] $value 247 * @return string|string[] 248 */ 249 protected function filterChangeToLike($value) { 250 $map = function ($input) { 251 return str_replace('*', '%', $input); 252 }; 253 254 if(is_array($value)) { 255 $value = array_map($map, $value); 256 } else { 257 $value = $map($value); 258 } 259 return $value; 260 } 261 262 /** 263 * Set offset for the results 264 * 265 * @param int $offset 266 */ 267 public function setOffset($offset) { 268 $limit = 0; 269 if($this->range_end) { 270 // if there was a limit set previously, the range_end needs to be recalculated 271 $limit = $this->range_end - $this->range_begin; 272 } 273 $this->range_begin = $offset; 274 if($limit) $this->setLimit($limit); 275 } 276 277 /** 278 * Limit results to this number 279 * 280 * @param int $limit Set to 0 to disable limit again 281 */ 282 public function setLimit($limit) { 283 if($limit) { 284 $this->range_end = $this->range_begin + $limit; 285 } else { 286 $this->range_end = 0; 287 } 288 } 289 290 /** 291 * Return the number of results (regardless of limit and offset settings) 292 * 293 * Use this to implement paging. Important: this may only be called after running @see execute() 294 * 295 * @return int 296 */ 297 public function getCount() { 298 if($this->count < 0) throw new StructException('Count is only accessible after executing the search'); 299 return $this->count; 300 } 301 302 /** 303 * Returns the PID associated with each result row 304 * 305 * Important: this may only be called after running @see execute() 306 * 307 * @return \string[] 308 */ 309 public function getPids() { 310 if($this->result_pids === null) throw new StructException('PIDs are only accessible after executing the search'); 311 return $this->result_pids; 312 } 313 314 /** 315 * Returns the rid associated with each result row 316 * 317 * Important: this may only be called after running @see execute() 318 * 319 * @return array 320 */ 321 public function getRids() { 322 if($this->result_rids === null) throw new StructException('rids are only accessible after executing the search'); 323 return $this->result_rids; 324 } 325 326 /** 327 * Returns the rid associated with each result row 328 * 329 * Important: this may only be called after running @see execute() 330 * 331 * @return array 332 */ 333 public function getRevs() { 334 if($this->result_revs === null) throw new StructException('revs are only accessible after executing the search'); 335 return $this->result_revs; 336 } 337 338 /** 339 * Execute this search and return the result 340 * 341 * The result is a two dimensional array of Value()s. 342 * 343 * This will always query for the full result (not using offset and limit) and then 344 * return the wanted range, setting the count (@see getCount) to the whole result number 345 * 346 * @param string $idColumn Column on which to join tables 347 * @return Value[][] 348 */ 349 public function execute($idColumn = 'rid') { 350 list($sql, $opts) = $this->getSQL($idColumn); 351 352 /** @var \PDOStatement $res */ 353 $res = $this->sqlite->query($sql, $opts); 354 if($res === false) throw new StructException("SQL execution failed for\n\n$sql"); 355 356 $this->result_pids = array(); 357 $result = array(); 358 $cursor = -1; 359 $pageidAndRevOnly = array_reduce($this->columns, function ($pageidAndRevOnly, Column $col) { 360 return $pageidAndRevOnly && ($col->getTid() == 0); 361 }, true); 362 while($row = $res->fetch(\PDO::FETCH_ASSOC)) { 363 $cursor++; 364 if($cursor < $this->range_begin) continue; 365 if($this->range_end && $cursor >= $this->range_end) continue; 366 367 $C = 0; 368 $resrow = array(); 369 $isempty = true; 370 foreach($this->columns as $col) { 371 $val = $row["C$C"]; 372 if($col->isMulti()) { 373 $val = explode(self::CONCAT_SEPARATOR, $val); 374 } 375 $value = new Value($col, $val); 376 $isempty &= $this->isEmptyValue($value); 377 $resrow[] = $value; 378 $C++; 379 } 380 381 // skip empty rows 382 if($isempty && !$pageidAndRevOnly) { 383 $cursor--; 384 continue; 385 } 386 387 $this->result_pids[] = $row['PID']; 388 $this->result_rids[] = $row['rid']; 389 $this->result_revs[] = $row['rev']; 390 $result[] = $resrow; 391 } 392 393 $this->sqlite->res_close($res); 394 $this->count = $cursor + 1; 395 return $result; 396 } 397 398 /** 399 * Transform the set search parameters into a statement 400 * 401 * @param string $idColumn Column on which to join tables 402 * @return array ($sql, $opts) The SQL and parameters to execute 403 */ 404 public function getSQL($idColumn) { 405 if(!$this->columns) throw new StructException('nocolname'); 406 407 $QB = new QueryBuilder(); 408 409 // basic tables 410 $first_table = ''; 411 foreach($this->schemas as $schema) { 412 $datatable = 'data_' . $schema->getTable(); 413 if($first_table) { 414 // follow up tables 415 $QB->addLeftJoin($first_table, $datatable, $datatable, "$first_table.$idColumn = $datatable.$idColumn"); 416 } else { 417 // first table 418 419 // FIXME this breaks page search, a different check is needed 420 if(false) { 421 $QB->addTable('schema_assignments'); 422 $QB->filters()->whereAnd("$datatable.pid = schema_assignments.pid"); 423 $QB->filters()->whereAnd("schema_assignments.tbl = '{$schema->getTable()}'"); 424 $QB->filters()->whereAnd("schema_assignments.assigned = 1"); 425 $QB->filters()->whereAnd("GETACCESSLEVEL($datatable.pid) > 0"); 426 $QB->filters()->whereAnd("PAGEEXISTS($datatable.pid) = 1"); 427 } 428 429 $QB->addTable($datatable); 430 $QB->addSelectColumn($datatable, 'rid'); 431 $QB->addSelectColumn($datatable, 'pid', 'PID'); 432 $QB->addSelectColumn($datatable, 'rev'); 433 $QB->addGroupByColumn($datatable, 'pid'); 434 435 $first_table = $datatable; 436 } 437 $QB->filters()->whereAnd("$datatable.latest = 1"); 438 } 439 440 // columns to select, handling multis 441 $sep = self::CONCAT_SEPARATOR; 442 $n = 0; 443 foreach($this->columns as $col) { 444 $CN = 'C' . $n++; 445 446 if($col->isMulti()) { 447 $datatable = "data_{$col->getTable()}"; 448 $multitable = "multi_{$col->getTable()}"; 449 $MN = $QB->generateTableAlias('M'); 450 451 $QB->addLeftJoin( 452 $datatable, 453 $multitable, 454 $MN, 455 "$datatable.$idColumn = $MN.$idColumn AND 456 $datatable.rev = $MN.rev AND 457 $MN.colref = {$col->getColref()}" 458 ); 459 460 $col->getType()->select($QB, $MN, 'value', $CN); 461 $sel = $QB->getSelectStatement($CN); 462 $QB->addSelectStatement("GROUP_CONCAT($sel, '$sep')", $CN); 463 } else { 464 $col->getType()->select($QB, 'data_' . $col->getTable(), $col->getColName(), $CN); 465 $QB->addGroupByStatement($CN); 466 } 467 } 468 469 // where clauses 470 if (!empty($this->filter)) { 471 $userWHERE = $QB->filters()->where('AND'); 472 } 473 foreach($this->filter as $filter) { 474 /** @var Column $col */ 475 list($col, $value, $comp, $op) = $filter; 476 477 $datatable = "data_{$col->getTable()}"; 478 $multitable = "multi_{$col->getTable()}"; 479 480 /** @var $col Column */ 481 if($col->isMulti()) { 482 $MN = $QB->generateTableAlias('MN'); 483 484 $QB->addLeftJoin( 485 $datatable, 486 $multitable, 487 $MN, 488 "$datatable.$idColumn = $MN.$idColumn AND 489 $datatable.rev = $MN.rev AND 490 $MN.colref = {$col->getColref()}" 491 ); 492 $coltbl = $MN; 493 $colnam = 'value'; 494 } else { 495 $coltbl = $datatable; 496 $colnam = $col->getColName(); 497 } 498 499 $col->getType()->filter($userWHERE, $coltbl, $colnam, $comp, $value, $op); // type based filter 500 } 501 502 // sorting - we always sort by the single val column 503 foreach($this->sortby as $sort) { 504 list($col, $asc, $nc) = $sort; 505 /** @var $col Column */ 506 $colname = $col->getColName(false); 507 if($nc) $colname .= ' COLLATE NOCASE'; 508 $col->getType()->sort($QB, 'data_' . $col->getTable(), $colname, $asc ? 'ASC' : 'DESC'); 509 } 510 511 return $QB->getSQL(); 512 } 513 514 /** 515 * Returns all the columns that where added to the search 516 * 517 * @return Column[] 518 */ 519 public function getColumns() { 520 return $this->columns; 521 } 522 523 /** 524 * All the schemas currently added 525 * 526 * @return Schema[] 527 */ 528 public function getSchemas() { 529 return array_values($this->schemas); 530 } 531 532 /** 533 * Checks if the given column is a * wildcard 534 * 535 * If it's a wildcard all matching columns are added to the column list, otherwise 536 * nothing happens 537 * 538 * @param string $colname 539 * @return bool was wildcard? 540 */ 541 protected function processWildcard($colname) { 542 list($colname, $table) = $this->resolveColumn($colname); 543 if($colname !== '*') return false; 544 545 // no table given? assume the first is meant 546 if($table === null) { 547 $schema_list = array_keys($this->schemas); 548 $table = $schema_list[0]; 549 } 550 551 $schema = $this->schemas[$table]; 552 if(!$schema) return false; 553 $this->columns = array_merge($this->columns, $schema->getColumns(false)); 554 return true; 555 } 556 557 /** 558 * Split a given column name into table and column 559 * 560 * Handles Aliases. Table might be null if none given. 561 * 562 * @param $colname 563 * @return array (colname, table) 564 */ 565 protected function resolveColumn($colname) { 566 if(!$this->schemas) throw new StructException('noschemas'); 567 568 // resolve the alias or table name 569 @list($table, $colname) = explode('.', $colname, 2); 570 if(!$colname) { 571 $colname = $table; 572 $table = null; 573 } 574 if($table && isset($this->aliases[$table])) { 575 $table = $this->aliases[$table]; 576 } 577 578 if(!$colname) throw new StructException('nocolname'); 579 580 return array($colname, $table); 581 } 582 583 /** 584 * Find a column to be used in the search 585 * 586 * @param string $colname may contain an alias 587 * @return bool|Column 588 */ 589 public function findColumn($colname) { 590 if(!$this->schemas) throw new StructException('noschemas'); 591 $schema_list = array_keys($this->schemas); 592 593 // add "fake" column for special col 594 if($colname == '%pageid%') { 595 return new PageColumn(0, new Page(), $schema_list[0]); 596 } 597 if($colname == '%title%') { 598 return new PageColumn(0, new Page(array('usetitles' => true)), $schema_list[0]); 599 } 600 if($colname == '%lastupdate%') { 601 return new RevisionColumn(0, new DateTime(), $schema_list[0]); 602 } 603 if ($colname == '%lasteditor%') { 604 return new UserColumn(0, new User(), $schema_list[0]); 605 } 606 if ($colname == '%lastsummary%') { 607 return new SummaryColumn(0, new AutoSummary(), $schema_list[0]); 608 } 609 if($colname == '%rowid%') { 610 return new RowColumn(0, new Decimal(), $schema_list[0]); 611 } 612 613 list($colname, $table) = $this->resolveColumn($colname); 614 615 // if table name given search only that, otherwise try all for matching column name 616 if($table !== null) { 617 $schemas = array($table => $this->schemas[$table]); 618 } else { 619 $schemas = $this->schemas; 620 } 621 622 // find it 623 $col = false; 624 foreach($schemas as $schema) { 625 if(empty($schema)) { 626 continue; 627 } 628 $col = $schema->findColumn($colname); 629 if($col) break; 630 } 631 632 return $col; 633 } 634 635 /** 636 * Check if the given row is empty or references our own row 637 * 638 * @param Value $value 639 * @return bool 640 */ 641 protected function isEmptyValue(Value $value) { 642 if ($value->isEmpty()) return true; 643 if ($value->getColumn()->getTid() == 0) return true; 644 return false; 645 } 646} 647 648 649