Repository: oom-ai/oomstore
Pull request #1048 (2022-01-21 02:59:25)

Test/export
this PR does:
1. improve oomcli export integraioin-test
2. resolve: #1055 


Patch:
diff --git a/internal/database/offline/bigquery/store.go b/internal/database/offline/bigquery/store.go
index 515a3225..692b4477 100644
--- a/internal/database/offline/bigquery/store.go
+++ b/internal/database/offline/bigquery/store.go
@@ -44,8 +44,8 @@ func (db *DB) Ping(ctx context.Context) error {
 	return errdefs.WithStack(err)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	q := fmt.Sprintf(`SELECT column_name, data_type FROM %s.INFORMATION_SCHEMA.COLUMNS WHERE table_name = "%s"`, db.datasetID, opt.TableName)
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
+	q := fmt.Sprintf(`SELECT column_name, data_type FROM %s.INFORMATION_SCHEMA.COLUMNS WHERE table_name = "%s"`, db.datasetID, tableName)
 	rows, err := db.Query(q).Read(ctx)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
@@ -69,26 +69,7 @@ func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*typ
 			ValueType: valueType,
 		})
 	}
-	if opt.CheckTimeRange {
-		q = fmt.Sprintf(`
-		SELECT
-			MIN(%s) AS %s,
-			MAX(%s) AS %s
-		FROM %s.%s`, "unix_milli", "min_unix_milli", "unix_milli", "max_unix_milli", db.datasetID, opt.TableName)
-		rows, err = db.Query(q).Read(ctx)
-		if err != nil {
-			return nil, errdefs.WithStack(err)
-		}
-		recordMap := make(map[string]bigquery.Value)
-		err = rows.Next(&recordMap)
-		if err != nil {
-			return nil, errdefs.WithStack(err)
-		}
-		schema.TimeRange = types.DataTableTimeRange{
-			MinUnixMilli: cast.ToInt64(recordMap["min_unix_milli"]),
-			MaxUnixMilli: cast.ToInt64(recordMap["max_unix_milli"]),
-		}
-	}
+
 	return &schema, nil
 }
 
diff --git a/internal/database/offline/mock_offline/store.go b/internal/database/offline/mock_offline/store.go
index 2ecbc4a2..e1d6438d 100644
--- a/internal/database/offline/mock_offline/store.go
+++ b/internal/database/offline/mock_offline/store.go
@@ -152,16 +152,16 @@ func (mr *MockStoreMockRecorder) Snapshot(ctx, opt interface{}) *gomock.Call {
 }
 
 // TableSchema mocks base method.
-func (m *MockStore) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
+func (m *MockStore) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "TableSchema", ctx, opt)
+	ret := m.ctrl.Call(m, "TableSchema", ctx, tableName)
 	ret0, _ := ret[0].(*types.DataTableSchema)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // TableSchema indicates an expected call of TableSchema.
-func (mr *MockStoreMockRecorder) TableSchema(ctx, opt interface{}) *gomock.Call {
+func (mr *MockStoreMockRecorder) TableSchema(ctx, tableName interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TableSchema", reflect.TypeOf((*MockStore)(nil).TableSchema), ctx, opt)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TableSchema", reflect.TypeOf((*MockStore)(nil).TableSchema), ctx, tableName)
 }
diff --git a/internal/database/offline/mysql/store.go b/internal/database/offline/mysql/store.go
index 9e1599f1..118d29de 100644
--- a/internal/database/offline/mysql/store.go
+++ b/internal/database/offline/mysql/store.go
@@ -45,12 +45,12 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	rows, err := db.QueryxContext(ctx, "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = ?", opt.TableName)
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
+	rows, err := db.QueryxContext(ctx, "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = ?", tableName)
 	if err != nil {
 		return nil, err
 	}
-	return sqlutil.SqlxTableSchema(ctx, db.DB, Backend, rows, opt)
+	return sqlutil.SqlxTableSchema(ctx, db, types.BackendMySQL, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
diff --git a/internal/database/offline/postgres/store.go b/internal/database/offline/postgres/store.go
index 0aa502ff..7acc460c 100644
--- a/internal/database/offline/postgres/store.go
+++ b/internal/database/offline/postgres/store.go
@@ -43,12 +43,12 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	rows, err := db.QueryxContext(ctx, "select column_name, data_type from information_schema.columns where table_name = $1", opt.TableName)
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
+	rows, err := db.QueryxContext(ctx, "select column_name, data_type from information_schema.columns where table_name = $1", tableName)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
 	}
-	return sqlutil.SqlxTableSchema(ctx, db.DB, Backend, rows, opt)
+	return sqlutil.SqlxTableSchema(ctx, db, Backend, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
diff --git a/internal/database/offline/redshift/store.go b/internal/database/offline/redshift/store.go
index 0c787e3a..648049df 100644
--- a/internal/database/offline/redshift/store.go
+++ b/internal/database/offline/redshift/store.go
@@ -45,12 +45,12 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	rows, err := db.QueryxContext(ctx, "select column_name, data_type from information_schema.columns where table_name = $1", opt.TableName)
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
+	rows, err := db.QueryxContext(ctx, "select column_name, data_type from information_schema.columns where table_name = $1", tableName)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
 	}
-	return sqlutil.SqlxTableSchema(ctx, db.DB, Backend, rows, opt)
+	return sqlutil.SqlxTableSchema(ctx, db, Backend, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
diff --git a/internal/database/offline/snowflake/store.go b/internal/database/offline/snowflake/store.go
index 49591e31..d7c75921 100644
--- a/internal/database/offline/snowflake/store.go
+++ b/internal/database/offline/snowflake/store.go
@@ -59,12 +59,12 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	rows, err := db.QueryxContext(ctx, "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = ?", opt.TableName)
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
+	rows, err := db.QueryxContext(ctx, "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = ?", tableName)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
 	}
