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