~drizzle-trunk/drizzle/development : revision 1909

52

namespace drizzled

53

{

54

55

56

/* Defines used by filesort and uniques */

57

#define MERGEBUFF 7

58

#define MERGEBUFF2 15

59

60

class BufferCompareContext

61

{

62

public:

63

qsort_cmp2 key_compare;

64

void *key_compare_arg;

65

66

BufferCompareContext() :

67

key_compare(0),

68

key_compare_arg(0)

69

{ }

70

71

};

72

73

class SortParam {

74

public:

75

uint32_t rec_length; /* Length of sorted records */

76

uint32_t sort_length; /* Length of sorted columns */

77

uint32_t ref_length; /* Length of record ref. */

78

uint32_t addon_length; /* Length of added packed fields */

79

uint32_t res_length; /* Length of records in final sorted file/buffer */

80

uint32_t keys; /* Max keys / buffer */

81

ha_rows max_rows,examined_rows;

82

Table *sort_form; /* For quicker make_sortkey */

83

SortField *local_sortorder;

84

SortField *end;

85

sort_addon_field *addon_field; /* Descriptors for companion fields */

86

unsigned char *unique_buff;

87

bool not_killable;

88

char *tmp_buffer;

89

/* The fields below are used only by Unique class */

90

qsort2_cmp compare;

91

BufferCompareContext cmp_context;

92

93

SortParam() :

94

rec_length(0),

95

sort_length(0),

96

ref_length(0),

97

addon_length(0),

98

res_length(0),

99

keys(0),

100

max_rows(0),

101

examined_rows(0),

102

sort_form(0),

103

local_sortorder(0),

104

end(0),

105

addon_field(0),

106

unique_buff(0),

107

not_killable(0),

108

tmp_buffer(0),

109

compare(0)

110

{

111

}

112

113

~SortParam()

114

{

115

if (tmp_buffer)

116

free(tmp_buffer);

117

}

118

119

int write_keys(unsigned char * *sort_keys,

120

uint32_t count,

121

internal::IO_CACHE *buffer_file,

122

internal::IO_CACHE *tempfile);

123

124

void make_sortkey(unsigned char *to,

125

unsigned char *ref_pos);

126

void register_used_fields();

127

bool save_index(unsigned char **sort_keys,

128

uint32_t count,

129

filesort_info *table_sort);

130

131

};

132

55

133

/* functions defined in this file */

56

134

57

135

static char **make_char_array(char **old_pos, register uint32_t fields,

61

139

uint32_t count,

62

140

unsigned char *buf);

63

141

64

static int write_keys(SORTPARAM *param,

65

unsigned char * *sort_keys,

66

uint32_t count,

67

internal::IO_CACHE *buffer_file,

68

internal::IO_CACHE *tempfile);

69

70

static void make_sortkey(SORTPARAM *param,

71

unsigned char *to,

72

unsigned char *ref_pos);

73

static void register_used_fields(SORTPARAM *param);

74

static bool save_index(SORTPARAM *param,

75

unsigned char **sort_keys,

76

uint32_t count,

77

filesort_info *table_sort);

78

142

static uint32_t suffix_length(uint32_t string_length);

79

143

static void unpack_addon_fields(sort_addon_field *addon_field,

80

144

unsigned char *buff);

121

185

122

186

ha_rows FileSort::run(Table *table, SortField *sortorder, uint32_t s_length,

123

187

optimizer::SqlSelect *select, ha_rows max_rows,

124

bool sort_positions, ha_rows *examined_rows)

188

bool sort_positions, ha_rows &examined_rows)

125

189

