Repository: matrixorigin/matrixone
Pull request #3906 (2022-07-12 05:29:46)

Add objectio for distributed tae
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available


CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/matrixorigin/matrixone?pullRequest=3906) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our [Contributor License Agreement](https://cla-assistant.io/matrixorigin/matrixone?pullRequest=3906) before we can accept your contribution.<br/>**1** out of **2** committers have signed the CLA.<br/><br/>:white_check_mark: LeftHandCold<br/>:x: 申江伟<br/><hr/>**申江伟** seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please [add the email address used for this commit to your account](https://help.github.com/articles/why-are-my-commits-linked-to-the-wrong-user/#commits-are-not-linked-to-any-user).<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/matrixorigin/matrixone?pullRequest=3906) it.</sub>

Patch:
diff --git a/pkg/vm/engine/tae/dataio/objectio/allocator.go b/pkg/vm/engine/tae/dataio/objectio/allocator.go
new file mode 100644
index 000000000..cd40790d1
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/allocator.go
@@ -0,0 +1,50 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import "sync"
+
+type ObjectAllocator struct {
+	mutex     sync.RWMutex
+	available uint64
+	pageSize  uint32
+}
+
+func NewObjectAllocator(capacity uint64, pageSize uint32) *ObjectAllocator {
+	allocator := &ObjectAllocator{
+		pageSize: pageSize,
+	}
+	allocator.Init(capacity, pageSize)
+	return allocator
+}
+
+func (o *ObjectAllocator) Init(capacity uint64, pageSize uint32) {
+	o.pageSize = pageSize
+	o.available = 0
+}
+
+func (o *ObjectAllocator) Allocate(needLen uint64) (uint64, uint64) {
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	offset := o.available
+	o.available += needLen
+	return offset, needLen
+}
+
+func (o *ObjectAllocator) GetAvailable() uint64 {
+	o.mutex.RLock()
+	defer o.mutex.RUnlock()
+	return o.available
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/block.go b/pkg/vm/engine/tae/dataio/objectio/block.go
new file mode 100644
index 000000000..3b8feddb2
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/block.go
@@ -0,0 +1,276 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"bytes"
+
+	"github.com/RoaringBitmap/roaring"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/file"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/stl/adaptors"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/types"
+)
+
+type blockFile struct {
+	common.RefHelper
+	seg     *segmentFile
+	rows    uint32
+	id      *common.ID
+	ts      uint64
+	columns []*columnBlock
+	writer  *Writer
+	reader  *Reader
+}
+
+func newBlock(id uint64, seg *segmentFile, colCnt int, indexCnt map[int]int) *blockFile {
+	blockID := &common.ID{
+		SegmentID: seg.id.SegmentID,
+		BlockID:   id,
+	}
+	bf := &blockFile{
+		seg:     seg,
+		id:      blockID,
+		columns: make([]*columnBlock, colCnt),
+	}
+	bf.reader = NewReader(seg.fs, bf)
+	bf.writer = NewWriter(seg.fs)
+	bf.OnZeroCB = bf.close
+	for i := range bf.columns {
+		cnt := 0
+		if indexCnt != nil {
+			cnt = indexCnt[i]
+		}
+		bf.columns[i] = newColumnBlock(bf, cnt, i)
+	}
+	bf.Ref()
+	return bf
+}
+
+func (bf *blockFile) Fingerprint() *common.ID {
+	return bf.id
+}
+
+func (bf *blockFile) close() {
+	bf.Close()
+	err := bf.Destroy()
+	if err != nil {
+		panic(any("Destroy error"))
+	}
+}
+
+func (bf *blockFile) WriteRows(rows uint32) (err error) {
+	bf.rows = rows
+	return nil
+}
+
+func (bf *blockFile) ReadRows() uint32 {
+	return bf.rows
+}
+
+func (bf *blockFile) WriteTS(ts uint64) (err error) {
+	bf.ts = ts
+	return
+}
+
+func (bf *blockFile) ReadTS() (ts uint64, err error) {
+	ts = bf.ts
+	return
+}
+
+func (bf *blockFile) WriteDeletes(buf []byte) (err error) {
+	return bf.writer.WriteDeletes(bf.ts, bf.id, buf)
+}
+
+func (bf *blockFile) ReadDeletes(buf []byte) (err error) {
+	return
+}
+
+func (bf *blockFile) GetDeletesFileStat() (stat common.FileInfo) {
+	return nil
+}
+
+func (bf *blockFile) WriteIndexMeta(buf []byte) (err error) {
+	return bf.writer.WriteIndexMeta(bf.id, buf)
+}
+
+func (bf *blockFile) LoadIndexMeta() (any, error) {
+	return bf.reader.LoadIndexMeta(bf.id)
+}
+
+func (bf *blockFile) OpenColumn(colIdx int) (colBlk file.ColumnBlock, err error) {
+	if colIdx >= len(bf.columns) {
+		err = file.ErrInvalidParam
+		return
+	}
+	bf.columns[colIdx].Ref()
+	colBlk = bf.columns[colIdx]
+	return
+}
+
+func (bf *blockFile) Close() error {
+	return nil
+}
+
+func (bf *blockFile) Destroy() error {
+	return nil
+}
+
+func (bf *blockFile) Sync() error { return nil }
+
+func (bf *blockFile) LoadBatch(
+	colTypes []types.Type,
+	colNames []string,
+	nullables []bool,
+	opts *containers.Options) (bat *containers.Batch, err error) {
+	return bf.reader.LoadABlkColumns(colTypes, colNames, nullables, opts)
+}
+
+func (bf *blockFile) WriteColumnVec(ts uint64, colIdx int, vec containers.Vector) (err error) {
+	cb, err := bf.OpenColumn(colIdx)
+	if err != nil {
+		return err
+	}
+	defer cb.Close()
+	err = cb.WriteTS(ts)
+	w := adaptors.NewBuffer(nil)
+	defer w.Close()
+	if _, err = vec.WriteTo(w); err != nil {
+		return
+	}
+	err = cb.WriteData(w.Bytes())
+	return
+}
+
+func (bf *blockFile) WriteBatch(bat *containers.Batch, ts uint64) (err error) {
+	if err = bf.WriteTS(ts); err != nil {
+		return
+	}
+	if err = bf.WriteRows(uint32(bat.Length())); err != nil {
+		return
+	}
+	for colIdx := range bat.Attrs {
+		if err = bf.WriteColumnVec(ts, colIdx, bat.Vecs[colIdx]); err != nil {
+			return
+		}
+	}
+	return
+}
+
+func (bf *blockFile) PrepareUpdates(
+	tool *containers.CodecTool,
+	buffer *bytes.Buffer,
+	typ types.Type,
+	nullable bool,
+	mask *roaring.Bitmap,
+	vals map[uint32]any) (err error) {
+	_, _ = tool.Write(types.EncodeType(typ))
+	_, _ = tool.Write(types.EncodeFixed(nullable))
+	buf, err := mask.ToBytes()
+	if err != nil {
+		return
+	}
+	_, _ = tool.Write(buf)
+	tmp := containers.MakeVector(typ, nullable)
+	defer tmp.Close()
+	it := mask.Iterator()
+	for it.HasNext() {
+		row := it.Next()
+		v := vals[row]
+		tmp.Append(v)
+	}
+	if _, err = tmp.WriteTo(buffer); err != nil {
+		return
+	}
+	_, _ = tool.Write(buffer.Bytes())
+	return
+}
+
+func (bf *blockFile) WriteSnapshot(
+	bat *containers.Batch,
+	ts uint64,
+	masks map[uint16]*roaring.Bitmap,
+	vals map[uint16]map[uint32]any,
+	deletes *roaring.Bitmap) (err error) {
+	if err = bf.WriteTS(ts); err != nil {
+		return err
+	}
+	if err = bf.WriteRows(uint32(bat.Length())); err != nil {
+		return err
+	}
+
+	buffer := new(bytes.Buffer)
+	if deletes != nil {
+		if _, err = deletes.WriteTo(buffer); err != nil {
+			return
+		}
+		if err = bf.WriteDeletes(buffer.Bytes()); err != nil {
+			return
+		}
+	}
+	tool := containers.NewCodecTool()
+	defer tool.Close()
+	for colIdx := range bat.Attrs {
+		var cb file.ColumnBlock
+		cb, err = bf.OpenColumn(colIdx)
+		if err != nil {
+			return err
+		}
+		defer cb.Close()
+		err = cb.WriteTS(ts)
+		if err != nil {
+			return err
+		}
+		vec := bat.Vecs[colIdx]
+		updates := vals[uint16(colIdx)]
+		if updates != nil {
+			var uf common.IRWFile
+			uf, err = cb.OpenUpdateFile()
+			if err != nil {
+				panic(err)
+			}
+			defer uf.Unref()
+			mask := masks[uint16(colIdx)]
+			buffer.Reset()
+			if err = bf.PrepareUpdates(tool, buffer, vec.GetType(), vec.Nullable(), mask, updates); err != nil {
+				return
+			}
+			buffer.Reset()
+			if _, err = tool.WriteTo(buffer); err != nil {
+				return
+			}
+			if _, err = uf.Write(buffer.Bytes()); err != nil {
+				return
+			}
+		}
+		buffer.Reset()
+		if _, err = vec.WriteTo(buffer); err != nil {
+			return err
+		}
+		if err = cb.WriteData(buffer.Bytes()); err != nil {
+			return err
+		}
+	}
+	return
+}
+
+func (bf *blockFile) LoadDeletes() (mask *roaring.Bitmap, err error) {
+	return bf.reader.LoadDeletes(bf.id)
+}
+
+func (bf *blockFile) LoadUpdates() (masks map[uint16]*roaring.Bitmap, vals map[uint16]map[uint32]any) {
+	return bf.reader.LoadUpdates()
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/colblk.go b/pkg/vm/engine/tae/dataio/objectio/colblk.go
new file mode 100644
index 000000000..b14ab627a
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/colblk.go
@@ -0,0 +1,127 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"os"
+)
+
+type columnBlock struct {
+	common.RefHelper
+	block   *blockFile
+	ts      uint64
+	indexes int
+	id      *common.ID
+	vfs     *VFS
+}
+
+func newColumnBlock(block *blockFile, indexCnt int, col int) *columnBlock {
+	cId := &common.ID{
+		SegmentID: block.id.SegmentID,
+		BlockID:   block.id.BlockID,
+		Idx:       uint16(col),
+	}
+	cb := &columnBlock{
+		block:   block,
+		indexes: indexCnt,
+		id:      cId,
+	}
+	cb.vfs = &VFS{cb: cb}
+	cb.OnZeroCB = cb.close
+	cb.Ref()
+	return cb
+}
+
+func (cb *columnBlock) WriteTS(ts uint64) (err error) {
+	cb.ts = ts
+	return
+}
+
+func (cb *columnBlock) WriteData(buf []byte) (err error) {
+	return cb.block.writer.WriteData(cb.ts, cb.id, buf)
+}
+
+func (cb *columnBlock) WriteUpdates(buf []byte) (err error) {
+	return cb.block.writer.WriteUpdates(cb.ts, cb.id, buf)
+}
+
+func (cb *columnBlock) WriteIndex(idx int, buf []byte) (err error) {
+	return cb.block.writer.WriteIndex(cb.id, idx, buf)
+}
+
+func (cb *columnBlock) ReadTS() uint64 {
+	return cb.ts
+}
+
+func (cb *columnBlock) ReadData(buf []byte) (err error) {
+	return cb.block.reader.Read(cb.ts, cb.id, buf)
+}
+
+func (cb *columnBlock) ReadUpdates(buf []byte) (err error) {
+	return
+}
+
+func (cb *columnBlock) ReadIndex(idx int, buf []byte) (err error) {
+	return
+}
+
+func (cb *columnBlock) GetDataFileStat() (stat common.FileInfo) {
+	return nil
+}
+
+func (cb *columnBlock) OpenIndexFile(idx int) (vfile common.IRWFile, err error) {
+	name := EncodeIndexName(cb.id, idx, nil)
+	vfile, err = cb.block.seg.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return
+	}
+	vfile.Ref()
+	return
+}
+
+func (cb *columnBlock) OpenUpdateFile() (vfile common.IRWFile, err error) {
+	name := EncodeUpdateNameWithVersion(cb.id, cb.ts, nil)
+	vfile, err = cb.block.seg.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return
+	}
+	vfile.Ref()
+	return
+}
+
+func (cb *columnBlock) OpenDataFile() (vfile common.IRWFile, err error) {
+	cb.vfs.Ref()
+	return cb.vfs, nil
+}
+
+func (cb *columnBlock) Close() error {
+	cb.Unref()
+	// cb.data.Unref()
+	// cb.updates.Unref()
+	// for _, index := range cb.indexes {
+	// 	index.Unref()
+	// }
+	return nil
+}
+
+func (cb *columnBlock) close() {
+	cb.Destroy()
+}
+
+func (cb *columnBlock) Destroy() {
+	logutil.Infof("Destroying Block %d Col @ TS %d", cb.block.id, cb.ts)
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/dir.go b/pkg/vm/engine/tae/dataio/objectio/dir.go
new file mode 100644
index 000000000..313d83700
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/dir.go
@@ -0,0 +1,102 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"os"
+)
+
+type ObjectDir struct {
+	common.RefHelper
+	nodes map[string]tfs.File
+	inode *Inode
+	fs    *ObjectFS
+	stat  *objectFileStat
+}
+
+func openObjectDir(fs *ObjectFS, name string) *ObjectDir {
+	inode := &Inode{
+		magic: MAGIC,
+		inode: fs.lastInode,
+		typ:   DIR,
+		name:  name,
+	}
+	file := &ObjectDir{}
+	file.fs = fs
+	file.inode = inode
+	file.nodes = make(map[string]tfs.File)
+	fs.lastInode++
+	return file
+}
+
+func (d *ObjectDir) Stat() common.FileInfo {
+	d.inode.mutex.RLock()
+	defer d.inode.mutex.RUnlock()
+	stat := &objectFileStat{}
+	stat.size = int64(len(d.nodes))
+	stat.dataSize = int64(len(d.nodes))
+	stat.oType = d.inode.typ
+	return stat
+}
+
+func (d *ObjectDir) Read(bytes []byte) (int, error) {
+	return 0, nil
+}
+
+func (d *ObjectDir) Close() error {
+	return nil
+}
+
+func (d *ObjectDir) Write(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (d *ObjectDir) Sync() error {
+	return nil
+}
+
+func (d *ObjectDir) Delete(name string) {
+	d.inode.mutex.Lock()
+	defer d.inode.mutex.Unlock()
+	delete(d.nodes, name)
+}
+
+func (d *ObjectDir) Remove(name string) error {
+	file := d.nodes[name]
+	if file != nil {
+		return os.ErrNotExist
+	}
+	err := d.fs.Delete(file)
+	if err != nil {
+		return err
+	}
+	d.Delete(name)
+	return nil
+}
+
+func (d *ObjectDir) OpenFile(fs *ObjectFS, name string) tfs.File {
+	file := d.nodes[name]
+	if file == nil {
+		file = openObjectFile(fs, name)
+		d.nodes[name] = file
+	}
+	return file
+}
+
+func (d *ObjectDir) GetFileType() common.FileType {
+	return common.DiskFile
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/extent.go b/pkg/vm/engine/tae/dataio/objectio/extent.go
new file mode 100644
index 000000000..ce1463e9e
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/extent.go
@@ -0,0 +1,58 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+type ExtentType uint8
+
+const (
+	APPEND ExtentType = iota
+	UPDATE
+)
+
+type entry struct {
+	offset uint32
+	length uint32
+}
+
+type Extent struct {
+	typ    ExtentType
+	offset uint32
+	length uint32
+	data   entry
+}
+
+func (ex *Extent) End() uint32 {
+	return ex.offset + ex.length
+}
+
+func (ex *Extent) Offset() uint32 {
+	return ex.offset
+}
+
+func (ex *Extent) Length() uint32 {
+	return ex.length
+}
+
+func (ex *Extent) GetData() *entry {
+	return &ex.data
+}
+
+func (en *entry) GetOffset() uint32 {
+	return en.offset
+}
+
+func (en *entry) GetLength() uint32 {
+	return en.length
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/file.go b/pkg/vm/engine/tae/dataio/objectio/file.go
new file mode 100644
index 000000000..8e12ed825
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/file.go
@@ -0,0 +1,109 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+)
+
+type ObjectFile struct {
+	common.RefHelper
+	nodes map[string]*ObjectFile
+	inode *Inode
+	fs    *ObjectFS
+	stat  *objectFileStat
+}
+
+func openObjectFile(fs *ObjectFS, name string) *ObjectFile {
+	inode := &Inode{
+		magic: MAGIC,
+		inode: fs.lastInode,
+		typ:   FILE,
+		name:  name,
+	}
+	file := &ObjectFile{}
+	file.fs = fs
+	file.inode = inode
+	fs.lastInode++
+	return file
+}
+
+func (b *ObjectFile) GetFS() *ObjectFS {
+	return b.fs
+}
+
+func (b *ObjectFile) GetInode() *Inode {
+	b.inode.mutex.RLock()
+	defer b.inode.mutex.RUnlock()
+	return b.inode
+}
+
+func (b *ObjectFile) SetRows(rows uint32) {
+	b.inode.rows = rows
+}
+
+func (b *ObjectFile) SetCols(cols uint32) {
+	b.inode.cols = cols
+}
+
+func (b *ObjectFile) SetIdxs(idxs uint32) {
+	b.inode.idxs = idxs
+}
+
+func (b *ObjectFile) Stat() common.FileInfo {
+	b.inode.mutex.RLock()
+	defer b.inode.mutex.RUnlock()
+	stat := &objectFileStat{}
+	stat.size = int64(b.inode.size)
+	stat.dataSize = int64(b.inode.size)
+	stat.oType = b.inode.typ
+	return stat
+}
+
+func (b *ObjectFile) GetName() string {
+	return b.inode.name
+}
+
+func (b *ObjectFile) Write(data []byte) (n int, err error) {
+	return b.fs.Append(b, data)
+}
+
+func (b *ObjectFile) GetExtents() *[]Extent {
+	extents := &b.inode.extents
+	return extents
+}
+
+func (b *ObjectFile) Read(data []byte) (n int, err error) {
+	return b.fs.Read(b, data)
+}
+
+func (b *ObjectFile) close() {
+	b.Destroy()
+}
+
+func (b *ObjectFile) Destroy() {
+}
+
+func (b *ObjectFile) Close() error {
+	return nil
+}
+
+func (b *ObjectFile) Sync() error {
+	return b.fs.Sync(b)
+}
+
+func (b *ObjectFile) GetFileType() common.FileType {
+	return common.DiskFile
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/fs.go b/pkg/vm/engine/tae/dataio/objectio/fs.go
new file mode 100644
index 000000000..7eb678552
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/fs.go
@@ -0,0 +1,191 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/compress"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"os"
+	"strings"
+	"sync"
+)
+
+type ObjectFS struct {
+	sync.RWMutex
+	common.RefHelper
+	dirs      map[string]tfs.File
+	data      []*Object
+	meta      []*Object
+	attr      *Attr
+	lastId    uint64
+	lastInode uint64
+}
+
+type Attr struct {
+	algo uint8
+	dir  string
+}
+
+func NewObjectFS() tfs.FS {
+	fs := &ObjectFS{
+		attr: &Attr{
+			algo: compress.None,
+		},
+		dirs: make(map[string]tfs.File),
+	}
+	fs.lastId = 1
+	fs.lastInode = 1
+	return fs
+}
+
+func (o *ObjectFS) SetDir(dir string) {
+	o.attr.dir = dir
+}
+
+func (o *ObjectFS) OpenFile(name string, flag int) (tfs.File, error) {
+	o.RWMutex.Lock()
+	defer o.RWMutex.Unlock()
+	fileName := strings.Split(name, "/")
+	dir := o.dirs[fileName[0]]
+	if dir == nil {
+		dir = openObjectDir(o, fileName[0])
+		o.dirs[fileName[0]] = dir
+	}
+	if len(fileName) == 1 {
+		return dir, nil
+	}
+	file := dir.(*ObjectDir).OpenFile(o, fileName[1])
+	return file, nil
+}
+
+func (o *ObjectFS) ReadDir(dir string) ([]common.FileInfo, error) {
+	fileInfos := make([]common.FileInfo, 0)
+	entry := o.dirs[dir]
+	info := entry.Stat()
+	fileInfos = append(fileInfos, info)
+	return fileInfos, nil
+}
+
+func (o *ObjectFS) Remove(name string) error {
+	o.RWMutex.Lock()
+	defer o.RWMutex.Unlock()
+	fileName := strings.Split(name, "/")
+	dir := o.dirs[fileName[0]]
+	if dir == nil {
+		return os.ErrNotExist
+	}
+	return dir.(*ObjectDir).Remove(fileName[1])
+}
+
+func (o *ObjectFS) RemoveAll(dir string) error {
+	return nil
+}
+
+func (o *ObjectFS) MountInfo() *tfs.MountInfo {
+	return nil
+}
+
+func (o *ObjectFS) GetData(size uint64) (object *Object, err error) {
+	o.RWMutex.Lock()
+	defer o.RWMutex.Unlock()
+	if len(o.data) == 0 ||
+		o.data[len(o.data)-1].GetSize()+size >= OBJECT_SIZE {
+		object, err = OpenObject(o.lastId, DATATYPE, o.attr.dir)
+		o.data = append(o.data, object)
+		o.lastId++
+		return
+	}
+	return o.data[len(o.data)-1], nil
+}
+
+func (o *ObjectFS) GetMeta(size uint64) (object *Object, err error) {
+	o.RWMutex.Lock()
+	defer o.RWMutex.Unlock()
+	if len(o.meta) == 0 ||
+		o.meta[len(o.meta)-1].GetSize()+size >= OBJECT_SIZE {
+		object, err = OpenObject(o.lastId, METADATA, o.attr.dir)
+		o.meta = append(o.meta, object)
+		o.lastId++
+		return
+	}
+	return o.meta[len(o.meta)-1], nil
+}
+
+func (o *ObjectFS) GetDataWithId(id uint64) *Object {
+	o.RWMutex.RLock()
+	defer o.RWMutex.RUnlock()
+	for _, object := range o.data {
+		if object.id == id {
+			return object
+		}
+	}
+	return nil
+}
+
+func (o *ObjectFS) Append(file *ObjectFile, data []byte) (n int, err error) {
+	dataObject, err := o.GetData(uint64(len(data)))
+	if err != nil {
+		return
+	}
+	offset, allocated, err := dataObject.Append(data)
+	if err != nil {
+		return int(allocated), err
+	}
+	file.inode.mutex.Lock()
+	file.inode.extents = append(file.inode.extents, Extent{
+		typ:    APPEND,
+		offset: uint32(offset),
+		length: uint32(len(data)),
+		data:   entry{offset: 0, length: uint32(len(data))},
+	})
+	file.inode.size += uint64(len(data))
+	file.inode.dataSize += uint64(len(data))
+	file.inode.seq++
+	file.inode.objectId = dataObject.id
+	file.inode.mutex.Unlock()
+	inode, err := file.inode.Marshal()
+	if err != nil {
+		return int(allocated), err
+	}
+	metaObject, err := o.GetMeta(uint64(len(inode)))
+	if err != nil {
+		return
+	}
+	_, _, err = metaObject.Append(inode)
+	return int(allocated), err
+}
+
+func (o *ObjectFS) Read(file *ObjectFile, data []byte) (n int, err error) {
+	bufLen := len(data)
+	if bufLen == 0 {
+		return 0, nil
+	}
+	inode := file.GetInode()
+	dataObject := o.GetDataWithId(inode.objectId)
+	return dataObject.oFile.ReadAt(data, int64(inode.extents[0].offset))
+}
+
+func (o *ObjectFS) Sync(file *ObjectFile) error {
+	data := o.GetDataWithId(file.inode.objectId)
+	if data == nil {
+		return nil
+	}
+	return data.oFile.Sync()
+}
+
+func (o *ObjectFS) Delete(file tfs.File) error {
+	return nil
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/inode.go b/pkg/vm/engine/tae/dataio/objectio/inode.go
new file mode 100644
index 000000000..c9459527e
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/inode.go
@@ -0,0 +1,230 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"bytes"
+	"encoding/binary"
+	"sync"
+	"unsafe"
+)
+
+type StateType uint8
+type InodeType uint8
+
+const (
+	RESIDENT StateType = iota
+	REMOVE
+)
+
+const (
+	FILE InodeType = iota
+	DIR
+)
+
+const MAGIC = 0xFFFFFFFF
+
+type Inode struct {
+	magic    uint64
+	inode    uint64
+	name     string
+	algo     uint8
+	size     uint64
+	dataSize uint64
+	rows     uint32
+	cols     uint32
+	idxs     uint32
+	mutex    sync.RWMutex
+	extents  []Extent
+	typ      InodeType
+	seq      uint64
+	objectId uint64
+}
+
+func (i *Inode) GetFileSize() int64 {
+	return int64(i.size)
+}
+
+func (i *Inode) GetDataSize() int64 {
+	return int64(i.dataSize)
+}
+
+func (i *Inode) GetAlgo() uint8 {
+	return i.algo
+}
+
+func (i *Inode) GetRows() uint32 {
+	return i.rows
+}
+
+func (i *Inode) GetCols() uint32 {
+	return i.cols
+}
+
+func (i *Inode) GetIdxs() uint32 {
+	return i.idxs
+}
+
+func (i *Inode) Marshal() (buf []byte, err error) {
+	var (
+		buffer bytes.Buffer
+	)
+	if err = binary.Write(&buffer, binary.BigEndian, i.magic); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.inode); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.typ); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, uint32(len([]byte(i.name)))); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, []byte(i.name)); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.seq); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.algo); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.size); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.dataSize); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.rows); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.cols); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, i.idxs); err != nil {
+		return
+	}
+	if err = binary.Write(&buffer, binary.BigEndian, uint64(len(i.extents))); err != nil {
+		return
+	}
+	i.mutex.RLock()
+	extents := i.extents
+	i.mutex.RUnlock()
+	for _, ext := range extents {
+		if err = binary.Write(&buffer, binary.BigEndian, ext.typ); err != nil {
+			return
+		}
+		if err = binary.Write(&buffer, binary.BigEndian, ext.offset); err != nil {
+			return
+		}
+		if err = binary.Write(&buffer, binary.BigEndian, ext.length); err != nil {
+			return
+		}
+		if err = binary.Write(&buffer, binary.BigEndian, ext.data.offset); err != nil {
+			return
+		}
+		if err = binary.Write(&buffer, binary.BigEndian, ext.data.length); err != nil {
+			return
+		}
+	}
+	return buffer.Bytes(), err
+}
+func (i *Inode) UnMarshal(cache *bytes.Buffer, inode *Inode) (n int, err error) {
+	var nameLen uint32
+	var extentLen uint64
+	n = 0
+	if err = binary.Read(cache, binary.BigEndian, &inode.magic); err != nil {
+		return
+	}
+	if inode.magic != MAGIC {
+		return 0, nil
+	}
+	n += int(unsafe.Sizeof(inode.magic))
+	if err = binary.Read(cache, binary.BigEndian, &inode.inode); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.inode))
+	if err = binary.Read(cache, binary.BigEndian, &inode.typ); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.typ))
+	if err = binary.Read(cache, binary.BigEndian, &nameLen); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(nameLen))
+	name := make([]byte, nameLen)
+	if err = binary.Read(cache, binary.BigEndian, name); err != nil {
+		return
+	}
+	n += len(name)
+	inode.name = string(name)
+	if err = binary.Read(cache, binary.BigEndian, &inode.seq); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.seq))
+	if err = binary.Read(cache, binary.BigEndian, &inode.algo); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.algo))
+	if err = binary.Read(cache, binary.BigEndian, &inode.size); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.size))
+	if err = binary.Read(cache, binary.BigEndian, &inode.dataSize); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.dataSize))
+	if err = binary.Read(cache, binary.BigEndian, &inode.rows); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.rows))
+	if err = binary.Read(cache, binary.BigEndian, &inode.cols); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.cols))
+	if err = binary.Read(cache, binary.BigEndian, &inode.idxs); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(inode.idxs))
+	if err = binary.Read(cache, binary.BigEndian, &extentLen); err != nil {
+		return
+	}
+	n += int(unsafe.Sizeof(extentLen))
+	inode.extents = make([]Extent, extentLen)
+	for i := 0; i < int(extentLen); i++ {
+		if err = binary.Read(cache, binary.BigEndian, &inode.extents[i].typ); err != nil {
+			return
+		}
+		n += int(unsafe.Sizeof(inode.extents[i].typ))
+		if err = binary.Read(cache, binary.BigEndian, &inode.extents[i].offset); err != nil {
+			return
+		}
+		n += int(unsafe.Sizeof(inode.extents[i].offset))
+		if err = binary.Read(cache, binary.BigEndian, &inode.extents[i].length); err != nil {
+			return
+		}
+		n += int(unsafe.Sizeof(inode.extents[i].length))
+		if err = binary.Read(cache, binary.BigEndian, &inode.extents[i].data.offset); err != nil {
+			return
+		}
+		n += int(unsafe.Sizeof(inode.extents[i].data.offset))
+		if err = binary.Read(cache, binary.BigEndian, &inode.extents[i].data.length); err != nil {
+			return
+		}
+		n += int(unsafe.Sizeof(inode.extents[i].data.length))
+	}
+	return
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/object.go b/pkg/vm/engine/tae/dataio/objectio/object.go
new file mode 100644
index 000000000..acb0ac184
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/object.go
@@ -0,0 +1,109 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/file"
+	"os"
+	"path"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+const OBJECT_SIZE = 64 * 1024 * 1024
+const PAGE_SIZE = 4096
+
+type ObjectType uint8
+
+const (
+	DATATYPE ObjectType = iota
+	METADATA
+)
+
+const (
+	DATA = "data"
+	META = "meta"
+)
+
+type Object struct {
+	id        uint64
+	mutex     sync.Mutex
+	oFile     *os.File
+	allocator *ObjectAllocator
+	oType     ObjectType
+}
+
+func encodeName(id uint64, oType ObjectType) string {
+	if oType == DATATYPE {
+		return fmt.Sprintf("%d.%s", id, DATA)
+	}
+	return fmt.Sprintf("%d.%s", id, META)
+}
+
+func decodeName(name string) (id uint64, oType ObjectType, err error) {
+	oName := strings.Split(name, ".")
+	if len(oName) != 2 {
+		err = fmt.Errorf("%w: %s", file.ErrInvalidName, name)
+		return
+	}
+	id, err = strconv.ParseUint(oName[0], 10, 64)
+	if err != nil {
+		err = fmt.Errorf("%w: %s", file.ErrInvalidName, name)
+	}
+	if oName[1] == DATA {
+		oType = DATATYPE
+	} else {
+		oType = METADATA
+	}
+	return
+}
+
+func OpenObject(id uint64, oType ObjectType, dir string) (object *Object, err error) {
+	object = &Object{
+		id:    id,
+		oType: oType,
+	}
+	object.allocator = NewObjectAllocator(OBJECT_SIZE, PAGE_SIZE)
+	path := path.Join(dir, encodeName(id, oType))
+	if _, err = os.Stat(path); os.IsNotExist(err) {
+		object.oFile, err = os.Create(path)
+		return
+	}
+
+	if object.oFile, err = os.OpenFile(path, os.O_RDWR, os.ModePerm); err != nil {
+		return
+	}
+	return
+}
+
+func (o *Object) Append(data []byte) (offset, allocated uint64, err error) {
+	offset, allocated = o.allocator.Allocate(uint64(len(data)))
+	_, err = o.oFile.WriteAt(data, int64(offset))
+	if err != nil {
+		return
+	}
+	return
+}
+
+func (o *Object) Read(offset int64, data []byte) (length int, err error) {
+	length, err = o.oFile.ReadAt(data, offset)
+	return
+}
+
+func (o *Object) GetSize() uint64 {
+	return o.allocator.GetAvailable()
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/reader.go b/pkg/vm/engine/tae/dataio/objectio/reader.go
new file mode 100644
index 000000000..c7a31b014
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/reader.go
@@ -0,0 +1,172 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"bytes"
+	"github.com/RoaringBitmap/roaring"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/tables/indexwrapper"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/types"
+	"os"
+)
+
+type Reader struct {
+	block *blockFile
+	fs    tfs.FS
+}
+
+func NewReader(fs tfs.FS, block *blockFile) *Reader {
+	return &Reader{
+		fs:    fs,
+		block: block,
+	}
+}
+
+func (r *Reader) LoadIndexMeta(id *common.ID) (any, error) {
+	name := EncodeMetaIndexName(id, r.fs)
+	f, err := r.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return nil, err
+	}
+	size := f.Stat().Size()
+	buf := make([]byte, size)
+	_, err = f.Read(buf)
+	if err != nil {
+		return nil, err
+	}
+	indices := indexwrapper.NewEmptyIndicesMeta()
+	if err = indices.Unmarshal(buf); err != nil {
+		return nil, err
+	}
+	return indices, nil
+}
+
+func (r *Reader) LoadDeletes(id *common.ID) (mask *roaring.Bitmap, err error) {
+	name := EncodeDeleteName(id, r.fs)
+	f, err := r.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return nil, err
+	}
+	size := f.Stat().Size()
+	if size == 0 {
+		return
+	}
+	node := common.GPool.Alloc(uint64(size))
+	defer common.GPool.Free(node)
+	if _, err = f.Read(node.Buf[:size]); err != nil {
+		return
+	}
+	mask = roaring.New()
+	err = mask.UnmarshalBinary(node.Buf[:size])
+	return
+}
+
+func (r *Reader) Read(
+	version uint64,
+	id *common.ID,
+	data []byte) (err error) {
+	name := EncodeColBlkNameWithVersion(id, version, r.fs)
+	f, err := r.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Read(data)
+	return
+}
+
+func (r *Reader) LoadABlkColumns(
+	colTypes []types.Type,
+	colNames []string,
+	nullables []bool,
+	opts *containers.Options) (bat *containers.Batch, err error) {
+	bat = containers.NewBatch()
+	var f common.IRWFile
+	for i, colBlk := range r.block.columns {
+		if f, err = colBlk.OpenDataFile(); err != nil {
+			return
+		}
+		defer f.Unref()
+		vec := containers.MakeVector(colTypes[i], nullables[i], opts)
+		bat.AddVector(colNames[i], vec)
+		size := f.Stat().Size()
+		if size == 0 {
+			continue
+		}
+		buf := make([]byte, size)
+		if _, err = f.Read(buf); err != nil {
+			return
+		}
+		r := bytes.NewBuffer(buf)
+		if _, err = vec.ReadFrom(r); err != nil {
+			return
+		}
+		bat.Vecs[i] = vec
+	}
+	return
+}
+
+func (r *Reader) LoadUpdates() (masks map[uint16]*roaring.Bitmap, vals map[uint16]map[uint32]any) {
+	tool := containers.NewCodecTool()
+	defer tool.Close()
+	for i, cb := range r.block.columns {
+		tool.Reset()
+		uf, err := cb.OpenUpdateFile()
+		if err != nil {
+			panic(any(err))
+		}
+		defer uf.Unref()
+		if uf.Stat().OriginSize() == 0 {
+			continue
+		}
+		if err := tool.ReadFromFile(uf, nil); err != nil {
+			panic(any(err))
+		}
+		buf := tool.Get(0)
+		typ := types.DecodeType(buf)
+		nullable := containers.GetValueFrom[bool](tool, 1)
+		mask := roaring.New()
+		buf = tool.Get(2)
+		if err = mask.UnmarshalBinary(buf); err != nil {
+			panic(any(err))
+		}
+		buf = tool.Get(3)
+		r := bytes.NewBuffer(buf)
+		vec := containers.MakeVector(typ, nullable)
+		if _, err = vec.ReadFrom(r); err != nil {
+			panic(any(err))
+		}
+		defer vec.Close()
+		val := make(map[uint32]any)
+		it := mask.Iterator()
+		pos := 0
+		for it.HasNext() {
+			row := it.Next()
+			v := vec.Get(pos)
+			val[row] = v
+			pos++
+		}
+		if masks == nil {
+			masks = make(map[uint16]*roaring.Bitmap)
+			vals = make(map[uint16]map[uint32]any)
+		}
+		vals[uint16(i)] = val
+		masks[uint16(i)] = mask
+	}
+	return
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/segment.go b/pkg/vm/engine/tae/dataio/objectio/segment.go
new file mode 100644
index 000000000..0a9a1ac1e
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/segment.go
@@ -0,0 +1,137 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"sync"
+
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/file"
+)
+
+var SegmentFactory file.SegmentFactory
+
+func init() {
+	SegmentFactory = &segmentFactory{fs: NewObjectFS()}
+}
+
+type segmentFactory struct {
+	fs tfs.FS
+}
+
+func (factory *segmentFactory) EncodeName(id uint64) string {
+	return ""
+}
+
+func (factory *segmentFactory) DecodeName(name string) (id uint64, err error) {
+	return 0, nil
+}
+
+func (factory *segmentFactory) Build(dir string, id uint64) file.Segment {
+	factory.fs.(*ObjectFS).SetDir(dir)
+	return openSegment(dir, id, factory.fs)
+}
+
+type segmentFile struct {
+	sync.RWMutex
+	common.RefHelper
+	id     *common.ID
+	ts     uint64
+	blocks map[uint64]*blockFile
+	name   string
+	fs     tfs.FS
+}
+
+func openSegment(name string, id uint64, fs tfs.FS) *segmentFile {
+	sf := &segmentFile{
+		blocks: make(map[uint64]*blockFile),
+		name:   name,
+	}
+	sf.fs = fs
+	sf.id = &common.ID{
+		SegmentID: id,
+	}
+	sf.Ref()
+	sf.OnZeroCB = sf.close
+	return sf
+}
+
+func (sf *segmentFile) Name() string { return sf.name }
+
+func (sf *segmentFile) RemoveBlock(id uint64) {
+	sf.Lock()
+	defer sf.Unlock()
+	block := sf.blocks[id]
+	if block == nil {
+		return
+	}
+	delete(sf.blocks, id)
+}
+
+func (sf *segmentFile) Fingerprint() *common.ID { return sf.id }
+func (sf *segmentFile) Close() error            { return nil }
+
+func (sf *segmentFile) close() {
+	sf.Destroy()
+}
+func (sf *segmentFile) Destroy() {
+	logutil.Infof("Destroying Driver %d", sf.id.SegmentID)
+	sf.RLock()
+	blocks := sf.blocks
+	sf.RUnlock()
+	for _, block := range blocks {
+		if err := block.Destroy(); err != nil {
+			panic(any(err))
+		}
+	}
+	sf.fs = nil
+}
+
+func (sf *segmentFile) OpenBlock(id uint64, colCnt int, indexCnt map[int]int) (block file.Block, err error) {
+	sf.Lock()
+	defer sf.Unlock()
+	bf := sf.blocks[id]
+	if bf == nil {
+		bf = newBlock(id, sf, colCnt, indexCnt)
+		sf.blocks[id] = bf
+	}
+	block = bf
+	return
+}
+
+func (sf *segmentFile) WriteTS(ts uint64) error {
+	sf.ts = ts
+	return nil
+}
+
+func (sf *segmentFile) ReadTS() uint64 {
+	return sf.ts
+}
+
+func (sf *segmentFile) String() string {
+	s := fmt.Sprintf("SegmentFile[%d][\"%s\"][TS=%d][BCnt=%d]", sf.id, sf.name, sf.ts, len(sf.blocks))
+	return s
+}
+
+func (sf *segmentFile) GetFs() tfs.FS {
+	return sf.fs
+}
+
+func (sf *segmentFile) Sync() error {
+	return nil
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/stat.go b/pkg/vm/engine/tae/dataio/objectio/stat.go
new file mode 100644
index 000000000..48da54d7b
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/stat.go
@@ -0,0 +1,52 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/compress"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"io/fs"
+	"time"
+)
+
+type objectFileStat struct {
+	common.FileInfo
+	name     string
+	size     int64
+	dataSize int64
+	oType    InodeType
+}
+
+func (stat *objectFileStat) Mode() fs.FileMode {
+	return 0
+}
+
+func (stat *objectFileStat) ModTime() time.Time {
+	return time.Time{}
+}
+
+func (stat *objectFileStat) IsDir() bool {
+	return stat.oType == DIR
+}
+
+func (stat *objectFileStat) Sys() any {
+	return nil
+}
+
+func (stat *objectFileStat) Name() string      { return stat.name }
+func (stat *objectFileStat) Size() int64       { return stat.size }
+func (stat *objectFileStat) DataSize() int64   { return stat.dataSize }
+func (stat *objectFileStat) OriginSize() int64 { return stat.size }
+func (stat *objectFileStat) CompressAlgo() int { return compress.None }
diff --git a/pkg/vm/engine/tae/dataio/objectio/types.go b/pkg/vm/engine/tae/dataio/objectio/types.go
new file mode 100644
index 000000000..e838825c0
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/types.go
@@ -0,0 +1,103 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"path/filepath"
+
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+)
+
+type Extension int16
+
+const (
+	ColumnBlockExt Extension = iota + 1000
+	ComposedUpdatesExt
+	ColumnUpdatesExt
+	DeletesExt
+	IndexExt
+	MetaIndexExt
+)
+
+var ExtName map[Extension]string = map[Extension]string{
+	ColumnBlockExt:     "cblk",
+	ComposedUpdatesExt: "cus",
+	ColumnUpdatesExt:   "cu",
+	DeletesExt:         "del",
+	IndexExt:           "idx",
+	MetaIndexExt:       "midx",
+}
+
+func ExtensionName(ext Extension) (name string) {
+	name, found := ExtName[ext]
+	if !found {
+		panic(any(fmt.Sprintf("Unknown ext: %d", ext)))
+	}
+	return
+}
+
+func EncodeDir(id *common.ID) (dir string) {
+	return fmt.Sprintf("%d-%d", id.SegmentID, id.BlockID)
+}
+
+func EncodeColBlkNameWithVersion(id *common.ID, version uint64, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d-%d.%s", id.Idx, version, ExtensionName(ColumnBlockExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+func EncodeIndexName(id *common.ID, idx int, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d-%d.%s", id.Idx, idx, ExtensionName(IndexExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+func EncodeMetaIndexName(id *common.ID, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d.%s", id.BlockID, ExtensionName(MetaIndexExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+func EncodeDeleteName(id *common.ID, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d.%s", id.BlockID, ExtensionName(DeletesExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+func EncodeDeleteNameWithVersion(id *common.ID, version uint64, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d-%d.%s", id.BlockID, version, ExtensionName(DeletesExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+func EncodeUpdateNameWithVersion(id *common.ID, version uint64, fs tfs.FS) (name string) {
+	dir := EncodeDir(id)
+	basename := fmt.Sprintf("%d-%d.%s", id.Idx, version, ExtensionName(ColumnUpdatesExt))
+	name = filepath.Join(dir, basename)
+	return
+}
+
+// func EncodeComposedColumnFileName(id *common.ID, attrs []int, fs file.FS) (name string) {
+// 	dir := fs.EncodeDir(id)
+// 	name = fs.Join(dir, common.IntsToStr(attrs, "-"))
+// 	return
+// }
diff --git a/pkg/vm/engine/tae/dataio/objectio/vfs.go b/pkg/vm/engine/tae/dataio/objectio/vfs.go
new file mode 100644
index 000000000..fe956db8b
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/vfs.go
@@ -0,0 +1,55 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"os"
+)
+
+type VFS struct {
+	common.RefHelper
+	cb *columnBlock
+}
+
+func (v *VFS) Write(p []byte) (n int, err error) {
+	return 0, v.cb.WriteData(p)
+}
+
+func (v *VFS) Read(p []byte) (n int, err error) {
+	err = v.cb.ReadData(p)
+	if err != nil {
+		return 0, err
+	}
+	n = len(p)
+	return
+}
+
+func (v *VFS) RefCount() int64 {
+	return 0
+}
+
+func (v *VFS) Stat() common.FileInfo {
+	name := EncodeColBlkNameWithVersion(v.cb.id, v.cb.ts, nil)
+	vfile, err := v.cb.block.seg.fs.OpenFile(name, os.O_RDWR)
+	if err != nil {
+		return nil
+	}
+	return vfile.Stat()
+}
+
+func (v *VFS) GetFileType() common.FileType {
+	return common.DiskFile
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/writer.go b/pkg/vm/engine/tae/dataio/objectio/writer.go
new file mode 100644
index 000000000..e00751124
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/writer.go
@@ -0,0 +1,172 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/tfs"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/stl/adaptors"
+	"os"
+
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/catalog"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+)
+
+type Writer struct {
+	fs    tfs.FS
+	block blockFile
+}
+
+func NewWriter(fs tfs.FS) *Writer {
+	return &Writer{
+		fs: fs,
+	}
+}
+
+func (w *Writer) WriteABlkColumn(
+	version uint64,
+	id *common.ID,
+	column containers.Vector) (info common.FileInfo, err error) {
+	name := EncodeColBlkNameWithVersion(id, version, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	info = f.Stat()
+	writerBuffer := adaptors.NewBuffer(nil)
+	defer writerBuffer.Close()
+	if _, err = column.WriteTo(writerBuffer); err != nil {
+		return
+	}
+	if _, err = f.Write(writerBuffer.Bytes()); err != nil {
+		return
+	}
+	return
+}
+
+func (w *Writer) WriteABlkColumns(
+	version uint64,
+	id *common.ID,
+	columns *containers.Batch) (infos []common.FileInfo, err error) {
+	var info common.FileInfo
+	for colIdx := range columns.Attrs {
+		colId := *id
+		colId.Idx = uint16(colIdx)
+		if info, err = w.WriteABlkColumn(version, &colId, columns.Vecs[colIdx]); err != nil {
+			return
+		}
+		infos = append(infos, info)
+	}
+	return
+}
+
+func (w *Writer) WriteBlockColumn(
+	version uint64,
+	id *common.ID,
+	column *vector.Vector) (err error) {
+	name := EncodeColBlkNameWithVersion(id, version, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	buf, err := column.Show()
+	if err != nil {
+		return
+	}
+	_, err = f.Write(buf)
+	return
+}
+
+func (w *Writer) WriteUpdates(
+	version uint64,
+	id *common.ID,
+	data []byte) (err error) {
+	name := EncodeUpdateNameWithVersion(id, version, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Write(data)
+	return
+}
+
+func (w *Writer) WriteDeletes(
+	version uint64,
+	id *common.ID,
+	data []byte) (err error) {
+	name := EncodeDeleteNameWithVersion(id, version, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Write(data)
+	return
+}
+
+func (w *Writer) WriteIndexMeta(
+	id *common.ID,
+	data []byte) (err error) {
+	name := EncodeMetaIndexName(id, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Write(data)
+	return
+}
+
+func (w *Writer) WriteIndex(
+	id *common.ID,
+	idx int,
+	data []byte) (err error) {
+	name := EncodeIndexName(id, idx, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Write(data)
+	return
+}
+
+func (w *Writer) WriteData(
+	version uint64,
+	id *common.ID,
+	data []byte) (err error) {
+	name := EncodeColBlkNameWithVersion(id, version, w.fs)
+	f, err := w.fs.OpenFile(name, os.O_CREATE)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+	_, err = f.Write(data)
+	return
+}
+
+func (w *Writer) WriteZonemapIndexFromSource(
+	version uint64,
+	cols []int,
+	indexT catalog.IndexT,
+	id *common.ID,
+	source *vector.Vector) (err error) {
+	// TODO
+	return
+}
diff --git a/pkg/vm/engine/tae/iface/tfs/types.go b/pkg/vm/engine/tae/iface/tfs/types.go
new file mode 100644
index 000000000..851414317
--- /dev/null
+++ b/pkg/vm/engine/tae/iface/tfs/types.go
@@ -0,0 +1,29 @@
+package tfs
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"io"
+)
+
+type File interface {
+	common.IVFile
+	io.Writer
+	Sync() error
+	Close() error
+}
+
+type MountInfo struct {
+	Dir  string
+	Misc []byte
+}
+
+type MkFsFuncT = func(name string) (FS, error)
+type MountFsFuncT = func(name string) (FS, error)
+
+type FS interface {
+	OpenFile(name string, flag int) (File, error)
+	ReadDir(dir string) ([]common.FileInfo, error)
+	Remove(name string) error
+	RemoveAll(dir string) error
+	MountInfo() *MountInfo
+}


Test patch:
diff --git a/pkg/vm/engine/tae/dataio/objectio/block_test.go b/pkg/vm/engine/tae/dataio/objectio/block_test.go
new file mode 100644
index 000000000..f065682aa
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/block_test.go
@@ -0,0 +1,79 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"bytes"
+	"github.com/RoaringBitmap/roaring"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/file"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/testutils"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestBlock1(t *testing.T) {
+	dir := testutils.InitTestEnv(ModuleName, t)
+	colCnt := 4
+	indexCnt := make(map[int]int)
+	for col := 0; col < colCnt; col++ {
+		indexCnt[col] = 2
+	}
+	var block file.Block
+	id := common.NextGlobalSeqNum()
+	seg := SegmentFactory.Build(dir, id).(*segmentFile)
+	block = newBlock(common.NextGlobalSeqNum(), seg, colCnt, indexCnt)
+	blockTs := common.NextGlobalSeqNum()
+	err := block.WriteTS(blockTs)
+	assert.Nil(t, err)
+	readTs, _ := block.ReadTS()
+	assert.Equal(t, blockTs, readTs)
+
+	deletes := roaring.New()
+	deletes.Add(10)
+	deletes.Add(20)
+	deletesBuf, _ := deletes.ToBytes()
+	err = block.WriteDeletes(deletesBuf)
+	assert.Nil(t, err)
+
+	_, err = block.OpenColumn(colCnt)
+	assert.NotNil(t, err)
+
+	colBlk0, err := block.OpenColumn(0)
+	assert.Nil(t, err)
+	assert.NotNil(t, colBlk0)
+	var w bytes.Buffer
+	dataStr := "hello tae"
+	w.WriteString(dataStr)
+	err = colBlk0.WriteTS(blockTs)
+	assert.Nil(t, err)
+	err = colBlk0.WriteData(w.Bytes())
+	assert.Nil(t, err)
+
+	dataFile, err := colBlk0.OpenDataFile()
+	assert.Nil(t, err)
+	size := dataFile.Stat().Size()
+	assert.Equal(t, int64(len(dataStr)), dataFile.Stat().OriginSize())
+	buf := make([]byte, size)
+	_, err = dataFile.Read(buf)
+	assert.Nil(t, err)
+	assert.Equal(t, dataStr, string(buf))
+	t.Log(string(buf))
+
+	dataFile.Unref()
+	colBlk0.Close()
+
+	block.Unref()
+}
diff --git a/pkg/vm/engine/tae/dataio/objectio/segment_test.go b/pkg/vm/engine/tae/dataio/objectio/segment_test.go
new file mode 100644
index 000000000..1acfac881
--- /dev/null
+++ b/pkg/vm/engine/tae/dataio/objectio/segment_test.go
@@ -0,0 +1,47 @@
+// Copyright 2021 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package objectio
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/testutils"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+const (
+	ModuleName = "OBJECTIO"
+)
+
+func TestSgment1(t *testing.T) {
+	dir := testutils.InitTestEnv(ModuleName, t)
+	id := common.NextGlobalSeqNum()
+	seg := SegmentFactory.Build(dir, id)
+	fp := seg.Fingerprint()
+	assert.Equal(t, id, fp.SegmentID)
+
+	blkId1 := common.NextGlobalSeqNum()
+	blk1, err := seg.OpenBlock(blkId1, 2, nil)
+	assert.Nil(t, err)
+	blkTs1 := common.NextGlobalSeqNum()
+	err = blk1.WriteTS(blkTs1)
+	assert.Nil(t, err)
+
+	ts, _ := blk1.ReadTS()
+	assert.Equal(t, blkTs1, ts)
+	blk1.Close()
+	t.Log(seg.String())
+	seg.Unref()
+}