123123

gistfile1.txt

/*
mysql

drop table if exists file;
create table file
(
    id        bigint primary key,
    name      varchar(200),
    parent_id bigint,
    is_dir    bigint,
    index idx_parent_id (parent_id),
    index idx_parent_id_id (parent_id, id),
    index idx_id_parent_id (id, parent_id)
);


desc
SELECT count(*)
FROM (SELECT id, parent_id, `name` FROM file where parent_id != 0) a,
     (SELECT @pv := 31) b
WHERE (FIND_IN_SET(parent_id, @pv) != 0 AND @pv := concat(@pv, ',', id));


desc
WITH RECURSIVE temp_dept AS (SELECT id, `name`, parent_id
                             FROM file
                             WHERE id = 2
                             UNION ALL
                             SELECT d.id, d.`name`, d.parent_id
                             FROM file d
                                      JOIN temp_dept sd ON sd.id = d.parent_id)
SELECT count(*)
FROM temp_dept;

select parent_id,count(*) c
from file
group by parent_id
order by c desc;


select *
from file
order by id;
*/

/*

drop table file;
create table file
(
    id        Int64,
    name      String,
    parent_id Int64,
    is_dir    Int64
) ENGINE MergeTree()
      PRIMARY KEY (id)
      order by id;
alter table file
    add index idx_parent_id parent_id type minmax granularity 3;


select *
from file
-- group by parent_id, id, name
order by id;

select *
from file where parent_id = 0 order by id;

select parent_id,count() as c
from file group by parent_id order by c desc ;


*/

package iter

import (
	"bytes"
	"database/sql"
	"fmt"
	"github.com/ClickHouse/clickhouse-go/v2"
	"github.com/samber/lo"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"math/rand"
	"strconv"
	"sync"
	"testing"
)

type File struct {
	Id       int64  `gorm:"column:id"`
	Name     string `gorm:"column:name"`
	ParentId int64  `gorm:"column:parent_id"`
	IsDir    int64  `gorm:"column:is_dir"`
}

func (File) TableName() string {
	return "file"
}

var (
	max   = 10002000
	start = 20
)

func nextFile(id int64, idDir bool, pid int64) File {
	var dir = 0
	if idDir {
		dir = 1
	}
	if pid == id || pid < 0 {
		pid = 0
	}
	return File{
		Id:       id,
		Name:     fmt.Sprintf("file_%d", id),
		ParentId: pid,
		IsDir:    int64(dir),
	}
}

func nextStr(id int64, idDir bool, pid int64) string {
	var dir = 0
	if idDir {
		dir = 1
	}
	if pid == id || pid < 0 {
		pid = 0
	}
	return fmt.Sprintf("(%d,'%s',%d,%d)", id, "file_"+strconv.FormatInt(id, 10), pid, dir)
}

func Test(t *testing.T) {
	// 参考 https://github.com/go-sql-driver/mysql#dsn-data-source-name 获取详情
	dsn := "root:qwe@tcp(127.0.0.1:3306)/fzdwx?charset=utf8mb4&parseTime=True&loc=Local"
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		panic(err)
	}

	var files []File
	var dirsArray = make([]int64, 0)
	var pid int64 = 0
	for i := start; i < max; i++ {
		if len(dirsArray) > 100 {
			pid = dirsArray[rand.Int63n(int64(len(dirsArray)))]
		}
		if i%5000 == 0 {
			if err := db.CreateInBatches(files, 5000).Error; err != nil {
				panic(err)
			}
			files = []File{}
		}

		id := int64(i)
		dir := rand.Int()%3 == 0
		if dir {
			dirsArray = append(dirsArray, id)
		}
		files = append(files, nextFile(int64(i), dir, pid))
	}
}

