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