laskoviymishka commented on code in PR #1229:
URL: https://github.com/apache/iceberg-go/pull/1229#discussion_r3448979477
##########
table/arrow_scanner.go:
##########
@@ -251,48 +252,127 @@ func sameDVBlob(a, b iceberg.DataFile) bool {
return *a.ContentOffset() == *b.ContentOffset()
}
-func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(_ map[string]*arrow.Chunked, err error) {
- src, err := internal.GetFile(ctx, fs, dataFile, true)
- if err != nil {
- return nil, err
+type releasableScalar interface {
+ scalar.Scalar
+ scalar.Releasable
+}
+
+func filePathScalar(v string, dt arrow.DataType) releasableScalar {
+ if dictType, ok := dt.(*arrow.DictionaryType); ok {
+ dt = dictType.ValueType
}
- rdr, err := src.GetReader(ctx)
- if err != nil {
- return nil, err
+ if dt.ID() == arrow.LARGE_STRING {
+ return scalar.NewLargeStringScalar(v)
}
- defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
- if err != nil {
- return nil, err
+ return scalar.NewStringScalar(v)
+}
+
+func appendUniqueFilePaths(paths []string, values arrow.Array) ([]string,
error) {
+ switch arr := values.(type) {
+ case array.StringLike:
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, arr.Value(i))
+ }
+ case *array.Dictionary:
+ wrapped, err := array.NewDictWrapper[string](arr)
+ if err != nil {
+ return nil, fmt.Errorf("%w: file_path column is not
string: %w",
+ iceberg.ErrInvalidSchema, err)
+ }
+
+ dict := arr.Dictionary()
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ if dict.IsNull(arr.GetValueIndex(i)) {
+ return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, wrapped.Value(i))
+ }
+ default:
+ return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
+ iceberg.ErrInvalidSchema, values.DataType())
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ return paths, nil
+}
+
+func distinctPosDeleteFilePaths(ctx context.Context, filePathCol
*arrow.Chunked) ([]string, error) {
+ if filePathCol.NullN() > 0 {
+ return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+ }
+
+ unique, err := compute.Unique(compute.WithAllocator(ctx,
memory.DefaultAllocator),
+ compute.NewDatumWithoutOwning(filePathCol))
if err != nil {
return nil, err
}
- defer tbl.Release()
+ defer unique.Release()
+
+ var chunks []arrow.Array
+ var releaseChunks func()
+ switch result := unique.(type) {
+ case *compute.ArrayDatum:
+ arr := result.MakeArray()
+ chunks = []arrow.Array{arr}
+ releaseChunks = arr.Release
+ case *compute.ChunkedDatum:
Review Comment:
I think this branch is unreachable — `compute.Unique` registers with
`OutputChunked = false`, so `WrapResults` hands back the single datum
un-wrapped and the result is always `*compute.ArrayDatum`.
The asymmetry bugs me a little too: the array branch sets `releaseChunks =
arr.Release` while this one is a no-op. I'd collapse to an
`*compute.ArrayDatum` assert plus the `default` error case and drop this. wdyt?
##########
table/arrow_scanner_posdelete_regression_test.go:
##########
@@ -21,12 +21,139 @@ import (
"testing"
"github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+func TestGroupPosDeletesByFilePathSupportsStringLayouts(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ filePathCol func(memory.Allocator) (*arrow.Chunked, func())
+ }{
+ {
+ name: "string chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ chunkA := stringArray(mem, "file-a.parquet",
"file-b.parquet")
+ chunkB := stringArray(mem, "file-a.parquet",
"file-c.parquet")
+ chunked :=
arrow.NewChunked(arrow.BinaryTypes.String, []arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ }
+ },
+ },
+ {
+ name: "large string chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ chunkA := largeStringArray(mem,
"file-a.parquet", "file-b.parquet")
+ chunkB := largeStringArray(mem,
"file-a.parquet", "file-c.parquet")
+ chunked :=
arrow.NewChunked(arrow.BinaryTypes.LargeString, []arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ }
+ },
+ },
+ {
+ name: "dictionary chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ dict := stringArray(mem, "file-a.parquet",
"file-b.parquet", "file-c.parquet")
+ idxA := int32Array(mem, 0, 1)
+ idxB := int32Array(mem, 0, 2)
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ chunkA := array.NewDictionaryArray(dictType,
idxA, dict)
+ chunkB := array.NewDictionaryArray(dictType,
idxB, dict)
+ chunked := arrow.NewChunked(dictType,
[]arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ dict.Release()
+ idxA.Release()
+ idxB.Release()
+ }
+ },
+ },
+ {
+ name: "large string dictionary chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ dict := largeStringArray(mem, "file-a.parquet",
"file-b.parquet", "file-c.parquet")
+ idxA := int32Array(mem, 0, 1)
+ idxB := int32Array(mem, 0, 2)
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType:
arrow.BinaryTypes.LargeString,
+ }
+ chunkA := array.NewDictionaryArray(dictType,
idxA, dict)
+ chunkB := array.NewDictionaryArray(dictType,
idxB, dict)
+ chunked := arrow.NewChunked(dictType,
[]arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ dict.Release()
+ idxA.Release()
+ idxB.Release()
+ }
+ },
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem := memory.DefaultAllocator
+ ctx := t.Context()
+
+ filePathCol, releaseFilePathCol := tc.filePathCol(mem)
+ posA := int64Array(mem, 1, 2)
+ posB := int64Array(mem, 3, 4)
+ posCol := arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{posA, posB})
+
+ got, err := groupPosDeletesByFilePath(ctx, filePathCol,
posCol)
+ require.NoError(t, err)
+
+ assert.Equal(t, []int64{1, 3},
int64Values(got["file-a.parquet"]))
+ assert.Equal(t, []int64{2},
int64Values(got["file-b.parquet"]))
+ assert.Equal(t, []int64{4},
int64Values(got["file-c.parquet"]))
+
+ releasePosDeletes(got)
+ posCol.Release()
+ posA.Release()
+ posB.Release()
+ releaseFilePathCol()
+ })
+ }
+}
+
+func TestGroupPosDeletesByFilePathRejectsUnsupportedFilePathLayout(t
*testing.T) {
+ mem := memory.DefaultAllocator
Review Comment:
Same here — checked allocator + `defer mem.AssertSize(t, 0)`. The error path
allocates and rejects, so it's exactly where you'd want to confirm nothing's
left dangling.
##########
table/arrow_scanner.go:
##########
@@ -251,48 +252,127 @@ func sameDVBlob(a, b iceberg.DataFile) bool {
return *a.ContentOffset() == *b.ContentOffset()
}
-func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(_ map[string]*arrow.Chunked, err error) {
- src, err := internal.GetFile(ctx, fs, dataFile, true)
- if err != nil {
- return nil, err
+type releasableScalar interface {
+ scalar.Scalar
+ scalar.Releasable
+}
+
+func filePathScalar(v string, dt arrow.DataType) releasableScalar {
+ if dictType, ok := dt.(*arrow.DictionaryType); ok {
+ dt = dictType.ValueType
}
- rdr, err := src.GetReader(ctx)
- if err != nil {
- return nil, err
+ if dt.ID() == arrow.LARGE_STRING {
+ return scalar.NewLargeStringScalar(v)
}
- defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
- if err != nil {
- return nil, err
+ return scalar.NewStringScalar(v)
+}
+
+func appendUniqueFilePaths(paths []string, values arrow.Array) ([]string,
error) {
+ switch arr := values.(type) {
+ case array.StringLike:
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, arr.Value(i))
+ }
+ case *array.Dictionary:
+ wrapped, err := array.NewDictWrapper[string](arr)
+ if err != nil {
+ return nil, fmt.Errorf("%w: file_path column is not
string: %w",
+ iceberg.ErrInvalidSchema, err)
+ }
+
+ dict := arr.Dictionary()
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ if dict.IsNull(arr.GetValueIndex(i)) {
+ return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, wrapped.Value(i))
+ }
+ default:
+ return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
+ iceberg.ErrInvalidSchema, values.DataType())
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ return paths, nil
+}
+
+func distinctPosDeleteFilePaths(ctx context.Context, filePathCol
*arrow.Chunked) ([]string, error) {
+ if filePathCol.NullN() > 0 {
+ return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+ }
+
+ unique, err := compute.Unique(compute.WithAllocator(ctx,
memory.DefaultAllocator),
Review Comment:
I'd pass `ctx` straight through here rather than forcing
`memory.DefaultAllocator`.
There's no leak today — `unique` is released via `defer unique.Release()`
and the array via `releaseChunks` — but hardcoding the allocator means
`Unique`'s allocations bypass whatever the caller wired in: a
`CheckedAllocator` in tests, or a bounded/tracked allocator in production. It's
defensive code that quietly defeats the thing it looks like it's protecting.
This pairs with the two regression tests below — once they use a checked
allocator, this override has to go or the `AssertSize` can't see `Unique`'s
allocations. wdyt?
##########
table/row_delta_test.go:
##########
@@ -490,10 +508,13 @@ func TestRowDeltaIntegrationPosDeleteRoundTrip(t
*testing.T) {
// (0-indexed: positions 1 and 3 → "beta" and "delta")
posDelArrowSc := table.PositionalDeleteArrowSchema
posDelPath := location + "/data/pos-del-001.parquet"
- writeParquetFile(t, posDelPath, posDelArrowSc, `[
+ writeParquetFileWithProperties(t, posDelPath, posDelArrowSc, `[
{"file_path": "`+dataPath+`", "pos": 1},
{"file_path": "`+dataPath+`", "pos": 3}
- ]`)
+ ]`, 1, parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithDictionaryDefault(false),
Review Comment:
Nice touch forcing plain-string encoding with `WithDictionaryDefault(false)`
— that's exactly the layout the old code panicked on.
One gap: the assertion downstream is `assertRowCount(t, tbl, 3)`, so if the
filter deleted the wrong two rows the test would still pass. Since the
survivors should be "alpha", "gamma", "epsilon", I'd assert the actual values,
not just the count — that's what proves we dropped positions 1 and 3 and not
some others.
##########
table/arrow_scanner_posdelete_regression_test.go:
##########
@@ -21,12 +21,139 @@ import (
"testing"
"github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
"github.com/apache/arrow-go/v18/arrow/compute"
"github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/iceberg-go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
+func TestGroupPosDeletesByFilePathSupportsStringLayouts(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ filePathCol func(memory.Allocator) (*arrow.Chunked, func())
+ }{
+ {
+ name: "string chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ chunkA := stringArray(mem, "file-a.parquet",
"file-b.parquet")
+ chunkB := stringArray(mem, "file-a.parquet",
"file-c.parquet")
+ chunked :=
arrow.NewChunked(arrow.BinaryTypes.String, []arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ }
+ },
+ },
+ {
+ name: "large string chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ chunkA := largeStringArray(mem,
"file-a.parquet", "file-b.parquet")
+ chunkB := largeStringArray(mem,
"file-a.parquet", "file-c.parquet")
+ chunked :=
arrow.NewChunked(arrow.BinaryTypes.LargeString, []arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ }
+ },
+ },
+ {
+ name: "dictionary chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ dict := stringArray(mem, "file-a.parquet",
"file-b.parquet", "file-c.parquet")
+ idxA := int32Array(mem, 0, 1)
+ idxB := int32Array(mem, 0, 2)
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType: arrow.BinaryTypes.String,
+ }
+ chunkA := array.NewDictionaryArray(dictType,
idxA, dict)
+ chunkB := array.NewDictionaryArray(dictType,
idxB, dict)
+ chunked := arrow.NewChunked(dictType,
[]arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ dict.Release()
+ idxA.Release()
+ idxB.Release()
+ }
+ },
+ },
+ {
+ name: "large string dictionary chunks",
+ filePathCol: func(mem memory.Allocator)
(*arrow.Chunked, func()) {
+ dict := largeStringArray(mem, "file-a.parquet",
"file-b.parquet", "file-c.parquet")
+ idxA := int32Array(mem, 0, 1)
+ idxB := int32Array(mem, 0, 2)
+ dictType := &arrow.DictionaryType{
+ IndexType: arrow.PrimitiveTypes.Int32,
+ ValueType:
arrow.BinaryTypes.LargeString,
+ }
+ chunkA := array.NewDictionaryArray(dictType,
idxA, dict)
+ chunkB := array.NewDictionaryArray(dictType,
idxB, dict)
+ chunked := arrow.NewChunked(dictType,
[]arrow.Array{chunkA, chunkB})
+
+ return chunked, func() {
+ chunked.Release()
+ chunkA.Release()
+ chunkB.Release()
+ dict.Release()
+ idxA.Release()
+ idxB.Release()
+ }
+ },
+ },
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ mem := memory.DefaultAllocator
Review Comment:
This is a "does it crash" test rather than a "does it leak" test — with
`memory.DefaultAllocator` and no `AssertSize`, a leak anywhere in
`groupPosDeletesByFilePath` would pass silently.
`TestProcessPositionalDeletesAcrossBatches` just below in this file already
does it the way I'd want: `memory.NewCheckedAllocator(memory.DefaultAllocator)`
with `defer mem.AssertSize(t, 0)`. I'd mirror that here (needs the
allocator-override fix above so the checked allocator actually sees `Unique`'s
allocations).
##########
table/arrow_scanner.go:
##########
@@ -251,48 +252,127 @@ func sameDVBlob(a, b iceberg.DataFile) bool {
return *a.ContentOffset() == *b.ContentOffset()
}
-func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(_ map[string]*arrow.Chunked, err error) {
- src, err := internal.GetFile(ctx, fs, dataFile, true)
- if err != nil {
- return nil, err
+type releasableScalar interface {
+ scalar.Scalar
+ scalar.Releasable
+}
+
+func filePathScalar(v string, dt arrow.DataType) releasableScalar {
+ if dictType, ok := dt.(*arrow.DictionaryType); ok {
+ dt = dictType.ValueType
}
- rdr, err := src.GetReader(ctx)
- if err != nil {
- return nil, err
+ if dt.ID() == arrow.LARGE_STRING {
+ return scalar.NewLargeStringScalar(v)
}
- defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
- if err != nil {
- return nil, err
+ return scalar.NewStringScalar(v)
+}
+
+func appendUniqueFilePaths(paths []string, values arrow.Array) ([]string,
error) {
+ switch arr := values.(type) {
+ case array.StringLike:
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, arr.Value(i))
+ }
+ case *array.Dictionary:
+ wrapped, err := array.NewDictWrapper[string](arr)
+ if err != nil {
+ return nil, fmt.Errorf("%w: file_path column is not
string: %w",
+ iceberg.ErrInvalidSchema, err)
+ }
+
+ dict := arr.Dictionary()
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+ if dict.IsNull(arr.GetValueIndex(i)) {
+ return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, wrapped.Value(i))
+ }
+ default:
+ return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
+ iceberg.ErrInvalidSchema, values.DataType())
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ return paths, nil
+}
+
+func distinctPosDeleteFilePaths(ctx context.Context, filePathCol
*arrow.Chunked) ([]string, error) {
+ if filePathCol.NullN() > 0 {
+ return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+ }
+
+ unique, err := compute.Unique(compute.WithAllocator(ctx,
memory.DefaultAllocator),
+ compute.NewDatumWithoutOwning(filePathCol))
if err != nil {
return nil, err
}
- defer tbl.Release()
+ defer unique.Release()
+
+ var chunks []arrow.Array
+ var releaseChunks func()
+ switch result := unique.(type) {
+ case *compute.ArrayDatum:
+ arr := result.MakeArray()
+ chunks = []arrow.Array{arr}
+ releaseChunks = arr.Release
+ case *compute.ChunkedDatum:
+ chunks = result.Chunks()
+ releaseChunks = func() {}
+ default:
+ return nil, fmt.Errorf("%w: unique file_path result is %s",
+ iceberg.ErrInvalidSchema, unique.Kind())
+ }
+ defer releaseChunks()
- filePathCol :=
tbl.Column(tbl.Schema().FieldIndices("file_path")[0]).Data()
- posCol := tbl.Column(tbl.Schema().FieldIndices("pos")[0]).Data()
- dict :=
filePathCol.Chunk(0).(*array.Dictionary).Dictionary().(*array.String)
+ paths := make([]string, 0, int(unique.Len()))
+ for _, chunk := range chunks {
+ paths, err = appendUniqueFilePaths(paths, chunk)
+ if err != nil {
+ return nil, err
+ }
+ }
- results := make(map[string]*arrow.Chunked)
- for i := 0; i < dict.Len(); i++ {
- v := dict.Value(i)
+ return paths, nil
+}
+
+func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (map[string]*arrow.Chunked, error) {
+ paths, err := distinctPosDeleteFilePaths(ctx, filePathCol)
+ if err != nil {
+ return nil, err
+ }
+ results := make(map[string]*arrow.Chunked, len(paths))
+ for _, v := range paths {
+ sc := filePathScalar(v, filePathCol.DataType())
+ scDatum := compute.NewDatum(sc)
+ sc.Release()
mask, err := compute.CallFunction(ctx, "equal", nil,
- compute.NewDatumWithoutOwning(filePathCol),
compute.NewDatum(v))
+ compute.NewDatumWithoutOwning(filePathCol), scDatum)
+ scDatum.Release()
if err != nil {
+ releasePosDeletes(results)
+
return nil, err
}
defer mask.Release()
Review Comment:
`defer` here fires at function return, not per loop iteration, so we hold
one full boolean mask (one bit per row of `filePathCol`) live for every
distinct path at once. For a delete file referencing thousands of data files
that's a real spike.
I'd release it right after `Filter` instead, and on the error paths:
```go
filtered, err := compute.Filter(ctx, compute.NewDatumWithoutOwning(posCol),
mask, *compute.DefaultFilterOptions())
mask.Release()
if err != nil {
releasePosDeletes(results)
return nil, err
}
```
Pre-existing, but the refactor is the natural moment to fix it.
##########
table/row_delta_test.go:
##########
@@ -434,9 +447,14 @@ func writeParquetFile(t testing.TB, path string, sc
*arrow.Schema, jsonData stri
tbl := array.NewTableFromRecords(sc, []arrow.RecordBatch{rec})
defer tbl.Release()
- require.NoError(t, pqarrow.WriteTable(tbl, fw, rec.NumRows(),
- parquet.NewWriterProperties(parquet.WithStats(true)),
- pqarrow.DefaultWriterProps()))
+ if rowGroupSize <= 0 {
+ rowGroupSize = rec.NumRows()
+ }
+ if writerProps == nil {
+ writerProps =
parquet.NewWriterProperties(parquet.WithStats(true))
+ }
+
+ require.NoError(t, pqarrow.WriteTable(tbl, fw, rowGroupSize,
writerProps, pqarrow.DefaultWriterProps()))
Review Comment:
`fw` from `fs.Create(path)` just above is never closed — copied from
`writeParquetFile`, so pre-existing, but worth a `defer fw.Close()` while we're
in here. Tempdir so it's low-stakes, but it leaks FDs and has bitten Windows CI
before.
##########
table/arrow_scanner.go:
##########
@@ -251,48 +252,127 @@ func sameDVBlob(a, b iceberg.DataFile) bool {
return *a.ContentOffset() == *b.ContentOffset()
}
-func readDeletes(ctx context.Context, fs iceio.IO, dataFile iceberg.DataFile)
(_ map[string]*arrow.Chunked, err error) {
- src, err := internal.GetFile(ctx, fs, dataFile, true)
- if err != nil {
- return nil, err
+type releasableScalar interface {
+ scalar.Scalar
+ scalar.Releasable
+}
+
+func filePathScalar(v string, dt arrow.DataType) releasableScalar {
+ if dictType, ok := dt.(*arrow.DictionaryType); ok {
+ dt = dictType.ValueType
}
- rdr, err := src.GetReader(ctx)
- if err != nil {
- return nil, err
+ if dt.ID() == arrow.LARGE_STRING {
+ return scalar.NewLargeStringScalar(v)
}
- defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
- if err != nil {
- return nil, err
+ return scalar.NewStringScalar(v)
+}
+
+func appendUniqueFilePaths(paths []string, values arrow.Array) ([]string,
error) {
+ switch arr := values.(type) {
+ case array.StringLike:
+ for i := 0; i < arr.Len(); i++ {
+ if arr.IsNull(i) {
+ return nil, fmt.Errorf("%w: null file_path in
position delete file",
+ iceberg.ErrInvalidSchema)
+ }
+
+ paths = append(paths, arr.Value(i))
+ }
+ case *array.Dictionary:
Review Comment:
The `NullN()>0` guard, the per-row null checks, and this
`NewDictWrapper[string]` error branch all carry `ErrInvalidSchema` paths that
nothing exercises — they'd still pass if we deleted them entirely.
I'd add a case feeding a `file_path` chunk with a null and asserting
`ErrInvalidSchema`, and optionally a dictionary with a non-string value type to
hit the `NewDictWrapper` failure. The reject test already gives you the pattern
to extend.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