func TestQueryMySQL(t *testing.T) {
	// 参考 https://github.com/go-sql-driver/mysql#dsn-data-source-name 获取详情
	dsn := "root:qwe@tcp(127.0.0.1:3306)/fzdwx?charset=utf8mb4&parseTime=True&loc=Local"
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		panic(err)
	}

	conn, err := db.DB()
	if err != nil {
		panic(err)
	}

	var f f
	err = f.findAllChildren(conn, []int64{67})
	if err != nil {
		panic(err)
	}

	fmt.Println(len(f.files))
}

func TestCK(t *testing.T) {
	conn, err := connect()
	if err != nil {
		panic(err)
	}

	var dirsArray = make([]int64, 0)

	var buf bytes.Buffer
	var pid int64 = 0
	buf.WriteString("insert into file values")
	for i := 1; i < max; i++ {
		if len(dirsArray) > 100 {
			pid = dirsArray[rand.Int63n(int64(len(dirsArray)))]
		}
		if i%5000 == 0 {
			_, err = conn.Query(buf.String())
			if err != nil {
				panic(err)
			}
			buf.Reset()
			buf.WriteString("insert into file values")
		}
		id := int64(i)
		dir := rand.Int()%2 == 0
		if dir {
			dirsArray = append(dirsArray, id)
		}
		str := nextStr(id, dir, pid)
		buf.WriteString(str)
	}
}

func TestQueryCK(t *testing.T) {
	conn, err := connect()
	if err != nil {
		panic(err)
	}

	var q f
	err = q.findAllChildren(conn, []int64{206})
	q.Wait.Wait()
	if err != nil {
		panic(err)
	}
	fmt.Println(len(q.files))
}

type f struct {
	files []File
	async bool
	err   error

	filesMx sync.Mutex
	Wait    sync.WaitGroup
}

func (f *f) findAllChildren(conn *sql.DB, pids []int64) error {
	f.Wait.Add(1)
	defer f.Wait.Done()
	if f.err != nil {
		return f.err
	}

	children, err := findByPid(conn, pids)
	if err != nil {
		return err
	}
	if len(children) == 0 {
		return nil
	}
	f.filesMx.Lock()
	f.files = append(f.files, children...)
	f.filesMx.Unlock()

	var ids []int64
	for i := range children {
		child := children[i]
		if child.IsDir == 1 {
			ids = append(ids, child.Id)
		}
	}

	lo.ForEach(lo.Chunk(ids, 200000), func(item []int64, index int) {
		f.Wait.Add(1)
		go func(item []int64) {
			defer f.Wait.Done()
			if f.err != nil {
				return
			}
			if err = f.findAllChildren(conn, item); err != nil {
				f.err = err
			}
		}(item)
	})

	return nil
}

func findByPid(conn *sql.DB, pids []int64) ([]File, error) {
	if len(pids) == 0 {
		return []File{}, nil
	}

	query, err := conn.Query("select * from file where parent_id in " + toArray(pids))
	if err != nil {
		return nil, err
	}

	var result []File
	for query.Next() {
		var file = File{}
		if err := query.Scan(&file.Id, &file.Name, &file.ParentId, &file.IsDir); err != nil {
			panic(err)
		}
		result = append(result, file)
	}

	return result, nil
}

func toArray(pids []int64) string {
	var buf bytes.Buffer
	buf.WriteString("(")
	for i := range pids {
		if i != 0 {
			buf.WriteByte(',')
		}
		buf.WriteString(strconv.FormatInt(pids[i], 10))
	}
	buf.WriteString(")")

	return buf.String()
}

func connect() (*sql.DB, error) {
	var (
		conn = clickhouse.OpenDB(&clickhouse.Options{
			Addr: []string{fmt.Sprintf("localhost:8123")},
			Auth: clickhouse.Auth{
				Database: "default",
				Username: "default",
			},
			TLS:      nil,
			Protocol: clickhouse.HTTP,
			Settings: map[string]any{
				"max_query_size": 1024 * 1024 * 1024 * 1024,
			},
		})
	)

	if err := conn.Ping(); err != nil {
		if exception, ok := err.(*clickhouse.Exception); ok {
			fmt.Printf("Exception [%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
		}
		return nil, err
	}
	return conn, nil
}