From de77ed1fd44ebfee485ce255b5b8f758b3079755 Mon Sep 17 00:00:00 2001
From: Thomas Klausner <domm@plix.at>
Date: Wed, 15 Nov 2023 22:16:23 +0100
Subject: [PATCH] Bug 35345: Add --where option to rebuild_elasticsearch.pl
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Sometimes we need to only re-index a subset of our bibliographic data or authorities. Currently this is only possible by enumerating all id (-bn or -ai), which does not work well when indexing eg 100.000 items of a 2.000.000 DB. Re-indexing everything is also overkill.

This patch adds an `--where` flag to misc/search_tools/rebuild_elasticsearch.pl which can take arbitrary SQL (that of course has to match the respective tables) and adds it as an additional param to the resultset to index

To test, start koha-testing-docker with ElasticSearch enabled, for example via `ktd --es7 up

Before applying the patch, rebuild_elasticsearch will index all data:

Biblios:
$ misc/search_tools/rebuild_elasticsearch.pl -b -v
[12387] Checking state of biblios index
[12387] Indexing biblios
[12387] Committing final records...
[12387] Total 435 records indexed
(there might be a waring regarding a broken biblio, which can be ignored)

Auth:
$ misc/search_tools/rebuild_elasticsearch.pl -a -v
[12546] Checking state of authorities index
[12546] Indexing authorities
[12546] 1000 records processed
[12546] Committing final records...
[12546] Total 1706 records indexed

Now apply the patch

Biblio, limit by range of biblioid:
$ misc/search_tools/rebuild_elasticsearch.pl -b -v --where "biblionumber between 100 and 150"
[12765] Checking state of biblios index
[12765] Indexing biblios
[12765] Committing final records...
[12765] Total 50 records indexed

Note that only 50 records where indexed (instead of the whole set of 435 records)

Auth, limit by authtypecode:
$ misc/search_tools/rebuild_elasticsearch.pl -a -v --where "authtypecode = 'GEOGR_NAME'"
[12848] Checking state of authorities index
[12848] Indexing authorities
[12848] Committing final records...
[12848] Total 142 records indexed

Again, only 142 have been indexed.

Note that this patch also includes perltidied versions of the affected files (because the qa script complained)

Sponsored-by: Steiermärkische Landesbibliothek
Signed-off-by: David Nind <david@davidnind.com>
---
 Koha/BiblioUtils.pm                        |  43 +++++----
 Koha/MetadataRecord/Authority.pm           |  99 +++++++++++--------
 misc/search_tools/rebuild_elasticsearch.pl | 105 ++++++++++++---------
 3 files changed, 148 insertions(+), 99 deletions(-)

diff --git a/Koha/BiblioUtils.pm b/Koha/BiblioUtils.pm
index cb7baed0db..d28dbf7db3 100644
--- a/Koha/BiblioUtils.pm
+++ b/Koha/BiblioUtils.pm
@@ -34,7 +34,6 @@ use Koha::MetadataIterator;
 use Koha::Database;
 use Modern::Perl;
 
-
 use base qw(Koha::MetadataRecord);
 
 __PACKAGE__->mk_accessors(qw( record schema id datatype ));
@@ -51,8 +50,8 @@ the biblionumber can be provided too.
 =cut
 
 sub new {
-    my $class = shift;
-    my $record = shift;
+    my $class        = shift;
+    my $record       = shift;
     my $biblionumber = shift;
 
     my $self = $class->SUPER::new(
@@ -89,10 +88,10 @@ It will return C<undef> if the biblio doesn't exist.
 =cut
 
 sub get_from_biblionumber {
-    my ($class, $bibnum, %options) = @_;
+    my ( $class, $bibnum, %options ) = @_;
 
-    my $marc = $class->get_marc_biblio($bibnum, %options);
-    return $class->new($marc, $bibnum);
+    my $marc = $class->get_marc_biblio( $bibnum, %options );
+    return $class->new( $marc, $bibnum );
 }
 
 =head2 get_all_biblios_iterator
@@ -120,34 +119,41 @@ records instead of all.
 =cut
 
 sub get_all_biblios_iterator {
-    my ($class, %options) = @_;
+    my ( $class, %options ) = @_;
 
     my $search_terms = {};
-    my ($slice_modulo, $slice_count);
-    if ($options{slice}) {
-        $slice_count = $options{slice}->{count};
+    my ( $slice_modulo, $slice_count );
+    if ( $options{slice} ) {
+        $slice_count  = $options{slice}->{count};
         $slice_modulo = $options{slice}->{index};
         $search_terms = \[ 'mod(biblionumber, ?) = ?', $slice_count, $slice_modulo ];
     }
 
     my $search_options = { columns => [qw/ biblionumber /] };
-    if ( $options{desc} ){
-        $search_options->{order_by}  = { -desc => 'biblionumber' };
+    if ( $options{desc} ) {
+        $search_options->{order_by} = { -desc => 'biblionumber' };
     }
 
     my $database = Koha::Database->new();
     my $schema   = $database->schema();
-    my $rs = Koha::Biblios->search(
+    my $rs       = Koha::Biblios->search(
         $search_terms,
-        $search_options );
+        $search_options
+    );
+
+    if ( my $sql = $options{where} ) {
+        $rs = $rs->search( \[$sql] );
+    }
+
     my $next_func = sub {
+
         # Warn and skip bad records, otherwise we break the loop
         while (1) {
             my $row = $rs->next();
             return if !$row;
             my $next = eval {
-                my $marc = $row->metadata->record({ embed_items => 1 });
-                $class->new($marc, $row->biblionumber);
+                my $marc = $row->metadata->record( { embed_items => 1 } );
+                $class->new( $marc, $row->biblionumber );
             };
             if ($@) {
                 warn sprintf "Something went wrong reading record for biblio %s: %s\n", $row->biblionumber, $@;
@@ -180,10 +186,9 @@ If set to true, item data is embedded in the record. Default is to not do this.
 =cut
 
 sub get_marc_biblio {
-    my ($class, $bibnum, %options) = @_;
+    my ( $class, $bibnum, %options ) = @_;
 
-    my $record = Koha::Biblios->find($bibnum)
-      ->metadata->record( { $options{item_data} ? ( embed_items => 1 ) : () } );
+    my $record = Koha::Biblios->find($bibnum)->metadata->record( { $options{item_data} ? ( embed_items => 1 ) : () } );
     return $record;
 }
 
diff --git a/Koha/MetadataRecord/Authority.pm b/Koha/MetadataRecord/Authority.pm
index d16970a828..e9a8466c43 100644
--- a/Koha/MetadataRecord/Authority.pm
+++ b/Koha/MetadataRecord/Authority.pm
@@ -67,7 +67,6 @@ sub new {
     return $self;
 }
 
-
 =head2 get_from_authid
 
     my $auth = Koha::MetadataRecord::Authority->get_from_authid($authid);
@@ -80,23 +79,35 @@ unfortunate but unavoidable fact.
 =cut
 
 sub get_from_authid {
-    my $class = shift;
-    my $authid = shift;
+    my $class       = shift;
+    my $authid      = shift;
     my $marcflavour = lc C4::Context->preference("marcflavour");
 
-    my $dbh=C4::Context->dbh;
-    my $sth=$dbh->prepare("select authtypecode, marcxml from auth_header where authid=?");
+    my $dbh = C4::Context->dbh;
+    my $sth = $dbh->prepare("select authtypecode, marcxml from auth_header where authid=?");
     $sth->execute($authid);
-    my ($authtypecode, $marcxml) = $sth->fetchrow;
-    my $record=eval {MARC::Record->new_from_xml(StripNonXmlChars($marcxml),'UTF-8',
-        (C4::Context->preference("marcflavour") eq "UNIMARC"?"UNIMARCAUTH":C4::Context->preference("marcflavour")))};
+    my ( $authtypecode, $marcxml ) = $sth->fetchrow;
+    my $record = eval {
+        MARC::Record->new_from_xml(
+            StripNonXmlChars($marcxml), 'UTF-8',
+            (
+                C4::Context->preference("marcflavour") eq "UNIMARC"
+                ? "UNIMARCAUTH"
+                : C4::Context->preference("marcflavour")
+            )
+        );
+    };
     return if ($@);
     $record->encoding('UTF-8');
 
-    my $self = $class->SUPER::new( { authid => $authid,
-                                     authtypecode => $authtypecode,
-                                     schema => $marcflavour,
-                                     record => $record });
+    my $self = $class->SUPER::new(
+        {
+            authid       => $authid,
+            authtypecode => $authtypecode,
+            schema       => $marcflavour,
+            record       => $record
+        }
+    );
 
     bless $self, $class;
     return $self;
@@ -111,16 +122,24 @@ Create the Koha::MetadataRecord::Authority object associated with the provided a
 =cut
 
 sub get_from_breeding {
-    my $class = shift;
+    my $class            = shift;
     my $import_record_id = shift;
-    my $marcflavour = lc C4::Context->preference("marcflavour");
+    my $marcflavour      = lc C4::Context->preference("marcflavour");
 
-    my $dbh=C4::Context->dbh;
-    my $sth=$dbh->prepare("select marcxml from import_records where import_record_id=? and record_type='auth';");
+    my $dbh = C4::Context->dbh;
+    my $sth = $dbh->prepare("select marcxml from import_records where import_record_id=? and record_type='auth';");
     $sth->execute($import_record_id);
     my $marcxml = $sth->fetchrow;
-    my $record=eval {MARC::Record->new_from_xml(StripNonXmlChars($marcxml),'UTF-8',
-        (C4::Context->preference("marcflavour") eq "UNIMARC"?"UNIMARCAUTH":C4::Context->preference("marcflavour")))};
+    my $record  = eval {
+        MARC::Record->new_from_xml(
+            StripNonXmlChars($marcxml), 'UTF-8',
+            (
+                C4::Context->preference("marcflavour") eq "UNIMARC"
+                ? "UNIMARCAUTH"
+                : C4::Context->preference("marcflavour")
+            )
+        );
+    };
     return if ($@);
     $record->encoding('UTF-8');
 
@@ -131,10 +150,13 @@ sub get_from_breeding {
     require C4::AuthoritiesMarc;
     my $authtypecode = C4::AuthoritiesMarc::GuessAuthTypeCode($record);
 
-    my $self = $class->SUPER::new( {
-                                     schema => $marcflavour,
-                                     authtypecode => $authtypecode,
-                                     record => $record });
+    my $self = $class->SUPER::new(
+        {
+            schema       => $marcflavour,
+            authtypecode => $authtypecode,
+            record       => $record
+        }
+    );
 
     bless $self, $class;
     return $self;
@@ -142,8 +164,8 @@ sub get_from_breeding {
 
 sub authorized_heading {
     my ($self) = @_;
-    if ($self->schema =~ m/marc/) {
-        return Koha::Util::MARC::getAuthorityAuthorizedHeading($self->record, $self->schema);
+    if ( $self->schema =~ m/marc/ ) {
+        return Koha::Util::MARC::getAuthorityAuthorizedHeading( $self->record, $self->schema );
     }
     return;
 }
@@ -173,14 +195,12 @@ records instead of all.
 =cut
 
 sub get_all_authorities_iterator {
-    my ($self, %options) = @_;
+    my ( $self, %options ) = @_;
 
-    my $search_terms = {
-        marcxml => { '!=', undef }
-    };
-    my ($slice_modulo, $slice_count);
-    if ($options{slice}) {
-        $slice_count = $options{slice}->{count};
+    my $search_terms = { marcxml => { '!=', undef } };
+    my ( $slice_modulo, $slice_count );
+    if ( $options{slice} ) {
+        $slice_count  = $options{slice}->{count};
         $slice_modulo = $options{slice}->{index};
         $search_terms = {
             '-and' => [
@@ -191,24 +211,29 @@ sub get_all_authorities_iterator {
     }
 
     my $search_options->{columns} = [qw/ authid /];
-    if ($options{desc}) {
+    if ( $options{desc} ) {
         $search_options->{order_by} = { -desc => 'authid' };
     }
 
     my $database = Koha::Database->new();
     my $schema   = $database->schema();
-    my $rs =
-      $schema->resultset('AuthHeader')->search(
+    my $rs       = $schema->resultset('AuthHeader')->search(
         $search_terms,
-        $search_options);
+        $search_options
+    );
+
+    if ( my $sql = $options{where} ) {
+        $rs = $rs->search( \[$sql] );
+    }
     my $next_func = sub {
+
         # Warn and skip bad records, otherwise we break the loop
         while (1) {
             my $row = $rs->next();
             return if !$row;
 
-            my $auth = __PACKAGE__->get_from_authid($row->authid);
-            if (!$auth) {
+            my $auth = __PACKAGE__->get_from_authid( $row->authid );
+            if ( !$auth ) {
                 warn "Something went wrong reading record for authority $row->authid: $@\n";
                 next;
             }
diff --git a/misc/search_tools/rebuild_elasticsearch.pl b/misc/search_tools/rebuild_elasticsearch.pl
index 337ac19dd2..ffb7301238 100755
--- a/misc/search_tools/rebuild_elasticsearch.pl
+++ b/misc/search_tools/rebuild_elasticsearch.pl
@@ -34,6 +34,7 @@ B<rebuild_elasticsearch.pl>
 [B<--desc>]
 [B<-bn|--bnumber>]
 [B<-ai|--authid>]
+[B<-w|--where SQL>]
 [B<-p|--processes>]
 [B<-v|--verbose>]
 [B<-h|--help>]
@@ -87,6 +88,10 @@ repeated.
 Only index the supplied authority id, mostly for testing purposes. May be
 repeated.
 
+=item B<-w|--where>
+
+Pass some additional SQL to limit the records to be indexed.
+
 =item B<-p|--processes>
 
 Number of processes to use for indexing. This can be used to do more indexing
@@ -125,13 +130,13 @@ use Pod::Usage qw( pod2usage );
 use Try::Tiny qw( catch try );
 
 my $verbose = 0;
-my $commit = 5000;
-my ($delete, $reset, $help, $man, $processes);
-my ($index_biblios, $index_authorities);
-my (@biblionumbers,@authids);
+my $commit  = 5000;
+my ( $delete, $reset, $help, $man, $processes );
+my ( $index_biblios, $index_authorities );
+my ( @biblionumbers, @authids, $where );
 my $desc;
 
-$|=1; # flushes output
+$| = 1;    # flushes output
 
 GetOptions(
     'c|commit=i'    => \$commit,
@@ -142,6 +147,7 @@ GetOptions(
     'desc'          => \$desc,
     'bn|bnumber=i'  => \@biblionumbers,
     'ai|authid=i'   => \@authids,
+    'w|where=s'     => \$where,
     'p|processes=i' => \$processes,
     'v|verbose+'    => \$verbose,
     'h|help'        => \$help,
@@ -149,62 +155,69 @@ GetOptions(
 );
 
 # Default is to do both
-unless ($index_authorities || $index_biblios) {
+unless ( $index_authorities || $index_biblios ) {
     $index_authorities = $index_biblios = 1;
 }
 
-if ($processes && ( @biblionumbers || @authids) ) {
+if ( $processes && ( @biblionumbers || @authids ) ) {
     die "Argument p|processes cannot be combined with bn|bnumber or ai|authid";
 }
 
-pod2usage(1) if $help;
+pod2usage(1)                                 if $help;
 pod2usage( -exitstatus => 0, -verbose => 2 ) if $man;
 
 _sanity_check();
 
-if ($reset){
+if ($reset) {
     Koha::SearchEngine::Elasticsearch->reset_elasticsearch_mappings;
     $delete = 1;
 }
 
-_verify_index_state($Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX, $delete) if ($index_biblios);
-_verify_index_state($Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX, $delete) if ($index_authorities);
+_verify_index_state( $Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX,     $delete ) if ($index_biblios);
+_verify_index_state( $Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX, $delete ) if ($index_authorities);
 
 my $slice_index = 0;
 my $slice_count = ( $processes //= 1 );
 my %iterator_options;
 
-if ($slice_count > 1) {
+if ( $slice_count > 1 ) {
+
     # Fire up child processes for processing slices from 2 on. This main process will handle slice 1.
     $slice_index = 0;
-    for (my $proc = 1; $proc < $slice_count; $proc++) {
+    for ( my $proc = 1 ; $proc < $slice_count ; $proc++ ) {
         my $pid = fork();
         die "Failed to fork a child process\n" unless defined $pid;
-        if ($pid == 0) {
+        if ( $pid == 0 ) {
+
             # Child process, give it a slice to process
             $slice_index = $proc;
             last;
         }
     }
+
     # Fudge the commit count a bit to spread out the Elasticsearch commits
     $commit *= 1 + 0.10 * $slice_index;
-    $commit = int( $commit );
-    _log(1, "Processing slice @{[$slice_index + 1]} of $slice_count\n");
+    $commit = int($commit);
+    _log( 1, "Processing slice @{[$slice_index + 1]} of $slice_count\n" );
     $iterator_options{slice} = { index => $slice_index, count => $slice_count };
 }
 
-if( $desc ){
+if ($desc) {
     $iterator_options{desc} = 1;
 }
 
+if ($where) {
+    $iterator_options{where} = $where;
+}
+
 my $next;
 if ($index_biblios) {
-    _log(1, "Indexing biblios\n");
+    _log( 1, "Indexing biblios\n" );
     if (@biblionumbers) {
         $next = sub {
             my $r = shift @biblionumbers;
             return () unless defined $r;
-            return ($r, Koha::BiblioUtils->get_from_biblionumber($r, item_data => 1 ));
+            return ( $r, Koha::BiblioUtils->get_from_biblionumber( $r, item_data => 1 ) );
         };
     } else {
         my $records = Koha::BiblioUtils->get_all_biblios_iterator(%iterator_options);
@@ -212,16 +225,16 @@ if ($index_biblios) {
             $records->next();
         }
     }
-    _do_reindex($next, $Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX);
+    _do_reindex( $next, $Koha::SearchEngine::Elasticsearch::BIBLIOS_INDEX );
 }
 if ($index_authorities) {
-    _log(1, "Indexing authorities\n");
+    _log( 1, "Indexing authorities\n" );
     if (@authids) {
         $next = sub {
             my $r = shift @authids;
             return () unless defined $r;
             my $a = Koha::MetadataRecord::Authority->get_from_authid($r);
-            return ($r, $a);
+            return ( $r, $a );
         };
     } else {
         my $records = Koha::MetadataRecord::Authority->get_all_authorities_iterator(%iterator_options);
@@ -229,12 +242,13 @@ if ($index_authorities) {
             $records->next();
         }
     }
-    _do_reindex($next, $Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX);
+    _do_reindex( $next, $Koha::SearchEngine::Elasticsearch::AUTHORITIES_INDEX );
 }
 
-if ($slice_index == 0) {
+if ( $slice_index == 0 ) {
+
     # Main process, wait for children
-    for (my $proc = 1; $proc < $processes; $proc++) {
+    for ( my $proc = 1 ; $proc < $processes ; $proc++ ) {
         wait();
     }
 }
@@ -252,21 +266,22 @@ Checks the index state and recreates it if requested.
 sub _verify_index_state {
     my ( $index_name, $recreate ) = @_;
 
-    _log(1, "Checking state of $index_name index\n");
+    _log( 1, "Checking state of $index_name index\n" );
     my $indexer = Koha::SearchEngine::Elasticsearch::Indexer->new( { index => $index_name } );
 
     if ($recreate) {
-        _log(1, "Dropping and recreating $index_name index\n");
+        _log( 1, "Dropping and recreating $index_name index\n" );
         $indexer->drop_index() if $indexer->index_exists();
         $indexer->create_index();
-    }
-    elsif (!$indexer->index_exists) {
+    } elsif ( !$indexer->index_exists ) {
+
         # Create index if does not exist
         $indexer->create_index();
-    } elsif ($indexer->is_index_status_ok) {
+    } elsif ( $indexer->is_index_status_ok ) {
+
         # Update mapping unless index is some kind of problematic state
         $indexer->update_mappings();
-    } elsif ($indexer->is_index_status_recreate_required) {
+    } elsif ( $indexer->is_index_status_recreate_required ) {
         warn qq/Index "$index_name" has status "recreate required", suggesting it should be recreated/;
     }
 }
@@ -293,7 +308,7 @@ sub _do_reindex {
         my $record = $record->record;
         $count++;
         if ( $verbose == 1 ) {
-            _log( 1, "$count records processed\n" ) if ( $count % 1000 == 0);
+            _log( 1, "$count records processed\n" ) if ( $count % 1000 == 0 );
         } else {
             _log( 2, "$id\n" );
         }
@@ -303,13 +318,13 @@ sub _do_reindex {
         if ( !( --$commit_count ) ) {
             _log( 1, "Committing $commit records...\n" );
             my $response;
-            try{
+            try {
                 $response = $indexer->update_index( \@id_buffer, \@commit_buffer );
                 _handle_response($response);
                 _log( 1, "Commit complete\n" );
             } catch {
-                _log(1,"Elasticsearch exception thrown: ".$_->type."\n");
-                _log(2,"Details: ".$_->details."\n");
+                _log( 1, "Elasticsearch exception thrown: " . $_->type . "\n" );
+                _log( 2, "Details: " . $_->details . "\n" );
             };
             $commit_count  = $commit;
             @id_buffer     = ();
@@ -333,6 +348,7 @@ Checks some basic stuff to ensure that it's sane before we start.
 =cut
 
 sub _sanity_check {
+
     # Do we have an elasticsearch block defined?
     my $conf = C4::Context->config('elasticsearch');
     die "No 'elasticsearch' block is defined in koha-conf.xml.\n" if ( !$conf );
@@ -346,14 +362,17 @@ Parse the return from update_index and display errors depending on verbosity of
 
 sub _handle_response {
     my ($response) = @_;
-    if( $response->{errors} eq 'true' ){
+    if ( $response->{errors} eq 'true' ) {
         _log( 1, "There were errors during indexing\n" );
-        if ( $verbose > 1 ){
-            foreach my $item (@{$response->{items}}){
+        if ( $verbose > 1 ) {
+            foreach my $item ( @{ $response->{items} } ) {
                 next unless defined $item->{index}->{error};
-                print "Record #" . $item->{index}->{_id} . " " .
-                      $item->{index}->{error}->{reason} . " (" . $item->{index}->{error}->{type} . ") : " .
-                      $item->{index}->{error}->{caused_by}->{type} . " (" . $item->{index}->{error}->{caused_by}->{reason} . ")\n";
+                print "Record #"
+                    . $item->{index}->{_id} . " "
+                    . $item->{index}->{error}->{reason} . " ("
+                    . $item->{index}->{error}->{type} . ") : "
+                    . $item->{index}->{error}->{caused_by}->{type} . " ("
+                    . $item->{index}->{error}->{caused_by}->{reason} . ")\n";
             }
         }
     }
@@ -371,7 +390,7 @@ include a trailing newline automatically.
 =cut
 
 sub _log {
-    my ($level, $msg) = @_;
+    my ( $level, $msg ) = @_;
 
-    print "[$$] $msg" if ($verbose >= $level);
+    print "[$$] $msg" if ( $verbose >= $level );
 }
-- 
2.30.2