1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge

Browse Source
This commit is contained in:
patg@krsna.patg.net 2005-02-20 14:20:05 -08:00
commit e6352ee568
7 changed files with 522 additions and 389 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/mysql.h
View File

@ -499,6 +499,8 @@ MYSQL_FIELD_OFFSET STDCALL mysql_field_seek(MYSQL_RES *result,
MYSQL_FIELD_OFFSET offset);
MYSQL_ROW STDCALL mysql_fetch_row(MYSQL_RES *result);
unsigned long * STDCALL mysql_fetch_lengths(MYSQL_RES *result);
void STDCALL cli_fetch_lengths(unsigned long *to, MYSQL_ROW column,
unsigned int field_count);
MYSQL_FIELD * STDCALL mysql_fetch_field(MYSQL_RES *result);
MYSQL_RES * STDCALL mysql_list_fields(MYSQL *mysql, const char *table,
const char *wild);

26
mysql-test/r/federated.result
View File

@ -644,14 +644,19 @@ select * from federated.t1 where fileguts = 'jimbob';
id code fileguts creation_date entered_time
3 DEUEUEUEUEUEUEUEUEU jimbob 2004-04-04 04:04:04 2004-04-04 04:04:04
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id));
CREATE TABLE federated.t1 (a BLOB);
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id) ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
insert into federated.t1 (name, country_id, other) values ('Kumar', 1, 11111);
insert into federated.t1 (name, country_id, other) values ('Lenz', 2, 22222);
insert into federated.t1 (name, country_id, other) values ('Marizio', 3, 33333);
insert into federated.t1 (name, country_id, other) values ('Monty', 4, 33333);
insert into federated.t1 (name, country_id, other) values ('Sanja', 5, 33333);
CREATE TABLE federated.t1 (a BLOB) ENGINE="FEDERATED" COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
INSERT INTO federated.t1 VALUES (0x00);
INSERT INTO federated.t1 VALUES (0x0001);
INSERT INTO federated.t1 VALUES (0x0100);
SELECT HEX(a) FROM federated.t1;
HEX(a)
00
0001
0100
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id));
drop table if exists federated.countries;
Warnings:
Note 1051 Unknown table 'countries'
@ -661,6 +666,13 @@ insert into federated.countries (country) values ('Germany');
insert into federated.countries (country) values ('Italy');
insert into federated.countries (country) values ('Finland');
insert into federated.countries (country) values ('Ukraine');
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id) ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
insert into federated.t1 (name, country_id, other) values ('Kumar', 1, 11111);
insert into federated.t1 (name, country_id, other) values ('Lenz', 2, 22222);
insert into federated.t1 (name, country_id, other) values ('Marizio', 3, 33333);
insert into federated.t1 (name, country_id, other) values ('Monty', 4, 33333);
insert into federated.t1 (name, country_id, other) values ('Sanja', 5, 33333);
select federated.t1.*, federated.countries.country from federated.t1 left join federated.countries on federated.t1.country_id = federated.countries.id;
id country_id name other country
1 1 Kumar 11111 India

33
mysql-test/t/federated.test
View File

@ -522,8 +522,22 @@ insert into federated.t1 (code, fileguts, creation_date) values ('ASDFWERQWETWET
insert into federated.t1 (code, fileguts, creation_date) values ('DEUEUEUEUEUEUEUEUEU', '*()w*09*$()*#)(*09*^90*d)(*s()d8g)(s*ned)(*)(s*d)(*hn(d*)(*sbn)D((#$*(#*%%&#&^$#&#&#&#&^&#*&*#$*&^*(&#(&Q*&&(*!&!(*&*(#&*(%&#<S-F8>*<S-F8><S-F8><S-F8>#<S-F8>#<S-F8>#<S-F8>[[', '2004-04-04 04:04:04');
insert into federated.t1 (code, fileguts, creation_date) values ('DEUEUEUEUEUEUEUEUEU', 'jimbob', '2004-04-04 04:04:04');
select * from federated.t1;
select * from federated.t1 where fileguts = 'jimbob';
# test blob indexes
select * from federated.t1 where fileguts = 'jimbob';
# test blob with binary
connection slave;
drop table if exists federated.t1;
CREATE TABLE federated.t1 (a BLOB);
connection master;
drop table if exists federated.t1;
CREATE TABLE federated.t1 (a BLOB) ENGINE="FEDERATED" COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
INSERT INTO federated.t1 VALUES (0x00);
INSERT INTO federated.t1 VALUES (0x0001);
INSERT INTO federated.t1 VALUES (0x0100);
SELECT HEX(a) FROM federated.t1;
# TODO
#
@ -559,15 +573,6 @@ drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id));
connection master;
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id) ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
insert into federated.t1 (name, country_id, other) values ('Kumar', 1, 11111);
insert into federated.t1 (name, country_id, other) values ('Lenz', 2, 22222);
insert into federated.t1 (name, country_id, other) values ('Marizio', 3, 33333);
insert into federated.t1 (name, country_id, other) values ('Monty', 4, 33333);
insert into federated.t1 (name, country_id, other) values ('Sanja', 5, 33333);
drop table if exists federated.countries;
CREATE TABLE federated.countries ( `id` int(20) NOT NULL auto_increment, `country` varchar(32), primary key (id));
insert into federated.countries (country) values ('India');
@ -576,6 +581,14 @@ insert into federated.countries (country) values ('Italy');
insert into federated.countries (country) values ('Finland');
insert into federated.countries (country) values ('Ukraine');
drop table if exists federated.t1;
CREATE TABLE federated.t1 ( `id` int(20) NOT NULL auto_increment, `country_id` int(20) NOT NULL DEFAULT 0, `name` varchar(32), `other` varchar(20), PRIMARY KEY (`id`), key (country_id) ) ENGINE="FEDERATED" DEFAULT CHARSET=latin1 COMMENT='mysql://root@127.0.0.1:9308/federated/t1';
insert into federated.t1 (name, country_id, other) values ('Kumar', 1, 11111);
insert into federated.t1 (name, country_id, other) values ('Lenz', 2, 22222);
insert into federated.t1 (name, country_id, other) values ('Marizio', 3, 33333);
insert into federated.t1 (name, country_id, other) values ('Monty', 4, 33333);
insert into federated.t1 (name, country_id, other) values ('Sanja', 5, 33333);
select federated.t1.*, federated.countries.country from federated.t1 left join federated.countries on federated.t1.country_id = federated.countries.id;
drop table federated.countries;

