Bug#45983 ibmdb2i_create_index_option=1 not working for primary key

With ibmdb2i_create_index_option set to 1, creating an IBMDB2I table
with a primary key should produce an additional index that uses EBCDIC
hexadecimal sorting. However, this does not work. Adding indexes that
are not primary keys does work. The ibmdb2i_create_index_option should
be honoured when creating a table with a primary key.

This patch adds code to the create() function to check for the value
of the ibmdb2i_create_index_option variable and, when appropriate, to 
generate a *HEX-based shadow index in DB2 for the primary key. Previously 
this behavior was limited to secondary indexes.

Additionally, this patch restricts the creation of shadow indexes to
cases in which a non-*HEX sort sequence is used, as the documentation
for ibmdb2i_create_index_option describes. Previously, the shadow index
would in some cases be created even when the MySQL-specific index used
*HEX sorting, leading to redundant indexes.

Finally, the code used to generate the list of fields for indexes 
and the code used to generate the SQL statement for the shadow
indexes has been refactored into individual functions.

mysql-test/suite/ibmdb2i/r/ibmdb2i_bug_45983.result

+set ibmdb2i_create_index_option=1;
+drop schema if exists test1;
+create schema test1;
+use test1;
+CREATE TABLE t1 (f int primary key, index(f)) engine=ibmdb2i;
+drop table t1;
+CREATE TABLE t1 (f char(10) collate utf8_bin primary key, index(f)) engine=ibmdb2i;
+drop table t1;
+CREATE TABLE t1 (f char(10) collate latin1_swedish_ci primary key, index(f)) engine=ibmdb2i;
+drop table t1;
+CREATE TABLE t1 (f char(10) collate latin1_swedish_ci primary key, i int, index i(i,f)) engine=ibmdb2i;
+drop table t1;
+create table fd (SQSSEQ CHAR(10)) engine=ibmdb2i;
+select * from fd;
+SQSSEQ
+*HEX
+*HEX
+*HEX
+*HEX
+drop table fd;

mysql-test/suite/ibmdb2i/t/ibmdb2i_bug_45983.test

