From 0351559f3e2f92c76744c61ec1cb0e4fb7e1120e Mon Sep 17 00:00:00 2001 From: David Li Date: Fri, 12 Jan 2024 10:50:58 -0500 Subject: [PATCH] feat(go/adbc/driver/flightsql): enable incremental queries Closes #1451. --- CONTRIBUTING.md | 2 +- go/adbc/adbc.go | 4 +- go/adbc/driver/driverbase/database.go | 2 +- go/adbc/driver/driverbase/driver.go | 2 +- .../driver/flightsql/cmd/testserver/main.go | 10 +- .../flightsql/flightsql_adbc_server_test.go | 302 +++++++++++++++++- .../driver/flightsql/flightsql_adbc_test.go | 12 +- .../driver/flightsql/flightsql_connection.go | 20 +- .../driver/flightsql/flightsql_database.go | 6 +- go/adbc/driver/flightsql/flightsql_driver.go | 2 +- .../driver/flightsql/flightsql_statement.go | 146 ++++++++- go/adbc/driver/flightsql/record_reader.go | 10 +- .../driver/flightsql/record_reader_test.go | 12 +- go/adbc/driver/internal/shared_utils.go | 6 +- go/adbc/driver/panicdummy/panicdummy_adbc.go | 6 +- go/adbc/driver/snowflake/connection.go | 4 +- go/adbc/driver/snowflake/driver.go | 2 +- go/adbc/driver/snowflake/driver_test.go | 8 +- go/adbc/driver/snowflake/record_reader.go | 10 +- go/adbc/driver/snowflake/statement.go | 6 +- go/adbc/drivermgr/wrapper.go | 6 +- go/adbc/drivermgr/wrapper_sqlite_test.go | 6 +- go/adbc/go.mod | 3 +- go/adbc/go.sum | 4 +- go/adbc/pkg/_tmpl/driver.go.tmpl | 8 +- go/adbc/pkg/flightsql/driver.go | 8 +- go/adbc/pkg/panicdummy/driver.go | 8 +- go/adbc/pkg/snowflake/driver.go | 8 +- go/adbc/sqldriver/driver.go | 10 +- go/adbc/sqldriver/driver_internals_test.go | 10 +- go/adbc/sqldriver/flightsql/flightsql.go | 2 +- go/adbc/sqldriver/flightsql/flightsql_test.go | 8 +- go/adbc/standard_schemas.go | 2 +- go/adbc/utils/utils.go | 2 +- go/adbc/validation/validation.go | 6 +- .../tools/create-go-vendor-archive.R | 2 +- 36 files changed, 549 insertions(+), 116 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 6c5ccaf219..b7c5249d4a 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -347,7 +347,7 @@ $ cd go/adbc && go-licenses report ./... \ --ignore github.com/apache/arrow/go/v11 \ --ignore github.com/apache/arrow/go/v12 \ --ignore github.com/apache/arrow/go/v13 \ - --ignore github.com/apache/arrow/go/v14 \ + --ignore github.com/apache/arrow/go/v15 \ --template ../../license.tpl > ../../LICENSE.txt 2> /dev/null ``` diff --git a/go/adbc/adbc.go b/go/adbc/adbc.go index 3fb61d692d..4e3636b1ed 100644 --- a/go/adbc/adbc.go +++ b/go/adbc/adbc.go @@ -40,8 +40,8 @@ import ( "context" "fmt" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) diff --git a/go/adbc/driver/driverbase/database.go b/go/adbc/driver/driverbase/database.go index e3a96ff16c..eda99b3e64 100644 --- a/go/adbc/driver/driverbase/database.go +++ b/go/adbc/driver/driverbase/database.go @@ -21,7 +21,7 @@ import ( "context" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory" "golang.org/x/exp/slog" ) diff --git a/go/adbc/driver/driverbase/driver.go b/go/adbc/driver/driverbase/driver.go index c47677948f..d043b39446 100644 --- a/go/adbc/driver/driverbase/driver.go +++ b/go/adbc/driver/driverbase/driver.go @@ -22,7 +22,7 @@ package driverbase import ( "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory" ) // DriverImpl is an interface that drivers implement to provide diff --git a/go/adbc/driver/flightsql/cmd/testserver/main.go b/go/adbc/driver/flightsql/cmd/testserver/main.go index bb3053d6c6..0ecda6c3b4 100644 --- a/go/adbc/driver/flightsql/cmd/testserver/main.go +++ b/go/adbc/driver/flightsql/cmd/testserver/main.go @@ -31,11 +31,11 @@ import ( "strconv" "strings" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/memory" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" diff --git a/go/adbc/driver/flightsql/flightsql_adbc_server_test.go b/go/adbc/driver/flightsql/flightsql_adbc_server_test.go index a591f1cae2..8216bf6851 100644 --- a/go/adbc/driver/flightsql/flightsql_adbc_server_test.go +++ b/go/adbc/driver/flightsql/flightsql_adbc_server_test.go @@ -26,17 +26,20 @@ import ( "net/textproto" "os" "strings" + "strconv" "testing" "time" + "sync" + "github.com/google/uuid" "github.com/apache/arrow-adbc/go/adbc" driver "github.com/apache/arrow-adbc/go/adbc/driver/flightsql" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql/schema_ref" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql/schema_ref" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/golang/protobuf/ptypes/wrappers" "github.com/stretchr/testify/suite" "golang.org/x/exp/maps" @@ -108,6 +111,10 @@ func TestExecuteSchema(t *testing.T) { suite.Run(t, &ExecuteSchemaTests{}) } +func TestIncrementalPoll(t *testing.T) { + suite.Run(t, &IncrementalPollTests{}) +} + func TestTimeout(t *testing.T) { suite.Run(t, &TimeoutTests{}) } @@ -427,6 +434,289 @@ func (ts *ExecuteSchemaTests) TestQuery() { ts.True(expectedSchema.Equal(schema), schema.String()) } +// ---- IncrementalPoll Tests -------------------- + +type IncrementalQuery struct { + query string + nextIndex int +} + +type IncrementalPollTestServer struct { + flightsql.BaseServer + mu sync.Mutex + queries map[string]*IncrementalQuery + testCases map[string]IncrementalPollTestCase +} + +func (srv *IncrementalPollTestServer) PollFlightInfo(ctx context.Context, desc *flight.FlightDescriptor) (*flight.PollInfo, error) { + srv.mu.Lock() + defer srv.mu.Unlock() + + var val wrapperspb.StringValue + var err error + if err = proto.Unmarshal(desc.Cmd, &val); err != nil { + return nil, err + } + queryId := val.Value + progress := int64(0) + if strings.Contains(queryId, ";") { + parts := strings.SplitN(queryId, ";", 2) + queryId = parts[0] + progress, err = strconv.ParseInt(parts[1], 10, 32) + if err != nil { + return nil, err + } + } + + query, ok := srv.queries[queryId] + if !ok { + return nil, status.Errorf(codes.NotFound, "Query ID not found") + } + + testCase, ok := srv.testCases[query.query] + if !ok { + return nil, status.Errorf(codes.Unimplemented, fmt.Sprintf("Invalid case %s", query.query)) + } + + if testCase.differentRetryDescriptor && progress != int64(query.nextIndex) { + return nil, status.Errorf(codes.InvalidArgument, fmt.Sprintf("Used wrong retry descriptor, expected %d but got %d", query.nextIndex, progress)) + } + + return srv.MakePollInfo(&testCase, query, queryId) +} + +func (srv *IncrementalPollTestServer) PollFlightInfoStatement(ctx context.Context, query flightsql.StatementQuery, desc *flight.FlightDescriptor) (*flight.PollInfo, error) { + queryId := uuid.New().String() + + testCase, ok := srv.testCases[query.GetQuery()] + if !ok { + return nil, status.Errorf(codes.Unimplemented, fmt.Sprintf("Invalid case %s", query.GetQuery())) + } + + srv.mu.Lock() + defer srv.mu.Unlock() + + srv.queries[queryId] = &IncrementalQuery{ + query: query.GetQuery(), + nextIndex: 0, + } + + return srv.MakePollInfo(&testCase, srv.queries[queryId], queryId) +} + +func (srv *IncrementalPollTestServer) MakePollInfo(testCase *IncrementalPollTestCase, query *IncrementalQuery, queryId string) (*flight.PollInfo, error) { + schema := flight.SerializeSchema(arrow.NewSchema([]arrow.Field{ + {Name: "ints", Type: arrow.PrimitiveTypes.Int32}, + }, nil), srv.Alloc) + + pb := wrapperspb.StringValue{Value: queryId} + if testCase.differentRetryDescriptor { + pb.Value = queryId + ";" + strconv.Itoa(query.nextIndex + 1) + } + descriptor, err := proto.Marshal(&pb) + if err != nil { + return nil, err + } + + numEndpoints := 0 + for i := 0; i <= query.nextIndex; i++ { + if i >= len(testCase.progress) { + break + } + numEndpoints += testCase.progress[i] + } + endpoints := make([]*flight.FlightEndpoint, numEndpoints) + for i := range endpoints { + endpoints[i] = &flight.FlightEndpoint{ + Ticket: &flight.Ticket{ + Ticket: []byte{}, + }, + } + } + + query.nextIndex++ + pollInfo := flight.PollInfo{ + Info: &flight.FlightInfo { + Schema: schema, + Endpoint: endpoints, + }, + FlightDescriptor: &flight.FlightDescriptor{ + Type: flight.DescriptorCMD, + Cmd: descriptor, + }, + Progress: proto.Float64(float64(query.nextIndex) / float64(len(testCase.progress))), + } + + if query.nextIndex >= len(testCase.progress) { + if testCase.completeLazily { + if query.nextIndex == len(testCase.progress) { + // Make the client poll one more time + } else { + pollInfo.FlightDescriptor = nil + delete(srv.queries, queryId) + } + + } else { + pollInfo.FlightDescriptor = nil + delete(srv.queries, queryId) + } + } + + fmt.Printf("Returning %d endpoints, has retry? %t\n", numEndpoints, pollInfo.FlightDescriptor != nil) + + return &pollInfo, nil +} + +type IncrementalPollTestCase struct { + // on each poll (including the first), this many new endpoints complete + // making 0 progress is allowed, but not recommended (allow clients to 'long poll') + progress []int + + // use a different retry descriptor for each poll + differentRetryDescriptor bool + + // require one extra poll to get completion (i.e. the last poll will have a nil FlightInfo) + completeLazily bool +} + +type IncrementalPollTests struct { + ServerBasedTests + testCases map[string]IncrementalPollTestCase +} + +func (suite *IncrementalPollTests) SetupSuite() { + suite.testCases = map[string]IncrementalPollTestCase{ + "basic": { + progress: []int{1, 1, 1, 1}, + }, + "basic 2": { + progress: []int{2, 3, 4, 5}, + }, + "basic 3": { + progress: []int{2}, + }, + "descriptor changes": { + progress: []int{1, 1, 1, 1}, + differentRetryDescriptor: true, + }, + "lazy": { + progress: []int{1, 1, 1, 1}, + completeLazily: true, + }, + "lazy 2": { + progress: []int{1, 1, 1, 0}, + completeLazily: true, + }, + "no progress": { + progress: []int{0, 1, 1, 1}, + }, + "no progress 2": { + progress: []int{0, 0, 1, 1}, + }, + "no progress 3": { + progress: []int{0, 0, 1, 0}, + }, + } + + srv := IncrementalPollTestServer{ + queries: make(map[string]*IncrementalQuery), + testCases: suite.testCases, + } + srv.Alloc = memory.DefaultAllocator + suite.DoSetupSuite(&srv, nil, nil) +} + +func (ts *IncrementalPollTests) TestMaxProgress() { + stmt, err := ts.cnxn.NewStatement() + ts.NoError(err) + defer stmt.Close() + opts := stmt.(adbc.GetSetOptions) + + val, err := opts.GetOptionDouble(adbc.OptionKeyMaxProgress) + ts.NoError(err) + ts.Equal(1.0, val) +} + +func (ts *IncrementalPollTests) TestOptionValue() { + stmt, err := ts.cnxn.NewStatement() + ts.NoError(err) + defer stmt.Close() + opts := stmt.(adbc.GetSetOptions) + + val, err := opts.GetOption(adbc.OptionKeyIncremental) + ts.NoError(err) + ts.Equal(adbc.OptionValueDisabled, val) + + ts.NoError(stmt.SetOption(adbc.OptionKeyIncremental, adbc.OptionValueEnabled)) + + val, err = opts.GetOption(adbc.OptionKeyIncremental) + ts.NoError(err) + ts.Equal(adbc.OptionValueEnabled, val) + + var adbcErr adbc.Error + ts.ErrorAs(stmt.SetOption(adbc.OptionKeyIncremental, "foobar"), &adbcErr) + ts.Equal(adbc.StatusInvalidArgument, adbcErr.Code) +} + +func (ts *IncrementalPollTests) RunOneTestCase(ctx context.Context, stmt adbc.Statement, name string, testCase *IncrementalPollTestCase) { + opts := stmt.(adbc.GetSetOptions) + ts.NoError(stmt.SetSqlQuery(name)) + + for idx, progress := range testCase.progress { + fmt.Printf("Poll %d/%d\n", idx + 1, len(testCase.progress)) + if progress == 0 { + // the driver hides this from us + continue + } + + _, partitions, _, err := stmt.ExecutePartitions(ctx) + ts.NoError(err) + + ts.Equal(uint64(progress), partitions.NumPartitions) + + val, err := opts.GetOptionDouble(adbc.OptionKeyProgress) + ts.NoError(err) + ts.Equal(float64(idx + 1) / float64(len(testCase.progress)), val) + } + + fmt.Println("Poll last") + // Query completed, but we find out by getting no partitions in this call + _, partitions, _, err := stmt.ExecutePartitions(ctx) + ts.NoError(err) + + ts.Equal(uint64(0), partitions.NumPartitions) +} + +func (ts *IncrementalPollTests) TestQuery() { + ctx := context.Background() + for name, testCase := range ts.testCases { + ts.Run(name, func() { + stmt, err := ts.cnxn.NewStatement() + ts.NoError(err) + defer stmt.Close() + + ts.NoError(stmt.SetOption(adbc.OptionKeyIncremental, adbc.OptionValueEnabled)) + + // Run the query multiple times (we should be able to reuse the statement) + for i := 0; i < 2; i++ { + ts.RunOneTestCase(ctx, stmt, name, &testCase) + } + }) + } +} + +func (ts *IncrementalPollTests) TestQueryPrepared() { +} + +func (ts *IncrementalPollTests) TestQueryPreparedTransaction() { +} + +func (ts *IncrementalPollTests) TestQueryTransaction() { +} + +func (ts *IncrementalPollTests) TestStatementReuse() { +} + // ---- Timeout Tests -------------------- type TimeoutTestServer struct { diff --git a/go/adbc/driver/flightsql/flightsql_adbc_test.go b/go/adbc/driver/flightsql/flightsql_adbc_test.go index 1619f8fabb..163d03d9b2 100644 --- a/go/adbc/driver/flightsql/flightsql_adbc_test.go +++ b/go/adbc/driver/flightsql/flightsql_adbc_test.go @@ -42,12 +42,12 @@ import ( "github.com/apache/arrow-adbc/go/adbc" driver "github.com/apache/arrow-adbc/go/adbc/driver/flightsql" "github.com/apache/arrow-adbc/go/adbc/validation" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql/example" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql/example" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "google.golang.org/grpc" diff --git a/go/adbc/driver/flightsql/flightsql_connection.go b/go/adbc/driver/flightsql/flightsql_connection.go index ee24b75ee2..dbb05fe7f2 100644 --- a/go/adbc/driver/flightsql/flightsql_connection.go +++ b/go/adbc/driver/flightsql/flightsql_connection.go @@ -27,12 +27,12 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/internal" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql/schema_ref" - "github.com/apache/arrow/go/v14/arrow/ipc" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql/schema_ref" + "github.com/apache/arrow/go/v15/arrow/ipc" "github.com/bluele/gcache" "google.golang.org/grpc" grpccodes "google.golang.org/grpc/codes" @@ -886,6 +886,14 @@ func (c *cnxn) executeSubstraitUpdate(ctx context.Context, plan flightsql.Substr return c.cl.ExecuteSubstraitUpdate(ctx, plan, opts...) } +func (c *cnxn) poll(ctx context.Context, query string,retryDescriptor *flight.FlightDescriptor, opts ...grpc.CallOption) (*flight.PollInfo, error) { + // if c.txn != nil { + // return c.txn.Execute(ctx, query, opts...) + // } + + return c.cl.ExecutePoll(ctx, query, retryDescriptor, opts...) +} + func (c *cnxn) prepare(ctx context.Context, query string, opts ...grpc.CallOption) (*flightsql.PreparedStatement, error) { if c.txn != nil { return c.txn.Prepare(ctx, query, opts...) diff --git a/go/adbc/driver/flightsql/flightsql_database.go b/go/adbc/driver/flightsql/flightsql_database.go index 8b6ab2ccb9..91d40e7c8c 100644 --- a/go/adbc/driver/flightsql/flightsql_database.go +++ b/go/adbc/driver/flightsql/flightsql_database.go @@ -30,9 +30,9 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/driverbase" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" "github.com/bluele/gcache" "google.golang.org/grpc" "google.golang.org/grpc/credentials" diff --git a/go/adbc/driver/flightsql/flightsql_driver.go b/go/adbc/driver/flightsql/flightsql_driver.go index 0060c04055..b50097e586 100644 --- a/go/adbc/driver/flightsql/flightsql_driver.go +++ b/go/adbc/driver/flightsql/flightsql_driver.go @@ -38,7 +38,7 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/driverbase" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory" "golang.org/x/exp/maps" "google.golang.org/grpc/metadata" ) diff --git a/go/adbc/driver/flightsql/flightsql_statement.go b/go/adbc/driver/flightsql/flightsql_statement.go index 9edcf439a5..da28c290a6 100644 --- a/go/adbc/driver/flightsql/flightsql_statement.go +++ b/go/adbc/driver/flightsql/flightsql_statement.go @@ -22,13 +22,14 @@ import ( "fmt" "strconv" "strings" + "time" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/bluele/gcache" "google.golang.org/grpc" "google.golang.org/grpc/metadata" @@ -109,6 +110,19 @@ func (s *sqlOrSubstrait) executeUpdate(ctx context.Context, cnxn *cnxn, opts ... } } +func (s *sqlOrSubstrait) poll(ctx context.Context, cnxn *cnxn, retryDescriptor *flight.FlightDescriptor, opts ...grpc.CallOption) (*flight.PollInfo, error) { + if s.sqlQuery != "" { + return cnxn.poll(ctx, s.sqlQuery, retryDescriptor, opts...) + } // else if s.substraitPlan != nil { + // return cnxn.executeSubstrait(ctx, flightsql.SubstraitPlan{Plan: s.substraitPlan, Version: s.substraitVersion}, opts...) + // } + + return nil, adbc.Error{ + Code: adbc.StatusInvalidState, + Msg: "[Flight SQL] cannot call ExecuteQuery without a query or prepared statement", + } +} + func (s *sqlOrSubstrait) prepare(ctx context.Context, cnxn *cnxn, opts ...grpc.CallOption) (*flightsql.PreparedStatement, error) { if s.sqlQuery != "" { return cnxn.prepare(ctx, s.sqlQuery, opts...) @@ -122,6 +136,13 @@ func (s *sqlOrSubstrait) prepare(ctx context.Context, cnxn *cnxn, opts ...grpc.C } } +type incrementalState struct { + schema *arrow.Schema + previousInfo *flight.FlightInfo + retryDescriptor *flight.FlightDescriptor + complete bool +} + type statement struct { alloc memory.Allocator cnxn *cnxn @@ -132,6 +153,9 @@ type statement struct { prepared *flightsql.PreparedStatement queueSize int timeouts timeoutOption + // TODO: this needs to be cleared at appropriate times + incrementalState *incrementalState + progress float64 } func (s *statement) closePreparedStatement() error { @@ -173,6 +197,11 @@ func (s *statement) GetOption(key string) (string, error) { return s.timeouts.queryTimeout.String(), nil case OptionTimeoutUpdate: return s.timeouts.updateTimeout.String(), nil + case adbc.OptionKeyIncremental: + if s.incrementalState != nil { + return adbc.OptionValueEnabled, nil + } + return adbc.OptionValueDisabled, nil } if strings.HasPrefix(key, OptionRPCCallHeaderPrefix) { @@ -221,6 +250,10 @@ func (s *statement) GetOptionDouble(key string) (float64, error) { return s.timeouts.queryTimeout.Seconds(), nil case OptionTimeoutUpdate: return s.timeouts.updateTimeout.Seconds(), nil + case adbc.OptionKeyProgress: + return s.progress, nil + case adbc.OptionKeyMaxProgress: + return 1.0, nil } return 0, adbc.Error{ @@ -260,6 +293,18 @@ func (s *statement) SetOption(key string, val string) error { return s.SetOptionInt(key, int64(size)) case OptionStatementSubstraitVersion: s.query.substraitVersion = val + case adbc.OptionKeyIncremental: + switch val { + case adbc.OptionValueEnabled: + s.incrementalState = &incrementalState{} + case adbc.OptionValueDisabled: + s.incrementalState = nil + default: + return adbc.Error{ + Msg: fmt.Sprintf("[Flight SQL] Invalid statement option value %s=%s", key, val), + Code: adbc.StatusInvalidArgument, + } + } default: return adbc.Error{ Msg: "[Flight SQL] Unknown statement option '" + key + "'", @@ -485,13 +530,98 @@ func (s *statement) ExecutePartitions(ctx context.Context) (*arrow.Schema, adbc. var ( info *flight.FlightInfo + poll *flight.PollInfo out adbc.Partitions sc *arrow.Schema err error ) var header, trailer metadata.MD - if s.prepared != nil { + if s.incrementalState != nil { + if s.incrementalState.complete { + fmt.Println("complete, reset") + schema := s.incrementalState.schema + totalRecords := s.incrementalState.previousInfo.TotalRecords + // Reset the statement for reuse + s.incrementalState = &incrementalState{} + s.progress = 0.0 + return schema, adbc.Partitions{}, totalRecords, nil + } + + if s.prepared != nil { + return nil, out, -1, adbc.Error{ + Msg: "[Flight SQL] Cannot enable incremental execution for a prepared statement", + Code: adbc.StatusNotImplemented, + } + } else { + backoff := 100 * time.Millisecond + for { + fmt.Println("polling") + // Keep polling until the query completes or we get new partitions + poll, err = s.query.poll(ctx, s.cnxn, s.incrementalState.retryDescriptor, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts) + if err != nil { + break + } + info = poll.GetInfo() + if info == nil { + // The server is misbehaving + // XXX: should we also issue a query cancellation? + s.incrementalState = &incrementalState{} + s.progress = 0.0 + return nil, adbc.Partitions{}, -1, adbc.Error{ + Msg: "[Flight SQL] Server returned a PollInfo with no FlightInfo", + Code: adbc.StatusInternal, + } + } + info = &flight.FlightInfo{ + Schema: info.Schema, + FlightDescriptor: info.FlightDescriptor, + Endpoint: info.Endpoint, + TotalRecords: info.TotalRecords, + TotalBytes: info.TotalBytes, + Ordered: info.Ordered, + AppMetadata: info.AppMetadata, + } + // We only return the new endpoints each time + if s.incrementalState.previousInfo != nil { + offset := len(s.incrementalState.previousInfo.Endpoint) + if offset >= len(info.Endpoint) { + info.Endpoint = []*flight.FlightEndpoint{} + } else { + info.Endpoint = info.Endpoint[offset:] + } + } + s.incrementalState.previousInfo = poll.GetInfo() + s.incrementalState.retryDescriptor = poll.GetFlightDescriptor() + s.progress = poll.GetProgress() + + if s.incrementalState.retryDescriptor == nil { + // Query is finished + fmt.Println("query completed") + s.incrementalState.complete = true + break + } else if len(info.Endpoint) > 0 { + // Query made progress + fmt.Printf("query made progress %d\n", len(info.Endpoint)) + break + } + // Back off before next poll + time.Sleep(backoff) + backoff *= 2 + if backoff > 5000 * time.Millisecond { + backoff = 5000 * time.Millisecond + } + } + + // Special case: the query completed but there were no new endpoints. We + // return 0 new partitions, and also reset the statement (because + // returning 0 partitions implies completion) + if s.incrementalState.complete && len(info.Endpoint) == 0 { + s.incrementalState = &incrementalState{} + s.progress = 0.0 + } + } + } else if s.prepared != nil { info, err = s.prepared.Execute(ctx, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts) } else { info, err = s.query.execute(ctx, s.cnxn, grpc.Header(&header), grpc.Trailer(&trailer), s.timeouts) @@ -508,6 +638,10 @@ func (s *statement) ExecutePartitions(ctx context.Context) (*arrow.Schema, adbc. } } + if s.incrementalState != nil { + s.incrementalState.schema = sc + } + out.NumPartitions = uint64(len(info.Endpoint)) out.PartitionIDs = make([][]byte, out.NumPartitions) for i, e := range info.Endpoint { diff --git a/go/adbc/driver/flightsql/record_reader.go b/go/adbc/driver/flightsql/record_reader.go index 9a10e3708d..c2ae474a41 100644 --- a/go/adbc/driver/flightsql/record_reader.go +++ b/go/adbc/driver/flightsql/record_reader.go @@ -24,11 +24,11 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/utils" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/bluele/gcache" "golang.org/x/sync/errgroup" "google.golang.org/grpc" diff --git a/go/adbc/driver/flightsql/record_reader_test.go b/go/adbc/driver/flightsql/record_reader_test.go index 28c9046275..83104e63fd 100644 --- a/go/adbc/driver/flightsql/record_reader_test.go +++ b/go/adbc/driver/flightsql/record_reader_test.go @@ -24,12 +24,12 @@ import ( "testing" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/ipc" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/ipc" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/bluele/gcache" "github.com/stretchr/testify/suite" "google.golang.org/grpc" diff --git a/go/adbc/driver/internal/shared_utils.go b/go/adbc/driver/internal/shared_utils.go index 3a5791192c..52d74d9ca7 100644 --- a/go/adbc/driver/internal/shared_utils.go +++ b/go/adbc/driver/internal/shared_utils.go @@ -26,9 +26,9 @@ import ( "time" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/memory" ) type CatalogAndSchema struct { diff --git a/go/adbc/driver/panicdummy/panicdummy_adbc.go b/go/adbc/driver/panicdummy/panicdummy_adbc.go index f0513cd6f3..47b1460e34 100644 --- a/go/adbc/driver/panicdummy/panicdummy_adbc.go +++ b/go/adbc/driver/panicdummy/panicdummy_adbc.go @@ -25,9 +25,9 @@ import ( "os" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/memory" ) func maybePanic(fname string) { diff --git a/go/adbc/driver/snowflake/connection.go b/go/adbc/driver/snowflake/connection.go index 73c31604de..f96aeb437c 100644 --- a/go/adbc/driver/snowflake/connection.go +++ b/go/adbc/driver/snowflake/connection.go @@ -30,8 +30,8 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/internal" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" "github.com/snowflakedb/gosnowflake" ) diff --git a/go/adbc/driver/snowflake/driver.go b/go/adbc/driver/snowflake/driver.go index db5efed456..e880c04fb5 100644 --- a/go/adbc/driver/snowflake/driver.go +++ b/go/adbc/driver/snowflake/driver.go @@ -24,7 +24,7 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/driverbase" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/snowflakedb/gosnowflake" "golang.org/x/exp/maps" ) diff --git a/go/adbc/driver/snowflake/driver_test.go b/go/adbc/driver/snowflake/driver_test.go index 61e944866d..f1bf87a90e 100644 --- a/go/adbc/driver/snowflake/driver_test.go +++ b/go/adbc/driver/snowflake/driver_test.go @@ -37,10 +37,10 @@ import ( "github.com/apache/arrow-adbc/go/adbc/driver/internal" driver "github.com/apache/arrow-adbc/go/adbc/driver/snowflake" "github.com/apache/arrow-adbc/go/adbc/validation" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/decimal128" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/decimal128" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/google/uuid" "github.com/snowflakedb/gosnowflake" "github.com/stretchr/testify/require" diff --git a/go/adbc/driver/snowflake/record_reader.go b/go/adbc/driver/snowflake/record_reader.go index c77b578ffc..44057d4ee0 100644 --- a/go/adbc/driver/snowflake/record_reader.go +++ b/go/adbc/driver/snowflake/record_reader.go @@ -27,11 +27,11 @@ import ( "time" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/compute" - "github.com/apache/arrow/go/v14/arrow/ipc" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/compute" + "github.com/apache/arrow/go/v15/arrow/ipc" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/snowflakedb/gosnowflake" "golang.org/x/sync/errgroup" ) diff --git a/go/adbc/driver/snowflake/statement.go b/go/adbc/driver/snowflake/statement.go index 57ac079d44..edbaa826a4 100644 --- a/go/adbc/driver/snowflake/statement.go +++ b/go/adbc/driver/snowflake/statement.go @@ -25,9 +25,9 @@ import ( "strings" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/snowflakedb/gosnowflake" "golang.org/x/exp/constraints" ) diff --git a/go/adbc/drivermgr/wrapper.go b/go/adbc/drivermgr/wrapper.go index 07bb94b814..9294716567 100644 --- a/go/adbc/drivermgr/wrapper.go +++ b/go/adbc/drivermgr/wrapper.go @@ -43,9 +43,9 @@ import ( "unsafe" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/cdata" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/cdata" ) type option struct { diff --git a/go/adbc/drivermgr/wrapper_sqlite_test.go b/go/adbc/drivermgr/wrapper_sqlite_test.go index c33adf2792..4fa516501c 100644 --- a/go/adbc/drivermgr/wrapper_sqlite_test.go +++ b/go/adbc/drivermgr/wrapper_sqlite_test.go @@ -27,9 +27,9 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/drivermgr" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" diff --git a/go/adbc/go.mod b/go/adbc/go.mod index 69e9eaf591..6754457344 100644 --- a/go/adbc/go.mod +++ b/go/adbc/go.mod @@ -20,7 +20,7 @@ module github.com/apache/arrow-adbc/go/adbc go 1.19 require ( - github.com/apache/arrow/go/v14 v14.0.0 + github.com/apache/arrow/go/v15 v15.0.0 github.com/bluele/gcache v0.0.2 github.com/golang/protobuf v1.5.3 github.com/google/uuid v1.3.1 @@ -105,3 +105,4 @@ require ( ) replace github.com/snowflakedb/gosnowflake => github.com/snowflakedb/gosnowflake v1.6.23-0.20231106142408-8445dcaf46d9 +replace github.com/apache/arrow/go/v15 => /home/lidavidm/Code/arrow/go \ No newline at end of file diff --git a/go/adbc/go.sum b/go/adbc/go.sum index d8e09e3ffb..8f73a46f10 100644 --- a/go/adbc/go.sum +++ b/go/adbc/go.sum @@ -17,8 +17,8 @@ github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/ github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= github.com/apache/arrow/go/v12 v12.0.1 h1:JsR2+hzYYjgSUkBSaahpqCetqZMr76djX80fF/DiJbg= github.com/apache/arrow/go/v12 v12.0.1/go.mod h1:weuTY7JvTG/HDPtMQxEUp7pU73vkLWMLpY67QwZ/WWw= -github.com/apache/arrow/go/v14 v14.0.0 h1:NXfgmvrHAWSzPO1YNjDhO9VwYrUQI/kRvzy5dJuCIaY= -github.com/apache/arrow/go/v14 v14.0.0/go.mod h1:u3fgh3EdgN/YQ8cVQRguVW3R+seMybFg8QBQ5LU+eBY= +github.com/apache/arrow/go/v15 v14.0.0 h1:NXfgmvrHAWSzPO1YNjDhO9VwYrUQI/kRvzy5dJuCIaY= +github.com/apache/arrow/go/v15 v14.0.0/go.mod h1:u3fgh3EdgN/YQ8cVQRguVW3R+seMybFg8QBQ5LU+eBY= github.com/apache/thrift v0.17.0 h1:cMd2aj52n+8VoAtvSvLn4kDC3aZ6IAkBuqWQ2IDu7wo= github.com/apache/thrift v0.17.0/go.mod h1:OLxhMRJxomX+1I/KUw03qoV3mMz16BwaKI+d4fPBx7Q= github.com/aws/aws-sdk-go-v2 v1.19.0 h1:klAT+y3pGFBU/qVf1uzwttpBbiuozJYWzNLHioyDJ+k= diff --git a/go/adbc/pkg/_tmpl/driver.go.tmpl b/go/adbc/pkg/_tmpl/driver.go.tmpl index 4b7008ea9e..b87856d27c 100644 --- a/go/adbc/pkg/_tmpl/driver.go.tmpl +++ b/go/adbc/pkg/_tmpl/driver.go.tmpl @@ -58,10 +58,10 @@ import ( "unsafe" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/cdata" - "github.com/apache/arrow/go/v14/arrow/memory" - "github.com/apache/arrow/go/v14/arrow/memory/mallocator" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/cdata" + "github.com/apache/arrow/go/v15/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory/mallocator" "golang.org/x/exp/slog" ) diff --git a/go/adbc/pkg/flightsql/driver.go b/go/adbc/pkg/flightsql/driver.go index 2847274c30..619f9d6295 100644 --- a/go/adbc/pkg/flightsql/driver.go +++ b/go/adbc/pkg/flightsql/driver.go @@ -61,10 +61,10 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/flightsql" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/cdata" - "github.com/apache/arrow/go/v14/arrow/memory" - "github.com/apache/arrow/go/v14/arrow/memory/mallocator" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/cdata" + "github.com/apache/arrow/go/v15/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory/mallocator" "golang.org/x/exp/slog" ) diff --git a/go/adbc/pkg/panicdummy/driver.go b/go/adbc/pkg/panicdummy/driver.go index 399d0edcdb..2a1ea4b2ed 100644 --- a/go/adbc/pkg/panicdummy/driver.go +++ b/go/adbc/pkg/panicdummy/driver.go @@ -61,10 +61,10 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/panicdummy" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/cdata" - "github.com/apache/arrow/go/v14/arrow/memory" - "github.com/apache/arrow/go/v14/arrow/memory/mallocator" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/cdata" + "github.com/apache/arrow/go/v15/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory/mallocator" "golang.org/x/exp/slog" ) diff --git a/go/adbc/pkg/snowflake/driver.go b/go/adbc/pkg/snowflake/driver.go index b591018106..c13bd50a74 100644 --- a/go/adbc/pkg/snowflake/driver.go +++ b/go/adbc/pkg/snowflake/driver.go @@ -61,10 +61,10 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/driver/snowflake" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/cdata" - "github.com/apache/arrow/go/v14/arrow/memory" - "github.com/apache/arrow/go/v14/arrow/memory/mallocator" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/cdata" + "github.com/apache/arrow/go/v15/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory/mallocator" "golang.org/x/exp/slog" ) diff --git a/go/adbc/sqldriver/driver.go b/go/adbc/sqldriver/driver.go index 775f3f78a8..abc8404111 100644 --- a/go/adbc/sqldriver/driver.go +++ b/go/adbc/sqldriver/driver.go @@ -31,11 +31,11 @@ import ( "unsafe" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/decimal128" - "github.com/apache/arrow/go/v14/arrow/decimal256" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/decimal128" + "github.com/apache/arrow/go/v15/arrow/decimal256" + "github.com/apache/arrow/go/v15/arrow/memory" ) func getIsolationlevel(lvl sql.IsolationLevel) adbc.OptionIsolationLevel { diff --git a/go/adbc/sqldriver/driver_internals_test.go b/go/adbc/sqldriver/driver_internals_test.go index 9981a40ddb..c7418b69a1 100644 --- a/go/adbc/sqldriver/driver_internals_test.go +++ b/go/adbc/sqldriver/driver_internals_test.go @@ -26,11 +26,11 @@ import ( "time" "github.com/apache/arrow-adbc/go/adbc" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/decimal128" - "github.com/apache/arrow/go/v14/arrow/decimal256" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/decimal128" + "github.com/apache/arrow/go/v15/arrow/decimal256" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) diff --git a/go/adbc/sqldriver/flightsql/flightsql.go b/go/adbc/sqldriver/flightsql/flightsql.go index 6ad00e9c20..23425fbe6c 100644 --- a/go/adbc/sqldriver/flightsql/flightsql.go +++ b/go/adbc/sqldriver/flightsql/flightsql.go @@ -22,7 +22,7 @@ import ( "github.com/apache/arrow-adbc/go/adbc/driver/flightsql" "github.com/apache/arrow-adbc/go/adbc/sqldriver" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/memory" ) func init() { diff --git a/go/adbc/sqldriver/flightsql/flightsql_test.go b/go/adbc/sqldriver/flightsql/flightsql_test.go index 7d0b86b23a..4c0b7f68b3 100644 --- a/go/adbc/sqldriver/flightsql/flightsql_test.go +++ b/go/adbc/sqldriver/flightsql/flightsql_test.go @@ -23,10 +23,10 @@ import ( "testing" _ "github.com/apache/arrow-adbc/go/adbc/sqldriver/flightsql" - "github.com/apache/arrow/go/v14/arrow/flight" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql" - "github.com/apache/arrow/go/v14/arrow/flight/flightsql/example" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow/flight" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql" + "github.com/apache/arrow/go/v15/arrow/flight/flightsql/example" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/stretchr/testify/suite" "google.golang.org/grpc" ) diff --git a/go/adbc/standard_schemas.go b/go/adbc/standard_schemas.go index 264f76140c..a441ae0f47 100644 --- a/go/adbc/standard_schemas.go +++ b/go/adbc/standard_schemas.go @@ -17,7 +17,7 @@ package adbc -import "github.com/apache/arrow/go/v14/arrow" +import "github.com/apache/arrow/go/v15/arrow" var ( GetInfoSchema = arrow.NewSchema([]arrow.Field{ diff --git a/go/adbc/utils/utils.go b/go/adbc/utils/utils.go index e1c372c32b..4dd69216b3 100644 --- a/go/adbc/utils/utils.go +++ b/go/adbc/utils/utils.go @@ -17,7 +17,7 @@ package utils -import "github.com/apache/arrow/go/v14/arrow" +import "github.com/apache/arrow/go/v15/arrow" func RemoveSchemaMetadata(schema *arrow.Schema) *arrow.Schema { fields := make([]arrow.Field, len(schema.Fields())) diff --git a/go/adbc/validation/validation.go b/go/adbc/validation/validation.go index 8925f72532..1572ed74e3 100644 --- a/go/adbc/validation/validation.go +++ b/go/adbc/validation/validation.go @@ -28,9 +28,9 @@ import ( "github.com/apache/arrow-adbc/go/adbc" "github.com/apache/arrow-adbc/go/adbc/utils" - "github.com/apache/arrow/go/v14/arrow" - "github.com/apache/arrow/go/v14/arrow/array" - "github.com/apache/arrow/go/v14/arrow/memory" + "github.com/apache/arrow/go/v15/arrow" + "github.com/apache/arrow/go/v15/arrow/array" + "github.com/apache/arrow/go/v15/arrow/memory" "github.com/stretchr/testify/suite" ) diff --git a/r/adbcsnowflake/tools/create-go-vendor-archive.R b/r/adbcsnowflake/tools/create-go-vendor-archive.R index a9045b0598..aa8ba7959a 100644 --- a/r/adbcsnowflake/tools/create-go-vendor-archive.R +++ b/r/adbcsnowflake/tools/create-go-vendor-archive.R @@ -41,7 +41,7 @@ withr::with_dir("src/go/adbc", { "arrow-go-v14.0.2/go/arrow/cdata/arrow" ) - dst_go_arrow_cdata_dir <- "vendor/github.com/apache/arrow/go/v14/arrow/cdata/" + dst_go_arrow_cdata_dir <- "vendor/github.com/apache/arrow/go/v15/arrow/cdata/" stopifnot(file.copy(src_go_arrow_cdata_arrow_dir, dst_go_arrow_cdata_dir, recursive = TRUE)) })