Resync with influxdata/flux 0.194.4

.circleci/config.yml

@@ -78,7 +78,9 @@ jobs:
           keys:
             - flux-gomod-{{checksum "go.sum"}}
       # Run tests
-      - run: make test-bench
+      - run:
+          command: make test-bench
+          no_output_timeout: 20m
       # No need to save the pkg/mod cache since the other job does it
   test-build-reproducibility:
     docker:
@@ -105,9 +107,10 @@ jobs:
             git clone https://github.com/influxdata/rustup.install.git "${TMP}/rustup.install"
             cd "${TMP}/rustup.install"
               git checkout bump-v1.25.1
-              choco install -dvf ./rustup.install.nuspec
+              choco pack --pre ./rustup.install.nuspec
+              choco install rustup.install --source="C:\Users\circleci\AppData\Local\Temp\rustup.install"
             cd -
-            choco upgrade golang --version=1.18 --allow-downgrade
+            choco upgrade golang --version=1.21 --allow-downgrade
 
             choco install \
               grep \
@@ -185,7 +188,8 @@ jobs:
 
   build-ci-image:
     machine:
-      image: ubuntu-2204:2022.04.1
+      image: ubuntu-2204:current
+      resource_class: medium
     steps:
       - checkout
       - run:

Dockerfile_build

@@ -4,14 +4,14 @@
 # and verification, we can list the rust container as a prior build stage, and
 # then pull in the artifacts we need. There is an added benefit that tagged versions
 # also include minor releases, so 1.2 includes 1.2.1 and so on, for bugfix releases.
-FROM rust:1.68 as RUSTBUILD
+FROM rust:1.78 as RUSTBUILD
 
-FROM golang:1.19 as PKGCONFIG
+FROM golang:1.21 as PKGCONFIG
 COPY go.mod go.sum /go/src/github.com/influxdata/flux/
 RUN cd /go/src/github.com/influxdata/flux && \
     go build -o /usr/local/bin/cgo-pkgbuild github.com/influxdata/pkg-config
 
-FROM golang:1.19
+FROM golang:1.21
 
 # Install common packages
 RUN apt-get update && \

README.md