2
sql-common/client.c
View File

@ -1121,7 +1121,7 @@ void mysql_read_default_options(struct st_mysql_options *options,
else the lengths are calculated from the offset between pointers.
**************************************************************************/
static void cli_fetch_lengths(ulong *to, MYSQL_ROW column,
void cli_fetch_lengths(ulong *to, MYSQL_ROW column,
unsigned int field_count)
{
ulong *prev_length;

434
sql/ha_federated.cc
View File

@ -355,9 +355,12 @@
#include "ha_federated.h"
#define MAX_REMOTE_SIZE IO_SIZE
/* Variables for federated share methods */
static HASH federated_open_tables; // Hash used to track open tables
pthread_mutex_t federated_mutex; // This is the mutex we use to init the hash
static int federated_init= 0; // Variable for checking the init state of hash
static HASH federated_open_tables; // Hash used to track open
// tables
pthread_mutex_t federated_mutex; // This is the mutex we use to
// init the hash
static int federated_init= 0; // Variable for checking the
// init state of hash
/*
Function we use in the creation of our hash to get key.
@ -369,6 +372,49 @@ static byte* federated_get_key(FEDERATED_SHARE *share,uint *length,
return (byte*) share->table_name;
}
/*
Initialize the federated handler.
SYNOPSIS
federated_db_init()
void
RETURN
FALSE OK
TRUE Error
*/
bool federated_db_init()
{
federated_init= 1;
VOID(pthread_mutex_init(&federated_mutex, MY_MUTEX_INIT_FAST));
return (hash_init(&federated_open_tables, system_charset_info, 32, 0, 0,
(hash_get_key) federated_get_key, 0, 0));
}
/*
Release the federated handler.
SYNOPSIS
federated_db_end()
void
RETURN
FALSE OK
*/
bool federated_db_end()
{
if (federated_init)
{
hash_free(&federated_open_tables);
VOID(pthread_mutex_destroy(&federated_mutex));
}
federated_init= 0;
return FALSE;
}
/*
Parse connection info from table->s->comment
@ -376,6 +422,7 @@ static byte* federated_get_key(FEDERATED_SHARE *share,uint *length,
parse_url()
share pointer to FEDERATED share
table pointer to current TABLE class
table_create_flag determines what error to throw
DESCRIPTION
populates the share with information about the connection
@ -401,31 +448,28 @@ scheme://username:password@hostname/database/table
RETURN VALUE
0 success
-1 failure, wrong string format
1 failure, wrong string format
*/
static int parse_url(FEDERATED_SHARE *share, TABLE *table, uint table_create_flag)
static int parse_url(FEDERATED_SHARE *share, TABLE *table,
uint table_create_flag)
{
DBUG_ENTER("ha_federated::parse_url");
// This either get set or will remain the same.
share->port= 0;
uint error_num= table_create_flag ? ER_CANT_CREATE_TABLE : ER_CONNECT_TO_MASTER ;
uint error_num= table_create_flag ? ER_CANT_CREATE_TABLE :
ER_CONNECT_TO_MASTER;
share->scheme= my_strdup(table->s->comment, MYF(0));
if ((share->username= strstr(share->scheme, "://")))
{
share->scheme[share->username - share->scheme]= '\0';
if (strcmp(share->scheme, "mysql") != 0)
{
DBUG_PRINT("ha_federated::parse_url",
("The federated handler currently only supports connecting\
to a MySQL database!!!\n"));
my_error(error_num, MYF(0),
"ERROR: federated handler only supports remote 'mysql://' database");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
share->username+= 3;
@ -439,14 +483,14 @@ static int parse_url(FEDERATED_SHARE *share, TABLE *table, uint table_create_fla
share->username[share->password - share->username]= '\0';
share->password++;
share->username= share->username;
// make sure there isn't an extra / or @
/*
make sure there isn't an extra / or @
*/
if ((strchr(share->password, '/') || strchr(share->hostname, '@')))
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
/*
Found that if the string is:
@ -459,14 +503,14 @@ Then password is a null string, so set to NULL
else
share->username= share->username;
// make sure there isn't an extra / or @
/*
make sure there isn't an extra / or @
*/
if ((strchr(share->username, '/')) || (strchr(share->hostname, '@')))
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
if ((share->database= strchr(share->hostname, '/')))
@ -491,55 +535,56 @@ Then password is a null string, so set to NULL
}
else
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
}
else
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
// make sure there's not an extra /
/*
make sure there's not an extra /
*/
if ((strchr(share->table_base_name, '/')))
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
if (share->hostname[0] == '\0')
share->hostname= NULL;
if (!share->port)
{
if (strcmp(share->hostname, "localhost") == 0)
share->socket= my_strdup("/tmp/mysql.sock", MYF(0));
else
share->port= 3306;
}
DBUG_PRINT("ha_federated::parse_url",
("scheme %s username %s password %s \
hostname %s port %d database %s tablename %s\n",
share->scheme, share->username, share->password, share->hostname,
share->port, share->database, share->table_base_name));
share->scheme, share->username, share->password,
share->hostname, share->port, share->database,
share->table_base_name));
}
else
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
}
else
{
DBUG_PRINT("ha_federated::parse_url",
("this connection string is not in the correct format!!!\n"));
my_error(error_num, MYF(0),
"this connection string is not in the correct format!!!\n");
DBUG_RETURN(-1);
DBUG_RETURN(1);
}
DBUG_RETURN(0);
}
@ -564,11 +609,16 @@ Then password is a null string, so set to NULL
*/
uint ha_federated::convert_row_to_internal_format(byte *record, MYSQL_ROW row)
{
unsigned long len;
int x= 0;
ulong *lengths;
uint num_fields;
uint x= 0;
DBUG_ENTER("ha_federated::convert_row_to_internal_format");
// Question this
num_fields= mysql_num_fields(result);
lengths= (ulong*) my_malloc(num_fields * sizeof(ulong), MYF(0));
cli_fetch_lengths((ulong*) lengths, row, num_fields);
memset(record, 0, table->s->null_bytes);
for (Field **field= table->field; *field; field++, x++)
@ -576,12 +626,10 @@ uint ha_federated::convert_row_to_internal_format(byte *record, MYSQL_ROW row)
if (!row[x])
(*field)->set_null();
else
/*
changed system_charset_info to default_charset_info because
testing revealed that german text was not being retrieved properly
*/
(*field)->store(row[x], strlen(row[x]), &my_charset_bin);
(*field)->store(row[x], lengths[x], &my_charset_bin);
}
my_free((gptr) lengths, MYF(0));
lengths= 0;
DBUG_RETURN(0);
}
@ -600,13 +648,13 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
Field *field= key_part->field;
needs_quotes= field->needs_quotes();
//bool needs_quotes= type_quote(field->type());
DBUG_PRINT("ha_federated::create_where_from_key", ("key name %s type %d", field->field_name, field->type()));
DBUG_PRINT("ha_federated::create_where_from_key",
("key name %s type %d", field->field_name, field->type()));
uint length= key_part->length;
if (second_loop++ && to->append(" AND ", 5))
DBUG_RETURN(1);
if (to->append('`') || to->append(field->field_name) ||
to->append("` ",2))
if (to->append('`') || to->append(field->field_name) || to->append("` ", 2))
DBUG_RETURN(1); // Out of memory
if (key_part->null_bit)
@ -614,8 +662,10 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
if (*key++)
{
if (to->append("IS NULL", 7))
DBUG_PRINT("ha_federated::create_where_from_key", ("NULL type %s", to->c_ptr_quick()));
DBUG_RETURN(1);
DBUG_PRINT("ha_federated::create_where_from_key",
("NULL type %s", to->c_ptr_quick()));
key_length-= key_part->store_length;
key+= key_part->store_length - 1;
continue;
@ -644,7 +694,8 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
if (to->append(buff, (uint)(ptr - buff)))
DBUG_RETURN(1);
DBUG_PRINT("ha_federated::create_where_from_key", ("bit type %s", to->c_ptr_quick()));
DBUG_PRINT("ha_federated::create_where_from_key",
("bit type %s", to->c_ptr_quick()));
key_length-= length;
continue;
}
@ -658,7 +709,8 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
if (append_escaped(to, &tmp))
DBUG_RETURN(1);
DBUG_PRINT("ha_federated::create_where_from_key", ("blob type %s", to->c_ptr_quick()));
DBUG_PRINT("ha_federated::create_where_from_key",
("blob type %s", to->c_ptr_quick()));
length= key_part->length;
}
else if (key_part->key_part_flag & HA_VAR_LENGTH_PART)
@ -669,11 +721,13 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
if (append_escaped(to, &tmp))
DBUG_RETURN(1);
DBUG_PRINT("ha_federated::create_where_from_key", ("varchar type %s", to->c_ptr_quick()));
DBUG_PRINT("ha_federated::create_where_from_key",
("varchar type %s", to->c_ptr_quick()));
}
else
{
DBUG_PRINT("ha_federated::create_where_from_key", ("else block, unknown type so far"));
DBUG_PRINT("ha_federated::create_where_from_key",
("else block, unknown type so far"));
char buff[MAX_FIELD_WIDTH];
String str(buff, sizeof(buff), field->charset()), *res;
@ -684,14 +738,16 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
DBUG_RETURN(1);
res= field->val_str(&str, (char*) (key));
DBUG_PRINT("ha_federated::create_where_from_key", ("else block, string type", to->c_ptr_quick()));
DBUG_PRINT("ha_federated::create_where_from_key",
("else block, string type", to->c_ptr_quick()));
}
else if (to->append(res->ptr(), res->length()))
DBUG_RETURN(1);
}
if (needs_quotes && to->append("'"))
DBUG_RETURN(1);
DBUG_PRINT("ha_federated::create_where_from_key", ("final value for 'to' %s", to->c_ptr_quick()));
DBUG_PRINT("ha_federated::create_where_from_key",
("final value for 'to' %s", to->c_ptr_quick()));
key+= length;
key_length-= length;
DBUG_RETURN(0);
@ -699,32 +755,6 @@ bool ha_federated::create_where_from_key(String *to, KEY *key_info,
DBUG_RETURN(1);
}
int load_conn_info(FEDERATED_SHARE *share, TABLE *table)
{
DBUG_ENTER("ha_federated::load_conn_info");
int retcode;
retcode= parse_url(share, table, 0);
if (retcode < 0)
{
DBUG_PRINT("ha_federated::load_conn_info",
("retcode %d, setting defaults", retcode));
/* sanity checks to make sure all needed pieces are present */
if (!share->port)
{
if (strcmp(share->hostname, "localhost") == 0)
share->socket= my_strdup("/tmp/mysql.sock",MYF(0));
else
share->port= 3306;
}
}
DBUG_PRINT("ha_federated::load_conn_info",
("returned from retcode %d", retcode));
DBUG_RETURN(retcode);
}
/*
Example of simple lock controls. The "share" it creates is structure we will
pass to each federated handler. Do you have to have one of these? Well, you
@ -740,8 +770,10 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
uint table_name_length, table_base_name_length;
char *tmp_table_name, *tmp_table_base_name, *table_base_name, *select_query;
// share->table_name has the file location - we want the actual table's
// name!
/*
share->table_name has the file location - we want the actual table's
name!
*/
table_base_name= (char*) table->s->table_name;
DBUG_PRINT("ha_federated::get_share", ("table_name %s", table_base_name));
/*
@ -750,19 +782,6 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
do this. Since you will not want to do this, this is probably the next
best method.
*/
if (!federated_init)
{
/* Hijack a mutex for init'ing the storage engine */
pthread_mutex_lock(&LOCK_mysql_create_db);
if (!federated_init)
{
federated_init++;
VOID(pthread_mutex_init(&federated_mutex,MY_MUTEX_INIT_FAST));
(void) hash_init(&federated_open_tables,system_charset_info,32,0,0,
(hash_get_key) federated_get_key,0,0);
}
pthread_mutex_unlock(&LOCK_mysql_create_db);
}
pthread_mutex_lock(&federated_mutex);
table_name_length= (uint) strlen(table_name);
table_base_name_length= (uint) strlen(table_base_name);
@ -780,14 +799,15 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
&share, sizeof(*share),
&tmp_table_name, table_name_length + 1,
&tmp_table_base_name, table_base_name_length + 1,
&select_query, query.length()+1,
NullS)))
&select_query, query.length() + 1, NullS)))
{
pthread_mutex_unlock(&federated_mutex);
return NULL;
}
load_conn_info(share, table);
if (parse_url(share, table, 0))
goto error;
share->use_count= 0;
share->table_name_length= table_name_length;
share->table_name= tmp_table_name;
@ -796,8 +816,9 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
share->select_query= select_query;
strmov(share->table_name, table_name);
strmov(share->table_base_name, table_base_name);
strmov(share->select_query,query.c_ptr_quick());
DBUG_PRINT("ha_federated::get_share",("share->select_query %s", share->select_query));
strmov(share->select_query, query.ptr());
DBUG_PRINT("ha_federated::get_share",
("share->select_query %s", share->select_query));
if (my_hash_insert(&federated_open_tables, (byte*) share))
goto error;
thr_lock_init(&share->lock);
@ -808,11 +829,11 @@ static FEDERATED_SHARE *get_share(const char *table_name, TABLE *table)
return share;
error2:
thr_lock_delete(&share->lock);
pthread_mutex_destroy(&share->mutex);
error:
pthread_mutex_unlock(&federated_mutex);
if (share->scheme)
my_free((gptr) share->scheme, MYF(0));
VOID(pthread_mutex_destroy(&share->mutex));
my_free((gptr) share, MYF(0));
return NULL;
@ -827,11 +848,15 @@ error:
static int free_share(FEDERATED_SHARE *share)
{
pthread_mutex_lock(&federated_mutex);
if (!--share->use_count)
{
if (share->scheme)
my_free((gptr) share->scheme, MYF(0));
hash_delete(&federated_open_tables, (byte*) share);
thr_lock_delete(&share->lock);
pthread_mutex_destroy(&share->mutex);
VOID(pthread_mutex_destroy(&share->mutex));
my_free((gptr) share, MYF(0));
}
pthread_mutex_unlock(&federated_mutex);
@ -847,7 +872,13 @@ static int free_share(FEDERATED_SHARE *share)
in handler.cc.
*/
const char **ha_federated::bas_ext() const
{ static const char *ext[]= { NullS }; return ext; }
{
static const char *ext[]=
{
NullS
};
return ext;
}
/*
@ -882,8 +913,7 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
share->password,
share->database,
share->port,
NULL,
0))
share->socket, 0))
{
my_error(ER_CONNECT_TO_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_CONNECT_TO_MASTER);
@ -905,6 +935,15 @@ int ha_federated::open(const char *name, int mode, uint test_if_locked)
int ha_federated::close(void)
{
DBUG_ENTER("ha_federated::close");
/* free the result set */
if (result)
{
DBUG_PRINT("ha_federated::close",
("mysql_free_result result at address %lx", result));
mysql_free_result(result);
result= 0;
}
/* Disconnect from mysql */
mysql_close(mysql);
DBUG_RETURN(free_share(share));
@ -929,10 +968,9 @@ int ha_federated::close(void)
1 if NULL
0 otherwise
*/
inline uint field_in_record_is_null (
TABLE* table, /* in: MySQL table object */
Field* field, /* in: MySQL field object */
char* record) /* in: a row in MySQL format */
inline uint field_in_record_is_null(TABLE *table,
Field *field,
char *record)
{
int null_offset;
DBUG_ENTER("ha_federated::field_in_record_is_null");
@ -963,31 +1001,29 @@ inline uint field_in_record_is_null (
*/
int ha_federated::write_row(byte *buf)
{
int x= 0, num_fields= 0;
uint x= 0, num_fields= 0;
Field **field;
ulong current_query_id= 1;
ulong tmp_query_id= 1;
int all_fields_have_same_query_id= 1;
uint all_fields_have_same_query_id= 1;
char insert_buffer[IO_SIZE];
char values_buffer[IO_SIZE], insert_field_value_buffer[IO_SIZE];
// The main insert query string
/* The main insert query string */
String insert_string(insert_buffer, sizeof(insert_buffer), &my_charset_bin);
insert_string.length(0);
// The string containing the values to be added to the insert
/* The string containing the values to be added to the insert */
String values_string(values_buffer, sizeof(values_buffer), &my_charset_bin);
values_string.length(0);
// The actual value of the field, to be added to the values_string
/* The actual value of the field, to be added to the values_string */
String insert_field_value_string(insert_field_value_buffer,
sizeof(insert_field_value_buffer), &my_charset_bin);
sizeof(insert_field_value_buffer),
&my_charset_bin);
insert_field_value_string.length(0);
DBUG_ENTER("ha_federated::write_row");
/*
I want to use this and the next line, but the repository needs to be
updated to do so
*/
statistic_increment(table->in_use->status_var.ha_write_count, &LOCK_status);
if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_INSERT)
table->timestamp_field->set_time();
@ -1001,10 +1037,10 @@ int ha_federated::write_row(byte * buf)
DBUG_PRINT("ha_federated::write_row", ("current query id %d",
current_query_id));
// start off our string
/* start off our string */
insert_string.append("INSERT INTO ");
insert_string.append(share->table_base_name);
// start both our field and field values strings
/* start both our field and field values strings */
insert_string.append(" (");
values_string.append(" VALUES (");
@ -1046,18 +1082,17 @@ int ha_federated::write_row(byte * buf)
("current query id %d field is not null query ID %d",
current_query_id, (*field)->query_id));
(*field)->val_str(&insert_field_value_string);
/* quote these fields if they require it */
(*field)->quote_data(&insert_field_value_string);
}
// append the field name
/* append the field name */
insert_string.append((*field)->field_name);
// quote these fields if they require it
(*field)->quote_data(&insert_field_value_string);
// append the value
/* append the value */
values_string.append(insert_field_value_string);
insert_field_value_string.length(0);
// append commas between both fields and fieldnames
/* append commas between both fields and fieldnames */
insert_string.append(',');
values_string.append(',');
DBUG_PRINT("ha_federated::write_row",
@ -1076,31 +1111,28 @@ int ha_federated::write_row(byte * buf)
*/
insert_string.chop();
/*
if there were no fields, we don't want to add a closing paren
AND, we don't want to chop off the last char '('
insert will be "INSERT INTO t1 VALUES ();"
*/
DBUG_PRINT("ha_federated::write_row",("x %d num fields %d",
x, num_fields));
DBUG_PRINT("ha_federated::write_row", ("x %d num fields %d", x, num_fields));
if (num_fields > 0)
{
// chops off leading commas
/* chops off leading commas */
values_string.chop();
insert_string.append(')');
}
// we always want to append this, even if there aren't any fields
/* we always want to append this, even if there aren't any fields */
values_string.append(')');
// add the values
/* add the values */
insert_string.append(values_string);
DBUG_PRINT("ha_federated::write_row", ("insert query %s",
insert_string.c_ptr_quick()));
if (mysql_real_query(mysql, insert_string.c_ptr_quick(),
insert_string.length()))
if (mysql_real_query(mysql, insert_string.ptr(), insert_string.length()))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
@ -1125,27 +1157,28 @@ int ha_federated::write_row(byte * buf)
Called from sql_select.cc, sql_acl.cc, sql_update.cc, and sql_insert.cc.
*/
int ha_federated::update_row(
const byte * old_data,
byte * new_data
)
int ha_federated::update_row(const byte *old_data, byte *new_data)
{
int x= 0;
uint x= 0;
uint has_a_primary_key= 0;
int primary_key_field_num;
uint primary_key_field_num;
char old_field_value_buffer[IO_SIZE], new_field_value_buffer[IO_SIZE];
char update_buffer[IO_SIZE], where_buffer[IO_SIZE];
// stores the value to be replaced of the field were are updating
String old_field_value(old_field_value_buffer, sizeof(old_field_value_buffer), &my_charset_bin);
/*
stores the value to be replaced of the field were are updating
*/
String old_field_value(old_field_value_buffer, sizeof(old_field_value_buffer),
&my_charset_bin);
old_field_value.length(0);
// stores the new value of the field
String new_field_value(new_field_value_buffer, sizeof(new_field_value_buffer), &my_charset_bin);
/* stores the new value of the field */
String new_field_value(new_field_value_buffer, sizeof(new_field_value_buffer),
&my_charset_bin);
new_field_value.length(0);
// stores the update query
/* stores the update query */
String update_string(update_buffer, sizeof(update_buffer), &my_charset_bin);
update_string.length(0);
// stores the WHERE clause
/* stores the WHERE clause */
String where_string(where_buffer, sizeof(where_buffer), &my_charset_bin);
where_string.length(0);
@ -1205,8 +1238,7 @@ int ha_federated::update_row(
if (x == primary_key_field_num)
where_string.append("=");
}
else
if (! field_in_record_is_null(table, *field, (char*) old_data))
else if (!field_in_record_is_null(table, *field, (char*) old_data))
where_string.append("=");
}
@ -1235,7 +1267,7 @@ int ha_federated::update_row(
update_string.append(new_field_value);
new_field_value.length(0);
if ((uint) x+1 < table->s->fields)
if (x + 1 < table->s->fields)
{
update_string.append(", ");
if (!has_a_primary_key)
@ -1244,14 +1276,13 @@ int ha_federated::update_row(
old_field_value.length(0);
}
update_string.append(" WHERE ");
update_string.append(where_string.c_ptr_quick());
update_string.append(where_string);
if (! has_a_primary_key)
update_string.append(" LIMIT 1");
DBUG_PRINT("ha_federated::update_row", ("Final update query: %s",
update_string.c_ptr_quick()));
if (mysql_real_query(mysql, update_string.c_ptr_quick(),
update_string.length()))
if (mysql_real_query(mysql, update_string.ptr(), update_string.length()))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
@ -1277,7 +1308,7 @@ int ha_federated::update_row(
*/
int ha_federated::delete_row(const byte *buf)
{
int x= 0;
uint x= 0;
char delete_buffer[IO_SIZE];
char data_buffer[IO_SIZE];
@ -1311,15 +1342,14 @@ int ha_federated::delete_row(const byte * buf)
delete_string.append(data_string);
data_string.length(0);
if ((uint) x+1 < table->s->fields)
if (x + 1 < table->s->fields)
delete_string.append(" AND ");
}
delete_string.append(" LIMIT 1");
DBUG_PRINT("ha_federated::delete_row",
("Delete sql: %s", delete_string.c_ptr_quick()));
if ( mysql_real_query(mysql, delete_string.c_ptr_quick(),
delete_string.length()))
if (mysql_real_query(mysql, delete_string.ptr(), delete_string.length()))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
@ -1370,7 +1400,8 @@ int ha_federated::index_read_idx(byte * buf, uint index, const byte * key,
sql_query.length(0);
DBUG_ENTER("ha_federated::index_read_idx");
statistic_increment(table->in_use->status_var.ha_read_key_count,&LOCK_status);
statistic_increment(table->in_use->status_var.ha_read_key_count,
&LOCK_status);
sql_query.append(share->select_query);
sql_query.append(" WHERE ");
@ -1380,14 +1411,20 @@ int ha_federated::index_read_idx(byte * buf, uint index, const byte * key,
sql_query.append(index_string);
DBUG_PRINT("ha_federated::index_read_idx",
("current key %d key value %s index_string value %s length %d", index, (char *)(key),index_string.c_ptr_quick(),
("current key %d key value %s index_string value %s length %d",
index, (char*) key, index_string.c_ptr_quick(),
index_string.length()));
DBUG_PRINT("ha_federated::index_read_idx",
("current position %d sql_query %s", current_position,
sql_query.c_ptr_quick()));
if (mysql_real_query(mysql, sql_query.c_ptr_quick(), sql_query.length()))
if (result)
{
mysql_free_result(result);
result= 0;
}
if (mysql_real_query(mysql, sql_query.ptr(), sql_query.length()))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
@ -1449,9 +1486,31 @@ int ha_federated::rnd_init(bool scan)
DBUG_ENTER("ha_federated::rnd_init");
int num_fields, rows;
/*
This 'scan' flag is incredibly important for this handler to work properly,
especially with updates that are called with indexes, because what happens
without this is index_read_idx gets called, does a query using the
index in a where clause, calls mysql_store_result, which then rnd_init
(from sql_update.cc) is called after this, which would do a
"select * from table" then a mysql_store_result, wiping out the result
set from index_read_idx's query, which causes the subsequent update_row
to update the wrong row!
*/
scan_flag= scan;
if (scan)
{
DBUG_PRINT("ha_federated::rnd_init",
("share->select_query %s", share->select_query));
if (mysql_real_query(mysql, share->select_query, strlen(share->select_query)))
if (result)
{
DBUG_PRINT("ha_federated::rnd_init",
("mysql_free_result address %lx", result));
mysql_free_result(result);
result= 0;
}
if (mysql_real_query
(mysql, share->select_query, strlen(share->select_query)))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
@ -1460,12 +1519,21 @@ int ha_federated::rnd_init(bool scan)
if (mysql_errno(mysql))
DBUG_RETURN(mysql_errno(mysql));
}
DBUG_RETURN(0);
}
int ha_federated::rnd_end()
{
DBUG_ENTER("ha_federated::rnd_end");
if (result)
{
DBUG_PRINT("ha_federated::index_end",
("mysql_free_result address %lx", result));
mysql_free_result(result);
result= 0;
}
mysql_free_result(result);
DBUG_RETURN(index_end());
}
@ -1493,6 +1561,8 @@ int ha_federated::rnd_next(byte *buf)
// Fetch a row, insert it back in a row format.
current_position= result->data_cursor;
DBUG_PRINT("ha_federated::rnd_next",
("current position %d", current_position));
if (!(row= mysql_fetch_row(result)))
DBUG_RETURN(HA_ERR_END_OF_FILE);
@ -1535,12 +1605,23 @@ void ha_federated::position(const byte *record)
int ha_federated::rnd_pos(byte *buf, byte *pos)
{
DBUG_ENTER("ha_federated::rnd_pos");
statistic_increment(table->in_use->status_var.ha_read_rnd_count,&LOCK_status);
memcpy_fixed(&current_position, pos, sizeof(MYSQL_ROW_OFFSET)); // pos is not aligned
/*
we do not need to do any of this if there has been a scan performed already, or
if this is an update and index_read_idx already has a result set in which to build
it's update query from
*/
if (scan_flag)
{
statistic_increment(table->in_use->status_var.ha_read_rnd_count,
&LOCK_status);
memcpy_fixed(&current_position, pos, sizeof(MYSQL_ROW_OFFSET)); // pos
/* is not aligned */
result->current_row= 0;
result->data_cursor= current_position;
DBUG_RETURN(rnd_next(buf));
}
DBUG_RETURN(0);
}
/*
@ -1586,11 +1667,11 @@ int ha_federated::rnd_pos(byte * buf, byte *pos)
sql_update.cc
*/
// FIX: later version provide better information to the optimizer
/* FIX: later version provide better information to the optimizer */
void ha_federated::info(uint flag)
{
DBUG_ENTER("ha_federated::info");
records= 10000; // Fake!
records= 10000; // fix later
DBUG_VOID_RETURN;
}
@ -1618,7 +1699,8 @@ int ha_federated::delete_all_rows()
query.append("TRUNCATE ");
query.append(share->table_base_name);
if (mysql_real_query(mysql, query.c_ptr_quick(), query.length())) {
if (mysql_real_query(mysql, query.ptr(), query.length()))
{
my_error(ER_QUERY_ON_MASTER, MYF(0), mysql_error(mysql));
DBUG_RETURN(ER_QUERY_ON_MASTER);
}
@ -1670,8 +1752,7 @@ THR_LOCK_DATA **ha_federated::store_lock(THD *thd,
*/
if ((lock_type >= TL_WRITE_CONCURRENT_INSERT &&
lock_type <= TL_WRITE) && !thd->in_lock_tables
&& !thd->tablespace_op)
lock_type <= TL_WRITE) && !thd->in_lock_tables && !thd->tablespace_op)
lock_type= TL_WRITE_ALLOW_WRITE;
/*
@ -1701,17 +1782,14 @@ THR_LOCK_DATA **ha_federated::store_lock(THD *thd,
int ha_federated::create(const char *name, TABLE *table_arg,
HA_CREATE_INFO *create_info)
{
int retcode;
FEDERATED_SHARE tmp;
DBUG_ENTER("ha_federated::create");
retcode= parse_url(&tmp, table_arg, 1);
if (retcode < 0)
if (parse_url(&tmp, table_arg, 1))
{
DBUG_PRINT("ha_federated::create",
("ERROR: on table creation for %s called parse_url, retcode %d",
create_info->data_file_name, retcode));
my_error(ER_CANT_CREATE_TABLE, MYF(0));
DBUG_RETURN(ER_CANT_CREATE_TABLE);
}
my_free((gptr) tmp.scheme, MYF(0));
DBUG_RETURN(0);
}
#endif /* HAVE_FEDERATED_DB */

26
sql/ha_federated.h
View File

@ -32,13 +32,16 @@
FEDERATED_SHARE is a structure that will be shared amoung all open handlers
The example implements the minimum of what you will probably need.
*/
//FIX document
typedef struct st_federated_share {
char *table_name;
char *table_base_name;
// the primary select query to be used in rnd_init
/*
the primary select query to be used in rnd_init
*/
char *select_query;
// remote host info, parse_url supplies
/*
remote host info, parse_url supplies
*/
char *scheme;
char *hostname;
char *username;
@ -62,6 +65,7 @@ class ha_federated: public handler
FEDERATED_SHARE *share; /* Shared lock info */
MYSQL *mysql;
MYSQL_RES *result;
bool scan_flag;
uint ref_length;
uint fetch_num; // stores the fetch num
MYSQL_ROW_OFFSET current_position; // Current position used by ::position()
@ -73,10 +77,12 @@ private:
*/
uint convert_row_to_internal_format(byte *buf, MYSQL_ROW row);
bool create_where_from_key(String *to, KEY *key_info, const byte *key, uint key_length);
bool create_where_from_key(String *to, KEY *key_info,
const byte *key, uint key_length);
public:
ha_federated(TABLE *table): handler(table),
mysql(0),
mysql(0), result(0), scan_flag(0),
ref_length(sizeof(MYSQL_ROW_OFFSET)), current_position(0)
{
}
@ -126,11 +132,16 @@ public:
/*
Called in test_quick_select to determine if indexes should be used.
*/
virtual double scan_time() { DBUG_PRINT("ha_federated::scan_time", ("rows %d", records)); return (double)(records*2); }
virtual double scan_time()
{
DBUG_PRINT("ha_federated::scan_time",
("rows %d", records)); return (double)(records*2);
}
/*
The next method will never be called if you do not implement indexes.
*/
virtual double read_time(uint index, uint ranges, ha_rows rows) { return (double) rows / 20.0+1; }
virtual double read_time(uint index, uint ranges, ha_rows rows)
{ return (double) rows / 20.0+1; }
/*
Everything below are methods that we implment in ha_federated.cc.
@ -173,3 +184,6 @@ public:
THR_LOCK_DATA **store_lock(THD *thd, THR_LOCK_DATA **to,
enum thr_lock_type lock_type); //required
};
bool federated_db_init(void);
bool federated_db_end(void);

14
sql/handler.cc
View File

@ -395,6 +395,16 @@ int ha_init()
ha_was_inited_ok(ht++);
}
#endif
#ifdef HAVE_FEDERATED_DB
if (have_federated_db == SHOW_OPTION_YES)
{
if (federated_db_init())
{
have_federated_db= SHOW_OPTION_DISABLED;
error= 1;
}
}
#endif
#ifdef HAVE_ARCHIVE_DB
if (have_archive_db == SHOW_OPTION_YES)
{
@ -441,6 +451,10 @@ int ha_panic(enum ha_panic_function flag)
if (have_ndbcluster == SHOW_OPTION_YES)
error|=ndbcluster_end();
#endif
#ifdef HAVE_FEDERATED_DB
if (have_federated_db == SHOW_OPTION_YES)
error|= federated_db_end();
#endif
#ifdef HAVE_ARCHIVE_DB
if (have_archive_db == SHOW_OPTION_YES)
error|= archive_db_end();