arrow go parquet wirting leads to memory leak

[darren2013]

Describe the bug, including details regarding any error messages, version, and platform.

darren@darrendus-Mac-mini ~ % go tool pprof -text http://10.20.183.250:6060/debug/pprof/heap | head -30
Fetching profile over HTTP from http://10.20.183.250:6060/debug/pprof/heap
Saved profile in /Users/darren/pprof/pprof.log_fusion.alloc_objects.alloc_space.inuse_objects.inuse_space.012.pb.gz
File: log_fusion
Build ID: aee9d00c9e5d647228e1f8922ddc80e2115f794f
Type: inuse_space
Time: 2025-09-12 19:04:55 CST
Showing nodes accounting for 14863.66MB, 95.57% of 15552.03MB total
Dropped 241 nodes (cum <= 77.76MB)
flat flat% sum% cum cum%
3141.12MB 20.20% 20.20% 3141.12MB 20.20% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewColumnMetaData (inline)
2053.06MB 13.20% 33.40% 2053.06MB 13.20% github.com/apache/arrow-go/v18/parquet/metadata.(*ColumnChunkMetaDataBuilder).Finish
2022.75MB 13.01% 46.41% 2022.75MB 13.01% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewStatistics (inline)
1683.18MB 10.82% 57.23% 6348.41MB 40.82% github.com/apache/arrow-go/v18/parquet/file.(*columnWriter).Close
1630.68MB 10.49% 67.71% 5779.84MB 37.16% github.com/apache/arrow-go/v18/parquet/metadata.(*RowGroupMetaDataBuilder).NextColumnChunk
1389.17MB 8.93% 76.65% 14417.33MB 92.70% github.com/apache/arrow-go/v18/parquet/file.(*rowGroupWriter).NextColumn
871.98MB 5.61% 82.25% 871.98MB 5.61% github.com/apache/arrow-go/v18/arrow/memory.(*GoAllocator).Allocate (partial-inline)
764.03MB 4.91% 87.17% 1008.04MB 6.48% github.com/apache/arrow-go/v18/parquet/metadata.NewColumnChunkMetaDataBuilderWithContents (inline)
344.82MB 2.22% 89.38% 344.82MB 2.22% bufio.NewWriterSize (inline)
244MB 1.57% 90.95% 244MB 1.57% github.com/apache/arrow-go/v18/parquet/schema.ColumnPathFromNode
215.05MB 1.38% 92.33% 269.10MB 1.73% github.com/apache/arrow-go/v18/parquet/pqarrow.writeDenseArrow
128.99MB 0.83% 93.16% 128.99MB 0.83% github.com/apache/arrow-go/v18/parquet/metadata.NewRowGroupMetaDataBuilder (inline)
97.67MB 0.63% 93.79% 97.67MB 0.63% reflect.mapassign_faststr0
89.58MB 0.58% 94.37% 89.58MB 0.58% github.com/apache/arrow-go/v18/internal/hashing.NewHashTable[go.shape.int32] (inline)
37.51MB 0.24% 94.61% 428.80MB 2.76% github.com/apache/arrow-go/v18/parquet/metadata.NewByteArrayStatistics
29.01MB 0.19% 94.80% 824.28MB 5.30% github.com/apache/arrow-go/v18/parquet/file.NewByteArrayColumnChunkWriter
29MB 0.19% 94.98% 179.68MB 1.16% encoding/json.(*decodeState).object
23.50MB 0.15% 95.13% 711.76MB 4.58% github.com/apache/arrow-go/v18/parquet/internal/encoding.byteArrayEncoderTraits.Encoder
16.50MB 0.11% 95.24% 14825.96MB 95.33% github.com/apache/arrow-go/v18/parquet/pqarrow.(*arrowColumnWriter).Write
16MB 0.1% 95.34% 701.82MB 4.51% github.com/apache/arrow-go/v18/parquet/internal/encoding.NewPooledBufferWriter (inline)
14.52MB 0.093% 95.44% 194.20MB 1.25% github.com/darren2013/log_fusion_source/internal/source.FileConsumer
7MB 0.045% 95.48% 351.82MB 2.26% github.com/apache/thrift/lib/go/thrift.NewStreamTransportW (inline)
5MB 0.032% 95.51% 95.58MB 0.61% github.com/apache/arrow-go/v18/internal/hashing.NewBinaryMemoTable

the momry is increasing unitil there is no free memory to use

