SQLite

Check-in [c6fa01c28e]
Login

Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Add a hack to debug out a description of the WHERE clause of a SELECT (or other) statement. Use this in script tool/schemalint.tcl to automatically recommend indexes that might speed up specific queries.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | schemalint
Files: files | file ages | folders
SHA1: c6fa01c28ef7ceea2963a92dfffe62eed451b05c
User & Date: dan 2015-11-11 18:08:58.267
Context
2015-11-20
20:55
Add support for ORDER BY clauses to schemalint.tcl. (check-in: 93bdf70e85 user: dan tags: schemalint)
2015-11-11
18:08
Add a hack to debug out a description of the WHERE clause of a SELECT (or other) statement. Use this in script tool/schemalint.tcl to automatically recommend indexes that might speed up specific queries. (check-in: c6fa01c28e user: dan tags: schemalint)
15:28
Improvements to the SQLITE_CONFIG_PAGECACHE documentation. Enhance the command-line shell to be able to take advantage of the full range of SQLITE_CONFIG_PAGECACHE capabilities, such as setting pMem==NULL and N<0. (check-in: 2518d5c971 user: drh tags: trunk)
Changes
Unified Diff Ignore Whitespace Patch
Changes to src/where.c.
3897
3898
3899
3900
3901
3902
3903





















































































































































3904
3905
3906
3907
3908
3909
3910
    pLoop->cId = '0';
#endif
    return 1;
  }
  return 0;
}






















































































































































/*
** Generate the beginning of the loop used for WHERE clause processing.
** The return value is a pointer to an opaque structure that contains
** information needed to terminate the loop.  Later, the calling routine
** should invoke sqlite3WhereEnd() with the return value of this function
** in order to complete the WHERE clause processing.
**







>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>







3897
3898
3899
3900
3901
3902
3903
3904
3905
3906
3907
3908
3909
3910
3911
3912
3913
3914
3915
3916
3917
3918
3919
3920
3921
3922
3923
3924
3925
3926
3927
3928
3929
3930
3931
3932
3933
3934
3935
3936
3937
3938
3939
3940
3941
3942
3943
3944
3945
3946
3947
3948
3949
3950
3951
3952
3953
3954
3955
3956
3957
3958
3959
3960
3961
3962
3963
3964
3965
3966
3967
3968
3969
3970
3971
3972
3973
3974
3975
3976
3977
3978
3979
3980
3981
3982
3983
3984
3985
3986
3987
3988
3989
3990
3991
3992
3993
3994
3995
3996
3997
3998
3999
4000
4001
4002
4003
4004
4005
4006
4007
4008
4009
4010
4011
4012
4013
4014
4015
4016
4017
4018
4019
4020
4021
4022
4023
4024
4025
4026
4027
4028
4029
4030
4031
4032
4033
4034
4035
4036
4037
4038
4039
4040
4041
4042
4043
4044
4045
4046
4047
4048
4049
4050
4051
4052
4053
4054
4055
4056
4057
4058
4059
    pLoop->cId = '0';
#endif
    return 1;
  }
  return 0;
}

#ifdef SQLITE_SCHEMA_LINT
static char *whereAppendPrintf(sqlite3 *db, const char *zFmt, ...){
  va_list ap;
  char *zRes = 0;
  va_start(ap, zFmt);
  zRes = sqlite3_vmprintf(zFmt, ap);
  if( zRes==0 ){
    db->mallocFailed = 1;
  }else if( db->mallocFailed ){
    sqlite3_free(zRes);
    zRes = 0;
  }
  va_end(ap);
  return zRes;
}

/*
** Append a representation of term pTerm to the string in zIn and return
** the result. Or, if an OOM occurs, free zIn and return a NULL pointer.
*/
static char *whereAppendSingleTerm(
  Parse *pParse,
  Table *pTab,
  int bOr,
  char *zIn,
  WhereTerm *pTerm
){
  char *zBuf;
  sqlite3 *db = pParse->db;
  Expr *pX = pTerm->pExpr;
  CollSeq *pColl;
  const char *zOp = 0;

  if( pTerm->eOperator & (WO_IS|WO_EQ|WO_IN) ){
    zOp = "eq";
  }else if( pTerm->eOperator & (WO_LT|WO_LE|WO_GE|WO_GT) ){
    zOp = "range";
  }
  pColl = sqlite3BinaryCompareCollSeq(pParse, pX->pLeft, pX->pRight);

  if( zOp ){
    const char *zFmt = bOr ? "%z{{%s %s %s %lld}}" : "%z{%s %s %s %lld}";
    zBuf = whereAppendPrintf(db, zFmt, zIn, 
        zOp, pTab->aCol[pTerm->u.leftColumn].zName, 
        (pColl ? pColl->zName : "BINARY"),
        pTerm->prereqRight
    );
  }else{
    zBuf = zIn;
  }

  return zBuf;
}

