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 * @return Value[][] 347 */ 348 public function execute() { 349 list($sql, $opts) = $this->getSQL(); 350 351 /** @var \PDOStatement $res */ 352 $res = $this->sqlite->query($sql, $opts); 353 if($res === false) throw new StructException("SQL execution failed for\n\n$sql"); 354 355 $this->result_pids = array(); 356 $result = array(); 357 $cursor = -1; 358 $pageidAndRevOnly = array_reduce($this->columns, function ($pageidAndRevOnly, Column $col) { 359 return $pageidAndRevOnly && ($col->getTid() == 0); 360 }, true); 361 while($row = $res->fetch(\PDO::FETCH_ASSOC)) { 362 $cursor++; 363 if($cursor < $this->range_begin) continue; 364 if($this->range_end && $cursor >= $this->range_end) continue; 365 366 $C = 0; 367 $resrow = array(); 368 $isempty = true; 369 foreach($this->columns as $col) { 370 $val = $row["C$C"]; 371 if($col->isMulti()) { 372 $val = explode(self::CONCAT_SEPARATOR, $val); 373 } 374 $value = new Value($col, $val); 375 $isempty &= $this->isEmptyValue($value); 376 $resrow[] = $value; 377 $C++; 378 } 379 380 // skip empty rows 381 if($isempty && !$pageidAndRevOnly) { 382 $cursor--; 383 continue; 384 } 385 386 $this->result_pids[] = $row['PID']; 387 $this->result_rids[] = $row['rid']; 388 $this->result_revs[] = $row['rev']; 389 $result[] = $resrow; 390 } 391 392 $this->sqlite->res_close($res); 393 $this->count = $cursor + 1; 394 return $result; 395 } 396 397 /** 398 * Transform the set search parameters into a statement 399 * 400 * @return array ($sql, $opts) The SQL and parameters to execute 401 */ 402 public function getSQL() { 403 if(!$this->columns) throw new StructException('nocolname'); 404 405 $QB = new QueryBuilder(); 406 407 // basic tables 408 $first_table = ''; 409 foreach($this->schemas as $schema) { 410 $datatable = 'data_' . $schema->getTable(); 411 if($first_table) { 412 // follow up tables 413 $QB->addLeftJoin($first_table, $datatable, $datatable, "$first_table.pid = $datatable.pid"); 414 } else { 415 // first table 416 417 // FIXME this breaks page search, a different check is needed 418 if(false) { 419 $QB->addTable('schema_assignments'); 420 $QB->filters()->whereAnd("$datatable.pid = schema_assignments.pid"); 421 $QB->filters()->whereAnd("schema_assignments.tbl = '{$schema->getTable()}'"); 422 $QB->filters()->whereAnd("schema_assignments.assigned = 1"); 423 $QB->filters()->whereAnd("GETACCESSLEVEL($datatable.pid) > 0"); 424 $QB->filters()->whereAnd("PAGEEXISTS($datatable.pid) = 1"); 425 } 426 427 $QB->addTable($datatable); 428 $QB->addSelectColumn($datatable, 'rid'); 429 $QB->addSelectColumn($datatable, 'pid', 'PID'); 430 $QB->addSelectColumn($datatable, 'rev'); 431 $QB->addGroupByColumn($datatable, 'pid'); 432 433 $first_table = $datatable; 434 } 435 $QB->filters()->whereAnd("$datatable.latest = 1"); 436 } 437 438 // columns to select, handling multis 439 $sep = self::CONCAT_SEPARATOR; 440 $n = 0; 441 foreach($this->columns as $col) { 442 $CN = 'C' . $n++; 443 444 if($col->isMulti()) { 445 $datatable = "data_{$col->getTable()}"; 446 $multitable = "multi_{$col->getTable()}"; 447 $MN = $QB->generateTableAlias('M'); 448 449 $QB->addLeftJoin( 450 $datatable, 451 $multitable, 452 $MN, 453 "$datatable.pid = $MN.pid AND 454 $datatable.rev = $MN.rev AND 455 $MN.colref = {$col->getColref()}" 456 ); 457 458 $col->getType()->select($QB, $MN, 'value', $CN); 459 $sel = $QB->getSelectStatement($CN); 460 $QB->addSelectStatement("GROUP_CONCAT($sel, '$sep')", $CN); 461 } else { 462 $col->getType()->select($QB, 'data_' . $col->getTable(), $col->getColName(), $CN); 463 $QB->addGroupByStatement($CN); 464 } 465 } 466 467 // where clauses 468 if (!empty($this->filter)) { 469 $userWHERE = $QB->filters()->where('AND'); 470 } 471 foreach($this->filter as $filter) { 472 /** @var Column $col */ 473 list($col, $value, $comp, $op) = $filter; 474 475 $datatable = "data_{$col->getTable()}"; 476 $multitable = "multi_{$col->getTable()}"; 477 478 /** @var $col Column */ 479 if($col->isMulti()) { 480 $MN = $QB->generateTableAlias('MN'); 481 482 $QB->addLeftJoin( 483 $datatable, 484 $multitable, 485 $MN, 486 "$datatable.pid = $MN.pid AND 487 $datatable.rev = $MN.rev AND 488 $MN.colref = {$col->getColref()}" 489 ); 490 $coltbl = $MN; 491 $colnam = 'value'; 492 } else { 493 $coltbl = $datatable; 494 $colnam = $col->getColName(); 495 } 496 497 $col->getType()->filter($userWHERE, $coltbl, $colnam, $comp, $value, $op); // type based filter 498 } 499 500 // sorting - we always sort by the single val column 501 foreach($this->sortby as $sort) { 502 list($col, $asc, $nc) = $sort; 503 /** @var $col Column */ 504 $colname = $col->getColName(false); 505 if($nc) $colname .= ' COLLATE NOCASE'; 506 $col->getType()->sort($QB, 'data_' . $col->getTable(), $colname, $asc ? 'ASC' : 'DESC'); 507 } 508 509 return $QB->getSQL(); 510 } 511 512 /** 513 * Returns all the columns that where added to the search 514 * 515 * @return Column[] 516 */ 517 public function getColumns() { 518 return $this->columns; 519 } 520 521 /** 522 * All the schemas currently added 523 * 524 * @return Schema[] 525 */ 526 public function getSchemas() { 527 return array_values($this->schemas); 528 } 529 530 /** 531 * Checks if the given column is a * wildcard 532 * 533 * If it's a wildcard all matching columns are added to the column list, otherwise 534 * nothing happens 535 * 536 * @param string $colname 537 * @return bool was wildcard? 538 */ 539 protected function processWildcard($colname) { 540 list($colname, $table) = $this->resolveColumn($colname); 541 if($colname !== '*') return false; 542 543 // no table given? assume the first is meant 544 if($table === null) { 545 $schema_list = array_keys($this->schemas); 546 $table = $schema_list[0]; 547 } 548 549 $schema = $this->schemas[$table]; 550 if(!$schema) return false; 551 $this->columns = array_merge($this->columns, $schema->getColumns(false)); 552 return true; 553 } 554 555 /** 556 * Split a given column name into table and column 557 * 558 * Handles Aliases. Table might be null if none given. 559 * 560 * @param $colname 561 * @return array (colname, table) 562 */ 563 protected function resolveColumn($colname) { 564 if(!$this->schemas) throw new StructException('noschemas'); 565 566 // resolve the alias or table name 567 @list($table, $colname) = explode('.', $colname, 2); 568 if(!$colname) { 569 $colname = $table; 570 $table = null; 571 } 572 if($table && isset($this->aliases[$table])) { 573 $table = $this->aliases[$table]; 574 } 575 576 if(!$colname) throw new StructException('nocolname'); 577 578 return array($colname, $table); 579 } 580 581 /** 582 * Find a column to be used in the search 583 * 584 * @param string $colname may contain an alias 585 * @return bool|Column 586 */ 587 public function findColumn($colname) { 588 if(!$this->schemas) throw new StructException('noschemas'); 589 $schema_list = array_keys($this->schemas); 590 591 // add "fake" column for special col 592 if($colname == '%pageid%') { 593 return new PageColumn(0, new Page(), $schema_list[0]); 594 } 595 if($colname == '%title%') { 596 return new PageColumn(0, new Page(array('usetitles' => true)), $schema_list[0]); 597 } 598 if($colname == '%lastupdate%') { 599 return new RevisionColumn(0, new DateTime(), $schema_list[0]); 600 } 601 if ($colname == '%lasteditor%') { 602 return new UserColumn(0, new User(), $schema_list[0]); 603 } 604 if ($colname == '%lastsummary%') { 605 return new SummaryColumn(0, new AutoSummary(), $schema_list[0]); 606 } 607 if($colname == '%rowid%') { 608 return new RowColumn(0, new Decimal(), $schema_list[0]); 609 } 610 611 list($colname, $table) = $this->resolveColumn($colname); 612 613 // if table name given search only that, otherwise try all for matching column name 614 if($table !== null) { 615 $schemas = array($table => $this->schemas[$table]); 616 } else { 617 $schemas = $this->schemas; 618 } 619 620 // find it 621 $col = false; 622 foreach($schemas as $schema) { 623 if(empty($schema)) { 624 continue; 625 } 626 $col = $schema->findColumn($colname); 627 if($col) break; 628 } 629 630 return $col; 631 } 632 633 /** 634 * Check if the given row is empty or references our own row 635 * 636 * @param Value $value 637 * @return bool 638 */ 639 protected function isEmptyValue(Value $value) { 640 if ($value->isEmpty()) return true; 641 if ($value->getColumn()->getTid() == 0) return true; 642 return false; 643 } 644} 645 646 647