the witie code as follows
`func (p *ParquetProcessor) flushBatch(entities []model.Event) error {
if len(entities) == 0 {
return nil
}

// 检查Arrow Schema是否已初始化
if p.arrowSchema == nil {
	return fmt.Errorf("arrow schema not initialized")
}

// 检查Arrow写入器是否已初始化
if p.arrowWriter == nil {
	return fmt.Errorf("arrow writer not initialized")
}

// 当前时间
now := time.Now().UnixMilli()

// 优化Arrow Record构建器的使用
// 每次创建新的RecordBuilder以确保内存正确释放
if p.recordBuilder != nil {
	p.recordBuilder.Release()
	p.recordBuilder = nil
}
p.recordBuilder = array.NewRecordBuilder(p.mem, p.arrowSchema)

// 按字段填充数据
for i, field := range p.arrowSchema.Fields() {
	fieldName := field.Name

	// timestamp字段特殊处理
	if fieldName == "timestamp" {
		builder := p.recordBuilder.Field(i).(*array.Int64Builder)
		for j := 0; j < len(entities); j++ {
			builder.Append(now)
		}
		continue
	}

	// 根据字段类型填充数据
	switch field.Type.ID() {
	case arrow.STRING:
		builder := p.recordBuilder.Field(i).(*array.StringBuilder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var strVal string
				switch v := val.(type) {
				case string:
					strVal = v
				default:
					strVal = fmt.Sprintf("%v", v)
				}
				builder.Append(strVal)
			}
		}

	case arrow.INT32:
		builder := p.recordBuilder.Field(i).(*array.Int32Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var intVal int32
				switch v := val.(type) {
				case int:
					intVal = int32(v)
				case int64:
					intVal = int32(v)
				case float64:
					intVal = int32(v)
				case string:
					var f float64
					if _, err := fmt.Sscanf(v, "%f", &f); err == nil {
						intVal = int32(f)
					}
				}
				builder.Append(intVal)
			}
		}

	case arrow.INT64:
		builder := p.recordBuilder.Field(i).(*array.Int64Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var int64Val int64
				switch v := val.(type) {
				case int:
					int64Val = int64(v)
				case int64:
					int64Val = v
				case float64:
					int64Val = int64(v)
				case time.Time:
					int64Val = v.UnixMilli()
				case string:
					var f float64
					if _, err := fmt.Sscanf(v, "%f", &f); err == nil {
						int64Val = int64(f)
					} else {
						// 尝试解析时间字符串
						formats := []string{
							time.RFC3339,
							"2006-01-02 15:04:05",
							"2006/01/02 15:04:05",
							"01/02/2006 15:04:05",
							"02/01/2006 15:04:05",
						}
						for _, format := range formats {
							if t, err := time.Parse(format, v); err == nil {
								int64Val = t.UnixMilli()
								break
							}
						}
					}
				}
				builder.Append(int64Val)
			}
		}

	case arrow.FLOAT64:
		builder := p.recordBuilder.Field(i).(*array.Float64Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var floatVal float64
				switch v := val.(type) {
				case float64:
					floatVal = v
				case float32:
					floatVal = float64(v)
				case int:
					floatVal = float64(v)
				case int64:
					floatVal = float64(v)
				case string:
					fmt.Sscanf(v, "%f", &floatVal)
				}
				builder.Append(floatVal)
			}
		}

	case arrow.BOOL:
		builder := p.recordBuilder.Field(i).(*array.BooleanBuilder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var boolVal bool
				switch v := val.(type) {
				case bool:
					boolVal = v
				case string:
					switch v {
					case "true", "True", "TRUE", "1":
						boolVal = true
					case "false", "False", "FALSE", "0":
						boolVal = false
					}
				case int:
					boolVal = v != 0
				case float64:
					boolVal = v != 0
				}
				builder.Append(boolVal)
			}
		}

	default:
		log.Printf("警告: 字段 %s 的类型 %s 不支持", fieldName, field.Type.ID())
	}
}

// 构建Arrow RecordBatch
recordBatch := p.recordBuilder.NewRecordBatch()
defer recordBatch.Release()

// 使用Arrow Go v18最新的方式写入Parquet文件
if err := p.arrowWriter.Write(recordBatch); err != nil {
	return fmt.Errorf("failed to write record batch to parquet: %w", err)
}

// 更新写入计数器和时间
p.writeCount++
p.lastFlushTime = time.Now()
p.currentRowGroupSize += int64(len(entities))
p.totalRecords += int64(len(entities))

// 温和的内存管理策略：记录写入统计信息
if p.writeCount%1000 == 0 {
	log.Printf("ParquetProcessor[%d] 已写入 %d 次，行组大小: %d，总记录: %d",
		p.instanceID, p.writeCount, p.currentRowGroupSize, p.totalRecords)
}

// 内存优化策略：通过控制批处理大小来减少内存压力
if len(entities) > 1000 {
	log.Printf("ParquetProcessor[%d] 警告：批处理大小较大 (%d)，可能增加内存压力", p.instanceID, len(entities))
}

// 🔥 核心内存优化：智能批处理级别的C++资源释放
// 每10批数据或每1000条记录后释放C++对象资源，平衡性能和内存
if p.writeCount%10 == 0 || len(entities) >= 1000 {
	if err := p.forceCppMemoryRelease(); err != nil {
		log.Printf("ParquetProcessor[%d] C++资源释放失败: %v", p.instanceID, err)
		// 不返回错误，继续处理后续数据
	}
}

// 层级2: 每10万条记录进行深度内存释放（文件分段）- 降低阈值以防止内存溢出
/*if p.totalRecords > 0 && p.totalRecords%100000 == 0 {
	log.Printf("ParquetProcessor[%d] 总记录数达到 %d，开始深度内存释放", p.instanceID, p.totalRecords)

	if err := p.forceRowGroupFlush(); err != nil {
		log.Printf("ParquetProcessor[%d] 深度内存释放失败: %v", p.instanceID, err)
	} else {
		p.currentRowGroupSize = 0 // 重置行组计数器
		log.Printf("ParquetProcessor[%d] 深度内存释放成功，新文件已创建", p.instanceID)
	}
}*/

// 调用TotalCounter.CountN方法，记录成功写入的记录数
(*p.TotalCounter).CountN(int64(len(entities)))

return nil

}`

Component(s)

Parquet

[zeroshade]

Can you share the rest of the relevant code for the object? Such as where the arrowWriter is created and the forceRowGroupFlush method?

[darren2013]

Can you share the rest of the relevant code for the object? Such as where the arrowWriter is created and the forceRowGroupFlush method?

The extra messsage:

1. the entrypoint method is flushBatch,the model.event represent a log recored,contains 909 fields.The record size is 1.7k on average
2. There are 4 gouritines,every gouritine has a ParquetProcessor instance
3. pqarrow.FileWriter has a long lifecycle,until the program exit or terminate it manually
4. I tried every 500000 records,close the filewirter,and recreate a new filewrite,the memory will release significantly.In this way,will create many small files,and the important,I am not sure when the memory is full,may be 300000 or 1000000.Ideally,
   The memory is released,when the batch writing finished
   5.the attachment is pprof file,you can analyze by execute "go tool pprof -http=:8080 pprof.log_fusion.alloc_objects.alloc_space.inuse_objects.inuse_space.012.pb.gz"

Thank you for your help!

pprof.log_fusion.alloc_objects.alloc_space.inuse_objects.inuse_space.012.pb.gz

