PHRAS-1797 porting searchengine:index to 4.1 (#2454)

* portage commande searchengine:index vers 4.1

* FIX Indexer

* Update DataboxFetcherFactory.php

bin/console

@@ -18,6 +18,7 @@ use Alchemy\Phrasea\Command\SearchEngine\Debug\QueryParseCommand;
 use Alchemy\Phrasea\Command\SearchEngine\Debug\QuerySampleCommand;
 use Alchemy\Phrasea\Command\SearchEngine\IndexCreateCommand;
 use Alchemy\Phrasea\Command\SearchEngine\IndexDropCommand;
+use Alchemy\Phrasea\Command\SearchEngine\IndexManipulateCommand;
 use Alchemy\Phrasea\Command\SearchEngine\IndexPopulateCommand;
 use Alchemy\Phrasea\Command\Thesaurus\FindConceptsCommand;
 use Alchemy\Phrasea\Core\Version;
@@ -123,6 +124,7 @@ $cli->command(new H264MappingGenerator());
 $cli->command(new XSendFileConfigurationDumper());
 $cli->command(new XSendFileMappingGenerator());
 
+$cli->command(new IndexManipulateCommand());
 $cli->command(new IndexCreateCommand());
 $cli->command(new IndexDropCommand());
 $cli->command(new MappingUpdateCommand());

lib/Alchemy/Phrasea/Command/SearchEngine/IndexCreateCommand.php

@@ -24,7 +24,7 @@ class IndexCreateCommand extends Command
     {
         $this
             ->setName('searchengine:index:create')
-            ->setDescription('Creates search index')
+            ->setDescription('Creates search index <fg=yellow;>(Deprecated use searchengine:index instead)</>')
             ->addOption('drop', 'd', InputOption::VALUE_NONE, 'Drops the index if it already exists.');
     }
 

lib/Alchemy/Phrasea/Command/SearchEngine/IndexDropCommand.php

@@ -23,12 +23,12 @@ class IndexDropCommand extends Command
     {
         $this
             ->setName('searchengine:index:drop')
-            ->setDescription('Deletes the search index')
+            ->setDescription('Deletes the search index <fg=yellow;>(Deprecated use searchengine:index instead)</>')
             ->addOption(
                'force',
                null,
                InputOption::VALUE_NONE,
-               "Don't ask for for the dropping of the index, but force the operation to run."
+               "Don't ask for the dropping of the index, but force the operation to run."
             )
         ;
     }

lib/Alchemy/Phrasea/Command/SearchEngine/IndexManipulateCommand.php

@@ -0,0 +1,215 @@
+<?php
+
+/*
+ * This file is part of Phraseanet
+ *
+ * (c) 2005-2014 Alchemy
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Alchemy\Phrasea\Command\SearchEngine;
+
+use Alchemy\Phrasea\Command\Command;
+use Alchemy\Phrasea\SearchEngine\Elastic\ElasticsearchOptions;
+use Alchemy\Phrasea\SearchEngine\Elastic\Indexer;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class IndexManipulateCommand extends Command
+{
+    /** @var  OutputInterface */
+    private $output = null;
+
+    /**
+     * print a string if verbosity >= verbose (-v)
+     *
+     * @param string $s
+     * @param int $verbosity
+     */
+    private function verbose($s, $verbosity = OutputInterface::VERBOSITY_VERBOSE)
+    {
+        if ($this->output->getVerbosity() >= $verbosity) {
+            $this->output->writeln($s);
+        }
+    }
+
+    protected function configure()
+    {
+        $this
+            ->setName('searchengine:index')
+            ->setDescription('Manipulates search index')
+            ->addOption('drop',      'd', InputOption::VALUE_NONE, 'Drops the index.')
+            ->addOption('create',    'c', InputOption::VALUE_NONE, 'Creates the index.')
+            ->addOption('populate',  'p', InputOption::VALUE_NONE, 'Populates the index.')
+            ->addOption('temporary', 't', InputOption::VALUE_NONE, 'Populates using temporary index.')
+            ->addOption('name', null, InputOption::VALUE_REQUIRED, 'index name', null)
+            ->addOption('host', null, InputOption::VALUE_REQUIRED, 'host', null)
+            ->addOption('port', null, InputOption::VALUE_REQUIRED, 'port', null)
+            ->addOption('order', null, InputOption::VALUE_REQUIRED, 'order (record_id|modification_date)[.asc|.desc]', null)
+            ->addOption(
+                'databox_id',
+                null,
+                InputOption::VALUE_OPTIONAL | InputOption::VALUE_IS_ARRAY,
+                'Only populate chosen databox'
+            )->addOption(
+                'force',
+                null,
+                InputOption::VALUE_NONE,
+                "Don't ask for for the dropping of the index, but force the operation to run."
+            );
+
+    }
+
+    protected function doExecute(InputInterface $input, OutputInterface $output)
+    {
+        $this->output = $output;
+
+        /** @var Indexer $indexer */
+        $indexer = $this->container['elasticsearch.indexer'];
+        /** @var ElasticsearchOptions $options */
+        $options = $indexer->getIndex()->getOptions();
+
+        if($input->getOption('name')) {
+            $options->setIndexName($input->getOption('name'));
+        }
+        if($input->getOption('host')) {
+            $options->setHost($input->getOption('host'));
+        }
+        if($input->getOption('port')) {
+            $options->setPort($input->getOption('port'));
+        }
+
+        if($input->getOption('order')) {
+            $order = explode('.', $input->getOption('order'));
+            if (!$options->setPopulateOrder($order[0])) {
+                $output->writeln(sprintf('<error>bad order value for --order</error>'));
+
+                return 1;
+            }
+            if (count($order) > 1) {
+                if (!$options->setPopulateDirection($order[1])) {
+                    $output->writeln(sprintf('<error>bad direction value for --order</error>'));
+
+                    return 1;
+                }
+            }
+        }
+
+        $idx = sprintf("%s@%s:%s", $options->getIndexName(), $options->getHost(), $options->getPort());
+
+        $drop         = $input->getOption('drop');
+        $create       = $input->getOption('create');
+        $populate     = $input->getOption('populate');
+        $temporary    = $input->getOption('temporary');
+        $databoxes_id = $input->getOption('databox_id');
+
+        if($temporary && (!$populate || $databoxes_id)) {
+            $output->writeln(sprintf('<error>temporary must be used to populate all databoxes</error>', $idx));
+
+            return 1;
+        }
+
+        $indexExists = $indexer->indexExists();
+
+        if ($drop && $indexExists) {
+            if ($input->getOption('force')) {
+                $confirmation = true;
+            }
+            else {
+                $question = '<question>You are about to delete the index and all contained data. Are you sure you wish to continue? (y/n)</question>';
+                $confirmation = $this->getHelper('dialog')->askConfirmation($output, $question, false);
+            }
+
+            if ($confirmation) {
+                $indexer->deleteIndex();
+                $this->verbose(sprintf('<info>Search index "%s" was dropped.</info>', $idx));
+            }
+            else {
+                $this->verbose('Canceled.');
+
+                return 0;
+            }
+        }
+
+        $indexExists = $indexer->indexExists();
+
+        if ($create) {
+            if($indexExists) {
+                $output->writeln(sprintf('<error>The search index "%s" already exists.</error>', $idx));
+
+                return 1;
+            }
+            else {
+                $r = $indexer->createIndex();
+                $this->verbose(sprintf('<info>Search index "%s@%s:%s" -> "%s" was created</info>'
+                    , $r['alias']
+                    , $options->getHost()
+                    , $options->getPort()
+                    , $r['index']
+                ));
+            }
+        }
+
+        $indexExists = $indexer->indexExists();
+
+        if($populate) {
+            if(!$indexExists) {
+                $r = $indexer->createIndex();
+                $this->verbose(sprintf('<info>Search index "%s@%s:%s" -> "%s" was created</info>'
+                    , $r['alias']
+                    , $options->getHost()
+                    , $options->getPort()
+                    , $r['index']
+                ));
+            }
+
+            $oldAliasName = $indexer->getIndex()->getName();
+            $newAliasName = $newIndexName = null;
+            if($temporary) {
+                // change the name to create a new index
+                $now = explode(' ', microtime());
+                $now = sprintf("%X%X", $now[1], 1000000*$now[0]);
+                $indexer->getIndex()->getOptions()->setIndexName($oldAliasName . "_T" . $now);
+
+                $r = $indexer->createIndex($oldAliasName);
+                $newIndexName = $r['index'];
+                $newAliasName = $r['alias'];
+
+                $this->verbose(sprintf('<info>Temporary index "%s@%s:%s" -> "%s" was created</info>'
+                    , $r['alias']
+                    , $options->getHost()
+                    , $options->getPort()
+                    , $r['index']
+                ));
+            }
+
+            foreach ($this->container->getDataboxes() as $databox) {
+                if (!$databoxes_id || in_array($databox->get_sbas_id(), $databoxes_id)) {
+                    $r = $indexer->populateIndex(Indexer::THESAURUS | Indexer::RECORDS, $databox, false); // , $temporary);
+                    $output->writeln(sprintf(
+                            "Indexation of databox \"%s\" finished in %0.2f sec (Mem. %0.2f Mo)",
+                            $databox->get_dbname(),
+                            $r['duration']/1000,
+                            $r['memory']/1048576)
+                    );
+                }
+            }
+
+            if($temporary) {
+                $this->verbose('<info>Renaming temporary :</info>');
+
+                $indexer->getIndex()->getOptions()->setIndexName($oldAliasName);
+
+                $r = $indexer->replaceIndex($newIndexName, $newAliasName);
+                foreach($r as $action) {
+                    $this->verbose(sprintf('  <info>%s</info>', $action['msg']));
+                }
+            }
+        }
+
+        return 0;
+    }
+}