-	return sqlutil.SqlxTableSchema(ctx, db.DB, Backend, rows, opt)
+	return sqlutil.SqlxTableSchema(ctx, db, types.BackendSnowflake, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
diff --git a/internal/database/offline/sqlite/store.go b/internal/database/offline/sqlite/store.go
index 0af6cff5..60547458 100644
--- a/internal/database/offline/sqlite/store.go
+++ b/internal/database/offline/sqlite/store.go
@@ -45,7 +45,7 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
+func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
 	query := `
 		SELECT
 			p.name as column_name,
@@ -54,11 +54,11 @@ func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*typ
 		LEFT OUTER JOIN pragma_table_info((m.name)) AS p
 		WHERE m.type = 'table' AND m.name = ?
 `
-	rows, err := db.QueryxContext(ctx, query, opt.TableName)
+	rows, err := db.QueryxContext(ctx, query, tableName)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
 	}
-	return sqlutil.SqlxTableSchema(ctx, db.DB, Backend, rows, opt)
+	return sqlutil.SqlxTableSchema(ctx, db, types.BackendSQLite, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
diff --git a/internal/database/offline/sqlutil/export_helper.go b/internal/database/offline/sqlutil/export_helper.go
index e2dda196..41d1755d 100644
--- a/internal/database/offline/sqlutil/export_helper.go
+++ b/internal/database/offline/sqlutil/export_helper.go
@@ -13,6 +13,7 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/pkg/errors"
 )
 
 const UNION_ENTITY_QUERY = `
@@ -97,7 +98,7 @@ func buildUnionEntityQuery(params unionEntityQueryParams) (string, []interface{}
 
 	buf := bytes.NewBuffer(nil)
 	if err := t.Execute(buf, params); err != nil {
-		return "", nil, errdefs.WithStack(err)
+		return "", nil, errors.WithStack(err)
 	}
 	return buf.String(), args, nil
 }
@@ -135,7 +136,7 @@ func buildExportQuery(params exportQueryParams) (string, error) {
 
 	buf := bytes.NewBuffer(nil)
 	if err := t.Execute(buf, params); err != nil {
-		return "", errdefs.WithStack(err)
+		return "", errors.WithStack(err)
 	}
 	return buf.String(), nil
 }
@@ -171,10 +172,10 @@ func prepareEntityTable(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.Exp
 		DatasetID:      dbOpt.DatasetID,
 	})
 	if err != nil {
-		return "", errdefs.WithStack(err)
+		return "", err
 	}
 	if err = dbOpt.ExecContext(ctx, query, args); err != nil {
-		return "", errdefs.WithStack(err)
+		return "", err
 	}
 
 	// Step 3: create index on table entity_rows
diff --git a/internal/database/offline/sqlutil/table_schema.go b/internal/database/offline/sqlutil/table_schema.go
index dad6b6ef..b3f1bb57 100644
--- a/internal/database/offline/sqlutil/table_schema.go
+++ b/internal/database/offline/sqlutil/table_schema.go
@@ -2,19 +2,17 @@ package sqlutil
 
 import (
 	"context"
-	"fmt"
-
-	"github.com/oom-ai/oomstore/internal/database/offline"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/offline"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 // SqlxTableSchema returns the schema of the given table.
-func SqlxTableSchema(ctx context.Context, db *sqlx.DB, backend types.BackendType, rows *sqlx.Rows, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
+func SqlxTableSchema(ctx context.Context, store offline.Store, backend types.BackendType, rows *sqlx.Rows) (*types.DataTableSchema, error) {
 	defer rows.Close()
 
 	var schema types.DataTableSchema
@@ -35,27 +33,5 @@ func SqlxTableSchema(ctx context.Context, db *sqlx.DB, backend types.BackendType
 	if len(schema.Fields) == 0 {
 		return nil, errdefs.Errorf("table not found")
 	}
-	if opt.CheckTimeRange {
-		timeRange, err := getCdcTimeRange(ctx, db, opt.TableName, backend)
-		if err != nil {
-			return nil, err
-		}
-		schema.TimeRange = *timeRange
-	}
 	return &schema, nil
 }
-
-func getCdcTimeRange(ctx context.Context, db *sqlx.DB, tableName string, backend types.BackendType) (*types.DataTableTimeRange, error) {
-	qt := dbutil.QuoteFn(backend)
-	var timeRange types.DataTableTimeRange
-	query := fmt.Sprintf(`
-		SELECT
-			MIN(%s) AS %s,
-			MAX(%s) AS %s
-		FROM %s`, qt("unix_milli"), qt("min_unix_milli"), qt("unix_milli"), qt("max_unix_milli"), qt(tableName))
-
-	if err := db.GetContext(ctx, &timeRange, query); err != nil {
-		return nil, errdefs.WithStack(err)
-	}
-	return &timeRange, nil
-}
diff --git a/internal/database/offline/store.go b/internal/database/offline/store.go
index d675cd90..13a1e573 100644
--- a/internal/database/offline/store.go
+++ b/internal/database/offline/store.go
@@ -14,7 +14,7 @@ type Store interface {
 	Push(ctx context.Context, opt PushOpt) error
 
 	CreateTable(ctx context.Context, opt CreateTableOpt) error
-	TableSchema(ctx context.Context, opt TableSchemaOpt) (*types.DataTableSchema, error)
+	TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error)
 	Snapshot(ctx context.Context, opt SnapshotOpt) error
 
 	Ping(ctx context.Context) error
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 8d378f4a..dc213622 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -77,8 +77,3 @@ type CreateTableOpt struct {
 	Features  types.FeatureList
 	TableType types.TableType
 }
-
-type TableSchemaOpt struct {
-	TableName      string
-	CheckTimeRange bool
-}
diff --git a/pkg/oomstore/export.go b/pkg/oomstore/export.go
index 61b915a2..aff8b4f3 100644
--- a/pkg/oomstore/export.go
+++ b/pkg/oomstore/export.go
@@ -64,25 +64,21 @@ func (s *OomStore) ChannelExport(ctx context.Context, opt types.ChannelExportOpt
 		}
 		revision := revisions.Before(opt.UnixMilli)
 		if revision == nil {
-			empty := make(chan types.ExportRecord)
-			close(empty)
-			return &types.ExportResult{
-				Header: append([]string{features[0].Entity().Name}, features.FullNames()...),
-				Data:   empty,
-			}, nil
-		}
-		if revision.SnapshotTable == "" {
-			if err = s.Snapshot(ctx, group.Name); err != nil {
+			return nil, errdefs.Errorf("group %s no feature values up to %d, use a later timestamp", group.Name, opt.UnixMilli)
+		} else {
+			if revision.SnapshotTable == "" {
+				if err = s.Snapshot(ctx, group.Name); err != nil {
+					return nil, err
+				}
+			}
+
+			if revision, err = s.GetRevision(ctx, revision.ID); err != nil {
 				return nil, err
 			}
-		}
-		revision, err = s.GetRevision(ctx, revision.ID)
-		if err != nil {
-			return nil, err
-		}
-		snapshotTables[group.ID] = revision.SnapshotTable
-		if group.Category == types.CategoryStream {
-			cdcTables[group.ID] = revision.CdcTable
+			snapshotTables[group.ID] = revision.SnapshotTable
+			if group.Category == types.CategoryStream {
+				cdcTables[group.ID] = revision.CdcTable
+			}
 		}
 	}
 
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index 62f0fe3d..d95a8093 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -97,15 +97,27 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 
 func (s *OomStore) tableLinkImport(ctx context.Context, opt *importOpt, dataSource *types.TableLinkDataSource) (int, error) {
 	// Make sure all features existing with correct value type
-	tableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
-		TableName: dataSource.TableName,
-	})
+	tableSchema, err := s.offline.TableSchema(ctx, dataSource.TableName)
 	if err != nil {
 		return 0, err
 	}
-	if err = validateTableSchema(tableSchema, opt.features); err != nil {
-		return 0, err
+	validate := func(f *types.Feature) error {
+		for _, field := range tableSchema.Fields {
+			if field.Name == f.Name {
+				if field.ValueType != f.ValueType {
+					return errdefs.Errorf("expect value type '%s', got '%s'", f.ValueType, field.ValueType)
+				}
+				return nil
+			}
+		}
+		return errdefs.Errorf("field '%s' found in target table", f.Name)
+	}
+	for _, feature := range opt.features {
+		if err := validate(feature); err != nil {
+			return 0, err
+		}
 	}
+
 	var revision int64
 	if opt.Revision == nil {
 		revision = time.Now().UnixMilli()
@@ -126,25 +138,6 @@ func (s *OomStore) tableLinkImport(ctx context.Context, opt *importOpt, dataSour
 	return newRevisionID, nil
 }
 
-func validateTableSchema(schema *types.DataTableSchema, features types.FeatureList) error {
-	validate := func(f *types.Feature) error {
-		for _, field := range schema.Fields {
-			if field.Name == f.Name {
-				if field.ValueType != f.ValueType {
-					return errdefs.Errorf("expect value type '%s', got '%s'", f.ValueType, field.ValueType)
-				}
-				return nil
-			}
-		}
-		return errdefs.Errorf("field '%s' found in target table", f.Name)
-	}
-	for _, feature := range features {
-		if err := validate(feature); err != nil {
-			return err
-		}
-	}
-	return nil
-}
 func hasDup(a []string) bool {
 	s := make(map[string]bool)
 	for _, e := range a {
diff --git a/pkg/oomstore/import_stream.go b/pkg/oomstore/import_stream.go
index cbe062e9..47a03bcb 100644
--- a/pkg/oomstore/import_stream.go
+++ b/pkg/oomstore/import_stream.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/internal/database/offline"
 	"github.com/spf13/cast"
 
@@ -37,8 +36,6 @@ func (s *OomStore) ImportStream(ctx context.Context, opt types.ImportStreamOpt)
 		})
 	case types.CSV_READER:
 		return s.csvReaderImportStream(ctx, opt, opt.CsvReaderDataSource)
-	case types.TABLE_LINK:
-		return s.tableLinkImportStream(ctx, opt, opt.TableLinkDataSource)
 	default:
 		return errdefs.Errorf("unsupported data source: %v", opt.DataSourceType)
 	}
@@ -91,41 +88,6 @@ func (s *OomStore) csvReaderImportStream(ctx context.Context, opt types.ImportSt
 	return s.pushStreamingRecords(ctx, records, entity.Name, group, features)
 }
 
-func (s *OomStore) tableLinkImportStream(ctx context.Context, opt types.ImportStreamOpt, dataSource *types.TableLinkDataSource) error {
-	// get group information
-	_, group, features, err := s.getGroupInfo(ctx, opt.GroupName)
-	if err != nil {
-		return err
-	}
-	// get linked table schema
-	tableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
-		TableName:      dataSource.TableName,
-		CheckTimeRange: true,
-	})
-	if err != nil {
-		return err
-	}
-	// validation
-	if err = validateTableSchema(tableSchema, features); err != nil {
-		return err
-	}
-	if err = s.validateRevisions(ctx, group.ID, tableSchema); err != nil {
-		return err
-	}
-
-	_, _, err = s.metadata.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:    tableSchema.TimeRange.MinUnixMilli,
-		GroupID:     group.ID,
-		CdcTable:    &dataSource.TableName,
-		Description: opt.Description,
-		Anchored:    true,
-	})
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
 func (s *OomStore) pushStreamingRecords(ctx context.Context, records []types.StreamRecord, entityName string, group *types.Group, features types.FeatureList) error {
 	buckets := make(map[int64][]types.StreamRecord)
 	for _, record := range records {
@@ -160,7 +122,6 @@ func (s *OomStore) pushStreamingRecords(ctx context.Context, records []types.Str
 	}
 	return nil
 }
-
 func generateStreamRecord(line []interface{}, header []string, group *types.Group, features types.FeatureList) types.StreamRecord {
 	var (
 		entityKey string
@@ -188,29 +149,3 @@ func generateStreamRecord(line []interface{}, header []string, group *types.Grou
 		Values:    values,
 	}
 }
-
-func (s *OomStore) validateRevisions(ctx context.Context, groupID int, schema *types.DataTableSchema) error {
-	revisions, err := s.ListRevision(ctx, &groupID)
-	if err != nil {
-		return err
-	}
-
-	revisionBeforeMin := revisions.Before(schema.TimeRange.MinUnixMilli)
-	revisionBeforeMax := revisions.Before(schema.TimeRange.MaxUnixMilli)
-	if revisionBeforeMax != revisionBeforeMin {
-		return errdefs.Errorf("data table crosses with another offline table %s", revisionBeforeMax.CdcTable)
-	}
-	if revisionBeforeMin != nil {
-		beforeTableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
-			TableName:      revisionBeforeMin.CdcTable,
-			CheckTimeRange: true,
-		})
-		if err != nil {
-			return err
-		}
-		if beforeTableSchema.TimeRange.MaxUnixMilli >= schema.TimeRange.MinUnixMilli {
-			return errdefs.Errorf("data table crosses with another offline table %s", revisionBeforeMin.CdcTable)
-		}
-	}
-	return nil
-}
diff --git a/pkg/oomstore/types/import.go b/pkg/oomstore/types/import.go
index 49e721c9..1f346ee6 100644
--- a/pkg/oomstore/types/import.go
+++ b/pkg/oomstore/types/import.go
@@ -42,5 +42,4 @@ type ImportStreamOpt struct {
 	DataSourceType      DataSourceType
 	CsvFileDataSource   *CsvFileDataSource
 	CsvReaderDataSource *CsvReaderDataSource
-	TableLinkDataSource *TableLinkDataSource
 }
diff --git a/pkg/oomstore/types/types.go b/pkg/oomstore/types/types.go
index 1b391e71..1348c05f 100644
--- a/pkg/oomstore/types/types.go
+++ b/pkg/oomstore/types/types.go
@@ -66,14 +66,8 @@ func (e *ExportResult) GetErrorChannel() <-chan error {
 	return e.error
 }
 
-type DataTableTimeRange struct {
-	MinUnixMilli int64 `db:"min_unix_milli"`
-	MaxUnixMilli int64 `db:"max_unix_milli"`
-}
-
 type DataTableSchema struct {
-	Fields    []DataTableFieldSchema
-	TimeRange DataTableTimeRange
+	Fields []DataTableFieldSchema
 }
 
 type DataTableFieldSchema struct {
diff --git a/sdk/python_v2/Cargo.lock b/sdk/python_v2/Cargo.lock
index 1b300aca..7daec46f 100644
--- a/sdk/python_v2/Cargo.lock
+++ b/sdk/python_v2/Cargo.lock
@@ -516,9 +516,9 @@ checksum = "da32515d9f6e6e489d7bc9d84c71b060db7247dc035bbe44eac88cf87486d8d5"
 
 [[package]]
 name = "oomclient"
-version = "0.1.0-rc.4"
+version = "0.1.0-rc.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e31e967d74a159fed20efe160e826fa9459f6a9b4937975bc5aff01978b657f"
+checksum = "419dde096185087b0ef00cde23a28ad6825193038c9fa5aafec38967998bf8bf"
 dependencies = [
  "async-stream",
  "futures-core",
@@ -538,7 +538,6 @@ dependencies = [
  "oomclient",
  "pyo3",
  "pyo3-asyncio",
- "thiserror",
  "tokio",
 ]
 
diff --git a/sdk/python_v2/Cargo.toml b/sdk/python_v2/Cargo.toml
index e942ab51..dd50b235 100644
--- a/sdk/python_v2/Cargo.toml
+++ b/sdk/python_v2/Cargo.toml
@@ -9,8 +9,7 @@ name = "oomclient"
 crate-type = ["cdylib"]
 
 [dependencies]
-oomclient = "0.1.0-rc.4"
+oomclient = "0.1.0-rc.1"
 pyo3-asyncio = { version = "0.15", features = ["tokio-runtime"] }
 pyo3 = { version = "0.15", features = ["extension-module"] }
 tokio = { version = "1.15", features = ["macros", "rt-multi-thread"] }
-thiserror = "1.0"
diff --git a/sdk/python_v2/Makefile.toml b/sdk/python_v2/Makefile.toml
index 11bc20b2..941cac62 100644
--- a/sdk/python_v2/Makefile.toml
+++ b/sdk/python_v2/Makefile.toml
@@ -1,4 +1,6 @@
-[tasks.maturin]
+[tasks.pyshell]
+
+[tasks.post-build]
 script = '''
 source .env/bin/activate
 maturin develop
diff --git a/sdk/python_v2/examples/export.py b/sdk/python_v2/examples/export.py
deleted file mode 100644
index a2119147..00000000
--- a/sdk/python_v2/examples/export.py
+++ /dev/null
@@ -1,20 +0,0 @@
-#!/usr/bin/env python3
-import asyncio
-import time
-from oomclient import Client
-
-
-async def main():
-    client = await Client.connect("http://localhost:50051")
-    features = [
-        "account.state",
-        "account.has_2fa_installed",
-        "transaction_stats.transaction_count_7d",
-    ]
-    output_path = "/tmp/output.csv"
-    now = round(time.time() * 1000)
-    await client.export(features, now, output_path, 20)
-    print(open(output_path).read())
-
-
-asyncio.run(main())
diff --git a/sdk/python_v2/examples/import.py b/sdk/python_v2/examples/import.py
deleted file mode 100644
index 4019cbe3..00000000
--- a/sdk/python_v2/examples/import.py
+++ /dev/null
@@ -1,26 +0,0 @@
-#!/usr/bin/env python3
-import asyncio
-from oomclient import Client
-
-async def main():
-    client = await Client.connect("http://localhost:50051")
-    group = "account"
-    path = "/tmp/demo.csv"
-    with open(path, "w") as f:
-        rows = """
-user,state,credit_score,account_age_days,has_2fa_installed
-1,Nevada,530,242,true
-2,South Carolina,520,268,false
-3,New Jersey,655,84,false
-4,Ohio,677,119,true
-5,California,566,289,false
-6,North Carolina,533,155,true
-7,North Dakota,605,334,true
-8,West Virginia,664,282,false
-9,Alabama,577,150,true
-10,Idaho,693,212,true
-""".lstrip()
-        f.write(rows)
-    await client.import_(group, None, None, path, None)
-
-asyncio.run(main())
diff --git a/sdk/python_v2/examples/join.py b/sdk/python_v2/examples/join.py
deleted file mode 100644
index 0b3d4070..00000000
--- a/sdk/python_v2/examples/join.py
+++ /dev/null
@@ -1,43 +0,0 @@
-#!/usr/bin/env python3
-import asyncio
-from oomclient import Client
-
-
-async def main():
-    client = await Client.connect("http://localhost:50051")
-    features = [
-        "driver_stats.conv_rate",
-        "driver_stats.acc_rate",
-        "driver_stats.avg_daily_trips",
-    ]
-    input_path = "/tmp/input.csv"
-    output_path = "/tmp/output.csv"
-    with open(input_path, "w") as f:
-        rows = """
-entity_key,unix_milli,age
-1,0,20
-1,3,44
-1,4,37
-2,3,28
-3,3,48
-4,4,22
-5,2,41
-5,2,43
-5,3,47
-5,4,59
-5,5,55
-6,1,27
-6,3,46
-7,0,42
-7,4,40
-7,4,49
-8,4,39
-9,3,35
-10,4,54
-10,4,33
-""".lstrip()
-        f.write(rows)
-    await client.join(features, input_path, output_path)
-    print(open(output_path).read())
-
-asyncio.run(main())
diff --git a/sdk/python_v2/examples/push.py b/sdk/python_v2/examples/push.py
deleted file mode 100644
index 95df04a5..00000000
--- a/sdk/python_v2/examples/push.py
+++ /dev/null
@@ -1,15 +0,0 @@
-#!/usr/bin/env python3
-import asyncio
-from oomclient import Client
-
-
-async def main():
-    client = await Client.connect("http://localhost:50051")
-    kv_pairs = {
-        "last_5_click_posts": "1,3,5,7,9",
-        "number_of_user_starred_posts": 30,
-    }
-    await client.push("user-click", "929", kv_pairs)
-
-
-asyncio.run(main())
diff --git a/sdk/python_v2/examples/snapshot.py b/sdk/python_v2/examples/snapshot.py
deleted file mode 100644
index e48255dc..00000000
--- a/sdk/python_v2/examples/snapshot.py
+++ /dev/null
@@ -1,9 +0,0 @@
-#!/usr/bin/env python3
-import asyncio
-from oomclient import Client
-
-async def main():
-    client = await Client.connect("http://localhost:50051")
-    await client.snapshot("user-click")
-
-asyncio.run(main())
diff --git a/sdk/python_v2/src/convert.rs b/sdk/python_v2/src/convert.rs
index d1b549b1..e75b6a27 100644
--- a/sdk/python_v2/src/convert.rs
+++ b/sdk/python_v2/src/convert.rs
@@ -1,9 +1,10 @@
+use oomclient::error::OomError;
 use std::collections::HashMap;
 
 use oomclient::Value;
 use pyo3::{exceptions::PyException, prelude::*};
 
-pub fn value_to_py(value: Option<&Value>, py: Python) -> PyObject {
+pub fn value_to_py(value: Option<Value>, py: Python) -> PyObject {
     value
         .map(|value| match value {
             Value::Int64(v) => v.to_object(py),
@@ -18,32 +19,11 @@ pub fn value_to_py(value: Option<&Value>, py: Python) -> PyObject {
 
 pub fn value_map_to_py(m: HashMap<String, Option<Value>>, py: Python) -> PyObject {
     m.into_iter()
-        .map(|(k, v)| (k, value_to_py(v.as_ref(), py)))
+        .map(|(k, v)| (k, value_to_py(v, py)))
         .collect::<HashMap<_, _>>()
         .into_py(py)
 }
 
-pub fn err_to_py(err: impl std::error::Error) -> PyErr {
+pub fn oom_err_to_py(err: OomError) -> PyErr {
     PyException::new_err(format!("{:?}", err))
 }
-
-pub fn py_to_value(obj: &PyAny) -> PyResult<Value> {
-    obj.extract::<ValueWrapper>().map(|wrapper| match wrapper {
-        ValueWrapper::Int64(v) => Value::Int64(v),
-        ValueWrapper::Double(v) => Value::Double(v),
-        ValueWrapper::String(v) => Value::String(v),
-        ValueWrapper::Bool(v) => Value::Bool(v),
-        ValueWrapper::UnixMilli(v) => Value::UnixMilli(v),
-        ValueWrapper::Bytes(v) => Value::Bytes(v),
-    })
-}
-
-#[derive(FromPyObject, Debug)]
-enum ValueWrapper {
-    Int64(i64),
-    Double(f64),
-    String(String),
-    Bool(bool),
-    UnixMilli(i64),
-    Bytes(Vec<u8>),
-}
diff --git a/sdk/python_v2/src/error.rs b/sdk/python_v2/src/error.rs
deleted file mode 100644
index e3820880..00000000
--- a/sdk/python_v2/src/error.rs
+++ /dev/null
@@ -1,12 +0,0 @@
-use thiserror::Error;
-
-#[derive(Error, Debug)]
-pub enum Error {
-    #[error("{0}")]
-    Custom(String),
-}
-
-#[macro_export]
-macro_rules! err {
-    ($($tt:tt)*) => { Err(Error::Custom(format!($($tt)*).into())) };
-}
diff --git a/sdk/python_v2/src/lib.rs b/sdk/python_v2/src/lib.rs
index e6788cd7..d07cf9f0 100644
--- a/sdk/python_v2/src/lib.rs
+++ b/sdk/python_v2/src/lib.rs
@@ -1,13 +1,8 @@
 mod convert;
-mod error;
 
-use convert::{err_to_py, py_to_value, value_map_to_py};
-use error::Error;
+use convert::{oom_err_to_py, value_map_to_py};
 use oomclient::Client as OomClient;
-use pyo3::{
-    prelude::*,
-    types::{PyDict, PyType},
-};
+use pyo3::{prelude::*, types::PyType};
 use pyo3_asyncio::tokio::future_into_py;
 use std::collections::HashMap;
 
@@ -21,7 +16,7 @@ impl Client {
     #[classmethod]
     pub fn connect<'p>(_cls: &PyType, py: Python<'p>, endpoint: String) -> PyResult<&'p PyAny> {
         future_into_py(py, async {
-            let inner = OomClient::connect(endpoint).await.map_err(err_to_py)?;
+            let inner = OomClient::connect(endpoint).await.map_err(oom_err_to_py)?;
             let client = Client { inner };
             Python::with_gil(|py| PyCell::new(py, client).map(|py_cell| py_cell.to_object(py)))
         })
@@ -31,7 +26,7 @@ impl Client {
         // Don't panic, it's cheap:
         // https://github.com/hyperium/tonic/issues/285#issuecomment-595880400
         let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move { inner.health_check().await.map_err(err_to_py) })
+        future_into_py(py, async move { inner.health_check().await.map_err(oom_err_to_py) })
     }
 
     pub fn online_get<'p>(&self, py: Python<'p>, entity_key: String, features: Vec<String>) -> PyResult<&'p PyAny> {
@@ -40,7 +35,7 @@ impl Client {
             inner
                 .online_get(entity_key, features)
                 .await
-                .map_err(err_to_py)
+                .map_err(oom_err_to_py)
                 .map(|m| Python::with_gil(|py| value_map_to_py(m, py)))
         })
     }
@@ -56,7 +51,7 @@ impl Client {
             inner
                 .online_multi_get(entity_keys, features)
                 .await
-                .map_err(err_to_py)
+                .map_err(oom_err_to_py)
                 .map(|r| {
                     Python::with_gil(|py| {
                         r.into_iter()
@@ -70,93 +65,9 @@ impl Client {
     pub fn sync<'p>(&mut self, py: Python<'p>, revision_id: u32, purge_delay: u32) -> PyResult<&'p PyAny> {
         let mut inner = OomClient::clone(&self.inner);
         future_into_py(py, async move {
-            inner.sync(revision_id, purge_delay).await.map_err(err_to_py)
+            inner.sync(revision_id, purge_delay).await.map_err(oom_err_to_py)
         })
     }
-
-    #[pyo3(name = "import_")]
-    pub fn import<'p>(
-        &mut self,
-        py: Python<'p>,
-        group: String,
-        revision: Option<i64>,
-        description: Option<String>,
-        input_file: String,
-        delimiter: Option<String>,
-    ) -> PyResult<&'p PyAny> {
-        let delimiter = delimiter
-            .map(|s| {
-                let mut chars = s.chars();
-                match (chars.next(), chars.next()) {
-                    (Some(c), None) => Ok(c),
-                    _ => err!("delimiter must be exactly one character"),
-                }
-            })
-            .transpose()
-            .map_err(err_to_py)?;
-        let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move {
-            inner
-                .import(group, revision, description, input_file, delimiter)
-                .await
-                .map_err(err_to_py)
-        })
-    }
-
-    pub fn push<'p>(
-        &mut self,
-        py: Python<'p>,
-        group: String,
-        entity_key: String,
-        kv_pairs: &PyDict,
-    ) -> PyResult<&'p PyAny> {
-        let mut kvs = Vec::with_capacity(kv_pairs.len());
-        for (k, v) in kv_pairs {
-            let name = k.extract::<String>()?;
-            let value = py_to_value(v)?;
-            kvs.push((name, value));
-        }
-
-        let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move {
-            inner.push(entity_key, group, kvs).await.map_err(err_to_py)
-        })
-    }
-
-    pub fn join<'p>(
-        &mut self,
-        py: Python<'p>,
-        features: Vec<String>,
-        input_file: String,
-        output_file: String,
-    ) -> PyResult<&'p PyAny> {
-        let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move {
-            inner.join(features, input_file, output_file).await.map_err(err_to_py)
-        })
-    }
-
-    pub fn export<'p>(
-        &mut self,
-        py: Python<'p>,
-        features: Vec<String>,
-        unix_milli: u64,
-        output_file: String,
-        limit: Option<usize>,
-    ) -> PyResult<&'p PyAny> {
-        let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move {
-            inner
-                .export(features, unix_milli, output_file, limit)
-                .await
-                .map_err(err_to_py)
-        })
-    }
-
-    pub fn snapshot<'p>(&mut self, py: Python<'p>, group: String) -> PyResult<&'p PyAny> {
-        let mut inner = OomClient::clone(&self.inner);
-        future_into_py(py, async move { inner.snapshot(group).await.map_err(err_to_py) })
-    }
 }
 
 /// OomClient python module implemented in Rust.
diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml
index 7d8c5e81..48ed35a5 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "oomclient"
-version = "0.1.0-rc.4"
+version = "0.1.0-rc.3"
 edition = "2021"
 authors = ["Wenxuan Zhang <wenxuangm@gmail.com>"]
 description = "oomstore client for rust"
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index 375bd3d9..fc15711a 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -130,8 +130,8 @@ impl Client {
     pub async fn import(
         &mut self,
         group: impl Into<String>,
-        revision: impl Into<Option<i64>>,
         description: impl Into<Option<String>>,
+        revision: impl Into<Option<i64>>,
         input_file: impl AsRef<Path>,
         delimiter: impl Into<Option<char>>,
     ) -> Result<u32> {


Test patch:
diff --git a/internal/database/offline/bigquery/store_test.go b/internal/database/offline/bigquery/store_test.go
index cecd3a2c..9e62a068 100644
--- a/internal/database/offline/bigquery/store_test.go
+++ b/internal/database/offline/bigquery/store_test.go
@@ -96,13 +96,10 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		query := fmt.Sprintf("CREATE TABLE %s.offline_batch_1_1(`user` STRING, `age` BIGINT, `unix_milli` BIGINT)", DATASET_ID)
+		query := fmt.Sprintf("CREATE TABLE %s.offline_batch_1_1(`user` STRING, `age` BIGINT)", DATASET_ID)
 		if _, err = db.Query(query).Read(ctx); err != nil {
 			t.Fatal(err)
 		}
-		if _, err = db.Query(fmt.Sprintf("insert into %s.offline_batch_1_1 VALUES ('1', 1, 1), ('2', 2, 100)", DATASET_ID)).Read(ctx); err != nil {
-			t.Fatal(err)
-		}
 	})
 }
 
diff --git a/internal/database/offline/mysql/store_test.go b/internal/database/offline/mysql/store_test.go
index 5cdfd90e..3d738f69 100644
--- a/internal/database/offline/mysql/store_test.go
+++ b/internal/database/offline/mysql/store_test.go
@@ -61,10 +61,7 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer db.Close()
-		if _, err = db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint, `unix_milli` int)"); err != nil {
-			t.Fatal(err)
-		}
-		if _, err = db.ExecContext(ctx, "insert into `offline_batch_1_1` VALUES ('1', 1, 1), ('2', 2, 100)"); err != nil {
+		if _, err := db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint)"); err != nil {
 			t.Fatal(err)
 		}
 	})
diff --git a/internal/database/offline/postgres/store_test.go b/internal/database/offline/postgres/store_test.go
index aaeae5fe..69486466 100644
--- a/internal/database/offline/postgres/store_test.go
+++ b/internal/database/offline/postgres/store_test.go
@@ -53,10 +53,7 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer db.Close()
-		if _, err = db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint, "unix_milli" int)`); err != nil {
-			t.Fatal(err)
-		}
-		if _, err = db.ExecContext(ctx, `insert into "offline_batch_1_1" VALUES ('1', 1, 1), ('2', 2, 100)`); err != nil {
+		if _, err := db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint)`); err != nil {
 			t.Fatal(err)
 		}
 	})
diff --git a/internal/database/offline/redshift/store_test.go b/internal/database/offline/redshift/store_test.go
index d23f4127..e9df60ad 100644
--- a/internal/database/offline/redshift/store_test.go
+++ b/internal/database/offline/redshift/store_test.go
@@ -97,10 +97,7 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer db.Close()
-		if _, err = db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint, "unix_milli" int)`); err != nil {
-			t.Fatal(err)
-		}
-		if _, err = db.ExecContext(ctx, `insert into "offline_batch_1_1" VALUES ('1', 1, 1), ('2', 2, 100)`); err != nil {
+		if _, err = db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint)`); err != nil {
 			t.Fatal(err)
 		}
 	})
diff --git a/internal/database/offline/snowflake/store_test.go b/internal/database/offline/snowflake/store_test.go
index 1f0485d3..8b56293e 100644
--- a/internal/database/offline/snowflake/store_test.go
+++ b/internal/database/offline/snowflake/store_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/offline"
 	"github.com/oom-ai/oomstore/internal/database/offline/snowflake"
+	"github.com/oom-ai/oomstore/internal/database/offline/test_impl"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -53,25 +54,25 @@ func destroyStore(database string) func() {
 	}
 }
 
-//func TestPing(t *testing.T) {
-//	test_impl.TestPing(t, prepareStore, destroyStore(DATABASE))
-//}
-//
-//func TestExport(t *testing.T) {
-//	test_impl.TestExport(t, prepareStore, destroyStore(DATABASE))
-//}
-//
-//func TestImport(t *testing.T) {
-//	test_impl.TestImport(t, prepareStore, destroyStore(DATABASE))
-//}
-//
-//func TestJoin(t *testing.T) {
-//	test_impl.TestJoin(t, prepareStore, destroyStore(DATABASE))
-//}
-//
-//func TestSnapshot(t *testing.T) {
-//	test_impl.TestSnapshot(t, prepareStore, destroyStore(DATABASE))
-//}
+func TestPing(t *testing.T) {
+	test_impl.TestPing(t, prepareStore, destroyStore(DATABASE))
+}
+
+func TestExport(t *testing.T) {
+	test_impl.TestExport(t, prepareStore, destroyStore(DATABASE))
+}
+
+func TestImport(t *testing.T) {
+	test_impl.TestImport(t, prepareStore, destroyStore(DATABASE))
+}
+
+func TestJoin(t *testing.T) {
+	test_impl.TestJoin(t, prepareStore, destroyStore(DATABASE))
+}
+
+func TestSnapshot(t *testing.T) {
+	test_impl.TestSnapshot(t, prepareStore, destroyStore(DATABASE))
+}
 
 // We don't use test_impl.TestTableSchema because snowflake cannot be
 // accessed by two different sessions.
@@ -82,19 +83,13 @@ func TestTableSchema(t *testing.T) {
 	defer store.Close()
 	db := store.(*snowflake.DB)
 
-	if _, err := db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint, "unix_milli" int)`); err != nil {
-		t.Fatal(err)
-	}
-	if _, err := db.ExecContext(ctx, `insert into "offline_batch_1_1" VALUES ('1', 1, 1), ('2', 2, 100)`); err != nil {
+	if _, err := db.ExecContext(ctx, `create table "offline_batch_1_1"("user" varchar(16), "age" smallint)`); err != nil {
 		t.Fatal(err)
 	}
 
-	actual, err := store.TableSchema(ctx, offline.TableSchemaOpt{
-		TableName:      "offline_batch_1_1",
-		CheckTimeRange: true,
-	})
+	actual, err := store.TableSchema(ctx, "offline_batch_1_1")
 	require.NoError(t, err)
-	require.Equal(t, 3, len(actual.Fields))
+	require.Equal(t, 2, len(actual.Fields))
 
 	expected := types.DataTableSchema{
 		Fields: []types.DataTableFieldSchema{
@@ -106,21 +101,11 @@ func TestTableSchema(t *testing.T) {
 				Name:      "age",
 				ValueType: types.Int64,
 			},
-			{
-				Name:      "unix_milli",
-				ValueType: types.Int64,
-			},
-		},
-		TimeRange: types.DataTableTimeRange{
-			MinUnixMilli: 1,
-			MaxUnixMilli: 100,
 		},
 	}
 	require.ElementsMatch(t, expected.Fields, actual.Fields)
-	require.Equal(t, expected.TimeRange, actual.TimeRange)
-
 }
 
