Fixed parsing of column names and foreign key constraints in Innobase to handle quoted identifiers and identifiers with space. (Bug #1725)

Fix optimizer tuning bug when first used key part was a constant. (Bug #1679)

innobase/dict/dict0dict.c

@@ -2138,19 +2138,37 @@ dict_scan_col(
 		return(ptr);
 	}
 
-	if (*ptr == '`') {
-		ptr++;
-	}
+	if (*ptr == '`' || *ptr == '"') {
+	  /*
+	    The identifier is quoted. Search for end quote.
+	    We can't use the general code here as the name may contain
+	    special characters like space.
+	  */
+	  char quote= *ptr++;
 
-	old_ptr = ptr;
+	  old_ptr= ptr;
+	  /*
+	    The colum name should always end with 'quote' but we check for
+	    end zero just to be safe if this is called outside of MySQL
+	  */
+	  while (*ptr && *ptr != quote)
+	    ptr++;
+	  *column_name_len = (ulint)(ptr - old_ptr);
+
+	  if (*ptr)				/* Skip end quote */
+	    ptr++;
+	}
+	else
+	{
+	  old_ptr = ptr;
 	
-	while (!isspace(*ptr) && *ptr != ',' && *ptr != ')' && 	*ptr != '`'
-	       && *ptr != '\0') {
-
+	  while (!isspace(*ptr) && *ptr != ',' && *ptr != ')'
+		 && *ptr != '\0') {
 		ptr++;
+	  }
+	  *column_name_len = (ulint)(ptr - old_ptr);
 	}
 
-	*column_name_len = (ulint)(ptr - old_ptr);
 	
 	if (table == NULL) {
 		*success = TRUE;
@@ -2161,9 +2179,9 @@ dict_scan_col(
 
 			col = dict_table_get_nth_col(table, i);
 
-			if (ut_strlen(col->name) == (ulint)(ptr - old_ptr)
+			if (ut_strlen(col->name) == *column_name_len
 			    && 0 == ut_cmp_in_lower_case(col->name, old_ptr,
-						(ulint)(ptr - old_ptr))) {
+						*column_name_len)) {
 		    		/* Found */
 
 		    		*success = TRUE;
@@ -2175,10 +2193,6 @@ dict_scan_col(
 		}
 	}
 	
-	if (*ptr == '`') {
-		ptr++;
-	}
-	
 	return(ptr);
 }
 
@@ -2200,6 +2214,7 @@ dict_scan_table_name(
 	char*	dot_ptr			= NULL;
 	char*	old_ptr;
 	ulint	i;
+	char	quote			= 0;
 	
 	*success = FALSE;
 	*table = NULL;
@@ -2213,14 +2228,16 @@ dict_scan_table_name(
 		return(ptr);
 	}
 
-	if (*ptr == '`') {
-		ptr++;
+	if (*ptr == '`' || *ptr == '"') {
+		quote= *ptr++;
 	}
 
 	old_ptr = ptr;
 	
-	while (!isspace(*ptr) && *ptr != '(' && *ptr != '`' && *ptr != '\0') {
-		if (*ptr == '.') {
+	while (*ptr != quote &&
+	       (quote || (!isspace(*ptr) && *ptr != '(')) &&
+	       *ptr != '\0') {
+		if (!quote && *ptr == '.') {
 			dot_ptr = ptr;
 		}
 
@@ -2273,7 +2290,7 @@ dict_scan_table_name(
 
 	*table = dict_table_get_low(second_table_name);
 
-	if (*ptr == '`') {
+	if (*ptr && *ptr == quote) {
 		ptr++;
 	}
 
@@ -2293,7 +2310,7 @@ dict_scan_id(
 			scannable */
 	ulint*	len)	/* out: length of the id */
 {
-	ibool	scanned_backquote	= FALSE;
+	char quote	= 0;
 
 	*start = NULL;
 
@@ -2306,23 +2323,23 @@ dict_scan_id(
 		return(ptr);
 	}
 
-	if (*ptr == '`') {
-		scanned_backquote = TRUE;
-		ptr++;
+	if (*ptr == '`' || *ptr == '"') {
+		quote = *ptr++;
 	}
 	
 	*start = ptr;
 
-	while (!isspace(*ptr) && *ptr != ',' && *ptr != '(' && *ptr != ')'
-		&& *ptr != '\0' && *ptr != '`') {
-
+	while (*ptr != quote &&
+	       (!quote || (!isspace(*ptr) && *ptr != ',' && *ptr != '(' &&
+			   *ptr != ')'))
+	       && *ptr != '\0') {
 		ptr++;
 	}
 
 	*len = (ulint) (ptr - *start);
 	
-	if (scanned_backquote) {
-		if (*ptr == '`') {
+	if (quote) {
+		if (*ptr == quote) {
 			ptr++;
 		} else {
 			/* Syntax error */

mysql-test/r/innodb.result

@@ -1243,3 +1243,6 @@ a
 3
 4
 drop table t1;
+CREATE TABLE t1 (`id 1` INT NOT NULL, PRIMARY KEY (`id 1`)) TYPE=INNODB;
+CREATE TABLE t2 (id INT PRIMARY KEY, t1_id INT, INDEX par_ind (t1_id), FOREIGN KEY (`t1_id`) REFERENCES `t1`(`id 1`)  ON DELETE CASCADE ) TYPE=INNODB;
+drop table t1,t2;

mysql-test/t/innodb.test

@@ -869,3 +869,12 @@ truncate table t1;
 insert into t1 (a) values (NULL),(NULL);
 SELECT * from t1;
 drop table t1;
+
+#
+# Test dictionary handling with spaceand quoting
+#
+
+CREATE TABLE t1 (`id 1` INT NOT NULL, PRIMARY KEY (`id 1`)) TYPE=INNODB;
+CREATE TABLE t2 (id INT PRIMARY KEY, t1_id INT, INDEX par_ind (t1_id), FOREIGN KEY (`t1_id`) REFERENCES `t1`(`id 1`)  ON DELETE CASCADE ) TYPE=INNODB;
+#show create table t2;
+drop table t1,t2;

sql/sql_class.cc

@@ -538,7 +538,6 @@ select_export::~select_export()
 int
 select_export::prepare(List<Item> &list)
 {
-  char path[FN_REFLEN];
   uint option=4;
   bool blob_flag=0;
 #ifdef DONT_ALLOW_FULL_LOAD_DATA_PATHS
@@ -739,9 +738,13 @@ err:
 void select_export::send_error(uint errcode,const char *err)
 {
   ::send_error(&thd->net,errcode,err);
-  (void) end_io_cache(&cache);
-  (void) my_close(file,MYF(0));
-  file= -1;
+  if (file > 0)
+  {
+    (void) end_io_cache(&cache);
+    (void) my_close(file,MYF(0));
+    (void) my_delete(path,MYF(0));		// Delete file on error
+    file= -1;
+  }
 }
 
 
@@ -849,10 +852,13 @@ err:
 void select_dump::send_error(uint errcode,const char *err)
 {
   ::send_error(&thd->net,errcode,err);
-  (void) end_io_cache(&cache);
-  (void) my_close(file,MYF(0));
-  (void) my_delete(path,MYF(0));		// Delete file on error
-  file= -1;
+  if (file > 0)
+  {
+    (void) end_io_cache(&cache);
+    (void) my_close(file,MYF(0));
+    (void) my_delete(path,MYF(0));		// Delete file on error
+    file= -1;
+  }
 }
 
 

sql/sql_class.h

@@ -665,11 +665,13 @@ class select_export :public select_result {
   File file;
   IO_CACHE cache;
   ha_rows row_count;
+  char path[FN_REFLEN];
   uint field_term_length;
   int field_sep_char,escape_char,line_sep_char;
   bool fixed_row_size;
 public:
-  select_export(sql_exchange *ex) :exchange(ex),file(-1),row_count(0L) {}
+  select_export(sql_exchange *ex) :exchange(ex),file(-1),row_count(0L)
+  { path[0]=0; }
   ~select_export();
   int prepare(List<Item> &list);
   bool send_fields(List<Item> &list,

sql/sql_select.cc

@@ -1912,7 +1912,8 @@ find_best(JOIN *join,table_map rest_tables,uint idx,double record_count,
 
     read_time+=record_count/(double) TIME_FOR_COMPARE;
     if (join->sort_by_table &&
-	join->sort_by_table != join->positions[join->const_tables].table->table)
+	join->sort_by_table !=
+	join->positions[join->const_tables].table->table)
       read_time+=record_count;			// We have to make a temp table
     if (read_time < join->best_read)
     {
@@ -1946,7 +1947,7 @@ find_best(JOIN *join,table_map rest_tables,uint idx,double record_count,
 	uint max_key_part=0;
 
 	/* Test how we can use keys */
-	rec= s->records/MATCHING_ROWS_IN_OTHER_TABLE;  /* Assumed records/key */
+	rec= s->records/MATCHING_ROWS_IN_OTHER_TABLE;  // Assumed records/key
 	for (keyuse=s->keyuse ; keyuse->table == table ;)
 	{
 	  key_map found_part=0;
@@ -2085,7 +2086,7 @@ find_best(JOIN *join,table_map rest_tables,uint idx,double record_count,
 		will match
 	      */
 	      if (table->quick_keys & ((key_map) 1 << key) &&
-		  table->quick_key_parts[key] <= max_key_part)
+		  table->quick_key_parts[key] == max_key_part)
 		tmp=records= (double) table->quick_rows[key];
 	      else
 	      {
@@ -2127,6 +2128,14 @@ find_best(JOIN *join,table_map rest_tables,uint idx,double record_count,
 		  }
 		  records=(ulong) tmp;
 		}
+		/*
+		  If quick_select was used on a part of this key, we know
+		  the maximum number of rows that the key can match.
+		*/
+		if (table->quick_keys & ((key_map) 1 << key) &&
+		    table->quick_key_parts[key] <= max_key_part &&
+		    records > (double) table->quick_rows[key])
+		  tmp= records= (double) table->quick_rows[key];
 	      }
 	      /* Limit the number of matched rows */
 	      set_if_smaller(tmp, (double) thd->variables.max_seeks_for_key);