static char *whereTraceWC(
  Parse *pParse, 
  struct SrcList_item *pItem,
  char *zIn,
  WhereClause *pWC
){
  sqlite3 *db = pParse->db;
  Table *pTab = pItem->pTab;
  char *zBuf = zIn;
  int iCol;
  int ii;
  int bFirst = 1;

  /* List of WO_SINGLE constraints */
  for(iCol=0; iCol<pTab->nCol; iCol++){
    int opMask = WO_SINGLE; 
    WhereScan scan;
    WhereTerm *pTerm;
    for(pTerm=whereScanInit(&scan, pWC, pItem->iCursor, iCol, opMask, 0);
        pTerm;
        pTerm=whereScanNext(&scan)
    ){
      assert( iCol==pTerm->u.leftColumn );
      if( bFirst==0 ) zBuf = whereAppendPrintf(db, "%z ", zBuf);
      zBuf = whereAppendSingleTerm(pParse, pTab, pWC->op==TK_OR, zBuf, pTerm);
      bFirst = 0;
    }
  }

  /* Add composite - (WO_OR|WO_AND) - constraints */
  for(ii=0; ii<pWC->nTerm; ii++){
    WhereTerm *pTerm = &pWC->a[ii];
    if( pTerm->eOperator & (WO_OR|WO_AND) ){
      const char *zFmt = ((pTerm->eOperator&WO_OR) ? "%z%s{or " : "%z%s{");
      zBuf = whereAppendPrintf(db, zFmt, zBuf, bFirst ? "" : " ");
      zBuf = whereTraceWC(pParse, pItem, zBuf, &pTerm->u.pOrInfo->wc);
      zBuf = whereAppendPrintf(db, "%z}", zBuf);
      bFirst = 0;
    }
  }

  return zBuf;
}

static void whereTraceBuilder(
  Parse *pParse,
  WhereLoopBuilder *p
){
  sqlite3 *db = pParse->db;
  if( db->xTrace ){
    WhereInfo *pWInfo = p->pWInfo;
    int nTablist = pWInfo->pTabList->nSrc;
    int ii;

    /* Loop through each element of the FROM clause. Ignore any sub-selects
    ** or views. Invoke the xTrace() callback once for each real table. */
    for(ii=0; ii<nTablist; ii++){
      char *zBuf = 0;
      int iCol;
      int nCol;
      Table *pTab;

      struct SrcList_item *pItem = &pWInfo->pTabList->a[ii];
      if( pItem->pSelect ) continue;
      pTab = pItem->pTab;
      nCol = pTab->nCol;

      /* Append the table name to the buffer. */
      zBuf = whereAppendPrintf(db, "%s", pTab->zName);

      /* Append the list of columns required to create a covering index */
      zBuf = whereAppendPrintf(db, "%z {cols", zBuf);
      if( 0==(pItem->colUsed & ((u64)1 << (sizeof(Bitmask)*8-1))) ){
        for(iCol=0; iCol<nCol; iCol++){
          if( iCol==(sizeof(Bitmask)*8-1) ) break;
          if( pItem->colUsed & ((u64)1 << iCol) ){
            zBuf = whereAppendPrintf(db, "%z %s", zBuf, pTab->aCol[iCol].zName);
          }
        }
      }
      zBuf = whereAppendPrintf(db, "%z} ", zBuf);

      /* Append the contents of WHERE clause */
      zBuf = whereTraceWC(pParse, pItem, zBuf, p->pWC);

      /* Pass the buffer to the xTrace() callback, then free it */
      db->xTrace(db->pTraceArg, zBuf);
      sqlite3DbFree(db, zBuf);
    }
  }
}
#else
# define whereTraceBuilder(x,y)
#endif

