feat: Change Iterator base on google guideline (#131)

.golangci.yml

@@ -80,7 +80,7 @@ issues:
   # Excluding configuration per-path, per-linter, per-text and per-source
   exclude-rules:
     # Some code is just exceptional. Review whenever the code is changed.
-    - path: influxdb3/query_iterator\.go
+    - path: influxdb3/query_iterator\.go,influxdb3/point_value_iterator\.go
       linters:
         - gocyclo
     # Exclude some linters from running on tests files.

CHANGELOG.md

@@ -12,6 +12,11 @@
 
 1. [#127](https://github.com/InfluxCommunity/influxdb3-go/pull/127): LPBatcher now returns first line of the internal buffer when the line length exceeds the batch size.
 
+### Features
+
+1. [#131](https://github.com/InfluxCommunity/influxdb3-go/pull/131): Add new PointValueIterator based on google
+   guidelines [Guidelines](https://github.com/googleapis/google-cloud-go/wiki/Iterator-Guidelines)
+
 ## 2.0.0 [2024-12-13]
 
 ### Breaking Changes

influxdb3/client_e2e_test.go

@@ -27,6 +27,7 @@ package influxdb3_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"math"
@@ -239,6 +240,112 @@ func TestQueryWithParameters(t *testing.T) {
 	assert.True(t, iterator.Done())
 }
 
+func TestQueryPointValue(t *testing.T) {
+	SkipCheck(t)
+	now := time.Now().UTC()
+	testId := now.UnixNano()
+
+	url := os.Getenv("TESTING_INFLUXDB_URL")
+	token := os.Getenv("TESTING_INFLUXDB_TOKEN")
+	database := os.Getenv("TESTING_INFLUXDB_DATABASE")
+
+	client, err := influxdb3.New(influxdb3.ClientConfig{
+		Host:     url,
+		Token:    token,
+		Database: database,
+	})
+	require.NoError(t, err)
+	defer client.Close()
+
+	p := influxdb3.NewPointWithMeasurement("weather5").
+		SetField("text", "a1").
+		SetField("testId", testId).
+		SetTimestamp(now)
+	err = client.WritePoints(context.Background(), []*influxdb3.Point{p})
+	require.NoError(t, err)
+	query := fmt.Sprintf(`
+			SELECT *
+				FROM weather5
+			WHERE
+				time >= now() - interval '10 minute'
+			AND
+			    "testId" = %d
+  `, testId)
+
+	sleepTime := 5 * time.Second
+	time.Sleep(sleepTime)
+
+	pointValueIterator, err := client.QueryPointValue(context.Background(), query)
+	require.NoError(t, err)
+	require.NotNil(t, pointValueIterator)
+
+	PointValue, err := pointValueIterator.Next()
+	assert.NoError(t, err)
+	assert.NotNil(t, PointValue)
+	assert.Equal(t, PointValue.GetField("text"), "a1")
+
+	PointValue, err = pointValueIterator.Next()
+	assert.Equal(t, influxdb3.Done, errors.New("no more items in iterator"))
+	assert.Nil(t, PointValue)
+}
+
+func TestQueryPointValueWithParameters(t *testing.T) {
+	SkipCheck(t)
+	now := time.Now().UTC()
+	testId := now.UnixNano()
+
+	url := os.Getenv("TESTING_INFLUXDB_URL")
+	token := os.Getenv("TESTING_INFLUXDB_TOKEN")
+	database := os.Getenv("TESTING_INFLUXDB_DATABASE")
+
+	client, err := influxdb3.New(influxdb3.ClientConfig{
+		Host:     url,
+		Token:    token,
+		Database: database,
+	})
+	require.NoError(t, err)
+	defer client.Close()
+
+	p := influxdb3.NewPointWithMeasurement("weather5").
+		SetField("text", "a1").
+		SetField("testId", testId).
+		SetTimestamp(now)
+	err = client.WritePoints(context.Background(), []*influxdb3.Point{p})
+	require.NoError(t, err)
+
+	query := `
+		SELECT *
+		FROM weather5
+		WHERE
+		time >= now() - interval '10 minute'
+		AND
+		text = $text
+		AND
+		"testId" = $testId
+		ORDER BY time
+	`
+	parameters := influxdb3.QueryParameters{
+		"text":   "a1",
+		"testId": testId,
+	}
+
+	sleepTime := 5 * time.Second
+	time.Sleep(sleepTime)
+
+	pointValueIterator, err := client.QueryPointValueWithParameters(context.Background(), query, parameters)
+	require.NoError(t, err)
+	require.NotNil(t, pointValueIterator)
+
+	PointValue, err := pointValueIterator.Next()
+	assert.NoError(t, err)
+	assert.NotNil(t, PointValue)
+	assert.Equal(t, PointValue.GetField("text"), "a1")
+
+	PointValue, err = pointValueIterator.Next()
+	assert.Equal(t, influxdb3.Done, errors.New("no more items in iterator"))
+	assert.Nil(t, PointValue)
+}
+
 func TestQueryDatabaseDoesNotExist(t *testing.T) {
 	SkipCheck(t)
 	url := os.Getenv("TESTING_INFLUXDB_URL")

influxdb3/point_value_iterator.go

@@ -0,0 +1,91 @@
+/*
+ The MIT License
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy
+ of this software and associated documentation files (the "Software"), to deal
+ in the Software without restriction, including without limitation the rights
+ to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ copies of the Software, and to permit persons to whom the Software is
+ furnished to do so, subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in
+ all copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ THE SOFTWARE.
+*/
+
+package influxdb3
+
+import (
+	"errors"
+
+	"github.com/apache/arrow/go/v15/arrow"
+	"github.com/apache/arrow/go/v15/arrow/flight"
+)
+
+var Done = errors.New("no more items in iterator") //nolint:all
+
+// PointValueIterator is a custom query iterator that encapsulates and simplifies the logic for
+// the flight reader. It provides method Next to consume the flight reader,
+//
+// The PointValueIterator Next function will return response as a *PointValues object representing the current row
+type PointValueIterator struct {
+	reader *flight.Reader
+	// Index of row of current object in current record
+	index int
+	// Current record
+	record arrow.Record
+}
+
+// Return a new PointValueIterator
+func newPointValueIterator(reader *flight.Reader) *PointValueIterator {
+	return &PointValueIterator{
+		reader: reader,
+		index:  -1,
+		record: nil,
+	}
+}
+
+// Next returns the next result.
+// Its second return value is iterator.Done if there are no more results.
+// Once Next returns Done in the second parameter, all subsequent calls will return Done.
+//
+//	it := newPointValueIterator(flightReader)
+//	for {
+//		PointValue, err := it.Next()
+//		if err == iterator.Done {
+//			break
+//		}
+//		if err != nil {
+//			return err
+//		}
+//		process(PointValue)
+//	}
+func (it *PointValueIterator) Next() (*PointValues, error) {
+	it.index++
+
+	for it.record == nil || it.index >= int(it.record.NumRows()) {
+		if !it.reader.Next() {
+			if err := it.reader.Err(); err != nil {
+				return nil, err
+			}
+			return nil, Done
+		}
+		it.record = it.reader.Record()
+		it.index = 0
+	}
+
+	pointValues := rowToPointValue(it.reader.Record(), it.index)
+	return pointValues, nil
+}
+
+// Index return the current index of PointValueIterator
+func (it *PointValueIterator) Index() int {
+	return it.index
+}