Repository: matrixorigin/matrixone
Pull request #5764 (2022-10-19 10:59:50)

refactor log data store path as account/log/yyyy/mm/dd/datatype/*.csv
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue https://github.com/matrixorigin/matrixone/issues/5598

## What this PR does / why we need it:
adapt log data write in filepath {account}/log/yyyy/mm/dd/metric/*.csv
- result as Table `SingleStatementTable`, `SingleRawLogTable`, and View `logView`, `errorView`, `spanView`
    - Table `SingleStatementTable` handle `statement` data, storing in path `account/log/yyyy/mm/dd/statement_info/*.csv`
    - Table `SingleRawLogTable` handle `log`, `error`, `span` data, storing in path `account/log/yyyy/mm/dd/rawlog/*.csv`
    - View `logView`, only show `log` data in Table`SingleRawLogTable`
        - view schema like `create view logView as select ... from rawlog where raw_item = 'log_info'`
    - View `errorView` like logView
    - View `spanView` like logView

others
- add config `pathBuilder` chose which rule exportor would use.
    - value `DBTable`, build path like `{db_name}/{tbl_name}_*.csv`
    - value `AccountDate`,  build path like `account/{log,merged}/yyyy/mm/dd/datatype/*.csv`

xzxiong: @matrix-meow rerun failed

Patch:
diff --git a/cmd/mo-service/main.go b/cmd/mo-service/main.go
index 0299c02ae..8f60451bc 100644
--- a/cmd/mo-service/main.go
+++ b/cmd/mo-service/main.go
@@ -267,6 +267,7 @@ func initTraceMetric(ctx context.Context, cfg *Config, stopper *stopper.Stopper,
 
 	if !SV.DisableTrace || !SV.DisableMetric {
 		writerFactory = export.GetFSWriterFactory(fs, UUID, nodeRole)
+		_ = export.SetPathBuilder(SV.PathBuilder)
 	}
 	if !SV.DisableTrace {
 		initWG.Add(1)
@@ -279,7 +280,6 @@ func initTraceMetric(ctx context.Context, cfg *Config, stopper *stopper.Stopper,
 				trace.WithFSWriterFactory(writerFactory),
 				trace.WithExportInterval(SV.TraceExportInterval),
 				trace.WithLongQueryTime(SV.LongQueryTime),
-				trace.DebugMode(SV.EnableTraceDebug),
 				trace.WithSQLExecutor(nil),
 			); err != nil {
 				panic(err)
@@ -302,7 +302,6 @@ func initTraceMetric(ctx context.Context, cfg *Config, stopper *stopper.Stopper,
 	if SV.MergeCycle > 0 {
 		stopper.RunNamedTask("merge", func(ctx context.Context) {
 			merge, inited := export.NewMergeService(ctx,
-				export.WithDB(metric.MetricDBConst),
 				export.WithTable(metric.SingleMetricTable),
 				export.WithFileService(fs),
 				export.WithMinFilesMerge(1),
diff --git a/etc/launch-tae-logservice/log1.toml b/etc/launch-tae-logservice/log1.toml
index 37497bc47..9b649ed41 100644
--- a/etc/launch-tae-logservice/log1.toml
+++ b/etc/launch-tae-logservice/log1.toml
@@ -3,7 +3,7 @@ service-type = "LOG"
 
 [log]
 level = "info"
-format = "json"
+format = "console"
 max-size = 512
 
 [logservice]
diff --git a/pkg/config/configuration.go b/pkg/config/configuration.go
index 973cc68ac..871aa4f15 100644
--- a/pkg/config/configuration.go
+++ b/pkg/config/configuration.go
@@ -118,6 +118,9 @@ var (
 	// defaultMergeCycle default: 0 sec, means disable merge as service
 	defaultMergeCycle = 0
 
+	// defaultPathBuilder, val in [DBTable, AccountDate]
+	defaultPathBuilder = "AccountDate"
+
 	// defaultSessionTimeout default: 10 minutes
 	defaultSessionTimeout = 10 * time.Minute
 )
@@ -396,6 +399,9 @@ type ObservabilityParameters struct {
 	MetricGatherInterval int `toml:"metricGatherInterval"`
 
 	MergeCycle int `toml:"mergeCycle"`
+
+	// PathBuilder default: DBTable. Support val in [DBTable, AccountDate]
+	PathBuilder string `toml:"PathBuilder"`
 }
 
 func (op *ObservabilityParameters) SetDefaultValues(version string) {
@@ -428,6 +434,10 @@ func (op *ObservabilityParameters) SetDefaultValues(version string) {
 	if op.MergeCycle <= 0 {
 		op.MergeCycle = defaultMergeCycle
 	}
+
+	if op.PathBuilder == "" {
+		op.PathBuilder = defaultPathBuilder
+	}
 }
 
 type ParameterUnit struct {
diff --git a/pkg/logutil/api.go b/pkg/logutil/api.go
index 4adab105e..e9f80fd89 100644
--- a/pkg/logutil/api.go
+++ b/pkg/logutil/api.go
@@ -132,7 +132,6 @@ func GetLoggerWithOptions(level zapcore.LevelEnabler, encoder zapcore.Encoder, s
 		cores = append(cores, zapcore.NewCore(encoder, syncer, level))
 	}
 
-	GetLevelChangeFunc()(level)
 	return zap.New(zapcore.NewTee(cores...), options...)
 }
 
diff --git a/pkg/logutil/internal.go b/pkg/logutil/internal.go
index 0736c2267..a4a47bf4a 100644
--- a/pkg/logutil/internal.go
+++ b/pkg/logutil/internal.go
@@ -48,7 +48,7 @@ var _errorLogger atomic.Value
 
 // init initializes a default zap logger before set up logger.
 func init() {
-	SetLogReporter(&TraceReporter{noopReportLog, noopReportZap, noopLevelSignal, noopContextField})
+	SetLogReporter(&TraceReporter{noopReportZap, noopContextField})
 	conf := &LogConfig{Level: "info", Format: "console"}
 	setGlobalLogConfig(conf)
 	logger, _ := initMOLogger(conf)
diff --git a/pkg/logutil/report.go b/pkg/logutil/report.go
index 1470512c2..77e9160a5 100644
--- a/pkg/logutil/report.go
+++ b/pkg/logutil/report.go
@@ -43,64 +43,38 @@ type ZapSink struct {
 	out zapcore.WriteSyncer
 }
 
-// logReporter should be trace.ReportLog
-var logReporter atomic.Value
-
 // zapReporter should be trace.ReportZap
 var zapReporter atomic.Value
 
-// levelChangeFunc should be trace.SetLogLevel function
-var levelChangeFunc atomic.Value
-
 // contextField should be trace.ContextField function
 var contextField atomic.Value
 
 type TraceReporter struct {
-	ReportLog    reportLogFunc
 	ReportZap    reportZapFunc
-	LevelSignal  levelChangeSignal
 	ContextField contextFieldFunc
 }
 
-type reportLogFunc func(context.Context, zapcore.Level, int, string, ...any)
 type reportZapFunc func(zapcore.Encoder, zapcore.Entry, []zapcore.Field) (*buffer.Buffer, error)
-type levelChangeSignal func(zapcore.LevelEnabler)
 type contextFieldFunc func(context.Context) zap.Field
 
-func noopReportLog(context.Context, zapcore.Level, int, string, ...any) {}
 func noopReportZap(zapcore.Encoder, zapcore.Entry, []zapcore.Field) (*buffer.Buffer, error) {
 	return buffer.NewPool().Get(), nil
 }
-func noopLevelSignal(zapcore.LevelEnabler)       {}
 func noopContextField(context.Context) zap.Field { return zap.String("span", "{}") }
 
 func SetLogReporter(r *TraceReporter) {
-	if r.ReportLog != nil {
-		logReporter.Store(r.ReportLog)
-	}
 	if r.ReportZap != nil {
 		zapReporter.Store(r.ReportZap)
 	}
-	if r.LevelSignal != nil {
-		levelChangeFunc.Store(r.LevelSignal)
-	}
 	if r.ContextField != nil {
 		contextField.Store(r.ContextField)
 	}
 }
 
-func GetReportLogFunc() reportLogFunc {
-	return logReporter.Load().(reportLogFunc)
-}
-
 func GetReportZapFunc() reportZapFunc {
 	return zapReporter.Load().(reportZapFunc)
 }
 
-func GetLevelChangeFunc() levelChangeSignal {
-	return levelChangeFunc.Load().(levelChangeSignal)
-}
-
 func GetContextFieldFunc() contextFieldFunc {
 	return contextField.Load().(contextFieldFunc)
 }
@@ -149,6 +123,7 @@ func newTraceLogEncoder() *TraceLogEncoder {
 		Encoder: zapcore.NewJSONEncoder(
 			zapcore.EncoderConfig{
 				StacktraceKey:  "stacktrace",
+				SkipLineEnding: true,
 				LineEnding:     zapcore.DefaultLineEnding,
 				EncodeLevel:    zapcore.LowercaseLevelEncoder,
 				EncodeTime:     zapcore.EpochTimeEncoder,
diff --git a/pkg/util/export/merge.go b/pkg/util/export/merge.go
index 92ec79dbf..16e3a8dfa 100644
--- a/pkg/util/export/merge.go
+++ b/pkg/util/export/merge.go
@@ -28,7 +28,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
 	"github.com/matrixorigin/matrixone/pkg/fileservice"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
-
 	"github.com/matrixorigin/simdcsv"
 )
 
@@ -37,22 +36,28 @@ import (
 // ========================
 
 // Merge like a compaction, merge input files into one/two/... files.
+//   - `NewMergeService` init merge as service, with param `serviceInit` to avoid multi init.
+//   - `NewMerge` handle merge obj init.
+//   - `Merge::Start` as service loop, trigger `Merge::Main` each cycle
+//   - `Merge::Main` handle handle job,
+//     1. foreach account, build `rootPath` with tuple {account, date, Table }
+//     2. call `Merge::doMergeFiles` with all files in `rootPath`,  do merge job
+//   - `Merge::doMergeFiles` handle one job flow: read each file, merge in cache, write into file.
 type Merge struct {
-	Table       *Table                  // see WithTable
-	DB          string                  // see WithDB
-	FS          fileservice.FileService // see WithFileService
-	FSName      string                  // see WithFileServiceName, cooperate with FS
+	Table       *Table                  // WithTable
+	FS          fileservice.FileService // WithFileService
+	FSName      string                  // WithFileServiceName, cooperate with FS
 	datetime    time.Time               // see Main
-	pathBuilder PathBuilder             // const as NewMetricLogPathBuilder()
+	pathBuilder PathBuilder             // const as NewAccountDatePathBuilder()
 
 	// MaxFileSize 控制合并后最大文件大小, default: 128 MB
-	MaxFileSize int64 // see WithMaxFileSize
+	MaxFileSize int64 // WithMaxFileSize
 	// MaxMergeJobs 允许进行的Merge的任务个数，default: 16
-	MaxMergeJobs int64 // see WithMaxMergeJobs
+	MaxMergeJobs int64 // WithMaxMergeJobs
 	// MinFilesMerge 控制Merge最少合并文件个数，default：2
 	//
 	// Deprecated: useless in Merge all in one file
-	MinFilesMerge int // see WithMinFilesMerge
+	MinFilesMerge int // WithMinFilesMerge
 	// FileCacheSize 控制Merge 过程中, 允许缓存的文件大小，default: 16 MB
 	//
 	// Deprecated: useless while NOT support multiParts upload
@@ -76,11 +81,6 @@ func WithTable(tbl *Table) MergeOption {
 		m.Table = tbl
 	})
 }
-func WithDB(db string) MergeOption {
-	return MergeOption(func(m *Merge) {
-		m.DB = db
-	})
-}
 func WithFileService(fs fileservice.FileService) MergeOption {
 	return MergeOption(func(m *Merge) {
 		m.FS = fs
@@ -123,7 +123,7 @@ func NewMerge(ctx context.Context, opts ...MergeOption) *Merge {
 	m := &Merge{
 		FSName:        etlFileServiceName,
 		datetime:      time.Now(),
-		pathBuilder:   NewMetricLogPathBuilder(),
+		pathBuilder:   NewAccountDatePathBuilder(),
 		MaxFileSize:   128 * mpool.MB,
 		MaxMergeJobs:  16,
 		MinFilesMerge: 2,
@@ -184,7 +184,7 @@ func (m *Merge) Main(ts time.Time) error {
 	if m.datetime.IsZero() {
 		return moerr.NewInternalError("Merge Task missing input 'datetime'")
 	}
-	logutil.Debugf("Merge start on %s.%s, %v", m.DB, m.Table.GetName(), m.datetime)
+	logutil.Debugf("Merge start on %s, %v", m.Table.GetIdentify(), m.datetime)
 	accounts, err := m.FS.List(m.ctx, "/")
 	if err != nil {
 		return err
@@ -194,7 +194,7 @@ func (m *Merge) Main(ts time.Time) error {
 			logutil.Warnf("path is not dir: %s", account.Name)
 			continue
 		}
-		rootPath := m.pathBuilder.Build(account.Name, MergeLogTypeLog, m.datetime, m.DB, m.Table.GetName())
+		rootPath := m.pathBuilder.Build(account.Name, MergeLogTypeLog, m.datetime, m.Table.GetDatabase(), m.Table.GetName())
 		// get all file entry
 
 		fileEntrys, err := m.FS.List(m.ctx, rootPath)
@@ -255,7 +255,7 @@ func (m *Merge) doMergeFiles(account string, paths []string) error {
 	timestampEnd := timestamps[len(timestamps)-1]
 
 	// Step 2. new filename, file writer
-	prefix := m.pathBuilder.Build(account, MergeLogTypeMerged, m.datetime, m.DB, m.Table.GetName())
+	prefix := m.pathBuilder.Build(account, MergeLogTypeMerged, m.datetime, m.Table.GetDatabase(), m.Table.GetName())
 	mergeFilename := m.pathBuilder.NewMergeFilename(timestampStart, timestampEnd)
 	mergeFilepath := path.Join(prefix, mergeFilename)
 	newFileWriter, _ := NewCSVWriter(m.ctx, m.FS, mergeFilepath)
diff --git a/pkg/util/export/table.go b/pkg/util/export/table.go
index de95d6376..f555fecfe 100644
--- a/pkg/util/export/table.go
+++ b/pkg/util/export/table.go
@@ -17,6 +17,7 @@ package export
 import (
 	"fmt"
 	"strings"
+	"sync"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
@@ -42,11 +43,21 @@ type Column struct {
 	Comment string
 }
 
+// ToCreateSql return column scheme in create sql
+//   - case 1: `column_name` varchar(36) DEFAULT "def_val" COMMENT "what am I, with default."
+//   - case 2: `column_name` varchar(36) NOT NULL COMMENT "what am I. Without default, SO NOT NULL."
 func (col *Column) ToCreateSql() string {
 	sb := strings.Builder{}
 	sb.WriteString(fmt.Sprintf("`%s` %s ", col.Name, col.Type))
-	if len(col.Default) > 0 {
+	if col.Type == "JSON" {
+		sb.WriteString("NOT NULL ")
+		if len(col.Default) == 0 {
+			panic(moerr.NewNotSupported("json column need default in csv, but not in schema"))
+		}
+	} else if len(col.Default) > 0 {
 		sb.WriteString(fmt.Sprintf("DEFAULT %q ", col.Default))
+	} else {
+		sb.WriteString("NOT NULL ")
 	}
 	sb.WriteString(fmt.Sprintf("COMMENT %q", col.Comment))
 	return sb.String()
@@ -64,15 +75,24 @@ type Table struct {
 	PrimaryKeyColumn []Column
 	Engine           string
 	Comment          string
-	TableOptions     TableOptions
-	PathBuilder      PathBuilder
-
+	// PathBuilder help to desc param 'infile'
+	PathBuilder PathBuilder
+	// AccountColumn help to split data in account's filepath
 	AccountColumn *Column
+	// TableOptions default is nil, see GetTableOptions
+	TableOptions TableOptions
 }
 
 func (tbl *Table) GetName() string {
 	return tbl.Table
 }
+func (tbl *Table) GetDatabase() string {
+	return tbl.Database
+}
+
+func (tbl *Table) GetIdentify() string {
+	return fmt.Sprintf("%s.%s", tbl.Database, tbl.Table)
+}
 
 type TableOptions interface {
 	FormatDdl(ddl string) string
@@ -83,13 +103,15 @@ type TableOptions interface {
 
 func (tbl *Table) ToCreateSql(ifNotExists bool) string {
 
+	TableOptions := tbl.GetTableOptions()
+
 	const newLineCharacter = ",\n"
 	sb := strings.Builder{}
 	// create table
 	sb.WriteString("CREATE ")
 	switch strings.ToUpper(tbl.Engine) {
 	case ExternalTableEngine:
-		sb.WriteString(tbl.TableOptions.GetCreateOptions())
+		sb.WriteString(TableOptions.GetCreateOptions())
 	default:
 		panic(moerr.NewInternalError("NOT support engine: %s", tbl.Engine))
 	}
@@ -111,7 +133,7 @@ func (tbl *Table) ToCreateSql(ifNotExists bool) string {
 	// primary key
 	if len(tbl.PrimaryKeyColumn) > 0 {
 		sb.WriteString(newLineCharacter)
-		sb.WriteString("PRIMARY KEY (`")
+		sb.WriteString("PRIMARY KEY (")
 		for idx, col := range tbl.PrimaryKeyColumn {
 			if idx > 0 {
 				sb.WriteString(`, `)
@@ -121,11 +143,18 @@ func (tbl *Table) ToCreateSql(ifNotExists bool) string {
 		sb.WriteString(`)`)
 	}
 	sb.WriteString("\n)")
-	sb.WriteString(tbl.TableOptions.GetTableOptions(tbl.PathBuilder))
+	sb.WriteString(TableOptions.GetTableOptions(tbl.PathBuilder))
 
 	return sb.String()
 }
 
+func (tbl *Table) GetTableOptions() TableOptions {
+	if tbl.TableOptions != nil {
+		return tbl.TableOptions
+	}
+	return GetOptionFactory(tbl.Engine)(tbl.Database, tbl.Table)
+}
+
 type ViewOption func(view *View)
 
 func (opt ViewOption) Apply(view *View) {
@@ -173,6 +202,15 @@ func (tbl *View) ToCreateSql(ifNotExists bool) string {
 	return sb.String()
 }
 
+type ViewSingleCondition struct {
+	Column Column
+	Table  string
+}
+
+func (tbl *ViewSingleCondition) String() string {
+	return fmt.Sprintf("`%s` = %q", tbl.Column.Name, tbl.Table)
+}
+
 type Row struct {
 	Table          *Table
 	AccountIdx     int
@@ -206,9 +244,10 @@ func (r *Row) Reset() {
 	}
 }
 
-// GetAccount return r.Columns[r.AccountIdx] if r.AccountIdx >= 0, else return "sys"
+// GetAccount return r.Columns[r.AccountIdx] if r.AccountIdx >= 0 and r.Table.PathBuilder.SupportAccountStrategy,
+// else return "sys"
 func (r *Row) GetAccount() string {
-	if r.AccountIdx >= 0 {
+	if r.Table.PathBuilder.SupportAccountStrategy() && r.AccountIdx >= 0 {
 		return r.Columns[r.AccountIdx]
 	}
 	return "sys"
@@ -222,6 +261,14 @@ func (r *Row) SetVal(col string, val string) {
 	}
 }
 
+func (r *Row) SetColumnVal(col Column, val string) {
+	if idx, exist := r.Name2ColumnIdx[col.Name]; !exist {
+		logutil.Fatalf("column(%s) not exist in table(%s)", col.Name, r.Table.Table)
+	} else {
+		r.Columns[idx] = val
+	}
+}
+
 func (r *Row) SetFloat64(col string, val float64) {
 	r.SetVal(col, fmt.Sprintf("%f", val))
 }
@@ -247,3 +294,83 @@ type NoopTableOptions struct{}
 func (o NoopTableOptions) FormatDdl(ddl string) string        { return ddl }
 func (o NoopTableOptions) GetCreateOptions() string           { return "" }
 func (o NoopTableOptions) GetTableOptions(PathBuilder) string { return "" }
+
+var _ TableOptions = (*CsvTableOptions)(nil)
+
+type CsvTableOptions struct {
+	Formatter string
+	DbName    string
+	TblName   string
+}
+
+func getExternalTableDDLPrefix(sql string) string {
+	return strings.Replace(sql, "CREATE TABLE", "CREATE EXTERNAL TABLE", 1)
+}
+
+func (o *CsvTableOptions) FormatDdl(ddl string) string {
+	return getExternalTableDDLPrefix(ddl)
+}
+
+func (o *CsvTableOptions) GetCreateOptions() string {
+	return "EXTERNAL "
+}
+
+func (o *CsvTableOptions) GetTableOptions(builder PathBuilder) string {
+	if builder == nil {
+		builder = NewDBTablePathBuilder()
+	}
+	if len(o.Formatter) > 0 {
+		return fmt.Sprintf(o.Formatter, builder.BuildETLPath(o.DbName, o.TblName))
+	}
+	return ""
+}
+
+func GetOptionFactory(engine string) func(db, tbl string) TableOptions {
+	var infileFormatter = ` infile{"filepath"="etl:%s","compression"="none"}` +
+		` FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n' IGNORE 0 lines`
+	switch engine {
+	case NormalTableEngine:
+		return func(_, _ string) TableOptions { return NoopTableOptions{} }
+	case ExternalTableEngine:
+		return func(db, tbl string) TableOptions {
+			return &CsvTableOptions{Formatter: infileFormatter, DbName: db, TblName: tbl}
+		}
+	default:
+		panic(moerr.NewInternalError("unknown engine: %s", engine))
+	}
+}
+
+var gTable map[string]*Table
+var mux sync.Mutex
+
+// RegisterTableDefine return old one, if already registered
+var RegisterTableDefine = func(table *Table) *Table {
+	mux.Lock()
+	defer mux.Unlock()
+	if len(gTable) == 0 {
+		gTable = make(map[string]*Table)
+	}
+	id := table.GetIdentify()
+	old := gTable[id]
+	gTable[id] = table
+	return old
+}
+
+var GetAllTable = func() []*Table {
+	mux.Lock()
+	defer mux.Unlock()
+	tables := make([]*Table, 0, len(gTable))
+	for _, tbl := range gTable {
+		tables = append(tables, tbl)
+	}
+	return tables
+}
+
+func SetPathBuilder(pathBuilder string) error {
+	tables := GetAllTable()
+	bp := PathBuilderFactory(pathBuilder)
+	for _, tbl := range tables {
+		tbl.PathBuilder = bp
+	}
+	return nil
+}
diff --git a/pkg/util/export/type.go b/pkg/util/export/type.go
index 3c9895848..714e8ef44 100644
--- a/pkg/util/export/type.go
+++ b/pkg/util/export/type.go
@@ -124,6 +124,12 @@ type PathBuilder interface {
 	ParsePath(path string) (CSVPath, error)
 	NewMergeFilename(timestampStart, timestampEnd string) string
 	NewLogFilename(name, nodeUUID, nodeType string, ts time.Time) string
+	// SupportMergeSplit const. if false, not support SCV merge|split task
+	SupportMergeSplit() bool
+	// SupportAccountStrategy const
+	SupportAccountStrategy() bool
+	// GetName const
+	GetName() string
 }
 
 type CSVPath interface {
@@ -197,15 +203,15 @@ func (p *MetricLogPath) Timestamp() []string {
 	return p.timestamps
 }
 
-var _ PathBuilder = (*MetricLogPathBuilder)(nil)
+var _ PathBuilder = (*AccountDatePathBuilder)(nil)
 
-type MetricLogPathBuilder struct{}
+type AccountDatePathBuilder struct{}
 
-func NewMetricLogPathBuilder() *MetricLogPathBuilder {
-	return &MetricLogPathBuilder{}
+func NewAccountDatePathBuilder() *AccountDatePathBuilder {
+	return &AccountDatePathBuilder{}
 }
 
-func (m *MetricLogPathBuilder) Build(account string, typ MergeLogType, ts time.Time, db string, name string) string {
+func (b *AccountDatePathBuilder) Build(account string, typ MergeLogType, ts time.Time, db string, name string) string {
 	if ts != ETLParamTSAll {
 		return path.Join(account,
 			typ.String(),
@@ -223,25 +229,29 @@ func (m *MetricLogPathBuilder) Build(account string, typ MergeLogType, ts time.T
 //
 // #     account | typ | ts   | table | filename
 // like: *       /*    /*/*/* /metric /*.csv
-func (m *MetricLogPathBuilder) BuildETLPath(db string, name string) string {
-	etlDirectory := m.Build(ETLParamAccountAll, ETLParamTypeAll, ETLParamTSAll, db, name)
+func (b *AccountDatePathBuilder) BuildETLPath(db string, name string) string {
+	etlDirectory := b.Build(ETLParamAccountAll, ETLParamTypeAll, ETLParamTSAll, db, name)
 	etlFilename := "*" + CsvExtension
 	return path.Join(etlDirectory, etlFilename)
 }
 
-func (m *MetricLogPathBuilder) ParsePath(path string) (CSVPath, error) {
+func (b *AccountDatePathBuilder) ParsePath(path string) (CSVPath, error) {
 	p := NewMetricLogPath(path)
 	return p, p.Parse()
 }
 
-func (m *MetricLogPathBuilder) NewMergeFilename(timestampStart, timestampEnd string) string {
+func (b *AccountDatePathBuilder) NewMergeFilename(timestampStart, timestampEnd string) string {
 	return strings.Join([]string{timestampStart, timestampEnd, string(MergeLogTypeMerged)}, FilenameSeparator) + CsvExtension
 }
 
-func (m *MetricLogPathBuilder) NewLogFilename(name, nodeUUID, nodeType string, ts time.Time) string {
+func (b *AccountDatePathBuilder) NewLogFilename(name, nodeUUID, nodeType string, ts time.Time) string {
 	return strings.Join([]string{fmt.Sprintf("%d", ts.Unix()), nodeUUID, nodeType}, FilenameSeparator) + CsvExtension
 }
 
+func (b *AccountDatePathBuilder) SupportMergeSplit() bool      { return true }
+func (b *AccountDatePathBuilder) SupportAccountStrategy() bool { return true }
+func (b *AccountDatePathBuilder) GetName() string              { return "AccountDate" }
+
 var _ PathBuilder = (*DBTablePathBuilder)(nil)
 
 type DBTablePathBuilder struct{}
@@ -272,3 +282,18 @@ func (m *DBTablePathBuilder) NewMergeFilename(timestampStart, timestampEnd strin
 func (m *DBTablePathBuilder) NewLogFilename(name, nodeUUID, nodeType string, ts time.Time) string {
 	return fmt.Sprintf(`%s_%s_%s_%s`, name, nodeUUID, nodeType, ts.Format("20060102.150405.000000")) + CsvExtension
 }
+
+func (m *DBTablePathBuilder) SupportMergeSplit() bool      { return false }
+func (m *DBTablePathBuilder) SupportAccountStrategy() bool { return false }
+func (m *DBTablePathBuilder) GetName() string              { return "DBTable" }
+
+func PathBuilderFactory(pathBuilder string) PathBuilder {
+	switch pathBuilder {
+	case (*DBTablePathBuilder)(nil).GetName():
+		return NewDBTablePathBuilder()
+	case (*AccountDatePathBuilder)(nil).GetName():
+		return NewAccountDatePathBuilder()
+	default:
+		panic(moerr.NewNotSupported("not support PathBuilder: %s", pathBuilder))
+	}
+}
diff --git a/pkg/util/metric/metric.go b/pkg/util/metric/metric.go
index 4b12eed3a..81267c4e2 100644
--- a/pkg/util/metric/metric.go
+++ b/pkg/util/metric/metric.go
@@ -24,6 +24,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/config"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
@@ -208,7 +209,7 @@ func initTables(ctx context.Context, ieFactory func() ie.InternalExecutor, batch
 			mustExec(sql)
 		}
 	} else {
-		optFactory := trace.GetOptionFactory(export.ExternalTableEngine)
+		optFactory := export.GetOptionFactory(export.ExternalTableEngine)
 		buf := new(bytes.Buffer)
 		for desc := range descChan {
 			sql := createTableSqlFromMetricFamily(desc, buf, optFactory)
@@ -340,13 +341,8 @@ var SingleMetricTable = &export.Table{
 	PrimaryKeyColumn: []export.Column{},
 	Engine:           export.ExternalTableEngine,
 	Comment:          `metric data`,
-	TableOptions:     trace.GetOptionFactory(export.ExternalTableEngine)(MetricDBConst, `metric`),
-	PathBuilder:      export.NewDBTablePathBuilder(),
-	AccountColumn:    nil,
-}
-
-type ViewWhereCondition struct {
-	Table string
+	PathBuilder:      export.NewAccountDatePathBuilder(),
+	AccountColumn:    &metricAccountColumn,
 }
 
 func NewMetricView(tbl string, opts ...export.ViewOption) *export.View {
@@ -355,7 +351,7 @@ func NewMetricView(tbl string, opts ...export.ViewOption) *export.View {
 		Table:       tbl,
 		OriginTable: SingleMetricTable,
 		Columns:     []export.Column{metricCollectTimeColumn, metricValueColumn, metricNodeColumn, metricRoleColumn},
-		Condition:   &ViewWhereCondition{Table: tbl},
+		Condition:   &export.ViewSingleCondition{Column: metricNameColumn, Table: tbl},
 	}
 	for _, opt := range opts {
 		opt.Apply(view)
@@ -375,10 +371,6 @@ func NewMetricViewWithLabels(tbl string, lbls []string) *export.View {
 	return NewMetricView(tbl, options...)
 }
 
-func (tbl *ViewWhereCondition) String() string {
-	return fmt.Sprintf("`%s` = %q", metricNameColumn.Name, tbl.Table)
-}
-
 var gView struct {
 	content map[string]*export.View
 	mu      sync.Mutex
@@ -397,3 +389,9 @@ func GetMetricViewWithLabels(tbl string, lbls []string) *export.View {
 	}
 	return view
 }
+
+func init() {
+	if export.RegisterTableDefine(SingleMetricTable) != nil {
+		panic(moerr.NewInternalError("metric table already registered"))
+	}
+}
diff --git a/pkg/util/trace/buffer_pipe.go b/pkg/util/trace/buffer_pipe.go
index 856615808..41c796015 100644
--- a/pkg/util/trace/buffer_pipe.go
+++ b/pkg/util/trace/buffer_pipe.go
@@ -77,8 +77,6 @@ func (t batchSqlHandler) NewItemBuffer(name string) bp.ItemBuffer[bp.HasName, an
 	switch name {
 	case MOSpanType:
 		f = genSpanBatchSql
-	case MORawLogType:
-		f = genLogBatchSql
 	case MOLogType:
 		f = genZapLogBatchSql
 	case MOStatementType:
@@ -192,50 +190,6 @@ func genSpanBatchSql(in []IBuffer2SqlItem, buf *bytes.Buffer) any {
 
 var logStackFormatter atomic.Value
 
-func genLogBatchSql(in []IBuffer2SqlItem, buf *bytes.Buffer) any {
-	buf.Reset()
-	if len(in) == 0 {
-		logutil.Debugf("genLogBatchSql empty")
-		return ""
-	}
-
-	buf.WriteString(fmt.Sprintf("insert into %s.%s ", StatsDatabase, logInfoTbl))
-	buf.WriteString("(")
-	buf.WriteString("`span_id`")
-	buf.WriteString(", `statement_id`")
-	buf.WriteString(", `node_uuid`")
-	buf.WriteString(", `node_type`")
-	buf.WriteString(", `timestamp`")
-	buf.WriteString(", `name`")
-	buf.WriteString(", `level`")
-	buf.WriteString(", `caller`")
-	buf.WriteString(", `message`")
-	buf.WriteString(", `extra`")
-	buf.WriteString(") values ")
-
-	moNode := GetNodeResource()
-
-	for _, item := range in {
-		s, ok := item.(*MOLog)
-		if !ok {
-			panic("Not MOLog")
-		}
-		buf.WriteString("(")
-		buf.WriteString(fmt.Sprintf(`"%s"`, s.SpanID.String()))
-		buf.WriteString(fmt.Sprintf(`, "%s"`, s.TraceID.String()))
-		buf.WriteString(fmt.Sprintf(`, "%s"`, moNode.NodeUuid))                                                 // node_uuid
-		buf.WriteString(fmt.Sprintf(`, "%s"`, moNode.NodeType))                                                 // node_type
-		buf.WriteString(fmt.Sprintf(`, "%s"`, nanoSec2DatetimeString(s.Timestamp)))                             // timestamp
-		buf.WriteString(fmt.Sprintf(`, "%s"`, quote(s.Name)))                                                   // log level
-		buf.WriteString(fmt.Sprintf(`, "%s"`, s.Level.String()))                                                // log level
-		buf.WriteString(fmt.Sprintf(`, "%s"`, quote(fmt.Sprintf(logStackFormatter.Load().(string), s.Caller)))) // caller
-		buf.WriteString(fmt.Sprintf(`, "%s"`, quote(s.Message)))                                                // message
-		buf.WriteString(fmt.Sprintf(`, "%s"`, quote(s.Extra)))                                                  // extra
-		buf.WriteString("),")
-	}
-	return string(buf.Next(buf.Len() - 1))
-}
-
 func genZapLogBatchSql(in []IBuffer2SqlItem, buf *bytes.Buffer) any {
 	buf.Reset()
 	if len(in) == 0 {
@@ -407,12 +361,12 @@ func (t batchCSVHandler) NewItemBuffer(name string) bp.ItemBuffer[bp.HasName, an
 	var f genBatchFunc = genCsvData
 	logutil.Debugf("NewItemBuffer name: %s", name)
 	switch name {
-	case MOSpanType:
-	case MORawLogType:
-	case MOLogType:
-	case MOStatementType:
+	case MOStatementType, SingleStatementTable.GetName():
 		opts = append(opts, bufferWithFilterItemFunc(filterTraceInsertSql))
 	case MOErrorType:
+	case MOSpanType:
+	case MOLogType:
+	case MORawLogType:
 	default:
 		panic(moerr.NewInternalError("unknown type %s", name))
 	}
@@ -463,7 +417,8 @@ func (t batchCSVHandler) NewItemBatchHandler(ctx context.Context) func(b any) {
 
 type CsvFields interface {
 	bp.HasName
-	CsvFields() []string
+	GetRow() *export.Row
+	CsvFields(row *export.Row) []string
 }
 
 var QuoteFieldFunc = func(buf *bytes.Buffer, value string, enclose rune) string {
@@ -497,14 +452,14 @@ func genCsvData(in []IBuffer2SqlItem, buf *bytes.Buffer) any {
 	}
 	opts := export.CommonCsvOptions
 
-	writer := GetTracerProvider().writerFactory(DefaultContext(), StatsDatabase, i)
-
+	ts := util.Now()
+	row := i.GetRow()
 	for _, i := range in {
 		item, ok := i.(CsvFields)
 		if !ok {
 			panic("not MalCsv, dont support output CSV")
 		}
-		fields := item.CsvFields()
+		fields := item.CsvFields(row)
 		for idx, field := range fields {
 			if idx > 0 {
 				buf.WriteRune(opts.FieldTerminator)
@@ -519,6 +474,10 @@ func genCsvData(in []IBuffer2SqlItem, buf *bytes.Buffer) any {
 		}
 		buf.WriteRune(opts.Terminator)
 	}
+
+	writer := GetTracerProvider().writerFactory(DefaultContext(), StatsDatabase, i,
+		export.WithTimestamp(ts),
+		export.WithPathBuilder(row.Table.PathBuilder))
 	return NewCSVRequest(writer, buf.String())
 }
 
diff --git a/pkg/util/trace/config.go b/pkg/util/trace/config.go
index a4f4096c4..7beba2f5f 100644
--- a/pkg/util/trace/config.go
+++ b/pkg/util/trace/config.go
@@ -16,6 +16,7 @@ package trace
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
@@ -39,9 +40,9 @@ const (
 const (
 	MOStatementType = "statement"
 	MOSpanType      = "span"
-	MORawLogType    = "raw_log"
 	MOLogType       = "log"
 	MOErrorType     = "error"
+	MORawLogType    = rawLogTbl
 )
 
 // tracerProviderConfig.
@@ -53,29 +54,29 @@ type tracerProviderConfig struct {
 	// registered.
 	spanProcessors []SpanProcessor
 
-	enable bool // see SetEnable
+	enable bool // SetEnable
 
 	// idGenerator is used to generate all Span and Trace IDs when needed.
 	idGenerator IDGenerator
 
 	// resource contains attributes representing an entity that produces telemetry.
-	resource *Resource // see WithMOVersion, WithNode,
+	resource *Resource // WithMOVersion, WithNode,
 
 	// TODO: can check span's END
-	debugMode bool // see DebugMode
+	debugMode bool // DebugMode
 
-	batchProcessMode string // see WithBatchProcessMode
+	batchProcessMode string // WithBatchProcessMode
 
 	// writerFactory gen writer for CSV output
-	writerFactory export.FSWriterFactory // see WithFSWriterFactory, default from export.GetFSWriterFactory result
+	writerFactory export.FSWriterFactory // WithFSWriterFactory, default from export.GetFSWriterFactory result
 
-	sqlExecutor func() ie.InternalExecutor // see WithSQLExecutor
+	sqlExecutor func() ie.InternalExecutor // WithSQLExecutor
 	// needInit control table schema create
-	needInit bool // see WithInitAction
+	needInit bool // WithInitAction
 
-	exportInterval time.Duration // see WithExportInterval
+	exportInterval time.Duration //  WithExportInterval
 	// longQueryTime unit ns
-	longQueryTime int64 // see WithLongQueryTime
+	longQueryTime int64 //  WithLongQueryTime
 
 	mux sync.RWMutex
 }
@@ -113,20 +114,20 @@ type TracerProviderOption interface {
 	apply(*tracerProviderConfig)
 }
 
-type tracerProviderOptionFunc func(config *tracerProviderConfig)
+type tracerProviderOption func(config *tracerProviderConfig)
 
-func (f tracerProviderOptionFunc) apply(config *tracerProviderConfig) {
+func (f tracerProviderOption) apply(config *tracerProviderConfig) {
 	f(config)
 }
 
-func WithMOVersion(v string) tracerProviderOptionFunc {
+func WithMOVersion(v string) tracerProviderOption {
 	return func(config *tracerProviderConfig) {
 		config.resource.Put("version", v)
 	}
 }
 
 // WithNode give id as NodeId, t as NodeType
-func WithNode(uuid string, t string) tracerProviderOptionFunc {
+func WithNode(uuid string, t string) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.resource.Put("Node", &MONodeResource{
 			NodeUuid: uuid,
@@ -135,43 +136,43 @@ func WithNode(uuid string, t string) tracerProviderOptionFunc {
 	}
 }
 
-func EnableTracer(enable bool) tracerProviderOptionFunc {
+func EnableTracer(enable bool) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.SetEnable(enable)
 	}
 }
 
-func WithFSWriterFactory(f export.FSWriterFactory) tracerProviderOptionFunc {
-	return tracerProviderOptionFunc(func(cfg *tracerProviderConfig) {
+func WithFSWriterFactory(f export.FSWriterFactory) tracerProviderOption {
+	return tracerProviderOption(func(cfg *tracerProviderConfig) {
 		cfg.writerFactory = f
 	})
 }
 
-func WithExportInterval(secs int) tracerProviderOptionFunc {
-	return tracerProviderOptionFunc(func(cfg *tracerProviderConfig) {
+func WithExportInterval(secs int) tracerProviderOption {
+	return tracerProviderOption(func(cfg *tracerProviderConfig) {
 		cfg.exportInterval = time.Second * time.Duration(secs)
 	})
 }
 
-func WithLongQueryTime(secs float64) tracerProviderOptionFunc {
-	return tracerProviderOptionFunc(func(cfg *tracerProviderConfig) {
+func WithLongQueryTime(secs float64) tracerProviderOption {
+	return tracerProviderOption(func(cfg *tracerProviderConfig) {
 		cfg.longQueryTime = int64(float64(time.Second) * secs)
 	})
 }
 
-func DebugMode(debug bool) tracerProviderOptionFunc {
+func DebugMode(debug bool) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.debugMode = debug
 	}
 }
 
-func WithBatchProcessMode(mode string) tracerProviderOptionFunc {
+func WithBatchProcessMode(mode string) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.batchProcessMode = mode
 	}
 }
 
-func WithSQLExecutor(f func() ie.InternalExecutor) tracerProviderOptionFunc {
+func WithSQLExecutor(f func() ie.InternalExecutor) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.mux.Lock()
 		defer cfg.mux.Unlock()
@@ -179,7 +180,7 @@ func WithSQLExecutor(f func() ie.InternalExecutor) tracerProviderOptionFunc {
 	}
 }
 
-func WithInitAction(init bool) tracerProviderOptionFunc {
+func WithInitAction(init bool) tracerProviderOption {
 	return func(cfg *tracerProviderConfig) {
 		cfg.mux.Lock()
 		defer cfg.mux.Unlock()
@@ -216,6 +217,9 @@ func (t TraceID) IsZero() bool {
 }
 
 func (t TraceID) String() string {
+	if t.IsZero() {
+		return "0"
+	}
 	return uuid.UUID(t).String()
 }
 
@@ -237,6 +241,9 @@ func (s *SpanID) IsZero() bool {
 }
 
 func (s SpanID) String() string {
+	if s.IsZero() {
+		return "0"
+	}
 	return hex.EncodeToString(s[:])
 }
 
@@ -252,6 +259,10 @@ func IsSpanField(field zapcore.Field) bool {
 	return field.Key == SpanFieldKey
 }
 
+func ContextField(ctx context.Context) zap.Field {
+	return SpanField(SpanFromContext(ctx).SpanContext())
+}
+
 // SpanContext contains identifying trace information about a Span.
 type SpanContext struct {
 	TraceID TraceID `json:"trace_id"`
@@ -321,7 +332,7 @@ type SpanConfig struct {
 	// NewRoot identifies a Span as the root Span for a new trace. This is
 	// commonly used when an existing trace crosses trust boundaries and the
 	// remote parent span context should be ignored for security.
-	NewRoot bool `json:"NewRoot"` // see WithNewRoot
+	NewRoot bool `json:"NewRoot"` // WithNewRoot
 	parent  Span `json:"-"`
 }
 
diff --git a/pkg/util/trace/mo_trace.go b/pkg/util/trace/mo_trace.go
index 5c417cfbe..7e50e348c 100644
--- a/pkg/util/trace/mo_trace.go
+++ b/pkg/util/trace/mo_trace.go
@@ -26,6 +26,7 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/util/export"
 	"sync"
 	"unsafe"
 
@@ -150,22 +151,25 @@ func (s *MOSpan) Free() {
 }
 
 func (s *MOSpan) GetName() string {
-	return MOSpanType
-}
-
-func (s *MOSpan) CsvFields() []string {
-	var result []string
-	result = append(result, s.SpanID.String())
-	result = append(result, s.TraceID.String())
-	result = append(result, s.parent.SpanContext().SpanID.String())
-	result = append(result, GetNodeResource().NodeUuid)
-	result = append(result, GetNodeResource().NodeType)
-	result = append(result, s.Name.String())
-	result = append(result, nanoSec2DatetimeString(s.StartTimeNS))
-	result = append(result, nanoSec2DatetimeString(s.EndTimeNS))
-	result = append(result, fmt.Sprintf("%d", s.Duration)) // Duration
-	result = append(result, s.tracer.provider.resource.String())
-	return result
+	return spanView.OriginTable.GetName()
+}
+
+func (s *MOSpan) GetRow() *export.Row { return spanView.OriginTable.GetRow() }
+
+func (s *MOSpan) CsvFields(row *export.Row) []string {
+	row.Reset()
+	row.SetColumnVal(rawItemCol, spanView.Table)
+	row.SetColumnVal(spanIDCol, s.SpanID.String())
+	row.SetColumnVal(stmtIDCol, s.TraceID.String())
+	row.SetColumnVal(parentSpanIDCol, s.parent.SpanContext().SpanID.String())
+	row.SetColumnVal(nodeUUIDCol, GetNodeResource().NodeUuid)
+	row.SetColumnVal(nodeTypeCol, GetNodeResource().NodeType)
+	row.SetColumnVal(spanNameCol, s.Name.String())
+	row.SetColumnVal(startTimeCol, nanoSec2DatetimeString(s.StartTimeNS))
+	row.SetColumnVal(endTimeCol, nanoSec2DatetimeString(s.EndTimeNS))
+	row.SetColumnVal(durationCol, fmt.Sprintf("%d", s.Duration)) // Duration
+	row.SetColumnVal(resourceCol, s.tracer.provider.resource.String())
+	return row.ToStrings()
 }
 
 func (s *MOSpan) End(options ...SpanEndOption) {
diff --git a/pkg/util/trace/report_error.go b/pkg/util/trace/report_error.go
index defc8bdf5..23df63dbb 100644
--- a/pkg/util/trace/report_error.go
+++ b/pkg/util/trace/report_error.go
@@ -16,9 +16,11 @@ package trace
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"unsafe"
 
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/util"
 	"github.com/matrixorigin/matrixone/pkg/util/errutil"
@@ -37,7 +39,7 @@ type MOErrorHolder struct {
 }
 
 func (h *MOErrorHolder) GetName() string {
-	return MOErrorType
+	return errorView.OriginTable.GetName()
 }
 
 func (h *MOErrorHolder) Size() int64 {
@@ -47,22 +49,26 @@ func (h *MOErrorHolder) Free() {
 	h.Error = nil
 }
 
-func (h *MOErrorHolder) CsvFields() []string {
-	var span Span
+func (h *MOErrorHolder) GetRow() *export.Row { return errorView.OriginTable.GetRow() }
+
+func (h *MOErrorHolder) CsvFields(row *export.Row) []string {
+	row.Reset()
+	row.SetColumnVal(rawItemCol, errorView.Table)
+	row.SetColumnVal(timestampCol, nanoSec2DatetimeString(h.Timestamp))
+	row.SetColumnVal(nodeUUIDCol, GetNodeResource().NodeUuid)
+	row.SetColumnVal(nodeTypeCol, GetNodeResource().NodeType)
+	row.SetColumnVal(errorCol, h.Error.Error())
+	row.SetColumnVal(stackCol, fmt.Sprintf(errorFormatter.Load().(string), h.Error))
+	var moError *moerr.Error
+	if errors.As(h.Error, &moError) {
+		row.SetColumnVal(errCodeCol, fmt.Sprintf("%d", moError.ErrorCode()))
+	}
 	if ct := errutil.GetContextTracer(h.Error); ct != nil && ct.Context() != nil {
-		span = SpanFromContext(ct.Context())
-	} else {
-		span = SpanFromContext(DefaultContext())
+		span := SpanFromContext(ct.Context())
+		row.SetColumnVal(stmtIDCol, span.SpanContext().TraceID.String())
+		row.SetColumnVal(spanIDCol, span.SpanContext().SpanID.String())
 	}
-	var result []string
-	result = append(result, span.SpanContext().TraceID.String())
-	result = append(result, span.SpanContext().SpanID.String())
-	result = append(result, GetNodeResource().NodeUuid)
-	result = append(result, GetNodeResource().NodeType)
-	result = append(result, h.Error.Error())
-	result = append(result, fmt.Sprintf(errorFormatter.Load().(string), h.Error))
-	result = append(result, nanoSec2DatetimeString(h.Timestamp))
-	return result
+	return row.ToStrings()
 }
 
 func (h *MOErrorHolder) Format(s fmt.State, verb rune) { errbase.FormatError(h.Error, s, verb) }
diff --git a/pkg/util/trace/report_log.go b/pkg/util/trace/report_log.go
index 9a7e77ebc..486e31890 100644
--- a/pkg/util/trace/report_log.go
+++ b/pkg/util/trace/report_log.go
@@ -15,106 +15,17 @@
 package trace
 
 import (
-	"context"
-	"fmt"
-	"sync/atomic"
 	"time"
 	"unsafe"
 
-	"github.com/matrixorigin/matrixone/pkg/util/stack"
-	"go.uber.org/zap/buffer"
-
-	"github.com/matrixorigin/matrixone/pkg/util"
 	"github.com/matrixorigin/matrixone/pkg/util/batchpipe"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
 
 	"go.uber.org/zap"
+	"go.uber.org/zap/buffer"
 	"go.uber.org/zap/zapcore"
 )
 
-func init() {
-	SetLogLevel(zapcore.DebugLevel)
-}
-
-var _ batchpipe.HasName = (*MOLog)(nil)
-var _ IBuffer2SqlItem = (*MOLog)(nil)
-var _ CsvFields = (*MOLog)(nil)
-
-type MOLog struct {
-	TraceID   TraceID       `json:"trace_id"`
-	SpanID    SpanID        `json:"span_id"`
-	Timestamp util.TimeNano `json:"timestamp"`
-	Level     zapcore.Level `json:"level"`
-	Caller    stack.Frame   `json:"caller"` // like "util/trace/trace.go:666"
-	Name      string        `json:"name"`
-	Message   string        `json:"Message"`
-	Extra     string        `json:"extra"` // like json text
-}
-
-func newMOLog() *MOLog {
-	return &MOLog{}
-}
-
-func (MOLog) GetName() string {
-	return MORawLogType
-}
-
-func (l MOLog) Size() int64 {
-	return int64(unsafe.Sizeof(l)) + int64(len(l.Message))
-}
-
-func (l MOLog) Free() {}
-
-func (l MOLog) CsvFields() []string {
-	var result []string
-	result = append(result, l.TraceID.String())
-	result = append(result, l.SpanID.String())
-	result = append(result, GetNodeResource().NodeUuid)
-	result = append(result, GetNodeResource().NodeType)
-	result = append(result, nanoSec2DatetimeString(l.Timestamp))
-	result = append(result, l.Name)
-	result = append(result, l.Level.String())
-	result = append(result, fmt.Sprintf(logStackFormatter.Load().(string), l.Caller))
-	result = append(result, l.Message)
-	result = append(result, l.Extra)
-	return result
-}
-
-var logLevelEnabler atomic.Value
-
-func SetLogLevel(l zapcore.LevelEnabler) {
-	logLevelEnabler.Store(l)
-}
-
-func ReportLog(ctx context.Context, level zapcore.Level, depth int, formatter string, args ...any) {
-	if !logLevelEnabler.Load().(zapcore.LevelEnabler).Enabled(level) {
-		return
-	}
-	if !GetTracerProvider().IsEnable() {
-		return
-	}
-	_, newSpan := Start(DefaultContext(), "ReportLog")
-	defer newSpan.End()
-
-	sc := SpanFromContext(ctx).SpanContext()
-	if sc.IsEmpty() {
-		sc = *DefaultSpanContext()
-	}
-	log := newMOLog()
-	log.TraceID = sc.TraceID
-	log.SpanID = sc.SpanID
-	log.Timestamp = util.NowNS()
-	log.Level = level
-	log.Caller = stack.Caller(depth + 1)
-	log.Message = fmt.Sprintf(formatter, args...)
-	log.Extra = "{}"
-	export.GetGlobalBatchProcessor().Collect(DefaultContext(), log)
-}
-
-func ContextField(ctx context.Context) zap.Field {
-	return SpanField(SpanFromContext(ctx).SpanContext())
-}
-
 var _ batchpipe.HasName = (*MOZapLog)(nil)
 var _ IBuffer2SqlItem = (*MOZapLog)(nil)
 var _ CsvFields = (*MOZapLog)(nil)
@@ -134,7 +45,7 @@ func newMOZap() *MOZapLog {
 }
 
 func (m *MOZapLog) GetName() string {
-	return MOLogType
+	return logView.OriginTable.GetName()
 }
 
 // Size 计算近似值
@@ -150,19 +61,22 @@ func (m *MOZapLog) Free() {
 	m.Extra = ""
 }
 
-func (m *MOZapLog) CsvFields() []string {
-	var result []string
-	result = append(result, m.SpanContext.TraceID.String())
-	result = append(result, m.SpanContext.SpanID.String())
-	result = append(result, GetNodeResource().NodeUuid)
-	result = append(result, GetNodeResource().NodeType)
-	result = append(result, time2DatetimeString(m.Timestamp))
-	result = append(result, m.LoggerName)
-	result = append(result, m.Level.String())
-	result = append(result, m.Caller)
-	result = append(result, m.Message)
-	result = append(result, m.Extra)
-	return result
+func (m *MOZapLog) GetRow() *export.Row { return logView.OriginTable.GetRow() }
+
+func (m *MOZapLog) CsvFields(row *export.Row) []string {
+	row.Reset()
+	row.SetColumnVal(rawItemCol, logView.Table)
+	row.SetColumnVal(stmtIDCol, m.SpanContext.TraceID.String())
+	row.SetColumnVal(spanIDCol, m.SpanContext.SpanID.String())
+	row.SetColumnVal(nodeUUIDCol, GetNodeResource().NodeUuid)
+	row.SetColumnVal(nodeTypeCol, GetNodeResource().NodeType)
+	row.SetColumnVal(timestampCol, time2DatetimeString(m.Timestamp))
+	row.SetColumnVal(loggerNameCol, m.LoggerName)
+	row.SetColumnVal(levelCol, m.Level.String())
+	row.SetColumnVal(callerCol, m.Caller)
+	row.SetColumnVal(messageCol, m.Message)
+	row.SetColumnVal(extraCol, m.Extra)
+	return row.ToStrings()
 }
 
 func ReportZap(jsonEncoder zapcore.Encoder, entry zapcore.Entry, fields []zapcore.Field) (*buffer.Buffer, error) {
diff --git a/pkg/util/trace/report_statement.go b/pkg/util/trace/report_statement.go
index 14ce49554..748f91b31 100644
--- a/pkg/util/trace/report_statement.go
+++ b/pkg/util/trace/report_statement.go
@@ -17,6 +17,7 @@ package trace
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -69,7 +70,7 @@ type StatementInfo struct {
 }
 
 func (s *StatementInfo) GetName() string {
-	return MOStatementType
+	return SingleStatementTable.GetName()
 }
 
 func (s *StatementInfo) Size() int64 {
@@ -87,37 +88,43 @@ func (s *StatementInfo) Free() {
 	s.Error = nil
 }
 
-func (s *StatementInfo) CsvFields() []string {
+func (s *StatementInfo) GetRow() *export.Row { return SingleStatementTable.GetRow() }
+
+func (s *StatementInfo) CsvFields(row *export.Row) []string {
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.exported = true
-	var result []string
-	result = append(result, uuid.UUID(s.StatementID).String())
-	result = append(result, uuid.UUID(s.TransactionID).String())
-	result = append(result, uuid.UUID(s.SessionID).String())
-	result = append(result, s.Account)
-	result = append(result, s.User)
-	result = append(result, s.Host)
-	result = append(result, s.Database)
-	result = append(result, s.Statement)
-	result = append(result, s.StatementTag)
-	result = append(result, s.StatementFingerprint)
-	result = append(result, GetNodeResource().NodeUuid)
-	result = append(result, GetNodeResource().NodeType)
-	result = append(result, nanoSec2DatetimeString(s.RequestAt))
-	result = append(result, nanoSec2DatetimeString(s.ResponseAt))
-	result = append(result, fmt.Sprintf("%d", s.Duration))
-	result = append(result, s.Status.String())
-	if s.Error == nil {
-		result = append(result, "")
-	} else {
-		result = append(result, fmt.Sprintf("%s", s.Error))
+	row.Reset()
+	row.SetColumnVal(stmtIDCol, uuid.UUID(s.StatementID).String())
+	row.SetColumnVal(txnIDCol, uuid.UUID(s.TransactionID).String())
+	row.SetColumnVal(sesIDCol, uuid.UUID(s.SessionID).String())
+	row.SetColumnVal(accountCol, s.Account)
+	row.SetColumnVal(userCol, s.User)
+	row.SetColumnVal(hostCol, s.Host)
+	row.SetColumnVal(dbCol, s.Database)
+	row.SetColumnVal(stmtCol, s.Statement)
+	row.SetColumnVal(stmtTagCol, s.StatementTag)
+	row.SetColumnVal(stmtFgCol, s.StatementFingerprint)
+	row.SetColumnVal(nodeUUIDCol, GetNodeResource().NodeUuid)
+	row.SetColumnVal(nodeTypeCol, GetNodeResource().NodeType)
+	row.SetColumnVal(reqAtCol, nanoSec2DatetimeString(s.RequestAt))
+	row.SetColumnVal(respAtCol, nanoSec2DatetimeString(s.ResponseAt))
+	row.SetColumnVal(durationCol, fmt.Sprintf("%d", s.Duration))
+	row.SetColumnVal(statusCol, s.Status.String())
+	if s.Error != nil {
+		var moError *moerr.Error
+		errCode := moerr.ErrInfo
+		if errors.As(s.Error, &moError) {
+			errCode = moError.ErrorCode()
+		}
+		row.SetColumnVal(errCodeCol, fmt.Sprintf("%d", errCode))
+		row.SetColumnVal(errorCol, fmt.Sprintf("%s", s.Error))
 	}
-	result = append(result, s.ExecPlan2Json())
-	result = append(result, fmt.Sprintf("%d", s.RowsRead))
-	result = append(result, fmt.Sprintf("%d", s.BytesScan))
+	row.SetColumnVal(execPlanCol, s.ExecPlan2Json())
+	row.SetColumnVal(rowsReadCol, fmt.Sprintf("%d", s.RowsRead))
+	row.SetColumnVal(bytesScanCol, fmt.Sprintf("%d", s.BytesScan))
 
-	return result
+	return row.ToStrings()
 }
 
 // ExecPlan2Json return ExecPlan Serialized json-str
diff --git a/pkg/util/trace/schema.go b/pkg/util/trace/schema.go
index 68c4a8e20..382e55223 100644
--- a/pkg/util/trace/schema.go
+++ b/pkg/util/trace/schema.go
@@ -16,9 +16,7 @@ package trace
 
 import (
 	"context"
-	"fmt"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
-	"strings"
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
@@ -27,12 +25,181 @@ import (
 )
 
 const (
-	SystemDBConst    = "system"
-	StatsDatabase    = SystemDBConst
-	spanInfoTbl      = "span_info"
-	logInfoTbl       = "log_info"
+	SystemDBConst = "system"
+	StatsDatabase = SystemDBConst
+	// statementInfoTbl is an EXTERNAL table
 	statementInfoTbl = "statement_info"
-	errorInfoTbl     = "error_info"
+	rawLogTbl        = "rawlog"
+
+	// spanInfoTbl is a view
+	spanInfoTbl  = "span_info"
+	logInfoTbl   = "log_info"
+	errorInfoTbl = "error_info"
+
+	uuidColType        = "varchar(36)"
+	spanIDType         = "varchar(16)"
+	datetime6Type      = "datetime(6)"
+	bigintUnsignedType = "bigint unsigned"
+	stringType         = "varchar(1024)"
+
+	jsonColumnDEFAULT = "{}"
+)
+
+var (
+	stmtIDCol    = export.Column{Name: "statement_id", Type: uuidColType, Default: "0", Comment: "statement uniq id"}
+	txnIDCol     = export.Column{Name: "transaction_id", Type: uuidColType, Default: "0", Comment: "txn uniq id"}
+	sesIDCol     = export.Column{Name: "sessionID", Type: uuidColType, Default: "0", Comment: "session uniq id"}
+	accountCol   = export.Column{Name: "account", Type: stringType, Default: "", Comment: "account name"}
+	userCol      = export.Column{Name: "user", Type: stringType, Default: "", Comment: "user name"}
+	hostCol      = export.Column{Name: "host", Type: stringType, Default: "", Comment: "user client ip"}
+	dbCol        = export.Column{Name: "database", Type: stringType, Default: "", Comment: "what database current session stay in."}
+	stmtCol      = export.Column{Name: "statement", Type: "TEXT", Default: "", Comment: "sql statement"}
+	stmtTagCol   = export.Column{Name: "statement_tag", Type: "TEXT", Default: "", Comment: "note tag in statement(Reserved)"}
+	stmtFgCol    = export.Column{Name: "statement_fingerprint", Type: "TEXT", Default: "", Comment: "note tag in statement(Reserved)"}
+	nodeUUIDCol  = export.Column{Name: "node_uuid", Type: uuidColType, Default: "0", Comment: "node uuid, which node gen this data."}
+	nodeTypeCol  = export.Column{Name: "node_type", Type: "varchar(64)", Default: "node", Comment: "node type in MO, val in [DN, CN, LOG]"}
+	reqAtCol     = export.Column{Name: "request_at", Type: datetime6Type, Default: "", Comment: "request accept datetime"}
+	respAtCol    = export.Column{Name: "response_at", Type: datetime6Type, Default: "", Comment: "response send datetime"}
+	durationCol  = export.Column{Name: "duration", Type: bigintUnsignedType, Default: "0", Comment: "exec time, unit: ns"}
+	statusCol    = export.Column{Name: "status", Type: "varchar(32)", Default: "Running", Comment: "sql statement running status, enum: Running, Success, Failed"}
+	errorCol     = export.Column{Name: "error", Type: "TEXT", Default: "", Comment: "error message"}
+	execPlanCol  = export.Column{Name: "exec_plan", Type: "JSON", Default: jsonColumnDEFAULT, Comment: "statement execution plan"}
+	rowsReadCol  = export.Column{Name: "rows_read", Type: bigintUnsignedType, Default: "0", Comment: "rows read total"}
+	bytesScanCol = export.Column{Name: "bytes_scan", Type: bigintUnsignedType, Default: "0", Comment: "bytes scan total"}
+
+	SingleStatementTable = &export.Table{
+		Database: StatsDatabase,
+		Table:    statementInfoTbl,
+		Columns: []export.Column{
+			stmtIDCol,
+			txnIDCol,
+			sesIDCol,
+			accountCol,
+			userCol,
+			hostCol,
+			dbCol,
+			stmtCol,
+			stmtTagCol,
+			stmtFgCol,
+			nodeUUIDCol,
+			nodeTypeCol,
+			reqAtCol,
+			respAtCol,
+			durationCol,
+			statusCol,
+			errCodeCol,
+			errorCol,
+			execPlanCol,
+			rowsReadCol,
+			bytesScanCol,
+		},
+		PrimaryKeyColumn: []export.Column{stmtIDCol},
+		Engine:           export.ExternalTableEngine,
+		Comment:          "record each statement and stats info",
+		PathBuilder:      export.NewAccountDatePathBuilder(),
+		AccountColumn:    &accountCol,
+	}
+
+	rawItemCol      = export.Column{Name: "raw_item", Type: stringType, Comment: "raw log item"}
+	timestampCol    = export.Column{Name: "timestamp", Type: datetime6Type, Comment: "timestamp of action"}
+	loggerNameCol   = export.Column{Name: "logger_name", Type: stringType, Comment: "logger name"}
+	levelCol        = export.Column{Name: "level", Type: stringType, Comment: "log level, enum: debug, info, warn, error, panic, fatal"}
+	callerCol       = export.Column{Name: "caller", Type: stringType, Comment: "where it log, like: package/file.go:123"}
+	messageCol      = export.Column{Name: "message", Type: "TEXT", Comment: "log message"}
+	extraCol        = export.Column{Name: "extra", Type: "JSON", Default: jsonColumnDEFAULT, Comment: "log dynamic fields"}
+	errCodeCol      = export.Column{Name: "err_code", Type: stringType, Default: "0"}
+	stackCol        = export.Column{Name: "stack", Type: "varchar(4096)"}
+	spanIDCol       = export.Column{Name: "span_id", Type: spanIDType, Default: "0", Comment: "span uniq id"}
+	parentSpanIDCol = export.Column{Name: "parent_span_id", Type: spanIDType, Default: "0", Comment: "parent span uniq id"}
+	spanNameCol     = export.Column{Name: "span_name", Type: stringType, Default: "", Comment: "span name, for example: step name of execution plan, function name in code, ..."}
+	startTimeCol    = export.Column{Name: "start_time", Type: datetime6Type, Default: ""}
+	endTimeCol      = export.Column{Name: "end_time", Type: datetime6Type, Default: ""}
+	resourceCol     = export.Column{Name: "resource", Type: "JSON", Default: jsonColumnDEFAULT, Comment: "static resource information"}
+
+	SingleRowLogTable = &export.Table{
+		Database: StatsDatabase,
+		Table:    rawLogTbl,
+		Columns: []export.Column{
+			rawItemCol,
+			nodeUUIDCol,
+			nodeTypeCol,
+			spanIDCol,
+			stmtIDCol,
+			loggerNameCol,
+			timestampCol,
+			levelCol,
+			callerCol,
+			messageCol,
+			extraCol,
+			errCodeCol,
+			errorCol,
+			stackCol,
+			spanNameCol,
+			parentSpanIDCol,
+			startTimeCol,
+			endTimeCol,
+			durationCol,
+			resourceCol,
+		},
+		PrimaryKeyColumn: nil,
+		Engine:           export.ExternalTableEngine,
+		Comment:          "read merge data from log, error, span",
+		PathBuilder:      export.NewAccountDatePathBuilder(),
+		AccountColumn:    nil,
+	}
+
+	logView = &export.View{
+		Database:    StatsDatabase,
+		Table:       logInfoTbl,
+		OriginTable: SingleRowLogTable,
+		Columns: []export.Column{
+			stmtIDCol,
+			spanIDCol,
+			nodeUUIDCol,
+			nodeTypeCol,
+			timestampCol,
+			loggerNameCol,
+			levelCol,
+			callerCol,
+			messageCol,
+			extraCol,
+		},
+		Condition: &export.ViewSingleCondition{Column: rawItemCol, Table: logInfoTbl},
+	}
+
+	errorView = &export.View{
+		Database:    StatsDatabase,
+		Table:       errorInfoTbl,
+		OriginTable: SingleRowLogTable,
+		Columns: []export.Column{
+			timestampCol,
+			errCodeCol,
+			errorCol,
+			nodeUUIDCol,
+			nodeTypeCol,
+			stackCol,
+		},
+		Condition: &export.ViewSingleCondition{Column: rawItemCol, Table: errorInfoTbl},
+	}
+
+	spanView = &export.View{
+		Database:    StatsDatabase,
+		Table:       spanInfoTbl,
+		OriginTable: SingleRowLogTable,
+		Columns: []export.Column{
+			spanIDCol,
+			stmtIDCol,
+			parentSpanIDCol,
+			nodeUUIDCol,
+			nodeTypeCol,
+			spanNameCol,
+			startTimeCol,
+			endTimeCol,
+			durationCol,
+			resourceCol,
+		},
+		Condition: &export.ViewSingleCondition{Column: rawItemCol, Table: spanInfoTbl},
+	}
 )
 
 const (
@@ -96,12 +263,8 @@ PRIMARY KEY (statement_id)
 )`
 )
 
-var initDDLs = []struct{ sqlPrefix, filePrefix string }{
-	{sqlCreateStatementInfoTable, MOStatementType},
-	{sqlCreateSpanInfoTable, MOSpanType},
-	{sqlCreateLogInfoTable, MOLogType},
-	{sqlCreateErrorInfoTable, MOErrorType},
-}
+var tables = []*export.Table{SingleStatementTable, SingleRowLogTable}
+var views = []*export.View{logView, errorView, spanView}
 
 // InitSchemaByInnerExecutor init schema, which can access db by io.InternalExecutor on any Node.
 func InitSchemaByInnerExecutor(ctx context.Context, ieFactory func() ie.InternalExecutor) error {
@@ -117,8 +280,6 @@ func InitSchemaByInnerExecutor(ctx context.Context, ieFactory func() ie.Internal
 		return nil
 	}
 
-	optFactory := GetOptionFactory(export.ExternalTableEngine)
-
 	if err := mustExec(sqlCreateDBConst); err != nil {
 		return err
 	}
@@ -129,10 +290,13 @@ func InitSchemaByInnerExecutor(ctx context.Context, ieFactory func() ie.Internal
 	}()
 	instant := time.Now()
 
-	for _, ddl := range initDDLs {
-		opts := optFactory(StatsDatabase, ddl.filePrefix)
-		sql := opts.FormatDdl(ddl.sqlPrefix) + opts.GetTableOptions(nil)
-		if err := mustExec(sql); err != nil {
+	for _, tbl := range tables {
+		if err := mustExec(tbl.ToCreateSql(true)); err != nil {
+			return err
+		}
+	}
+	for _, v := range views {
+		if err := mustExec(v.ToCreateSql(true)); err != nil {
 			return err
 		}
 	}
@@ -141,47 +305,10 @@ func InitSchemaByInnerExecutor(ctx context.Context, ieFactory func() ie.Internal
 	return nil
 }
 
-var _ export.TableOptions = (*CsvTableOptions)(nil)
-
-type CsvTableOptions struct {
-	Formatter string
-	DbName    string
-	TblName   string
-}
-
-func getExternalTableDDLPrefix(sql string) string {
-	return strings.Replace(sql, "CREATE TABLE", "CREATE EXTERNAL TABLE", 1)
-}
-
-func (o *CsvTableOptions) FormatDdl(ddl string) string {
-	return getExternalTableDDLPrefix(ddl)
-}
-
-func (o *CsvTableOptions) GetCreateOptions() string {
-	return "EXTERNAL "
-}
-
-func (o *CsvTableOptions) GetTableOptions(builder export.PathBuilder) string {
-	if builder == nil {
-		builder = export.NewDBTablePathBuilder()
-	}
-	if len(o.Formatter) > 0 {
-		return fmt.Sprintf(o.Formatter, builder.BuildETLPath(o.DbName, o.TblName))
-	}
-	return ""
-}
-
-func GetOptionFactory(engine string) func(db, tbl string) export.TableOptions {
-	var infileFormatter = ` infile{"filepath"="etl:%s","compression"="none"}` +
-		` FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n' IGNORE 0 lines`
-	switch engine {
-	case export.NormalTableEngine:
-		return func(_, _ string) export.TableOptions { return export.NoopTableOptions{} }
-	case export.ExternalTableEngine:
-		return func(db, tbl string) export.TableOptions {
-			return &CsvTableOptions{Formatter: infileFormatter, DbName: db, TblName: tbl}
+func init() {
+	for _, tbl := range tables {
+		if old := export.RegisterTableDefine(tbl); old != nil {
+			panic(moerr.NewInternalError("table already registered: %s", old.GetIdentify()))
 		}
-	default:
-		panic(moerr.NewInternalError("unknown engine: %s", engine))
 	}
 }
diff --git a/pkg/util/trace/trace.go b/pkg/util/trace/trace.go
index 0489424b0..b77b7289b 100644
--- a/pkg/util/trace/trace.go
+++ b/pkg/util/trace/trace.go
@@ -76,7 +76,7 @@ func Init(ctx context.Context, opts ...TracerProviderOption) (context.Context, e
 	}
 
 	// init tool dependence
-	logutil.SetLogReporter(&logutil.TraceReporter{ReportLog: ReportLog, ReportZap: ReportZap, LevelSignal: SetLogLevel, ContextField: ContextField})
+	logutil.SetLogReporter(&logutil.TraceReporter{ReportZap: ReportZap, ContextField: ContextField})
 	logutil.SpanFieldKey.Store(SpanFieldKey)
 	errutil.SetErrorReporter(ReportError)
 	export.SetDefaultContextFunc(DefaultContext)
@@ -103,13 +103,11 @@ func initExporter(ctx context.Context, config *tracerProviderConfig) error {
 	case config.batchProcessMode == InternalExecutor:
 		// register buffer pipe implements
 		export.Register(&MOSpan{}, NewBufferPipe2SqlWorker(defaultOptions...))
-		export.Register(&MOLog{}, NewBufferPipe2SqlWorker(defaultOptions...))
 		export.Register(&MOZapLog{}, NewBufferPipe2SqlWorker(defaultOptions...))
 		export.Register(&StatementInfo{}, NewBufferPipe2SqlWorker(defaultOptions...))
 		export.Register(&MOErrorHolder{}, NewBufferPipe2SqlWorker(defaultOptions...))
 	case config.batchProcessMode == FileService:
 		export.Register(&MOSpan{}, NewBufferPipe2CSVWorker(defaultOptions...))
-		export.Register(&MOLog{}, NewBufferPipe2CSVWorker(defaultOptions...))
 		export.Register(&MOZapLog{}, NewBufferPipe2CSVWorker(defaultOptions...))
 		export.Register(&StatementInfo{}, NewBufferPipe2CSVWorker(defaultOptions...))
 		export.Register(&MOErrorHolder{}, NewBufferPipe2CSVWorker(defaultOptions...))


Test patch:
diff --git a/pkg/logutil/report_test.go b/pkg/logutil/report_test.go
index 1a652a3bb..ddf32beee 100644
--- a/pkg/logutil/report_test.go
+++ b/pkg/logutil/report_test.go
@@ -24,12 +24,10 @@ import (
 )
 
 func TestNoop(t *testing.T) {
-	noopReportLog(context.Background(), zapcore.DebugLevel, 0, "msg")
 	require.Equal(t, zap.String("span", "{}"), noopContextField(context.Background()))
 	buf, err := noopReportZap(nil, zapcore.Entry{}, nil)
 	require.Equal(t, nil, err)
 	require.Equal(t, "", buf.String())
-	noopLevelSignal(zap.ErrorLevel)
 	require.Equal(t, zap.Bool(MOInternalFiledKeyNoopReport, true), NoReportFiled())
 }
 
diff --git a/pkg/util/export/table_test.go b/pkg/util/export/table_test.go
index d154f4123..41839c344 100644
--- a/pkg/util/export/table_test.go
+++ b/pkg/util/export/table_test.go
@@ -51,7 +51,7 @@ func TestNoopTableOptions_FormatDdl(t *testing.T) {
 }
 
 var dummyStrColumn = Column{Name: "str", Type: "varchar(32)", Default: "", Comment: "str column"}
-var dummyStrCreateSql = "`str` varchar(32) COMMENT \"str column\""
+var dummyStrCreateSql = "`str` varchar(32) NOT NULL COMMENT \"str column\""
 var dummyInt64Column = Column{Name: "int64", Type: "BIGINT", Default: "0", Comment: "int64 column"}
 var dummyInt64CreateSql = "`int64` BIGINT DEFAULT \"0\" COMMENT \"int64 column\""
 var dummyFloat64Column = Column{Name: "float64", Type: "DOUBLE", Default: "0.0", Comment: "float64 column"}
@@ -61,7 +61,7 @@ var dummyTable = &Table{
 	Database:         "db_dummy",
 	Table:            "tbl_dummy",
 	Columns:          []Column{dummyStrColumn, dummyInt64Column, dummyFloat64Column},
-	PrimaryKeyColumn: nil,
+	PrimaryKeyColumn: []Column{dummyStrColumn, dummyInt64Column},
 	Engine:           ExternalTableEngine,
 	Comment:          "dummy table",
 	TableOptions:     NoopTableOptions{},
@@ -71,11 +71,13 @@ var dummyTableCreateExistsSql = "CREATE TABLE IF NOT EXISTS `db_dummy`.`tbl_dumm
 	"\n" + dummyStrCreateSql +
 	",\n" + dummyInt64CreateSql +
 	",\n" + dummyFloat64CreateSql +
+	",\nPRIMARY KEY (`" + dummyStrColumn.Name + "`, `" + dummyInt64Column.Name + "`)" +
 	"\n)"
 var dummyTableCreateSql = "CREATE TABLE `db_dummy`.`tbl_dummy`(" +
 	"\n" + dummyStrCreateSql +
 	",\n" + dummyInt64CreateSql +
 	",\n" + dummyFloat64CreateSql +
+	",\nPRIMARY KEY (`" + dummyStrColumn.Name + "`, `" + dummyInt64Column.Name + "`)" +
 	"\n)"
 
 type dummyCondition struct{}
diff --git a/pkg/util/export/type_test.go b/pkg/util/export/type_test.go
index 87264f73d..b818c273a 100644
--- a/pkg/util/export/type_test.go
+++ b/pkg/util/export/type_test.go
@@ -167,7 +167,7 @@ func TestPathBuilder(t *testing.T) {
 		},
 		{
 			name:  "metric_log",
-			field: field{builder: NewMetricLogPathBuilder()},
+			field: field{builder: NewAccountDatePathBuilder()},
 			args: args{
 				account:  "user",
 				typ:      MergeLogTypeLog,
diff --git a/pkg/util/metric/metric_test.go b/pkg/util/metric/metric_test.go
index 0f1916daf..8af84cc62 100644
--- a/pkg/util/metric/metric_test.go
+++ b/pkg/util/metric/metric_test.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
-	"github.com/matrixorigin/matrixone/pkg/util/trace"
 	"io"
 	"net/http"
 	"strings"
@@ -127,7 +126,7 @@ func TestDescExtra(t *testing.T) {
 	assert.Equal(t, extra.labels[2].GetName(), "xy")
 }
 
-var dummyOptionsFactory = trace.GetOptionFactory(export.NormalTableEngine)
+var dummyOptionsFactory = export.GetOptionFactory(export.NormalTableEngine)
 
 func TestCreateTable(t *testing.T) {
 	buf := new(bytes.Buffer)
diff --git a/pkg/util/trace/buffer_pipe_test.go b/pkg/util/trace/buffer_pipe_test.go
index 0f2ba159f..71acf29f8 100644
--- a/pkg/util/trace/buffer_pipe_test.go
+++ b/pkg/util/trace/buffer_pipe_test.go
@@ -25,9 +25,7 @@ import (
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
-	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/util/export"
-	"github.com/matrixorigin/matrixone/pkg/util/stack"
 	"github.com/stretchr/testify/require"
 
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
@@ -49,10 +47,7 @@ var testBaseBuffer2SqlOption = []bufferOption{bufferWithSizeThreshold(1 * mpool.
 var traceIDSpanIDColumnStr string
 var traceIDSpanIDCsvStr string
 
-var gCtrlSqlCh = make(chan struct{}, 1)
-
-func noopReportLog(context.Context, zapcore.Level, int, string, ...any) {}
-func noopReportError(context.Context, error, int)                       {}
+func noopReportError(context.Context, error, int) {}
 
 func init() {
 	time.Local = time.FixedZone("CST", 0) // set time-zone +0000
@@ -61,7 +56,7 @@ func init() {
 		EnableTracer(true),
 		WithMOVersion("v0.test.0"),
 		WithNode("node_uuid", NodeTypeStandalone),
-		WithBatchProcessMode(InternalExecutor),
+		WithBatchProcessMode(FileService),
 		WithFSWriterFactory(func(ctx context.Context, _ string, _ batchpipe.HasName, _ ...export.FSWriterOption) io.StringWriter {
 			return os.Stdout
 		}),
@@ -74,7 +69,6 @@ func init() {
 	); err != nil {
 		panic(err)
 	}
-	logutil.SetLogReporter(&logutil.TraceReporter{ReportLog: noopReportLog, LevelSignal: SetLogLevel})
 	errutil.SetErrorReporter(noopReportError)
 
 	sc := SpanFromContext(DefaultContext()).SpanContext()
@@ -140,7 +134,7 @@ func Test_batchSqlHandler_NewItemBuffer_Check_genBatchFunc(t1 *testing.T) {
 		want genBatchFunc
 	}{
 		{name: "span_type", args: args{opt: opts, name: MOSpanType}, want: genSpanBatchSql},
-		{name: "log_type", args: args{opt: opts, name: MORawLogType}, want: genLogBatchSql},
+		{name: "log_type", args: args{opt: opts, name: MOLogType}, want: genZapLogBatchSql},
 		{name: "statement_type", args: args{opt: opts, name: MOStatementType},
 			want: genStatementBatchSql},
 		{name: "error_type", args: args{opt: opts, name: MOErrorType},
@@ -158,309 +152,6 @@ func Test_batchSqlHandler_NewItemBuffer_Check_genBatchFunc(t1 *testing.T) {
 	}
 }
 
-func Test_buffer2Sql_GetBatch_AllType(t *testing.T) {
-	gCtrlSqlCh <- struct{}{}
-	type fields struct {
-		Reminder      batchpipe.Reminder
-		sizeThreshold int64
-	}
-	sc := SpanContextWithIDs(_1TraceID, _1SpanID)
-	defaultFields := fields{
-		Reminder:      batchpipe.NewConstantClock(15 * time.Second),
-		sizeThreshold: mpool.MB,
-	}
-	type args struct {
-		in  []IBuffer2SqlItem
-		buf *bytes.Buffer
-	}
-	tests := []struct {
-		name     string
-		fields   fields
-		args     args
-		wantFunc genBatchFunc
-		want     string
-	}{
-		{
-			name:   "single_error",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOErrorHolder{Error: err1, Timestamp: uint64(0)},
-				},
-				buf: buf,
-			},
-			wantFunc: genErrorBatchSql,
-			want: `insert into system.error_info (` +
-				"`statement_id`, `span_id`, `node_uuid`, `node_type`, `err_code`, `stack`, `timestamp`" +
-				`) values (` + traceIDSpanIDColumnStr + `, "node_uuid", "Standalone", "internal error: test1", "internal error: test1", "1970-01-01 00:00:00.000000")`,
-		},
-		{
-			name:   "multi_error",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOErrorHolder{Error: err1, Timestamp: uint64(0)},
-					&MOErrorHolder{Error: err2, Timestamp: uint64(time.Millisecond + time.Microsecond)},
-				},
-				buf: buf,
-			},
-			wantFunc: genErrorBatchSql,
-			want: `insert into system.error_info (` +
-				"`statement_id`, `span_id`, `node_uuid`, `node_type`, `err_code`, `stack`, `timestamp`" +
-				`) values (` + traceIDSpanIDColumnStr + `, "node_uuid", "Standalone", "internal error: test1", "internal error: test1", "1970-01-01 00:00:00.000000")` +
-				`,(` + traceIDSpanIDColumnStr + `, "node_uuid", "Standalone", "test2: internal error: test1", "test2: internal error: test1", "1970-01-01 00:00:00.001001")`,
-		},
-		{
-			name:   "single_log",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(0),
-						Level:     zapcore.InfoLevel,
-						Caller:    stack.Caller(0),
-						Message:   "info message",
-						Extra:     "{}",
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genLogBatchSql,
-			want: `insert into system.log_info (` +
-				"`span_id`, `statement_id`, `node_uuid`, `node_type`, `timestamp`, `name`, `level`, `caller`, `message`, `extra`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "", "info", "Test_buffer2Sql_GetBatch_AllType", "info message", "{}")`,
-		},
-		{
-			name:   "multi_log",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(0),
-						Level:     zapcore.InfoLevel,
-						Caller:    stack.Caller(0),
-						Message:   "info message",
-						Extra:     "{}",
-					},
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(time.Millisecond + time.Microsecond),
-						Level:     zapcore.DebugLevel,
-						Caller:    stack.Caller(0),
-						Message:   "debug message",
-						Extra:     "{}",
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genLogBatchSql,
-			want: `insert into system.log_info (` +
-				"`span_id`, `statement_id`, `node_uuid`, `node_type`, `timestamp`, `name`, `level`, `caller`, `message`, `extra`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "", "info", "Test_buffer2Sql_GetBatch_AllType", "info message", "{}")` +
-				`,("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.001001", "", "debug", "Test_buffer2Sql_GetBatch_AllType", "debug message", "{}")`,
-		},
-		{
-			name:   "single_span",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOSpan{
-						SpanConfig:  SpanConfig{SpanContext: SpanContext{TraceID: _1TraceID, SpanID: _1SpanID}, parent: noopSpan{}},
-						Name:        *bytes.NewBuffer([]byte("span1")),
-						StartTimeNS: util.TimeNano(0),
-						EndTimeNS:   util.TimeNano(time.Microsecond),
-						Duration:    util.TimeNano(time.Microsecond),
-						tracer:      gTracer.(*MOTracer),
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genSpanBatchSql,
-			want: `insert into system.span_info (` +
-				"`span_id`, `statement_id`, `parent_span_id`, `node_uuid`, `node_type`, `resource`, `name`, `start_time`, `end_time`, `duration`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "0000000000000000", "node_uuid", "Standalone", "{\"Node\":{\"node_uuid\":\"node_uuid\",\"node_type\":\"Standalone\"},\"version\":\"v0.test.0\"}", "span1", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00.000001", 1000)`,
-		},
-		{
-			name:   "multi_span",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOSpan{
-						SpanConfig:  SpanConfig{SpanContext: SpanContext{TraceID: _1TraceID, SpanID: _1SpanID}, parent: noopSpan{}},
-						Name:        *bytes.NewBuffer([]byte("span1")),
-						StartTimeNS: util.TimeNano(0),
-						EndTimeNS:   util.TimeNano(time.Microsecond),
-						Duration:    util.TimeNano(time.Microsecond),
-						tracer:      gTracer.(*MOTracer),
-					},
-					&MOSpan{
-						SpanConfig:  SpanConfig{SpanContext: SpanContext{TraceID: _1TraceID, SpanID: _2SpanID}, parent: noopSpan{}},
-						Name:        *bytes.NewBuffer([]byte("span2")),
-						StartTimeNS: util.TimeNano(time.Microsecond),
-						EndTimeNS:   util.TimeNano(time.Millisecond),
-						Duration:    util.TimeNano(time.Millisecond - time.Microsecond),
-						tracer:      gTracer.(*MOTracer),
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genSpanBatchSql,
-			want: `insert into system.span_info (` +
-				"`span_id`, `statement_id`, `parent_span_id`, `node_uuid`, `node_type`, `resource`, `name`, `start_time`, `end_time`, `duration`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "0000000000000000", "node_uuid", "Standalone", "{\"Node\":{\"node_uuid\":\"node_uuid\",\"node_type\":\"Standalone\"},\"version\":\"v0.test.0\"}", "span1", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00.000001", 1000)` +
-				`,("0000000000000002", "00000000-0000-0000-0000-000000000001", "0000000000000000", "node_uuid", "Standalone", "{\"Node\":{\"node_uuid\":\"node_uuid\",\"node_type\":\"Standalone\"},\"version\":\"v0.test.0\"}", "span2", "1970-01-01 00:00:00.000001", "1970-01-01 00:00:00.001000", 999000)`,
-		},
-		{
-			name:   "single_statement",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&StatementInfo{
-						StatementID:          _1TraceID,
-						TransactionID:        _1TxnID,
-						SessionID:            _1SesID,
-						Account:              "MO",
-						User:                 "moroot",
-						Database:             "system",
-						Statement:            "show tables",
-						StatementFingerprint: "show tables",
-						StatementTag:         "",
-						RequestAt:            util.TimeNano(0),
-						ExecPlan:             nil,
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genStatementBatchSql,
-			want: `insert into system.statement_info (` +
-				"`statement_id`, `transaction_id`, `session_id`, `account`, `user`, `host`, `database`, `statement`, `statement_tag`, `statement_fingerprint`, `node_uuid`, `node_type`, `request_at`, `response_at`, `status`, `error`, `duration`, `exec_plan`" +
-				`) values ("00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000001", "MO", "moroot", "", "system", "show tables", "show tables", "", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00.000000", 0, "Running", "", "{\"code\":200,\"message\":\"NO ExecPlan Serialize function\",\"steps\":null,\"success\":false,\"uuid\":\"00000000-0000-0000-0000-000000000001\"}")`,
-		},
-		{
-			name:   "multi_statement",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&StatementInfo{
-						StatementID:          _1TraceID,
-						TransactionID:        _1TxnID,
-						SessionID:            _1SesID,
-						Account:              "MO",
-						User:                 "moroot",
-						Database:             "system",
-						Statement:            "show tables",
-						StatementFingerprint: "show tables",
-						StatementTag:         "",
-						RequestAt:            util.TimeNano(0),
-						ExecPlan:             nil,
-					},
-					&StatementInfo{
-						StatementID:          _2TraceID,
-						TransactionID:        _1TxnID,
-						SessionID:            _1SesID,
-						Account:              "MO",
-						User:                 "moroot",
-						Database:             "system",
-						Statement:            "show databases",
-						StatementFingerprint: "show databases",
-						StatementTag:         "dcl",
-						RequestAt:            util.TimeNano(time.Microsecond),
-						ResponseAt:           util.TimeNano(time.Microsecond + time.Second),
-						Duration:             uint64(time.Second),
-						ExecPlan:             nil,
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genStatementBatchSql,
-			want: `insert into system.statement_info (` +
-				"`statement_id`, `transaction_id`, `session_id`, `account`, `user`, `host`, `database`, `statement`, `statement_tag`, `statement_fingerprint`, `node_uuid`, `node_type`, `request_at`, `response_at`, `status`, `error`, `duration`, `exec_plan`" +
-				`) values ("00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000001", "MO", "moroot", "", "system", "show tables", "show tables", "", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "1970-01-01 00:00:00.000000", 0, "Running", "", "{\"code\":200,\"message\":\"NO ExecPlan Serialize function\",\"steps\":null,\"success\":false,\"uuid\":\"00000000-0000-0000-0000-000000000001\"}")` +
-				`,("00000000-0000-0000-0000-000000000002", "00000000-0000-0000-0000-000000000001", "00000000-0000-0000-0000-000000000001", "MO", "moroot", "", "system", "show databases", "show databases", "dcl", "node_uuid", "Standalone", "1970-01-01 00:00:00.000001", "1970-01-01 00:00:01.000001", 1000000000, "Running", "", "{\"code\":200,\"message\":\"NO ExecPlan Serialize function\",\"steps\":null,\"success\":false,\"uuid\":\"00000000-0000-0000-0000-000000000002\"}")`,
-		},
-		{
-			name:   "single_zap",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOZapLog{
-						Level:       zapcore.InfoLevel,
-						SpanContext: &sc,
-						Timestamp:   time.Unix(0, 0),
-						Caller:      "trace/buffer_pipe_sql_test.go:100",
-						Message:     "info message",
-						Extra:       "{}",
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genZapLogBatchSql,
-			want: `insert into system.log_info (` +
-				"`span_id`, `statement_id`, `node_uuid`, `node_type`, `timestamp`, `name`, `level`, `caller`, `message`, `extra`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "", "info", "trace/buffer_pipe_sql_test.go:100", "info message", "{}")`,
-		},
-		{
-			name:   "multi_zap",
-			fields: defaultFields,
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOZapLog{
-						Level:       zapcore.InfoLevel,
-						SpanContext: &sc,
-						Timestamp:   time.Unix(0, 0),
-						Caller:      "trace/buffer_pipe_sql_test.go:100",
-						Message:     "info message",
-						Extra:       "{}",
-					},
-					&MOZapLog{
-						Level:       zapcore.DebugLevel,
-						SpanContext: &sc,
-						Timestamp:   time.Unix(0, int64(time.Microsecond+time.Millisecond)),
-						Caller:      "trace/buffer_pipe_sql_test.go:100",
-						Message:     "debug message",
-						Extra:       "{}",
-					},
-				},
-				buf: buf,
-			},
-			wantFunc: genZapLogBatchSql,
-			want: `insert into system.log_info (` +
-				"`span_id`, `statement_id`, `node_uuid`, `node_type`, `timestamp`, `name`, `level`, `caller`, `message`, `extra`" +
-				`) values ("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.000000", "", "info", "trace/buffer_pipe_sql_test.go:100", "info message", "{}")` +
-				`,("0000000000000001", "00000000-0000-0000-0000-000000000001", "node_uuid", "Standalone", "1970-01-01 00:00:00.001001", "", "debug", "trace/buffer_pipe_sql_test.go:100", "debug message", "{}")`,
-		},
-	}
-
-	errorFormatter.Store("%v")
-	logStackFormatter.Store("%n")
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			bpImpl := NewBufferPipe2SqlWorker(
-				bufferWithReminder(tt.fields.Reminder),
-				bufferWithSizeThreshold(tt.fields.sizeThreshold),
-			)
-			b := bpImpl.NewItemBuffer(tt.args.in[0].GetName())
-			t.Logf("buffer.type: %s", b.(*buffer2Sql).GetBufferType())
-			for _, i := range tt.args.in {
-				b.Add(i)
-			}
-			if got := b.(*buffer2Sql).genBatchFunc; reflect.ValueOf(got).Pointer() != reflect.ValueOf(tt.wantFunc).Pointer() {
-				t.Errorf("buffer2Sql's genBatchFunc = %v, want %v", got, tt.wantFunc)
-			}
-
-			got := b.GetBatch(tt.args.buf)
-			require.Equal(t, tt.want, got)
-			t.Logf("GetBatch() = %v", got)
-		})
-	}
-	<-gCtrlSqlCh
-}
-
 func Test_buffer2Sql_IsEmpty(t *testing.T) {
 	type fields struct {
 		Reminder      batchpipe.Reminder
@@ -487,7 +178,7 @@ func Test_buffer2Sql_IsEmpty(t *testing.T) {
 			name: "not_empty",
 			fields: fields{
 				Reminder:      batchpipe.NewConstantClock(time.Hour),
-				buf:           []IBuffer2SqlItem{&MOLog{}},
+				buf:           []IBuffer2SqlItem{&MOZapLog{}},
 				sizeThreshold: mpool.GB,
 				batchFunc:     nil,
 			},
@@ -535,7 +226,7 @@ func Test_buffer2Sql_Reset(t *testing.T) {
 			name: "not_empty",
 			fields: fields{
 				Reminder:      batchpipe.NewConstantClock(time.Hour),
-				buf:           []IBuffer2SqlItem{&MOLog{}},
+				buf:           []IBuffer2SqlItem{&MOZapLog{}},
 				sizeThreshold: mpool.GB,
 				batchFunc:     nil,
 			},
@@ -704,7 +395,7 @@ func Test_withGenBatchFunc(t *testing.T) {
 		want genBatchFunc
 	}{
 		{name: "genSpanBatchSql", args: args{f: genSpanBatchSql}, want: genSpanBatchSql},
-		{name: "genLogBatchSql", args: args{f: genLogBatchSql}, want: genLogBatchSql},
+		{name: "genLogBatchSql", args: args{f: genZapLogBatchSql}, want: genZapLogBatchSql},
 		{name: "genStatementBatchSql", args: args{f: genStatementBatchSql}, want: genStatementBatchSql},
 		{name: "genErrorBatchSql", args: args{f: genErrorBatchSql}, want: genErrorBatchSql},
 	}
@@ -747,6 +438,7 @@ func Test_withSizeThreshold(t *testing.T) {
 }
 
 /*
+var gCtrlSqlCh = make(chan struct{}, 1)
 func Test_batchSqlHandler_NewItemBatchHandler(t1 *testing.T) {
 	gCtrlSqlCh <- struct{}{}
 	type fields struct {
@@ -825,7 +517,7 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `0000000000000001,00000000-0000-0000-0000-000000000001,0000000000000000,node_uuid,Standalone,span1,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000001,1000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
+			want: `span_info,node_uuid,Standalone,0000000000000001,00000000-0000-0000-0000-000000000001,,,,,,{},0,,,span1,0,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000001,1000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
 `,
 		},
 		{
@@ -851,56 +543,8 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `0000000000000001,00000000-0000-0000-0000-000000000001,0000000000000000,node_uuid,Standalone,span1,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000001,1000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
-0000000000000002,00000000-0000-0000-0000-000000000001,0000000000000000,node_uuid,Standalone,span2,1970-01-01 00:00:00.000001,1970-01-01 00:00:00.001000,999000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
-`,
-		},
-		{
-			name: "single_log",
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(0),
-						Level:     zapcore.InfoLevel,
-						Caller:    stack.Caller(0),
-						Message:   "info message",
-						Extra:     "{}",
-					},
-				},
-				buf: buf,
-			},
-			want: `00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.000000,,info,Test_genCsvData,info message,{}
-`,
-		},
-		{
-			name: "multi_log",
-			args: args{
-				in: []IBuffer2SqlItem{
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(0),
-						Level:     zapcore.InfoLevel,
-						Caller:    stack.Caller(0),
-						Message:   "info message",
-						Extra:     "{}",
-					},
-					&MOLog{
-						TraceID:   _1TraceID,
-						SpanID:    _1SpanID,
-						Timestamp: uint64(time.Millisecond + time.Microsecond),
-						Level:     zapcore.DebugLevel,
-						Caller:    stack.Caller(0),
-						Message:   "debug message",
-						Extra:     "{}",
-					},
-				},
-				buf: buf,
-			},
-			want: `00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.000000,,info,Test_genCsvData,info message,{}
-00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.001001,,debug,Test_genCsvData,debug message,{}
+			want: `span_info,node_uuid,Standalone,0000000000000001,00000000-0000-0000-0000-000000000001,,,,,,{},0,,,span1,0,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000001,1000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
+span_info,node_uuid,Standalone,0000000000000002,00000000-0000-0000-0000-000000000001,,,,,,{},0,,,span2,0,1970-01-01 00:00:00.000001,1970-01-01 00:00:00.001000,999000,"{""Node"":{""node_uuid"":""node_uuid"",""node_type"":""Standalone""},""version"":""v0.test.0""}"
 `,
 		},
 		{
@@ -918,7 +562,7 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.000000,,info,trace/buffer_pipe_sql_test.go:912,info message,{}
+			want: `log_info,node_uuid,Standalone,0000000000000001,00000000-0000-0000-0000-000000000001,,1970-01-01 00:00:00.000000,info,trace/buffer_pipe_sql_test.go:912,info message,{},0,,,,0,,,0,{}
 `,
 		},
 		{
@@ -944,8 +588,8 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.000000,,info,trace/buffer_pipe_sql_test.go:939,info message,{}
-00000000-0000-0000-0000-000000000001,0000000000000001,node_uuid,Standalone,1970-01-01 00:00:00.001001,,debug,trace/buffer_pipe_sql_test.go:939,debug message,{}
+			want: `log_info,node_uuid,Standalone,0000000000000001,00000000-0000-0000-0000-000000000001,,1970-01-01 00:00:00.000000,info,trace/buffer_pipe_sql_test.go:939,info message,{},0,,,,0,,,0,{}
+log_info,node_uuid,Standalone,0000000000000001,00000000-0000-0000-0000-000000000001,,1970-01-01 00:00:00.001001,debug,trace/buffer_pipe_sql_test.go:939,debug message,{},0,,,,0,,,0,{}
 `,
 		},
 		{
@@ -968,7 +612,7 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,0,Running,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
+			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,0,Running,0,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
 `,
 		},
 		{
@@ -1008,8 +652,8 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,0,Running,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
-00000000-0000-0000-0000-000000000002,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show databases,dcl,show databases,node_uuid,Standalone,1970-01-01 00:00:00.000001,1970-01-01 00:00:01.000001,1000001000,Failed,internal error: test error,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000002""}",0,0
+			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,0,Running,0,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
+00000000-0000-0000-0000-000000000002,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show databases,dcl,show databases,node_uuid,Standalone,1970-01-01 00:00:00.000001,1970-01-01 00:00:01.000001,1000001000,Failed,20101,internal error: test error,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000002""}",0,0
 `,
 		},
 		{
@@ -1020,7 +664,7 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: traceIDSpanIDCsvStr + `,node_uuid,Standalone,internal error: test1,internal error: test1,1970-01-01 00:00:00.000000
+			want: `error_info,node_uuid,Standalone,0,0,,1970-01-01 00:00:00.000000,,,,{},20101,internal error: test1,internal error: test1,,0,,,0,{}
 `,
 		},
 		{
@@ -1032,8 +676,8 @@ func Test_genCsvData(t *testing.T) {
 				},
 				buf: buf,
 			},
-			want: traceIDSpanIDCsvStr + `,node_uuid,Standalone,internal error: test1,internal error: test1,1970-01-01 00:00:00.000000
-` + traceIDSpanIDCsvStr + `,node_uuid,Standalone,test2: internal error: test1,test2: internal error: test1,1970-01-01 00:00:00.001001
+			want: `error_info,node_uuid,Standalone,0,0,,1970-01-01 00:00:00.000000,,,,{},20101,internal error: test1,internal error: test1,,0,,,0,{}
+error_info,node_uuid,Standalone,0,0,,1970-01-01 00:00:00.001001,,,,{},20101,test2: internal error: test1,test2: internal error: test1,,0,,,0,{}
 `,
 		},
 	}
@@ -1117,9 +761,9 @@ func Test_genCsvData_LongQueryTime(t *testing.T) {
 				buf:    buf,
 				queryT: int64(time.Second),
 			},
-			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,999999999,Running,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
-00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,999999999,Running,,"{""code"":200,""message"":""no exec plan""}",0,0
-00000000-0000-0000-0000-000000000002,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show databases,dcl,show databases,node_uuid,Standalone,1970-01-01 00:00:00.000001,1970-01-01 00:00:01.000001,1000000000,Failed,internal error: test error,"{""key"":""val""}",1,1
+			want: `00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,999999999,Running,0,,"{""code"":200,""message"":""NO ExecPlan Serialize function"",""steps"":null,""success"":false,""uuid"":""00000000-0000-0000-0000-000000000001""}",0,0
+00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show tables,,show tables,node_uuid,Standalone,1970-01-01 00:00:00.000000,1970-01-01 00:00:00.000000,999999999,Running,0,,"{""code"":200,""message"":""no exec plan""}",0,0
+00000000-0000-0000-0000-000000000002,00000000-0000-0000-0000-000000000001,00000000-0000-0000-0000-000000000001,MO,moroot,,system,show databases,dcl,show databases,node_uuid,Standalone,1970-01-01 00:00:00.000001,1970-01-01 00:00:01.000001,1000000000,Failed,20101,internal error: test error,"{""key"":""val""}",1,1
 `,
 		},
 	}
diff --git a/pkg/util/trace/report_log_test.go b/pkg/util/trace/report_log_test.go
index 1778f9a29..584993a63 100644
--- a/pkg/util/trace/report_log_test.go
+++ b/pkg/util/trace/report_log_test.go
@@ -15,7 +15,6 @@
 package trace
 
 import (
-	"context"
 	"testing"
 	"time"
 
@@ -25,50 +24,6 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
-func TestReportLog(t *testing.T) {
-	type args struct {
-		ctx       context.Context
-		level     zapcore.Level
-		depth     int
-		formatter string
-		args      []any
-	}
-	tests := []struct {
-		name         string
-		enableTracer bool
-		args         args
-	}{
-		{
-			name:         "close",
-			enableTracer: false,
-			args: args{
-				ctx:       ContextWithSpanContext(context.Background(), SpanContextWithIDs(nilTraceID, nilSpanID)),
-				level:     zapcore.InfoLevel,
-				depth:     3,
-				formatter: "info message",
-				args:      []any{},
-			},
-		},
-		{
-			name:         "collect",
-			enableTracer: true,
-			args: args{
-				ctx:       ContextWithSpanContext(context.Background(), SpanContextWithIDs(nilTraceID, nilSpanID)),
-				level:     zapcore.InfoLevel,
-				depth:     3,
-				formatter: "info message",
-				args:      []any{},
-			},
-		},
-	}
-	for _, tt := range tests {
-		GetTracerProvider().SetEnable(tt.enableTracer)
-		t.Run(tt.name, func(t *testing.T) {
-			ReportLog(tt.args.ctx, tt.args.level, tt.args.depth, tt.args.formatter, tt.args.args...)
-		})
-	}
-}
-
 func TestReportZap(t *testing.T) {
 	type args struct {
 		jsonEncoder zapcore.Encoder
diff --git a/pkg/util/trace/schema_test.go b/pkg/util/trace/schema_test.go
index b82cf7280..208b4dae2 100644
--- a/pkg/util/trace/schema_test.go
+++ b/pkg/util/trace/schema_test.go
@@ -89,7 +89,7 @@ func TestInitSchemaByInnerExecutor(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			var wg sync.WaitGroup
-			wg.Add(1 + len(initDDLs))
+			wg.Add(1 + len(tables) + len(views))
 			err := InitSchemaByInnerExecutor(tt.args.ctx, newDummyExecutorFactory(tt.args.ch))
 			require.Equal(t, nil, err)
 			go func() {
@@ -107,7 +107,8 @@ func TestInitSchemaByInnerExecutor(t *testing.T) {
 						continue
 					}
 					idx := strings.Index(sql, "CREATE EXTERNAL TABLE")
-					require.Equal(t, 0, idx)
+					viewIdx := strings.Index(sql, "CREATE VIEW")
+					require.Equal(t, -1, idx|viewIdx)
 				} else {
 					t.Log("exec sql Done.")
 					wg.Wait()
diff --git a/test/cases/tenant/privilege/operate_catalog.result b/test/cases/tenant/privilege/operate_catalog.result
index 1b757dd8d..b50bbf967 100644
--- a/test/cases/tenant/privilege/operate_catalog.result
+++ b/test/cases/tenant/privilege/operate_catalog.result
@@ -80,17 +80,21 @@ drop table system.error_info;
 internal error: do not have privilege to execute the statement
 create table system.A(a int);
 internal error: do not have privilege to execute the statement
-update error_info set statement_id = "1111";
+update statement_info set statement_id = "1111";
 internal error: the external table is not support update operation
-update system.error_info set statement_id = "1111";
-internal error: the external table is not support update operation
-insert into error_info values ("1","1","1","1","1","1","2022-10-09 00:00:00");
-invalid input: cannot insert into external table 'error_info'
-insert into system.error_info values ("1","1","1","1","1","1","2022-10-09 00:00:00");
-invalid input: cannot insert into external table 'error_info'
+update error_info set err_code = "1111";
+internal error: view is not support update operation
+update system.error_info set err_code = "1111";
+internal error: view is not support update operation
+insert into error_info values ("2022-10-09 00:00:00", "1", "1", "1", "1", "1");
+invalid input: cannot insert into view 'error_info'
+insert into system.error_info values ("2022-10-09 00:00:00", "1", "1", "1", "1", "1");
+invalid input: cannot insert into view 'error_info'
 delete from error_info;
-invalid input: cannot delete from external table
+invalid input: cannot delete from view
 delete from system.error_info;
+invalid input: cannot delete from view
+delete from system.statement_info;
 invalid input: cannot delete from external table
 use system_metrics;
 drop table sql_statement_total;
diff --git a/test/cases/tenant/privilege/operate_catalog.sql b/test/cases/tenant/privilege/operate_catalog.sql
index 852caf067..e0e05bf0b 100644
--- a/test/cases/tenant/privilege/operate_catalog.sql
+++ b/test/cases/tenant/privilege/operate_catalog.sql
@@ -43,12 +43,14 @@ drop table system.span_info;
 drop table system.log_info;
 drop table system.error_info;
 create table system.A(a int);
-update error_info set statement_id = "1111";
-update system.error_info set statement_id = "1111";
-insert into error_info values ("1","1","1","1","1","1","2022-10-09 00:00:00");
-insert into system.error_info values ("1","1","1","1","1","1","2022-10-09 00:00:00");
+update statement_info set statement_id = "1111";
+update error_info set err_code = "1111";
+update system.error_info set err_code = "1111";
+insert into error_info values ("2022-10-09 00:00:00", "1", "1", "1", "1", "1");
+insert into system.error_info values ("2022-10-09 00:00:00", "1", "1", "1", "1", "1");
 delete from error_info;
 delete from system.error_info;
+delete from system.statement_info;
 
 use system_metrics;
 drop table sql_statement_total;
@@ -120,4 +122,4 @@ drop table if exists INFORMATION_SCHEMA.CHARACTER_SETS;
 drop table if exists INFORMATION_SCHEMA.TRIGGERS;
 drop table if exists INFORMATION_SCHEMA.TABLES;
 create table INFORMATION_SCHEMA.A(a int);
--- add update,insert,delete
\ No newline at end of file
+-- add update,insert,delete