Add support for (variable length) integer keys in LSM1.

FossilOrigin-Name: 32f3daec0a8084b258b2513c8025bc4d95a5d757

ext/lsm1/lsm_vtab.c

@@ -199,6 +199,135 @@ static int lsm1Rowid(sqlite3_vtab_cursor *cur, sqlite_int64 *pRowid){
 #define LSM1_TYPE_TEXT       2
 #define LSM1_TYPE_BLOB       3
 
+/*
+** Write a 32-bit unsigned integer as 4 big-endian bytes.
+*/
+static void varintWrite32(unsigned char *z, unsigned int y){
+  z[0] = (unsigned char)(y>>24);
+  z[1] = (unsigned char)(y>>16);
+  z[2] = (unsigned char)(y>>8);
+  z[3] = (unsigned char)(y);
+}
+
+/*
+** Write a varint into z[].  The buffer z[] must be at least 9 characters
+** long to accommodate the largest possible varint.  Return the number of
+** bytes of z[] used.
+*/
+static int lsm1PutVarint64(unsigned char *z, sqlite3_uint64 x){
+  unsigned int w, y;
+  if( x<=240 ){
+    z[0] = (unsigned char)x;
+    return 1;
+  }
+  if( x<=2287 ){
+    y = (unsigned int)(x - 240);
+    z[0] = (unsigned char)(y/256 + 241);
+    z[1] = (unsigned char)(y%256);
+    return 2;
+  }
+  if( x<=67823 ){
+    y = (unsigned int)(x - 2288);
+    z[0] = 249;
+    z[1] = (unsigned char)(y/256);
+    z[2] = (unsigned char)(y%256);
+    return 3;
+  }
+  y = (unsigned int)x;
+  w = (unsigned int)(x>>32);
+  if( w==0 ){
+    if( y<=16777215 ){
+      z[0] = 250;
+      z[1] = (unsigned char)(y>>16);
+      z[2] = (unsigned char)(y>>8);
+      z[3] = (unsigned char)(y);
+      return 4;
+    }
+    z[0] = 251;
+    varintWrite32(z+1, y);
+    return 5;
+  }
+  if( w<=255 ){
+    z[0] = 252;
+    z[1] = (unsigned char)w;
+    varintWrite32(z+2, y);
+    return 6;
+  }
+  if( w<=65535 ){
+    z[0] = 253;
+    z[1] = (unsigned char)(w>>8);
+    z[2] = (unsigned char)w;
+    varintWrite32(z+3, y);
+    return 7;
+  }
+  if( w<=16777215 ){
+    z[0] = 254;
+    z[1] = (unsigned char)(w>>16);
+    z[2] = (unsigned char)(w>>8);
+    z[3] = (unsigned char)w;
+    varintWrite32(z+4, y);
+    return 8;
+  }
+  z[0] = 255;
+  varintWrite32(z+1, w);
+  varintWrite32(z+5, y);
+  return 9;
+}
+
+/*
+** Decode the varint in the first n bytes z[].  Write the integer value
+** into *pResult and return the number of bytes in the varint.
+**
+** If the decode fails because there are not enough bytes in z[] then
+** return 0;
+*/
+static int lsm1GetVarint64(
+  const unsigned char *z,
+  int n,
+  sqlite3_uint64 *pResult
+){
+  unsigned int x;
+  if( n<1 ) return 0;
+  if( z[0]<=240 ){
+    *pResult = z[0];
+    return 1;
+  }
+  if( z[0]<=248 ){
+    if( n<2 ) return 0;
+    *pResult = (z[0]-241)*256 + z[1] + 240;
+    return 2;
+  }
+  if( n<z[0]-246 ) return 0;
+  if( z[0]==249 ){
+    *pResult = 2288 + 256*z[1] + z[2];
+    return 3;
+  }
+  if( z[0]==250 ){
+    *pResult = (z[1]<<16) + (z[2]<<8) + z[3];
+    return 4;
+  }
+  x = (z[1]<<24) + (z[2]<<16) + (z[3]<<8) + z[4];
+  if( z[0]==251 ){
+    *pResult = x;
+    return 5;
+  }
+  if( z[0]==252 ){
+    *pResult = (((sqlite3_uint64)x)<<8) + z[5];
+    return 6;
+  }
+  if( z[0]==253 ){
+    *pResult = (((sqlite3_uint64)x)<<16) + (z[5]<<8) + z[6];
+    return 7;
+  }
+  if( z[0]==254 ){
+    *pResult = (((sqlite3_uint64)x)<<24) + (z[5]<<16) + (z[6]<<8) + z[7];
+    return 8;
+  }
+  *pResult = (((sqlite3_uint64)x)<<32) +
+               (0xffffffff & ((z[5]<<24) + (z[6]<<16) + (z[7]<<8) + z[8]));
+  return 9;
+}
+
 /*
 ** Generate a key encoding for pValue such that all keys compare in
 ** lexicographical order.  Return an SQLite error code or SQLITE_OK.
@@ -218,6 +347,7 @@ static int lsm1EncodeKey(
   int eType = sqlite3_value_type(pValue);
   *ppKey = 0;
   *pnKey = 0;
+  assert( nSpace>=32 );
   switch( eType ){
     default: {
       return SQLITE_ERROR;  /* We cannot handle NULL keys */
@@ -244,6 +374,21 @@ static int lsm1EncodeKey(
       *pnKey = nVal+1;
       break;
     }
+    case SQLITE_INTEGER: {
+      sqlite3_int64 iVal = sqlite3_value_int64(pValue);
+      sqlite3_uint64 uVal;
+      if( iVal<0 ){
+        if( iVal==0xffffffffffffffffLL ) return SQLITE_ERROR;
+        uVal = -iVal;
+        eType = LSM1_TYPE_NEGATIVE;
+      }else{
+        uVal = iVal;
+        eType = LSM1_TYPE_POSITIVE;
+      }
+      pSpace[0] = eType;
+      *ppKey = pSpace;
+      *pnKey = 1 + lsm1PutVarint64(&pSpace[1], uVal);
+    }
   }
   return SQLITE_OK;
 }
@@ -279,7 +424,17 @@ static int lsm1Column(
         }else if( pVal[0]==LSM1_TYPE_TEXT ){
           sqlite3_result_text(ctx, (const char*)&pVal[1],nVal-1,
                               SQLITE_TRANSIENT);
-        }
+        }else if( nVal>=2 && nVal<=9 &&
+           (pVal[0]==LSM1_TYPE_POSITIVE || pVal[0]==LSM1_TYPE_NEGATIVE)
+        ){
+          sqlite3_uint64 uVal = 0;
+          lsm1GetVarint64(pVal+1, nVal-1, &uVal);
+          if( pVal[0]==LSM1_TYPE_NEGATIVE ){
+            sqlite3_result_int64(ctx, -(sqlite3_int64)uVal);
+          }else{
+            sqlite3_result_int64(ctx, (sqlite3_int64)uVal); 
+          }
+        }         
       }
       break;
     }