fallintoplace commented on code in PR #1229:
URL: https://github.com/apache/iceberg-go/pull/1229#discussion_r3449017695
##########
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:
Thanks, pushed `37c45a8`.
One note: I left the happy-path layout matrix on the default allocator. When
I ran that through a checked allocator, Arrow's string/dictionary `Unique` path
still showed live allocations after the returned objects were released. The
reject cases use a checked allocator now, so the cleanup/error paths are still
covered.
--
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]
