diff --git a/catalog/statistics.go b/catalog/statistics.go
index f25cbf91..7694ad62 100644
--- a/catalog/statistics.go
+++ b/catalog/statistics.go
@@ -1,6 +1,7 @@
 package catalog
 
 import (
+	"errors"
 	"github.com/ryogrid/SamehadaDB/common"
 	"github.com/ryogrid/SamehadaDB/execution/expression"
 	"github.com/ryogrid/SamehadaDB/samehada/samehada_util"
@@ -10,6 +11,8 @@ import (
 	"math"
 )
 
+var AbortedError error = errors.New("aborted")
+
 type distinctCounter struct {
 	max      *types.Value
 	min      *types.Value
@@ -159,9 +162,9 @@ func (cs *columnStats) EstimateCount(from *types.Value, to *types.Value) float64
 		to = retValAccordingToCompareResult(to.CompareLessThan(*cs.max), to, cs.max)
 		tmpVal := to.Sub(from)
 		if cs.colType == types.Integer {
-			return float64(tmpVal.ToInteger()+1) * float64(cs.count) / float64(cs.distinct)
+			return float64(tmpVal.ToInteger()+1) * float64(cs.count) / guardNotZeroReturn(float64(cs.distinct))
 		} else { // Float
-			return float64(tmpVal.ToFloat()+1) * float64(cs.count) / float64(cs.distinct)
+			return float64(tmpVal.ToFloat()+1) * float64(cs.count) / guardNotZeroReturn(float64(cs.distinct))
 		}
 	} else if cs.colType == types.Varchar {
 		if to.CompareLessThanOrEqual(*from) {
@@ -207,6 +210,9 @@ func (ts *TableStatistics) Update(target *TableMetadata, txn *access.Transaction
 	}
 
 	for t := it.Current(); !it.End(); t = it.Next() {
+		if txn.GetState() == access.ABORTED {
+			return AbortedError
+		}
 		for ii := 0; ii < len(ts.colStats); ii++ {
 			distCounters[ii].Add(samehada_util.GetPonterOfValue(t.GetValue(schema_, uint32(ii))))
 		}
@@ -237,6 +243,13 @@ func isBinaryExp(exp expression.Expression) bool {
 	return exp.GetType() == expression.EXPRESSION_TYPE_COMPARISON || exp.GetType() == expression.EXPRESSION_TYPE_LOGICAL_OP
 }
 
+func guardNotZeroReturn(val float64) float64 {
+	if val == 0 {
+		return math.MaxFloat64
+	}
+	return val
+}
+
 // Returns estimated inverted selection ratio if the `sc` is selected by
 // `predicate`. If the predicate selects rows to 1 / x, returns x.
 // Returning 1 means no selection (pass through).
@@ -253,21 +266,21 @@ func (ts *TableStatistics) ReductionFactor(sc *schema.Schema, predicate expressi
 				samehada_util.SHAssert(colIndexLeft >= 0 && int(colIndexLeft) < len(ts.colStats), "invalid column index (Left)")
 				colIndexRight := rcv.GetColIndex()
 				samehada_util.SHAssert(colIndexRight >= 0 && int(colIndexRight) < len(ts.colStats), "invalid column index (Right)")
-				return math.Min(float64(ts.colStats[colIndexLeft].Distinct()), float64(ts.colStats[colIndexRight].Distinct()))
+				return guardNotZeroReturn(math.Min(float64(ts.colStats[colIndexLeft].Distinct()), float64(ts.colStats[colIndexRight].Distinct())))
 			}
 			if boCmp.GetChildAt(0).GetType() == expression.EXPRESSION_TYPE_COLUMN_VALUE {
 				lcv := boCmp.GetChildAt(0).(*expression.ColumnValue)
 				colIndexLeft := lcv.GetColIndex()
 				samehada_util.SHAssert(colIndexLeft >= 0 && int(colIndexLeft) < len(ts.colStats), "invalid column index (Left)")
 				// return static_cast<double>(stats_[offset_left].distinct());
-				return float64(ts.colStats[colIndexLeft].Distinct())
+				return guardNotZeroReturn(float64(ts.colStats[colIndexLeft].Distinct()))
 			}
 			if boCmp.GetChildAt(1).GetType() == expression.EXPRESSION_TYPE_COLUMN_VALUE {
 				rcv := boCmp.GetChildAt(1).(*expression.ColumnValue)
 				colIndexRight := rcv.GetColIndex()
 				samehada_util.SHAssert(colIndexRight >= 0 && int(colIndexRight) < len(ts.colStats), "invalid column index (Right)")
 				// return static_cast<double>(stats_[offset_right].distinct());
-				return float64(ts.colStats[colIndexRight].Distinct())
+				return guardNotZeroReturn(float64(ts.colStats[colIndexRight].Distinct()))
 			}
 			if boCmp.GetChildAt(0).GetType() == expression.EXPRESSION_TYPE_CONSTANT_VALUE &&
 				boCmp.GetChildAt(1).GetType() == expression.EXPRESSION_TYPE_CONSTANT_VALUE {
@@ -284,11 +297,11 @@ func (ts *TableStatistics) ReductionFactor(sc *schema.Schema, predicate expressi
 		boLogi, okLogi := predicate.(*expression.LogicalOp)
 		if okLogi {
 			if boLogi.GetLogicalOpType() == expression.AND {
-				return ts.ReductionFactor(sc, boLogi.GetChildAt(0)) * ts.ReductionFactor(sc, boLogi.GetChildAt(1))
+				return guardNotZeroReturn(ts.ReductionFactor(sc, boLogi.GetChildAt(0)) * ts.ReductionFactor(sc, boLogi.GetChildAt(1)))
 			}
 			if boLogi.GetLogicalOpType() == expression.OR {
 				// TODO: what should be returned?
-				return ts.ReductionFactor(sc, boLogi.GetChildAt(0)) * ts.ReductionFactor(sc, boLogi.GetChildAt(1))
+				return guardNotZeroReturn(ts.ReductionFactor(sc, boLogi.GetChildAt(0)) * ts.ReductionFactor(sc, boLogi.GetChildAt(1)))
 			}
 		}
 	}
@@ -315,6 +328,9 @@ func (ts *TableStatistics) EstimateCount(col_idx int32, from *types.Value, to *t
 func (ts *TableStatistics) TransformBy(col_idx int32, from *types.Value, to *types.Value) *TableStatistics {
 	multiplier := ts.EstimateCount(col_idx, from, to)
 	for _, st := range ts.colStats {
+		if st.Count() == 0 {
+			continue
+		}
 		st.Multiply(multiplier / float64(st.Count()))
 	}
 
diff --git a/catalog/table_catalog.go b/catalog/table_catalog.go
index e750d685..05601cf6 100644
--- a/catalog/table_catalog.go
+++ b/catalog/table_catalog.go
@@ -65,26 +65,26 @@ func RecoveryCatalogFromCatalogPage(bpm *buffer.BufferPoolManager, log_manager *
 	tableIds := make(map[uint32]*TableMetadata)
 	tableNames := make(map[string]*TableMetadata)
 
-	for tuple := tableCatalogHeapIt.Current(); !tableCatalogHeapIt.End(); tuple = tableCatalogHeapIt.Next() {
-		oid := tuple.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("oid")).ToInteger()
-		name := tuple.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("name")).ToVarchar()
-		firstPage := tuple.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("first_page")).ToInteger()
+	for tuple_outer := tableCatalogHeapIt.Current(); !tableCatalogHeapIt.End(); tuple_outer = tableCatalogHeapIt.Next() {
+		oid := tuple_outer.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("oid")).ToInteger()
+		name := tuple_outer.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("name")).ToVarchar()
+		firstPage := tuple_outer.GetValue(TableCatalogSchema(), TableCatalogSchema().GetColIndex("first_page")).ToInteger()
 
 		columns := []*column.Column{}
 		columnsCatalogHeapIt := access.InitTableHeap(bpm, ColumnsCatalogPageId, log_manager, lock_manager).Iterator(txn)
-		for tuple := columnsCatalogHeapIt.Current(); !columnsCatalogHeapIt.End(); tuple = columnsCatalogHeapIt.Next() {
-			tableOid := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("table_oid")).ToInteger()
+		for tuple_inner := columnsCatalogHeapIt.Current(); !columnsCatalogHeapIt.End(); tuple_inner = columnsCatalogHeapIt.Next() {
+			tableOid := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("table_oid")).ToInteger()
 			if tableOid != oid {
 				continue
 			}
-			columnType := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("type")).ToInteger()
-			columnName := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("name")).ToVarchar()
-			fixedLength := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("fixed_length")).ToInteger()
-			variableLength := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("variable_length")).ToInteger()
-			columnOffset := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("offset")).ToInteger()
-			hasIndex := Int32toBool(tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("has_index")).ToInteger())
-			indexKind := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("index_kind")).ToInteger()
-			indexHeaderPageId := tuple.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("index_header_page_id")).ToInteger()
+			columnType := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("type")).ToInteger()
+			columnName := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("name")).ToVarchar()
+			fixedLength := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("fixed_length")).ToInteger()
+			variableLength := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("variable_length")).ToInteger()
+			columnOffset := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("offset")).ToInteger()
+			hasIndex := Int32toBool(tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("has_index")).ToInteger())
+			indexKind := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("index_kind")).ToInteger()
+			indexHeaderPageId := tuple_inner.GetValue(ColumnsCatalogSchema(), ColumnsCatalogSchema().GetColIndex("index_header_page_id")).ToInteger()
 
 			column_ := column.NewColumn(columnName, types.TypeID(columnType), false, index_constants.INDEX_KIND_INVALID, types.PageID(indexHeaderPageId), nil)
 			column_.SetFixedLength(uint32(fixedLength))
