プルリクエスト #23 のマージ

embeded form front end uses optimizer if usable case

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()))
 	}
 

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))

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!")
 			}

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
 			}

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
+}

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 {

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
 )