feat(storage/dataflux): add dataflux interface

storage/dataflux/doc.go

@@ -0,0 +1,27 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/*
+Package dataflux provides an easy way to parallelize listing in Google
+Cloud Storage.
+
+More information about Google Cloud Storage is available at
+https://cloud.google.com/storage/docs.
+
+See https://pkg.go.dev/cloud.google.com/go for authentication, timeouts,
+connection pooling and similar aspects of this package.
+
+NOTE: This package is in preview. It is not stable, and is likely to change.
+*/
+package dataflux // import "cloud.google.com/go/storage/dataflux"

storage/dataflux/fast_list.go

@@ -0,0 +1,95 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dataflux
+
+import (
+	"context"
+
+	"cloud.google.com/go/storage"
+)
+
+// listingMethod represents the method of listing.
+type listingMethod int
+
+const (
+	// open when any method can be used to list.
+	open listingMethod = iota
+	// sequential when the listing is done sequentially.
+	sequential
+	// worksteal when the listing is done using work stealing algorithm.
+	worksteal
+)
+
+// Input contains options for listing objects.
+type Input struct {
+	// Client is the storage client to list objects from.
+	Client *storage.Client
+	// BucketName is the name of the bucket to list objects from.
+	BucketName string
+	// Parallelism is number of parallel workers to use for listing.
+	Parallelism int
+	// pageSize is the number of objects to list.
+	BatchSize int
+	// query is the query to filter objects for listing.
+	Query storage.Query
+	// SkipDirectoryObjects is to indicate whether to list directory objects.
+	SkipDirectoryObjects bool
+}
+
+// Lister is used for interacting with Dataflux fast-listing.
+// The caller should initialize it with NewLister() instead of creating it directly.
+type Lister struct {
+	// method indicates the listing method(open, sequential, worksteal) to be used for listing.
+	method listingMethod
+	// pageToken is the token to use for sequential listing.
+	pageToken string
+	// ranges is the channel to store the start and end ranges to be listed by the workers in worksteal listing.
+	ranges chan *listRange
+	// bucket is the bucket handle to list objects from.
+	bucket *storage.BucketHandle
+	// parallelism is number of parallel workers to use for listing.
+	parallelism int
+	// batchSize is the number of objects to list.
+	batchSize int
+	// query is the query to filter objects for listing.
+	query storage.Query
+	// skipDirectoryObjects is to indicate whether to list directory objects.
+	skipDirectoryObjects bool
+}
+
+// CloseFunc is the function to close the range channel of a Lister.
+type CloseFunc func()
+
+// NewLister creates a new dataflux Lister to list objects in the give bucket.
+func NewLister(opts *Input) (*Lister, CloseFunc) {
+
+	lister := &Lister{}
+	return lister, func() { lister.Close() }
+}
+
+// NextBatch runs worksteal algorithm and sequential listing in parallel to quickly
+// return a list of objects in the bucket. For buckets with smaller dataset,
+// sequential listing is expected to be faster. For buckets with larger dataset,
+// worksteal listing is expected to be faster.
+func (c *Lister) NextBatch(ctx context.Context) ([]*storage.ObjectAttrs, error) {
+	return nil, nil
+}
+
+// Close closes the range channel of the Lister.
+func (c *Lister) Close() {
+	if c.ranges != nil {
+		close(c.ranges)
+	}
+}

storage/dataflux/object_lister.go

@@ -0,0 +1,62 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dataflux
+
+import (
+	"context"
+
+	"cloud.google.com/go/storage"
+)
+
+// newObjectListerOpts specifies options for instantiating the NewObjectLister.
+type newObjectListerOpts struct {
+	// startRange is the start offset of the objects to be listed.
+	startRange string
+	// endRange is the end offset of the objects to be listed.
+	endRange string
+	// bucketHandle is the bucket handle of the bucket to be listed.
+	bucketHandle *storage.BucketHandle
+	// query is the storage.Query to filter objects for listing.
+	query storage.Query
+	// skipDirectoryObjects is to indicate whether to list directory objects.
+	skipDirectoryObjects bool
+	// generation is the generation number of the last object in the page.
+	generation int64
+}
+
+// nextPageResult holds the next page of object names and indicates whether the
+// lister has completed listing (no more objects to retrieve).
+type nextPageResult struct {
+	// items is the list of objects listed.
+	items []*storage.ObjectAttrs
+	// doneListing indicates whether the lister has completed listing.
+	doneListing bool
+	// nextStartRange is the start offset of the next page of objects to be listed.
+	nextStartRange string
+	// generation is the generation number of the last object in the page.
+	generation int64
+}
+
+// newObjectLister creates a new ObjectLister using the given lister options.
+func newObjectLister(ctx context.Context, opts newObjectListerOpts) (*nextPageResult, error) {
+	return &nextPageResult{}, nil
+}
+
+func addPrefix(name, prefix string) string {
+	if name != "" {
+		return prefix + name
+	}
+	return name
+}

storage/dataflux/range_splitter.go

@@ -0,0 +1,43 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dataflux
+
+import (
+	"sync"
+)
+
+// rangeSplitter specifies the a list and a map of sorted alphabets.
+type rangeSplitter struct {
+	mu             sync.Mutex
+	sortedAlphabet *[]rune
+	alphabetMap    map[rune]int
+}
+
+// listRange specifies the start and end range for the range splitter.
+type listRange struct {
+	startRange string
+	endRange   string
+}
+
+// newRangeSplitter creates a new RangeSplitter with the given alphabets.
+func newRangeSplitter(alphabet string) (*rangeSplitter, error) {
+
+	return &rangeSplitter{}, nil
+}
+
+// splitRange creates a given number of splits based on a provided start and end range.
+func (rs *rangeSplitter) splitRange(startRange, endRange string, numSplits int) ([]string, error) {
+	return nil, nil
+}

storage/dataflux/sequential.go

@@ -0,0 +1,28 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dataflux
+
+import (
+	"context"
+
+	"cloud.google.com/go/storage"
+)
+
+// Listing performs a sequential listing on the given bucket.
+// It returns a list of objects and the next token to use to continue listing.
+// If the next token is empty, then listing is complete.
+func Listing(ctx context.Context, opts Lister) ([]*storage.ObjectAttrs, string, error) {
+	return nil, "", nil
+}

storage/dataflux/worksteal.go

@@ -0,0 +1,54 @@
+// Copyright 2024 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dataflux
+
+import (
+	"context"
+	"sync"
+
+	"cloud.google.com/go/storage"
+)
+
+// WorkerStatus indicates the status of a worker.
+type WorkerStatus int
+
+const (
+	// Idle status shows that the worker is currently not listing.
+	Idle WorkerStatus = iota
+	// Active status shows that the worker is currently listing objects within assigned range.
+	Active
+)
+
+type listerResult struct {
+	mu      sync.Mutex
+	objects []*storage.ObjectAttrs
+}
+
+type worker struct {
+	goroutineID   int
+	startRange    string
+	endRange      string
+	status        WorkerStatus
+	rangesplitter *rangeSplitter
+	idleChannel   chan int
+	result        *listerResult
+	generation    int64
+}
+
+// workstealListing is the main entry point of the worksteal algorithm.
+// It performs worksteal to achieve highly dynamic object listing.
+func workstealListing(ctx context.Context, opts Lister) ([]*storage.ObjectAttrs, error) {
+	return nil, nil
+}