`package parquet

import (
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"sync/atomic"
"time"

"github.com/darren2013/log_fusion_core/pkg/common"

"github.com/apache/arrow-go/v18/arrow"
"github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
"github.com/apache/arrow-go/v18/parquet/compress"
"github.com/apache/arrow-go/v18/parquet/pqarrow"
"github.com/apache/arrow-go/v18/parquet/schema"
"github.com/darren2013/log_fusion_core/pkg/api"
"github.com/darren2013/log_fusion_core/pkg/model"

)

// 全局实例计数器，用于生成唯一的实例ID
var instanceCounter int64

// ParquetProcessor 实现日志数据到Parquet文件的处理
type ParquetProcessor struct {
Counter *api.Counter
TotalCounter *api.Counter
FailCounter *api.Counter

config     *ParquetConfig
instanceID int64 // 实例唯一ID
batchSize  int
dictionary *model.Dictionary // 字典对象，包含字段定义和类型
currentDay string            // 当前日期，格式 YYYYMMDD
dataDir    string            // 数据存储基础目录
filePath   string            // 当前正在写入的文件路径

// Arrow相关
mem           memory.Allocator     // Arrow内存分配器
arrowSchema   *arrow.Schema        // Arrow Schema
recordBuilder *array.RecordBuilder // Arrow记录构建器

// Parquet相关
arrowWriter   *pqarrow.FileWriter       // Arrow Parquet文件写入器
currentFile   *os.File                  // 当前打开的文件
parquetSchema *schema.GroupNode         // Parquet Schema (保持向后兼容)
writerProps   *parquet.WriterProperties // 写入器属性

// 注意：移除了内存监控相关字段，因为内存泄漏的根本原因
// 在于 Parquet 库内部的 C++ 内存管理

}

// ParquetConfig 保存Parquet处理器的配置信息
type ParquetConfig struct {
DictionaryPath string json:"dictionary_path" // 字典文件路径
DataDir string json:"data_dir" // 数据存储基础目录
BatchSize int json:"batch_size" // 批处理缓冲区大小
Compression string json:"compression" // 压缩算法, 可用选项: SNAPPY, GZIP, ZSTD
}

// NewParquetProcessor 创建一个新的Parquet处理器
func NewParquetProcessor(counter, totalCounter, failCounter *api.Counter) (*ParquetProcessor, error) {
// 生成唯一的实例ID
instanceID := atomic.AddInt64(&instanceCounter, 1)

// 创建处理器实例
processor := &ParquetProcessor{
	Counter:      counter,
	TotalCounter: totalCounter,
	FailCounter:  failCounter,
	config:       &ParquetConfig{},
	instanceID:   instanceID,
	batchSize:    1000,                    // 默认值，会被UpdateConfig覆盖
	dataDir:      "./data",                // 默认值，会被UpdateConfig覆盖
	mem:          memory.NewGoAllocator(), // 初始化Arrow内存分配器
}

log.Printf("创建ParquetProcessor实例，ID: %d", instanceID)
return processor, nil

}

// Process 处理单个事件
func (p *ParquetProcessor) Process(entity model.Event) {
// 直接调用 BatchProcess 方法处理单个事件
p.BatchProcess([]model.Event{entity})
}

// BatchProcess 批量处理事件
func (p *ParquetProcessor) BatchProcess(entities []model.Event) {
// 注意：移除了内存监控，因为内存泄漏的根本原因
// 在于 Parquet 库内部的 C++ 内存管理

// 检查是否需要旋转到新的日期
day := time.Now().Format("20060102")

// 检查是否需要切换到新的一天
if p.currentDay != day {
	if err := p.rotateToNewDay(day); err != nil {
		log.Printf("ParquetProcessor[%d] Error rotating to new day: %v", p.instanceID, err)
		(*p.FailCounter).Count()
		return
	}
}

// 如果arrowWriter为nil，需要重新初始化
if p.arrowWriter == nil {
	if err := p.initializeParquetFile(); err != nil {
		log.Printf("ParquetProcessor[%d] Error initializing Parquet file: %v", p.instanceID, err)
		(*p.FailCounter).Count()
		return
	}
}

// 实现真正的批处理逻辑
if len(entities) == 0 {
	return
}

// 一次性写入所有数据
if err := p.flushBatch(entities); err != nil {
	log.Printf("ParquetProcessor[%d] Error flushing batch: %v", p.instanceID, err)
	(*p.FailCounter).Count()
	return
}

// 更新计数器
(*p.Counter).CountN(int64(len(entities)))

}

// GetName 返回处理器名称
func (p *ParquetProcessor) GetName() string {
return "parquet-processor"
}

// Start 启动处理器
func (p *ParquetProcessor) Start() error {
return nil
}

// rotateToNewDay 旋转到新的一天，包括刷新批次、关闭当前文件、初始化新文件
func (p *ParquetProcessor) rotateToNewDay(day string) error {
// 在旋转之前不需要刷新，因为没有批处理缓冲区

// 关闭当前的Arrow Parquet文件
if p.arrowWriter != nil {
	if err := p.arrowWriter.Close(); err != nil {
		log.Printf("ParquetProcessor[%d] Error closing Arrow Parquet writer: %v", p.instanceID, err)
	}
	p.arrowWriter = nil
}

if p.currentFile != nil {
	if err := p.currentFile.Close(); err != nil {
		log.Printf("ParquetProcessor[%d] Error closing file: %v", p.instanceID, err)
	}
	p.currentFile = nil
}

// 使用新的日期初始化
p.currentDay = day
if err := p.initializeParquetFile(); err != nil {
	log.Printf("ParquetProcessor[%d] Error initializing Parquet file for new day: %v", p.instanceID, err)
	(*p.FailCounter).Count()
	return err
}

return nil

}

// initializeParquetFile 初始化当前日期的Parquet文件
func (p *ParquetProcessor) initializeParquetFile() error {
p.currentDay = time.Now().Format("20060102")

// 为当前日期创建目录
dayDir := filepath.Join(p.dataDir, p.currentDay)
if err := os.MkdirAll(dayDir, 0755); err != nil {
	return fmt.Errorf("failed to create directory: %w", err)
}

// 创建具有时间戳和实例ID的文件名，确保唯一性
timestamp := time.Now().Format("150405")
fileName := fmt.Sprintf("logs_%s_inst%d.parquet", timestamp, p.instanceID)
filePath := filepath.Join(dayDir, fileName)

// 记录文件路径
p.filePath = filePath

// 创建Arrow和Parquet schema
if p.arrowSchema == nil {
	// 构建Arrow schema
	p.arrowSchema = buildArrowSchema(p.dictionary)
	// 构建Parquet schema（保持向后兼容）
	p.parquetSchema = buildParquetSchema(p.dictionary)
}

// 创建用于写入的文件
f, err := os.Create(filePath)
if err != nil {
	return fmt.Errorf("failed to create file: %w", err)
}
p.currentFile = f

// 设置Writer属性
writerProps := parquet.NewWriterProperties(
	parquet.WithCompression(getCompressionType(p.config.Compression)),
	parquet.WithVersion(parquet.V2_LATEST),
)
p.writerProps = writerProps

// 创建Arrow Parquet文件写入器
arrowWriter, err := pqarrow.NewFileWriter(p.arrowSchema, f, writerProps, pqarrow.DefaultWriterProps())
if err != nil {
	return fmt.Errorf("failed to create arrow parquet writer: %w", err)
}
p.arrowWriter = arrowWriter

log.Printf("ParquetProcessor[%d] Created new Parquet file at %s with compression %s", p.instanceID, filePath, p.config.Compression)

return nil

}

// buildArrowSchema 构建Arrow Schema
func buildArrowSchema(dictionary *model.Dictionary) *arrow.Schema {
// 记录字段名和对应的字段类型，保持一致的顺序
fieldNames := []string{}
fieldTypes := make(map[string]string)

// 添加timestamp字段（始终是第一个字段）
fieldNames = append(fieldNames, "timestamp")
fieldTypes["timestamp"] = "timestamp"

// 获取字段列表并按照名称进行排序，确保顺序稳定
remainingFields := make([]string, 0, len(dictionary.FieldTypes))
for fieldName := range dictionary.FieldTypes {
	if fieldName != "timestamp" { // 跳过timestamp，已经添加过了
		remainingFields = append(remainingFields, fieldName)
	}
}

// 确保字段顺序稳定，这里可以按照字段名称排序，或者按照字典中的顺序添加
// 这里简单地按照ID排序
for _, item := range dictionary.Items {
	if item.WordEnName != "timestamp" && item.Enable {
		// 只有没有添加过的字段才添加（避免重复）
		isAdded := false
		for _, name := range fieldNames {
			if name == item.WordEnName {
				isAdded = true
				break
			}
		}

		if !isAdded {
			fieldNames = append(fieldNames, item.WordEnName)
			fieldTypes[item.WordEnName] = item.WordType
		}
	}
}

// 记录所有字段和对应的类型方便调试
log.Printf("构建 Arrow Schema，字段数量: %d", len(fieldNames))

// 按照记录的顺序创建字段
fields := make([]arrow.Field, 0, len(fieldNames))

// 遍历字段名列表，按顺序创建字段
for i, fieldName := range fieldNames {
	fieldType := fieldTypes[fieldName]
	var arrowType arrow.DataType

	// 映射字段类型到Arrow类型
	switch fieldType {
	case "string":
		arrowType = arrow.BinaryTypes.String
	case "int", "integer":
		arrowType = arrow.PrimitiveTypes.Int32
	case "float", "double":
		arrowType = arrow.PrimitiveTypes.Float64
	case "bool", "boolean":
		arrowType = arrow.FixedWidthTypes.Boolean
	case "timestamp":
		arrowType = arrow.PrimitiveTypes.Int64
	default:
		// 对于未知类型，默认为字符串
		arrowType = arrow.BinaryTypes.String
	}

	// 创建Arrow字段
	field := arrow.Field{
		Name:     fieldName,
		Type:     arrowType,
		Nullable: fieldName != "timestamp" || i != 0, // timestamp字段作为第一个字段时不可为空
	}
	fields = append(fields, field)
}

// 创建Arrow Schema
schema := arrow.NewSchema(fields, nil)

log.Printf("Arrow Schema 构建完成, 总字段数: %d", len(fields))

return schema

}

// buildParquetSchema 构建Parquet Schema（保持向后兼容）
func buildParquetSchema(dictionary *model.Dictionary) *schema.GroupNode {
// 使用Arrow Schema构建Parquet Schema
arrowSchema := buildArrowSchema(dictionary)

// 将Arrow Schema转换为Parquet Schema
fields := make(schema.FieldList, 0, len(arrowSchema.Fields()))

for _, field := range arrowSchema.Fields() {
	var parquetField schema.Node

	// 映射Arrow类型到Parquet类型
	switch field.Type.ID() {
	case arrow.STRING:
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewByteArrayNode(field.Name, repetition, -1)
	case arrow.INT32:
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewInt32Node(field.Name, repetition, -1)
	case arrow.INT64:
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewInt64Node(field.Name, repetition, -1)
	case arrow.FLOAT64:
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewFloat64Node(field.Name, repetition, -1)
	case arrow.BOOL:
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewBooleanNode(field.Name, repetition, -1)
	default:
		// 对于未知类型，默认为字符串
		repetition := parquet.Repetitions.Optional
		if !field.Nullable {
			repetition = parquet.Repetitions.Required
		}
		parquetField = schema.NewByteArrayNode(field.Name, repetition, -1)
	}

	fields = append(fields, parquetField)
}

// 创建根节点
node, err := schema.NewGroupNode("schema", parquet.Repetitions.Required, fields, -1)
if err != nil {
	log.Fatalf("构建 Parquet Schema 失败: %v", err)
}

log.Printf("Parquet Schema 构建完成, 总字段数: %d", len(fields))

return node

}

// 注意：移除了 checkMemoryUsage 方法
// 因为内存泄漏的根本原因在于 Parquet 库内部的 C++ 内存管理
// 所以内存监控不是有效的解决方案

// flushBatch 使用Arrow Go v18最新的方式将指定批次写入Parquet文件
func (p *ParquetProcessor) flushBatch(entities []model.Event) error {
if len(entities) == 0 {
return nil
}

// 检查Arrow Schema是否已初始化
if p.arrowSchema == nil {
	return fmt.Errorf("arrow schema not initialized")
}

// 检查Arrow写入器是否已初始化
if p.arrowWriter == nil {
	return fmt.Errorf("arrow writer not initialized")
}

// 当前时间
now := time.Now().UnixMilli()

// 创建Arrow Record构建器
if p.recordBuilder == nil {
	p.recordBuilder = array.NewRecordBuilder(p.mem, p.arrowSchema)
}

// 清理之前的构建器状态
p.recordBuilder.Release()
p.recordBuilder = array.NewRecordBuilder(p.mem, p.arrowSchema)

// 按字段填充数据
for i, field := range p.arrowSchema.Fields() {
	fieldName := field.Name

	// timestamp字段特殊处理
	if fieldName == "timestamp" {
		builder := p.recordBuilder.Field(i).(*array.Int64Builder)
		for j := 0; j < len(entities); j++ {
			builder.Append(now)
		}
		continue
	}

	// 根据字段类型填充数据
	switch field.Type.ID() {
	case arrow.STRING:
		builder := p.recordBuilder.Field(i).(*array.StringBuilder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var strVal string
				switch v := val.(type) {
				case string:
					strVal = v
				default:
					strVal = fmt.Sprintf("%v", v)
				}
				builder.Append(strVal)
			}
		}

	case arrow.INT32:
		builder := p.recordBuilder.Field(i).(*array.Int32Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var intVal int32
				switch v := val.(type) {
				case int:
					intVal = int32(v)
				case int64:
					intVal = int32(v)
				case float64:
					intVal = int32(v)
				case string:
					var f float64
					if _, err := fmt.Sscanf(v, "%f", &f); err == nil {
						intVal = int32(f)
					}
				}
				builder.Append(intVal)
			}
		}

	case arrow.INT64:
		builder := p.recordBuilder.Field(i).(*array.Int64Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var int64Val int64
				switch v := val.(type) {
				case int:
					int64Val = int64(v)
				case int64:
					int64Val = v
				case float64:
					int64Val = int64(v)
				case time.Time:
					int64Val = v.UnixMilli()
				case string:
					var f float64
					if _, err := fmt.Sscanf(v, "%f", &f); err == nil {
						int64Val = int64(f)
					} else {
						// 尝试解析时间字符串
						formats := []string{
							time.RFC3339,
							"2006-01-02 15:04:05",
							"2006/01/02 15:04:05",
							"01/02/2006 15:04:05",
							"02/01/2006 15:04:05",
						}
						for _, format := range formats {
							if t, err := time.Parse(format, v); err == nil {
								int64Val = t.UnixMilli()
								break
							}
						}
					}
				}
				builder.Append(int64Val)
			}
		}

	case arrow.FLOAT64:
		builder := p.recordBuilder.Field(i).(*array.Float64Builder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var floatVal float64
				switch v := val.(type) {
				case float64:
					floatVal = v
				case float32:
					floatVal = float64(v)
				case int:
					floatVal = float64(v)
				case int64:
					floatVal = float64(v)
				case string:
					fmt.Sscanf(v, "%f", &floatVal)
				}
				builder.Append(floatVal)
			}
		}

	case arrow.BOOL:
		builder := p.recordBuilder.Field(i).(*array.BooleanBuilder)
		for _, entity := range entities {
			val, ok := entity.Values[fieldName]
			if !ok || val == nil {
				builder.AppendNull()
			} else {
				var boolVal bool
				switch v := val.(type) {
				case bool:
					boolVal = v
				case string:
					switch v {
					case "true", "True", "TRUE", "1":
						boolVal = true
					case "false", "False", "FALSE", "0":
						boolVal = false
					}
				case int:
					boolVal = v != 0
				case float64:
					boolVal = v != 0
				}
				builder.Append(boolVal)
			}
		}

	default:
		log.Printf("警告: 字段 %s 的类型 %s 不支持", fieldName, field.Type.ID())
	}
}

// 构建Arrow RecordBatch
recordBatch := p.recordBuilder.NewRecordBatch()
defer recordBatch.Release()

// 使用Arrow Go v18最新的方式写入Parquet文件
if err := p.arrowWriter.Write(recordBatch); err != nil {
	return fmt.Errorf("failed to write record batch to parquet: %w", err)
}

// 调用TotalCounter.CountN方法，记录成功写入的记录数
(*p.TotalCounter).CountN(int64(len(entities)))

return nil

}

// getCompressionType 将压缩算法名称转换为对应的Parquet压缩类型
func getCompressionType(compression string) compress.Compression {
switch compression {
case "SNAPPY":
return compress.Codecs.Snappy
case "GZIP":
return compress.Codecs.Gzip
case "ZSTD":
return compress.Codecs.Zstd
default:
// 默认SNAPPY压缩
return compress.Codecs.Snappy
}
}

// Close 完成处理器的工作，刷新所有挂起的数据
func (p *ParquetProcessor) Close() error {
var closeErrors []error

// 释放Arrow Record构建器
if p.recordBuilder != nil {
	p.recordBuilder.Release()
	p.recordBuilder = nil
}

// 关闭Arrow Parquet写入器
if p.arrowWriter != nil {
	log.Printf("ParquetProcessor[%d] 正在关闭Arrow Parquet写入器...当前文件: %s", p.instanceID, p.filePath)
	if err := p.arrowWriter.Close(); err != nil {
		log.Printf("ParquetProcessor[%d] Error closing Arrow Parquet writer: %v", p.instanceID, err)
		closeErrors = append(closeErrors, fmt.Errorf("failed to close arrow parquet writer: %w", err))
	}
	p.arrowWriter = nil
}

// 关闭文件
if p.currentFile != nil {
	log.Printf("ParquetProcessor[%d] 正在关闭文件: %s", p.instanceID, p.filePath)
	if err := p.currentFile.Close(); err != nil {
		log.Printf("ParquetProcessor[%d] Error closing file: %v", p.instanceID, err)
		closeErrors = append(closeErrors, fmt.Errorf("failed to close file: %w", err))
	}
	p.currentFile = nil
}

// 释放Arrow Schema引用
if p.arrowSchema != nil {
	p.arrowSchema = nil
}

// 释放Parquet Schema引用
if p.parquetSchema != nil {
	p.parquetSchema = nil
}

// 释放Writer Properties引用
if p.writerProps != nil {
	p.writerProps = nil
}

if len(closeErrors) > 0 {
	log.Printf("ParquetProcessor[%d] 关闭时遇到 %d 个错误", p.instanceID, len(closeErrors))
	return fmt.Errorf("close errors: %v", closeErrors)
}

log.Printf("ParquetProcessor[%d] 已完全关闭，文件: %s", p.instanceID, p.filePath)
return nil

}

// UpdateConfig 更新处理器配置
func (p *ParquetProcessor) UpdateConfig(configData map[string]interface{}) error {
// 解析配置
configBytes, err := json.Marshal(configData)
if err != nil {
return fmt.Errorf("failed to marshal config: %w", err)
}

var newConfig ParquetConfig
if err := json.Unmarshal(configBytes, &newConfig); err != nil {
	return fmt.Errorf("failed to unmarshal config: %w", err)
}

// 不需要刷新，因为没有批处理缓冲区

// 仅在首次配置时需要初始化
needReinitialization := p.dictionary == nil

if needReinitialization {
	// 关闭现有的资源
	if p.arrowWriter != nil {
		if err := p.arrowWriter.Close(); err != nil {
			log.Printf("ParquetProcessor[%d] Error closing Arrow Parquet writer during config update: %v", p.instanceID, err)
		}
		p.arrowWriter = nil
	}

	if p.currentFile != nil {
		if err := p.currentFile.Close(); err != nil {
			log.Printf("ParquetProcessor[%d] Error closing file during config update: %v", p.instanceID, err)
		}
		p.currentFile = nil
	}

	// 更新数据目录
	if newConfig.DataDir != "" {
		p.dataDir = newConfig.DataDir
	}

	// 加载字典
	if p.dictionary == nil {

		dictionary, err := common.GetDic(newConfig.DictionaryPath)

		if err != nil {
			return fmt.Errorf("failed to load dictionary from %s: %w", newConfig.DictionaryPath, err)
		}
		p.dictionary = dictionary

		// 重新构建 Arrow 和 Parquet schema
		p.arrowSchema = buildArrowSchema(p.dictionary)
		p.parquetSchema = buildParquetSchema(p.dictionary)
	}
}

// 更新批处理大小（虽然不再使用批处理，但保留配置）
if newConfig.BatchSize > 0 {
	p.batchSize = newConfig.BatchSize
}

// 更新压缩算法
if newConfig.Compression != "" {
	p.config.Compression = newConfig.Compression
}

// 更新配置对象的其他字段
if newConfig.DataDir != "" {
	dirPath, pathErr := common.GetDataDirPath(p.dataDir)

	if pathErr != nil {
		fmt.Println("data pathErr,use default", pathErr)
	}
	p.config.DataDir = dirPath
}
if newConfig.DictionaryPath != "" {
	p.config.DictionaryPath = newConfig.DictionaryPath
}

// 如果需要重新初始化 Parquet 文件
if p.arrowWriter == nil || p.currentFile == nil {
	if err := p.initializeParquetFile(); err != nil {
		return fmt.Errorf("failed to initialize Parquet file: %w", err)
	}
}

log.Printf("ParquetProcessor[%d] configuration updated successfully", p.instanceID)
return nil

}

// ReadParquetFile 读取Parquet文件并返回事件数据
// 注意：此功能需要进一步实现，当前版本暂时注释掉以避免编译错误
/*
func (p *ParquetProcessor) ReadParquetFile(filePath string) ([]model.Event, error) {
// 实现读取功能需要正确的Arrow API使用
// 暂时返回空实现
return nil, fmt.Errorf("read functionality not implemented yet")
}
*/
`

