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