GH-39238:[Go] PATCH Prevents empty record to be appended to empty res…

…ultset (#39239)

### Rationale for this change
When having an empty resultset, the driver tries to include an empty record referece, that cannot be scanned.

So, any operation that relies on the returned Row(s) will trigger a "Index out of Range" error.

### What changes are included in this PR?
We're preventing to include an invalid record (that can't be scanned) in an empty resultset

### Are these changes tested?
Yes, there's a new test included

### Are there any user-facing changes?
No

**This PR contains a "Critical Fix".**
* Closes: #39238

Authored-by: miguel pragier <[email protected]>
Signed-off-by: Matt Topol <[email protected]>

go/arrow/flight/flightsql/driver/driver.go

@@ -487,9 +487,10 @@ func readEndpoint(ctx context.Context, client *flightsql.Client, endpoint *fligh
 	schema := reader.Schema()
 	var records []arrow.Record
 	for reader.Next() {
-		record := reader.Record()
-		record.Retain()
-		records = append(records, record)
+		if record := reader.Record(); record.NumRows() > 0 {
+			record.Retain()
+			records = append(records, record)
+		}
 	}
 
 	if err := reader.Err(); err != nil && !errors.Is(err, io.EOF) {

go/arrow/flight/flightsql/driver/driver_test.go

@@ -273,6 +273,50 @@ func (s *SqlTestSuite) TestQuery() {
 	wg.Wait()
 }
 
+func (s *SqlTestSuite) TestQueryWithEmptyResultset() {
+	t := s.T()
+
+	// Create and start the server
+	server, addr, err := s.createServer()
+	require.NoError(t, err)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		require.NoError(s.T(), s.startServer(server))
+	}()
+	defer s.stopServer(server)
+	time.Sleep(100 * time.Millisecond)
+
+	// Configure client
+	cfg := s.Config
+	cfg.Address = addr
+	db, err := sql.Open("flightsql", cfg.DSN())
+	require.NoError(t, err)
+	defer db.Close()
+
+	// Create the table
+	_, err = db.Exec(fmt.Sprintf(s.Statements["create table"], s.TableName))
+	require.NoError(t, err)
+
+	rows, err := db.Query(fmt.Sprintf(s.Statements["query"], s.TableName))
+	require.NoError(t, err)
+	require.False(t, rows.Next())
+
+	row := db.QueryRow(fmt.Sprintf(s.Statements["query"], s.TableName))
+	require.NotNil(t, row)
+	require.NoError(t, row.Err())
+
+	target := make(map[string]any)
+	err = row.Scan(&target)
+	require.ErrorIs(t, err, sql.ErrNoRows)
+
+	// Tear-down server
+	s.stopServer(server)
+	wg.Wait()
+}
+
 func (s *SqlTestSuite) TestPreparedQuery() {
 	t := s.T()