[zeroshade]

So a couple things:

Looking at the memory you posted

3141.12MB 20.20% 20.20% 3141.12MB 20.20% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewColumnMetaData (inline)
2053.06MB 13.20% 33.40% 2053.06MB 13.20% github.com/apache/arrow-go/v18/parquet/metadata.(*ColumnChunkMetaDataBuilder).Finish
2022.75MB 13.01% 46.41% 2022.75MB 13.01% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewStatistics (inline)
1683.18MB 10.82% 57.23% 6348.41MB 40.82% github.com/apache/arrow-go/v18/parquet/file.(*columnWriter).Close
1630.68MB 10.49% 67.71% 5779.84MB 37.16% github.com/apache/arrow-go/v18/parquet/metadata.(*RowGroupMetaDataBuilder).NextColumnChunk
1389.17MB 8.93% 76.65% 14417.33MB 92.70% github.com/apache/arrow-go/v18/parquet/file.(*rowGroupWriter).NextColumn

Nearly all of this memory is coming from the Column metadata, with 909 columns that makes sense. In addition, when you call Write(record) it is going to create a new row group every time unless you use WriteBuffered, and every time it creates a row group it's going to build and construct a bunch of metadata for all 909 columns. Because Parquet metadata goes at the end of the file it's going to keep it in memory until you finish and write it out by closing the file. The RowGroupWriter has TotalCompressedBytes and TotalBytesWritten methods which you could use to track when to break up row groups potentially.

