merge current message queue code changes (#6201)

* listing files to convert to parquet

* write parquet files

* save logs into parquet files

* pass by value

* compact logs into parquet format

* can skip existing files

* refactor

* refactor

* fix compilation

* when no partition found

* refactor

* add untested parquet file read

* rename package

* refactor

* rename files

* remove unused

* add merged log read func

* parquet wants to know the file size

* rewind by time

* pass in stop ts

* add stop ts

* adjust log

* minor

* adjust log

* skip .parquet files when reading message logs

* skip non message files

* Update subscriber_record.go

* send messages

* skip message data with only ts

* skip non log files

* update parquet-go package

* ensure a valid record type

* add new field to a record type

* Update read_parquet_to_log.go

* fix parquet file name generation

* separating reading parquet and logs

* add key field

* add skipped logs

* use in memory cache

* refactor

* refactor

* refactor

* refactor, and change compact log

* refactor

* rename

* refactor

* fix format

* prefix v to version directory

1 files changed, 20 insertions, 0 deletions

diff --git a/weed/util/chunk_cache/chunk_cache_in_memory.go b/weed/util/chunk_cache/chunk_cache_in_memory.go
index 2982d0979..dd101996e 100644
--- a/weed/util/chunk_cache/chunk_cache_in_memory.go
+++ b/weed/util/chunk_cache/chunk_cache_in_memory.go
@@ -5,11 +5,31 @@ import (
 	"time"
 )
 
+var (
+	_ ChunkCache = &ChunkCacheInMemory{}
+)
+
 // a global cache for recently accessed file chunks
 type ChunkCacheInMemory struct {
 	cache *ccache.Cache
 }
 
+func (c *ChunkCacheInMemory) ReadChunkAt(data []byte, fileId string, offset uint64) (n int, err error) {
+	return c.readChunkAt(data, fileId, offset)
+}
+
+func (c *ChunkCacheInMemory) IsInCache(fileId string, lockNeeded bool) (answer bool) {
+	item := c.cache.Get(fileId)
+	if item == nil {
+		return false
+	}
+	return true
+}
+
+func (c *ChunkCacheInMemory) GetMaxFilePartSizeInCache() (answer uint64) {
+	return 8 * 1024 * 1024
+}
+
 func NewChunkCacheInMemory(maxEntries int64) *ChunkCacheInMemory {
 	pruneCount := maxEntries >> 3
 	if pruneCount <= 0 {