diff --git a/catalog/table_metadata.go b/catalog/table_metadata.go
index 4e258ca7..fdfd1368 100644
--- a/catalog/table_metadata.go
+++ b/catalog/table_metadata.go
@@ -49,13 +49,19 @@ func NewTableMetadata(schema *schema.Schema, name string, table *access.TableHea
 			case index_constants.INDEX_KIND_UNIQ_SKIP_LIST:
 				// currently, SkipList Index always use new pages even if relaunch
 				im := index.NewIndexMetadata(column_.GetColumnName()+"_index", name, schema, []uint32{uint32(idx)})
+				// TODO: (SDB) need to add index headae ID argument like HashIndex (NewTableMetadata)
 				slIdx := index.NewUniqSkipListIndex(im, table.GetBufferPoolManager(), uint32(idx))
+
 				indexes = append(indexes, slIdx)
+				//column_.SetIndexHeaderPageId(slIdx.GetHeaderPageId())
 			case index_constants.INDEX_KIND_SKIP_LIST:
 				// currently, SkipList Index always use new pages even if relaunch
 				im := index.NewIndexMetadata(column_.GetColumnName()+"_index", name, schema, []uint32{uint32(idx)})
+				// TODO: (SDB) need to add index headae ID argument like HashIndex (NewTableMetadata)
 				slIdx := index.NewSkipListIndex(im, table.GetBufferPoolManager(), uint32(idx))
+
 				indexes = append(indexes, slIdx)
+				//column_.SetIndexHeaderPageId(slIdx.GetHeaderPageId())
 			default:
 				panic("illegal index kind!")
 			}