I tried every 500000 records,close the filewirter,and recreate a new filewrite,the memory will release significantly.In this way,will create many small files,and the important,I am not sure when the memory is full,may be 300000 or 1000000.Ideally, The memory is released,when the batch writing finished

If every record is ~1.7KB, then 500000 is ~830MB before compression and encoding. It also might make sense to disable dictionary encoding depending on the cardinality of your columns (use WithDictionaryDefault(false) when building the properties). Depending on your workflows, that might be a perfectly reasonable size for your parquet files, not a "small" file. The memory for the records will be released after the batches are written and flushed to the file (once the GC reclaims it), but you're still going to build up a large amount of metadata when you're writing that many row groups and columns.

Can I see the forceRowGroupFlush function? Also what batch size are you using?

[darren2013]

So a couple things:

Looking at the memory you posted

3141.12MB 20.20% 20.20% 3141.12MB 20.20% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewColumnMetaData (inline)
2053.06MB 13.20% 33.40% 2053.06MB 13.20% github.com/apache/arrow-go/v18/parquet/metadata.(*ColumnChunkMetaDataBuilder).Finish
2022.75MB 13.01% 46.41% 2022.75MB 13.01% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewStatistics (inline)
1683.18MB 10.82% 57.23% 6348.41MB 40.82% github.com/apache/arrow-go/v18/parquet/file.(*columnWriter).Close
1630.68MB 10.49% 67.71% 5779.84MB 37.16% github.com/apache/arrow-go/v18/parquet/metadata.(*RowGroupMetaDataBuilder).NextColumnChunk
1389.17MB 8.93% 76.65% 14417.33MB 92.70% github.com/apache/arrow-go/v18/parquet/file.(*rowGroupWriter).NextColumn

