xref: /plugin/struct/meta/Search.php (revision 00bff81c57638c56aa4e85d91c0f80d92dd8338f) 
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        '<=', '>=', '=*', '=', '<', '>', '!=', '!~', '~',
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        // add the filter
180        $this->filter[] = array($col, $value, $comp, $op);
181    }
182
183    /**
184     * Wrap given value in asterisks
185     *
186     * @param string|string[] $value
187     * @return string|string[]
188     */
189    protected function filterWrapAsterisks($value) {
190        $map = function ($input) {
191            return "*$input*";
192        };
193
194        if(is_array($value)) {
195            $value = array_map($map, $value);
196        } else {
197            $value = $map($value);
198        }
199        return $value;
200    }
201
202    /**
203     * Change given string to use % instead of *
204     *
205     * @param string|string[] $value
206     * @return string|string[]
207     */
208    protected function filterChangeToLike($value) {
209        $map = function ($input) {
210            return str_replace('*', '%', $input);
211        };
212
213        if(is_array($value)) {
214            $value = array_map($map, $value);
215        } else {
216            $value = $map($value);
217        }
218        return $value;
219    }
220
221    /**
222     * Set offset for the results
223     *
224     * @param int $offset
225     */
226    public function setOffset($offset) {
227        $limit = 0;
228        if($this->range_end) {
229            // if there was a limit set previously, the range_end needs to be recalculated
230            $limit = $this->range_end - $this->range_begin;
231        }
232        $this->range_begin = $offset;
233        if($limit) $this->setLimit($limit);
234    }
235
236    /**
237     * Limit results to this number
238     *
239     * @param int $limit Set to 0 to disable limit again
240     */
241    public function setLimit($limit) {
242        if($limit) {
243            $this->range_end = $this->range_begin + $limit;
244        } else {
245            $this->range_end = 0;
246        }
247    }
248
249    /**
250     * Return the number of results (regardless of limit and offset settings)
251     *
252     * Use this to implement paging. Important: this may only be called after running @see execute()
253     *
254     * @return int
255     */
256    public function getCount() {
257        if($this->count < 0) throw new StructException('Count is only accessible after executing the search');
258        return $this->count;
259    }
260
261    /**
262     * Returns the PID associated with each result row
263     *
264     * Important: this may only be called after running @see execute()
265     *
266     * @return \string[]
267     */
268    public function getPids() {
269        if($this->result_pids === null) throw new StructException('PIDs are only accessible after executing the search');
270        return $this->result_pids;
271    }
272
273    /**
274     * Returns the rid associated with each result row
275     *
276     * Important: this may only be called after running @see execute()
277     *
278     * @return array
279     */
280    public function getRids() {
281        if($this->result_rids === null) throw new StructException('rids are only accessible after executing the search');
282        return $this->result_rids;
283    }
284
285    /**
286     * Returns the rid associated with each result row
287     *
288     * Important: this may only be called after running @see execute()
289     *
290     * @return array
291     */
292    public function getRevs() {
293        if($this->result_revs === null) throw new StructException('revs are only accessible after executing the search');
294        return $this->result_revs;
295    }
296
297    /**
298     * Execute this search and return the result
299     *
300     * The result is a two dimensional array of Value()s.
301     *
302     * This will always query for the full result (not using offset and limit) and then
303     * return the wanted range, setting the count (@see getCount) to the whole result number
304     *
305     * @return Value[][]
306     */
307    public function execute() {
308        list($sql, $opts) = $this->getSQL();
309
310        /** @var \PDOStatement $res */
311        $res = $this->sqlite->query($sql, $opts);
312        if($res === false) throw new StructException("SQL execution failed for\n\n$sql");
313
314        $this->result_pids = array();
315        $result = array();
316        $cursor = -1;
317        $pageidAndRevOnly = array_reduce($this->columns, function ($pageidAndRevOnly, Column $col) {
318            return $pageidAndRevOnly && ($col->getTid() == 0);
319        }, true);
320        while($row = $res->fetch(\PDO::FETCH_ASSOC)) {
321            $cursor++;
322            if($cursor < $this->range_begin) continue;
323            if($this->range_end && $cursor >= $this->range_end) continue;
324
325            $C = 0;
326            $resrow = array();
327            $isempty = true;
328            foreach($this->columns as $col) {
329                $val = $row["C$C"];
330                if($col->isMulti()) {
331                    $val = explode(self::CONCAT_SEPARATOR, $val);
332                }
333                $value = new Value($col, $val);
334                $isempty &= $this->isEmptyValue($value);
335                $resrow[] = $value;
336                $C++;
337            }
338
339            // skip empty rows
340            if($isempty && !$pageidAndRevOnly) {
341                $cursor--;
342                continue;
343            }
344
345            $this->result_pids[] = $row['PID'];
346            $this->result_rids[] = $row['rid'];
347            $this->result_revs[] = $row['rev'];
348            $result[] = $resrow;
349        }
350
351        $this->sqlite->res_close($res);
352        $this->count = $cursor + 1;
353        return $result;
354    }
355
356    /**
357     * Transform the set search parameters into a statement
358     *
359     * @return array ($sql, $opts) The SQL and parameters to execute
360     */
361    public function getSQL() {
362        if(!$this->columns) throw new StructException('nocolname');
363
364        $QB = new QueryBuilder();
365
366        // basic tables
367        $first_table = '';
368        foreach($this->schemas as $schema) {
369            $datatable = 'data_' . $schema->getTable();
370            if($first_table) {
371                // follow up tables
372                $QB->addLeftJoin($first_table, $datatable, $datatable, "$first_table.pid = $datatable.pid");
373            } else {
374                // first table
375
376                // FIXME this breaks page search, a different check is needed
377                if(false) {
378                    $QB->addTable('schema_assignments');
379                    $QB->filters()->whereAnd("$datatable.pid = schema_assignments.pid");
380                    $QB->filters()->whereAnd("schema_assignments.tbl = '{$schema->getTable()}'");
381                    $QB->filters()->whereAnd("schema_assignments.assigned = 1");
382                    $QB->filters()->whereAnd("GETACCESSLEVEL($datatable.pid) > 0");
383                    $QB->filters()->whereAnd("PAGEEXISTS($datatable.pid) = 1");
384                }
385
386                $QB->addTable($datatable);
387                $QB->addSelectColumn($datatable, 'rid');
388                $QB->addSelectColumn($datatable, 'pid', 'PID');
389                $QB->addSelectColumn($datatable, 'rev');
390                $QB->addGroupByColumn($datatable, 'pid');
391
392                $first_table = $datatable;
393            }
394            $QB->filters()->whereAnd("$datatable.latest = 1");
395        }
396
397        // columns to select, handling multis
398        $sep = self::CONCAT_SEPARATOR;
399        $n = 0;
400        foreach($this->columns as $col) {
401            $CN = 'C' . $n++;
402
403            if($col->isMulti()) {
404                $datatable = "data_{$col->getTable()}";
405                $multitable = "multi_{$col->getTable()}";
406                $MN = $QB->generateTableAlias('M');
407
408                $QB->addLeftJoin(
409                    $datatable,
410                    $multitable,
411                    $MN,
412                    "$datatable.pid = $MN.pid AND
413                     $datatable.rev = $MN.rev AND
414                     $MN.colref = {$col->getColref()}"
415                );
416
417                $col->getType()->select($QB, $MN, 'value', $CN);
418                $sel = $QB->getSelectStatement($CN);
419                $QB->addSelectStatement("GROUP_CONCAT($sel, '$sep')", $CN);
420            } else {
421                $col->getType()->select($QB, 'data_' . $col->getTable(), $col->getColName(), $CN);
422                $QB->addGroupByStatement($CN);
423            }
424        }
425
426        // where clauses
427        if (!empty($this->filter)) {
428            $userWHERE = $QB->filters()->where('AND');
429        }
430        foreach($this->filter as $filter) {
431            /** @var Column $col */
432            list($col, $value, $comp, $op) = $filter;
433
434            $datatable = "data_{$col->getTable()}";
435            $multitable = "multi_{$col->getTable()}";
436
437            /** @var $col Column */
438            if($col->isMulti()) {
439                $MN = $QB->generateTableAlias('MN');
440
441                $QB->addLeftJoin(
442                    $datatable,
443                    $multitable,
444                    $MN,
445                    "$datatable.pid = $MN.pid AND
446                     $datatable.rev = $MN.rev AND
447                     $MN.colref = {$col->getColref()}"
448                );
449                $coltbl = $MN;
450                $colnam = 'value';
451            } else {
452                $coltbl = $datatable;
453                $colnam = $col->getColName();
454            }
455
456            $col->getType()->filter($userWHERE, $coltbl, $colnam, $comp, $value, $op); // type based filter
457        }
458
459        // sorting - we always sort by the single val column
460        foreach($this->sortby as $sort) {
461            list($col, $asc, $nc) = $sort;
462            /** @var $col Column */
463            $colname = $col->getColName(false);
464            if($nc) $colname .= ' COLLATE NOCASE';
465            $col->getType()->sort($QB, 'data_' . $col->getTable(), $colname, $asc ? 'ASC' : 'DESC');
466        }
467
468        return $QB->getSQL();
469    }
470
471    /**
472     * Returns all the columns that where added to the search
473     *
474     * @return Column[]
475     */
476    public function getColumns() {
477        return $this->columns;
478    }
479
480    /**
481     * All the schemas currently added
482     *
483     * @return Schema[]
484     */
485    public function getSchemas() {
486        return array_values($this->schemas);
487    }
488
489    /**
490     * Checks if the given column is a * wildcard
491     *
492     * If it's a wildcard all matching columns are added to the column list, otherwise
493     * nothing happens
494     *
495     * @param string $colname
496     * @return bool was wildcard?
497     */
498    protected function processWildcard($colname) {
499        list($colname, $table) = $this->resolveColumn($colname);
500        if($colname !== '*') return false;
501
502        // no table given? assume the first is meant
503        if($table === null) {
504            $schema_list = array_keys($this->schemas);
505            $table = $schema_list[0];
506        }
507
508        $schema = $this->schemas[$table];
509        if(!$schema) return false;
510        $this->columns = array_merge($this->columns, $schema->getColumns(false));
511        return true;
512    }
513
514    /**
515     * Split a given column name into table and column
516     *
517     * Handles Aliases. Table might be null if none given.
518     *
519     * @param $colname
520     * @return array (colname, table)
521     */
522    protected function resolveColumn($colname) {
523        if(!$this->schemas) throw new StructException('noschemas');
524
525        // resolve the alias or table name
526        @list($table, $colname) = explode('.', $colname, 2);
527        if(!$colname) {
528            $colname = $table;
529            $table = null;
530        }
531        if($table && isset($this->aliases[$table])) {
532            $table = $this->aliases[$table];
533        }
534
535        if(!$colname) throw new StructException('nocolname');
536
537        return array($colname, $table);
538    }
539
540    /**
541     * Find a column to be used in the search
542     *
543     * @param string $colname may contain an alias
544     * @return bool|Column
545     */
546    public function findColumn($colname) {
547        if(!$this->schemas) throw new StructException('noschemas');
548        $schema_list = array_keys($this->schemas);
549
550        // add "fake" column for special col
551        if($colname == '%pageid%') {
552            return new PageColumn(0, new Page(), $schema_list[0]);
553        }
554        if($colname == '%title%') {
555            return new PageColumn(0, new Page(array('usetitles' => true)), $schema_list[0]);
556        }
557        if($colname == '%lastupdate%') {
558            return new RevisionColumn(0, new DateTime(), $schema_list[0]);
559        }
560        if ($colname == '%lasteditor%') {
561            return new UserColumn(0, new User(), $schema_list[0]);
562        }
563        if ($colname == '%lastsummary%') {
564            return new SummaryColumn(0, new AutoSummary(), $schema_list[0]);
565        }
566        if($colname == '%rowid%') {
567            return new RowColumn(0, new Decimal(), $schema_list[0]);
568        }
569
570        list($colname, $table) = $this->resolveColumn($colname);
571
572        // if table name given search only that, otherwise try all for matching column name
573        if($table !== null) {
574            $schemas = array($table => $this->schemas[$table]);
575        } else {
576            $schemas = $this->schemas;
577        }
578
579        // find it
580        $col = false;
581        foreach($schemas as $schema) {
582            if(empty($schema)) {
583                continue;
584            }
585            $col = $schema->findColumn($colname);
586            if($col) break;
587        }
588
589        return $col;
590    }
591
592    /**
593     * Check if the given row is empty or references our own row
594     *
595     * @param Value $value
596     * @return bool
597     */
598    protected function isEmptyValue(Value $value) {
599        if ($value->isEmpty()) return true;
600        if ($value->getColumn()->getTid() == 0) return true;
601        return false;
602    }
603}
604
605
606