/* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#define C_LUCY_SORTFIELDWRITER
#include "Lucy/Util/ToolSet.h"
#include <math.h>
#include "Lucy/Index/SortFieldWriter.h"
#include "Lucy/Index/Inverter.h"
#include "Lucy/Index/PolyReader.h"
#include "Lucy/Index/Segment.h"
#include "Lucy/Index/SegReader.h"
#include "Lucy/Index/Snapshot.h"
#include "Lucy/Index/SortCache.h"
#include "Lucy/Index/SortCache/NumericSortCache.h"
#include "Lucy/Index/SortCache/TextSortCache.h"
#include "Lucy/Index/SortReader.h"
#include "Lucy/Index/ZombieKeyedHash.h"
#include "Lucy/Plan/FieldType.h"
#include "Lucy/Plan/Schema.h"
#include "Lucy/Store/Folder.h"
#include "Lucy/Store/InStream.h"
#include "Lucy/Store/OutStream.h"
#include "Lucy/Util/Memory.h"
#include "Lucy/Util/MemoryPool.h"
#include "Lucy/Util/SortUtils.h"
// Prepare to read back a run.
static void
S_flip_run(SortFieldWriter *run, size_t sub_thresh, InStream *ord_in,
InStream *ix_in, InStream *dat_in);
// Write out a sort cache. Returns the number of unique values in the sort
// cache.
static int32_t
S_write_files(SortFieldWriter *self, OutStream *ord_out, OutStream *ix_out,
OutStream *dat_out);
typedef struct lucy_SFWriterElem {
Obj *value;
int32_t doc_id;
} lucy_SFWriterElem;
#define SFWriterElem lucy_SFWriterElem
SortFieldWriter*
SortFieldWriter_new(Schema *schema, Snapshot *snapshot, Segment *segment,
PolyReader *polyreader, const CharBuf *field,
MemoryPool *memory_pool, size_t mem_thresh,
OutStream *temp_ord_out, OutStream *temp_ix_out,
OutStream *temp_dat_out) {
SortFieldWriter *self
= (SortFieldWriter*)VTable_Make_Obj(SORTFIELDWRITER);
return SortFieldWriter_init(self, schema, snapshot, segment, polyreader,
field, memory_pool, mem_thresh, temp_ord_out,
temp_ix_out, temp_dat_out);
}
SortFieldWriter*
SortFieldWriter_init(SortFieldWriter *self, Schema *schema,
Snapshot *snapshot, Segment *segment,
PolyReader *polyreader, const CharBuf *field,
MemoryPool *memory_pool, size_t mem_thresh,
OutStream *temp_ord_out, OutStream *temp_ix_out,
OutStream *temp_dat_out) {
// Init.
SortEx_init((SortExternal*)self, sizeof(SFWriterElem));
self->null_ord = -1;
self->count = 0;
self->ord_start = 0;
self->ord_end = 0;
self->ix_start = 0;
self->ix_end = 0;
self->dat_start = 0;
self->dat_end = 0;
self->run_cardinality = -1;
self->run_max = -1;
self->sort_cache = NULL;
self->doc_map = NULL;
self->sorted_ids = NULL;
self->run_ord = 0;
self->run_tick = 0;
self->ord_width = 0;
// Assign.
self->field = CB_Clone(field);
self->schema = (Schema*)INCREF(schema);
self->snapshot = (Snapshot*)INCREF(snapshot);
self->segment = (Segment*)INCREF(segment);
self->polyreader = (PolyReader*)INCREF(polyreader);
self->mem_pool = (MemoryPool*)INCREF(memory_pool);
self->temp_ord_out = (OutStream*)INCREF(temp_ord_out);
self->temp_ix_out = (OutStream*)INCREF(temp_ix_out);
self->temp_dat_out = (OutStream*)INCREF(temp_dat_out);
self->mem_thresh = mem_thresh;
// Derive.
self->field_num = Seg_Field_Num(segment, field);
FieldType *type = (FieldType*)CERTIFY(
Schema_Fetch_Type(self->schema, field), FIELDTYPE);
self->type = (FieldType*)INCREF(type);
self->prim_id = FType_Primitive_ID(type);
if (self->prim_id == FType_TEXT || self->prim_id == FType_BLOB) {
self->var_width = true;
}
else {
self->var_width = false;
}
self->uniq_vals = (Hash*)ZKHash_new(memory_pool, self->prim_id);
return self;
}
void
SortFieldWriter_clear_cache(SortFieldWriter *self) {
if (self->uniq_vals) {
Hash_Clear(self->uniq_vals);
}
SortFieldWriter_clear_cache_t super_clear_cache
= (SortFieldWriter_clear_cache_t)SUPER_METHOD(
self->vtable, SortFieldWriter, Clear_Cache);
super_clear_cache(self);
}
void
SortFieldWriter_destroy(SortFieldWriter *self) {
DECREF(self->uniq_vals);
self->uniq_vals = NULL;
DECREF(self->field);
DECREF(self->schema);
DECREF(self->snapshot);
DECREF(self->segment);
DECREF(self->polyreader);
DECREF(self->type);
DECREF(self->mem_pool);
DECREF(self->temp_ord_out);
DECREF(self->temp_ix_out);
DECREF(self->temp_dat_out);
DECREF(self->ord_in);
DECREF(self->ix_in);
DECREF(self->dat_in);
DECREF(self->sort_cache);
DECREF(self->doc_map);
FREEMEM(self->sorted_ids);
SUPER_DESTROY(self, SORTFIELDWRITER);
}
int32_t
SortFieldWriter_get_null_ord(SortFieldWriter *self) {
return self->null_ord;
}
int32_t
SortFieldWriter_get_ord_width(SortFieldWriter *self) {
return self->ord_width;
}
static Obj*
S_find_unique_value(Hash *uniq_vals, Obj *val) {
int32_t hash_sum = Obj_Hash_Sum(val);
Obj *uniq_val = Hash_Find_Key(uniq_vals, val, hash_sum);
if (!uniq_val) {
Hash_Store(uniq_vals, val, INCREF(&EMPTY));
uniq_val = Hash_Find_Key(uniq_vals, val, hash_sum);
}
return uniq_val;
}
void
SortFieldWriter_add(SortFieldWriter *self, int32_t doc_id, Obj *value) {
// Uniq-ify the value, and record it for this document.
SFWriterElem elem;
elem.value = S_find_unique_value(self->uniq_vals, value);
elem.doc_id = doc_id;
SortFieldWriter_Feed(self, &elem);
self->count++;
}
void
SortFieldWriter_add_segment(SortFieldWriter *self, SegReader *reader,
I32Array *doc_map, SortCache *sort_cache) {
if (!sort_cache) { return; }
SortFieldWriter *run
= SortFieldWriter_new(self->schema, self->snapshot, self->segment,
self->polyreader, self->field, self->mem_pool,
self->mem_thresh, NULL, NULL, NULL);
run->sort_cache = (SortCache*)INCREF(sort_cache);
run->doc_map = (I32Array*)INCREF(doc_map);
run->run_max = SegReader_Doc_Max(reader);
run->run_cardinality = SortCache_Get_Cardinality(sort_cache);
run->null_ord = SortCache_Get_Null_Ord(sort_cache);
run->run_tick = 1;
SortFieldWriter_Add_Run(self, (SortExternal*)run);
}
static int32_t
S_calc_width(int32_t cardinality) {
if (cardinality <= 0x00000002) { return 1; }
else if (cardinality <= 0x00000004) { return 2; }
else if (cardinality <= 0x0000000F) { return 4; }
else if (cardinality <= 0x000000FF) { return 8; }
else if (cardinality <= 0x0000FFFF) { return 16; }
else { return 32; }
}
static void
S_write_ord(void *ords, int32_t width, int32_t doc_id, int32_t ord) {
switch (width) {
case 1:
if (ord) { NumUtil_u1set(ords, doc_id); }
else { NumUtil_u1clear(ords, doc_id); }
break;
case 2:
NumUtil_u2set(ords, doc_id, ord);
break;
case 4:
NumUtil_u4set(ords, doc_id, ord);
break;
case 8: {
uint8_t *ints = (uint8_t*)ords;
ints[doc_id] = ord;
}
break;
case 16: {
uint8_t *bytes = (uint8_t*)ords;
bytes += doc_id * sizeof(uint16_t);
NumUtil_encode_bigend_u16(ord, &bytes);
}
break;
case 32: {
uint8_t *bytes = (uint8_t*)ords;
bytes += doc_id * sizeof(uint32_t);
NumUtil_encode_bigend_u32(ord, &bytes);
}
break;
default:
THROW(ERR, "Invalid width: %i32", width);
}
}
static void
S_write_val(Obj *val, int8_t prim_id, OutStream *ix_out, OutStream *dat_out,
int64_t dat_start) {
if (val) {
switch (prim_id & FType_PRIMITIVE_ID_MASK) {
case FType_TEXT: {
CharBuf *string = (CharBuf*)val;
int64_t dat_pos = OutStream_Tell(dat_out) - dat_start;
OutStream_Write_I64(ix_out, dat_pos);
OutStream_Write_Bytes(dat_out, (char*)CB_Get_Ptr8(string),
CB_Get_Size(string));
break;
}
case FType_BLOB: {
ByteBuf *byte_buf = (ByteBuf*)val;
int64_t dat_pos = OutStream_Tell(dat_out) - dat_start;
OutStream_Write_I64(ix_out, dat_pos);
OutStream_Write_Bytes(dat_out, BB_Get_Buf(byte_buf),
BB_Get_Size(byte_buf));
break;
}
case FType_INT32: {
Integer32 *i32 = (Integer32*)val;
OutStream_Write_I32(dat_out, Int32_Get_Value(i32));
break;
}
case FType_INT64: {
Integer64 *i64 = (Integer64*)val;
OutStream_Write_I64(dat_out, Int64_Get_Value(i64));
break;
}
case FType_FLOAT64: {
Float64 *float64 = (Float64*)val;
OutStream_Write_F64(dat_out, Float64_Get_Value(float64));
break;
}
case FType_FLOAT32: {
Float32 *float32 = (Float32*)val;
OutStream_Write_F32(dat_out, Float32_Get_Value(float32));
break;
}
default:
THROW(ERR, "Unrecognized primitive id: %i32", (int32_t)prim_id);
}
}
else {
switch (prim_id & FType_PRIMITIVE_ID_MASK) {
case FType_TEXT:
case FType_BLOB: {
int64_t dat_pos = OutStream_Tell(dat_out) - dat_start;
OutStream_Write_I64(ix_out, dat_pos);
}
break;
case FType_INT32:
OutStream_Write_I32(dat_out, 0);
break;
case FType_INT64:
OutStream_Write_I64(dat_out, 0);
break;
case FType_FLOAT64:
OutStream_Write_F64(dat_out, 0.0);
break;
case FType_FLOAT32:
OutStream_Write_F32(dat_out, 0.0f);
break;
default:
THROW(ERR, "Unrecognized primitive id: %i32", (int32_t)prim_id);
}
}
}
int
SortFieldWriter_compare(SortFieldWriter *self, void *va, void *vb) {
SFWriterElem *a = (SFWriterElem*)va;
SFWriterElem *b = (SFWriterElem*)vb;
int32_t comparison
= FType_null_back_compare_values(self->type, a->value, b->value);
if (comparison == 0) { comparison = b->doc_id - a->doc_id; }
return comparison;
}
static int
S_compare_doc_ids_by_ord_rev(void *context, const void *va, const void *vb) {
SortCache *sort_cache = (SortCache*)context;
int32_t a = *(int32_t*)va;
int32_t b = *(int32_t*)vb;
int32_t ord_a = SortCache_Ordinal(sort_cache, a);
int32_t ord_b = SortCache_Ordinal(sort_cache, b);
return ord_a - ord_b;
}
static void
S_lazy_init_sorted_ids(SortFieldWriter *self) {
if (!self->sorted_ids) {
self->sorted_ids
= (int32_t*)MALLOCATE((self->run_max + 1) * sizeof(int32_t));
for (int32_t i = 0, max = self->run_max; i <= max; i++) {
self->sorted_ids[i] = i;
}
Sort_quicksort(self->sorted_ids + 1, self->run_max, sizeof(int32_t),
S_compare_doc_ids_by_ord_rev, self->sort_cache);
}
}
void
SortFieldWriter_flush(SortFieldWriter *self) {
// Don't add a run unless we have data to put in it.
if (SortFieldWriter_Cache_Count(self) == 0) { return; }
OutStream *const temp_ord_out = self->temp_ord_out;
OutStream *const temp_ix_out = self->temp_ix_out;
OutStream *const temp_dat_out = self->temp_dat_out;
SortFieldWriter_Sort_Cache(self);
SortFieldWriter *run
= SortFieldWriter_new(self->schema, self->snapshot, self->segment,
self->polyreader, self->field, self->mem_pool,
self->mem_thresh, NULL, NULL, NULL);
// Record stream starts and align.
run->ord_start = OutStream_Align(temp_ord_out, sizeof(int64_t));
if (self->var_width) {
run->ix_start = OutStream_Align(temp_ix_out, sizeof(int64_t));
}
run->dat_start = OutStream_Align(temp_dat_out, sizeof(int64_t));
// Have the run borrow the array of elems.
run->cache = self->cache;
run->cache_max = self->cache_max;
run->cache_tick = self->cache_tick;
run->cache_cap = self->cache_cap;
// Write files, record stats.
run->run_max = (int32_t)Seg_Get_Count(self->segment);
run->run_cardinality = S_write_files(run, temp_ord_out, temp_ix_out,
temp_dat_out);
// Reclaim the buffer from the run and empty it.
run->cache = NULL;
run->cache_max = 0;
run->cache_tick = 0;
run->cache_cap = 0;
self->cache_tick = self->cache_max;
SortFieldWriter_Clear_Cache(self);
// Record stream ends.
run->ord_end = OutStream_Tell(temp_ord_out);
if (self->var_width) {
run->ix_end = OutStream_Tell(temp_ix_out);
}
run->dat_end = OutStream_Tell(temp_dat_out);
// Add the run to the array.
SortFieldWriter_Add_Run(self, (SortExternal*)run);
}
uint32_t
SortFieldWriter_refill(SortFieldWriter *self) {
if (!self->sort_cache) { return 0; }
// Sanity check, then reset the cache and prepare to start loading items.
uint32_t cache_count = SortFieldWriter_Cache_Count(self);
if (cache_count) {
THROW(ERR, "Refill called but cache contains %u32 items",
cache_count);
}
SortFieldWriter_Clear_Cache(self);
MemPool_Release_All(self->mem_pool);
S_lazy_init_sorted_ids(self);
const int32_t null_ord = self->null_ord;
Hash *const uniq_vals = self->uniq_vals;
I32Array *const doc_map = self->doc_map;
SortCache *const sort_cache = self->sort_cache;
Obj *const blank = SortCache_Make_Blank(sort_cache);
while (self->run_ord < self->run_cardinality
&& MemPool_Get_Consumed(self->mem_pool) < self->mem_thresh
) {
Obj *val = SortCache_Value(sort_cache, self->run_ord, blank);
if (val) {
Hash_Store(uniq_vals, val, INCREF(&EMPTY));
break;
}
self->run_ord++;
}
uint32_t count = 0;
while (self->run_tick <= self->run_max) {
int32_t raw_doc_id = self->sorted_ids[self->run_tick];
int32_t ord = SortCache_Ordinal(sort_cache, raw_doc_id);
if (ord != null_ord) {
int32_t remapped = doc_map
? I32Arr_Get(doc_map, raw_doc_id)
: raw_doc_id;
if (remapped) {
Obj *val = SortCache_Value(sort_cache, ord, blank);
SortFieldWriter_Add(self, remapped, val);
count++;
}
}
else if (ord > self->run_ord) {
break;
}
self->run_tick++;
}
self->run_ord++;
SortFieldWriter_Sort_Cache(self);
if (self->run_ord >= self->run_cardinality) {
DECREF(self->sort_cache);
self->sort_cache = NULL;
}
DECREF(blank);
return count;
}
void
SortFieldWriter_flip(SortFieldWriter *self) {
uint32_t num_items = SortFieldWriter_Cache_Count(self);
uint32_t num_runs = VA_Get_Size(self->runs);
if (self->flipped) { THROW(ERR, "Can't call Flip() twice"); }
self->flipped = true;
// Sanity check.
if (num_runs && num_items) {
THROW(ERR, "Sanity check failed: num_runs: %u32 num_items: %u32",
num_runs, num_items);
}
if (num_items) {
SortFieldWriter_Sort_Cache(self);
}
else if (num_runs) {
Folder *folder = PolyReader_Get_Folder(self->polyreader);
CharBuf *seg_name = Seg_Get_Name(self->segment);
CharBuf *filepath = CB_newf("%o/sort_ord_temp", seg_name);
self->ord_in = Folder_Open_In(folder, filepath);
if (!self->ord_in) { RETHROW(INCREF(Err_get_error())); }
if (self->var_width) {
CB_setf(filepath, "%o/sort_ix_temp", seg_name);
self->ix_in = Folder_Open_In(folder, filepath);
if (!self->ix_in) { RETHROW(INCREF(Err_get_error())); }
}
CB_setf(filepath, "%o/sort_dat_temp", seg_name);
self->dat_in = Folder_Open_In(folder, filepath);
if (!self->dat_in) { RETHROW(INCREF(Err_get_error())); }
DECREF(filepath);
// Assign streams and a slice of mem_thresh.
size_t sub_thresh = self->mem_thresh / num_runs;
if (sub_thresh < 65536) { sub_thresh = 65536; }
for (uint32_t i = 0; i < num_runs; i++) {
SortFieldWriter *run = (SortFieldWriter*)VA_Fetch(self->runs, i);
S_flip_run(run, sub_thresh, self->ord_in, self->ix_in,
self->dat_in);
}
}
self->flipped = true;
}
static int32_t
S_write_files(SortFieldWriter *self, OutStream *ord_out, OutStream *ix_out,
OutStream *dat_out) {
int8_t prim_id = self->prim_id;
int32_t doc_max = (int32_t)Seg_Get_Count(self->segment);
bool_t has_nulls = self->count == doc_max ? false : true;
size_t size = (doc_max + 1) * sizeof(int32_t);
int32_t *ords = (int32_t*)MALLOCATE(size);
int32_t ord = 0;
int64_t dat_start = OutStream_Tell(dat_out);
// Assign -1 as a stand-in for the NULL ord.
for (int32_t i = 0; i <= doc_max; i++) {
ords[i] = -1;
}
// Grab the first item and record its ord. Add a dummy ord for invalid
// doc id 0.
SFWriterElem *elem = (SFWriterElem*)SortFieldWriter_Fetch(self);
ords[elem->doc_id] = ord;
ords[0] = 0;
// Build array of ords, write non-NULL sorted values.
Obj *val = Obj_Clone(elem->value);
Obj *last_val_address = elem->value;
S_write_val(elem->value, prim_id, ix_out, dat_out, dat_start);
while (NULL != (elem = (SFWriterElem*)SortFieldWriter_Fetch(self))) {
if (elem->value != last_val_address) {
int32_t comparison
= FType_Compare_Values(self->type, elem->value, val);
if (comparison != 0) {
ord++;
S_write_val(elem->value, prim_id, ix_out, dat_out, dat_start);
Obj_Mimic(val, elem->value);
}
last_val_address = elem->value;
}
ords[elem->doc_id] = ord;
}
DECREF(val);
// If there are NULL values, write one now and record the NULL ord.
if (has_nulls) {
S_write_val(NULL, prim_id, ix_out, dat_out, dat_start);
ord++;
self->null_ord = ord;
}
int32_t null_ord = self->null_ord;
// Write one extra file pointer so that we can always derive length.
if (self->var_width) {
OutStream_Write_I64(ix_out, OutStream_Tell(dat_out) - dat_start);
}
// Calculate cardinality and ord width.
int32_t cardinality = ord + 1;
self->ord_width = S_calc_width(cardinality);
int32_t ord_width = self->ord_width;
// Write ords.
const double BITS_PER_BYTE = 8.0;
double bytes_per_doc = ord_width / BITS_PER_BYTE;
double byte_count = ceil((doc_max + 1) * bytes_per_doc);
char *compressed_ords
= (char*)CALLOCATE((size_t)byte_count, sizeof(char));
for (int32_t i = 0; i <= doc_max; i++) {
int32_t real_ord = ords[i] == -1 ? null_ord : ords[i];
S_write_ord(compressed_ords, ord_width, i, real_ord);
}
OutStream_Write_Bytes(ord_out, compressed_ords, (size_t)byte_count);
FREEMEM(compressed_ords);
FREEMEM(ords);
return cardinality;
}
int32_t
SortFieldWriter_finish(SortFieldWriter *self) {
// Bail if there's no data.
if (!SortFieldWriter_Peek(self)) { return 0; }
int32_t field_num = self->field_num;
Folder *folder = PolyReader_Get_Folder(self->polyreader);
CharBuf *seg_name = Seg_Get_Name(self->segment);
CharBuf *path = CB_newf("%o/sort-%i32.ord", seg_name, field_num);
// Open streams.
OutStream *ord_out = Folder_Open_Out(folder, path);
if (!ord_out) { RETHROW(INCREF(Err_get_error())); }
OutStream *ix_out = NULL;
if (self->var_width) {
CB_setf(path, "%o/sort-%i32.ix", seg_name, field_num);
ix_out = Folder_Open_Out(folder, path);
if (!ix_out) { RETHROW(INCREF(Err_get_error())); }
}
CB_setf(path, "%o/sort-%i32.dat", seg_name, field_num);
OutStream *dat_out = Folder_Open_Out(folder, path);
if (!dat_out) { RETHROW(INCREF(Err_get_error())); }
DECREF(path);
int32_t cardinality = S_write_files(self, ord_out, ix_out, dat_out);
// Close streams.
OutStream_Close(ord_out);
if (ix_out) { OutStream_Close(ix_out); }
OutStream_Close(dat_out);
DECREF(dat_out);
DECREF(ix_out);
DECREF(ord_out);
return cardinality;
}
static void
S_flip_run(SortFieldWriter *run, size_t sub_thresh, InStream *ord_in,
InStream *ix_in, InStream *dat_in) {
if (run->flipped) { THROW(ERR, "Can't Flip twice"); }
run->flipped = true;
// Get our own MemoryPool, ZombieKeyedHash, and slice of mem_thresh.
DECREF(run->uniq_vals);
DECREF(run->mem_pool);
run->mem_pool = MemPool_new(0);
run->uniq_vals = (Hash*)ZKHash_new(run->mem_pool, run->prim_id);
run->mem_thresh = sub_thresh;
// Done if we already have a SortCache to read from.
if (run->sort_cache) { return; }
// Open the temp files for reading.
CharBuf *seg_name = Seg_Get_Name(run->segment);
CharBuf *alias = CB_newf("%o/sort_ord_temp-%i64-to-%i64", seg_name,
run->ord_start, run->ord_end);
InStream *ord_in_dupe = InStream_Reopen(ord_in, alias, run->ord_start,
run->ord_end - run->ord_start);
InStream *ix_in_dupe = NULL;
if (run->var_width) {
CB_setf(alias, "%o/sort_ix_temp-%i64-to-%i64", seg_name,
run->ix_start, run->ix_end);
ix_in_dupe = InStream_Reopen(ix_in, alias, run->ix_start,
run->ix_end - run->ix_start);
}
CB_setf(alias, "%o/sort_dat_temp-%i64-to-%i64", seg_name,
run->dat_start, run->dat_end);
InStream *dat_in_dupe = InStream_Reopen(dat_in, alias, run->dat_start,
run->dat_end - run->dat_start);
DECREF(alias);
// Get a SortCache.
CharBuf *field = Seg_Field_Name(run->segment, run->field_num);
switch (run->prim_id & FType_PRIMITIVE_ID_MASK) {
case FType_TEXT:
run->sort_cache = (SortCache*)TextSortCache_new(
field, run->type, run->run_cardinality,
run->run_max, run->null_ord,
run->ord_width, ord_in_dupe,
ix_in_dupe, dat_in_dupe);
break;
case FType_INT32:
run->sort_cache = (SortCache*)I32SortCache_new(
field, run->type, run->run_cardinality,
run->run_max, run->null_ord,
run->ord_width, ord_in_dupe,
dat_in_dupe);
break;
case FType_INT64:
run->sort_cache = (SortCache*)I64SortCache_new(
field, run->type, run->run_cardinality,
run->run_max, run->null_ord,
run->ord_width, ord_in_dupe,
dat_in_dupe);
break;
case FType_FLOAT32:
run->sort_cache = (SortCache*)F32SortCache_new(
field, run->type, run->run_cardinality,
run->run_max, run->null_ord,
run->ord_width, ord_in_dupe,
dat_in_dupe);
break;
case FType_FLOAT64:
run->sort_cache = (SortCache*)F64SortCache_new(
field, run->type, run->run_cardinality,
run->run_max, run->null_ord,
run->ord_width, ord_in_dupe,
dat_in_dupe);
break;
default:
THROW(ERR, "No SortCache class for %o", run->type);
}
DECREF(ord_in_dupe);
DECREF(ix_in_dupe);
DECREF(dat_in_dupe);
}