-//func TestCreateTable(t *testing.T) {
-//	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
-//}
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
+}
diff --git a/internal/database/offline/sqlite/store_test.go b/internal/database/offline/sqlite/store_test.go
index 94af4156..fcade997 100644
--- a/internal/database/offline/sqlite/store_test.go
+++ b/internal/database/offline/sqlite/store_test.go
@@ -63,10 +63,7 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer db.Close()
-		if _, err = db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint, `unix_milli` int)"); err != nil {
-			t.Fatal(err)
-		}
-		if _, err = db.ExecContext(ctx, "insert into `offline_batch_1_1` VALUES ('1', 1, 1), ('2', 2, 100)"); err != nil {
+		if _, err := db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint)"); err != nil {
 			t.Fatal(err)
 		}
 	})
diff --git a/internal/database/offline/test_impl/table_schema.go b/internal/database/offline/test_impl/table_schema.go
index e76c38f4..1fba90a9 100644
--- a/internal/database/offline/test_impl/table_schema.go
+++ b/internal/database/offline/test_impl/table_schema.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/oom-ai/oomstore/internal/database/offline"
-
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 	"github.com/stretchr/testify/require"
 )
@@ -18,12 +16,9 @@ func TestTableSchema(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	prepareSampleTable(ctx)
 
-	actual, err := store.TableSchema(ctx, offline.TableSchemaOpt{
-		TableName:      "offline_batch_1_1",
-		CheckTimeRange: true,
-	})
+	actual, err := store.TableSchema(ctx, "offline_batch_1_1")
 	require.NoError(t, err)
-	require.Equal(t, 3, len(actual.Fields))
+	require.Equal(t, 2, len(actual.Fields))
 
 	expected := types.DataTableSchema{
 		Fields: []types.DataTableFieldSchema{
@@ -35,14 +30,6 @@ func TestTableSchema(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 				Name:      "age",
 				ValueType: types.Int64,
 			},
-			{
-				Name:      "unix_milli",
-				ValueType: types.Int64,
-			},
-		},
-		TimeRange: types.DataTableTimeRange{
-			MinUnixMilli: 10,
-			MaxUnixMilli: 100,
 		},
 	}
 	require.ElementsMatch(t, expected.Fields, actual.Fields)
diff --git a/internal/database/offline/tidb/store_test.go b/internal/database/offline/tidb/store_test.go
index a6091d3e..241fb4f7 100644
--- a/internal/database/offline/tidb/store_test.go
+++ b/internal/database/offline/tidb/store_test.go
@@ -52,10 +52,7 @@ func TestTableSchema(t *testing.T) {
 			t.Fatal(err)
 		}
 		defer db.Close()
-		if _, err = db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint, `unix_milli` int)"); err != nil {
-			t.Fatal(err)
-		}
-		if _, err = db.ExecContext(ctx, "insert into `offline_batch_1_1` VALUES ('1', 1, 1), ('2', 2, 100)"); err != nil {
+		if _, err = db.ExecContext(ctx, "create table `offline_batch_1_1`(`user` varchar(16), `age` smallint)"); err != nil {
 			t.Fatal(err)
 		}
 	})
