Fix datatype reporting and collating sequence selection so that it works

correctly on views and with the UNION, EXCEPT, and INTERCEPT operators. (CVS 839)

FossilOrigin-Name: 71cc292dce59cf8224b205d1cdbff59ad12f1043

5 files changed, 158 insertions, 71 deletions

diff --git a/src/build.c b/src/build.c
index 87162601a..8a0cef5fd 100644
--- a/src/build.c
+++ b/src/build.c
@@ -25,7 +25,7 @@
 **     ROLLBACK
 **     PRAGMA
 **
-** $Id: build.c,v 1.123 2003/01/14 02:54:08 drh Exp $
+** $Id: build.c,v 1.124 2003/01/18 20:11:07 drh Exp $
 */
 #include "sqliteInt.h"
 #include <ctype.h>
@@ -530,40 +530,10 @@ void sqliteAddColumnType(Parse *pParse, Token *pFirst, Token *pLast){
     z[j++] = c;
   }
   z[j] = 0;
-  pCol->sortOrder = SQLITE_SO_NUM;
   if( pParse->db->file_format>=4 ){
-    for(i=0; z[i]; i++){
-      switch( z[i] ){
-        case 'b':
-        case 'B': {
-          if( sqliteStrNICmp(&z[i],"blob",4)==0 ){
-            pCol->sortOrder = SQLITE_SO_TEXT;
-            return;
-          }
-          break;
-        }
-        case 'c':
-        case 'C': {
-          if( sqliteStrNICmp(&z[i],"char",4)==0 ||
-                  sqliteStrNICmp(&z[i],"clob",4)==0 ){
-            pCol->sortOrder = SQLITE_SO_TEXT;
-            return;
-          }
-          break;
-        }
-        case 'x':
-        case 'X': {
-          if( i>=2 && sqliteStrNICmp(&z[i-2],"text",4)==0 ){
-            pCol->sortOrder = SQLITE_SO_TEXT;
-            return;
-          }
-          break;
-        }
-        default: {
-          break;
-        }
-      }
-    }
+    pCol->sortOrder = sqliteCollateType(z, n);
+  }else{
+    pCol->sortOrder = SQLITE_SO_NUM;
   }
 }
 
@@ -643,21 +613,47 @@ void sqliteAddPrimaryKey(Parse *pParse, IdList *pList, int onError){
 }
 
 /*
-** Return the appropriate collating type given the collation type token.
-** Report an error if the type is undefined.
+** Return the appropriate collating type given a type name.
+**
+** The collation type is text (SQLITE_SO_TEXT) if the type
+** name contains the character stream "text" or "blob" or
+** "clob".  Any other type name is collated as numeric
+** (SQLITE_SO_NUM).
 */