{

126

int error;

190

int error= 1;

127

191

uint32_t memavl= 0, min_sort_memory;

128

192

uint32_t maxbuffer;

129

193

size_t allocated_sort_memory= 0;

130

buffpek *buffpek_inst;

194

buffpek *buffpek_inst= 0;

131

195

ha_rows records= HA_POS_ERROR;

132

196

unsigned char **sort_keys= 0;

133

internal::IO_CACHE tempfile, buffpek_pointers, *selected_records_file, *outfile;

134

SORTPARAM param;

197

internal::IO_CACHE tempfile;

198

internal::IO_CACHE buffpek_pointers;

199

internal::IO_CACHE *selected_records_file;

200

internal::IO_CACHE *outfile;

201

SortParam param;

135

202

bool multi_byte_charset;

136

203

137

filesort_info table_sort;

204

/*

205

Don't use table->sort in filesort as it is also used by

206

QuickIndexMergeSelect. Work with a copy and put it back at the end

207

when index_merge select has finished with it.

208

*/

209

filesort_info table_sort(table->sort);

210

table->sort.io_cache= NULL;

211

138

212

TableList *tab= table->pos_in_table_list;

139

213

Item_subselect *subselect= tab ? tab->containing_subselect() : 0;

140

214

146

220

*/

147

221

plugin::TransactionalStorageEngine::releaseTemporaryLatches(&getSession());

148

222

149

/*

150

Don't use table->sort in filesort as it is also used by

151

QuickIndexMergeSelect. Work with a copy and put it back at the end

152

when index_merge select has finished with it.

153

*/

154

memcpy(&table_sort, &table->sort, sizeof(filesort_info));

155

table->sort.io_cache= NULL;

156

223

157

224

outfile= table_sort.io_cache;

158

my_b_clear(&tempfile);

159

my_b_clear(&buffpek_pointers);

160

buffpek_inst=0;

161

error= 1;

162

memset(&param, 0, sizeof(param));

225

assert(tempfile.buffer == 0);

226

assert(buffpek_pointers.buffer == 0);

227

163

228

param.sort_length= sortlength(sortorder, s_length, &multi_byte_charset);

164

229

param.ref_length= table->cursor->ref_length;

165

param.addon_field= 0;

166

param.addon_length= 0;

230

167

231

if (!(table->cursor->getEngine()->check_flag(HTON_BIT_FAST_KEY_READ)) && !sort_positions)

168

232

{

169

233

/*

183

247

{

184

248

param.res_length= param.addon_length;

185

249

if (!(table_sort.addon_buf= (unsigned char *) malloc(param.addon_length)))

250

{

186

251

goto err;

252

}

187

253

}

188

254

else

189

255

{

225

291

selected_records_file= 0;

226

292

}

227

293

228

if (multi_byte_charset &&

229

!(param.tmp_buffer= (char*) malloc(param.sort_length)))

294

if (multi_byte_charset && !(param.tmp_buffer= (char*) malloc(param.sort_length)))

295

{

230

296

goto err;

297

}

231

298

232

299

memavl= getSession().variables.sortbuff_size;

233

300

min_sort_memory= max((uint32_t)MIN_SORT_MEMORY, param.sort_length*MERGEBUFF2);

261

328

goto err;

262

329

}

263

330

264

if (open_cached_file(&buffpek_pointers,drizzle_tmpdir.c_str(),TEMP_PREFIX,

265

DISK_BUFFER_SIZE, MYF(MY_WME)))

331

if (open_cached_file(&buffpek_pointers,drizzle_tmpdir.c_str(),TEMP_PREFIX, DISK_BUFFER_SIZE, MYF(MY_WME)))

332

{

266

333

goto err;

334

}

267

335

268

336

param.keys--; /* TODO: check why we do this */

269

337

param.sort_form= table;

277

345

278

346

if (maxbuffer == 0) // The whole set is in memory

279

347

{

280

if (save_index(&param,sort_keys,(uint32_t) records, &table_sort))

348

if (param.save_index(sort_keys,(uint32_t) records, &table_sort))

349

{

281

350

goto err;

351

}

282

352

}

283

353

else

284

354

{

289

359

table_sort.buffpek = 0;

290

360

}

291

361

if (!(table_sort.buffpek=

292

(unsigned char *) read_buffpek_from_file(&buffpek_pointers, maxbuffer,

293

table_sort.buffpek)))

362

(unsigned char *) read_buffpek_from_file(&buffpek_pointers, maxbuffer, table_sort.buffpek)))

363

{

294

364

goto err;

365

}

295

366

buffpek_inst= (buffpek *) table_sort.buffpek;

296

367

table_sort.buffpek_len= maxbuffer;

297

368

close_cached_file(&buffpek_pointers);

298

369

/* Open cached file if it isn't open */

299

if (! my_b_inited(outfile) &&

300

open_cached_file(outfile,drizzle_tmpdir.c_str(),TEMP_PREFIX,READ_RECORD_BUFFER, MYF(MY_WME)))

370

if (! my_b_inited(outfile) && open_cached_file(outfile,drizzle_tmpdir.c_str(),TEMP_PREFIX,READ_RECORD_BUFFER, MYF(MY_WME)))

371

{

301

372

goto err;

373

}

374

302

375

if (reinit_io_cache(outfile,internal::WRITE_CACHE,0L,0,0))

376

{

303

377

goto err;

378

}