Nearly all of this memory is coming from the Column metadata, with 909 columns that makes sense. In addition, when you call Write(record) it is going to create a new row group every time unless you use WriteBuffered, and every time it creates a row group it's going to build and construct a bunch of metadata for all 909 columns. Because Parquet metadata goes at the end of the file it's going to keep it in memory until you finish and write it out by closing the file. The RowGroupWriter has TotalCompressedBytes and TotalBytesWritten methods which you could use to track when to break up row groups potentially.

I tried every 500000 records,close the filewirter,and recreate a new filewrite,the memory will release significantly.In this way,will create many small files,and the important,I am not sure when the memory is full,may be 300000 or 1000000.Ideally, The memory is released,when the batch writing finished

If every record is ~1.7KB, then 500000 is ~830MB before compression and encoding. It also might make sense to disable dictionary encoding depending on the cardinality of your columns (use WithDictionaryDefault(false) when building the properties). Depending on your workflows, that might be a perfectly reasonable size for your parquet files, not a "small" file. The memory for the records will be released after the batches are written and flushed to the file (once the GC reclaims it), but you're still going to build up a large amount of metadata when you're writing that many row groups and columns.

Can I see the forceRowGroupFlush function? Also what batch size are you using?
Thank you for your replying!

forceRowGroupFlush function as follows.The batch_size is configured 1500
`
func (p *ParquetProcessor) forceRowGroupFlush() error {
if p.arrowWriter == nil {
return fmt.Errorf("arrow writer is nil")
}

log.Printf("ParquetProcessor[%d] 开始深度内存释放，当前行组记录: %d", p.instanceID, p.currentRowGroupSize)

// 🔥 方案1: 尝试通过Close再Reopen来强制释放C++内存
// 这是最有效的方法，但需要考虑文件完整性

// 保存当前文件路径
currentFilePath := p.filePath

if err := p.arrowWriter.Close(); err != nil {
	log.Printf("ParquetProcessor[%d] Warning: error closing arrow writer: %v", p.instanceID, err)
}
p.arrowWriter = nil

// 关闭文件句柄
if p.currentFile != nil {
	if err := p.currentFile.Close(); err != nil {
		log.Printf("ParquetProcessor[%d] Warning: error closing file: %v", p.instanceID, err)
	}
	p.currentFile = nil
}

// 释放RecordBuilder
if p.recordBuilder != nil {
	p.recordBuilder.Release()
	p.recordBuilder = nil
}

// 🔥 方案2: 创建新的分段文件来避免内存累积
// 注意：Parquet格式不支持追加写入，所以我们创建新的分段文件
p.fileSequence++
timestamp := time.Now().Format("150405")
newFileName := fmt.Sprintf("logs_%s_inst%d_seg%d.parquet", timestamp, p.instanceID, p.fileSequence)
newFilePath := filepath.Join(filepath.Dir(currentFilePath), newFileName)

file, err := os.Create(newFilePath)
if err != nil {
	return fmt.Errorf("failed to create new segment file: %w", err)
}
p.currentFile = file
p.filePath = newFilePath
p.totalRecords = 0 // 新文件重置记录数

log.Printf("ParquetProcessor[%d] 创建新分段文件: %s -> %s", p.instanceID, currentFilePath, newFilePath)

// 重新创建Arrow Writer
arrowWriterProps := pqarrow.DefaultWriterProps()
arrowWriter, err := pqarrow.NewFileWriter(p.arrowSchema, p.currentFile, p.writerProps, arrowWriterProps)
if err != nil {
	// 如果创建失败，关闭文件并报错
	p.currentFile.Close()
	return fmt.Errorf("failed to recreate arrow writer: %w", err)
}
p.arrowWriter = arrowWriter

// 重新创建RecordBuilder
p.recordBuilder = array.NewRecordBuilder(p.mem, p.arrowSchema)

log.Printf("ParquetProcessor[%d] 深度内存释放完成，Writer已重新创建", p.instanceID)
return nil

}
`