-int sqliteCollateType(Parse *pParse, Token *pType){
-  if( pType==0 ) return SQLITE_SO_UNK;
-  if( pType->n==4 && sqliteStrNICmp(pType->z, "text", 4)==0 ){
-    return SQLITE_SO_TEXT;
-  }
-  if( pType->n==7 && sqliteStrNICmp(pType->z, "numeric", 7)==0 ){
-    return SQLITE_SO_NUM;
-  }
-  sqliteSetNString(&pParse->zErrMsg, "unknown collating type: ", -1,
-    pType->z, pType->n, 0);
-  pParse->nErr++;
-  return SQLITE_SO_UNK;
+int sqliteCollateType(const char *zType, int nType){
+  int i;
+  int sortOrder = SQLITE_SO_NUM;
+  for(i=0; i<nType-1; i++){
+    switch( zType[i] ){
+      case 'b':
+      case 'B': {
+        if( i<nType-3 && sqliteStrNICmp(&zType[i],"blob",4)==0 ){
+          return SQLITE_SO_TEXT;
+        }
+        break;
+      }
+      case 'c':
+      case 'C': {
+        if( i<nType-3 && (sqliteStrNICmp(&zType[i],"char",4)==0 ||
+                           sqliteStrNICmp(&zType[i],"clob",4)==0)
+        ){
+          return SQLITE_SO_TEXT;
+        }
+        break;
+      }
+      case 'x':
+      case 'X': {
+        if( i>=2 && sqliteStrNICmp(&zType[i-2],"text",4)==0 ){
+          return SQLITE_SO_TEXT;
+        }
+        break;
+      }
+      default: {
+        break;
+      }
+    }
+  }
+  return SQLITE_SO_NUM;
 }
 
 /*
diff --git a/src/expr.c b/src/expr.c
index e56d05e3f..415cc3a1f 100644
--- a/src/expr.c
+++ b/src/expr.c
@@ -12,7 +12,7 @@
 ** This file contains routines used for analyzing expressions and
 ** for generating VDBE code that evaluates expressions in SQLite.
 **
-** $Id: expr.c,v 1.85 2003/01/14 02:49:28 drh Exp $
+** $Id: expr.c,v 1.86 2003/01/18 20:11:07 drh Exp $
 */
 #include "sqliteInt.h"
 #include <ctype.h>
@@ -122,7 +122,7 @@ void sqliteExprDelete(Expr *p){
 Expr *sqliteExprDup(Expr *p){
   Expr *pNew;
   if( p==0 ) return 0;
-  pNew = sqliteMalloc( sizeof(*p) );
+  pNew = sqliteMallocRaw( sizeof(*p) );
   if( pNew==0 ) return 0;
   memcpy(pNew, p, sizeof(*pNew));
   if( p->token.z!=0 ){
diff --git a/src/parse.y b/src/parse.y
index 73732005b..0b26348ed 100644
--- a/src/parse.y
+++ b/src/parse.y
@@ -14,7 +14,7 @@
 ** the parser.  Lemon will also generate a header file containing
 ** numeric codes for all of the tokens.
 **
-** @(#) $Id: parse.y,v 1.87 2003/01/13 23:27:33 drh Exp $
+** @(#) $Id: parse.y,v 1.88 2003/01/18 20:11:07 drh Exp $
 */
 %token_prefix TK_
 %token_type {Token}
@@ -174,7 +174,7 @@ ccons ::= REFERENCES nm(T) idxlist_opt(TA) refargs(R).
                                 {sqliteCreateForeignKey(pParse,0,&T,TA,R);}
 ccons ::= defer_subclause(D).   {sqliteDeferForeignKey(pParse,D);}
 ccons ::= COLLATE id(C).  {
-   sqliteAddCollateType(pParse, sqliteCollateType(pParse, &C));
+   sqliteAddCollateType(pParse, sqliteCollateType(C.z, C.n));
 }
 
 // The next group of rules parses the arguments to a REFERENCES clause
@@ -417,7 +417,7 @@ sortorder(A) ::= ASC.           {A = SQLITE_SO_ASC;}
 sortorder(A) ::= DESC.          {A = SQLITE_SO_DESC;}
 sortorder(A) ::= .              {A = SQLITE_SO_ASC;}
 collate(C) ::= .                {C = SQLITE_SO_UNK;}
-collate(C) ::= COLLATE id(X).   {C = sqliteCollateType(pParse, &X);}
+collate(C) ::= COLLATE id(X).   {C = sqliteCollateType(X.z, X.n);}
 
 %type groupby_opt {ExprList*}
 %destructor groupby_opt {sqliteExprListDelete($$);}
diff --git a/src/select.c b/src/select.c
index c989bde83..5a72a1a44 100644
--- a/src/select.c
+++ b/src/select.c
@@ -12,7 +12,7 @@
 ** This file contains C code routines that are called by the parser
 ** to handle SELECT statements in SQLite.
 **
-** $Id: select.c,v 1.121 2003/01/13 23:27:33 drh Exp $
+** $Id: select.c,v 1.122 2003/01/18 20:11:07 drh Exp $
 */
 #include "sqliteInt.h"
 
@@ -675,9 +675,48 @@ static void generateSortTail(
 }
 
 /*
-** Generate code that will tell the VDBE how many columns there
-** are in the result and the name for each column.  This information
-** is used to provide "argc" and "azCol[]" values in the callback.
+** Generate code that will tell the VDBE the datatypes of
+** columns in the result set.
+*/
+static void generateColumnTypes(
+  Parse *pParse,      /* Parser context */
+  int base,           /* VDBE cursor corresponding to first entry in pTabList */
+  SrcList *pTabList,  /* List of tables */
+  ExprList *pEList    /* Expressions defining the result set */
+){
+  Vdbe *v = pParse->pVdbe;
+  int i;
+  if( (pParse->db->flags & SQLITE_ReportTypes)==0 ) return;
+  for(i=0; i<pEList->nExpr; i++){
+    Expr *p = pEList->a[i].pExpr;
+    char *zType = 0;
+    if( p==0 ) continue;
+    if( p->op==TK_COLUMN && pTabList ){
+      Table *pTab = pTabList->a[p->iTable - base].pTab;
+      int iCol = p->iColumn;
+      if( iCol<0 ) iCol = pTab->iPKey;
+      assert( iCol==-1 || (iCol>=0 && iCol<pTab->nCol) );
+      if( iCol<0 ){
+        zType = "INTEGER";
+      }else{
+        zType = pTab->aCol[iCol].zType;
+      }
+    }else{
+      if( sqliteExprType(p)==SQLITE_SO_TEXT ){
+        zType = "TEXT";
+      }else{
+        zType = "NUMERIC";
+      }
+    }
+    sqliteVdbeAddOp(v, OP_ColumnName, i + pEList->nExpr, 0);
+    sqliteVdbeChangeP3(v, -1, zType, P3_STATIC);
+  }
+}
+
+/*
+** Generate code that will tell the VDBE the names of columns
+** in the result set.  This information is used to provide the
+** azCol[] vaolues in the callback.
 */
 static void generateColumnNames(
   Parse *pParse,      /* Parser context */
@@ -695,17 +734,6 @@ static void generateColumnNames(
     int showFullNames;
     p = pEList->a[i].pExpr;
     if( p==0 ) continue;
-    if( pParse->db->flags & SQLITE_ReportTypes ){
-      if( zType==0 ){
-        if( sqliteExprType(p)==SQLITE_SO_TEXT ){
-          zType = "TEXT";
-        }else{
-          zType = "NUMERIC";
-        }
-      }
-      sqliteVdbeAddOp(v, OP_ColumnName, i + pEList->nExpr, 0);
-      sqliteVdbeChangeP3(v, -1, zType, P3_STATIC);
-    }
     if( pEList->a[i].zName ){
       char *zName = pEList->a[i].zName;
       sqliteVdbeAddOp(v, OP_ColumnName, i, 0);
@@ -1050,6 +1078,13 @@ void sqliteSelectUnbind(Select *p){
 **
 ** Any entry that does not match is flagged as an error.  The number
 ** of errors is returned.
+**
+** This routine does NOT correctly initialize the Expr.dataType  field
+** of the ORDER BY expressions.  The multiSelectSortOrder() routine
+** must be called to do that after the individual select statements
+** have all been analyzed.  This routine is unable to compute Expr.dataType
+** because it must be called before the individual select statements
+** have been analyzed.
 */
 static int matchOrderbyToColumn(
   Parse *pParse,          /* A place to leave error messages */
@@ -1089,6 +1124,7 @@ static int matchOrderbyToColumn(
         nErr++;
         break;
       }
+      if( !mustComplete ) continue;
       iCol--;
     }
     for(j=0; iCol<0 && j<pEList->nExpr; j++){
@@ -1137,7 +1173,43 @@ Vdbe *sqliteGetVdbe(Parse *pParse){
   }
   return v;
 }
-    
+
+/*
+** This routine sets the Expr.dataType field on all elements of
+** the pOrderBy expression list.  The pOrderBy list will have been
+** set up by matchOrderbyToColumn().  Hence each expression has
+** a TK_COLUMN as its root node.  The Expr.iColumn refers to a 
+** column in the result set.   The datatype is set to SQLITE_SO_TEXT
+** if the corresponding column in p and every SELECT to the left of
+** p has a datatype of SQLITE_SO_TEXT.  If the cooressponding column
+** in p or any of the left SELECTs is SQLITE_SO_NUM, then the datatype
+** of the order-by expression is set to SQLITE_SO_NUM.
+**
+** Examples:
+**
+**    SELECT a,b
+*/ 
+static void multiSelectSortOrder(Select *p, ExprList *pOrderBy){
+  int i;
+  ExprList *pEList;
+  if( pOrderBy==0 ) return;
+  if( p==0 ){
+    for(i=0; i<pOrderBy->nExpr; i++){
+      pOrderBy->a[i].pExpr->dataType = SQLITE_SO_TEXT;
+    }
+    return;
+  }
+  multiSelectSortOrder(p->pPrior, pOrderBy);
+  pEList = p->pEList;
+  for(i=0; i<pOrderBy->nExpr; i++){
+    Expr *pE = pOrderBy->a[i].pExpr;
+    if( pE->dataType==SQLITE_SO_NUM ) continue;
+    assert( pE->iColumn>=0 );
+    if( pEList->nExpr>pE->iColumn ){
+      pE->dataType = sqliteExprType(pEList->a[pE->iColumn].pExpr);
+    }
+  }
+}
 
 /*
 ** This routine is called to process a query that is really the union
@@ -1248,11 +1320,13 @@ static int multiSelect(Parse *pParse, Select *p, int eDest, int iParm){
         assert( p->pEList );
         if( eDest==SRT_Callback ){
           generateColumnNames(pParse, p->base, 0, p->pEList);
+          generateColumnTypes(pParse, p->base, p->pSrc, p->pEList);
         }
         iBreak = sqliteVdbeMakeLabel(v);
         iCont = sqliteVdbeMakeLabel(v);
         sqliteVdbeAddOp(v, OP_Rewind, unionTab, iBreak);
         iStart = sqliteVdbeCurrentAddr(v);
+        multiSelectSortOrder(p, p->pOrderBy);
         rc = selectInnerLoop(pParse, p, p->pEList, unionTab, p->pEList->nExpr,
                              p->pOrderBy, -1, eDest, iParm, 
                              iCont, iBreak);
@@ -1303,12 +1377,14 @@ static int multiSelect(Parse *pParse, Select *p, int eDest, int iParm){
       assert( p->pEList );
       if( eDest==SRT_Callback ){
         generateColumnNames(pParse, p->base, 0, p->pEList);
+        generateColumnTypes(pParse, p->base, p->pSrc, p->pEList);
       }
       iBreak = sqliteVdbeMakeLabel(v);
       iCont = sqliteVdbeMakeLabel(v);
       sqliteVdbeAddOp(v, OP_Rewind, tab1, iBreak);
       iStart = sqliteVdbeAddOp(v, OP_FullKey, tab1, 0);
       sqliteVdbeAddOp(v, OP_NotFound, tab2, iCont);
+      multiSelectSortOrder(p, p->pOrderBy);
       rc = selectInnerLoop(pParse, p, p->pEList, tab1, p->pEList->nExpr,
                              p->pOrderBy, -1, eDest, iParm, 
                              iCont, iBreak);
@@ -1331,6 +1407,12 @@ static int multiSelect(Parse *pParse, Select *p, int eDest, int iParm){
     pParse->nErr++;
     return 1;
   }
+
+  /* Issue a null callback if that is what the user wants.
+  */
+  if( (pParse->db->flags & SQLITE_NullCallback)!=0 && eDest==SRT_Callback ){
+    sqliteVdbeAddOp(v, OP_NullCallback, p->pEList->nExpr, 0);
+  }
   return 0;
 }
 
@@ -1383,6 +1465,7 @@ static void substExpr(Expr *pExpr, int iTable, ExprList *pEList, int iSub){
     pNew = pEList->a[pExpr->iColumn].pExpr;
     assert( pNew!=0 );
     pExpr->op = pNew->op;
+    pExpr->dataType = pNew->dataType;
     assert( pExpr->pLeft==0 );
     pExpr->pLeft = sqliteExprDup(pNew->pLeft);
     assert( pExpr->pRight==0 );
@@ -1686,6 +1769,7 @@ static int simpleMinMaxQuery(Parse *pParse, Select *p, int eDest, int iParm){
   if( v==0 ) return 0;
   if( eDest==SRT_Callback ){
     generateColumnNames(pParse, p->base, p->pSrc, p->pEList);
+    generateColumnTypes(pParse, p->base, p->pSrc, p->pEList);
   }
 
   /* Generating code to find the min or the max.  Basically all we have
@@ -1992,6 +2076,13 @@ int sqliteSelect(
     return rc;
   }
 
+  /* Identify column types if we will be using in the callback.  This
+  ** step is skipped if the output is going to a table or a memory cell.
+  */
+  if( eDest==SRT_Callback ){
+    generateColumnTypes(pParse, p->base, pTabList, pEList);
+  }
+
   /* If the output is destined for a temporary table, open that table.
   */
   if( eDest==SRT_TempTable ){
diff --git a/src/sqliteInt.h b/src/sqliteInt.h
index bec619c66..51d7a55ac 100644
--- a/src/sqliteInt.h
+++ b/src/sqliteInt.h
@@ -11,7 +11,7 @@
 *************************************************************************
 ** Internal interface definitions for SQLite.
 **
-** @(#) $Id: sqliteInt.h,v 1.155 2003/01/16 16:28:54 drh Exp $
+** @(#) $Id: sqliteInt.h,v 1.156 2003/01/18 20:11:07 drh Exp $
 */
 #include "config.h"
 #include "sqlite.h"
@@ -948,7 +948,7 @@ void sqliteAddNotNull(Parse*, int);
 void sqliteAddPrimaryKey(Parse*, IdList*, int);
 void sqliteAddColumnType(Parse*,Token*,Token*);
 void sqliteAddDefaultValue(Parse*,Token*,int);
-int sqliteCollateType(Parse*, Token*);
+int sqliteCollateType(const char*, int);
 void sqliteAddCollateType(Parse*, int);
 void sqliteEndTable(Parse*,Token*,Select*);
 void sqliteCreateView(Parse*,Token*,Token*,Select*,int);