feat(blob): support MAP<K, BLOB> reads in Go

.licenserc.yaml

@@ -28,6 +28,7 @@ header:
     - ".github/PULL_REQUEST_TEMPLATE.md"
     - "crates/paimon/tests/**/*.json"
     - "crates/paimon/testdata/**"
+    - "bindings/go/tests/testdata/**"
     - "third-party-licenses/jieba-rs-0.10.3.LICENSE"
     - "third-party-licenses/openssl-1.1.1.LICENSE"
     - "**/go.sum"

bindings/go/blob_reader.go

@@ -23,9 +23,12 @@ import (
 	"context"
 	"fmt"
 	"runtime"
+	"strings"
 	"sync"
 	"unsafe"
 
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/jupiterrider/ffi"
 )
 
@@ -83,6 +86,40 @@ func (r *BlobReader) ReadBlobs(descriptors [][]byte) ([][]byte, error) {
 	return ffiBlobReaderReadBlobs.symbol(r.ctx)(r.inner, descriptors)
 }
 
+// StringBlobMapDescriptors returns one MAP<STRING, BLOB> row.
+func StringBlobMapDescriptors(column arrow.Array, row int) (map[string][]byte, error) {
+	m, ok := column.(*array.Map)
+	if !ok {
+		return nil, fmt.Errorf("paimon: BLOB map column is %T, want *array.Map", column)
+	}
+	if row < 0 || row >= m.Len() {
+		return nil, fmt.Errorf("paimon: BLOB map row %d is out of range", row)
+	}
+	if m.IsNull(row) {
+		return nil, nil
+	}
+	keys, ok := m.Keys().(*array.String)
+	if !ok {
+		return nil, fmt.Errorf("paimon: BLOB map keys are %T, want *array.String", m.Keys())
+	}
+	descriptors, ok := m.Items().(*array.Binary)
+	if !ok {
+		return nil, fmt.Errorf("paimon: BLOB map values are %T, want *array.Binary", m.Items())
+	}
+	start, end := m.ValueOffsets(row)
+	result := make(map[string][]byte, end-start)
+	for index := start; index < end; index++ {
+		i := int(index)
+		key := strings.Clone(keys.Value(i))
+		if descriptors.IsNull(i) {
+			result[key] = nil
+			continue
+		}
+		result[key] = append([]byte(nil), descriptors.Value(i)...)
+	}
+	return result, nil
+}
+
 // Close releases the reader and is idempotent.
 func (r *BlobReader) Close() {
 	r.mu.Lock()

bindings/go/table.go

@@ -21,6 +21,7 @@ package paimon
 
 import (
 	"context"
+	"runtime"
 	"sync"
 	"unsafe"
 
@@ -54,8 +55,20 @@ func (t *Table) NewReadBuilder() (*ReadBuilder, error) {
 	if t.inner == nil {
 		return nil, ErrClosed
 	}
-	createFn := ffiTableNewReadBuilder.symbol(t.ctx)
-	inner, err := createFn(t.inner)
+	inner, err := ffiTableNewReadBuilder.symbol(t.ctx)(t.inner)
+	if err != nil {
+		return nil, err
+	}
+	t.lib.acquire()
+	return &ReadBuilder{ctx: t.ctx, lib: t.lib, inner: inner}, nil
+}
+
+// NewReadBuilderWithOptions creates a ReadBuilder with per-read options.
+func (t *Table) NewReadBuilderWithOptions(options map[string]string) (*ReadBuilder, error) {
+	if t.inner == nil {
+		return nil, ErrClosed
+	}
+	inner, err := ffiTableNewReadBuilderWithOptions.symbol(t.ctx)(t.inner, options)
 	if err != nil {
 		return nil, err
 	}
@@ -93,3 +106,49 @@ var ffiTableNewReadBuilder = newFFI(ffiOpts{
 		return result.readBuilder, nil
 	}
 })
+
+var ffiTableNewReadBuilderWithOptions = newFFI(ffiOpts{
+	sym:   "paimon_table_new_read_builder_with_options",
+	rType: &typeResultReadBuilder,
+	aTypes: []*ffi.Type{
+		&ffi.TypePointer,
+		&ffi.TypePointer,
+		&ffi.TypePointer,
+	},
+}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable, map[string]string) (*paimonReadBuilder, error) {
+	return func(table *paimonTable, options map[string]string) (*paimonReadBuilder, error) {
+		type paimonOption struct {
+			key   *byte
+			value *byte
+		}
+		opts := make([]paimonOption, 0, len(options))
+		for key, value := range options {
+			keyPtr, err := bytePtrFromString(key)
+			if err != nil {
+				return nil, err
+			}
+			valuePtr, err := bytePtrFromString(value)
+			if err != nil {
+				return nil, err
+			}
+			opts = append(opts, paimonOption{key: keyPtr, value: valuePtr})
+		}
+		var optsPtr unsafe.Pointer
+		if len(opts) > 0 {
+			optsPtr = unsafe.Pointer(&opts[0])
+		}
+		optsLen := uintptr(len(opts))
+		var result resultReadBuilder
+		ffiCall(
+			unsafe.Pointer(&result),
+			unsafe.Pointer(&table),
+			unsafe.Pointer(&optsPtr),
+			unsafe.Pointer(&optsLen),
+		)
+		runtime.KeepAlive(opts)
+		if result.error != nil {
+			return nil, parseError(ctx, result.error)
+		}
+		return result.readBuilder, nil
+	}
+})

bindings/go/tests/blob_reader_test.go

@@ -30,6 +30,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/apache/arrow-go/v18/arrow/array"
 	paimon "github.com/apache/paimon-rust/bindings/go"
 )
 