@@ -1,3 +1,5 @@
+:warning: InfluxData still support Flux, but we are not doing active development on it or adding any new features to it. We still do security patches and will address any critical defects through the maintenance period. Our focus now is on features related to our latest database engine, InfluxDB 3.0, and its associated products (Cloud Serverless, Cloud Dedicated, Clustered, Edge). There is a community owned fork of [Flux](https://github.com/InfluxCommunity/flux) where active development is happening.
+
 # Flux - Influx data language
 
 [![CircleCI](https://circleci.com/gh/influxdata/flux/tree/master.svg?style=svg)](https://circleci.com/gh/influxdata/flux/tree/master)

array/array.go

@@ -2,11 +2,10 @@ package array
 
 import (
 	"strconv"
-	"sync/atomic"
 
 	"github.com/apache/arrow/go/v7/arrow"
 	"github.com/apache/arrow/go/v7/arrow/array"
-	arrowmem "github.com/apache/arrow/go/v7/arrow/memory"
+
 	"github.com/influxdata/flux/codes"
 	"github.com/influxdata/flux/internal/errors"
 	"github.com/influxdata/flux/memory"
@@ -103,10 +102,23 @@ type Builder interface {
 	NewArray() Array
 }
 
+type binaryArray interface {
+	NullN() int
+	NullBitmapBytes() []byte
+	IsNull(i int) bool
+	IsValid(i int) bool
+	Data() arrow.ArrayData
+	Len() int
+	ValueBytes() []byte
+	ValueLen(i int) int
+	ValueOffset(i int) int
+	ValueString(i int) string
+	Retain()
+	Release()
+}
+
 type String struct {
-	length int
-	data   *array.Binary
-	value  *stringValue
+	binaryArray
 }
 
 // NewStringFromBinaryArray creates an instance of String from
@@ -118,140 +130,36 @@ type String struct {
 func NewStringFromBinaryArray(data *array.Binary) *String {
 	data.Retain()
 	return &String{
-		data: data,
+		binaryArray: data,
 	}
 }
 
 func (a *String) DataType() DataType {
 	return StringType
 }
-func (a *String) NullN() int {
-	if a.data != nil {
-		return a.data.NullN()
-	}
-	return 0
-}
-func (a *String) NullBitmapBytes() []byte {
-	if a.data != nil {
-		return a.data.NullBitmapBytes()
-	}
-	return nil
-}
-func (a *String) IsNull(i int) bool {
-	if a.data != nil {
-		return a.data.IsNull(i)
-	}
-	return false
-}
-func (a *String) IsValid(i int) bool {
-	if a.data != nil {
-		return a.data.IsValid(i)
-	}
-	return true
-}
-func (a *String) Data() arrow.ArrayData {
-	if a.data != nil {
-		return a.data.Data()
-	}
-	return nil
-}
-func (a *String) Len() int {
-	if a.data != nil {
-		return a.data.Len()
-	}
-	return a.length
-}
-func (a *String) Retain() {
-	if a.data != nil {
-		a.data.Retain()
-		return
-	}
-	a.value.Retain()
-}
-func (a *String) Release() {
-	if a.data != nil {
-		a.data.Release()
-		return
-	}
-	a.value.Release()
-}
+
 func (a *String) Slice(i, j int) Array {
-	if a.data != nil {
-		data := array.NewSliceData(a.data.Data(), int64(i), int64(j))
-		defer data.Release()
-		return &String{
-			data: array.NewBinaryData(data),
-		}
+	slice, ok := a.binaryArray.(interface{ Slice(i, j int) binaryArray })
+	if ok {
+		return &String{binaryArray: slice.Slice(i, j)}
 	}
-	a.value.Retain()
+	data := array.NewSliceData(a.binaryArray.Data(), int64(i), int64(j))
+	defer data.Release()
 	return &String{
-		value:  a.value,
-		length: j - i,
+		binaryArray: array.NewBinaryData(data),
 	}
 }
 
-// ValueBytes returns a byte slice containing the value of this string
-// at index i. This slice points to the contents of the data buffer and
-// is only valid for the lifetime of the array.
-func (a *String) ValueBytes(i int) []byte {
-	if a.data != nil {
-		return a.data.Value(i)
-	}
-	return a.value.Bytes()
-}
-
-// Value returns a string copy of the value stored at index i. The
-// returned value will outlive the array and is safe to use like any
-// other go string. The memory backing the string will be allocated by
-// the runtime, rather than any provided allocator.
+// Value returns a string view of the bytes in the array. The string
+// is only valid for the lifetime of the array. Care should be taken not
+// to store this string without also retaining the array.
 func (a *String) Value(i int) string {
-	return string(a.ValueBytes(i))
-}
-func (a *String) ValueLen(i int) int {
-	if a.data != nil {
-		return a.data.ValueLen(i)
-	}
-	return a.value.Len()
-}
-func (a *String) IsConstant() bool {
-	return a.data == nil
-}
-
-type stringValue struct {
-	rc   int64
-	data []byte
-
-	mem arrowmem.Allocator
+	return a.ValueString(i)
 }
 
-func (v *stringValue) Retain() {
-	if v == nil {
-		return
-	}
-	atomic.AddInt64(&v.rc, 1)
-}
-
-func (v *stringValue) Release() {
-	if v == nil {
-		return
-	}
-	if atomic.AddInt64(&v.rc, -1) == 0 {
-		v.mem.Free(v.data)
-	}
-}
-
-func (v *stringValue) Bytes() []byte {
-	if v == nil {
-		return nil
-	}
-	return v.data
-}
-
-func (v *stringValue) Len() int {
-	if v == nil {
-		return 0
-	}
-	return len(v.data)
+func (a *String) IsConstant() bool {
+	ic, ok := a.binaryArray.(interface{ IsConstant() bool })
+	return ok && ic.IsConstant()
 }
 
 type sliceable interface {

array/array_test.go

@@ -15,20 +15,30 @@ func TestString(t *testing.T) {
 	for _, tc := range []struct {
 		name  string
 		build func(b *array.StringBuilder)
+		bsz   int
 		sz    int
 		want  []interface{}
 	}{
 		{
 			name: "Constant",
 			build: func(b *array.StringBuilder) {
 				for i := 0; i < 10; i++ {
-					b.Append("a")
+					b.Append("abcdefghij")
 				}
 			},
-			sz: 1,
+			bsz: 64, // 64 bytes data.
+			sz:  64, // The minimum size of a buffer is 64 bytes
 			want: []interface{}{
-				"a", "a", "a", "a", "a",
-				"a", "a", "a", "a", "a",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
+				"abcdefghij",
 			},
 		},
 		{
@@ -41,7 +51,8 @@ func TestString(t *testing.T) {
 					b.Append("b")
 				}
 			},
-			sz: 192,
+			bsz: 192,
+			sz:  192,
 			want: []interface{}{
 				"a", "a", "a", "a", "a",
 				"b", "b", "b", "b", "b",
@@ -58,7 +69,8 @@ func TestString(t *testing.T) {
 					b.Append(v)
 				}
 			},
-			sz: 192,
+			bsz: 192,
+			sz:  192,
 			want: []interface{}{
 				"a", "b", "c", "d", "e",
 				nil, "g", "h", "i", "j",
@@ -90,7 +102,7 @@ func TestString(t *testing.T) {
 			if want, got := len(tc.want)+2, b.Cap(); want != got {
 				t.Errorf("unexpected builder cap -want/+got:\n\t- %d\n\t+ %d", want, got)
 			}
-			assert.Equal(t, tc.sz, mem.CurrentAlloc(), "unexpected memory allocation.")
+			assert.Equal(t, tc.bsz, mem.CurrentAlloc(), "unexpected memory allocation.")
 
 			arr := b.NewStringArray()
 			defer arr.Release()
@@ -165,7 +177,7 @@ func TestStringBuilder_NewArray(t *testing.T) {
 		}
 
 		arr := b.NewArray()
-		assert.Equal(t, 1, mem.CurrentAlloc(), "unexpected memory allocation.")
+		assert.Equal(t, 64, mem.CurrentAlloc(), "unexpected memory allocation.")
 		arr.Release()
 		mem.AssertSize(t, 0)
 

array/binary.go

@@ -155,11 +155,11 @@ func StringAdd(l, r *String, mem memory.Allocator) (*String, error) {
 	b.Resize(n)
 	for i := 0; i < n; i++ {
 		if l.IsValid(i) && r.IsValid(i) {
-			lb := l.ValueBytes(i)
-			rb := r.ValueBytes(i)
-			buf := make([]byte, len(lb)+len(rb))
-			copy(buf, lb)
-			copy(buf[len(lb):], rb)
+			ls := l.Value(i)
+			rs := r.Value(i)
+			buf := make([]byte, len(ls)+len(rs))
+			copy(buf, ls)
+			copy(buf[len(ls):], rs)
 			b.AppendBytes(buf)
 
 		} else {
@@ -177,10 +177,10 @@ func StringAddLConst(l string, r *String, mem memory.Allocator) (*String, error)
 	b.Resize(n)
 	for i := 0; i < n; i++ {
 		if r.IsValid(i) {
-			rb := r.ValueBytes(i)
-			buf := make([]byte, len(l)+len(rb))
+			rs := r.Value(i)
+			buf := make([]byte, len(l)+len(rs))
 			copy(buf, l)
-			copy(buf[len(l):], rb)
+			copy(buf[len(l):], rs)
 			b.AppendBytes(buf)
 
 		} else {
@@ -198,10 +198,10 @@ func StringAddRConst(l *String, r string, mem memory.Allocator) (*String, error)
 	b.Resize(n)
 	for i := 0; i < n; i++ {
 		if l.IsValid(i) {
-			lb := l.ValueBytes(i)
-			buf := make([]byte, len(lb)+len(r))
-			copy(buf, lb)
-			copy(buf[len(lb):], r)
+			ls := l.Value(i)
+			buf := make([]byte, len(ls)+len(r))
+			copy(buf, ls)
+			copy(buf[len(ls):], r)
 			b.AppendBytes(buf)
 
 		} else {