This is an automated email from the ASF dual-hosted git repository. zeroshade pushed a commit to branch 18.3.x-maint in repository https://gitbox.apache.org/repos/asf/arrow-go.git
commit 438c0cccaa8a6b67421af8ed4e03af0ee18d64ec Author: Victor Perez <[email protected]> AuthorDate: Tue May 20 16:56:58 2025 +0200 fix: list_columns.parquet testing (#378) add list_columns.parquet testing. `rr.Next()` is returning true when we've already reached the end of the file. This leads to a panic later on - Test that proves list_columns.parquet is not working as expected. - Use known path for partquet_testing if var is not defined. This ensures test can be run as usual but still enables custom locations for parquet_test_data. It also guarantees failure if not run correctly. That is the goal of the PR no? (cherry picked from commit 60e36a4043c442a218ec21c986d2211a435ca15f) --- parquet/file/file_reader_test.go | 117 ++++++++++++++++++++++++++++++++++++-- parquet/pqarrow/column_readers.go | 2 +- 2 files changed, 113 insertions(+), 6 deletions(-) diff --git a/parquet/file/file_reader_test.go b/parquet/file/file_reader_test.go index a850fab..45ce067 100644 --- a/parquet/file/file_reader_test.go +++ b/parquet/file/file_reader_test.go @@ -21,6 +21,7 @@ import ( "context" "crypto/rand" "encoding/binary" + "encoding/csv" "fmt" "io" "os" @@ -396,7 +397,8 @@ func TestDeltaLengthByteArrayPackingWithNulls(t *testing.T) { func TestRleBooleanEncodingFileRead(t *testing.T) { dir := os.Getenv("PARQUET_TEST_DATA") if dir == "" { - t.Skip("no path supplied with PARQUET_TEST_DATA") + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") } assert.DirExists(t, dir) @@ -472,7 +474,8 @@ func (m *mockBadReader) ReadAt(p []byte, off int64) (n int, err error) { func TestBadReader(t *testing.T) { dir := os.Getenv("PARQUET_TEST_DATA") if dir == "" { - t.Skip("no path supplied with PARQUET_TEST_DATA") + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") } require.DirExists(t, dir) @@ -504,7 +507,8 @@ func TestBadReader(t *testing.T) { func TestByteStreamSplitEncodingFileRead(t *testing.T) { dir := os.Getenv("PARQUET_TEST_DATA") if dir == "" { - t.Skip("no path supplied with PARQUET_TEST_DATA") + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") } require.DirExists(t, dir) @@ -699,7 +703,8 @@ func TestDeltaBinaryPackedMultipleBatches(t *testing.T) { func TestLZ4RawFileRead(t *testing.T) { dir := os.Getenv("PARQUET_TEST_DATA") if dir == "" { - t.Skip("no path supplied with PARQUET_TEST_DATA") + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") } require.DirExists(t, dir) @@ -782,7 +787,8 @@ func TestLZ4RawFileRead(t *testing.T) { func TestLZ4RawLargerFileRead(t *testing.T) { dir := os.Getenv("PARQUET_TEST_DATA") if dir == "" { - t.Skip("no path supplied with PARQUET_TEST_DATA") + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") } require.DirExists(t, dir) @@ -820,3 +826,104 @@ func TestLZ4RawLargerFileRead(t *testing.T) { } require.Equal(t, expectedValsHead, vals[:len(expectedValsHead)]) } + +func TestDeltaByteArray(t *testing.T) { + dir := os.Getenv("PARQUET_TEST_DATA") + if dir == "" { + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") + } + require.DirExists(t, dir) + + expected, err := os.ReadFile(path.Join(dir, "delta_byte_array_expect.csv")) + require.NoError(t, err) + csvReader := csv.NewReader(bytes.NewReader(expected)) + + records, err := csvReader.ReadAll() + require.NoError(t, err) + + records = records[1:] // skip header + + props := parquet.NewReaderProperties(memory.DefaultAllocator) + fileReader, err := file.OpenParquetFile(path.Join(dir, "delta_byte_array.parquet"), + false, file.WithReadProps(props)) + require.NoError(t, err) + defer fileReader.Close() + + nrows := fileReader.MetaData().NumRows + assert.Equal(t, nrows, int64(len(records)), "expected %d rows, got %d", len(records), nrows) + + arrowReader, err := pqarrow.NewFileReader( + fileReader, + pqarrow.ArrowReadProperties{BatchSize: 1024}, + memory.DefaultAllocator, + ) + require.NoError(t, err) + + rr, err := arrowReader.GetRecordReader(context.Background(), nil, nil) + require.NoError(t, err) + defer rr.Release() + + for rr.Next() { + rec := rr.Record() + for i := range int(rec.NumCols()) { + vals := rec.Column(i) + for j := range vals.Len() { + if vals.IsNull(j) { + require.Equal(t, records[j][i], "") + continue + } + require.Equal(t, records[j][i], vals.ValueStr(j)) + } + } + } +} + +func TestListColumns(t *testing.T) { + dir := os.Getenv("PARQUET_TEST_DATA") + if dir == "" { + dir = "../../parquet-testing/data" + t.Log("PARQUET_TEST_DATA not set, using ../../parquet-testing/data") + } + require.DirExists(t, dir) + + records := [][]string{ + {"[1,2,3]", "[\"abc\",\"efg\",\"hij\"]"}, + {"[null,1]", ""}, + {"[4]", "[\"efg\",null,\"hij\",\"xyz\"]"}, + } + + props := parquet.NewReaderProperties(memory.DefaultAllocator) + fileReader, err := file.OpenParquetFile(path.Join(dir, "list_columns.parquet"), + false, file.WithReadProps(props)) + require.NoError(t, err) + defer fileReader.Close() + + nrows := fileReader.MetaData().NumRows + assert.Equal(t, nrows, int64(len(records)), "expected %d rows, got %d", len(records), nrows) + + arrowReader, err := pqarrow.NewFileReader( + fileReader, + pqarrow.ArrowReadProperties{BatchSize: 1024}, + memory.DefaultAllocator, + ) + require.NoError(t, err) + + rr, err := arrowReader.GetRecordReader(context.Background(), nil, nil) + require.NoError(t, err) + defer rr.Release() + + for rr.Next() { + rec := rr.Record() + for i := range int(rec.NumCols()) { + vals := rec.Column(i) + for j := range vals.Len() { + if vals.IsNull(j) { + require.Equal(t, records[j][i], "") + continue + } + require.Equal(t, records[j][i], vals.ValueStr(j)) + } + } + } +} diff --git a/parquet/pqarrow/column_readers.go b/parquet/pqarrow/column_readers.go index ab08a7b..3ea81a7 100644 --- a/parquet/pqarrow/column_readers.go +++ b/parquet/pqarrow/column_readers.go @@ -93,7 +93,7 @@ func (lr *leafReader) IsOrHasRepeatedChild() bool { return false } func (lr *leafReader) LoadBatch(nrecords int64) (err error) { lr.releaseOut() - lr.recordRdr.ResetValues() + lr.recordRdr.Reset() if err := lr.recordRdr.Reserve(nrecords); err != nil { return err