/*
** Generate the beginning of the loop used for WHERE clause processing.
** The return value is a pointer to an opaque structure that contains
** information needed to terminate the loop.  Later, the calling routine
** should invoke sqlite3WhereEnd() with the return value of this function
** in order to complete the WHERE clause processing.
**
4156
4157
4158
4159
4160
4161
4162



4163
4164
4165
4166
4167
4168
4169
  if( sqlite3WhereTrace & 0x100 ){ /* Display all terms of the WHERE clause */
    int i;
    for(i=0; i<sWLB.pWC->nTerm; i++){
      whereTermPrint(&sWLB.pWC->a[i], i);
    }
  }
#endif




  if( nTabList!=1 || whereShortCut(&sWLB)==0 ){
    rc = whereLoopAddAll(&sWLB);
    if( rc ) goto whereBeginError;
  
#ifdef WHERETRACE_ENABLED
    if( sqlite3WhereTrace ){    /* Display all of the WhereLoop objects */







>
>
>







4305
4306
4307
4308
4309
4310
4311
4312
4313
4314
4315
4316
4317
4318
4319
4320
4321
  if( sqlite3WhereTrace & 0x100 ){ /* Display all terms of the WHERE clause */
    int i;
    for(i=0; i<sWLB.pWC->nTerm; i++){
      whereTermPrint(&sWLB.pWC->a[i], i);
    }
  }
#endif

  /* Schema-lint xTrace callback */
  whereTraceBuilder(pParse, &sWLB);

  if( nTabList!=1 || whereShortCut(&sWLB)==0 ){
    rc = whereLoopAddAll(&sWLB);
    if( rc ) goto whereBeginError;
  
#ifdef WHERETRACE_ENABLED
    if( sqlite3WhereTrace ){    /* Display all of the WhereLoop objects */
Added test/schemalint.test.


































































































































































>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81


set testdir [file dirname $argv0]
source $testdir/tester.tcl
set testprefix schemalint

proc xTrace {zMsg} {
  lappend ::trace_out $zMsg
}
db trace xTrace

proc do_trace_test {tn sql res} {
  uplevel [list do_test $tn [subst -nocommands {
    set ::trace_out [list]
    set stmt [sqlite3_prepare db "$sql" -1 x]
    sqlite3_finalize [set stmt]
    set ::trace_out
  }] [list {*}$res]]
}


do_execsql_test 1.0 {
  CREATE TABLE t1(a, b, c);
  CREATE TABLE t2(x, y, z);
}

do_trace_test 1.1 {
  SELECT b, c, y, z FROM t1, t2 WHERE c=? AND z=?
} {
  {t1 {cols b c} {eq c BINARY 0}}
  {t2 {cols y z} {eq z BINARY 0}}
}

do_trace_test 1.2 {
  SELECT a FROM t1 WHERE b>10
} {
  {t1 {cols a b} {range b BINARY 0}}
}

do_trace_test 1.3 {
  SELECT b FROM t1 WHERE b IN (10, 20, 30)
} {
  {t1 {cols b} {eq b BINARY 0}}
}

do_trace_test 1.4 {
  SELECT * FROM t1, t2 WHERE x=a
} {
  {t1 {cols a b c} {eq a BINARY 2}} 
  {t2 {cols x y z} {eq x BINARY 1}}
}

do_trace_test 1.5 {
  SELECT * FROM t1 WHERE a IN (1, 2, 3)
} {
  {t1 {cols a b c} {eq a BINARY 0}}
}

#-----------------------------------------------------------------------
# Cases involving OR clauses in the WHERE clause.
#
do_trace_test 2.1 {
  SELECT * FROM t1 WHERE a=? OR b=?
} {
  {t1 {cols a b c} {or {{eq a BINARY 0}} {{eq b BINARY 0}}}}
}

do_trace_test 2.2 {
  SELECT * FROM t1 WHERE a=? OR (b=? AND c=?)
} {
  {t1 {cols a b c} {or {{eq a BINARY 0}} {{eq b BINARY 0} {eq c BINARY 0}}}}
}

do_trace_test 2.3 {
  SELECT * FROM t1 WHERE (a=? AND b=?) OR c=?
} {
  {t1 {cols a b c} {or {{eq c BINARY 0}} {{eq a BINARY 0} {eq b BINARY 0}}}}
}

finish_test

Added tool/schemalint.tcl.






































































































































































































































































































































































































































































































































>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259



set ::G(lSelect)  [list]           ;# List of SELECT statements to analyze
set ::G(database) ""               ;# Name of database or SQL schema file
set ::G(trace)    [list]           ;# List of data from xTrace()
set ::G(verbose)  0                ;# True if -verbose option was passed 

proc usage {} {
  puts stderr "Usage: $::argv0 ?SWITCHES? DATABASE/SCHEMA"
  puts stderr "  Switches are:"
  puts stderr "  -select SQL     (recommend indexes for SQL statement)"
  puts stderr "  -verbose        (increase verbosity of output)"
  puts stderr ""
  exit
}

proc process_cmdline_args {argv} {
  global G
  set nArg [llength $argv]
  set G(database) [lindex $argv end]

  for {set i 0} {$i < [llength $argv]-1} {incr i} {
    set k [lindex $argv $i]
    switch -- $k {
      -select {
        incr i
        if {$i>=[llength $argv]-1} usage
        lappend G(lSelect) [lindex $argv $i]
      }
      -verbose {
        set G(verbose) 1
      }
      default {
        usage
      }
    }
  }
}

proc open_database {} {
  global G
  sqlite3 db ""

  # Check if the "database" file is really an SQLite database. If so, copy
  # it into the temp db just opened. Otherwise, assume that it is an SQL
  # schema and execute it directly.
  set fd [open $G(database)]
  set hdr [read $fd 16]
  if {$hdr == "SQLite format 3\000"} {
    close $fd
    sqlite3 db2 $G(database)
    sqlite3_backup B db main db2 main
    B step 2000000000
    set rc [B finish]
    db2 close
    if {$rc != "SQLITE_OK"} { error "Failed to load database $G(database)" }
  } else {
    append hdr [read $fd]
    db eval $hdr
    close $fd
  }
}

proc analyze_selects {} {
  global G
  set G(trace) ""

  # Collect a line of xTrace output for each loop in the set of SELECT
  # statements.
  proc xTrace {zMsg} { lappend ::G(trace) $zMsg }
  db trace "lappend ::G(trace)"
  foreach s $G(lSelect) {
    set stmt [sqlite3_prepare_v2 db $s -1 dummy]
    set rc [sqlite3_finalize $stmt]
    if {$rc!="SQLITE_OK"} {
      error "Failed to compile SQL: [sqlite3_errmsg db]"
    }
  }

  db trace ""
  if {$G(verbose)} {
    foreach t $G(trace) { puts "trace: $t" }
  }

  # puts $G(trace)
}

# The argument is a list of the form:
#
#    key1 {value1.1 value1.2} key2 {value2.1 value 2.2...}
#
# Values lists may be of any length greater than zero. This function returns
# a list of lists created by pivoting on each values list. i.e. a list
# consisting of the elements:
#
#   {{key1 value1.1} {key2 value2.1}}
#   {{key1 value1.2} {key2 value2.1}}
#   {{key1 value1.1} {key2 value2.2}}
#   {{key1 value1.2} {key2 value2.2}}
#
proc expand_eq_list {L} {
  set ll [list {}]
  for {set i 0} {$i < [llength $L]} {incr i 2} {
    set key [lindex $L $i]
    set new [list]
    foreach piv [lindex $L $i+1] {
      foreach l $ll {
        lappend new [concat $l [list [list $key $piv]]]
      }
    }
    set ll $new
  }

  return $ll
}

proc eqset_to_index {tname eqset {range {}}} {
  global G
  set lCols [list]
  set idxname $tname
  foreach e [concat [lsort $eqset] [list $range]] {
    if {[llength $e]==0} continue
    foreach {c collate} $e {}
    lappend lCols "$c collate $collate"
    append idxname "_$c"
    if {[string compare -nocase binary $collate]!=0} {
      append idxname [string tolower $collate]
    }
  }

  set create_index "CREATE INDEX $idxname ON ${tname}("
  append create_index [join $lCols ", "]
  append create_index ");"

  set G(trial.$idxname) $create_index
}

proc expand_or_cons {L} {
  set lRet [list [list]]
  foreach elem $L {
    set type [lindex $elem 0]
    if {$type=="eq" || $type=="range"} {
      set lNew [list]
      for {set i 0} {$i < [llength $lRet]} {incr i} {
        lappend lNew [concat [lindex $lRet $i] [list $elem]]
      }
      set lRet $lNew
    } elseif {$type=="or"} {
      set lNew [list]
      foreach branch [lrange $elem 1 end] {
        foreach b [expand_or_cons $branch] {
          for {set i 0} {$i < [llength $lRet]} {incr i} {
            lappend lNew [concat [lindex $lRet $i] $b]
          }
        }
      }
      set lRet $lNew
    } 
  }
  return $lRet
}

proc find_trial_indexes {} {
  global G
  foreach t $G(trace) {
    set tname [lindex $t 0]
    catch { array unset mask }

    foreach lCons [expand_or_cons [lrange $t 2 end]] {
      set constraints [list]

      foreach a $lCons {
        set type [lindex $a 0]
        if {$type=="eq" || $type=="range"} {
          set m [lindex $a 3]
          foreach k [array names mask] { set mask([expr ($k & $m)]) 1 }
          set mask($m) 1
          lappend constraints $a
        }
      }

      foreach k [array names mask] {
        catch {array unset eq}
        foreach a $constraints {
          foreach {type col collate m} $a {
            if {($m & $k)==$m} {
              if {$type=="eq"} {
                lappend eq($col) $collate
              } else {
                set range($col.$collate) 1
              }
            }
          }
        }

        #puts "mask=$k eq=[array get eq] range=[array get range]"
        
        set ranges [array names range]
        foreach eqset [expand_eq_list [array get eq]] {
          if {[llength $ranges]==0} {
            eqset_to_index $tname $eqset
          } else {
            foreach r $ranges {
              set bSeen 0
              foreach {c collate} [split $r .] {}
              foreach e $eqset {
                if {[lindex $e 0] == $c} {
                  set bSeen 1
                  break
                }
              }
              if {$bSeen} {
                eqset_to_index $tname $eqset
              } else {
                eqset_to_index $tname $eqset [list $c $collate]
              }
            }
          }
        }
      }
    }
  }

  if {$G(verbose)} {
    foreach k [array names G trial.*] { puts "index: $G($k)" }
  }
}

proc run_trials {} {
  global G

  foreach k [array names G trial.*] {
    set idxname [lindex [split $k .] 1]
    db eval $G($k)
    set pgno [db one {SELECT rootpage FROM sqlite_master WHERE name = $idxname}]
    set IDX($pgno) $idxname
  }
  db eval ANALYZE

  catch { array unset used }
  foreach s $G(lSelect) {
    db eval "EXPLAIN $s" x {
      if {($x(opcode)=="OpenRead" || $x(opcode)=="ReopenIdx")} {
        if {[info exists IDX($x(p2))]} { set used($IDX($x(p2))) 1 }
      }
    }
    foreach idx [array names used] {
      puts $G(trial.$idx)
    }
  }
}

process_cmdline_args $argv
open_database
analyze_selects
find_trial_indexes
run_trials