+source suite/ibmdb2i/include/have_ibmdb2i.inc;
+
+# Confirm that ibmdb2i_create_index_option causes additional *HEX sorted indexes to be created for all non-binary keys.
+
+set ibmdb2i_create_index_option=1;
+--disable_warnings
+drop schema if exists test1;
+create schema test1;
+use test1;
+--enable_warnings
+
+--disable_abort_on_error
+--error 0,255
+exec system "DLTF QGPL/FDOUT" > /dev/null;
+--enable_abort_on_error
+
+#No additional index because no string fields in key
+CREATE TABLE t1 (f int primary key, index(f)) engine=ibmdb2i;
+--error 255
+exec system "DSPFD FILE(\"test1\"/PRIM0001) TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+--error 255
+exec system "DSPFD FILE(\"test1\"/\"f___H_t1\") TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+drop table t1;
+
+#No additional index because binary sorting
+CREATE TABLE t1 (f char(10) collate utf8_bin primary key, index(f)) engine=ibmdb2i;
+--error 255
+exec system "DSPFD FILE(\"test1\"/PRIM0001) TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+--error 255
+exec system "DSPFD FILE(\"test1\"/\"f___H_t1\") TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+drop table t1;
+
+CREATE TABLE t1 (f char(10) collate latin1_swedish_ci primary key, index(f)) engine=ibmdb2i;
+exec system "DSPFD FILE(\"test1\"/PRIM0001) TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+exec system "DSPFD FILE(\"test1\"/\"f___H_t1\") TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+drop table t1;
+
+CREATE TABLE t1 (f char(10) collate latin1_swedish_ci primary key, i int, index i(i,f)) engine=ibmdb2i;
+exec system "DSPFD FILE(\"test1\"/PRIM0001) TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+exec system "DSPFD FILE(\"test1\"/\"i___H_t1\") TYPE(*SEQ) OUTPUT(*OUTFILE) OUTFILE(QGPL/FDOUT) OUTMBR(*FIRST *ADD)" > /dev/null;
+drop table t1;
+
+
+create table fd (SQSSEQ CHAR(10)) engine=ibmdb2i;
+system system "CPYF FROMFILE(QGPL/FDOUT) TOFILE(\"test1\"/\"fd\") mbropt(*replace) fmtopt(*drop *map)" > /dev/null;
+select * from fd;
+drop table fd;

storage/ibmdb2i/ha_ibmdb2i.cc

@@ -2230,34 +2230,19 @@ int ha_ibmdb2i::create(const char *name, TABLE *table_arg,
     } 
   }
   
-  bool primaryHasStringField = false;
+  String fieldDefinition(128);
   
   if (table_arg->s->primary_key != MAX_KEY && !isTemporary)
   {
-    KEY& curKey = table_arg->key_info[table_arg->s->primary_key];
-    query.append(STRING_WITH_LEN(", PRIMARY KEY( "));
-    for (int j = 0; j < curKey.key_parts; ++j)
-    {
-      if (j != 0)
-      {
-        query.append( STRING_WITH_LEN(" , ") );
-      }
-      Field* field = curKey.key_part[j].field;
-      convertMySQLNameToDB2Name(field->field_name, colName, sizeof(colName));
-      query.append(colName);
-      enum_field_types type = field->real_type();
-      if (type == MYSQL_TYPE_VARCHAR || type == MYSQL_TYPE_BLOB ||
-          type == MYSQL_TYPE_STRING)
-      {
-        rc = updateAssociatedSortSequence(field->charset(),
+    query.append(STRING_WITH_LEN(", PRIMARY KEY "));
+    rc = buildIndexFieldList(fieldDefinition, 
+                             table_arg->key_info[table_arg->s->primary_key],
+                             true,
                              &fileSortSequenceType,
                              fileSortSequence,
                              fileSortSequenceLibrary);
-        if (rc) DBUG_RETURN (rc);
-        primaryHasStringField = true;
-      }
-    }
-    query.append(STRING_WITH_LEN(" ) "));
+    if (rc) DBUG_RETURN(rc);
+    query.append(fieldDefinition);
   }
 
   rc = buildDB2ConstraintString(thd->lex, 
@@ -2283,6 +2268,19 @@ int ha_ibmdb2i::create(const char *name, TABLE *table_arg,
   SqlStatementStream sqlStream(query.length());
   sqlStream.addStatement(query,fileSortSequence,fileSortSequenceLibrary);
   
+  if (table_arg->s->primary_key != MAX_KEY && 
+      !isTemporary &&
+      (THDVAR(thd, create_index_option)==1) &&
+      (fileSortSequenceType != 'B') &&
+      (fileSortSequenceType != ' '))
+  {
+    rc = generateShadowIndex(sqlStream, 
+                             table_arg->key_info[table_arg->s->primary_key], 
+                             libName, 
+                             fileName, 
+                             fieldDefinition);
+    if (rc) DBUG_RETURN(rc);
+  }
   for (uint i = 0; i < table_arg->s->keys; ++i)
   {
     if (i != table_arg->s->primary_key || isTemporary)
@@ -3012,58 +3010,123 @@ int32 ha_ibmdb2i::buildCreateIndexStatement(SqlStatementStream& sqlStream,
   }
   
   String fieldDefinition(128);
-  fieldDefinition.length(0);
-  fieldDefinition.append(STRING_WITH_LEN(" ( "));
+  rc = buildIndexFieldList(fieldDefinition,
+                           key,
+                           isPrimary,
+                           &fileSortSequenceType, 
+                           fileSortSequence,
+                           fileSortSequenceLibrary);
+  
+  if (rc) DBUG_RETURN(rc);
+   
+  query.append(fieldDefinition);
+  
+  if ((THDVAR(ha_thd(), create_index_option)==1) &&
+      (fileSortSequenceType != 'B') &&
+      (fileSortSequenceType != ' '))
+  {
+    rc = generateShadowIndex(sqlStream, 
+                             key, 
+                             db2LibName, 
+                             db2FileName, 
+                             fieldDefinition);
+    if (rc) DBUG_RETURN(rc);
+  }
+    
+  DBUG_PRINT("ha_ibmdb2i::buildCreateIndexStatement", ("Sent to DB2: %s",query.c_ptr_safe()));
+  sqlStream.addStatement(query,fileSortSequence,fileSortSequenceLibrary);
+
+  DBUG_RETURN(0);
+}
+
+/**
+  Generate the SQL syntax for the list of fields to be assigned to the 
+  specified key. The corresponding sort sequence is also calculated.
+      
+  @param[out] appendHere  The string to receive the generated SQL
+  @param key  The key to evaluate
+  @param isPrimary  True if this is being generated on behalf of the primary key
+  @param[out] fileSortSequenceType  The type of the associated sort sequence
+  @param[out] fileSortSequence  The name of the associated sort sequence
+  @param[out] fileSortSequenceLibrary  The library of the associated sort sequence
+  
+  @return  0 if successful; error value otherwise
+*/
+int32 ha_ibmdb2i::buildIndexFieldList(String& appendHere,
+                                      const KEY& key,
+                                      bool isPrimary,
+                                      char* fileSortSequenceType, 
+                                      char* fileSortSequence, 
+                                      char* fileSortSequenceLibrary)
+{
+  DBUG_ENTER("ha_ibmdb2i::buildIndexFieldList");
+  appendHere.append(STRING_WITH_LEN(" ( "));
   for (int j = 0; j < key.key_parts; ++j)
   {
     char colName[MAX_DB2_COLNAME_LENGTH+1];
     if (j != 0)
     {
-      fieldDefinition.append(STRING_WITH_LEN(" , "));
+      appendHere.append(STRING_WITH_LEN(" , "));
     }
-    Field* field = key.key_part[j].field;
-    convertMySQLNameToDB2Name(field->field_name, colName, sizeof(colName));
-    fieldDefinition.append(colName);
+    
+    KEY_PART_INFO& kpi = key.key_part[j];
+    Field* field = kpi.field;
+    
+    convertMySQLNameToDB2Name(field->field_name, 
+                              colName, 
+                              sizeof(colName));
+    appendHere.append(colName);
+    
+    int32 rc;
     rc = updateAssociatedSortSequence(field->charset(),
-                                      &fileSortSequenceType,
+                                      fileSortSequenceType,
                                       fileSortSequence,
                                       fileSortSequenceLibrary);
     if (rc) DBUG_RETURN (rc);
   }
-  fieldDefinition.append(STRING_WITH_LEN(" ) "));
     
-  query.append(fieldDefinition);
+  appendHere.append(STRING_WITH_LEN(" ) "));
   
-  if ((THDVAR(ha_thd(), create_index_option)==1) &&
-      (fileSortSequenceType != 'B'))
-  {
+  DBUG_RETURN(0);
+}
+
+
+/**
+  Generate an SQL statement that defines a *HEX sorted index to implement 
+  the ibmdb2i_create_index.
+      
+  @param[out] stream  The stream to append the generated statement to
+  @param key  The key to evaluate
+  @param[out] libName  The library containg the table
+  @param[out] fileName  The DB2-compatible name of the table 
+  @param[out] fieldDefinition  The list of the fields in the index, in SQL syntax
+  
+  @return  0 if successful; error value otherwise
+*/
+int32 ha_ibmdb2i::generateShadowIndex(SqlStatementStream& stream, 
+                                      const KEY& key,
+                                      const char* libName,
+                                      const char* fileName,
+                                      const String& fieldDefinition)
+{
   String shadowQuery(256);
   shadowQuery.length(0);
-    
   shadowQuery.append(STRING_WITH_LEN("CREATE INDEX "));
-
-    shadowQuery.append(db2LibName);
+  shadowQuery.append(libName);
   shadowQuery.append('.');
-    if (db2i_table::appendQualifiedIndexFileName(key.name, db2FileName, shadowQuery, db2i_table::ASCII_SQL, typeHex))
+  if (db2i_table::appendQualifiedIndexFileName(key.name, fileName, shadowQuery, db2i_table::ASCII_SQL, typeHex))
   {
     getErrTxt(DB2I_ERR_INVALID_NAME,"index","*generated*");
-      DBUG_RETURN(DB2I_ERR_INVALID_NAME );
+    return DB2I_ERR_INVALID_NAME;
   }
-
   shadowQuery.append(STRING_WITH_LEN(" ON "));
-
-    shadowQuery.append(db2LibName);
+  shadowQuery.append(libName);
   shadowQuery.append('.');
-    shadowQuery.append(db2FileName);
+  shadowQuery.append(fileName);
   shadowQuery.append(fieldDefinition);
-    DBUG_PRINT("ha_ibmdb2i::buildCreateIndexStatement", ("Sent to DB2: %s",shadowQuery.c_ptr_safe()));
-    sqlStream.addStatement(shadowQuery,"*HEX","QSYS");
-  }
-    
-  DBUG_PRINT("ha_ibmdb2i::buildCreateIndexStatement", ("Sent to DB2: %s",query.c_ptr_safe()));
-  sqlStream.addStatement(query,fileSortSequence,fileSortSequenceLibrary);
-
-  DBUG_RETURN(0);
+  DBUG_PRINT("ha_ibmdb2i::generateShadowIndex", ("Sent to DB2: %s",shadowQuery.c_ptr_safe()));
+  stream.addStatement(shadowQuery,"*HEX","QSYS");
+  return 0;
 }
   
   

storage/ibmdb2i/ha_ibmdb2i.h

@@ -531,6 +531,13 @@ private:
                                  const char* db2LibName,    
                                  const char* db2FileName);
   
+  int32 buildIndexFieldList(String& appendHere,
+                            const KEY& key,
+                            bool isPrimary,
+                            char* fileSortSequenceType, 
+                            char* fileSortSequence, 
+                            char* fileSortSequenceLibrary);
+
   // Specify NULL for data when using the data pointed to by field
   int32 convertMySQLtoDB2(Field* field, const DB2Field& db2Field, char* db2Buf, const uchar* data = NULL); 
 
@@ -806,4 +813,10 @@ private:
     query.append(STRING_WITH_LEN(" RCDFMT "));
     query.append(rcdfmt);
   }
+  
+  int32 generateShadowIndex(SqlStatementStream& stream, 
+                           const KEY& key,
+                           const char* libName,
+                           const char* fileName,
+                           const String& fieldDefinition);
 };