@@ -101,6 +102,95 @@ func TestBlobReaderReadBlobAndBatch(t *testing.T) {
 	}
 }
 
+func TestStringBlobMapDescriptors(t *testing.T) {
+	source := filepath.Join("testdata", "map_blob_table")
+	warehouse := t.TempDir()
+	if err := copyDirectory(source, filepath.Join(warehouse, "default.db", "map_blob_table")); err != nil {
+		t.Fatal(err)
+	}
+	table := openTableAt(t, warehouse, "map_blob_table")
+	builder, err := table.NewReadBuilderWithOptions(map[string]string{
+		"blob-as-descriptor": "true",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer builder.Close()
+	if err := builder.WithProjection([]string{"id", "assets"}); err != nil {
+		t.Fatal(err)
+	}
+	scan, err := builder.NewScan()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer scan.Close()
+	plan, err := scan.Plan()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer plan.Close()
+	read, err := builder.NewRead()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer read.Close()
+	batches, err := read.NewRecordBatchReader(plan.Splits())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer batches.Close()
+
+	rows := make(map[int32]map[string][]byte)
+	for {
+		record, err := batches.NextRecord()
+		if errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		ids := record.Column(0).(*array.Int32)
+		for row := 0; row < int(record.NumRows()); row++ {
+			rows[ids.Value(row)], err = paimon.StringBlobMapDescriptors(record.Column(1), row)
+			if err != nil {
+				record.Release()
+				t.Fatal(err)
+			}
+		}
+		record.Release()
+	}
+
+	descriptors := rows[1]
+	if len(rows) != 3 {
+		t.Fatalf("read %d rows, want 3", len(rows))
+	}
+	if len(descriptors["first"]) == 0 || len(descriptors["tail"]) == 0 {
+		t.Fatalf("descriptor map is invalid after Arrow release: %#v", descriptors)
+	}
+	if rows[2] != nil || len(rows[3]) != 0 {
+		t.Fatalf("unexpected null or empty maps: %#v", rows)
+	}
+	reader, err := table.NewBlobReader()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer reader.Close()
+	resolved, err := reader.ReadBlobs([][]byte{
+		descriptors["tail"],
+		descriptors["first"],
+		descriptors["empty"],
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(resolved[0]) != "ghij" || string(resolved[1]) != "abc" || len(resolved[2]) != 0 {
+		t.Fatalf("unexpected values: %q", resolved)
+	}
+	if descriptors["null"] != nil {
+		t.Fatalf("null BLOB returned %#v", descriptors["null"])
+	}
+}
+
 func TestBlobReaderFromTableOutlivesTable(t *testing.T) {
 	file := writeBlobFile(t, "table", "abcdefghij")
 

bindings/go/tests/testdata/map_blob_table/schema/schema-0

@@ -0,0 +1,30 @@
+{
+  "version": 3,
+  "id": 0,
+  "fields": [
+    {
+      "id": 0,
+      "name": "id",
+      "type": "INT"
+    },
+    {
+      "id": 1,
+      "name": "assets",
+      "type": {
+        "type": "MAP",
+        "key": "STRING NOT NULL",
+        "value": "BLOB",
+        "nullable": true
+      }
+    }
+  ],
+  "highestFieldId": 1,
+  "partitionKeys": [],
+  "primaryKeys": [],
+  "options": {
+    "row-tracking.enabled": "true",
+    "data-evolution.enabled": "true"
+  },
+  "comment": null,
+  "timeMillis": 1788359448372
+}

bindings/go/tests/testdata/map_blob_table/snapshot/LATEST

@@ -0,0 +1 @@
+1

bindings/go/tests/testdata/map_blob_table/snapshot/snapshot-1

@@ -0,0 +1,15 @@
+{
+  "version": 3,
+  "id": 1,
+  "schemaId": 0,
+  "baseManifestList": "manifest-list-d854a2ec-c453-4ece-bcbe-5ad5411a8273-0",
+  "deltaManifestList": "manifest-list-d854a2ec-c453-4ece-bcbe-5ad5411a8273-1",
+  "totalRecordCount": 6,
+  "deltaRecordCount": 6,
+  "commitUser": "ab9bd0e7-c0bc-4695-84cf-44f36df42a6a",
+  "commitIdentifier": 9223372036854775807,
+  "commitKind": "APPEND",
+  "timeMillis": 1788359448379,
+  "nextRowId": 3,
+  "uuid": "19a3dfb1-eae3-4d7a-8329-26b85143edcc"
+}