diff --git a/concurrency/checkpoint_manager.go b/concurrency/checkpoint_manager.go
index 717bc1cc..ea4664d1 100644
--- a/concurrency/checkpoint_manager.go
+++ b/concurrency/checkpoint_manager.go
@@ -29,9 +29,7 @@ func NewCheckpointManager(
 func (checkpoint_manager *CheckpointManager) StartCheckpointTh() {
 	go func() {
 		for checkpoint_manager.IsCheckpointActive() {
-			// TODO: (SDB) for debugging
 			time.Sleep(time.Second * 30)
-			//time.Sleep(time.Minute * 5)
 			if !checkpoint_manager.IsCheckpointActive() {
 				break
 			}
diff --git a/concurrency/statistics_updater.go b/concurrency/statistics_updater.go
new file mode 100644
index 00000000..751af0a8
--- /dev/null
+++ b/concurrency/statistics_updater.go
@@ -0,0 +1,69 @@
+package concurrency
+
+import (
+	"fmt"
+	"github.com/ryogrid/SamehadaDB/catalog"
+	"github.com/ryogrid/SamehadaDB/storage/access"
+	"time"
+)
+
+type StatisticsUpdater struct {
+	transaction_manager *access.TransactionManager
+	c                   *catalog.Catalog
+	// updater thread works when this flag is true
+	isUpdaterActive bool
+}
+
+func NewStatisticsUpdater(
+	transaction_manager *access.TransactionManager, c *catalog.Catalog) *StatisticsUpdater {
+
+	return &StatisticsUpdater{transaction_manager, c, true}
+}
+
+func (updater *StatisticsUpdater) StartStaticsUpdaterTh() {
+	go func() {
+		for updater.IsUpdaterActive() {
+			if !updater.IsUpdaterActive() {
+				break
+			}
+			fmt.Println("StatisticsUpdaterTh: start updating.")
+			updater.BeginStatsUpdate()
+			updater.UpdateAllTablesStatistics()
+			updater.EndStatsUpdate()
+			fmt.Println("StatisticsUpdaterTh: finish updating.")
+			time.Sleep(time.Second * 10)
+		}
+	}()
+}
+
+func (updater *StatisticsUpdater) UpdateAllTablesStatistics() {
+	txn := updater.transaction_manager.Begin(nil)
+	defer updater.transaction_manager.Commit(updater.c, txn)
+
+	tables := updater.c.GetAllTables()
+	for _, table_ := range tables {
+		stat := table_.GetStatistics()
+		err := stat.Update(table_, txn)
+		if err != nil {
+			// note: already updated table's statistics are not rollbacked
+			//       in current impl
+			return
+		}
+	}
+}
+
+func (updater *StatisticsUpdater) BeginStatsUpdate() {
+	// do nothing
+}
+
+func (updater *StatisticsUpdater) EndStatsUpdate() {
+	// do nothing
+}
+
+func (updater *StatisticsUpdater) StopStatsUpdateTh() {
+	updater.isUpdaterActive = false
+}
+
+func (updater *StatisticsUpdater) IsUpdaterActive() bool {
+	return updater.isUpdaterActive
+}
diff --git a/execution/plans/range_scan_with_index.go b/execution/plans/range_scan_with_index.go
index fce8a9d9..7bd566cd 100644
--- a/execution/plans/range_scan_with_index.go
+++ b/execution/plans/range_scan_with_index.go
@@ -27,7 +27,13 @@ type RangeScanWithIndexPlanNode struct {
 
 func NewRangeScanWithIndexPlanNode(c *catalog.Catalog, schema *schema.Schema, tableOID uint32, colIdx int32, predicate expression.Expression, startRange *types.Value, endRange *types.Value) Plan {
 	tm := c.GetTableByOID(tableOID)
-	return &RangeScanWithIndexPlanNode{&AbstractPlanNode{schema, nil}, predicate, tableOID, colIdx, startRange, endRange, tm.GetStatistics().GetDeepCopy()}
+	ret := &RangeScanWithIndexPlanNode{&AbstractPlanNode{schema, nil}, predicate, tableOID, colIdx, startRange, endRange, tm.GetStatistics().GetDeepCopy()}
+	if startRange != nil && endRange != nil {
+		// when caller is optimizer, both startRange and endRange are not nil
+		// when not optimizer, this call is not needed
+		ret.stats_ = ret.stats_.TransformBy(colIdx, startRange, endRange)
+	}
+	return ret
 }
 
 func (p *RangeScanWithIndexPlanNode) GetPredicate() expression.Expression {
diff --git a/go.mod b/go.mod
index 4a9bd958..8e4cd03c 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
 	github.com/pingcap/tidb v1.1.0-beta.0.20200630082100-328b6d0a955c
 	github.com/sasha-s/go-deadlock v0.2.0
 	github.com/spaolacci/murmur3 v1.1.0
+	golang.org/x/exp v0.0.0-20230905200255-921286631fa9
 )
 
 require (
@@ -31,10 +32,9 @@ require (
 	go.uber.org/atomic v1.6.0 // indirect
 	go.uber.org/multierr v1.5.0 // indirect
 	go.uber.org/zap v1.15.0 // indirect
-	golang.org/x/mod v0.5.1 // indirect
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20211124211545-fe61309f8881 // indirect
-	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/tools v0.1.7 // indirect
+	golang.org/x/net v0.15.0 // indirect
+	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sys v0.12.0 // indirect
+	golang.org/x/text v0.13.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 )
diff --git a/go.sum b/go.sum
index 253dd792..ebb4d8b6 100644
--- a/go.sum
+++ b/go.sum
@@ -576,16 +576,17 @@ golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191205180655-e7c4368fe9dd/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72 h1:+ELyKg6m8UBf0nPFSqD0mi7zUfwPyXo23HNjMnXPz7w=
 golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.13.0 h1:mvySKfSWJ+UKUii46M40LOvyWfN0s2U+46/jDd0e6Ck=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
 golang.org/x/exp v0.0.0-20190829153037-c13cbed26979/go.mod h1:86+5VVa7VpoJ4kLfm080zCjGlMRFzhUhsZKEZO7MGek=
 golang.org/x/exp v0.0.0-20191030013958-a1ab85dbe136/go.mod h1:JXzH8nQsPlswgeRAPE3MuO9GYsAcnJvJ4vnMwN/5qkY=
 golang.org/x/exp v0.0.0-20191129062945-2f5052295587/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
-golang.org/x/exp v0.0.0-20191227195350-da58074b4299 h1:zQpM52jfKHG6II1ISZY1ZcpygvuSFZpLwfluuF89XOg=
 golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
+golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
+golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/image v0.0.0-20200119044424-58c23975cae1/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
@@ -605,8 +606,7 @@ golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKG
 golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.5.1 h1:OJxoQ/rynoF0dcCdI7cLPktw/hR2cueqYfjm43oqK38=
-golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
+golang.org/x/mod v0.12.0 h1:rmsUpXtvNzj340zd98LZ4KntptpfRHwpFOHG188oHXc=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -632,7 +632,8 @@ golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
-golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d h1:20cMwl2fHAzkJMEA+8J4JgqBQcQGzbisXo31MIeenXI=
+golang.org/x/net v0.15.0 h1:ugBLEUaxABaB5AJqW9enI0ACdci2RUd4eP51NTBvuJ8=
+golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -644,8 +645,8 @@ golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
+golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -674,13 +675,13 @@ golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200122134326-e047566fdf82/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20211124211545-fe61309f8881 h1:TyHqChC80pFkXWraUUf6RuB5IqFdQieMLwwCJokV2pc=
-golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
+golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
-golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
-golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 h1:SvFZT6jyqRaOeXpc5h/JSfZenJ2O330aBsf7JfSUXmQ=
@@ -721,12 +722,10 @@ golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20200225230052-807dcd883420/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200325010219-a49f79bcc224/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
 golang.org/x/tools v0.0.0-20200325203130-f53864d0dba1/go.mod h1:Sl4aGygMT6LrqrWclx+PTx3U+LnKx/seiNR+3G19Ar8=
-golang.org/x/tools v0.1.7 h1:6j8CgantCy3yc8JGBqkDLMKWqZ0RDU2g1HVgacojGWQ=
-golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
+golang.org/x/tools v0.13.0 h1:Iey4qkscZuv0VvIt8E0neZjtPVQFSc870HQ448QgEmQ=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 google.golang.org/api v0.4.0/go.mod h1:8k5glujaEP+g9n7WNsDg8QP6cUVNI86fCNMcbazEtwE=
 google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E3M=
 google.golang.org/api v0.8.0/go.mod h1:o4eAsZoiT+ibD93RtjEohWalFOjRDx6CVaqeizhEnKg=
diff --git a/parser/parser_expressions.go b/parser/parser_expressions.go
index 2a194740..0d03318a 100644
--- a/parser/parser_expressions.go
+++ b/parser/parser_expressions.go
@@ -115,6 +115,10 @@ func (expr *BinaryOpExpression) GetDeepCopy() *BinaryOpExpression {
 	return ret
 }
 
+func (expr *BinaryOpExpression) AppendBinaryOpExpWithAnd(expr2 *BinaryOpExpression) *BinaryOpExpression {
+	return &BinaryOpExpression{expression.AND, -1, expr, expr2}
+}
+
 type SetExpression struct {
 	ColName_     *string
 	UpdateValue_ *types.Value
diff --git a/planner/optimizer/optimizer_test.go b/planner/optimizer/optimizer_test.go
index dfdfe244..1fa39a8a 100644
--- a/planner/optimizer/optimizer_test.go
+++ b/planner/optimizer/optimizer_test.go
@@ -328,6 +328,7 @@ func TestFindBestScans(t *testing.T) {
 
 	testAQuery := func(queryStr string, patternName string) {
 		queryInfo := parser.ProcessSQLStr(&queryStr)
+		queryInfo, _ = RewriteQueryInfo(c, queryInfo)
 		optimalPlans := NewSelingerOptimizer(queryInfo, c).findBestScans()
 		testingpkg.Assert(t, len(optimalPlans) == len(queryInfo.JoinTables_), "len(optimalPlans) != len(query.JoinTables_) ["+patternName+"]")
 		printOptimalPlans(patternName, queryStr, optimalPlans)
@@ -337,6 +338,8 @@ func TestFindBestScans(t *testing.T) {
 	testAQuery("select Sc1.c1, Sc1.c3 from Sc1 where Sc1.c2 = 'c2-32';", "IndexScan")
 	testAQuery("select Sc2.d1, Sc2.d2, Sc2.d3, Sc2.d4 from Sc2 where Sc2.d3 >= 'd3-3' and Sc2.d3 <= 'd3-5';", "IndexScanInclude(1)")
 	testAQuery("select Sc2.d1, Sc2.d2, Sc2.d3, Sc2.d4 from Sc2 where Sc2.d3 >= 'd3-3' and Sc2.d3 < 'd3-5';", "IndexScanInclude(2)")
+	testAQuery("select Sc4.c1 from Sc4 where Sc4.c3 >= 5 and Sc4.c3 <= 10;", "IndexScanInclude(3)")
+	testAQuery("select Sc4.c1 from Sc4 where Sc4.c3 >= 5 and Sc4.c3 < 10;", "IndexScanInclude(4)")
 	testAQuery("select Sc1.c2, Sc2.d1, Sc2.d3 from Sc1, Sc2 where Sc1.c1 = Sc2.d1;", "Join(HashJoin)")
 	testAQuery("select Sc3.e2, Sc4.c1, Sc4.c2 from Sc3, Sc4 where Sc3.e1 = Sc4.c3;", "Join(IndexJoin)")
 	testAQuery("select Sc3.e2, Sc4.c1, Sc4.c2 from Sc3, Sc4 where Sc3.e1 = Sc4.c3 and Sc4.c3 = 5;", "JoinAndIndexScan(HashJoin)")
@@ -395,6 +398,7 @@ func TestSimplePlanOptimization(t *testing.T) {
 
 	testAQuery := func(queryStr string, patternName string) {
 		queryInfo := parser.ProcessSQLStr(&queryStr)
+		queryInfo, _ = RewriteQueryInfo(c, queryInfo)
 
 		optimizer := NewSelingerOptimizer(queryInfo, c)
 		solution, err := optimizer.Optimize()
@@ -422,14 +426,21 @@ func TestSimplePlanOptimization(t *testing.T) {
 	testAQuery("select Sc1.c1, Sc1.c3 from Sc1 where Sc1.c2 = 'c2-32';", "IndexScan")
 	testAQuery("select Sc2.d1, Sc2.d2, Sc2.d3, Sc2.d4 from Sc2 where Sc2.d3 >= 'd3-3' and Sc2.d3 <= 'd3-5';", "IndexScanInclude(1)")
 	testAQuery("select Sc2.d1, Sc2.d2, Sc2.d3, Sc2.d4 from Sc2 where Sc2.d3 >= 'd3-3' and Sc2.d3 < 'd3-5';", "IndexScanInclude(2)")
+	testAQuery("select Sc4.c1 from Sc4 where Sc4.c3 >= 5 and Sc4.c3 <= 10;", "IndexScanInclude(3)")
+	testAQuery("select Sc4.c1 from Sc4 where Sc4.c3 >= 5 and Sc4.c3 < 10;", "IndexScanInclude(4)")
 	testAQuery("select Sc1.c2, Sc2.d1, Sc2.d3 from Sc1, Sc2 where Sc1.c1 = Sc2.d1;", "Join(HashJoin)")
-
 	testAQuery("select Sc3.e2, Sc4.c1, Sc4.c2 from Sc3, Sc4 where Sc3.e1 = Sc4.c3;", "Join(IndexJoin)")
 	testAQuery("select Sc3.e2, Sc4.c1, Sc4.c2 from Sc3, Sc4 where Sc3.e1 = Sc4.c3 and Sc4.c3 = 5;", "JoinAndIndexScan(HashJoin)")
-
 	testAQuery("select Sc1.c2, Sc2.d1, Sc3.e2 from Sc1, Sc2, Sc3 where Sc1.c1 = Sc2.d1 and Sc2.d1 = Sc3.e1;", "ThreeJoin(HashJoin)")
 	testAQuery("select Sc1.c1, Sc1.c2, Sc2.d1, Sc2.d2, Sc2.d3 from Sc1, Sc2 where Sc1.c1 = 2;", "JoinWhere(NestedLoopJoin)")
 	testAQuery("select Sc1.c1, Sc1.c2, Sc1.c3, Sc4.c1, Sc4.c2 from Sc1, Sc4 where Sc1.c1 = Sc4.c1 and Sc4.c1 = 2;", "SameNameColumn")
 
+	testAQuery("select Sc1.c1, Sc1.c2, Sc1.c3, Sc4.c1, Sc4.c2 from Sc1 join Sc4 on Sc1.c1 = Sc4.c1 where Sc4.c1 = 2;", "SameNameColumn(2)")
+	testAQuery("select * from Sc1 join Sc4 on Sc1.c1 = Sc4.c1 where Sc4.c1 = 2;", "SameNameColumn(3)")
+
+	testAQuery("select c1 from Sc1 where c1 = 2;", "NoTablePrefixScan(SequentialScan)")
+	testAQuery("delete from Sc1 where c1 = 2;", "NoTablePrefixScanDelete(SequentialScan)")
+	testAQuery("update Sc1 set c1 = 5 where c1 = 2;", "NoTablePrefixScanUpdate(SequentialScan)")
+
 	// "select * from Sc1, Sc4 where Sc1.c1 = Sc4.c1 and Sc4.c1 = 2;" // Asterisk (Not supported now...)
 }
diff --git a/planner/optimizer/selinger_optimizer.go b/planner/optimizer/selinger_optimizer.go
index 9bd27b32..2fb1e23f 100644
--- a/planner/optimizer/selinger_optimizer.go
+++ b/planner/optimizer/selinger_optimizer.go
@@ -1,6 +1,7 @@
 package optimizer
 
 import (
+	"errors"
 	mapset "github.com/deckarep/golang-set/v2"
 	stack "github.com/golang-collections/collections/stack"
 	pair "github.com/notEpsilon/go-pair"
@@ -13,8 +14,10 @@ import (
 	"github.com/ryogrid/SamehadaDB/storage/table/column"
 	"github.com/ryogrid/SamehadaDB/storage/table/schema"
 	"github.com/ryogrid/SamehadaDB/types"
+	"golang.org/x/exp/slices"
 	"math"
 	"sort"
+	"strings"
 )
 
 type CostAndPlan struct {
@@ -480,9 +483,11 @@ func (so *SelingerOptimizer) findBestJoin(optimalPlans map[string]CostAndPlan) p
 	optimalPlan, ok := optimalPlans[samehada_util.StrSetToString(samehada_util.MakeSet(so.qi.JoinTables_))]
 	samehada_util.SHAssert(ok, "plan which includes all tables is not found")
 
-	// Attach final projection and emit the result
 	solution := optimalPlan.plan
-	solution = plans.NewProjectionPlanNode(solution, parser.ConvParsedSelectionExprToSchema(so.c, so.qi.SelectFields_))
+	// Attach final projection and emit the result
+	if int(solution.OutputSchema().GetColumnCount()) > len(so.qi.SelectFields_) {
+		solution = plans.NewProjectionPlanNode(solution, parser.ConvParsedSelectionExprToSchema(so.c, so.qi.SelectFields_))
+	}
 
 	return solution
 }
@@ -497,3 +502,144 @@ func (so *SelingerOptimizer) Optimize() (plans.Plan, error) {
 	optimalPlans := so.findBestScans()
 	return so.findBestJoin(optimalPlans), nil
 }
+
+var CantTableIdentifedErr = errors.New("tableName can't be identified!")
+var InvalidColNameErr = errors.New("invalid column name!")
+
+func attachTableNameIfNeeded(tableMap map[string][]*string, tgtStr *string) (*string, error) {
+	if strings.Contains(*tgtStr, ".") {
+		return tgtStr, nil
+	}
+	if val, ok := tableMap[*tgtStr]; ok {
+		if len(val) == 1 {
+			tmpStr := *val[0] + "." + *tgtStr
+			return &tmpStr, nil
+		} else {
+			return nil, CantTableIdentifedErr
+		}
+	} else {
+		return nil, InvalidColNameErr
+	}
+}
+
+func rewiteColNameStrOfBinaryOpExp(tableMap map[string][]*string, exp interface{}) error {
+	switch casted := exp.(type) {
+	case *parser.BinaryOpExpression:
+		var err error
+		if str, ok := casted.Left_.(*string); ok {
+			casted.Left_, err = attachTableNameIfNeeded(tableMap, str)
+		} else {
+			err = rewiteColNameStrOfBinaryOpExp(tableMap, casted.Left_)
+		}
+		if err != nil {
+			return err
+		}
+		if str, ok := casted.Right_.(*string); ok {
+			casted.Right_, err = attachTableNameIfNeeded(nil, str)
+		} else {
+			err = rewiteColNameStrOfBinaryOpExp(tableMap, casted.Right_)
+		}
+		return err
+	case *types.Value:
+		// do nothing
+		return nil
+	case nil:
+		return nil
+	default:
+		panic("invalid type")
+	}
+}
+
+func CheckIncludesORInPredicate(exp interface{}) bool {
+	switch casted := exp.(type) {
+	case *parser.BinaryOpExpression:
+		if casted.LogicalOperationType_ == expression.OR {
+			return true
+		} else {
+			return CheckIncludesORInPredicate(casted.Left_) || CheckIncludesORInPredicate(casted.Right_)
+		}
+	default:
+		return false
+	}
+}
+
+func genTableMapAndColList(c *catalog.Catalog, qi *parser.QueryInfo) (map[string][]*string, []*parser.SelectFieldExpression) {
+	tableMap := make(map[string][]*string, 0)
+	colList := make([]*parser.SelectFieldExpression, 0)
+	for _, tableName := range qi.JoinTables_ {
+		tm := c.GetTableByName(*tableName)
+		colNum := tm.GetColumnNum()
+
+		for ii := 0; ii < int(colNum); ii++ {
+			col := tm.Schema().GetColumn(uint32(ii))
+			colName := col.GetColumnName()
+
+			if strings.Contains(colName, ".") {
+				splited := strings.Split(colName, ".")
+				colName = splited[1]
+				colList = append(colList, &parser.SelectFieldExpression{false, -1, &splited[0], &colName})
+			} else {
+				panic("invalid column name")
+			}
+			if val, ok := tableMap[colName]; ok {
+				tableMap[colName] = append(val, tableName)
+			} else {
+				tableMap[colName] = []*string{tableName}
+			}
+		}
+	}
+	return tableMap, colList
+}
+
+// add table name prefix to column name if column name doesn't have it
+// and attach predicate of ON clause to one of WHERE clause
+// ATTENTION: this func modifies *qi* arg
+func RewriteQueryInfo(c *catalog.Catalog, qi *parser.QueryInfo) (*parser.QueryInfo, error) {
+	tableMap, colList := genTableMapAndColList(c, qi)
+	// SelectFields_
+	// when SelectFields_[x].TableName_ is empty, set appropriate value
+	for _, sfield := range qi.SelectFields_ {
+		if sfield.TableName_ == nil && *sfield.ColName_ != "*" {
+			if val, ok := tableMap[*sfield.ColName_]; ok {
+				if len(val) == 1 {
+					sfield.TableName_ = val[0]
+				} else {
+					return nil, CantTableIdentifedErr
+				}
+			} else {
+				return nil, InvalidColNameErr
+			}
+		}
+	}
+	// replace asterisk to column names (one asterisk only)
+	for ii := 0; ii < len(qi.SelectFields_); ii++ {
+		if *qi.SelectFields_[ii].ColName_ == "*" {
+			qi.SelectFields_ = append(qi.SelectFields_[:ii], qi.SelectFields_[ii+1:]...)
+			qi.SelectFields_ = slices.Insert(qi.SelectFields_, ii, colList...)
+			break
+		}
+	}
+	// DELETE and UPDATE query is processed with optimizer asterisk specified SELECT query
+	if *qi.QueryType_ == parser.DELETE || *qi.QueryType_ == parser.UPDATE {
+		qi.SelectFields_ = colList
+	}
+
+	// OnExpressions_
+	err := rewiteColNameStrOfBinaryOpExp(tableMap, qi.OnExpressions_)
+	if err != nil {
+		return nil, err
+	}
+
+	// WhereExpression_
+	err = rewiteColNameStrOfBinaryOpExp(tableMap, qi.WhereExpression_)
+	if err != nil {
+		return nil, err
+	}
+
+	if !(qi.OnExpressions_.Left_ == nil && qi.OnExpressions_.Right_ == nil) {
+		// attach predicate of ON clause to one of WHERE clause
+		qi.WhereExpression_ = qi.WhereExpression_.AppendBinaryOpExpWithAnd(qi.OnExpressions_)
+	}
+
+	return qi, nil
+}
diff --git a/planner/simple_planner.go b/planner/simple_planner.go
index e8639c01..e758f80b 100644
--- a/planner/simple_planner.go
+++ b/planner/simple_planner.go
@@ -7,6 +7,7 @@ import (
 	"github.com/ryogrid/SamehadaDB/execution/expression"
 	"github.com/ryogrid/SamehadaDB/execution/plans"
 	"github.com/ryogrid/SamehadaDB/parser"
+	"github.com/ryogrid/SamehadaDB/planner/optimizer"
 	"github.com/ryogrid/SamehadaDB/storage/access"
 	"github.com/ryogrid/SamehadaDB/storage/buffer"
 	"github.com/ryogrid/SamehadaDB/storage/index/index_constants"
@@ -89,6 +90,11 @@ func (pner *SimplePlanner) MakeSelectPlanWithoutJoin() (error, plans.Plan) {
 	return nil, plans.NewSeqScanPlanNode(pner.catalog_, outSchema, predicate, tableMetadata.OID())
 }
 
+func (pner *SimplePlanner) MakeOptimizedSelectPlanWithJoin() (error, plans.Plan) {
+	optPlan, err := optimizer.NewSelingerOptimizer(pner.qi, pner.catalog_).Optimize()
+	return err, optPlan
+}
+
 func (pner *SimplePlanner) MakeSelectPlanWithJoin() (error, plans.Plan) {
 	tblNameL := *pner.qi.JoinTables_[0]
 	tableMetadataL := pner.catalog_.GetTableByName(tblNameL)
@@ -218,13 +224,19 @@ func (pner *SimplePlanner) MakeSelectPlanWithJoin() (error, plans.Plan) {
 }
 
 func (pner *SimplePlanner) MakeSelectPlan() (error, plans.Plan) {
-	if len(pner.qi.JoinTables_) == 1 {
-		return pner.MakeSelectPlanWithoutJoin()
+	if optimizer.CheckIncludesORInPredicate(pner.qi.WhereExpression_) {
+		// optimizer does not support OR, so use planning logic without optimization...
+		if len(pner.qi.JoinTables_) == 1 {
+			return pner.MakeSelectPlanWithoutJoin()
+		} else {
+			return pner.MakeSelectPlanWithJoin()
+		}
 	} else {
-		return pner.MakeSelectPlanWithJoin()
+		return pner.MakeOptimizedSelectPlanWithJoin()
 	}
 }
 
+// TODO: (SDB) duplicated functionality with expression.ConvParsedBinaryOpExprToExpIFOne func???
 func processPredicateTreeNode(node *parser.BinaryOpExpression, tgtTblSchemas []*schema.Schema) expression.Expression {
 	if node.LogicalOperationType_ != -1 { // node of logical operation
 		left_side_pred := processPredicateTreeNode(node.Left_.(*parser.BinaryOpExpression), tgtTblSchemas)
@@ -257,7 +269,8 @@ func (pner *SimplePlanner) MakeCreateTablePlan() (error, plans.Plan) {
 
 	columns := make([]*column.Column, 0)
 	for _, cdefExp := range pner.qi.ColDefExpressions_ {
-		columns = append(columns, column.NewColumn(*cdefExp.ColName_, *cdefExp.ColType_, false, index_constants.INDEX_KIND_INVALID, types.PageID(-1), nil))
+		//columns = append(columns, column.NewColumn(*cdefExp.ColName_, *cdefExp.ColType_, false, index_constants.INDEX_KIND_INVALID, types.PageID(-1), nil))
+		columns = append(columns, column.NewColumn(*cdefExp.ColName_, *cdefExp.ColType_, true, index_constants.INDEX_KIND_SKIP_LIST, types.PageID(-1), nil))
 	}
 	schema_ := schema.NewSchema(columns)
 
@@ -307,22 +320,31 @@ func (pner *SimplePlanner) MakeInsertPlan() (error, plans.Plan) {
 }
 
 func (pner *SimplePlanner) MakeDeletePlan() (error, plans.Plan) {
-	tableMetadata := pner.catalog_.GetTableByName(*pner.qi.JoinTables_[0])
-	if tableMetadata == nil {
-		return PrintAndCreateError("table " + *pner.qi.JoinTables_[0] + " not found.")
-	}
+	if optimizer.CheckIncludesORInPredicate(pner.qi.WhereExpression_) {
+		// optimizer does not support OR, so use planning logic without optimization...
+		tableMetadata := pner.catalog_.GetTableByName(*pner.qi.JoinTables_[0])
+		if tableMetadata == nil {
+			return PrintAndCreateError("table " + *pner.qi.JoinTables_[0] + " not found.")
+		}
 
-	tgtTblSchema := tableMetadata.Schema()
+		tgtTblSchema := tableMetadata.Schema()
+
+		expression_ := pner.ConstructPredicate([]*schema.Schema{tgtTblSchema})
+		seqScanPlanP := plans.NewSeqScanPlanNode(pner.catalog_, tgtTblSchema, expression_, tableMetadata.OID())
+		deletePlan := plans.NewDeletePlanNode(seqScanPlanP)
 
-	expression_ := pner.ConstructPredicate([]*schema.Schema{tgtTblSchema})
-	//deletePlan := plans.NewDeletePlanNode(expression_, tableMetadata.OID())
-	seqScanPlanP := plans.NewSeqScanPlanNode(pner.catalog_, tgtTblSchema, expression_, tableMetadata.OID())
-	deletePlan := plans.NewDeletePlanNode(seqScanPlanP)
+		return nil, deletePlan
+	} else {
+		_, selectPlan := pner.MakeOptimizedSelectPlanWithJoin()
+		deletePlan := plans.NewDeletePlanNode(selectPlan)
+		return nil, deletePlan
+	}
 
-	return nil, deletePlan
 }
 
 func (pner *SimplePlanner) MakeUpdatePlan() (error, plans.Plan) {
+
+	// optimizer does not support OR, so use planning logic without optimization...
 	tableMetadata := pner.catalog_.GetTableByName(*pner.qi.JoinTables_[0])
 	if tableMetadata == nil {
 		return PrintAndCreateError("table " + *pner.qi.JoinTables_[0] + " not found.")
@@ -357,6 +379,12 @@ func (pner *SimplePlanner) MakeUpdatePlan() (error, plans.Plan) {
 		predicate = pner.ConstructPredicate([]*schema.Schema{tgtTblSchema})
 	}
 
-	seqScanPlan := plans.NewSeqScanPlanNode(pner.catalog_, tgtTblSchema, predicate, tableMetadata.OID())
-	return nil, plans.NewUpdatePlanNode(updateVals, updateColIdxs, seqScanPlan)
+	var scanPlan plans.Plan
+	if optimizer.CheckIncludesORInPredicate(pner.qi.WhereExpression_) {
+		scanPlan = plans.NewSeqScanPlanNode(pner.catalog_, tgtTblSchema, predicate, tableMetadata.OID())
+	} else {
+		_, scanPlan = pner.MakeOptimizedSelectPlanWithJoin()
+	}
+
+	return nil, plans.NewUpdatePlanNode(updateVals, updateColIdxs, scanPlan)
 }
diff --git a/samehada/samehada.go b/samehada/samehada.go
index a22e9631..575ec92e 100644
--- a/samehada/samehada.go
+++ b/samehada/samehada.go
@@ -9,6 +9,7 @@ import (
 	"github.com/ryogrid/SamehadaDB/execution/plans"
 	"github.com/ryogrid/SamehadaDB/parser"
 	"github.com/ryogrid/SamehadaDB/planner"
+	"github.com/ryogrid/SamehadaDB/planner/optimizer"
 	"github.com/ryogrid/SamehadaDB/recovery/log_recovery"
 	"github.com/ryogrid/SamehadaDB/samehada/samehada_util"
 	"github.com/ryogrid/SamehadaDB/storage/access"
@@ -26,8 +27,9 @@ type SamehadaDB struct {
 	shi_         *SamehadaInstance
 	catalog_     *catalog.Catalog
 	exec_engine_ *executors.ExecutionEngine
-	chkpntMgr    *concurrency.CheckpointManager
-	planner_     planner.Planner
+	//chkpntMgr    *concurrency.CheckpointManager
+	planner_           planner.Planner
+	statistics_updator *concurrency.StatisticsUpdater
 }
 
 func reconstructIndexDataOfATbl(t *catalog.TableMetadata, c *catalog.Catalog, dman disk.DiskManager, txn *access.Transaction) {
@@ -59,6 +61,10 @@ func reconstructIndexDataOfATbl(t *catalog.TableMetadata, c *catalog.Catalog, dm
 				// do nothing here
 				// (Since SkipList index can't reuse past allocated pages, data clear of allocated pages
 				//  are not needed...)
+			case index_constants.INDEX_KIND_SKIP_LIST:
+				// do nothing here
+				// (Since SkipList index can't reuse past allocated pages, data clear of allocated pages
+				//  are not needed...)
 			default:
 				panic("invalid index kind!")
 			}
@@ -137,10 +143,16 @@ func NewSamehadaDB(dbName string, memKBytes int) *SamehadaDB {
 	exec_engine := &executors.ExecutionEngine{}
 	pnner := planner.NewSimplePlanner(c, shi.GetBufferPoolManager())
 
-	chkpntMgr := concurrency.NewCheckpointManager(shi.GetTransactionManager(), shi.GetLogManager(), shi.GetBufferPoolManager())
-	chkpntMgr.StartCheckpointTh()
+	//chkpntMgr := concurrency.NewCheckpointManager(shi.GetTransactionManager(), shi.GetLogManager(), shi.GetBufferPoolManager())
+	//chkpntMgr.StartCheckpointTh()
+	shi.GetCheckpointManager().StartCheckpointTh()
 
-	return &SamehadaDB{shi, c, exec_engine, chkpntMgr, pnner}
+	// statics data is updated periodically by this thread with full scan of all tables
+	// this may be not good implementation of statistics, but it is enough for now...
+	statUpdater := concurrency.NewStatisticsUpdater(shi.GetTransactionManager(), c)
+	statUpdater.StartStaticsUpdaterTh()
+
+	return &SamehadaDB{shi, c, exec_engine, pnner, statUpdater}
 }
 
 func (sdb *SamehadaDB) ExecuteSQL(sqlStr string) (error, [][]interface{}) {
@@ -150,6 +162,7 @@ func (sdb *SamehadaDB) ExecuteSQL(sqlStr string) (error, [][]interface{}) {
 
 func (sdb *SamehadaDB) ExecuteSQLRetValues(sqlStr string) (error, [][]*types.Value) {
 	qi := parser.ProcessSQLStr(&sqlStr)
+	qi, _ = optimizer.RewriteQueryInfo(sdb.catalog_, qi)
 	txn := sdb.shi_.transaction_manager.Begin(nil)
 	err, plan := sdb.planner_.MakePlan(qi, txn)
 
@@ -185,21 +198,25 @@ func (sdb *SamehadaDB) ExecuteSQLRetValues(sqlStr string) (error, [][]*types.Val
 }
 
 func (sdb *SamehadaDB) Shutdown() {
-	// set a flag which is check by checkpointing thread
-	sdb.chkpntMgr.StopCheckpointTh()
+	// set a flag which is checked by checkpointing thread
+	sdb.statistics_updator.StopStatsUpdateTh()
+	sdb.shi_.GetCheckpointManager().StopCheckpointTh()
 	sdb.shi_.GetBufferPoolManager().FlushAllDirtyPages()
 	sdb.shi_.Shutdown(false)
 }
 
+// no flush of page buffer
 func (sdb *SamehadaDB) ShutdownForTescase() {
-	// set a flag which is check by checkpointing thread
-	sdb.chkpntMgr.StopCheckpointTh()
-	sdb.shi_.Shutdown(false)
+	// set a flag which is checked by checkpointing thread
+	sdb.shi_.GetCheckpointManager().StopCheckpointTh()
+	sdb.statistics_updator.StopStatsUpdateTh()
+	//sdb.shi_.Shutdown(false)
+	sdb.shi_.CloseFilesForTesting()
 }
 
 func (sdb *SamehadaDB) ForceCheckpointingForTestcase() {
-	sdb.chkpntMgr.BeginCheckpoint()
-	sdb.chkpntMgr.EndCheckpoint()
+	sdb.shi_.GetCheckpointManager().BeginCheckpoint()
+	sdb.shi_.GetCheckpointManager().EndCheckpoint()
 }
 
 func ConvTupleListToValues(schema_ *schema.Schema, result []*tuple.Tuple) [][]*types.Value {
diff --git a/samehada/samehada_test/samehada_test.go b/samehada/samehada_test/samehada_test.go
index d51aa526..d72ade55 100644
--- a/samehada/samehada_test/samehada_test.go
+++ b/samehada/samehada_test/samehada_test.go
@@ -154,7 +154,8 @@ func TestRebootWithLoadAndRecovery(t *testing.T) {
 	testingpkg.SimpleAssert(t, len(results1) == 3)
 
 	// close db and log file
-	db.Shutdown()
+	db.ShutdownForTescase()
+	//db.Shutdown()
 
 	// relaunch using TestRebootWithLoadAndRecovery.log files
 	// load of db file and redo/undo process runs
@@ -166,7 +167,8 @@ func TestRebootWithLoadAndRecovery(t *testing.T) {
 	testingpkg.SimpleAssert(t, len(results2) == 4)
 
 	// close db and log file
-	db2.Shutdown()
+	db2.ShutdownForTescase()
+	//db2.Shutdown()
 
 	// relaunch using TestRebootWithLoadAndRecovery.db and TestRebootWithLoadAndRecovery.log files
 	// load of db file and redo/undo process runs
@@ -210,7 +212,8 @@ func TestRebootAndReturnIFValues(t *testing.T) {
 	testingpkg.SimpleAssert(t, len(results1) == 3)
 
 	// close db and log file
-	db.Shutdown()
+	db.ShutdownForTescase()
+	//db.Shutdown()
 
 	// relaunch
 	// load of db file and redo/undo process runs
@@ -225,7 +228,8 @@ func TestRebootAndReturnIFValues(t *testing.T) {
 	testingpkg.SimpleAssert(t, len(results2) == 4)
 
 	// close db and log file
-	db2.Shutdown()
+	db2.ShutdownForTescase()
+	//db2.Shutdown()
 
 	// relaunch
 	// load of db file and redo/undo process runs
diff --git a/storage/access/table_heap.go b/storage/access/table_heap.go
index e013a527..3eb5a949 100644
--- a/storage/access/table_heap.go
+++ b/storage/access/table_heap.go
@@ -170,9 +170,6 @@ func (t *TableHeap) UpdateTuple(tuple_ *tuple.Tuple, update_col_idxs []int, sche
 		// and insert need_follow_tuple(new_rid)
 		// as updating
 
-		// TODO: (SDB) for debug
-		fmt.Println("TableHeap::UpdateTuple changing rid occured!")
-
 		// first, delete target tuple1 (old data)
 		var is_deleted bool
 		if isRollback {
diff --git a/storage/index/skip_list_index.go b/storage/index/skip_list_index.go
index 6c838bf5..452adace 100644
--- a/storage/index/skip_list_index.go
+++ b/storage/index/skip_list_index.go
@@ -53,8 +53,6 @@ func (slidx *SkipListIndex) DeleteEntry(key *tuple.Tuple, rid page.RID, txn inte
 
 	convedKeyVal := samehada_util.EncodeValueAndRIDToDicOrderComparableVarchar(&orgKeyVal, &rid)
 
-	// TODO: (SDB) for debug
-	//fmt.Printf("SkipListIndex::DeleteEntry: %v %v\n", convedKeyVal.ToIFValue(), rid)
 	revertedOrgKey := samehada_util.ExtractOrgKeyFromDicOrderComparableEncodedVarchar(convedKeyVal, orgKeyVal.ValueType())
 	if !revertedOrgKey.CompareEquals(orgKeyVal) {
 		panic("key conversion may fail!")