feat(substrait) Update to substrait v0.64.0

* Update proto definition to v0.64.0
* Fixed FetchRel to use OffsetMode and CountMode

doc.go

@@ -3,7 +3,7 @@
 // Package substraitgo contains the experimental go bindings for substrait
 // (https://substrait.io).
 //
-// Current generated proto substrait version: v0.55.0
+// Current generated proto substrait version: v0.64.0
 package substraitgo
 
-//go:generate buf generate https://github.com/substrait-io/substrait.git#tag=v0.59.0
+//go:generate buf generate https://github.com/substrait-io/substrait.git#tag=v0.64.0

go.mod

@@ -30,6 +30,7 @@ require (
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	github.com/substrait-io/substrait-go v1.2.0 // indirect
 	golang.org/x/crypto v0.30.0 // indirect
 	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/text v0.21.0 // indirect

go.sum

@@ -56,6 +56,8 @@ github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOf
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/substrait-io/substrait v0.63.1 h1:XNPvrEYNPjDqenK4TxqBDDUNzglafdjzjejzQqEwk5Y=
 github.com/substrait-io/substrait v0.63.1/go.mod h1:MPFNw6sToJgpD5Z2rj0rQrdP/Oq8HG7Z2t3CAEHtkHw=
+github.com/substrait-io/substrait-go v1.2.0 h1:3ZNRkc8FYD7ifCagKEOZQtUcgMceMQfwo2N1NGaK4Q4=
+github.com/substrait-io/substrait-go v1.2.0/go.mod h1:IPsy24rdjp/buXR+T8ENl6QCnSCS6h+uM8P+GaZez7c=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.30.0 h1:RwoQn3GkWiMkzlX562cLB7OxWvjH1L8xutO2WoJcRoY=
 golang.org/x/crypto v0.30.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=

plan/builders.go

@@ -4,7 +4,6 @@ package plan
 
 import (
 	"fmt"
-
 	substraitgo "github.com/substrait-io/substrait-go/v3"
 	"github.com/substrait-io/substrait-go/v3/expr"
 	"github.com/substrait-io/substrait-go/v3/extensions"
@@ -124,6 +123,7 @@ type Builder interface {
 	// Deprecated: Use VirtualTableFromExpr(...).Remap() instead.
 	VirtualTableFromExprRemap(fieldNames []string, remap []int32, values ...expr.VirtualTableExpressionValue) (*VirtualTableReadRel, error)
 	VirtualTableFromExpr(fieldNames []string, values ...expr.VirtualTableExpressionValue) (*VirtualTableReadRel, error)
+	IcebergTableFromMetadataFile(metadataURI string, schema types.NamedStruct) (*IcebergTableReadRel, error)
 	// Deprecated: Use Sort(...).Remap() instead.
 	SortRemap(input Rel, remap []int32, sorts ...expr.SortField) (*SortRel, error)
 	Sort(input Rel, sorts ...expr.SortField) (*SortRel, error)
@@ -587,6 +587,18 @@ func (b *builder) VirtualTable(fields []string, values ...expr.StructLiteralValu
 	return b.VirtualTableRemap(fields, nil, values...)
 }
 
+func (b *builder) IcebergTableFromMetadataFile(metadataURI string, schema types.NamedStruct) (*IcebergTableReadRel, error) {
+	tableType := &Direct{}
+	tableType.MetadataUri = metadataURI
+
+	return &IcebergTableReadRel{
+		baseReadRel: baseReadRel{
+			baseSchema: schema,
+		},
+		tableType: tableType,
+	}, nil
+}
+
 func (b *builder) SortRemap(input Rel, remap []int32, sorts ...expr.SortField) (*SortRel, error) {
 	if input == nil {
 		return nil, errNilInputRel

plan/plan.go

@@ -411,10 +411,24 @@ func RelFromProto(rel *proto.Rel, reg expr.ExtensionRegistry) (Rel, error) {
 			return nil, fmt.Errorf("error getting input to FetchRel: %w", err)
 		}
 
+		var offset int64
+		if off, ok := rel.Fetch.OffsetMode.(*proto.FetchRel_Offset); ok {
+			offset = off.Offset
+		} else {
+			return nil, fmt.Errorf("%w: missing required Offset field for Fetch Relation", substraitgo.ErrInvalidRel)
+		}
+
+		var count int64
+		if cnt, ok := rel.Fetch.CountMode.(*proto.FetchRel_Count); ok {
+			count = cnt.Count
+		} else {
+			return nil, fmt.Errorf("%w: missing required Count field for Fetch Relation", substraitgo.ErrInvalidRel)
+		}
+
 		out := &FetchRel{
 			input:        input,
-			offset:       rel.Fetch.Offset,
-			count:        rel.Fetch.Count,
+			offset:       offset,
+			count:        count,
 			advExtension: rel.Fetch.AdvancedExtension,
 		}
 		if rel.Fetch.Common != nil {

plan/relations.go

@@ -371,6 +371,107 @@ func (e *ExtensionTableReadRel) Remap(mapping ...int32) (Rel, error) {
 	return RemapHelper(e, mapping)
 }
 
+type (
+	SnapshotId        proto.ReadRel_IcebergTable_MetadataFileRead_SnapshotId
+	SnapshotTimestamp proto.ReadRel_IcebergTable_MetadataFileRead_SnapshotTimestamp
+
+	IcebergSnapshot interface {
+		isSnapshot()
+	}
+)
+
+func (*SnapshotId) isSnapshot()        {}
+func (*SnapshotTimestamp) isSnapshot() {}
+
+type (
+	Direct proto.ReadRel_IcebergTable_MetadataFileRead
+
+	IcebergTableType interface {
+		isTableType()
+	}
+)
+
+func (*Direct) isTableType() {}
+
+// IcebergTableReadRel is a scan on iceberg table.
+type IcebergTableReadRel struct {
+	baseReadRel
+
+	tableType    IcebergTableType
+	advExtension *extensions.AdvancedExtension
+}
+
+func (n *IcebergTableReadRel) NamedTableAdvancedExtension() *extensions.AdvancedExtension {
+	return n.advExtension
+}
+
+func (n *IcebergTableReadRel) RecordType() types.RecordType {
+	return n.remap(n.directOutputSchema())
+}
+
+func (n *IcebergTableReadRel) ToProtoPlanRel() *proto.PlanRel {
+	return &proto.PlanRel{
+		RelType: &proto.PlanRel_Rel{
+			Rel: n.ToProto(),
+		},
+	}
+}
+
+func (n *IcebergTableReadRel) ToProto() *proto.Rel {
+	readRel := n.toReadRelProto()
+
+	if n.tableType != nil {
+		if directTableType, ok := n.tableType.(*Direct); ok {
+			readRel.ReadType = &proto.ReadRel_IcebergTable_{
+				IcebergTable: &proto.ReadRel_IcebergTable{
+					TableType: &proto.ReadRel_IcebergTable_Direct{
+						Direct: (*proto.ReadRel_IcebergTable_MetadataFileRead)(directTableType),
+					},
+				},
+			}
+		}
+	}
+
+	readRel.ReadType = &proto.ReadRel_IcebergTable_{
+		IcebergTable: &proto.ReadRel_IcebergTable{
+			TableType: &proto.ReadRel_IcebergTable_Direct{
+				Direct: &proto.ReadRel_IcebergTable_MetadataFileRead{
+					MetadataUri: "metadata_uri",
+					Snapshot: &proto.ReadRel_IcebergTable_MetadataFileRead_SnapshotId{
+						SnapshotId: "snapshot_id",
+					},
+				},
+			},
+		},
+	}
+	return &proto.Rel{
+		RelType: &proto.Rel_Read{
+			Read: readRel,
+		},
+	}
+}
+
+func (n *IcebergTableReadRel) Copy(_ ...Rel) (Rel, error) {
+	return n, nil
+}
+
+func (n *IcebergTableReadRel) CopyWithExpressionRewrite(rewriteFunc RewriteFunc, _ ...Rel) (Rel, error) {
+	newExprs, err := n.copyExpressions(rewriteFunc)
+	if err != nil {
+		return nil, err
+	}
+	if slices.Equal(newExprs, n.getExpressions()) {
+		return n, nil
+	}
+	nt := *n
+	nt.updateFilters(newExprs)
+	return &nt, nil
+}
+
+func (n *IcebergTableReadRel) Remap(mapping ...int32) (Rel, error) {
+	return RemapHelper(n, mapping)
+}
+
 // PathType is the type of a LocalFileReadRel's uris.
 type PathType int8
 
@@ -897,10 +998,14 @@ func (f *FetchRel) ToProto() *proto.Rel {
 	return &proto.Rel{
 		RelType: &proto.Rel_Fetch{
 			Fetch: &proto.FetchRel{
-				Common:            f.toProto(),
-				Input:             f.input.ToProto(),
-				Offset:            f.offset,
-				Count:             f.count,
+				Common: f.toProto(),
+				Input:  f.input.ToProto(),
+				OffsetMode: &proto.FetchRel_Offset{
+					Offset: f.offset,
+				},
+				CountMode: &proto.FetchRel_Count{
+					Count: f.count,
+				},
 				AdvancedExtension: f.advExtension,
 			},
 		},