xref: /plugin/struct/meta/Search.php (revision 8b04ed8174b19d68efb6bd7e781d1086fe9750ca)
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) {
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