inboxwritable: eidx_key for external index

[public-inbox.git] / lib / PublicInbox / SearchIdxShard.pm

diff --git a/lib/PublicInbox/SearchIdxShard.pm b/lib/PublicInbox/SearchIdxShard.pm
index c0f8be8964939e5c95aa33b4894b0bde8cc05d40..fa77a9f9b96a0780d1484f16a2293df0f4e674b3 100644 (file)
--- a/lib/PublicInbox/SearchIdxShard.pm
+++ b/lib/PublicInbox/SearchIdxShard.pm
@@ -7,6 +7,7 @@ package PublicInbox::SearchIdxShard;
 use strict;
 use v5.10.1;
 use parent qw(PublicInbox::SearchIdx);
+use bytes qw(length);
 use IO::Handle (); # autoflush
 use PublicInbox::Eml;
 
@@ -47,6 +48,13 @@ sub spawn_worker {
        close $r or die "failed to close: $!";
 }
 
+sub eml ($$) {
+       my ($r, $len) = @_;
+       my $n = read($r, my $bref, $len) or die "read: $!\n";
+       $n == $len or die "short read: $n != $len\n";
+       PublicInbox::Eml->new(\$bref);
+}
+
 # this reads all the writes to $self->{w} from the parent process
 sub shard_worker_loop ($$$$$) {
        my ($self, $v2w, $r, $shard, $bnote) = @_;
@@ -65,47 +73,93 @@ sub shard_worker_loop ($$$$$) {
                                        die "write failed for barrier $!\n";
                } elsif ($line =~ /\AD ([a-f0-9]{40,}) ([0-9]+)\n\z/s) {
                        $self->remove_by_oid($1, $2 + 0);
+               } elsif ($line =~ s/\A\+X //) {
+                       my ($len, $docid, $xnum, $oid, $eidx_key) =
+                                                       split(/ /, $line, 5);
+                       $self->add_xref3($docid, $xnum, $oid, $eidx_key,
+                                               eml($r, $len));
+               } elsif ($line =~ s/\A-X //) {
+                       my ($len, $docid, $xnum, $oid, $eidx_key) =
+                                                       split(/ /, $line, 5);
+                       $self->remove_xref3($docid, $xnum, $oid,
+                                               $eidx_key, eml($r, $len));
                } else {
                        chomp $line;
+                       my $eidx_key;
+                       if ($line =~ s/\AX(.+)\0//) {
+                               $eidx_key = $1;
+                       }
                        # n.b. $mid may contain spaces(!)
-                       my ($to_read, $bytes, $num, $blob, $ds, $ts, $mid) =
-                                                       split(/ /, $line, 7);
+                       my ($len, $bytes, $num, $oid, $ds, $ts, $tid, $mid)
+                               = split(/ /, $line, 8);
                        $self->begin_txn_lazy;
-                       my $n = read($r, my $msg, $to_read) or die "read: $!\n";
-                       $n == $to_read or die "short read: $n != $to_read\n";
-                       my $mime = PublicInbox::Eml->new(\$msg);
                        my $smsg = bless {
                                bytes => $bytes,
                                num => $num + 0,
-                               blob => $blob,
+                               blob => $oid,
                                mid => $mid,
+                               tid => $tid,
                                ds => $ds,
                                ts => $ts,
                        }, 'PublicInbox::Smsg';
-                       $self->add_message($mime, $smsg);
+                       $smsg->{eidx_key} = $eidx_key if defined($eidx_key);
+                       $self->add_message(eml($r, $len), $smsg);
                }
        }
        $self->worker_done;
 }
 
 sub index_raw {
-       my ($self, $msgref, $eml, $smsg) = @_;
+       my ($self, $msgref, $eml, $smsg, $ibx) = @_;
        if (my $w = $self->{w}) {
+               if ($ibx) {
+                       print $w 'X', $ibx->eidx_key, "\0" or die
+                               "failed to write shard: $!\n";
+               }
+               $msgref //= \($eml->as_string);
+               $smsg->{raw_bytes} //= length($$msgref);
                # mid must be last, it can contain spaces (but not LF)
                print $w join(' ', @$smsg{qw(raw_bytes bytes
-                                               num blob ds ts mid)}),
+                                               num blob ds ts tid mid)}),
                        "\n", $$msgref or die "failed to write shard $!\n";
        } else {
                if ($eml) {
-                       undef $$msgref;
+                       undef($$msgref) if $msgref;
                } else { # --xapian-only + --sequential-shard:
                        $eml = PublicInbox::Eml->new($msgref);
                }
                $self->begin_txn_lazy;
+               $smsg->{eidx_key} = $ibx->eidx_key if $ibx;
                $self->add_message($eml, $smsg);
        }
 }
 
+sub shard_add_xref3 {
+       my ($self, $docid, $xnum, $oid, $xibx, $eml) = @_;
+       my $eidx_key = $xibx->eidx_key;
+       if (my $w = $self->{w}) {
+               my $hdr = $eml->header_obj->as_string;
+               my $len = length($hdr);
+               print $w "+X $len $docid $xnum $oid $eidx_key\n", $hdr or
+                       die "failed to write shard: $!";
+       } else {
+               $self->add_xref3($docid, $xnum, $oid, $eidx_key, $eml);
+       }
+}
+
+sub shard_remove_xref3 {
+       my ($self, $docid, $oid, $xibx, $eml) = @_;
+       my $eidx_key = $xibx->eidx_key;
+       if (my $w = $self->{w}) {
+               my $hdr = $eml->header_obj->as_string;
+               my $len = length($hdr);
+               print $w "-X $len $docid $oid $eidx_key\n", $hdr or
+                       die "failed to write shard: $!";
+       } else {
+               $self->remove_xref3($docid, $oid, $eidx_key, $eml);
+       }
+}
+
 sub atfork_child {
        close $_[0]->{w} or die "failed to close write pipe: $!\n";
 }