Drizzled Public API Documentation

join_cache.cc
Go to the documentation of this file.
1 /* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
3  *
4  * Copyright (C) 2008-2009 Sun Microsystems, Inc.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License as published by
8  * the Free Software Foundation; either version 2 of the License, or
9  * (at your option) any later version.
10  *
11  * This program is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14  * GNU General Public License for more details.
15  *
16  * You should have received a copy of the GNU General Public License
17  * along with this program; if not, write to the Free Software
18  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19  */
20 
30 #include <config.h>
31 
32 #include <drizzled/sql_select.h> /* include join.h */
33 #include <drizzled/field/blob.h>
34 #include <drizzled/drizzled.h>
35 #include <drizzled/internal/my_sys.h>
36 #include <drizzled/table.h>
37 #include <drizzled/session.h>
38 #include <drizzled/system_variables.h>
39 
40 #include <algorithm>
41 
42 using namespace std;
43 
44 namespace drizzled {
45 
46 static uint32_t used_blob_length(CacheField **ptr)
47 {
48  uint32_t length,blob_length;
49  for (length=0 ; *ptr ; ptr++)
50  {
51  (*ptr)->blob_length=blob_length=(*ptr)->blob_field->get_length();
52  length+=blob_length;
53  (*ptr)->str= (*ptr)->blob_field->get_ptr();
54  }
55  return length;
56 }
57 
58 /*****************************************************************************
59  Fill join cache with packed records
60  Records are stored in tab->cache.buffer and last record in
61  last record is stored with pointers to blobs to support very big
62  records
63 ******************************************************************************/
64 int join_init_cache(Session *session, JoinTable *tables, uint32_t table_count)
65 {
66  unsigned int length, blobs;
67  size_t size;
68  CacheField *copy,**blob_ptr;
69  JoinCache *cache;
70  JoinTable *join_tab;
71 
72  cache= &tables[table_count].cache;
73  cache->fields=blobs=0;
74 
75  join_tab= tables;
76  for (unsigned int i= 0; i < table_count ; i++, join_tab++)
77  {
78  if (!join_tab->used_fieldlength) /* Not calced yet */
79  calc_used_field_length(session, join_tab);
80  cache->fields+=join_tab->used_fields;
81  blobs+=join_tab->used_blobs;
82 
83  /* SemiJoinDuplicateElimination: reserve space for rowid */
84  if (join_tab->rowid_keep_flags & JoinTable::KEEP_ROWID)
85  {
86  cache->fields++;
87  join_tab->used_fieldlength += join_tab->table->cursor->ref_length;
88  }
89  }
90  if (!(cache->field=(CacheField*)
91  memory::sql_alloc(sizeof(CacheField)*(cache->fields+table_count*2)+(blobs+1)* sizeof(CacheField*))))
92  {
93  size= cache->end - cache->buff;
94  global_join_buffer.sub(size);
95  free((unsigned char*) cache->buff);
96  cache->buff=0;
97  return 1;
98  }
99  copy=cache->field;
100  blob_ptr=cache->blob_ptr=(CacheField**)
101  (cache->field+cache->fields+table_count*2);
102 
103  length=0;
104  for (unsigned int i= 0 ; i < table_count ; i++)
105  {
106  uint32_t null_fields=0, used_fields;
107  Field **f_ptr,*field;
108  for (f_ptr= tables[i].table->getFields(), used_fields= tables[i].used_fields; used_fields; f_ptr++)
109  {
110  field= *f_ptr;
111  if (field->isReadSet())
112  {
113  used_fields--;
114  length+=field->fill_cache_field(copy);
115  if (copy->blob_field)
116  (*blob_ptr++)=copy;
117  if (field->maybe_null())
118  null_fields++;
119  copy->get_rowid= NULL;
120  copy++;
121  }
122  }
123  /* Copy null bits from table */
124  if (null_fields && tables[i].table->getNullFields())
125  { /* must copy null bits */
126  copy->str= tables[i].table->null_flags;
127  copy->length= tables[i].table->getShare()->null_bytes;
128  copy->strip=0;
129  copy->blob_field=0;
130  copy->get_rowid= NULL;
131  length+=copy->length;
132  copy++;
133  cache->fields++;
134  }
135  /* If outer join table, copy null_row flag */
136  if (tables[i].table->maybe_null)
137  {
138  copy->str= (unsigned char*) &tables[i].table->null_row;
139  copy->length=sizeof(tables[i].table->null_row);
140  copy->strip=0;
141  copy->blob_field=0;
142  copy->get_rowid= NULL;
143  length+=copy->length;
144  copy++;
145  cache->fields++;
146  }
147  /* SemiJoinDuplicateElimination: Allocate space for rowid if needed */
148  if (tables[i].rowid_keep_flags & JoinTable::KEEP_ROWID)
149  {
150  copy->str= tables[i].table->cursor->ref;
151  copy->length= tables[i].table->cursor->ref_length;
152  copy->strip=0;
153  copy->blob_field=0;
154  copy->get_rowid= NULL;
155  if (tables[i].rowid_keep_flags & JoinTable::CALL_POSITION)
156  {
157  /* We will need to call h->position(): */
158  copy->get_rowid= tables[i].table;
159  /* And those after us won't have to: */
160  tables[i].rowid_keep_flags&= ~((int)JoinTable::CALL_POSITION);
161  }
162  copy++;
163  }
164  }
165 
166  cache->length= length+blobs*sizeof(char*);
167  cache->blobs= blobs;
168  *blob_ptr= NULL; /* End sequentel */
169  size= max((size_t) session->variables.join_buff_size, (size_t)cache->length);
170  if (not global_join_buffer.add(size))
171  {
172  my_error(ER_OUT_OF_GLOBAL_JOINMEMORY, MYF(ME_ERROR+ME_WAITTANG));
173  return 1;
174  }
175  cache->buff= (unsigned char*) malloc(size);
176  cache->end= cache->buff+size;
177  cache->reset_cache_write();
178 
179  return 0;
180 }
181 
182 bool JoinCache::store_record_in_cache()
183 {
184  JoinCache *cache= this;
185  unsigned char *local_pos;
186  CacheField *copy,*end_field;
187  bool last_record;
188 
189  local_pos= cache->pos;
190  end_field= cache->field+cache->fields;
191 
192  {
193  uint32_t local_length;
194 
195  local_length= cache->length;
196  if (cache->blobs)
197  {
198  local_length+= used_blob_length(cache->blob_ptr);
199  }
200 
201  if ((last_record= (local_length + cache->length > (size_t) (cache->end - local_pos))))
202  {
203  cache->ptr_record= cache->records;
204  }
205  }
206 
207  /*
208  There is room in cache. Put record there
209  */
210  cache->records++;
211  for (copy= cache->field; copy < end_field; copy++)
212  {
213  if (copy->blob_field)
214  {
215  if (last_record)
216  {
217  copy->blob_field->get_image(local_pos, copy->length+sizeof(char*), copy->blob_field->charset());
218  local_pos+= copy->length+sizeof(char*);
219  }
220  else
221  {
222  copy->blob_field->get_image(local_pos, copy->length, // blob length
223  copy->blob_field->charset());
224  memcpy(local_pos + copy->length,copy->str,copy->blob_length); // Blob data
225  local_pos+= copy->length+copy->blob_length;
226  }
227  }
228  else
229  {
230  // SemiJoinDuplicateElimination: Get the rowid into table->ref:
231  if (copy->get_rowid)
232  copy->get_rowid->cursor->position(copy->get_rowid->getInsertRecord());
233 
234  if (copy->strip)
235  {
236  unsigned char *str, *local_end;
237  for (str= copy->str,local_end= str+copy->length; local_end > str && local_end[-1] == ' '; local_end--) {}
238 
239  uint32_t local_length= (uint32_t) (local_end - str);
240  memcpy(local_pos+2, str, local_length);
241  int2store(local_pos, local_length);
242  local_pos+= local_length+2;
243  }
244  else
245  {
246  memcpy(local_pos, copy->str, copy->length);
247  local_pos+= copy->length;
248  }
249  }
250  }
251  cache->pos= local_pos;
252  return last_record || (size_t) (cache->end - local_pos) < cache->length;
253 }
254 
255 void JoinCache::reset_cache_read()
256 {
257  record_nr= 0;
258  pos= buff;
259 }
260 
261 void JoinCache::reset_cache_write()
262 {
263  reset_cache_read();
264  records= 0;
265  ptr_record= UINT32_MAX;
266 }
267 
272 } /* namespace drizzled */
void calc_used_field_length(Session *, JoinTable *join_tab)
Definition: sql_select.cc:883
TODO: Rename this file - func.h is stupid.