lib/Alchemy/Phrasea/Command/SearchEngine/IndexPopulateCommand.php

@@ -23,7 +23,7 @@ class IndexPopulateCommand extends Command
     {
         $this
             ->setName('searchengine:index:populate')
-            ->setDescription('Populate search index')
+            ->setDescription('Populate search index <fg=yellow;>(Deprecated use searchengine:index instead)</>')
             ->addOption(
                 'thesaurus',
                 null,

lib/Alchemy/Phrasea/Core/Provider/SearchEngineServiceProvider.php

@@ -146,6 +146,7 @@ class SearchEngineServiceProvider implements ServiceProviderInterface
         $app['elasticsearch.indexer.databox_fetcher_factory'] = $app->share(function ($app) {
             return new DataboxFetcherFactory(
                 $app['elasticsearch.record_helper'],
+                $app['elasticsearch.options'],
                 $app,
                 'search_engine.structure',
                 'thesaurus'

lib/Alchemy/Phrasea/SearchEngine/Elastic/DataboxFetcherFactory.php

@@ -35,16 +35,21 @@ class DataboxFetcherFactory
      */
     private $recordHelper;
 
+    /** @var  ElasticsearchOptions */
+    private $options;
+
     /**
      * @param RecordHelper $recordHelper
+     * @param ElasticsearchOptions $options
      * @param \ArrayAccess $container
      * @param string $structureKey
      * @param string $thesaurusKey
      */
-    public function __construct(RecordHelper $recordHelper, \ArrayAccess $container, $structureKey, $thesaurusKey)
+    public function __construct(RecordHelper $recordHelper, ElasticsearchOptions $options, \ArrayAccess $container, $structureKey, $thesaurusKey)
     {
         $this->recordHelper = $recordHelper;
-        $this->container = $container;
+        $this->options      = $options;
+        $this->container    = $container;
         $this->structureKey = $structureKey;
         $this->thesaurusKey = $thesaurusKey;
     }
@@ -59,14 +64,19 @@ class DataboxFetcherFactory
         $connection = $databox->get_connection();
 
         $candidateTerms = new CandidateTerms($databox);
-        $fetcher = new Fetcher($databox, array(
+        $fetcher = new Fetcher(
-            new CoreHydrator($databox->get_sbas_id(), $databox->get_viewname(), $this->recordHelper),
+            $databox,
-            new TitleHydrator($connection),
+            $this->options,
-            new MetadataHydrator($connection, $this->getStructure(), $this->recordHelper),
+            [
-            new FlagHydrator($this->getStructure(), $databox),
+                new CoreHydrator($databox->get_sbas_id(), $databox->get_viewname(), $this->recordHelper),
-            new ThesaurusHydrator($this->getStructure(), $this->getThesaurus(), $candidateTerms),
+                new TitleHydrator($connection, $this->recordHelper),
-            new SubDefinitionHydrator($connection)
+                new MetadataHydrator($connection, $this->getStructure(), $this->recordHelper),
-        ), $fetcherDelegate);
+                new FlagHydrator($this->getStructure(), $databox),
+                new ThesaurusHydrator($this->getStructure(), $this->getThesaurus(), $candidateTerms),
+                new SubDefinitionHydrator($connection)
+            ],
+            $fetcherDelegate
+        );
 
         $fetcher->setBatchSize(200);
         $fetcher->onDrain(function() use ($candidateTerms) {

lib/Alchemy/Phrasea/SearchEngine/Elastic/ElasticsearchOptions.php

@@ -11,6 +11,10 @@ namespace Alchemy\Phrasea\SearchEngine\Elastic;
 
 class ElasticsearchOptions
 {
+    const POPULATE_ORDER_RID = "RECORD_ID";
+    const POPULATE_ORDER_MODDATE = "MODIFICATION_DATE";
+    const POPULATE_DIRECTION_ASC = "ASC";
+    const POPULATE_DIRECTION_DESC = "DESC";
     /** @var string */
     private $host;
     /** @var int */
@@ -25,6 +29,10 @@ class ElasticsearchOptions
     private $minScore;
     /** @var  bool */
     private $highlight;
+    /** @var string */
+    private $populateOrder;
+    /** @var string */
+    private $populateDirection;
 
     /** @var  int[] */
     private $_customValues;
@@ -46,6 +54,8 @@ class ElasticsearchOptions
             'replicas' => 0,
             'minScore' => 4,
             'highlight' => true,
+            'populate_order'     => self::POPULATE_ORDER_RID,
+            'populate_direction' => self::POPULATE_DIRECTION_DESC,
             'activeTab' => null,
         ];
 
@@ -63,6 +73,8 @@ class ElasticsearchOptions
         $self->setReplicas($options['replicas']);
         $self->setMinScore($options['minScore']);
         $self->setHighlight($options['highlight']);
+        $self->setPopulateOrder($options['populate_order']);
+        $self->setPopulateDirection($options['populate_direction']);
         $self->setActiveTab($options['activeTab']);
         foreach(self::getAggregableTechnicalFields() as $k => $f) {
             $self->setAggregableFieldLimit($k, $options[$k.'_limit']);
@@ -85,6 +97,8 @@ class ElasticsearchOptions
             'replicas' => $this->replicas,
             'minScore' => $this->minScore,
             'highlight' => $this->highlight,
+            'populate_order'     => $this->populateOrder,
+            'populate_direction' => $this->populateDirection,
             'activeTab' => $this->activeTab
         ];
         foreach(self::getAggregableTechnicalFields() as $k => $f) {
@@ -322,4 +336,60 @@ class ElasticsearchOptions
         ];
     }
 
+    /**
+     * @param string $order
+     * @return bool returns false if order is invalid
+     */
+    public function setPopulateOrder($order)
+    {
+        $order = strtoupper($order);
+        if (in_array($order, [self::POPULATE_ORDER_RID, self::POPULATE_ORDER_MODDATE])) {
+            $this->populateOrder = $order;
+
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * @return string
+     */
+    public function getPopulateOrderAsSQL()
+    {
+        static $orderAsColumn = [
+            self::POPULATE_ORDER_RID     => "`record_id`",
+            self::POPULATE_ORDER_MODDATE => "`moddate`",
+        ];
+
+        // populateOrder IS one of the keys (ensured by setPopulateOrder)
+        return $orderAsColumn[$this->populateOrder];
+    }
+
+    /**
+     * @param string $direction
+     * @return bool returns false if direction is invalid
+     */
+    public function setPopulateDirection($direction)
+    {
+        $direction = strtoupper($direction);
+        if (in_array($direction, [self::POPULATE_DIRECTION_DESC, self::POPULATE_DIRECTION_ASC])) {
+            $this->populateDirection = $direction;
+
+            return true;
+        }
+
+        return false;
+    }
+
+
+    /**
+     * @return string
+     */
+    public function getPopulateDirectionAsSQL()
+    {
+        // already a SQL word
+        return $this->populateDirection;
+    }
+
 }

lib/Alchemy/Phrasea/SearchEngine/Elastic/Indexer.php

@@ -16,13 +16,11 @@ use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\BulkOperation;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\RecordIndexer;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\TermIndexer;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\RecordQueuer;
-use appbox;
 use Closure;
 use Elasticsearch\Client;
 use Psr\Log\LoggerInterface;
 use igorw;
 use Psr\Log\NullLogger;
-use record_adapter;
 use Symfony\Component\Stopwatch\Stopwatch;
 use SplObjectStorage;
 
@@ -84,20 +82,64 @@ class Indexer
         $this->deleteQueue = new SplObjectStorage();
     }
 
-    public function createIndex($withMapping = true)
+    /**
+     * @return Index
+     */
+    public function getIndex()
     {
-        $params = array();
+        return $this->index;
-        $params['index'] = $this->index->getName();
+    }
-        $params['body']['settings']['number_of_shards'] = $this->index->getOptions()->getShards();
-        $params['body']['settings']['number_of_replicas'] = $this->index->getOptions()->getReplicas();
-        $params['body']['settings']['analysis'] = $this->index->getAnalysis();
 
-        if ($withMapping) {
+    public function createIndex($indexName = null)
-            $params['body']['mappings'][RecordIndexer::TYPE_NAME] = $this->index->getRecordIndex()->getMapping()->export();
+    {
-            $params['body']['mappings'][TermIndexer::TYPE_NAME]   = $this->index->getTermIndex()->getMapping()->export();
+        $aliasName = $this->index->getName();
+        if($indexName === null) {
+            $indexName = $aliasName;
         }
 
+        $now = sprintf("%s.%06d", Date('YmdHis'), 1000000*explode(' ', microtime())[0]) ;
+        $indexName .=  ('_' . $now);
+
+        $params = [
+            'index' => $indexName,
+            'body'  => [
+                'settings' => [
+                    'number_of_shards'   => $this->index->getOptions()->getShards(),
+                    'number_of_replicas' => $this->index->getOptions()->getReplicas(),
+                    'analysis'           => $this->index->getAnalysis()
+                ],
+                'mappings' => [
+                    RecordIndexer::TYPE_NAME => $this->index->getRecordIndex()->getMapping()->export(),
+                    TermIndexer::TYPE_NAME   => $this->index->getTermIndex()->getMapping()->export()
+                ]
+                //    ,
+                //    'aliases' => [
+                //        $aliasName => []
+                //    ]
+            ]
+        ];
+
         $this->client->indices()->create($params);
+
+        $params = [
+            'body' => [
+                'actions' => [
+                    [
+                        'add' => [
+                            'index' => $indexName,
+                            'alias' => $aliasName
+                        ]
+                    ]
+                ]
+            ]
+        ];
+
+        $this->client->indices()->updateAliases($params);
+
+        return [
+            'index' => $indexName,
+            'alias' => $aliasName
+        ];
     }
 
     public function updateMapping()
@@ -126,38 +168,129 @@ class Indexer
         ]);
     }
 
+    /**
+     * @param string $newIndexName
+     * @param string $newAliasName
+     * @return array
+     */
+    public function replaceIndex($newIndexName, $newAliasName)
+    {
+        $ret = [];
+
+        $oldIndexes = $this->client->indices()->getAlias(
+            [
+                'index' => $this->index->getName()
+            ]
+        );
+
+        // delete old alias(es), only one alias on one index should exist
+        foreach($oldIndexes as $oldIndexName => $data) {
+            foreach($data['aliases'] as $oldAliasName => $data2) {
+                $params['body']['actions'][] = [
+                    'remove' => [
+                        'alias' => $oldAliasName,
+                        'index' => $oldIndexName,
+                    ]
+                ];
+                $ret[] = [
+                    'action' => "ALIAS_REMOVE",
+                    'msg'    => sprintf('alias "%s" -> "%s" removed', $oldAliasName, $oldIndexName),
+                    'alias'  => $oldAliasName,
+                    'index'  => $oldIndexName,
+                ];
+            }
+        }
+
+        // create new alias
+        $params['body']['actions'][] = [
+            'add' => [
+                'alias' => $this->index->getName(),
+                'index' => $newIndexName,
+            ]
+        ];
+        $ret[] = [
+            'action' => "ALIAS_ADD",
+            'msg'   => sprintf('alias "%s" -> "%s" added', $this->index->getName(), $newIndexName),
+            'alias' => $this->index->getName(),
+            'index' => $newIndexName,
+        ];
+
+        //
+        $params['body']['actions'][] = [
+            'remove' => [
+                'alias' => $newAliasName,
+                'index' => $newIndexName,
+            ]
+        ];
+        $ret[] = [
+            'action' => "ALIAS_REMOVE",
+            'msg'    => sprintf('alias "%s" -> "%s" removed', $newAliasName, $newIndexName),
+            'alias'  => $newAliasName,
+            'index'  => $newIndexName,
+        ];
+
+
+        $this->client->indices()->updateAliases($params);
+
+        // delete old index(es), only one index should exist
+        $params = [
+            'index' => []
+        ];
+        foreach($oldIndexes as $oldIndexName => $data) {
+            $params['index'][] = $oldIndexName;
+            $ret[] = [
+                'action' => "INDEX_DELETE",
+                'msg'    => sprintf('index "%s" deleted', $oldIndexName),
+                'index'  => $oldIndexName,
+            ];
+        }
+        $this->client->indices()->delete(
+            $params
+        );
+
+        return $ret;
+    }
+
     public function populateIndex($what, \databox $databox)
     {
         $stopwatch = new Stopwatch();
         $stopwatch->start('populate');
 
-        $this->apply(function (BulkOperation $bulk) use ($what, $databox) {
+        $this->apply(
-            if ($what & self::THESAURUS) {
+            function (BulkOperation $bulk) use ($what, $databox) {
-                $this->termIndexer->populateIndex($bulk, $databox);
+                if ($what & self::THESAURUS) {
+                    $this->termIndexer->populateIndex($bulk, $databox);
 
-                // Record indexing depends on indexed terms so we need to make
+                    // Record indexing depends on indexed terms so we need to make
-                // everything ready to search
+                    // everything ready to search
-                $bulk->flush();
+                    $bulk->flush();
-                $this->client->indices()->refresh();
+                    $this->client->indices()->refresh();
-                $this->client->indices()->clearCache();
+                }
-                $this->client->indices()->flushSynced();
-            }
 
-            if ($what & self::RECORDS) {
+                if ($what & self::RECORDS) {
-                $databox->clearCandidates();
+                    $databox->clearCandidates();
-                $this->recordIndexer->populateIndex($bulk, $databox);
+                    $this->recordIndexer->populateIndex($bulk, $databox);
 
-                // Final flush
+                    // Final flush
-                $bulk->flush();
+                    $bulk->flush();
-            }
+                }
-        }, $this->index);
+            },
+            $this->index
+        );
 
         // Optimize index
-        $params = array('index' => $this->index->getName());
+        $this->client->indices()->optimize(
-        $this->client->indices()->optimize($params);
+            [
+                'index' => $this->index->getName()
+            ]
+        );
 
         $event = $stopwatch->stop('populate');
-        printf("Indexation finished in %s min (Mem. %s Mo)", ($event->getDuration()/1000/60), bcdiv($event->getMemory(), 1048576, 2));
+
+        return [
+            'duration' => $event->getDuration(),
+            'memory'   => $event->getMemory()
+        ];
     }
 
     public function migrateMappingForDatabox($databox)

lib/Alchemy/Phrasea/SearchEngine/Elastic/Indexer/Record/Fetcher.php

@@ -12,6 +12,7 @@
 namespace Alchemy\Phrasea\SearchEngine\Elastic\Indexer\Record;
 
 use Alchemy\Phrasea\Core\PhraseaTokens;
+use Alchemy\Phrasea\SearchEngine\Elastic\ElasticsearchOptions;
 use Alchemy\Phrasea\SearchEngine\Elastic\Exception\Exception;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\Record\Delegate\FetcherDelegate;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer\Record\Delegate\FetcherDelegateInterface;
@@ -24,6 +25,7 @@ use PDO;
 class Fetcher
 {
     private $databox;
+    private $options;
     private $connection;
     private $statement;
     private $delegate;
@@ -36,9 +38,10 @@ class Fetcher
     private $postFetch;
     private $onDrain;
 
-    public function __construct(databox $databox, array $hydrators, FetcherDelegateInterface $delegate = null)
+    public function __construct(databox $databox,ElasticsearchOptions $options, array $hydrators, FetcherDelegateInterface $delegate = null)
     {
         $this->databox = $databox;
+        $this->options = $options;
         $this->connection = $databox->get_connection();;
         $this->hydrators  = $hydrators;
         $this->delegate   = $delegate ?: new FetcherDelegate();
@@ -136,7 +139,7 @@ class Fetcher
                  . " FROM (record r INNER JOIN coll c ON (c.coll_id = r.coll_id))"
                  . " LEFT JOIN subdef ON subdef.record_id=r.record_id AND subdef.name='document'"
                  . " -- WHERE"
-                 . " ORDER BY r.record_id DESC"
+                 . " ORDER BY " . $this->options->getPopulateOrderAsSQL() . " " . $this->options->getPopulateDirectionAsSQL()
                  . " LIMIT :offset, :limit";
 
             $where = $this->delegate->buildWhereClause();