同一文件上的多个 Arrow CSV 读取器返回 null

我正在尝试使用多个 Goroutine 读取同一个文件，其中每个 Goroutine 都被分配一个字节来开始读取，并指定要读取的行数lineLimit.

当文件适合内存时，我成功地通过设置csv.ChunkSize的选项chunkSize多变的。但是，当文件大于内存时，我需要减少csv.ChunkSize选项。我正在尝试这样的事情

package main

import (
    "io"
    "log"
    "os"
    "sync"

    "github.com/apache/arrow/go/v11/arrow"
    "github.com/apache/arrow/go/v11/arrow/csv"
)

// A reader to read lines from the file starting from the byteOffset. The number
// of lines is specified by linesLimit.
func produce(
    id int,
    ch chan<- arrow.Record,
    byteOffset int64,
    linesLimit int64,
    filename string,
    wg *sync.WaitGroup,
) {
    defer wg.Done()

    fd, _ := os.Open(filename)
    fd.Seek(byteOffset, io.SeekStart)

    var remainder int64 = linesLimit % 10
    limit := linesLimit - remainder
    chunkSize := limit / 10

    reader := csv.NewInferringReader(fd,
        csv.WithChunk(int(chunkSize)),
        csv.WithNullReader(true, ""),
        csv.WithComma(','),
        csv.WithHeader(true),
        csv.WithColumnTypes(map[string]arrow.DataType{
            "Start_Time":        arrow.FixedWidthTypes.Timestamp_ns,
            "End_Time":          arrow.FixedWidthTypes.Timestamp_ns,
            "Weather_Timestamp": arrow.FixedWidthTypes.Timestamp_ns,
        }))
    reader.Retain()
    defer reader.Release()

    var count int64
    for reader.Next() {
        rec := reader.Record()
        rec.Retain() // released at the other end of the channel
        ch <- rec
        count += rec.NumRows()
        if count == limit {
            if remainder != 0 {
                flush(id, ch, fd, remainder)
            }
            break
        } else if count > limit {
            log.Panicf("Reader %d read more than it should, expected=%d, read=%d", id, linesLimit, count)
        }
    }

    if reader.Err() != nil {
        log.Panicf("error: %s in line %d,%d", reader.Err().Error(), count, id)
    }
}

func flush(id int,
    ch chan<- arrow.Record,
    fd *os.File,
    limit int64,
) {
    reader := csv.NewInferringReader(fd,
        csv.WithChunk(int(limit)),
        csv.WithNullReader(true, ""),
        csv.WithComma(','),
        csv.WithHeader(false),
    )

    reader.Retain()
    defer reader.Release()

    record := reader.Record()
    record.Retain() // nil pointer dereference error here
    ch <- record
}

我尝试了先前代码的多个版本，包括：

1. 复制文件描述符
2. 复制文件描述符的偏移量，打开同一个文件 并寻求这种抵消。
3. 调用前关闭第一个阅读器flush或关闭第一个fd.

无论我如何更改代码，错误似乎都是相同的。请注意，任何调用flush的读者提出了一个错误。包括reader.Next, and reader.Err().

我使用 csv 阅读器是否错误？这是重复使用同一文件的问题吗？

编辑：我不知道这是否有帮助，但是在中打开一个新的 fdflush没有任何Seek避免错误（不知何故任何Seek导致出现原始错误）。但是，如果没有Seek（即删除Seek导致任何 Goroutine 根本无法读取文件的一部分）。

主要问题是，csv 阅读器使用bufio.Reader下面，它的默认缓冲区大小为 4096。这意味着reader.Next()将读取超出需要的字节，并缓存额外的字节。如果之后直接从文件中读取reader.Next()，您将错过缓存的字节。

下面的演示展示了这种行为：

package main

import (
    "bytes"
    "fmt"
    "io"
    "os"

    "github.com/apache/arrow/go/v11/arrow"
    "github.com/apache/arrow/go/v11/arrow/csv"
)

func main() {
    // Create a two-column csv file with this content (the second column has 1024 bytes):
    // 0,000000....
    // 1,111111....
    // 2,222222....
    // 3,333333....
    temp := createTempFile()

    schema := arrow.NewSchema(
        []arrow.Field{
            {Name: "i64", Type: arrow.PrimitiveTypes.Int64},
            {Name: "str", Type: arrow.BinaryTypes.String},
        },
        nil,
    )
    r := csv.NewReader(
        temp, schema,
        csv.WithComma(','),
        csv.WithChunk(3),
    )
    defer r.Release()

    r.Next()

    // To check what's left after the first chunk is read.
    // If the reader stop at the end of the chunk, the content left will be:
    // 3,333333....
    // But in fact, the content left is:
    // 33333333333
    buf, err := io.ReadAll(temp)
    if err != nil {
        panic(err)
    }

    fmt.Printf("%s\n", buf)
}

func createTempFile() *os.File {
    temp, err := os.CreateTemp("", "test*.csv")
    if err != nil {
        panic(err)
    }
    for i := 0; i < 4; i++ {
        fmt.Fprintf(temp, "%d,", i)
        if _, err := temp.Write(bytes.Repeat([]byte{byte('0' + i)}, 1024)); err != nil {
            panic(err)
        }
        if _, err := temp.Write([]byte("\n")); err != nil {
            panic(err)
        }
    }

    if _, err := temp.Seek(0, io.SeekStart); err != nil {
        panic(err)
    }

    return temp
}

看来第二个读取器的目的是防止它读取另一个 csv 数据块。如果您提前知道下一个 csv 数据块的偏移量，则可以将文件包装在io.SectionReader使其仅读取当前的 csv 数据块。当前的问题没有提供有关这部分的足够信息，也许我们应该把它留给另一个问题。

Notes:

1. fd, _ := os.Open(filename)：永远不要忽略错误。至少记录它们。
2. fd大多数时候表示文件描述符。不要将其用于类型变量*os.File，特别是当*os.File有一个方法Fd.