laskoviymishka commented on code in PR #1898:
URL: https://github.com/apache/iceberg-go/pull/1898#discussion_r3874730715
##########
table/arrow_scanner.go:
##########
@@ -361,106 +361,179 @@ func (c *posDeleteCursor) next() (int64, bool) {
return pos, true
}
-func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- if err := ctx.Err(); err != nil {
- return nil, err
+type posDeleteAccumulator struct {
+ mem memory.Allocator
+ builders map[string]*array.Int64Builder
+}
+
+func newPosDeleteAccumulator(ctx context.Context) *posDeleteAccumulator {
+ return &posDeleteAccumulator{
+ mem: compute.GetAllocator(ctx),
+ builders: make(map[string]*array.Int64Builder),
}
- if filePathCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+}
+
+func (a *posDeleteAccumulator) release() {
+ for _, builder := range a.builders {
+ builder.Release()
}
- if filePathValueType(filePathCol.DataType()).ID() == arrow.STRING_VIEW {
- return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
- iceberg.ErrInvalidSchema, filePathCol.DataType())
+ a.builders = nil
+}
+
+func (a *posDeleteAccumulator) finish() map[string]*arrow.Chunked {
Review Comment:
`finish()` and `release()` both set `a.builders = nil`, so if `finish()` is
ever called after `release()` it ranges a nil map and returns an empty result
with no error, and the accumulated deletes just vanish. Today's callers don't
do this (success calls `finish()`, the error defer calls `release()`), but
nothing on the type enforces it.
Since it's unexported, I'd rather a misuse fail loudly than return a silent
empty map. A `panic` in `finish()` when `builders == nil`, or at least a
contract comment, would cover it.
##########
table/arrow_scanner.go:
##########
@@ -361,106 +361,179 @@ func (c *posDeleteCursor) next() (int64, bool) {
return pos, true
}
-func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- if err := ctx.Err(); err != nil {
- return nil, err
+type posDeleteAccumulator struct {
+ mem memory.Allocator
+ builders map[string]*array.Int64Builder
+}
+
+func newPosDeleteAccumulator(ctx context.Context) *posDeleteAccumulator {
+ return &posDeleteAccumulator{
+ mem: compute.GetAllocator(ctx),
+ builders: make(map[string]*array.Int64Builder),
}
- if filePathCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+}
+
+func (a *posDeleteAccumulator) release() {
+ for _, builder := range a.builders {
+ builder.Release()
}
- if filePathValueType(filePathCol.DataType()).ID() == arrow.STRING_VIEW {
- return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
- iceberg.ErrInvalidSchema, filePathCol.DataType())
+ a.builders = nil
+}
+
+func (a *posDeleteAccumulator) finish() map[string]*arrow.Chunked {
+ results := make(map[string]*arrow.Chunked, len(a.builders))
+ for path, builder := range a.builders {
+ positions := builder.NewInt64Array()
+ builder.Release()
+
+ results[path] = arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{positions})
+ positions.Release()
+ }
+ a.builders = nil
+
+ return results
+}
+
+func validatePosDeleteColumns(filePathType arrow.DataType, filePathNulls int,
filePathLen int,
+ posType arrow.DataType, posNulls int, posLen int,
+) error {
+ if filePathNulls > 0 {
+ return fmt.Errorf("%w: null file_path in position delete file",
iceberg.ErrInvalidSchema)
+ }
+ if filePathValueType(filePathType).ID() == arrow.STRING_VIEW {
+ return fmt.Errorf("%w: unsupported file_path column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, filePathType)
}
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ if posNulls > 0 {
+ return fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
}
- if posCol.DataType().ID() != arrow.INT64 {
- return nil, fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
- iceberg.ErrInvalidSchema, posCol.DataType())
+ if posType.ID() != arrow.INT64 {
+ return fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, posType)
}
- if filePathCol.Len() != posCol.Len() {
- return nil, fmt.Errorf("%w: file_path and pos columns have
different lengths: %d and %d",
- iceberg.ErrInvalidSchema, filePathCol.Len(),
posCol.Len())
+ if filePathLen != posLen {
+ return fmt.Errorf("%w: file_path and pos columns have different
lengths: %d and %d",
+ iceberg.ErrInvalidSchema, filePathLen, posLen)
}
- mem := compute.GetAllocator(ctx)
- posCursor, err := newPosDeleteCursor(posCol)
+ return nil
+}
+
+func (a *posDeleteAccumulator) appendFilePathChunk(ctx context.Context,
filePathChunk arrow.Array,
+ posCursor *posDeleteCursor,
+) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ paths, err := filePathValues(filePathChunk)
if err != nil {
- return nil, err
+ return err
}
- builders := make(map[string]*array.Int64Builder)
- defer func() {
- if err != nil {
- for _, builder := range builders {
- builder.Release()
+ var dictionary arrow.Array
+ var indices *array.Dictionary
+ if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
+ dictionary = dict.Dictionary()
+ indices = dict
+ }
+
+ for i := range filePathChunk.Len() {
+ if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
Review Comment:
This `i & (positionalDeleteCancellationCheckInterval - 1) == 0` only works
while the interval is a power of two. It is (16384), but now that it lives in
the hot loop here I'd drop a one-line comment at the const so nobody swaps in a
non-power-of-2 and quietly breaks the cancellation cadence.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +590,42 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex}, nil)
Review Comment:
Worth a one-line comment that we intentionally don't `UnifyTableDicts` on
this path anymore. It's safe because `filePathValues` decodes each dictionary
batch down to string values and we key the map on the string, so cross-batch
dictionary independence doesn't matter. Without a note, I could see someone
re-adding the unify call thinking the streaming path needs it.
##########
table/arrow_scanner.go:
##########
@@ -361,106 +361,179 @@ func (c *posDeleteCursor) next() (int64, bool) {
return pos, true
}
-func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- if err := ctx.Err(); err != nil {
- return nil, err
+type posDeleteAccumulator struct {
+ mem memory.Allocator
+ builders map[string]*array.Int64Builder
+}
+
+func newPosDeleteAccumulator(ctx context.Context) *posDeleteAccumulator {
+ return &posDeleteAccumulator{
+ mem: compute.GetAllocator(ctx),
+ builders: make(map[string]*array.Int64Builder),
}
- if filePathCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+}
+
+func (a *posDeleteAccumulator) release() {
+ for _, builder := range a.builders {
+ builder.Release()
}
- if filePathValueType(filePathCol.DataType()).ID() == arrow.STRING_VIEW {
- return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
- iceberg.ErrInvalidSchema, filePathCol.DataType())
+ a.builders = nil
+}
+
+func (a *posDeleteAccumulator) finish() map[string]*arrow.Chunked {
+ results := make(map[string]*arrow.Chunked, len(a.builders))
+ for path, builder := range a.builders {
+ positions := builder.NewInt64Array()
+ builder.Release()
+
+ results[path] = arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{positions})
+ positions.Release()
+ }
+ a.builders = nil
+
+ return results
+}
+
+func validatePosDeleteColumns(filePathType arrow.DataType, filePathNulls int,
filePathLen int,
+ posType arrow.DataType, posNulls int, posLen int,
+) error {
+ if filePathNulls > 0 {
+ return fmt.Errorf("%w: null file_path in position delete file",
iceberg.ErrInvalidSchema)
+ }
+ if filePathValueType(filePathType).ID() == arrow.STRING_VIEW {
+ return fmt.Errorf("%w: unsupported file_path column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, filePathType)
}
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ if posNulls > 0 {
+ return fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
}
- if posCol.DataType().ID() != arrow.INT64 {
- return nil, fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
- iceberg.ErrInvalidSchema, posCol.DataType())
+ if posType.ID() != arrow.INT64 {
+ return fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, posType)
}
- if filePathCol.Len() != posCol.Len() {
- return nil, fmt.Errorf("%w: file_path and pos columns have
different lengths: %d and %d",
- iceberg.ErrInvalidSchema, filePathCol.Len(),
posCol.Len())
+ if filePathLen != posLen {
+ return fmt.Errorf("%w: file_path and pos columns have different
lengths: %d and %d",
+ iceberg.ErrInvalidSchema, filePathLen, posLen)
}
- mem := compute.GetAllocator(ctx)
- posCursor, err := newPosDeleteCursor(posCol)
+ return nil
+}
+
+func (a *posDeleteAccumulator) appendFilePathChunk(ctx context.Context,
filePathChunk arrow.Array,
+ posCursor *posDeleteCursor,
+) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ paths, err := filePathValues(filePathChunk)
if err != nil {
- return nil, err
+ return err
}
- builders := make(map[string]*array.Int64Builder)
- defer func() {
- if err != nil {
- for _, builder := range builders {
- builder.Release()
+ var dictionary arrow.Array
+ var indices *array.Dictionary
+ if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
+ dictionary = dict.Dictionary()
+ indices = dict
+ }
+
+ for i := range filePathChunk.Len() {
+ if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
}
}
- }()
- for _, filePathChunk := range filePathCol.Chunks() {
- if err := ctx.Err(); err != nil {
- return nil, err
+ pos, ok := posCursor.next()
+ if !ok {
+ return fmt.Errorf("%w: position delete columns ended
before file_path column",
+ iceberg.ErrInvalidSchema)
}
-
- paths, pathErr := filePathValues(filePathChunk)
- if pathErr != nil {
- return nil, pathErr
+ if pos < 0 {
+ return fmt.Errorf("%w: negative pos %d in position
delete file",
+ iceberg.ErrInvalidSchema, pos)
+ }
+ if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
+ return fmt.Errorf("%w: null file_path dictionary value
in position delete file",
+ iceberg.ErrInvalidSchema)
}
- var dictionary arrow.Array
- var indices *array.Dictionary
- if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
- dictionary = dict.Dictionary()
- indices = dict
+ path := paths.Value(i)
+ builder, ok := a.builders[path]
+ if !ok {
+ path = strings.Clone(path)
+ builder = array.NewInt64Builder(a.mem)
+ a.builders[path] = builder
}
+ builder.Append(pos)
+ }
- for i := range filePathChunk.Len() {
- if i&(positionalDeleteCancellationCheckInterval-1) == 0
{
- if err := ctx.Err(); err != nil {
- return nil, err
- }
- }
+ return nil
+}
- pos, ok := posCursor.next()
- if !ok {
- return nil, fmt.Errorf("%w: position delete
columns ended before file_path column",
- iceberg.ErrInvalidSchema)
- }
- if pos < 0 {
- return nil, fmt.Errorf("%w: negative pos %d in
position delete file",
- iceberg.ErrInvalidSchema, pos)
- }
- if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
- return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
- iceberg.ErrInvalidSchema)
- }
+func (a *posDeleteAccumulator) appendChunked(ctx context.Context, filePathCol,
posCol *arrow.Chunked) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if err := validatePosDeleteColumns(filePathCol.DataType(),
filePathCol.NullN(), filePathCol.Len(),
+ posCol.DataType(), posCol.NullN(), posCol.Len()); err != nil {
+ return err
+ }
- path := paths.Value(i)
- builder, ok := builders[path]
- if !ok {
- path = strings.Clone(path)
- builder = array.NewInt64Builder(mem)
- builders[path] = builder
- }
- builder.Append(pos)
+ posCursor, err := newPosDeleteCursor(posCol)
+ if err != nil {
+ return err
+ }
+
+ for _, filePathChunk := range filePathCol.Chunks() {
+ if err := a.appendFilePathChunk(ctx, filePathChunk,
&posCursor); err != nil {
+ return err
}
}
- if err := ctx.Err(); err != nil {
- return nil, err
+
+ return ctx.Err()
+}
+
+func (a *posDeleteAccumulator) appendRecord(ctx context.Context, record
arrow.RecordBatch) error {
+ if record.NumCols() != 2 {
+ return fmt.Errorf("%w: projected position delete record has %d
columns, expected 2",
+ iceberg.ErrInvalidSchema, record.NumCols())
}
- results = make(map[string]*arrow.Chunked, len(builders))
- for path, builder := range builders {
- positions := builder.NewInt64Array()
- builder.Release()
+ filePathCol := record.Column(0)
+ posCol := record.Column(1)
+ if err := validatePosDeleteColumns(filePathCol.DataType(),
filePathCol.NullN(), filePathCol.Len(),
Review Comment:
A couple of the checks in this path can't actually fire.
`validatePosDeleteColumns` compares `filePathLen != posLen`, but these are two
columns of the same `RecordBatch`, so their lengths are always equal (that
cross-column check only had teeth in the old whole-file path). And it rejects
`posType.ID() != arrow.INT64` right before the `posCol.(*array.Int64)`
assertion just below, which can't independently fail for real pqarrow output.
Not a correctness problem, just dead branches that read as if they're
guarding something. I'd either trim the length and INT64 checks from this call
or drop the assertion and lean on `validatePosDeleteColumns`, so the two paths
don't diverge on which one reports the error.
##########
table/arrow_scanner_posdelete_regression_test.go:
##########
@@ -97,6 +97,68 @@ func TestReadDeletesRejectsMissingFilePath(t *testing.T) {
assert.Contains(t, err.Error(), `exactly one "file_path" column, found
0`)
}
+func TestReadDeletesProjectsColumnsAndAccumulatesBatches(t *testing.T) {
Review Comment:
This is the only end-to-end test of the new streaming path, and it never
exercises dictionary-encoded `file_path`. That's the case that actually flows
through `appendFilePathChunk`'s dictionary branch and the
`dictionary.IsNull(...)` guard, and it's the primary memory win this PR is
after. All the existing dictionary coverage goes through
`groupPosDeletesByFilePath` into `appendChunked`, which `readDeletes` no longer
calls.
Could we add a case that writes a delete file with repeated `file_path`
values (so it comes back dictionary-encoded under `SetReadDict`) and runs it
through `readDeletes`, asserting the grouping plus `mem.AssertSize(t, 0)`? A
regression in the dict path wouldn't be caught today.
##########
table/arrow_scanner.go:
##########
@@ -517,29 +590,42 @@ func readDeletes(ctx context.Context, fs iceio.IO,
dataFile iceberg.DataFile) (_
}
defer iceinternal.CheckedClose(rdr, &err)
- tbl, err := rdr.ReadTable(ctx)
+ schema, err := rdr.Schema()
if err != nil {
return nil, err
}
- defer tbl.Release()
- tbl, err = array.UnifyTableDicts(compute.GetAllocator(ctx), tbl)
+ filePathIndex, posIndex, err := positionDeleteColumnIndices(schema)
if err != nil {
return nil, err
}
- defer tbl.Release()
- filePathIndex, posIndex, err :=
positionDeleteColumnIndices(tbl.Schema())
+ records, err := rdr.GetRecords(ctx, []int{filePathIndex, posIndex}, nil)
if err != nil {
return nil, err
}
- filePathCol := tbl.Column(filePathIndex).Data()
- posCol := tbl.Column(posIndex).Data()
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ defer records.Release()
+
+ acc := newPosDeleteAccumulator(ctx)
+ defer func() {
Review Comment:
This deferred `release()` only fires on error because it reads the named
return `err`, but every early return in here uses `err :=` in an inner scope.
It's correct per spec (the `return nil, err` assigns the named return before
defers run), but it's subtle, and a refactor to `err = ...` inside an `if`
could quietly change what gets released. This defer is the only thing
preventing an Arrow leak on the error path, so I'd add a short comment noting
the dependency.
##########
table/arrow_scanner.go:
##########
@@ -361,106 +361,179 @@ func (c *posDeleteCursor) next() (int64, bool) {
return pos, true
}
-func groupPosDeletesByFilePath(ctx context.Context, filePathCol, posCol
*arrow.Chunked) (results map[string]*arrow.Chunked, err error) {
- if err := ctx.Err(); err != nil {
- return nil, err
+type posDeleteAccumulator struct {
+ mem memory.Allocator
+ builders map[string]*array.Int64Builder
+}
+
+func newPosDeleteAccumulator(ctx context.Context) *posDeleteAccumulator {
+ return &posDeleteAccumulator{
+ mem: compute.GetAllocator(ctx),
+ builders: make(map[string]*array.Int64Builder),
}
- if filePathCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null file_path in position delete
file", iceberg.ErrInvalidSchema)
+}
+
+func (a *posDeleteAccumulator) release() {
+ for _, builder := range a.builders {
+ builder.Release()
}
- if filePathValueType(filePathCol.DataType()).ID() == arrow.STRING_VIEW {
- return nil, fmt.Errorf("%w: unsupported file_path column type
%s in position delete file",
- iceberg.ErrInvalidSchema, filePathCol.DataType())
+ a.builders = nil
+}
+
+func (a *posDeleteAccumulator) finish() map[string]*arrow.Chunked {
+ results := make(map[string]*arrow.Chunked, len(a.builders))
+ for path, builder := range a.builders {
+ positions := builder.NewInt64Array()
+ builder.Release()
+
+ results[path] = arrow.NewChunked(arrow.PrimitiveTypes.Int64,
[]arrow.Array{positions})
+ positions.Release()
+ }
+ a.builders = nil
+
+ return results
+}
+
+func validatePosDeleteColumns(filePathType arrow.DataType, filePathNulls int,
filePathLen int,
+ posType arrow.DataType, posNulls int, posLen int,
+) error {
+ if filePathNulls > 0 {
+ return fmt.Errorf("%w: null file_path in position delete file",
iceberg.ErrInvalidSchema)
+ }
+ if filePathValueType(filePathType).ID() == arrow.STRING_VIEW {
+ return fmt.Errorf("%w: unsupported file_path column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, filePathType)
}
- if posCol.NullN() > 0 {
- return nil, fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
+ if posNulls > 0 {
+ return fmt.Errorf("%w: null pos in position delete file",
iceberg.ErrInvalidSchema)
}
- if posCol.DataType().ID() != arrow.INT64 {
- return nil, fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
- iceberg.ErrInvalidSchema, posCol.DataType())
+ if posType.ID() != arrow.INT64 {
+ return fmt.Errorf("%w: unsupported pos column type %s in
position delete file",
+ iceberg.ErrInvalidSchema, posType)
}
- if filePathCol.Len() != posCol.Len() {
- return nil, fmt.Errorf("%w: file_path and pos columns have
different lengths: %d and %d",
- iceberg.ErrInvalidSchema, filePathCol.Len(),
posCol.Len())
+ if filePathLen != posLen {
+ return fmt.Errorf("%w: file_path and pos columns have different
lengths: %d and %d",
+ iceberg.ErrInvalidSchema, filePathLen, posLen)
}
- mem := compute.GetAllocator(ctx)
- posCursor, err := newPosDeleteCursor(posCol)
+ return nil
+}
+
+func (a *posDeleteAccumulator) appendFilePathChunk(ctx context.Context,
filePathChunk arrow.Array,
+ posCursor *posDeleteCursor,
+) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+
+ paths, err := filePathValues(filePathChunk)
if err != nil {
- return nil, err
+ return err
}
- builders := make(map[string]*array.Int64Builder)
- defer func() {
- if err != nil {
- for _, builder := range builders {
- builder.Release()
+ var dictionary arrow.Array
+ var indices *array.Dictionary
+ if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
+ dictionary = dict.Dictionary()
+ indices = dict
+ }
+
+ for i := range filePathChunk.Len() {
+ if i&(positionalDeleteCancellationCheckInterval-1) == 0 {
+ if err := ctx.Err(); err != nil {
+ return err
}
}
- }()
- for _, filePathChunk := range filePathCol.Chunks() {
- if err := ctx.Err(); err != nil {
- return nil, err
+ pos, ok := posCursor.next()
+ if !ok {
+ return fmt.Errorf("%w: position delete columns ended
before file_path column",
+ iceberg.ErrInvalidSchema)
}
-
- paths, pathErr := filePathValues(filePathChunk)
- if pathErr != nil {
- return nil, pathErr
+ if pos < 0 {
+ return fmt.Errorf("%w: negative pos %d in position
delete file",
+ iceberg.ErrInvalidSchema, pos)
+ }
+ if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
+ return fmt.Errorf("%w: null file_path dictionary value
in position delete file",
+ iceberg.ErrInvalidSchema)
}
- var dictionary arrow.Array
- var indices *array.Dictionary
- if dict, ok := filePathChunk.(*array.Dictionary); ok &&
dict.Dictionary().NullN() > 0 {
- dictionary = dict.Dictionary()
- indices = dict
+ path := paths.Value(i)
+ builder, ok := a.builders[path]
+ if !ok {
+ path = strings.Clone(path)
+ builder = array.NewInt64Builder(a.mem)
+ a.builders[path] = builder
}
+ builder.Append(pos)
+ }
- for i := range filePathChunk.Len() {
- if i&(positionalDeleteCancellationCheckInterval-1) == 0
{
- if err := ctx.Err(); err != nil {
- return nil, err
- }
- }
+ return nil
+}
- pos, ok := posCursor.next()
- if !ok {
- return nil, fmt.Errorf("%w: position delete
columns ended before file_path column",
- iceberg.ErrInvalidSchema)
- }
- if pos < 0 {
- return nil, fmt.Errorf("%w: negative pos %d in
position delete file",
- iceberg.ErrInvalidSchema, pos)
- }
- if dictionary != nil &&
dictionary.IsNull(indices.GetValueIndex(i)) {
- return nil, fmt.Errorf("%w: null file_path
dictionary value in position delete file",
- iceberg.ErrInvalidSchema)
- }
+func (a *posDeleteAccumulator) appendChunked(ctx context.Context, filePathCol,
posCol *arrow.Chunked) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if err := validatePosDeleteColumns(filePathCol.DataType(),
filePathCol.NullN(), filePathCol.Len(),
+ posCol.DataType(), posCol.NullN(), posCol.Len()); err != nil {
+ return err
+ }
- path := paths.Value(i)
- builder, ok := builders[path]
- if !ok {
- path = strings.Clone(path)
- builder = array.NewInt64Builder(mem)
- builders[path] = builder
- }
- builder.Append(pos)
+ posCursor, err := newPosDeleteCursor(posCol)
+ if err != nil {
+ return err
+ }
+
+ for _, filePathChunk := range filePathCol.Chunks() {
+ if err := a.appendFilePathChunk(ctx, filePathChunk,
&posCursor); err != nil {
+ return err
}
}
- if err := ctx.Err(); err != nil {
- return nil, err
+
+ return ctx.Err()
+}
+
+func (a *posDeleteAccumulator) appendRecord(ctx context.Context, record
arrow.RecordBatch) error {
+ if record.NumCols() != 2 {
+ return fmt.Errorf("%w: projected position delete record has %d
columns, expected 2",
+ iceberg.ErrInvalidSchema, record.NumCols())
}
- results = make(map[string]*arrow.Chunked, len(builders))
- for path, builder := range builders {
- positions := builder.NewInt64Array()
- builder.Release()
+ filePathCol := record.Column(0)
Review Comment:
I chased down whether this hardcoded ordering is safe, since `readDeletes`
projects with `GetRecords(ctx, []int{filePathIndex, posIndex}, nil)` and here
we assume col 0 is `file_path` and col 1 is `pos`. It's correct: arrow-go's
`SchemaManifest.GetFieldIndices` builds the projected field list by iterating
the request slice in order, and `GetFieldReaders` fills the output schema
positionally from that, so the projection always comes back as `[file_path,
pos]` regardless of the file's physical layout, even a reversed `[pos,
file_path]` delete file.
What worries me is that nothing in the new path pins that.
`TestReadDeletesProjectsColumnsAndAccumulatesBatches` only writes a
standard-order file, so if arrow-go ever changed its emission order this would
silently mis-assign the columns and surface as a confusing `unsupported pos
column type string`. I'd add a `readDeletes` case that writes a reversed-schema
file (pos physically first) and asserts the positions still group under the
right paths. wdyt?
--
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]
