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