nssalian commented on code in PR #2002:
URL: https://github.com/apache/iceberg-go/pull/2002#discussion_r4020193288
##########
table/variant_residual.go:
##########
@@ -68,7 +67,84 @@ func buildExtractColumn(col iceberg.VariantExtractColumn,
rec arrow.RecordBatch,
return nil, arrow.Field{}, fmt.Errorf("%w: variant extract
column %q is not a VariantArray (got %T)", iceberg.ErrInvalidArgument, varName,
arr)
}
- for i := range n {
+ out, err := extractColumnValues(ctx, varr, col, typ, dt, mem)
+ if err != nil {
+ return nil, arrow.Field{}, err
+ }
+
+ field := arrow.Field{
+ Name: col.Name,
+ Type: dt,
+ Nullable: true,
+ Metadata: arrow.NewMetadata([]string{ArrowParquetFieldIDKey},
[]string{strconv.Itoa(col.FieldID)}),
+ }
+
+ return out, field, nil
+}
+
+// extractColumnValues navigates columnarly then casts each leaf with
iceberg's cast; a VariantGet
+// navigation error (arrow-rs errors where the residual filter wants
null-on-miss) falls back to per-row.
+func extractColumnValues(ctx context.Context, varr *extensions.VariantArray,
col iceberg.VariantExtractColumn, typ iceberg.PrimitiveType, dt arrow.DataType,
mem memory.Allocator) (arrow.Array, error) {
+ if fast := tryShreddedTypedColumn(varr, col.Term.VariantPath(), dt,
mem); fast != nil {
+ return fast, nil
+ }
+ if !varr.IsShredded() {
+ return extractColumnValuesPerRow(varr, col, dt, mem)
+ }
+
+ extracted, err := compute.VariantGet(ctx, varr,
compute.VariantGetOptions{Path: col.Term.VariantPath()})
Review Comment:
Fixed this. `extractColumnValues` checks `ctx.Err()` up front; on a
`VariantGet` error it returns the error when it wraps
`context.Canceled`/`DeadlineExceeded` (`errors.Is`) and only falls back on a
real navigation miss; `extractColumnValuesPerRow` now checks `ctx.Err()` in its
loop. Test: `TestExtractColumnValuesContextCancelled`.
##########
table/variant_residual.go:
##########
@@ -91,18 +167,124 @@ func buildExtractColumn(col iceberg.VariantExtractColumn,
rec arrow.RecordBatch,
}
if aerr := appendExtractLiteral(bldr, lit); aerr != nil {
- return nil, arrow.Field{}, aerr
+ return nil, aerr
}
}
- field := arrow.Field{
- Name: col.Name,
- Type: dt,
- Nullable: true,
- Metadata: arrow.NewMetadata([]string{ArrowParquetFieldIDKey},
[]string{strconv.Itoa(col.FieldID)}),
+ return bldr.NewArray(), nil
+}
+
+// tryShreddedTypedColumn returns the field's typed leaf column when it is
shredded to exactly dt, else nil.
+func tryShreddedTypedColumn(varr *extensions.VariantArray, path
variant.VariantPath, dt arrow.DataType, mem memory.Allocator) arrow.Array {
+ if path.Len() == 0 || varr.Data().Offset() != 0 {
+ return nil
+ }
+ tv := varr.Shredded()
+ if tv == nil || rootResidualHidesRows(varr, tv) {
+ return nil
+ }
+ n := varr.Len()
+
+ var mask *memory.Buffer
+ mergeValidity := func(arr arrow.Array) {
+ if arr.NullN() == 0 {
+ return
+ }
+ vb := arr.Data().Buffers()[0]
+ if vb == nil {
+ return
+ }
+ if mask == nil {
+ mask = bitutil.BitmapAndAlloc(mem, vb.Bytes(),
vb.Bytes(), 0, 0, int64(n), 0)
Review Comment:
First merge clones the bitmap (`memory.NewResizableBuffer` + `copy`); the
AND is kept only for the merge path.
##########
table/variant_residual.go:
##########
@@ -91,18 +167,124 @@ func buildExtractColumn(col iceberg.VariantExtractColumn,
rec arrow.RecordBatch,
}
if aerr := appendExtractLiteral(bldr, lit); aerr != nil {
- return nil, arrow.Field{}, aerr
+ return nil, aerr
}
}
- field := arrow.Field{
- Name: col.Name,
- Type: dt,
- Nullable: true,
- Metadata: arrow.NewMetadata([]string{ArrowParquetFieldIDKey},
[]string{strconv.Itoa(col.FieldID)}),
+ return bldr.NewArray(), nil
+}
+
+// tryShreddedTypedColumn returns the field's typed leaf column when it is
shredded to exactly dt, else nil.
+func tryShreddedTypedColumn(varr *extensions.VariantArray, path
variant.VariantPath, dt arrow.DataType, mem memory.Allocator) arrow.Array {
+ if path.Len() == 0 || varr.Data().Offset() != 0 {
+ return nil
+ }
+ tv := varr.Shredded()
+ if tv == nil || rootResidualHidesRows(varr, tv) {
+ return nil
+ }
+ n := varr.Len()
+
+ var mask *memory.Buffer
+ mergeValidity := func(arr arrow.Array) {
+ if arr.NullN() == 0 {
+ return
+ }
+ vb := arr.Data().Buffers()[0]
+ if vb == nil {
+ return
+ }
+ if mask == nil {
+ mask = bitutil.BitmapAndAlloc(mem, vb.Bytes(),
vb.Bytes(), 0, 0, int64(n), 0)
+
+ return
+ }
+ merged := bitutil.BitmapAndAlloc(mem, mask.Bytes(), vb.Bytes(),
0, 0, int64(n), 0)
+ mask.Release()
+ mask = merged
+ }
+ bail := func() arrow.Array {
+ if mask != nil {
+ mask.Release()
+ }
+
+ return nil
+ }
+
+ mergeValidity(varr.Storage())
+
+ cur := tv
+ for i := range path.Len() {
+ name, _, isField := path.StepAt(i)
+ if !isField {
+ return bail()
+ }
+ st, ok := cur.(*array.Struct)
+ if !ok {
+ return bail()
+ }
+ mergeValidity(st)
+ idx, ok := st.DataType().(*arrow.StructType).FieldIdx(name)
+ if !ok {
+ return bail()
+ }
+ field, ok := st.Field(idx).(*array.Struct)
+ if !ok {
+ return bail()
+ }
+ fty := field.DataType().(*arrow.StructType)
Review Comment:
I don't think we should do this. The premise doesn't hold for arrow-go. I
built the exact non-cascaded shape and measured it against the per-row oracle:
`VariantArray.Value` reassembly reads the live child there and returns `5`, it
does not treat the null wrapper as absent. So `mergeValidity(field)` would make
the fast path return null while per-row returns `5`, it breaks the
fast==per-row parity invariant instead of fixing a bug. Test
`TestFastPathWrapperFieldNullMatchesPerRow` pins that both paths agree (both
read the child) for this spec-undefined shape. Making it spec-correct (null)
requires the fix in arrow-go's reassembly so both paths change together.
--
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]