I changed the method calling
- if err := p.arrowWriter.Write(recordBatch); err != nil { + if err := p.arrowWriter.WriteBuffered(recordBatch); err != nil { return fmt.Errorf("failed to write record batch to parquet: %w", err) }
and the max row group size setting
writerProps := parquet.NewWriterProperties( parquet.WithCompression(getCompressionType(p.config.Compression)), parquet.WithVersion(parquet.V2_LATEST), // 设置较小的行组大小以减少内存使用 parquet.WithMaxRowGroupLength(1000), )
The memory is still increasing,the pprof as follows. No matter Write(record) or WriteBuffered method,The column meta data will keep in memory unti the filewrite is closed,is it right?if this ,how to manage the memory correctly? If I close the file frequently,will result in many small files.
darren@darrendus-Mac-mini ~ % go tool pprof -text http://10.20.183.250:6060/debug/pprof/heap | head -30 Fetching profile over HTTP from http://10.20.183.250:6060/debug/pprof/heap Saved profile in /Users/darren/pprof/pprof.log_fusion.alloc_objects.alloc_space.inuse_objects.inuse_space.022.pb.gz File: log_fusion Build ID: bee848f2187c216cf4161642c602db04073fd81a Type: inuse_space Time: 2025-09-17 08:34:36 CST Showing nodes accounting for 2397.48MB, 96.82% of 2476.11MB total Dropped 193 nodes (cum <= 12.38MB) flat flat% sum% cum cum% 510.60MB 20.62% 20.62% 510.60MB 20.62% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewColumnMetaData (inline) 337.51MB 13.63% 34.25% 337.51MB 13.63% github.com/apache/arrow-go/v18/parquet/metadata.(*ColumnChunkMetaDataBuilder).Finish 329.54MB 13.31% 47.56% 329.54MB 13.31% github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet.NewStatistics (inline) 291.03MB 11.75% 59.31% 985.97MB 39.82% github.com/apache/arrow-go/v18/parquet/file.(*columnWriter).Close 251.58MB 10.16% 69.47% 1190.89MB 48.10% github.com/apache/arrow-go/v18/parquet/file.(*rowGroupWriter).initColumns 250.19MB 10.10% 79.58% 921.29MB 37.21% github.com/apache/arrow-go/v18/parquet/metadata.(*RowGroupMetaDataBuilder).NextColumnChunk 116.51MB 4.71% 84.28% 160.51MB 6.48% github.com/apache/arrow-go/v18/parquet/metadata.NewColumnChunkMetaDataBuilderWithContents (inline) 63.71MB 2.57% 86.86% 63.71MB 2.57% reflect.mapassign_faststr0 57.62MB 2.33% 89.18% 57.62MB 2.33% bufio.(*Scanner).Text (inline) 44.50MB 1.80% 90.98% 44.50MB 1.80% github.com/apache/arrow-go/v18/parquet/schema.ColumnPathFromNode 44.45MB 1.80% 92.78% 44.45MB 1.80% github.com/apache/arrow-go/v18/arrow/memory.(*GoAllocator).Allocate (partial-inline) 36.51MB 1.47% 94.25% 36.51MB 1.47% encoding/json.(*decodeState).literalStore 23.18MB 0.94% 95.19% 23.18MB 0.94% github.com/apache/arrow-go/v18/parquet/metadata.NewRowGroupMetaDataBuilder (inline) 21.56MB 0.87% 96.06% 30.94MB 1.25% github.com/apache/arrow-go/v18/parquet/pqarrow.writeDenseArrow 12.50MB 0.5% 96.56% 114.71MB 4.63% encoding/json.(*decodeState).object 3.50MB 0.14% 96.70% 44.94MB 1.82% github.com/apache/arrow-go/v18/parquet/pqarrow.(*arrowColumnWriter).Write 2.50MB 0.1% 96.80% 116.72MB 4.71% github.com/darren2013/log_fusion_source/internal/source.FileConsumer 0.50MB 0.02% 96.82% 13.52MB 0.55% github.com/apache/arrow-go/v18/parquet/file.NewByteArrayColumnChunkWriter 0 0% 96.82% 114.71MB 4.63% encoding/json.(*decodeState).unmarshal 0 0% 96.82% 114.71MB 4.63% encoding/json.(*decodeState).value 0 0% 96.82% 114.71MB 4.63% encoding/json.Unmarshal 0 0% 96.82% 19.01MB 0.77% github.com/apache/arrow-go/v18/arrow/array.(*BinaryBuilder).AppendNull 0 0% 96.82% 19.51MB 0.79% github.com/apache/arrow-go/v18/arrow/array.(*BinaryBuilder).Reserve

[tushartg]

code:

func getSmallArrowRecordFromScratch(mem memory.Allocator) arrow.Record {
	schema := arrow.NewSchema([]arrow.Field{
		{Name: "col0", Type: arrow.BinaryTypes.String, Nullable: false},
		{Name: "col1", Type: arrow.BinaryTypes.String, Nullable: false},
		{Name: "col2", Type: arrow.PrimitiveTypes.Int64, Nullable: false},
	}, nil)
	builder := array.NewRecordBuilder(mem, schema)
	defer builder.Release()
	builder.Field(0).(*array.StringBuilder).Append("hello")
	builder.Field(1).(*array.StringBuilder).Append("world")
	builder.Field(2).(*array.Int64Builder).Append(123)
	record := builder.NewRecord()
	return record
}

func TestArrowParquetMemoryLeak(t *testing.T) {
	mem := memory.NewCheckedAllocator(memory.DefaultAllocator)
	memory.DefaultAllocator = mem
	defer mem.AssertSize(t, 0)

	filePath := filepath.Join(os.TempDir(), "parquet-bench-*.parquet")
	file, err := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644)
	if err != nil {
		t.Fatalf("Error opening file: %v", err)
	}

	props := parquet.NewWriterProperties(parquet.WithAllocator(mem))
	arrowProps := pqarrow.NewArrowWriterProperties(pqarrow.WithAllocator(mem))

	record := getSmallArrowRecordFromScratch(mem)
	defer (*record).Release()

	writer, err := pqarrow.NewFileWriter((*record).Schema(), file, props, arrowProps)
	if err != nil {
		log.Logger().Errorf("Error creating parquet writer for file %s: %v", filePath, err)
		file.Close()
		t.Fatalf("Error creating parquet writer: %v", err)
	}
	writer.WriteBuffered(*record)
	writer.Close()
	file.Close()
	os.Remove(filePath)
}

If you run the above code it gives

=== RUN   TestArrowParquetMemoryLeak
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+130
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:164
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+b0
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:165
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+130
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:164
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.(*PooledBufferWriter).Reset+b3
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/types.go:212
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+130
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:164
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+b0
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:165
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 256 bytes FROM
                github.com/apache/arrow-go/v18/parquet/file.newColumnWriterBase+b0
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/file/column_writer.go:165
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/checked_allocator.go:186: LEAK of 2048 bytes FROM
                github.com/apache/arrow-go/v18/parquet/internal/encoding.newEncoderBase+7c
                        /home/repo/vendor/github.com/apache/arrow-go/v18/parquet/internal/encoding/encoder.go:79
    /home/repo/pkg/util/parquet_helper_test.go:65: invalid memory size exp=0, got=20224
--- FAIL: TestArrowParquetMemoryLeak (0.00s)
FAIL
FAIL    /home/repo/pkg/util      0.436s

if you comment out this code

memory.DefaultAllocator = mem

then it shows no memory Leak. So, I believe somewhere library is using the default allocator. If you increase the records written then the memory leaked increases linearly.

@zeroshade - am I doing something wrong?

[zeroshade]

So, I believe somewhere library is using the default allocator.

The default internal buffer pool uses the default allocator as there isn't currently a good way to pass an allocator down to it. If the memory use is increasing linearly, then there's likely buffers that aren't being released back to the pool.

The leak is likely because the buffer pool won't call Release on the buffers in the pool. We might need to create a better pool of some kind. I'll take a look into it.

That said, my earlier statements are still correct in this particular case. According to pprof the bulk of the memory usage and in-use objects are the column metadata objects.

If I close the file frequently,will result in many small files.

How small are you talking about? As I said earlier, based on what you said the estimated record sizes are, if you cut off the files after 1000000 rows, that should be a bit over 1.5GB before compression and encoding. That's definitely not "many small files" if you have a bunch of 1GB files hanging around. So, when you say "will result in many small files", how small are the files you're talking about if you close the file more frequently?