nssalian commented on code in PR #2002:
URL: https://github.com/apache/iceberg-go/pull/2002#discussion_r4040113479
##########
table/variant_residual.go:
##########
@@ -68,7 +68,112 @@ 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
+}
+
+// variantPathOf returns the extract term's member-name path (kept off the
public BoundExtract interface); false if the term has none.
+func variantPathOf(t iceberg.BoundExtract) (variant.VariantPath, bool) {
+ vp, ok := t.(interface {
+ VariantPath() variant.VariantPath
+ })
+ if !ok {
+ return variant.VariantPath{}, false
+ }
+
+ return vp.VariantPath(), true
+}
+
+// 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 err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ path, ok := variantPathOf(col.Term)
+ if !ok {
+ return extractColumnValuesPerRow(ctx, varr, col, dt, mem)
+ }
+ if fast := tryShreddedTypedColumn(varr, path, dt, mem); fast != nil {
+ return fast, nil
+ }
+ if !varr.IsShredded() {
+ return extractColumnValuesPerRow(ctx, varr, col, dt, mem)
+ }
+
+ extracted, err := compute.VariantGet(ctx, varr,
compute.VariantGetOptions{Path: path})
+ if err != nil {
+ if errors.Is(err, context.Canceled) || errors.Is(err,
context.DeadlineExceeded) {
+ return nil, err
+ }
+
+ return extractColumnValuesPerRow(ctx, varr, col, dt, mem)
+ }
+ defer extracted.Release()
+
+ leaves, ok := extracted.(*extensions.VariantArray)
+ if !ok {
+ return extractColumnValuesPerRow(ctx, varr, col, dt, mem)
+ }
+
+ bldr := array.NewBuilder(mem, dt)
+ defer bldr.Release()
+
+ varName := col.Term.Ref().Field().Name
+ for i := range leaves.Len() {
Review Comment:
Done. I added the same `if i%4096 == 0 { ctx.Err() }` poll to the
middle-tier leaf loop, so a cancel after `VariantGet` returns is noticed within
4096 rows.
--
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]