304

379

305

380

/*

306

381

Use also the space previously used by string pointers in sort_buffer

310

385

param.rec_length-1);

311

386

maxbuffer--; // Offset from 0

312

387

if (merge_many_buff(&param,(unsigned char*) sort_keys,buffpek_inst,&maxbuffer, &tempfile))

313

goto err;

314

if (flush_io_cache(&tempfile) ||

315

reinit_io_cache(&tempfile,internal::READ_CACHE,0L,0,0))

316

goto err;

388

{

389

goto err;

390

}

391

392

if (flush_io_cache(&tempfile) || reinit_io_cache(&tempfile,internal::READ_CACHE,0L,0,0))

393

{

394

goto err;

395

}

396

317

397

if (merge_index(&param,(unsigned char*) sort_keys,buffpek_inst,maxbuffer,&tempfile, outfile))

398

{

318

399

goto err;

400

}

319

401

}

402

320

403

if (records > param.max_rows)

321

records=param.max_rows;

404

{

405

records= param.max_rows;

406

}

322

407

error =0;

323

408

324

409

err:

325

if (param.tmp_buffer)

326

if (param.tmp_buffer)

327

free(param.tmp_buffer);

328

if (!subselect || !subselect->is_uncacheable())

410

if (not subselect || not subselect->is_uncacheable())

329

411

{

330

412

free(sort_keys);

331

413

table_sort.sort_keys= 0;

333

415

table_sort.buffpek= 0;

334

416

table_sort.buffpek_len= 0;

335

417

}

418

336

419

close_cached_file(&tempfile);

337

420

close_cached_file(&buffpek_pointers);

338

421

if (my_b_inited(outfile))

339

422

{

340

423

if (flush_io_cache(outfile))

424

{

341

425

error=1;

426

}

342

427

{

343

428

internal::my_off_t save_pos=outfile->pos_in_file;

344

429

/* For following reads */

345

430

if (reinit_io_cache(outfile,internal::READ_CACHE,0L,0,0))

431

{

346

432

error=1;

433

}

347

434

outfile->end_of_file=save_pos;

348

435

}

349

436

}

356

443

{

357

444

getSession().status_var.filesort_rows+= (uint32_t) records;

358

445

}

359

*examined_rows= param.examined_rows;

446

examined_rows= param.examined_rows;

360

447

global_sort_buffer.sub(allocated_sort_memory);

361

memcpy(&table->sort, &table_sort, sizeof(filesort_info));

448

table->sort= table_sort;

362

449

DRIZZLE_FILESORT_DONE(error, records);

363

450

return (error ? HA_POS_ERROR : records);

364

451

} /* filesort */

365

452

366

367

void Table::filesort_free_buffers(bool full)

368

{

369

if (sort.record_pointers)

370

{

371

free((unsigned char*) sort.record_pointers);

372

sort.record_pointers=0;

373

}

374

if (full)

375

{

376

if (sort.sort_keys )

377

{

378

if ((unsigned char*) sort.sort_keys)

379

free((unsigned char*) sort.sort_keys);

380

sort.sort_keys= 0;

381

}

382

if (sort.buffpek)

383

{

384

if ((unsigned char*) sort.buffpek)

385

free((unsigned char*) sort.buffpek);

386

sort.buffpek= 0;

387

sort.buffpek_len= 0;

388

}

389

}

390

391

if (sort.addon_buf)

392

{

393

free((char *) sort.addon_buf);

394

free((char *) sort.addon_field);

395

sort.addon_buf=0;

396

sort.addon_field=0;

397

}

398

}

399

400

453

/** Make a array of string pointers. */

401

454

402

455

