#define C_KINO_DOCWRITER
#include "KinoSearch/Util/ToolSet.h"
#include "KinoSearch/Index/DocWriter.h"
#include "KinoSearch/Document/Doc.h"
#include "KinoSearch/Index/DocReader.h"
#include "KinoSearch/Index/Inverter.h"
#include "KinoSearch/Index/PolyReader.h"
#include "KinoSearch/Index/Segment.h"
#include "KinoSearch/Index/SegReader.h"
#include "KinoSearch/Index/Snapshot.h"
#include "KinoSearch/Plan/FieldType.h"
#include "KinoSearch/Plan/Schema.h"
#include "KinoSearch/Store/Folder.h"
#include "KinoSearch/Store/OutStream.h"
static OutStream*
S_lazy_init(DocWriter *self);
int32_t DocWriter_current_file_format = 2;
DocWriter*
DocWriter_new(Schema *schema, Snapshot *snapshot, Segment *segment,
PolyReader *polyreader)
{
DocWriter *self = (DocWriter*)VTable_Make_Obj(DOCWRITER);
return DocWriter_init(self, schema, snapshot, segment, polyreader);
}
DocWriter*
DocWriter_init(DocWriter *self, Schema *schema, Snapshot *snapshot,
Segment *segment, PolyReader *polyreader)
{
DataWriter_init((DataWriter*)self, schema, snapshot, segment, polyreader);
return self;
}
void
DocWriter_destroy(DocWriter *self)
{
DECREF(self->dat_out);
DECREF(self->ix_out);
SUPER_DESTROY(self, DOCWRITER);
}
static OutStream*
S_lazy_init(DocWriter *self)
{
if (!self->dat_out) {
Folder *folder = self->folder;
CharBuf *seg_name = Seg_Get_Name(self->segment);
// Get streams.
{
CharBuf *ix_file = CB_newf("%o/documents.ix", seg_name);
self->ix_out = Folder_Open_Out(folder, ix_file);
DECREF(ix_file);
if (!self->ix_out) { RETHROW(INCREF(Err_get_error())); }
}
{
CharBuf *dat_file = CB_newf("%o/documents.dat", seg_name);
self->dat_out = Folder_Open_Out(folder, dat_file);
DECREF(dat_file);
if (!self->dat_out) { RETHROW(INCREF(Err_get_error())); }
}
// Go past non-doc #0.
OutStream_Write_I64(self->ix_out, 0);
}
return self->dat_out;
}
void
DocWriter_add_inverted_doc(DocWriter *self, Inverter *inverter,
int32_t doc_id)
{
OutStream *dat_out = S_lazy_init(self);
OutStream *ix_out = self->ix_out;
uint32_t num_stored = 0;
int64_t start = OutStream_Tell(dat_out);
int64_t expected = OutStream_Tell(ix_out) / 8;
// Verify doc id.
if (doc_id != expected)
THROW(ERR, "Expected doc id %i64 but got %i32", expected, doc_id);
// Write the number of stored fields.
Inverter_Iterate(inverter);
while (Inverter_Next(inverter)) {
FieldType *type = Inverter_Get_Type(inverter);
if (FType_Stored(type)) { num_stored++; }
}
OutStream_Write_C32(dat_out, num_stored);
Inverter_Iterate(inverter);
while (Inverter_Next(inverter)) {
// Only store fields marked as "stored".
FieldType *type = Inverter_Get_Type(inverter);
if (FType_Stored(type)) {
CharBuf *field = Inverter_Get_Field_Name(inverter);
Obj *value = Inverter_Get_Value(inverter);
CB_Serialize(field, dat_out);
Obj_Serialize(value, dat_out);
}
}
// Write file pointer.
OutStream_Write_I64(ix_out, start);
}
void
DocWriter_add_segment(DocWriter *self, SegReader *reader,
I32Array *doc_map)
{
int32_t doc_max = SegReader_Doc_Max(reader);
if (doc_max == 0) {
// Bail if the supplied segment is empty.
return;
}
else {
OutStream *const dat_out = S_lazy_init(self);
OutStream *const ix_out = self->ix_out;
ByteBuf *const buffer = BB_new(0);
DefaultDocReader *const doc_reader = (DefaultDocReader*)CERTIFY(
SegReader_Obtain(reader, VTable_Get_Name(DOCREADER)),
DEFAULTDOCREADER);
int32_t i, max;
for (i = 1, max = SegReader_Doc_Max(reader); i <= max; i++) {
if (I32Arr_Get(doc_map, i)) {
int64_t start = OutStream_Tell(dat_out);
char *buf;
size_t size;
// Copy record over.
DefDocReader_Read_Record(doc_reader, buffer, i);
buf = BB_Get_Buf(buffer);
size = BB_Get_Size(buffer);
OutStream_Write_Bytes(dat_out, buf, size);
// Write file pointer.
OutStream_Write_I64(ix_out, start);
}
}
DECREF(buffer);
}
}
void
DocWriter_finish(DocWriter *self)
{
if (self->dat_out) {
// Write one final file pointer, so that we can derive the length of
// the last record.
int64_t end = OutStream_Tell(self->dat_out);
OutStream_Write_I64(self->ix_out, end);
// Close down output streams.
OutStream_Close(self->dat_out);
OutStream_Close(self->ix_out);
Seg_Store_Metadata_Str(self->segment, "documents", 9,
(Obj*)DocWriter_Metadata(self));
}
}
int32_t
DocWriter_format(DocWriter *self)
{
UNUSED_VAR(self);
return DocWriter_current_file_format;
}
/* Copyright 2005-2011 Marvin Humphrey
*
* This program is free software; you can redistribute it and/or modify
* under the same terms as Perl itself.
*/