diff --git a/oomcli/test/test_export.sh b/oomcli/test/test_export.sh
index f03beed8..c5633544 100755
--- a/oomcli/test/test_export.sh
+++ b/oomcli/test/test_export.sh
@@ -1,23 +1,77 @@
 #!/usr/bin/env bash
+
 SDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) && cd "$SDIR" || exit 1
 source ./util.sh
 
 init_store
 register_features
-import_device_sample >> /dev/null
-
-case='get all'
-expected='device,phone.price
-1,3999
-2,5299
-3,3999
-4,1999
-5,999
-6,4999
-7,5999
-8,6500
-9,4500
+
+t0=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
+t1=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
+
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11
+t2=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
+
+import_student_sample
+t3=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
+
+oomcli push --entity-key 2 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
+t4=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
+
+oomcli_export_push_feature() {
+    case="push feature"
+    expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
+1,"1,2,3,4,5",10
+'
+    actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t1 -o csv)
+    assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
+}
+
+oomcli_export_update_feature() {
+    case="update feature"
+    expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
+1,"2,3,4,5,6",11
+'
+    actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t2 -o csv)
+    assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
+}
+
+oomcli_export_batch() {
+    case='export batch feature'
+    expected='user,student.name,student.gender,student.age
+1,lian,m,18
+2,gao,m,20
+3,zhang,f,17
+4,dong,m,25
+5,tang,f,18
+6,chen,m,25
+7,he,f,19
 '
-timestamp=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-actual=$(oomcli export --feature phone.price --unix-milli $timestamp -o csv)
-assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
+    actual=$(oomcli export --feature student.name,student.gender,student.age --unix-milli $t3 -o csv)
+    assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
+}
+
+oomcli_export_batch_and_stream() {
+    case="export batch and stream feature"
+    expected='user,student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
+1,lian,m,18,"2,3,4,5,6",11
+2,gao,m,20,,
+3,zhang,f,17,,
+4,dong,m,25,,
+5,tang,f,18,,
+6,chen,m,25,,
+7,he,f,19,,
+'
+    actual=$(oomcli export --feature student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t3 -o csv)
+    assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
+}
+
+main() {
+    oomcli_export_push_feature
+    oomcli_export_update_feature
+    oomcli_export_batch
+    oomcli_export_batch_and_stream
+}
+
+main
diff --git a/oomcli/test/test_stream_export.sh b/oomcli/test/test_stream_export.sh
deleted file mode 100755
index f4c0c096..00000000
--- a/oomcli/test/test_stream_export.sh
+++ /dev/null
@@ -1,46 +0,0 @@
-#!/usr/bin/env bash
-
-SDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) && cd "$SDIR" || exit 1
-source ./util.sh
-
-init_store
-register_features
-
-
-t0=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
-t1=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-
-oomcli push --entity-key 1 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11
-t2=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-
-oomcli push --entity-key 2 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
-t3=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-
-case="empty result"
-expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-'
-actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t0 -o csv)
-assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
-
-case="push feature"
-expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,"1,2,3,4,5",10
-'
-actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t1 -o csv)
-assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
-
-case="update feature"
-expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,"2,3,4,5,6",11
-'
-actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t2 -o csv)
-assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
-
-case="push new entity key feature "
-expected='user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,"2,3,4,5,6",11
-2,"1,2,3,4,5",10
-'
-actual=$(oomcli export --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t3 -o csv)
-assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"