trigger/KinoSearch.pm

   1 use KinoSearch::Index::Indexer;
   2 use KinoSearch::Plan::Schema;
   3 use KinoSearch::Analysis::PolyAnalyzer;
   4 use KinoSearch::Plan::FullTextType;
   5
   6 # Create a Schema which defines index fields.
   7 my $schema = KinoSearch::Plan::Schema->new;
   8 my $polyanalyzer = KinoSearch::Analysis::PolyAnalyzer->new(
   9         language => 'en',
  10 );
  11 my $type = KinoSearch::Plan::FullTextType->new(
  12         analyzer => $polyanalyzer,
  13 );
  14 my $blob_type = KinoSearch::Plan::BlobType->new( stored => 1 );
  15 my $string_type = KinoSearch::Plan::StringType->new;
  16 $schema->spec_field( name => '_id',   type => $string_type );
  17 $schema->spec_field( name => '_rev', type => $string_type );
  18 $schema->spec_field( name => 'doc', type => $blob_type );
  19
  20 # Create the index and add documents.
  21 our $indexer;
  22
  23
  24 sub _indexer {
  25         $indexer ||= KinoSearch::Index::Indexer->new(
  26                 schema => $schema,
  27                 index  => '/tmp/index',
  28                 create => 1,
  29         );
  30 };
  31
  32 sub flatten {
  33         my ($flat,$data,$prefix) = @_;
  34         if ( ref $data eq '' ) {
  35                 $$flat->{$prefix} .= "\n" . $data;
  36                 $$flat->{$prefix} =~ s/^\n//; # strip first
  37         } elsif ( ref $data eq 'HASH' ) {
  38                 foreach my $key ( keys %$data ) {
  39                         my $full_prefix = $prefix ? $prefix . '.' : '';
  40                         $full_prefix .= $key;
  41                         flatten( $flat, $data->{$key}, $full_prefix );
  42                 }
  43         } elsif ( ref $data eq 'ARRAY' ) {
  44                 foreach my $el ( @$data ) {
  45                         flatten( $flat, $el, $prefix );
  46                 }
  47         } elsif ( ref $data eq 'Mojo::JSON::_Bool' ) {
  48                 $$flat->{$prefix} = $data;
  49         } else {
  50                 die "unsupported ",ref($data)," from ",dump($data);
  51         }
  52 }
  53
  54 sub filter {
  55         my $change = shift;
  56         my $doc = $change->{doc} || next;
  57
  58         if ( $doc->{_deleted} ) {
  59                 warn "# filter DELETE\n";
  60                 _indexer->delete_by_term( field => '_id', term => $doc->{_id} );
  61                 return 0;
  62         }
  63
  64         my $flat;
  65         flatten( \$flat, $doc, '' );
  66         foreach my $field ( keys %$flat ) {
  67                 next if $schema->fetch_type($field);
  68                 $schema->spec_field( name => $field, type => $type );
  69                 warn "# +++ $field\n";
  70         }
  71         $flat->{doc} = $json->encode($doc);
  72         warn "# add_doc ",dump($flat);
  73         _indexer->add_doc($flat);
  74         return 0;
  75 }
  76
  77 sub commit {
  78         return unless $indexer;
  79         $indexer->commit;
  80         undef $indexer;
  81         warn "# commit index done\n";
  82 }
  83
  84 1;