static char **make_char_array(char **old_pos, register uint32_t fields,

477

530

HA_POS_ERROR on error.

478

531

*/

479

532

480

ha_rows FileSort::find_all_keys(SORTPARAM *param,

533

ha_rows FileSort::find_all_keys(SortParam *param,

481

534

optimizer::SqlSelect *select,

482

535

unsigned char **sort_keys,

483

536

internal::IO_CACHE *buffpek_pointers,

529

582

sort_form->tmp_set.reset();

530

583

/* Temporary set for register_used_fields and register_field_in_read_map */

531

584

sort_form->read_set= &sort_form->tmp_set;

532

register_used_fields(param);

585

param->register_used_fields();

533

586

if (select && select->cond)

534

587

select->cond->walk(&Item::register_field_in_read_map, 1,

535

588

(unsigned char*) sort_form);

588

641

{

589

642

if (idx == param->keys)

590

643

{

591

if (write_keys(param,sort_keys,idx,buffpek_pointers,tempfile))

644

if (param->write_keys(sort_keys, idx, buffpek_pointers, tempfile))

592

645

return(HA_POS_ERROR);

593

646

idx=0;

594

647

indexpos++;

595

648

}

596

make_sortkey(param,sort_keys[idx++],ref_pos);

649

param->make_sortkey(sort_keys[idx++], ref_pos);

597

650

}

598

651

else

652

{

599

653

file->unlock_row();

654

}

655

600

656

/* It does not make sense to read more keys in case of a fatal error */

601

657

if (getSession().is_error())

602

658

break;

628

684

return(HA_POS_ERROR);

629

685

}

630

686

if (indexpos && idx &&

631

write_keys(param,sort_keys,idx,buffpek_pointers,tempfile))

687

param->write_keys(sort_keys,idx,buffpek_pointers,tempfile))

632

688

return(HA_POS_ERROR);

633

689

return(my_b_inited(tempfile) ?

634

690

(ha_rows) (my_b_tell(tempfile)/param->rec_length) :

658

714

1 Error

659

715

*/

660

716

661

static int

662

write_keys(SORTPARAM *param, register unsigned char **sort_keys, uint32_t count,

663

internal::IO_CACHE *buffpek_pointers, internal::IO_CACHE *tempfile)

717

int SortParam::write_keys(register unsigned char **sort_keys, uint32_t count,

718

internal::IO_CACHE *buffpek_pointers, internal::IO_CACHE *tempfile)

664

719

{

665

size_t sort_length, rec_length;

666

unsigned char **end;

667

720

buffpek buffpek;

668

721

669

sort_length= param->sort_length;

670

rec_length= param->rec_length;

671

722

internal::my_string_ptr_sort((unsigned char*) sort_keys, (uint32_t) count, sort_length);

672

723

if (!my_b_inited(tempfile) &&

673

open_cached_file(tempfile, drizzle_tmpdir.c_str(), TEMP_PREFIX, DISK_BUFFER_SIZE,

674

MYF(MY_WME)))

724

open_cached_file(tempfile, drizzle_tmpdir.c_str(), TEMP_PREFIX, DISK_BUFFER_SIZE, MYF(MY_WME)))

675

725

{

676

726

return 1;

677

727

}

682

732

}

683

733

684

734

buffpek.file_pos= my_b_tell(tempfile);

685

if ((ha_rows) count > param->max_rows)

686

count=(uint32_t) param->max_rows;

735

if ((ha_rows) count > max_rows)

736

count=(uint32_t) max_rows;

737

687

738

buffpek.count=(ha_rows) count;

688

for (end=sort_keys+count ; sort_keys != end ; sort_keys++)

739

740

for (unsigned char **ptr= sort_keys + count ; sort_keys != ptr ; sort_keys++)

689

741

{

690

742

if (my_b_write(tempfile, (unsigned char*) *sort_keys, (uint32_t) rec_length))

691

743

{

727

779

728

780

/** Make a sort-key from record. */

729

781

730

static void make_sortkey(register SORTPARAM *param,

731

register unsigned char *to, unsigned char *ref_pos)

782

void SortParam::make_sortkey(register unsigned char *to, unsigned char *ref_pos)

732

783

{

733

784

Field *field;

734

785

SortField *sort_field;

735

786

size_t length;

736

787

737

for (sort_field=param->local_sortorder ;

738

sort_field != param->end ;

788

for (sort_field= local_sortorder ;

789

sort_field != end ;

739

790

sort_field++)

740

791

{

741

792

bool maybe_null=0;

811

862

if ((unsigned char*) from == to)

812

863

{

813

864

set_if_smaller(length,sort_field->length);

814

memcpy(param->tmp_buffer,from,length);

815

from=param->tmp_buffer;

865

memcpy(tmp_buffer,from,length);

866

from= tmp_buffer;

816

867

}

817

868

tmp_length= my_strnxfrm(cs,to,sort_field->length,

818

869

(unsigned char*) from, length);

911

962

to+= sort_field->length;

912

963

}

913

964

914

if (param->addon_field)

965

if (addon_field)

915

966

{

916

967

/*

917

968

Save field values appended to sorted fields.

919

970

In this implementation we use fixed layout for field values -

920

971

the same for all records.

921

972

*/

922

sort_addon_field *addonf= param->addon_field;

973

sort_addon_field *addonf= addon_field;

923

974

unsigned char *nulls= to;

924

975

assert(addonf != 0);

925

976

memset(nulls, 0, addonf->offset);

951

1002

else

952

1003

{

953

1004

/* Save filepos last */

954

memcpy(to, ref_pos, (size_t) param->ref_length);

1005

memcpy(to, ref_pos, (size_t) ref_length);

955

1006

}

956

return;

957

1007

}

958

1008

959

1009

961

1011

Register fields used by sorting in the sorted table's read set

962

1012

*/

963

1013

964

static void register_used_fields(SORTPARAM *param)

1014

void SortParam::register_used_fields()

965

1015

{

966

1016

SortField *sort_field;

967

Table *table=param->sort_form;

1017

Table *table= sort_form;

968

1018

969

for (sort_field= param->local_sortorder ;

970

sort_field != param->end ;

1019

for (sort_field= local_sortorder ;

1020

sort_field != end ;

971

1021

sort_field++)

972

1022

{

973

1023

Field *field;

983

1033

}

984

1034

}

985

1035

986

if (param->addon_field)

1036

if (addon_field)

987

1037

{

988

sort_addon_field *addonf= param->addon_field;

1038

sort_addon_field *addonf= addon_field;

989

1039

Field *field;

990

1040

for ( ; (field= addonf->field) ; addonf++)

991

1041

table->setReadSet(field->field_index);

998

1048

}

999

1049

1000

1050

1001

static bool save_index(SORTPARAM *param, unsigned char **sort_keys, uint32_t count,

1002

filesort_info *table_sort)

1051

bool SortParam::save_index(unsigned char **sort_keys, uint32_t count,

1052

filesort_info *table_sort)

1003

1053

{

1004

uint32_t offset,res_length;

1054

uint32_t offset;

1005

1055

unsigned char *to;

1006

1056

1007

internal::my_string_ptr_sort((unsigned char*) sort_keys, (uint32_t) count, param->sort_length);

1008

res_length= param->res_length;

1009

offset= param->rec_length-res_length;

1010

if ((ha_rows) count > param->max_rows)

1011

count=(uint32_t) param->max_rows;

1057

internal::my_string_ptr_sort((unsigned char*) sort_keys, (uint32_t) count, sort_length);

1058

offset= rec_length - res_length;

1059

if ((ha_rows) count > max_rows)

1060

count=(uint32_t) max_rows;

1012

1061

if (!(to= table_sort->record_pointers=

1013

1062

(unsigned char*) malloc(res_length*count)))

1014

return(1);

1015

for (unsigned char **end= sort_keys+count ; sort_keys != end ; sort_keys++)

1063

return true;

1064

1065

for (unsigned char **end_ptr= sort_keys+count ; sort_keys != end_ptr ; sort_keys++)

1016

1066

{

1017

1067

memcpy(to, *sort_keys+offset, res_length);

1018

1068

to+= res_length;

1019

1069

}

1020

return(0);

1070

return false;

1021

1071

}

1022

1072

1023

1073

1024

1074

/** Merge buffers to make < MERGEBUFF2 buffers. */

1025

1075

1026

int FileSort::merge_many_buff(SORTPARAM *param, unsigned char *sort_buffer,

1076

int FileSort::merge_many_buff(SortParam *param, unsigned char *sort_buffer,

1027

1077

buffpek *buffpek_inst, uint32_t *maxbuffer, internal::IO_CACHE *t_file)

1028

1078

{

1029

1079

internal::IO_CACHE t_file2,*from_file,*to_file,*temp;

1151

1201

other error

1152

1202

*/

1153

1203

1154

int FileSort::merge_buffers(SORTPARAM *param, internal::IO_CACHE *from_file,

1204

int FileSort::merge_buffers(SortParam *param, internal::IO_CACHE *from_file,

1155

1205

internal::IO_CACHE *to_file, unsigned char *sort_buffer,

1156

1206

buffpek *lastbuff, buffpek *Fb, buffpek *Tb,

1157

1207

int flag)

1364

1414

1365

1415

/* Do a merge to output-file (save only positions) */

1366

1416

1367

int FileSort::merge_index(SORTPARAM *param, unsigned char *sort_buffer,

1417

int FileSort::merge_index(SortParam *param, unsigned char *sort_buffer,

1368

1418

buffpek *buffpek_inst, uint32_t maxbuffer,

1369

1419

internal::IO_CACHE *tempfile, internal::IO_CACHE *outfile)

1370

1420

{