fgo

package
v0.1.0-beta.9 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 1, 2026 License: Apache-2.0 Imports: 30 Imported by: 0

Documentation

Overview

Package fgo provides the Apache Fluss 0.9.1 data client.

Open one Client for an application and derive table handles, writers, scanners, lookup clients, and administrative clients from it. A Client owns negotiated coordinator and tablet connections and is safe for concurrent use. Close child resources before calling Client.Close.

Tables and records

Client.OpenTable loads the authoritative Table and Schema. Log tables support LogWriter and LogScanner. Primary-key tables support KVWriter, LookupClient, BatchScanner, and snapshot scans. Row is the portable value representation; the generic typed wrappers adapt application values through explicit Codec and KeyCodec implementations.

Secure connections

WithTLSConfig enables certificate-verified TLS, while WithAuthenticator configures one authentication mechanism per connection. Use PlainAuthenticator with TLS when the server enables SASL PLAIN. The secure connection guide covers certificate, credential, and authentication error handling.

Cancellation and partial results

Blocking operations accept a context. Cancellation stops local waiting but cannot prove that an in-flight mutation was rejected by the server. Writers report ambiguous state through ErrWriterState. Multi-bucket reads preserve successful records alongside per-bucket errors. The error handling guide describes safe retries, writer reconciliation, and partial-result handling.

Arrow ownership

Arrow-backed APIs use explicit ownership. An input record passed to LogWriter.AppendArrow must remain valid until its WriteFuture completes. Call ScanResult.Release, BatchResult.Release, or ArrowLogBatch.Release for decoded records owned by returned results.

Errors and compatibility

Public errors support errors.Is and errors.As. The package negotiates protocol versions but intentionally supports only Apache Fluss 0.9.1. Later server versions require explicit protocol, fixture, and integration validation before they are supported.

The public Go API is experimental before v1. Applications should pin a release and review the project changelog before upgrading.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrClosed         = errors.New("fgo: client closed")
	ErrUnsupportedAPI = errors.New("fgo: server does not support API")
	ErrInvalidConfig  = errors.New("fgo: invalid client configuration")
	ErrAuthentication = errors.New("fgo: authentication failed")
)

Client lifecycle, protocol support, configuration, and authentication errors.

View Source 
var (
	ErrServerFailure = errors.New("fgo: server failure")
	ErrAuthorization = errors.New("fgo: authorization failure")
	ErrMetadata      = errors.New("fgo: stale or invalid metadata")
	ErrTimeout       = errors.New("fgo: server timeout")
	ErrStorage       = errors.New("fgo: storage failure")
	ErrSequence      = errors.New("fgo: sequence failure")
	ErrRecord        = errors.New("fgo: record failure")
	ErrValidation    = errors.New("fgo: validation failure")
)

Server-side error classifications used with errors.Is.

View Source 
var (
	ErrUnknownTable     = errors.New("fgo: unknown table")
	ErrUnknownPartition = errors.New("fgo: unknown partition")
	ErrUnknownBucket    = errors.New("fgo: unknown bucket")
	ErrNoBucketLeader   = errors.New("fgo: bucket has no tablet leader")
)

Metadata lookup and routing errors.

View Source 
var (
	ErrInvalidSchema = errors.New("fgo: invalid schema")
	ErrInvalidRow    = errors.New("fgo: invalid row")
	ErrTableKind     = errors.New("fgo: unsupported table operation")
)

Schema, row, and table-kind validation errors.

View Source 
var (
	ErrFileSystemSecurityToken       = errors.New("fgo: filesystem security token unavailable")
	ErrSecurityTokenProviderDisabled = errors.New("fgo: filesystem security token provider disabled")
)

Filesystem security-token availability errors.

View Source 
var ErrMalformedRecordBatch = errors.New("fgo: malformed record batch")

ErrMalformedRecordBatch reports an invalid KV, row, or Arrow batch encoding.

View Source 
var ErrMalformedRow = errors.New("fgo: malformed row encoding")

ErrMalformedRow reports an invalid compacted or indexed row encoding.

View Source 
var ErrNotFound = fmt.Errorf("fgo: record not found")

ErrNotFound reports a missing primary-key row.

View Source 
var ErrServerRole = errors.New("fgo: unexpected server role")

ErrServerRole reports that a connection negotiated an unexpected server role.

View Source 
var ErrWakeup = errors.New("fgo: log scanner wakeup")

ErrWakeup reports that LogScanner.Wakeup interrupted a poll.

View Source 
var ErrWriterState = errors.New("fgo: writer state is uncertain")

ErrWriterState reports that cancellation or transport failure left the server outcome of a mutation unknown.

Functions

func EncodeArrowLogBatch 

func EncodeArrowLogBatch(batch ArrowLogBatch, compression ArrowCompression, allocator memory.Allocator) ([]byte, error)

EncodeArrowLogBatch encodes an Arrow-backed Fluss log batch. The function borrows batch.Record only for the duration of the call.

func EncodeCompactedProjectedRow 

func EncodeCompactedProjectedRow(schema Schema, columns []string, row Row) ([]byte, error)

EncodeCompactedProjectedRow encodes values for the named columns in the given order. It is used by Fluss partial updates, where omitted columns must not be serialized as nulls.

func EncodeCompactedRow 

func EncodeCompactedRow(schema Schema, row Row) ([]byte, error)

EncodeCompactedRow encodes Fluss's compacted binary row format. The returned buffer is owned by the caller and remains valid after the input row is reused.

func EncodeIndexedProjectedRow 

func EncodeIndexedProjectedRow(schema Schema, columns []string, row Row) ([]byte, error)

EncodeIndexedProjectedRow encodes indexed values for columns.

func EncodeIndexedRow 

func EncodeIndexedRow(schema Schema, row Row) ([]byte, error)

EncodeIndexedRow encodes Fluss's indexed binary row format.

func EncodeLookupKey 

func EncodeLookupKey(schema Schema, key PrimaryKey, version int16) ([]byte, error)

EncodeLookupKey selects the key contract negotiated for Lookup v0 or v1. Fluss 0.9.1 uses the same compacted key bytes in both versions; keeping the version explicit prevents silent use of a future incompatible layout.

func EncodePrefixKey 

func EncodePrefixKey(schema Schema, prefix PrimaryKey) ([]byte, error)

EncodePrefixKey writes a non-empty leading subset of the primary key for PrefixLookup v0 and v1.

func EncodePrefixLookupKey 

func EncodePrefixLookupKey(schema Schema, prefix PrimaryKey, version int16) ([]byte, error)

EncodePrefixLookupKey selects the negotiated PrefixLookup key contract.

func EncodePrimaryKey 

func EncodePrimaryKey(schema Schema, key PrimaryKey) ([]byte, error)

EncodePrimaryKey writes the compacted key representation used by Lookup v0 and v1. Primary-key columns have no null bitmap and must all be non-null.

func ResponseError 

func ResponseError(code int32, message string, api fmsg.APIKey) error

ResponseError converts an error embedded in a successful Fluss response body. Most APIs report errors in the RPC envelope; multi-resource APIs use this helper for their per-resource fields.

Types

type ArrowCompression 

type ArrowCompression uint8

ArrowCompression identifies the Arrow IPC compression codec. 

const (
	ArrowCompressionNone ArrowCompression = iota
	ArrowCompressionLZ4
	ArrowCompressionZSTD
)

Arrow compression codecs supported by Apache Fluss 0.9.1.

type ArrowLogBatch 

type ArrowLogBatch struct {
	// Magic selects the Fluss v0 or v1 batch header.
	Magic byte
	// BaseOffset is the first row offset.
	BaseOffset int64
	// CommitTime is the server commit time in Unix milliseconds.
	CommitTime int64
	// LeaderEpoch is present only for magic 1.
	LeaderEpoch int32
	// SchemaID identifies the Arrow schema.
	SchemaID int16
	// AppendOnly omits per-row change bytes when true.
	AppendOnly bool
	// WriterID identifies the idempotent writer session.
	WriterID int64
	// BatchSequence is monotonic within WriterID and bucket.
	BatchSequence int32
	// Record is borrowed during encoding and owned after decoding.
	Record arrow.RecordBatch
	// Changes has one entry per row unless AppendOnly is true.
	Changes []ChangeType
	// contains filtered or unexported fields
}

ArrowLogBatch is the Arrow-backed variant of a Fluss log batch. EncodeArrowLogBatch only borrows Record for the duration of the call. DecodeArrowLogBatch returns an owned Record that remains valid until Release is called.

func DecodeArrowLogBatch 

func DecodeArrowLogBatch(schema *arrow.Schema, encoded []byte, allocator memory.Allocator) (*ArrowLogBatch, error)

DecodeArrowLogBatch decodes a Fluss Arrow log batch. The caller owns the returned batch and must call ArrowLogBatch.Release.

func (*ArrowLogBatch) Release 

func (b *ArrowLogBatch) Release()

Release releases a record owned by a decoded batch. It is safe to call more than once.

type AuthenticationError 

type AuthenticationError struct {
	// Err is the underlying mechanism or transport failure.
	Err error
	// Retriable reports whether a fresh connection may repeat authentication.
	Retriable bool
}

AuthenticationError reports whether a failed authentication exchange may be retried on a new connection. Its message intentionally never includes authentication tokens or credentials.

Example 
package main

import (
	"errors"
	"fmt"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	err := &fgo.AuthenticationError{
		Err:       errors.New("server rejected authentication"),
		Retriable: true,
	}
	var authentication *fgo.AuthenticationError
	fmt.Println(errors.Is(err, fgo.ErrAuthentication))
	fmt.Println(errors.As(err, &authentication), authentication.Retriable)
	fmt.Println(err)
}

Output:
true
true true
fgo: authentication failed (retriable)

func (*AuthenticationError) Error 

func (e *AuthenticationError) Error() string

Error returns a credential-safe authentication summary.

func (*AuthenticationError) Is 

func (e *AuthenticationError) Is(target error) bool

Is reports whether target is ErrAuthentication.

func (*AuthenticationError) Unwrap 

func (e *AuthenticationError) Unwrap() error

Unwrap returns the underlying authentication failure.

type Authenticator 

type Authenticator interface {
	// Protocol returns the SASL mechanism name sent to Fluss.
	Protocol() string
	// HasInitialResponse reports whether Authenticate should run before the
	// first server challenge.
	HasInitialResponse() bool
	// Authenticate consumes one challenge and returns the next response.
	// Implementations must not retain or expose challenge or response bytes.
	Authenticate(context.Context, []byte) ([]byte, error)
	// Complete reports whether the exchange has reached a terminal success
	// state.
	Complete() bool
	// Close releases mechanism-specific state and secret material.
	Close() error
}

Authenticator performs one Fluss Authenticate challenge exchange. An instance belongs to one server connection and must not be shared by concurrent connections.

type AuthenticatorFactory 

type AuthenticatorFactory func() (Authenticator, error)

AuthenticatorFactory creates a fresh authenticator for each server connection.

func PlainAuthenticator 

func PlainAuthenticator(username, password string) AuthenticatorFactory

PlainAuthenticator returns a factory for the SASL PLAIN mechanism supported by Fluss 0.9.1. Each connection receives its own credentials buffer, which is cleared when the connection closes.

type BatchResult 

type BatchResult struct {
	// Rows contains row-oriented results.
	Rows []Row
	// ArrowBatches contains owned Arrow results released by [BatchResult.Release].
	ArrowBatches []ArrowLogBatch
	// Done reports that no additional rows remain.
	Done bool
}

BatchResult owns any Arrow batches it contains. Call Release after consuming the result.

func (*BatchResult) Release 

func (r *BatchResult) Release()

Release frees Arrow records owned by the result. Release is safe to call more than once.

type BatchScanner 

type BatchScanner struct {
	// contains filtered or unexported fields
}

BatchScanner reads the bounded current state or one immutable snapshot of a table bucket. Poll calls are serialized; call Close to interrupt an active poll and release scanner resources.

func (*BatchScanner) Close 

func (s *BatchScanner) Close() error

Close stops the scanner and cancels an active poll. Close is idempotent.

func (*BatchScanner) Done 

func (s *BatchScanner) Done() bool

Done reports whether the scanner reached its configured bound.

func (*BatchScanner) Poll 

func (s *BatchScanner) Poll(ctx context.Context) (BatchResult, error)

Poll returns available rows. A completed scanner keeps returning Done without further I/O.

type BatchScannerConfig 

type BatchScannerConfig struct {
	// Limit is the maximum rows requested per poll.
	Limit int
	// Projection lists returned columns in result order; nil selects all.
	Projection []string
}

BatchScannerConfig contains limits and projection for a bounded scan.

type BatchScannerOption 

type BatchScannerOption func(*BatchScannerConfig) error

BatchScannerOption configures a bounded current-state or snapshot scan.

func WithBatchLimit 

func WithBatchLimit(limit int) BatchScannerOption

WithBatchLimit bounds the number of rows returned by a current-state request or snapshot poll.

func WithBatchProjection 

func WithBatchProjection(columns ...string) BatchScannerOption

WithBatchProjection selects result columns in the requested order.

type BucketAssignment 

type BucketAssignment string

BucketAssignment selects how unkeyed log rows are routed to buckets. 

const (
	AssignmentAuto       BucketAssignment = "auto"
	AssignmentSticky     BucketAssignment = "sticky"
	AssignmentRoundRobin BucketAssignment = "round-robin"
)

Bucket assignment strategies supported by LogWriter.

type BucketScanError 

type BucketScanError struct {
	// Bucket identifies the failed table bucket.
	Bucket int32
	// Err is the bucket-local fetch or decode failure.
	Err error
}

BucketScanError reports a bucket-local failure in a partial scan result.

type ChangeType 

type ChangeType int8

ChangeType identifies the row-level change carried by a log record. 

const (
	Append ChangeType = iota
	Insert
	UpdateBefore
	UpdateAfter
	Delete

	// Upsert is the idempotent update-after change used by primary-key tables.
	Upsert = UpdateAfter
)

Change types encoded by Apache Fluss 0.9.1.

func (ChangeType) Validate 

func (c ChangeType) Validate() error

Validate reports an error for an unsupported change type.

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client owns negotiated coordinator and tablet connections. A Client is safe for concurrent use. Close child resources before closing it.

func Open 

func Open(ctx context.Context, options ...Option) (*Client, error)

Open connects to a coordinator, negotiates protocol versions, and returns a shared client.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(
		ctx,
		fgo.WithSeedBrokers("coordinator.example:9123"),
		fgo.WithClientIdentity("orders-service", "1.0.0"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "orders",
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("opened %s with %d buckets", table.Path, table.BucketCount)
}
Example (TlsAndSASL) 
package main

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"log"
	"os"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	caPEM, err := os.ReadFile(os.Getenv("FLUSS_CA_FILE"))
	if err != nil {
		log.Fatal(err)
	}
	roots := x509.NewCertPool()
	if !roots.AppendCertsFromPEM(caPEM) {
		log.Fatal("Fluss CA file contains no certificates")
	}
	serverName := os.Getenv("FLUSS_TLS_SERVER_NAME")
	username := os.Getenv("FLUSS_SASL_USERNAME")
	password := os.Getenv("FLUSS_SASL_PASSWORD")
	if serverName == "" || username == "" || password == "" {
		log.Fatal("Fluss TLS and SASL configuration is incomplete")
	}

	client, err := fgo.Open(
		context.Background(),
		fgo.WithSeedBrokers("coordinator.example:9123"),
		fgo.WithTLSConfig(&tls.Config{
			MinVersion: tls.VersionTLS12,
			RootCAs:    roots,
			ServerName: serverName,
		}),
		fgo.WithAuthenticator(fgo.PlainAuthenticator(username, password)),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()
}

func (*Client) Close 

func (c *Client) Close() error

Close stops token refresh and closes all managed connections. Close is idempotent.

func (*Client) CurrentFileSystemSecurityToken 

func (c *Client) CurrentFileSystemSecurityToken() (FileSystemSecurityToken, bool)

CurrentFileSystemSecurityToken returns a clone of the current valid, non-revoked token.

func (*Client) NewBatchScanner 

func (c *Client) NewBatchScanner(
	ctx context.Context,
	table Table,
	bucket TableBucket,
	options ...BatchScannerOption,
) (*BatchScanner, error)

NewBatchScanner scans the current state of one table bucket with Fluss LIMIT_SCAN.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "customers",
	})
	if err != nil {
		log.Fatal(err)
	}
	buckets, err := client.ResolveTableBuckets(ctx, fgo.PhysicalTablePath{
		TablePath: table.Path,
	})
	if err != nil {
		log.Fatal(err)
	}
	if len(buckets) == 0 {
		log.Fatal("table has no buckets")
	}
	scanner, err := client.NewBatchScanner(
		ctx,
		table,
		buckets[0],
		fgo.WithBatchLimit(1_000),
		fgo.WithBatchProjection("customer_id", "name"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := scanner.Close(); err != nil {
			log.Printf("close batch scanner: %v", err)
		}
	}()

	for !scanner.Done() {
		result, err := scanner.Poll(ctx)
		if err != nil {
			log.Fatal(err)
		}
		for _, row := range result.Rows {
			log.Printf("row=%v", row)
		}
		result.Release()
	}
}

func (*Client) NewKVWriter 

func (c *Client) NewKVWriter(ctx context.Context, table Table, options ...KVWriterOption) (*KVWriter, error)

NewKVWriter creates a primary-key writer for table.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "customers",
	})
	if err != nil {
		log.Fatal(err)
	}
	writer, err := client.NewKVWriter(
		ctx,
		table,
		fgo.WithKVBatchLimits(1<<20, 500),
		fgo.WithKVMergeMode(fgo.MergeModeOverwrite),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := writer.Close(ctx); err != nil {
			log.Printf("close KV writer: %v", err)
		}
	}()

	upsert := writer.Upsert(ctx, fgo.Row{int64(42), "Ada"}).Await(ctx)
	if upsert.Err != nil {
		log.Fatal(upsert.Err)
	}
	deleted := writer.Delete(ctx, fgo.PrimaryKey{int64(42)}).Await(ctx)
	if deleted.Err != nil {
		log.Fatal(deleted.Err)
	}
	if err := writer.Flush(ctx); err != nil {
		log.Fatal(err)
	}
}

func (*Client) NewLogScanner 

func (c *Client) NewLogScanner(ctx context.Context, table Table, start ScanOffset, options ...LogScannerOption) (*LogScanner, error)

NewLogScanner creates a scanner subscribed to all current buckets at start.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "events",
	})
	if err != nil {
		log.Fatal(err)
	}
	scanner, err := client.NewLogScanner(
		ctx,
		table,
		fgo.Earliest(),
		fgo.WithScanProjection("event_id", "event_type"),
		fgo.WithScanRowLimit(100),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer scanner.Close()

	result, err := scanner.Poll(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer result.Release()
	for _, record := range result.Records {
		log.Printf(
			"bucket=%d offset=%d row=%v",
			record.Bucket,
			record.Record.Offset,
			record.Record.Value,
		)
	}
}

func (*Client) NewLogWriter 

func (c *Client) NewLogWriter(ctx context.Context, table Table, options ...LogWriterOption) (*LogWriter, error)

NewLogWriter creates an append writer for a log table.

Example 
package main

import (
	"context"
	"log"
	"time"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "events",
	})
	if err != nil {
		log.Fatal(err)
	}
	writer, err := client.NewLogWriter(
		ctx,
		table,
		fgo.WithLogBatchLimits(1<<20, 500),
		fgo.WithLogLinger(5*time.Millisecond),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := writer.Close(ctx); err != nil {
			log.Printf("close KV writer: %v", err)
		}
	}()

	result := writer.Append(ctx, fgo.Row{int64(42), "created"}).Await(ctx)
	if result.Err != nil {
		log.Fatal(result.Err)
	}
	log.Printf("appended at offset %d", result.BaseOffset)
}

func (*Client) NewLookupClient 

func (c *Client) NewLookupClient(ctx context.Context, table Table, options ...LookupOption) (*LookupClient, error)

NewLookupClient creates a point and prefix lookup client for table.

Example 
package main

import (
	"context"
	"errors"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "customers",
	})
	if err != nil {
		log.Fatal(err)
	}
	lookup, err := client.NewLookupClient(ctx, table, fgo.WithLookupBatch(100, 4))
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := lookup.Close(); err != nil {
			log.Printf("close lookup client: %v", err)
		}
	}()

	results := lookup.Lookup(
		ctx,
		fgo.PrimaryKey{int64(42)},
		fgo.PrimaryKey{int64(43)},
	)
	for _, result := range results {
		switch {
		case errors.Is(result.Err, fgo.ErrNotFound):
			log.Printf("lookup %v: not found", result.Key)
		case result.Err != nil:
			log.Printf("lookup %v: %v", result.Key, result.Err)
		default:
			log.Printf("lookup %v: %v", result.Key, result.Row)
		}
	}
}

func (*Client) NewSnapshotBatchScanner 

func (c *Client) NewSnapshotBatchScanner(
	ctx context.Context,
	table Table,
	bucket TableBucket,
	snapshotID int64,
	options ...BatchScannerOption,
) (*BatchScanner, error)

NewSnapshotBatchScanner scans one immutable primary-key snapshot through the configured provider.

Example 
package main

import (
	"context"
	"io"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

type exampleSnapshotReader struct{}

func (exampleSnapshotReader) ReadBatch(context.Context, int) ([]fgo.Row, error) {
	return nil, io.EOF
}

func (exampleSnapshotReader) Close() error { return nil }

func main() {
	ctx := context.Background()
	provider := fgo.SnapshotBatchProviderFunc(
		func(context.Context, fgo.SnapshotBatchRequest) (fgo.SnapshotBatchReader, error) {
			return exampleSnapshotReader{}, nil
		},
	)
	client, err := fgo.Open(
		ctx,
		fgo.WithSeedBrokers("coordinator.example:9123"),
		fgo.WithSnapshotBatchProvider(provider),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := client.Close(); err != nil {
			log.Printf("close Fluss client: %v", err)
		}
	}()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "customers",
	})
	if err != nil {
		log.Fatal(err)
	}
	buckets, err := client.ResolveTableBuckets(ctx, fgo.PhysicalTablePath{
		TablePath: table.Path,
	})
	if err != nil {
		log.Fatal(err)
	}
	if len(buckets) == 0 {
		log.Fatal("table has no buckets")
	}
	scanner, err := client.NewSnapshotBatchScanner(
		ctx,
		table,
		buckets[0],
		101,
		fgo.WithBatchLimit(1_000),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		if err := scanner.Close(); err != nil {
			log.Printf("close snapshot scanner: %v", err)
		}
	}()

	result, err := scanner.Poll(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer result.Release()
}

func (*Client) OpenTable 

func (c *Client) OpenTable(ctx context.Context, path TablePath) (Table, error)

OpenTable loads authoritative table metadata and schema from the coordinator.

func (*Client) Request 

func (c *Client) Request(ctx context.Context, request fmsg.Request) (fmsg.Response, error)

Request sends a coordinator-scoped protocol request.

func (*Client) RequestBucket 

func (c *Client) RequestBucket(ctx context.Context, path PhysicalTablePath, bucket int32, request fmsg.Request) (fmsg.Response, error)

RequestBucket sends a bucket-scoped request to its current tablet leader. A stale metadata response causes one cache invalidation and one bounded reroute.

func (*Client) RequestCoordinator 

func (c *Client) RequestCoordinator(ctx context.Context, request fmsg.Request) (fmsg.Response, error)

RequestCoordinator sends an administrative request to the currently advertised coordinator.

func (*Client) RequestTo 

func (c *Client) RequestTo(ctx context.Context, node Node, request fmsg.Request) (fmsg.Response, error)

RequestTo sends a raw request to the connection for node. It is intended for protocol helpers; higher-level clients select the appropriate coordinator or tablet server from metadata.

func (*Client) Requester 

func (c *Client) Requester() fmsg.Requester

Requester exposes the low-level protocol requester implemented by the client.

func (*Client) ResolveTableBuckets 

func (c *Client) ResolveTableBuckets(
	ctx context.Context,
	path PhysicalTablePath,
) ([]TableBucket, error)

ResolveTableBuckets returns a stable bucket-ID ordered snapshot of current tablet leaders.

type Codec 

type Codec[T any] interface {
	// Encode maps one application value to schema column order.
	Encode(T) (Row, error)
	// Decode maps one row to an application value without retaining the row.
	Decode(Row) (T, error)
}

Codec is the stable, reflection-free mapping between an application type and a Fluss row.

type CodecFuncs 

type CodecFuncs[T any] struct {
	// EncodeFunc implements [Codec.Encode].
	EncodeFunc func(T) (Row, error)
	// DecodeFunc implements [Codec.Decode].
	DecodeFunc func(Row) (T, error)
}

CodecFuncs adapts encode and decode functions to Codec.

Example 
package main

import (
	"fmt"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	type customer struct {
		ID   int64
		Name string
	}
	codec := fgo.CodecFuncs[customer]{
		EncodeFunc: func(value customer) (fgo.Row, error) {
			return fgo.Row{value.ID, value.Name}, nil
		},
		DecodeFunc: func(row fgo.Row) (customer, error) {
			return customer{ID: row[0].(int64), Name: row[1].(string)}, nil
		},
	}

	row, err := codec.Encode(customer{ID: 42, Name: "Ada"})
	if err != nil {
		log.Fatal(err)
	}
	decoded, err := codec.Decode(row)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("%d %s\n", decoded.ID, decoded.Name)
}

Output:
42 Ada

func (CodecFuncs[T]) Decode 

func (c CodecFuncs[T]) Decode(row Row) (T, error)

Decode calls DecodeFunc and rejects an unset function.

func (CodecFuncs[T]) Encode 

func (c CodecFuncs[T]) Encode(value T) (Row, error)

Encode calls EncodeFunc and rejects an unset function.

type Column 

type Column struct {
	// Name is unique within the enclosing schema.
	Name string
	// Type is the physical Fluss data type.
	Type DataType
	// Nullable reports whether nil is a valid column value.
	Nullable bool
	// LogicalType carries nested and precision metadata.
	LogicalType *LogicalType
	// Description is optional user metadata.
	Description string
	// ID is the stable Fluss field identifier.
	ID int
}

Column describes one ordered field in a table schema.

type DataType 

type DataType string

DataType is the portable root type of a Fluss column. 

const (
	BoolType         DataType = "BOOLEAN"
	CharType         DataType = "CHAR"
	IntType          DataType = "INT"
	TinyIntType      DataType = "TINYINT"
	SmallIntType     DataType = "SMALLINT"
	BigIntType       DataType = "BIGINT"
	FloatType        DataType = "FLOAT"
	DoubleType       DataType = "DOUBLE"
	StringType       DataType = "STRING"
	BinaryType       DataType = "BINARY"
	BytesType        DataType = "BYTES"
	DecimalType      DataType = "DECIMAL"
	DateType         DataType = "DATE"
	TimeType         DataType = "TIME_WITHOUT_TIME_ZONE"
	TimestampType    DataType = "TIMESTAMP"
	TimestampLTZType DataType = "TIMESTAMP_WITH_LOCAL_TIME_ZONE"
	ArrayType        DataType = "ARRAY"
	MapType          DataType = "MAP"
	RowType          DataType = "ROW"
)

Data types supported by Apache Fluss 0.9.1 schemas.

func (DataType) Validate 

func (t DataType) Validate() error

Validate reports an error when the data type is unsupported.

type DialContextFunc 

type DialContextFunc func(context.Context, string, string) (net.Conn, error)

DialContextFunc opens one network connection for the client.

type DynamicPartitionCreationConfig 

type DynamicPartitionCreationConfig struct {
	// MetadataAttempts bounds refreshes after a partition is created.
	MetadataAttempts int
	// RetryBackoff is the delay between metadata refresh attempts.
	RetryBackoff time.Duration
}

DynamicPartitionCreationConfig controls opt-in creation of missing writer partitions.

type FileSystemSecurityToken 

type FileSystemSecurityToken struct {
	// Schema identifies the filesystem credential format.
	Schema string
	// Token contains opaque secret bytes and must not be logged.
	Token []byte
	// ExpiresAt is the server expiration time.
	ExpiresAt time.Time
	// AdditionalInfo contains non-secret provider metadata.
	AdditionalInfo map[string]string
	// Revoked reports that the server invalidated the token.
	Revoked bool
}

FileSystemSecurityToken contains temporary filesystem credentials. String and GoString always redact Token.

func (FileSystemSecurityToken) Clone 

func (t FileSystemSecurityToken) Clone() FileSystemSecurityToken

Clone returns a deep copy suitable for handing to another component.

func (FileSystemSecurityToken) GoString 

func (t FileSystemSecurityToken) GoString() string

GoString returns a redacted Go-syntax representation.

func (FileSystemSecurityToken) String 

func (t FileSystemSecurityToken) String() string

String returns a representation with token bytes redacted.

type FileSystemSecurityTokenProvider 

type FileSystemSecurityTokenProvider interface {
	// AcquireFileSystemSecurityToken returns a new caller-owned token.
	AcquireFileSystemSecurityToken(context.Context) (FileSystemSecurityToken, error)
}

FileSystemSecurityTokenProvider acquires temporary filesystem credentials.

type FileSystemSecurityTokenProviderFunc 

type FileSystemSecurityTokenProviderFunc func(context.Context) (FileSystemSecurityToken, error)

FileSystemSecurityTokenProviderFunc adapts a function to FileSystemSecurityTokenProvider.

func (FileSystemSecurityTokenProviderFunc) AcquireFileSystemSecurityToken 

func (f FileSystemSecurityTokenProviderFunc) AcquireFileSystemSecurityToken(
	ctx context.Context,
) (FileSystemSecurityToken, error)

AcquireFileSystemSecurityToken calls f.

type FileSystemSecurityTokenReceiver 

type FileSystemSecurityTokenReceiver interface {
	// ReceiveFileSystemSecurityToken receives a deep clone after acquisition.
	// Returning an error rejects publication and schedules a bounded retry.
	// Receivers run sequentially and should return promptly. A receiver may
	// close the client; shutdown stops waiting for the callback but cannot
	// forcibly stop callback code that remains blocked.
	ReceiveFileSystemSecurityToken(FileSystemSecurityToken) error
}

FileSystemSecurityTokenReceiver receives a clone after each successful acquisition.

type FileSystemSecurityTokenReceiverFunc 

type FileSystemSecurityTokenReceiverFunc func(FileSystemSecurityToken) error

FileSystemSecurityTokenReceiverFunc adapts a function to FileSystemSecurityTokenReceiver.

func (FileSystemSecurityTokenReceiverFunc) ReceiveFileSystemSecurityToken 

func (f FileSystemSecurityTokenReceiverFunc) ReceiveFileSystemSecurityToken(
	token FileSystemSecurityToken,
) error

ReceiveFileSystemSecurityToken calls f with token.

type FileSystemSecurityTokenRefreshConfig 

type FileSystemSecurityTokenRefreshConfig struct {
	// RenewalRatio selects the fraction of token lifetime before renewal;
	// zero defaults to 0.75.
	RenewalRatio float64
	// RetryBackoff is the initial acquisition retry delay; zero defaults to 1m.
	RetryBackoff time.Duration
	// MaxRetryBackoff bounds exponential retry delay; zero defaults to 1h.
	MaxRetryBackoff time.Duration
	// ClockSkew renews this much earlier than calculated; zero defaults to 30s.
	ClockSkew time.Duration
	// Jitter randomizes renewal delay as a fraction in [0, 1].
	Jitter float64
	// contains filtered or unexported fields
}

FileSystemSecurityTokenRefreshConfig controls renewal timing and bounded retry backoff. Zero fields use documented defaults.

type FileSystemSecurityTokenSource 

type FileSystemSecurityTokenSource func() (FileSystemSecurityToken, bool)

FileSystemSecurityTokenSource returns a cloned current token when available.

type KVBatch 

type KVBatch struct {
	// SchemaID identifies the row schema used by Records.
	SchemaID int16
	// WriterID identifies the idempotent writer session.
	WriterID int64
	// BatchSequence is monotonic within WriterID and bucket.
	BatchSequence int32
	// Records are encoded in request order.
	Records []KVRecord
}

KVBatch carries the writer state required for idempotent KV writes.

func DecodeKVBatch 

func DecodeKVBatch(encoded []byte) (KVBatch, error)

DecodeKVBatch validates and decodes a KV batch. Returned key and value buffers are owned by the caller.

func (KVBatch) Encode 

func (b KVBatch) Encode() ([]byte, error)

Encode serializes the KV batch using the Apache Fluss 0.9.1 layout.

type KVRecord 

type KVRecord struct {
	// Key contains an encoded non-empty primary key.
	Key []byte
	// Value contains an encoded row; nil represents a delete.
	Value []byte
}

KVRecord is one raw primary-key record. A nil Value represents a delete. Buffers returned by a decoder are owned by the caller.

type KVWriter 

type KVWriter struct {
	// contains filtered or unexported fields
}

KVWriter batches primary-key upserts and deletes. It owns one background scheduler and must be closed after use.

func (*KVWriter) Close 

func (w *KVWriter) Close(ctx context.Context) error

Close flushes accepted mutations and stops the writer. Close is idempotent.

func (*KVWriter) Delete 

func (w *KVWriter) Delete(ctx context.Context, key PrimaryKey) *WriteFuture

Delete queues removal of the row identified by key.

func (*KVWriter) Flush 

func (w *KVWriter) Flush(ctx context.Context) error

Flush waits until every previously accepted mutation reaches a terminal result.

func (*KVWriter) PartialUpsert 

func (w *KVWriter) PartialUpsert(ctx context.Context, columns []string, values Row) *WriteFuture

PartialUpsert writes only columns named by columns. The columns must contain the complete primary key and must be in the same order as values.

func (*KVWriter) Upsert 

func (w *KVWriter) Upsert(ctx context.Context, row Row) *WriteFuture

Upsert queues a complete primary-key row for insertion or update.

type KVWriterConfig 

type KVWriterConfig struct {
	// MaxBatchBytes bounds encoded bytes in one put request.
	MaxBatchBytes int
	// MaxBatchRecords bounds mutations in one put request.
	MaxBatchRecords int
	// MaxBuffered bounds accepted mutations awaiting completion.
	MaxBuffered int
	// MaxConcurrentRequests bounds active put calls across distinct buckets.
	MaxConcurrentRequests int
	// Linger is the maximum delay used to fill a non-full batch.
	Linger time.Duration
	// Timeout bounds both the server-side put operation and the client call.
	Timeout time.Duration
	// Acks is 0, 1, or -1 using the Fluss acknowledgement contract.
	Acks int32
	// Retry controls bounded retries of idempotent batches. More than one
	// attempt requires Acks=-1.
	Retry WriterRetryPolicy
	// Partition selects one named physical partition; empty selects the table.
	Partition string
	// MergeMode selects merge-engine or overwrite semantics.
	MergeMode MergeMode
}

KVWriterConfig controls batching, buffering, acknowledgements, partition routing, and merge behavior for a primary-key writer.

type KVWriterOption 

type KVWriterOption func(*KVWriterConfig) error

KVWriterOption configures a KVWriter.

func WithKVBatchLimits 

func WithKVBatchLimits(bytes, records int) KVWriterOption

WithKVBatchLimits sets maximum encoded bytes and records in one request.

func WithKVBuffer 

func WithKVBuffer(records int) KVWriterOption

WithKVBuffer bounds the number of queued records awaiting completion.

func WithKVConcurrency 

func WithKVConcurrency(requests int) KVWriterOption

WithKVConcurrency bounds active put calls across distinct buckets. Calls for one bucket remain strictly ordered.

func WithKVLinger 

func WithKVLinger(linger time.Duration) KVWriterOption

WithKVLinger sets the maximum delay used to collect a non-full batch.

func WithKVMergeMode 

func WithKVMergeMode(mode MergeMode) KVWriterOption

WithKVMergeMode sets one merge mode for every record written by the writer.

func WithKVPartition 

func WithKVPartition(partition string) KVWriterOption

WithKVPartition routes writes to the named physical partition.

func WithKVPartitionSpec 

func WithKVPartitionSpec(schema Schema, spec PartitionSpec) KVWriterOption

WithKVPartitionSpec selects a partition using the table schema's partition-key order.

func WithKVRequest 

func WithKVRequest(timeout time.Duration, acks int32) KVWriterOption

WithKVRequest sets the server and client call timeout and acknowledgement mode.

func WithKVRetryPolicy 

func WithKVRetryPolicy(policy WriterRetryPolicy) KVWriterOption

WithKVRetryPolicy configures bounded idempotent put retries. Retries preserve the encoded batch, writer ID, and bucket sequence.

type KeyCodec 

type KeyCodec[K any] interface {
	// EncodeKey maps one application key to primary-key column order.
	EncodeKey(K) (PrimaryKey, error)
}

KeyCodec maps an application key to primary-key column order.

type KeyCodecFunc 

type KeyCodecFunc[K any] func(K) (PrimaryKey, error)

KeyCodecFunc adapts a function to KeyCodec.

func (KeyCodecFunc[K]) EncodeKey 

func (f KeyCodecFunc[K]) EncodeKey(key K) (PrimaryKey, error)

EncodeKey calls f and rejects a nil function.

type LocalRemoteFileReader 

type LocalRemoteFileReader struct{}

LocalRemoteFileReader supports absolute paths and file:// URIs without an external dependency.

func (LocalRemoteFileReader) OpenRemoteFile 

func (LocalRemoteFileReader) OpenRemoteFile(
	ctx context.Context,
	request RemoteFileRequest,
) (io.ReadCloser, error)

OpenRemoteFile opens an exact local-file range without buffering the object.

func (LocalRemoteFileReader) ReadRemoteFile 

func (LocalRemoteFileReader) ReadRemoteFile(
	ctx context.Context,
	request RemoteFileRequest,
) ([]byte, error)

ReadRemoteFile reads an absolute local path or file URI.

type LogBatch 

type LogBatch struct {
	// Magic selects the Fluss v0 or v1 batch header.
	Magic byte
	// BaseOffset is the first record offset.
	BaseOffset int64
	// CommitTime is the server commit time in Unix milliseconds.
	CommitTime int64
	// LeaderEpoch is present only for magic 1.
	LeaderEpoch int32
	// SchemaID identifies the row schema used by Records.
	SchemaID int16
	// AppendOnly omits per-record change bytes when true.
	AppendOnly bool
	// WriterID identifies the idempotent writer session.
	WriterID int64
	// BatchSequence is monotonic within WriterID and bucket.
	BatchSequence int32
	// Records are ordered by offset.
	Records []Record
}

LogBatch encodes compacted or indexed row records. Magic 0 and 1 use the Fluss 0.9.1 layouts.

func DecodeLogBatchRows 

func DecodeLogBatchRows(schema Schema, encoded []byte, compacted bool) (LogBatch, error)

DecodeLogBatchRows decodes a single Fluss v0 or v1 row-oriented log batch. It validates the CRC before allocating records and returns the schema ID advertised by the batch.

func (LogBatch) EncodeRows 

func (b LogBatch) EncodeRows(schema Schema, compacted bool) ([]byte, error)

EncodeRows serializes row records using compacted or indexed encoding.

type LogScanner 

type LogScanner struct {
	// contains filtered or unexported fields
}

LogScanner polls ordered records from subscribed log buckets. Poll calls are serialized; Wakeup and Close may be called concurrently.

func (*LogScanner) Close 

func (s *LogScanner) Close() error

Close stops the scanner and interrupts an active poll. Close is idempotent.

func (*LogScanner) Done 

func (s *LogScanner) Done() bool

Done reports whether a configured row or stopping-offset bound has completed.

func (*LogScanner) Poll 

func (s *LogScanner) Poll(ctx context.Context) (ScanResult, error)

Poll waits for records, a terminal bound, wakeup, close, or ctx cancellation.

func (*LogScanner) Schema 

func (s *LogScanner) Schema() Schema

Schema returns the result schema after projection.

func (*LogScanner) Subscribe 

func (s *LogScanner) Subscribe(ctx context.Context, bucket int32, start ScanOffset) error

Subscribe adds or resets one bucket at start.

func (*LogScanner) Unsubscribe 

func (s *LogScanner) Unsubscribe(bucket int32)

Unsubscribe removes bucket from subsequent polls.

func (*LogScanner) Wakeup 

func (s *LogScanner) Wakeup()

Wakeup interrupts an active Poll, or the next Poll when none is active.

type LogScannerConfig 

type LogScannerConfig struct {
	// Projection lists returned columns in result order; nil selects all.
	Projection []string
	// Partition selects one named physical partition; empty selects the table.
	Partition string
	// MaxBytes bounds aggregate encoded bytes per fetch.
	MaxBytes int32
	// MaxBucketBytes bounds encoded bytes per bucket per fetch.
	MaxBucketBytes int32
	// MinBytes is the aggregate byte threshold before a fetch may return.
	MinBytes int32
	// MaxWait bounds server waiting when MinBytes is not reached.
	MaxWait time.Duration
	// RowLimit is the total delivered row bound; zero is unbounded.
	RowLimit int64
	// StoppingOffsets maps every initial bucket to an exclusive end offset.
	StoppingOffsets map[int32]int64
}

LogScannerConfig controls projection, partition selection, fetch limits, and optional completion bounds.

type LogScannerOption 

type LogScannerOption func(*LogScannerConfig) error

LogScannerOption configures a LogScanner.

func WithScanLimits 

func WithScanLimits(maxBytes, maxBucketBytes, minBytes int32, maxWait time.Duration) LogScannerOption

WithScanLimits sets aggregate and per-bucket fetch byte limits and wait time.

func WithScanPartition 

func WithScanPartition(partition string) LogScannerOption

WithScanPartition scans the named physical partition.

func WithScanPartitionSpec 

func WithScanPartitionSpec(schema Schema, spec PartitionSpec) LogScannerOption

WithScanPartitionSpec selects a partition using the table schema's partition-key order.

func WithScanProjection 

func WithScanProjection(columns ...string) LogScannerOption

WithScanProjection selects result columns in the requested order.

func WithScanRowLimit 

func WithScanRowLimit(limit int64) LogScannerOption

WithScanRowLimit completes a scanner after it has delivered limit rows across all buckets.

func WithScanStoppingOffsets 

func WithScanStoppingOffsets(offsets map[int32]int64) LogScannerOption

WithScanStoppingOffsets sets the exclusive stopping offset for every initial bucket.

type LogWriteFormat 

type LogWriteFormat string

LogWriteFormat selects the row or Arrow encoding used for log batches. 

const (
	LogWriteFormatAuto      LogWriteFormat = "auto"
	LogWriteFormatArrow     LogWriteFormat = "arrow"
	LogWriteFormatIndexed   LogWriteFormat = "indexed"
	LogWriteFormatCompacted LogWriteFormat = "compacted"
)

Log write formats supported by Apache Fluss 0.9.1.

type LogWriter 

type LogWriter struct {
	// contains filtered or unexported fields
}

LogWriter batches append operations for a log table. It owns one background scheduler and must be closed after use.

func (*LogWriter) Append 

func (w *LogWriter) Append(ctx context.Context, row Row) *WriteFuture

Append queues one row. The returned future completes exactly once after acknowledgment or failure. The writer copies row bytes while encoding, so the caller may reuse the row afterward.

func (*LogWriter) AppendArrow 

func (w *LogWriter) AppendArrow(ctx context.Context, bucket int32, batch arrow.RecordBatch, changes []ChangeType) *WriteFuture

AppendArrow queues one Arrow record batch for an explicit bucket. The caller must retain the record batch until the returned future completes.

func (*LogWriter) Close 

func (w *LogWriter) Close(ctx context.Context) error

Close flushes accepted appends and stops the writer. Close is idempotent.

func (*LogWriter) Flush 

func (w *LogWriter) Flush(ctx context.Context) error

Flush waits until every previously accepted append reaches a terminal result.

type LogWriterConfig 

type LogWriterConfig struct {
	// MaxBatchBytes bounds encoded bytes in one produce request.
	MaxBatchBytes int
	// MaxBatchRecords bounds records in one produce request.
	MaxBatchRecords int
	// MaxBuffered bounds accepted records awaiting completion.
	MaxBuffered int
	// MaxConcurrentRequests bounds active produce calls across distinct buckets.
	MaxConcurrentRequests int
	// Linger is the maximum delay used to fill a non-full batch.
	Linger time.Duration
	// Timeout bounds both the server-side produce operation and the client call.
	Timeout time.Duration
	// Acks is 0, 1, or -1 using the Fluss acknowledgement contract.
	Acks int32
	// Retry controls bounded retries of idempotent batches. More than one
	// attempt requires Acks=-1.
	Retry WriterRetryPolicy
	// Assignment selects routing for records without a key.
	Assignment BucketAssignment
	// Partition selects one named physical partition; empty selects the table.
	Partition string
	// Format selects row or Arrow encoding.
	Format LogWriteFormat
	// ArrowCompression applies only when Format resolves to Arrow.
	ArrowCompression ArrowCompression
	// contains filtered or unexported fields
}

LogWriterConfig controls batching, buffering, acknowledgements, routing, and encoding for a log writer.

type LogWriterOption 

type LogWriterOption func(*LogWriterConfig) error

LogWriterOption configures a LogWriter.

func WithLogArrowCompression 

func WithLogArrowCompression(compression ArrowCompression) LogWriterOption

WithLogArrowCompression selects the compression for Arrow log batches.

func WithLogBatchLimits 

func WithLogBatchLimits(bytes, records int) LogWriterOption

WithLogBatchLimits sets maximum encoded bytes and records in one request.

func WithLogBucketAssignment 

func WithLogBucketAssignment(assignment BucketAssignment) LogWriterOption

WithLogBucketAssignment selects routing for rows without a key.

func WithLogBuffer 

func WithLogBuffer(records int) LogWriterOption

WithLogBuffer bounds the number of queued records awaiting completion.

func WithLogConcurrency 

func WithLogConcurrency(requests int) LogWriterOption

WithLogConcurrency bounds active produce calls across distinct buckets. Calls for one bucket remain strictly ordered.

func WithLogLinger 

func WithLogLinger(linger time.Duration) LogWriterOption

WithLogLinger sets the maximum delay used to collect a non-full batch.

func WithLogPartition 

func WithLogPartition(partition string) LogWriterOption

WithLogPartition routes writes to the named physical partition.

func WithLogPartitionSpec 

func WithLogPartitionSpec(schema Schema, spec PartitionSpec) LogWriterOption

WithLogPartitionSpec selects a partition using the table schema's partition-key order.

func WithLogRequest 

func WithLogRequest(timeout time.Duration, acks int32) LogWriterOption

WithLogRequest sets the server and client call timeout and acknowledgement mode.

func WithLogRetryPolicy 

func WithLogRetryPolicy(policy WriterRetryPolicy) LogWriterOption

WithLogRetryPolicy configures bounded idempotent produce retries. Retries preserve the encoded batch, writer ID, and bucket sequence.

func WithLogWriteFormat 

func WithLogWriteFormat(format LogWriteFormat) LogWriterOption

WithLogWriteFormat selects the server-supported encoding used by this writer.

type LogicalField 

type LogicalField struct {
	// Name is unique within the nested row.
	Name string `json:"name"`
	// Type describes the nested field value.
	Type LogicalType `json:"field_type"`
	// Description is optional user metadata.
	Description string `json:"description,omitempty"`
	// ID is the stable Fluss field identifier.
	ID int `json:"field_id"`
}

LogicalField describes one nested field of a ROW logical type.

type LogicalType 

type LogicalType struct {
	// Root identifies the Fluss logical type.
	Root string `json:"type"`
	// Nullable reports whether nil is valid.
	Nullable bool `json:"nullable,omitempty"`
	// Length applies to fixed and variable character or binary values.
	Length int `json:"length,omitempty"`
	// Precision applies to decimal, time, and timestamp values.
	Precision int `json:"precision,omitempty"`
	// Scale applies to decimal values.
	Scale int `json:"scale,omitempty"`
	// Element describes array elements.
	Element *LogicalType `json:"element_type,omitempty"`
	// Key describes map keys.
	Key *LogicalType `json:"key_type,omitempty"`
	// Value describes map values.
	Value *LogicalType `json:"value_type,omitempty"`
	// Fields describe nested row columns in encoding order.
	Fields []LogicalField `json:"fields,omitempty"`
}

LogicalType is the Fluss 0.9.1 schema JSON type representation. It supports parameterized and nested roots without exposing protobuf implementation details.

func ParseLogicalTypeJSON 

func ParseLogicalTypeJSON(data []byte) (LogicalType, error)

ParseLogicalTypeJSON decodes and validates a logical type.

func (LogicalType) JSON 

func (t LogicalType) JSON() ([]byte, error)

JSON validates and encodes the logical type.

func (LogicalType) MarshalJSON 

func (t LogicalType) MarshalJSON() ([]byte, error)

MarshalJSON encodes the Fluss logical-type JSON representation.

func (*LogicalType) UnmarshalJSON 

func (t *LogicalType) UnmarshalJSON(data []byte) error

UnmarshalJSON decodes the Fluss logical-type JSON representation.

func (LogicalType) Validate 

func (t LogicalType) Validate() error

Validate checks root-specific parameters and nested types.

type LookupClient 

type LookupClient struct {
	// contains filtered or unexported fields
}

LookupClient performs batched point and prefix lookups for a primary-key table. A LookupClient is safe for concurrent calls and must be closed.

func (*LookupClient) Close 

func (c *LookupClient) Close() error

Close cancels active requests and rejects new lookups. Close is idempotent.

func (*LookupClient) Lookup 

func (c *LookupClient) Lookup(ctx context.Context, keys ...PrimaryKey) []LookupResult

Lookup returns one result for each input key in input order.

func (*LookupClient) PrefixLookup 

func (c *LookupClient) PrefixLookup(ctx context.Context, prefixes ...PrimaryKey) []PrefixLookupResult

PrefixLookup returns one result for each leading primary-key prefix in input order.

type LookupConfig 

type LookupConfig struct {
	// MaxBatchKeys bounds keys in one lookup request.
	MaxBatchKeys int
	// MaxConcurrent bounds in-flight lookup requests.
	MaxConcurrent int
	// MaxQueuedKeys bounds accepted keys awaiting completion across callers.
	MaxQueuedKeys int
	// BatchDelay is the maximum time a partial cross-call batch waits.
	BatchDelay time.Duration
	// Retry controls safe read-only lookup retries. Insert-if-not-exists
	// clients require one attempt.
	Retry RetryPolicy
	// Partition selects one named physical partition; empty selects the table.
	Partition string
	// InsertIfNotExists enables atomic insertion for missing full keys.
	InsertIfNotExists bool
	// Timeout bounds each lookup request and is sent to insertion requests.
	Timeout time.Duration
	// Acks is the insertion acknowledgement mode.
	Acks int32
}

LookupConfig controls batching, concurrency, partition routing, and optional atomic insertion of missing rows.

type LookupOption 

type LookupOption func(*LookupConfig) error

LookupOption configures a LookupClient.

func WithLookupBatch 

func WithLookupBatch(keys, concurrent int) LookupOption

WithLookupBatch sets the maximum keys per request and concurrent requests.

func WithLookupInsertIfNotExists 

func WithLookupInsertIfNotExists(timeout time.Duration, acks int32) LookupOption

WithLookupInsertIfNotExists atomically inserts a missing primary-key row before returning it. Fluss fills auto-increment columns and sets nullable non-key columns to null.

func WithLookupPartition 

func WithLookupPartition(partition string) LookupOption

WithLookupPartition routes lookups to the named physical partition.

func WithLookupQueue 

func WithLookupQueue(keys int, delay time.Duration) LookupOption

WithLookupQueue sets the accepted-key limit and partial-batch delay used to combine compatible concurrent calls.

func WithLookupRetryPolicy 

func WithLookupRetryPolicy(policy RetryPolicy) LookupOption

WithLookupRetryPolicy configures bounded retries for read-only point and prefix lookups. Insert-if-not-exists clients cannot enable retries.

func WithLookupTimeout 

func WithLookupTimeout(timeout time.Duration) LookupOption

WithLookupTimeout sets the client-side timeout for each lookup request.

type LookupResult 

type LookupResult struct {
	// Key is the requested primary key.
	Key PrimaryKey
	// Row is valid only when Found is true and Err is nil.
	Row Row
	// Found distinguishes a missing row from an empty row.
	Found bool
	// Bucket is the routed bucket when routing succeeded.
	Bucket int32
	// Err is a key-local encoding, routing, request, or decoding failure.
	Err error
}

LookupResult is the outcome associated with one requested primary key.

Example 
package main

import (
	"errors"
	"fmt"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	results := []fgo.LookupResult{
		{Key: fgo.PrimaryKey{int64(41)}, Row: fgo.Row{int64(41), "Ada"}, Found: true},
		{Key: fgo.PrimaryKey{int64(42)}, Err: fgo.ErrNotFound},
		{Key: fgo.PrimaryKey{int64(43)}, Err: fgo.ErrTimeout},
	}
	for _, result := range results {
		switch {
		case errors.Is(result.Err, fgo.ErrNotFound):
			fmt.Printf("%d missing\n", result.Key[0])
		case result.Err != nil:
			fmt.Printf("%d failed\n", result.Key[0])
		default:
			fmt.Printf("%d found\n", result.Key[0])
		}
	}
}

Output:
41 found
42 missing
43 failed

type Map 

type Map []MapEntry

Map is an ordered collection of Fluss map entries.

type MapEntry 

type MapEntry struct {
	// Key is non-nil and conforms to the map key logical type.
	Key any
	// Value may be nil only when the map value type is nullable.
	Value any
}

MapEntry preserves Fluss map key types and iteration order. Go maps with string keys are also accepted as input, but decoded map values always use Map.

type MergeMode 

type MergeMode int32

MergeMode controls whether Fluss applies or bypasses a table's merge engine. 

const (
	// MergeModeDefault applies the table's configured merge engine.
	MergeModeDefault MergeMode = 0
	// MergeModeOverwrite bypasses the merge engine and replaces stored values.
	MergeModeOverwrite MergeMode = 1
)

type Metadata 

type Metadata struct {
	// Coordinator is the current coordinator node.
	Coordinator Node
	// Tablets maps tablet node IDs to endpoints.
	Tablets map[int32]Node
	// Tables maps logical paths to current metadata snapshots.
	Tables map[TablePath]TableMetadata
}

Metadata is an immutable snapshot of coordinator, tablet, and table routing.

type MetadataFetcher 

type MetadataFetcher func(context.Context, TablePath) (TableMetadata, error)

MetadataFetcher loads authoritative metadata for one logical table.

type MetricErrorClass 

type MetricErrorClass uint8

MetricErrorClass is a bounded-cardinality failure classification. 

const (
	MetricErrorNone MetricErrorClass = iota
	MetricErrorCanceled
	MetricErrorTimeout
	MetricErrorClosed
	MetricErrorServer
	MetricErrorProtocol
	MetricErrorOther
)

Metric error classes emitted by the client.

type MetricEvent 

type MetricEvent struct {
	// Kind identifies the event category.
	Kind MetricKind
	// Operation identifies the bounded high-level operation.
	Operation MetricOperation
	// APIKey is set for RPC events and zero otherwise.
	APIKey fmsg.APIKey
	// ServerRole is set for server-bound events.
	ServerRole ServerRole
	// Duration is the completed operation duration.
	Duration time.Duration
	// QueueTime is time spent waiting before execution.
	QueueTime time.Duration
	// Attempt starts at one and is set for request and retry events.
	Attempt int
	// QueueSize is the observed bounded queue depth.
	QueueSize int
	// Records is the event record count.
	Records int64
	// Bytes is the encoded or transferred byte count.
	Bytes int64
	// Lag is a non-negative offset lag when available.
	Lag int64
	// Failed reports whether the observed operation failed.
	Failed bool
	// ErrorClass classifies failures without exposing error text.
	ErrorClass MetricErrorClass
}

MetricEvent contains bounded-cardinality measurements and never includes request payloads, addresses, table paths, bucket IDs, credentials, or server error messages.

type MetricKind 

type MetricKind uint8

MetricKind identifies the client activity represented by an event. 

const (
	MetricRequest MetricKind = iota
	MetricRetry
	MetricRemoteIO
	MetricWriteBatch
	MetricScannerFetch
	MetricDecodeFailure
	MetricThrottle
)

Metric event kinds emitted by the client.

type MetricOperation 

type MetricOperation uint8

MetricOperation identifies the high-level operation producing an event. 

const (
	MetricOperationRPC MetricOperation = iota
	MetricOperationDial
	MetricOperationLogWrite
	MetricOperationKVWrite
	MetricOperationLogScan
	MetricOperationRemoteRead
	MetricOperationLookup
)

Metric operations emitted by the client.

type MetricsObserver 

type MetricsObserver interface {
	// ObserveMetric receives an event synchronously and should return quickly.
	// Panics are recovered and ignored by the client.
	ObserveMetric(MetricEvent)
}

MetricsObserver receives synchronous client events. Implementations should return quickly and must not retain sensitive data.

type MetricsObserverFunc 

type MetricsObserverFunc func(MetricEvent)

MetricsObserverFunc adapts a function to MetricsObserver.

func (MetricsObserverFunc) ObserveMetric 

func (f MetricsObserverFunc) ObserveMetric(event MetricEvent)

ObserveMetric calls f with event.

type NamedRow 

type NamedRow map[string]any

NamedRow maps every schema column name to one value.

type Node 

type Node struct {
	// ID is the server node identifier.
	ID int32
	// Address is the dialable host:port endpoint.
	Address string
	// Role is the negotiated Fluss server role.
	Role ServerRole
}

Node identifies one coordinator or tablet endpoint.

type OffsetSpec 

type OffsetSpec struct {
	// Offset is non-negative for an explicit lookup and -1 otherwise.
	Offset int64
	// Timestamp is set only for timestamp lookup.
	Timestamp time.Time
}

OffsetSpec selects an explicit offset or timestamp lookup.

func (OffsetSpec) Validate 

func (o OffsetSpec) Validate() error

Validate checks offset range and timestamp exclusivity.

type Option 

type Option func(*config) error

Option configures a Client.

func WithAuthenticator 

func WithAuthenticator(factory AuthenticatorFactory) Option

WithAuthenticator enables authentication with a fresh mechanism instance for each server connection.

func WithClientIdentity 

func WithClientIdentity(name, version string) Option

WithClientIdentity sets the name and version sent during API negotiation.

func WithDialContext 

func WithDialContext(dial DialContextFunc) Option

WithDialContext replaces the network dialer.

func WithDialTimeout 

func WithDialTimeout(timeout time.Duration) Option

WithDialTimeout bounds each connection attempt.

func WithDynamicPartitionCreation 

func WithDynamicPartitionCreation(settings DynamicPartitionCreationConfig) Option

WithDynamicPartitionCreation enables automatic creation for partitioned log and KV writers.

func WithFileSystemSecurityTokenProvider 

func WithFileSystemSecurityTokenProvider(
	provider FileSystemSecurityTokenProvider,
	refresh FileSystemSecurityTokenRefreshConfig,
	receivers ...FileSystemSecurityTokenReceiver,
) Option

WithFileSystemSecurityTokenProvider enables refresh through a custom provider.

func WithFileSystemSecurityTokenRefresh 

func WithFileSystemSecurityTokenRefresh(
	config FileSystemSecurityTokenRefreshConfig,
	receivers ...FileSystemSecurityTokenReceiver,
) Option

WithFileSystemSecurityTokenRefresh enables coordinator-backed token acquisition and optional receivers.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	receiver := fgo.FileSystemSecurityTokenReceiverFunc(
		func(token fgo.FileSystemSecurityToken) error {
			// Pass the cloned token to an external filesystem client.
			return nil
		},
	)
	reader := fgo.RemoteFileReaderFunc(
		func(ctx context.Context, request fgo.RemoteFileRequest) ([]byte, error) {
			// Read request.Path with request.Token using an application SDK.
			return nil, nil
		},
	)

	client, err := fgo.Open(
		context.Background(),
		fgo.WithSeedBrokers("coordinator.example:9123"),
		fgo.WithRemoteFileReader(reader, fgo.RemoteFileReadConfig{}),
		fgo.WithFileSystemSecurityTokenRefresh(
			fgo.FileSystemSecurityTokenRefreshConfig{},
			receiver,
		),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()
}

func WithMetricsObserver 

func WithMetricsObserver(observer MetricsObserver) Option

WithMetricsObserver registers a synchronous bounded-cardinality event observer. Observer panics are isolated from client operations.

func WithRemoteFileReader 

func WithRemoteFileReader(reader RemoteFileReader, settings RemoteFileReadConfig) Option

WithRemoteFileReader enables remote log reads and supplies the shared filesystem adapter.

func WithRetryPolicy 

func WithRetryPolicy(policy RetryPolicy) Option

WithRetryPolicy configures bounded retries for safe read-only requests. Mutations are not blindly retried.

func WithSeedBrokers 

func WithSeedBrokers(seeds ...string) Option

WithSeedBrokers sets coordinator bootstrap addresses.

func WithSnapshotBatchProvider 

func WithSnapshotBatchProvider(provider SnapshotBatchProvider) Option

WithSnapshotBatchProvider configures explicit snapshot-ID scans.

func WithTLSConfig 

func WithTLSConfig(tlsConfig *tls.Config) Option

WithTLSConfig enables TLS using a clone of tlsConfig.

func WithTransportLimits 

func WithTransportLimits(limits transport.Config) Option

WithTransportLimits sets request and response frame limits.

type PartitionMetadata 

type PartitionMetadata struct {
	// Path identifies the logical table and physical partition.
	Path PhysicalTablePath
	// ID is the server-assigned partition identifier.
	ID int64
	// Buckets maps bucket IDs to current tablet leaders.
	Buckets map[int32]Node
	// contains filtered or unexported fields
}

PartitionMetadata describes a named partition and its tablet leaders.

type PartitionSpec 

type PartitionSpec map[string]string

PartitionSpec maps every partition key to one value.

func PartitionSpecFromName 

func PartitionSpecFromName(keys []string, name string) (PartitionSpec, error)

PartitionSpecFromName decodes a Fluss partition name using the ordered schema keys.

type PhysicalMetadataFetcher 

type PhysicalMetadataFetcher func(context.Context, PhysicalTablePath) (PartitionMetadata, error)

PhysicalMetadataFetcher refreshes a single named partition. It is optional because a table metadata response can include all of its partitions.

type PhysicalTablePath 

type PhysicalTablePath struct {
	TablePath
	// Partition is an optional canonical physical partition name.
	Partition string
}

PhysicalTablePath identifies a logical table and optional named partition.

func (PhysicalTablePath) String 

func (p PhysicalTablePath) String() string

String returns the logical path with an optional partition suffix.

func (PhysicalTablePath) Validate 

func (p PhysicalTablePath) Validate() error

Validate checks the logical path and optional partition name.

type PrefixLookupResult 

type PrefixLookupResult struct {
	// Prefix is the requested leading primary-key prefix.
	Prefix PrimaryKey
	// Rows contains all decoded matches when Err is nil.
	Rows []Row
	// Bucket is the routed bucket when routing succeeded.
	Bucket int32
	// Err is a prefix-local encoding, routing, request, or decoding failure.
	Err error
}

PrefixLookupResult contains rows matching one leading primary-key prefix.

type PrimaryKey 

type PrimaryKey Row

PrimaryKey stores key values in schema primary-key order.

type Record 

type Record struct {
	// Key contains primary-key columns and may be nil for log records.
	Key PrimaryKey
	// Value contains schema-ordered columns and may be nil for deletes.
	Value Row
	// Change identifies append, upsert, update, or delete semantics.
	Change ChangeType
	// Timestamp is the server commit timestamp.
	Timestamp time.Time
	// Offset is the bucket-local log offset, or -1 before assignment.
	Offset int64
}

Record is one keyed or unkeyed row change and its log metadata.

func (Record) Validate 

func (r Record) Validate(schema Schema) error

Validate checks the change, offset, key, and value against schema.

type RemoteFileReadConfig 

type RemoteFileReadConfig struct {
	// MaxAttempts includes the initial read; zero defaults to 3.
	MaxAttempts int
	// RetryBackoff is the delay between attempts; zero defaults to 50ms.
	RetryBackoff time.Duration
	// MaxFileBytes bounds allocation per object; zero defaults to 256 MiB.
	MaxFileBytes int64
	// MaxTotalBytes bounds bytes retained for one snapshot or remote-log read;
	// zero defaults to 512 MiB.
	MaxTotalBytes int64
	// MaxFiles bounds objects referenced by one operation; zero defaults to 4096.
	MaxFiles int
	// MaxConcurrentReads bounds active object streams; zero defaults to 4.
	MaxConcurrentReads int
	// MaxConcurrentBytes bounds advertised bytes across active streams; zero
	// defaults to 256 MiB.
	MaxConcurrentBytes int64
}

RemoteFileReadConfig bounds retries, backoff, object size, aggregate bytes, and file count. Zero fields use documented defaults.

type RemoteFileReader 

type RemoteFileReader interface {
	// ReadRemoteFile returns the complete object and honors ctx cancellation.
	// The returned buffer becomes caller-owned.
	ReadRemoteFile(context.Context, RemoteFileRequest) ([]byte, error)
}

RemoteFileReader reads one complete Fluss-managed remote object. Filesystem-specific adapters can use the cloned security token without exposing it through errors or formatting.

type RemoteFileReaderFunc 

type RemoteFileReaderFunc func(context.Context, RemoteFileRequest) ([]byte, error)

RemoteFileReaderFunc adapts a function to RemoteFileReader.

func (RemoteFileReaderFunc) ReadRemoteFile 

func (f RemoteFileReaderFunc) ReadRemoteFile(
	ctx context.Context,
	request RemoteFileRequest,
) ([]byte, error)

ReadRemoteFile calls f with request.

type RemoteFileRequest 

type RemoteFileRequest struct {
	// Path is the storage-specific URI or absolute local path.
	Path string
	// ExpectedSize is the server-advertised size, or zero when unavailable.
	ExpectedSize int64
	// MaxBytes is the largest complete object the caller accepts. Zero defaults
	// to 256 MiB. Implementations must enforce this limit while reading, before
	// allocating or returning the complete object.
	MaxBytes int64
	// Offset is the first requested object byte for streaming readers.
	Offset int64
	// Length is the exact requested byte count for streaming readers. Zero
	// requests from Offset through the end of the object.
	Length int64
	// Token is an optional caller-owned credential clone.
	Token *FileSystemSecurityToken
}

RemoteFileRequest describes one complete remote object read. Token is a caller-owned clone and may be nil.

type RemoteFileStreamReader 

type RemoteFileStreamReader interface {
	// OpenRemoteFile opens the exact requested range and transfers response
	// body ownership to the caller.
	OpenRemoteFile(context.Context, RemoteFileRequest) (io.ReadCloser, error)
}

RemoteFileStreamReader opens a bounded object range. Implementations must honor Offset, Length, ExpectedSize, MaxBytes, and context cancellation. Close must release all SDK response bodies and transport resources.

type RemoteLogFetchInfo 

type RemoteLogFetchInfo struct {
	// TabletDirectory is the server-advertised remote tablet path.
	TabletDirectory string
	// PartitionName is the optional physical partition name.
	PartitionName string
	// FirstStartPosition is the byte position in the first segment.
	FirstStartPosition int
	// Segments are ordered by StartOffset.
	Segments []RemoteLogSegment
}

RemoteLogFetchInfo describes remote segments referenced by a fetch response.

type RemoteLogSegment 

type RemoteLogSegment struct {
	// ID identifies the segment in remote storage metadata.
	ID string
	// StartOffset is the first log offset in the segment.
	StartOffset int64
	// EndOffset is the exclusive segment end offset.
	EndOffset int64
	// SizeBytes is the expected encoded object size.
	SizeBytes int64
	// MaxTime is the greatest record timestamp in the segment.
	MaxTime time.Time
}

RemoteLogSegment describes one immutable remote log object and offset range.

type RemoteSnapshotBatchProvider 

type RemoteSnapshotBatchProvider struct {
	// contains filtered or unexported fields
}

RemoteSnapshotBatchProvider composes snapshot metadata and format adapters with the shared remote-file transport. Decoder-specific dependencies remain optional.

func NewRemoteSnapshotBatchProvider 

func NewRemoteSnapshotBatchProvider(
	reader RemoteFileReader,
	settings RemoteFileReadConfig,
	resolver RemoteSnapshotResolver,
	decoder RemoteSnapshotDecoder,
	tokenSource FileSystemSecurityTokenSource,
	observer MetricsObserver,
) (*RemoteSnapshotBatchProvider, error)

NewRemoteSnapshotBatchProvider composes object discovery, bounded downloads, and format-specific decoding.

func (*RemoteSnapshotBatchProvider) OpenSnapshot 

func (p *RemoteSnapshotBatchProvider) OpenSnapshot(
	ctx context.Context,
	request SnapshotBatchRequest,
) (SnapshotBatchReader, error)

OpenSnapshot downloads and opens the requested immutable snapshot.

type RemoteSnapshotDecoder 

type RemoteSnapshotDecoder interface {
	// OpenSnapshotFiles consumes downloaded file descriptors and returns a
	// reader owned by the caller.
	OpenSnapshotFiles(context.Context, SnapshotBatchRequest, []RemoteSnapshotFile) (SnapshotBatchReader, error)
}

RemoteSnapshotDecoder opens downloaded snapshot objects in a storage-specific format.

type RemoteSnapshotDecoderFunc 

type RemoteSnapshotDecoderFunc func(
	context.Context,
	SnapshotBatchRequest,
	[]RemoteSnapshotFile,
) (SnapshotBatchReader, error)

RemoteSnapshotDecoderFunc adapts a function to RemoteSnapshotDecoder.

func (RemoteSnapshotDecoderFunc) OpenSnapshotFiles 

func (f RemoteSnapshotDecoderFunc) OpenSnapshotFiles(
	ctx context.Context,
	request SnapshotBatchRequest,
	files []RemoteSnapshotFile,
) (SnapshotBatchReader, error)

OpenSnapshotFiles calls f with the downloaded files.

type RemoteSnapshotFile 

type RemoteSnapshotFile struct {
	// Path is the storage-specific object location.
	Path string
	// Size is the expected object size in bytes.
	Size int64
	// Data is populated with caller-owned downloaded bytes before decoding.
	Data []byte
}

RemoteSnapshotFile describes a snapshot object before and after download.

type RemoteSnapshotResolver 

type RemoteSnapshotResolver interface {
	// ResolveSnapshotFiles returns immutable files in decoder input order.
	ResolveSnapshotFiles(context.Context, SnapshotBatchRequest) ([]RemoteSnapshotFile, error)
}

RemoteSnapshotResolver discovers immutable objects for one snapshot request.

type RemoteSnapshotResolverFunc 

type RemoteSnapshotResolverFunc func(
	context.Context,
	SnapshotBatchRequest,
) ([]RemoteSnapshotFile, error)

RemoteSnapshotResolverFunc adapts a function to RemoteSnapshotResolver.

func (RemoteSnapshotResolverFunc) ResolveSnapshotFiles 

func (f RemoteSnapshotResolverFunc) ResolveSnapshotFiles(
	ctx context.Context,
	request SnapshotBatchRequest,
) ([]RemoteSnapshotFile, error)

ResolveSnapshotFiles calls f with request.

type RetryPolicy 

type RetryPolicy struct {
	// MaxAttempts includes the initial request and must be positive.
	MaxAttempts int
	// Backoff returns the delay before the numbered retry attempt.
	Backoff func(attempt int) time.Duration
}

RetryPolicy bounds automatic retries of safe, read-only requests.

type Router 

type Router struct {
	// contains filtered or unexported fields
}

Router caches table and partition leaders and coalesces concurrent refreshes. A Router is safe for concurrent use.

func NewRouter 

func NewRouter(coordinator Node, fetch MetadataFetcher) *Router

NewRouter creates an empty metadata router.

func (*Router) Coordinator 

func (r *Router) Coordinator() Node

Coordinator returns the current coordinator snapshot.

func (*Router) Invalidate 

func (r *Router) Invalidate(path TablePath)

Invalidate removes cached metadata for path.

func (*Router) InvalidatePhysical 

func (r *Router) InvalidatePhysical(path PhysicalTablePath)

InvalidatePhysical removes one partition while retaining the table's other cached locations.

func (*Router) Refresh 

func (r *Router) Refresh(ctx context.Context, path TablePath) error

Refresh replaces cached metadata for path with an authoritative snapshot.

func (*Router) RefreshPhysical 

func (r *Router) RefreshPhysical(ctx context.Context, path PhysicalTablePath) error

RefreshPhysical refreshes exactly one named partition. When a dedicated physical fetcher is unavailable, it performs one table refresh and uses the partition included in that response.

func (*Router) Route 

func (r *Router) Route(ctx context.Context, path TablePath, bucket int32) (Node, error)

Route returns the current tablet leader for a logical table bucket, refreshing missing metadata once.

func (*Router) RouteAfterMetadataError 

func (r *Router) RouteAfterMetadataError(ctx context.Context, path TablePath, bucket int32, cause error) (Node, error)

RouteAfterMetadataError invalidates one stale cache entry and performs at most one refresh. Callers should use it only for errors matching ErrMetadata.

func (*Router) RoutePhysical 

func (r *Router) RoutePhysical(ctx context.Context, path PhysicalTablePath, bucket int32) (Node, error)

RoutePhysical returns the leader for a bucket in a named partition. An empty partition names the unpartitioned table and behaves like Route.

func (*Router) WithPhysicalMetadataFetcher 

func (r *Router) WithPhysicalMetadataFetcher(fetch PhysicalMetadataFetcher) *Router

WithPhysicalMetadataFetcher configures refreshes for individual partitions. It is intended for construction time, before the router is shared by requests.

type Row 

type Row []any

Row stores values in schema or projection order.

func DecodeCompactedProjectedRow 

func DecodeCompactedProjectedRow(schema Schema, columns []string, encoded []byte) (Row, error)

DecodeCompactedProjectedRow decodes compacted values for columns.

func DecodeCompactedRow 

func DecodeCompactedRow(schema Schema, encoded []byte) (Row, error)

DecodeCompactedRow decodes a complete compacted Fluss row.

func DecodeIndexedProjectedRow 

func DecodeIndexedProjectedRow(schema Schema, columns []string, encoded []byte) (Row, error)

DecodeIndexedProjectedRow decodes indexed values for columns.

func DecodeIndexedRow 

func DecodeIndexedRow(schema Schema, encoded []byte) (Row, error)

DecodeIndexedRow decodes a complete indexed Fluss row.

type ScanArrowBatch 

type ScanArrowBatch struct {
	// Bucket identifies the source table bucket.
	Bucket int32
	// Batch is owned by the enclosing [ScanResult].
	Batch ArrowLogBatch
}

ScanArrowBatch associates one owned Arrow batch with its source bucket.

type ScanOffset 

type ScanOffset struct {
	// Kind selects which of Offset or Timestamp is meaningful.
	Kind ScanOffsetKind
	// Offset is an inclusive non-negative start for [ScanFromOffset].
	Offset int64
	// Timestamp is used only for [ScanFromTimestamp].
	Timestamp time.Time
}

ScanOffset describes an explicit, symbolic, or timestamp-based start.

func AtOffset 

func AtOffset(offset int64) ScanOffset

AtOffset starts at an explicit inclusive log offset.

func AtTimestamp 

func AtTimestamp(timestamp time.Time) ScanOffset

AtTimestamp starts at the first offset whose timestamp is not before timestamp.

func Earliest 

func Earliest() ScanOffset

Earliest starts at the oldest available log offset.

func Latest 

func Latest() ScanOffset

Latest starts at the current log end.

func (ScanOffset) Validate 

func (s ScanOffset) Validate() error

Validate checks that exactly the fields required by Kind are set.

type ScanOffsetKind 

type ScanOffsetKind uint8

ScanOffsetKind identifies how a scanner resolves its initial position. 

const (
	ScanFromOffset ScanOffsetKind = iota
	ScanFromEarliest
	ScanFromLatest
	ScanFromTimestamp
)

Supported initial-position strategies for log scans.

type ScanRecord 

type ScanRecord struct {
	// Bucket identifies the source table bucket.
	Bucket int32
	// Record contains the decoded row and log metadata.
	Record Record
}

ScanRecord associates one decoded row record with its source bucket.

type ScanResult 

type ScanResult struct {
	// Records contains successfully decoded row records.
	Records []ScanRecord
	// ArrowBatches contains owned batches released by [ScanResult.Release].
	ArrowBatches []ScanArrowBatch
	// BucketErrors contains failures that did not invalidate other buckets.
	BucketErrors []BucketScanError
	// HighWatermark maps bucket IDs to the observed log end offset.
	HighWatermark map[int32]int64
	// Done reports that configured row or stopping-offset bounds were reached.
	Done bool
}

ScanResult contains rows, owned Arrow batches, and per-bucket outcomes from one poll.

func (*ScanResult) Release 

func (r *ScanResult) Release()

Release frees Arrow records owned by the result. Release is safe to call more than once.

type Schema 

type Schema struct {
	// Columns are stored in row encoding order.
	Columns []Column
	// PrimaryKey lists primary-key column names in key encoding order.
	PrimaryKey []string
	// BucketKey lists columns used for bucket hashing.
	BucketKey []string
	// PartitionKey lists columns used to name physical partitions.
	PartitionKey []string
	// AutoIncrement lists server-generated columns.
	AutoIncrement []string
	// HighestFieldID is the greatest assigned stable field identifier.
	HighestFieldID int
}

Schema describes ordered columns and table key definitions.

func ParseSchemaJSON 

func ParseSchemaJSON(data []byte) (Schema, error)

ParseSchemaJSON decodes and validates a Fluss 0.9.1 schema.

func SchemaFromArrow 

func SchemaFromArrow(schema *arrow.Schema) (Schema, error)

SchemaFromArrow reconstructs all logical parameters represented by Arrow. CHAR is returned as STRING and local-zoned timestamps as TIMESTAMP because Arrow does not retain those distinctions.

func (Schema) ArrowSchema 

func (s Schema) ArrowSchema() (*arrow.Schema, error)

ArrowSchema converts a Fluss schema to the exact Arrow field layout used by Fluss 0.9.1.

func (Schema) JSON 

func (s Schema) JSON() ([]byte, error)

JSON validates and encodes the Fluss 0.9.1 schema JSON representation.

func (Schema) PartitionName 

func (s Schema) PartitionName(spec PartitionSpec) (string, error)

PartitionName resolves a spec in schema partition-key order. Fluss 0.9.1 joins values with "$".

func (Schema) PartitionNames 

func (s Schema) PartitionNames(specs ...PartitionSpec) ([]string, error)

PartitionNames resolves, deduplicates, and sorts specs for deterministic multi-partition use.

func (Schema) RowFromNamed 

func (s Schema) RowFromNamed(row NamedRow) (Row, error)

RowFromNamed validates a named row and returns values in schema order.

func (Schema) Validate 

func (s Schema) Validate() error

Validate checks columns, logical types, and key references.

func (Schema) ValidatePrimaryKey 

func (s Schema) ValidatePrimaryKey(key PrimaryKey) error

ValidatePrimaryKey checks key values against primary-key columns.

func (Schema) ValidateRow 

func (s Schema) ValidateRow(row Row, columns []string) error

ValidateRow checks row values against all columns or the named projection.

type ServerError 

type ServerError struct {
	// Code is the numeric Fluss protocol error code.
	Code fmsg.ErrorCode
	// Name is the stable Fluss error name when known.
	Name string
	// Message is the credential-safe server message.
	Message string
	// API identifies the failed operation.
	API fmsg.APIKey
	// Endpoint identifies the server when available.
	Endpoint string
	// Retriable reports whether the protocol class permits retry.
	Retriable bool
	// contains filtered or unexported fields
}

ServerError is a Fluss 0.9.1 server failure with safe request context. Unknown future error codes remain inspectable and are never retriable by default.

Example 
package main

import (
	"errors"
	"fmt"

	"github.com/pletorco/fluss-go/pkg/fgo"
	"github.com/pletorco/fluss-go/pkg/fmsg"
)

func main() {
	err := fgo.ResponseError(
		int32(fmsg.ErrorCodeRequestTimeOut),
		"tablet request timed out",
		fmsg.APIKeyListOffsets,
	)
	var server *fgo.ServerError
	fmt.Println(errors.Is(err, fgo.ErrServerFailure))
	fmt.Println(errors.Is(err, fgo.ErrTimeout))
	fmt.Println(errors.As(err, &server), server.Retriable)
}

Output:
true
true
true true

func (*ServerError) Error 

func (e *ServerError) Error() string

Error returns a credential-safe server error summary.

func (*ServerError) Is 

func (e *ServerError) Is(target error) bool

Is reports whether target matches the server error classification.

type ServerRole 

type ServerRole int32

ServerRole identifies a Fluss RPC server advertised by ApiVersions. 

const (
	// UnknownServerRole represents an unrecognized server role.
	UnknownServerRole ServerRole = -1
	Coordinator       ServerRole = 1
	TabletServer      ServerRole = 2
)

Fluss server roles reported during API negotiation.

func (ServerRole) String 

func (r ServerRole) String() string

String returns a stable human-readable server role.

type SnapshotBatchProvider 

type SnapshotBatchProvider interface {
	// OpenSnapshot returns a new reader owned by the caller.
	OpenSnapshot(context.Context, SnapshotBatchRequest) (SnapshotBatchReader, error)
}

SnapshotBatchProvider opens readers for implementation-specific Fluss snapshot storage.

type SnapshotBatchProviderFunc 

type SnapshotBatchProviderFunc func(context.Context, SnapshotBatchRequest) (SnapshotBatchReader, error)

SnapshotBatchProviderFunc adapts a function to SnapshotBatchProvider.

func (SnapshotBatchProviderFunc) OpenSnapshot 

func (f SnapshotBatchProviderFunc) OpenSnapshot(
	ctx context.Context,
	request SnapshotBatchRequest,
) (SnapshotBatchReader, error)

OpenSnapshot calls f with the requested snapshot and projection.

type SnapshotBatchReader 

type SnapshotBatchReader interface {
	// ReadBatch returns at most limit rows. io.EOF marks completion and may
	// accompany a final non-empty batch.
	ReadBatch(context.Context, int) ([]Row, error)
	// Close releases reader resources and is safe after completion.
	Close() error
}

SnapshotBatchReader streams rows from one immutable snapshot. io.EOF marks completion.

type SnapshotBatchRequest 

type SnapshotBatchRequest struct {
	// Table is the authoritative table metadata for the snapshot.
	Table Table
	// Bucket identifies the physical bucket being read.
	Bucket TableBucket
	// SnapshotID identifies the immutable server snapshot.
	SnapshotID int64
	// Projection lists requested columns in result order.
	Projection []string
	// Limit is the maximum rows requested from one reader call.
	Limit int
}

SnapshotBatchRequest identifies one immutable primary-key snapshot.

type Table 

type Table struct {
	// ID is the server-assigned logical table identifier.
	ID int64
	// SchemaID identifies the current schema version.
	SchemaID int32
	// Path is the logical database and table name.
	Path TablePath
	// Kind selects log or primary-key operations.
	Kind TableKind
	// Schema is the authoritative current schema.
	Schema Schema
	// BucketCount is the logical table bucket count.
	BucketCount int
	// Properties contains server-reported table properties.
	Properties map[string]string
}

Table is authoritative table metadata and schema loaded by Client.OpenTable.

func (Table) RequireLog 

func (t Table) RequireLog() error

RequireLog reports an error unless the table supports log operations.

func (Table) RequirePrimaryKey 

func (t Table) RequirePrimaryKey() error

RequirePrimaryKey reports an error unless the table supports primary-key operations.

type TableBucket 

type TableBucket struct {
	// TableID is the server-assigned table identifier.
	TableID int64
	// PartitionID is -1 for an unpartitioned table.
	PartitionID int64
	// BucketID identifies the logical table bucket.
	BucketID int32
	// Leader is the current tablet server leader.
	Leader Node
}

TableBucket is a resolved bucket and its current tablet leader.

func (TableBucket) Validate 

func (b TableBucket) Validate() error

Validate checks IDs and leader metadata.

type TableInfo 

type TableInfo struct {
	// ID is the server-assigned logical table identifier.
	ID int64
	// Table contains authoritative metadata and schema.
	Table Table
	// SchemaID identifies the current schema version.
	SchemaID int64
}

TableInfo combines table identity, schema identity, and table metadata.

type TableKind 

type TableKind string

TableKind distinguishes append-only log tables from primary-key tables. 

const (
	LogTable        TableKind = "LOG"
	PrimaryKeyTable TableKind = "PRIMARY_KEY"
)

Table kinds supported by Apache Fluss 0.9.1.

type TableMetadata 

type TableMetadata struct {
	// Path is the logical table path.
	Path TablePath
	// ID is the server-assigned table identifier.
	ID int64
	// SchemaID identifies the current schema version.
	SchemaID int32
	// Buckets maps bucket IDs to current tablet leaders.
	Buckets map[int32]Node
	// Partitions maps canonical partition names to physical metadata.
	Partitions map[string]PartitionMetadata
	// contains filtered or unexported fields
}

TableMetadata contains table IDs, bucket leaders, and named partitions.

type TablePath 

type TablePath struct {
	// Database is the logical database name.
	Database string
	// Table is the logical table name.
	Table string
}

TablePath identifies a logical table by database and table name.

func (TablePath) String 

func (p TablePath) String() string

String returns the canonical database.table path.

func (TablePath) Validate 

func (p TablePath) Validate() error

Validate checks that both path components are present.

type TypedBatchResult 

type TypedBatchResult[T any] struct {
	// Values contains successfully decoded rows from one poll.
	Values []T
	// Done reports that the bounded scan has completed.
	Done bool
}

TypedBatchResult contains decoded values from one bounded poll.

type TypedBatchScanner 

type TypedBatchScanner[T any] struct {
	// contains filtered or unexported fields
}

TypedBatchScanner decodes bounded current-state or snapshot scan rows.

func NewTypedBatchScanner 

func NewTypedBatchScanner[T any](
	ctx context.Context,
	client *Client,
	table Table,
	bucket TableBucket,
	codec Codec[T],
	options ...BatchScannerOption,
) (*TypedBatchScanner[T], error)

NewTypedBatchScanner opens a typed current-state scanner for one table bucket.

func NewTypedSnapshotBatchScanner 

func NewTypedSnapshotBatchScanner[T any](
	ctx context.Context,
	client *Client,
	table Table,
	bucket TableBucket,
	snapshotID int64,
	codec Codec[T],
	options ...BatchScannerOption,
) (*TypedBatchScanner[T], error)

NewTypedSnapshotBatchScanner opens a typed scanner for one immutable snapshot.

func WrapTypedBatchScanner 

func WrapTypedBatchScanner[T any](
	scanner *BatchScanner,
	codec Codec[T],
) (*TypedBatchScanner[T], error)

WrapTypedBatchScanner wraps an existing row-oriented batch scanner.

func (*TypedBatchScanner[T]) Close 

func (s *TypedBatchScanner[T]) Close() error

Close closes the wrapped scanner or snapshot reader.

func (*TypedBatchScanner[T]) Done 

func (s *TypedBatchScanner[T]) Done() bool

Done reports whether the bounded scan has completed.

func (*TypedBatchScanner[T]) Poll 

func (s *TypedBatchScanner[T]) Poll(ctx context.Context) (TypedBatchResult[T], error)

Poll returns the next decoded bounded-scan result. A decode failure returns no partial decoded values.

type TypedKVWriter 

type TypedKVWriter[T, K any] struct {
	// contains filtered or unexported fields
}

TypedKVWriter encodes application values and delete keys for a primary-key table.

func NewTypedKVWriter 

func NewTypedKVWriter[T, K any](
	ctx context.Context,
	client *Client,
	table Table,
	codec Codec[T],
	keyCodec KeyCodec[K],
	options ...KVWriterOption,
) (*TypedKVWriter[T, K], error)

NewTypedKVWriter opens a primary-key writer that encodes values and delete keys.

func (*TypedKVWriter[T, K]) Close 

func (w *TypedKVWriter[T, K]) Close(ctx context.Context) error

Close flushes accepted mutations and closes the wrapped writer.

func (*TypedKVWriter[T, K]) Delete 

func (w *TypedKVWriter[T, K]) Delete(ctx context.Context, key K) *WriteFuture

Delete encodes key and queues a primary-key delete. Encoding failures are returned through the completed future.

func (*TypedKVWriter[T, K]) Flush 

func (w *TypedKVWriter[T, K]) Flush(ctx context.Context) error

Flush waits until all mutations accepted before the call have completed.

func (*TypedKVWriter[T, K]) Upsert 

func (w *TypedKVWriter[T, K]) Upsert(ctx context.Context, value T) *WriteFuture

Upsert encodes value and queues a primary-key upsert. Encoding failures are returned through the completed future.

type TypedLogScanner 

type TypedLogScanner[T any] struct {
	// contains filtered or unexported fields
}

TypedLogScanner decodes row-based log scan results.

func NewTypedLogScanner 

func NewTypedLogScanner[T any](
	ctx context.Context,
	client *Client,
	table Table,
	start ScanOffset,
	codec Codec[T],
	options ...LogScannerOption,
) (*TypedLogScanner[T], error)

NewTypedLogScanner opens a log scanner that decodes every row with codec.

func (*TypedLogScanner[T]) Close 

func (s *TypedLogScanner[T]) Close() error

Close interrupts blocked work and releases scanner resources.

func (*TypedLogScanner[T]) Done 

func (s *TypedLogScanner[T]) Done() bool

Done reports whether configured scan bounds have been reached.

func (*TypedLogScanner[T]) Poll 

func (s *TypedLogScanner[T]) Poll(ctx context.Context) (TypedScanResult[T], error)

Poll waits for the next decoded result. A decode failure returns no partial decoded records.

func (*TypedLogScanner[T]) Wakeup 

func (s *TypedLogScanner[T]) Wakeup()

Wakeup interrupts a blocked Poll with ErrWakeup.

type TypedLogWriter 

type TypedLogWriter[T any] struct {
	// contains filtered or unexported fields
}

TypedLogWriter encodes application values before appending them.

func NewTypedLogWriter 

func NewTypedLogWriter[T any](
	ctx context.Context,
	client *Client,
	table Table,
	codec Codec[T],
	options ...LogWriterOption,
) (*TypedLogWriter[T], error)

NewTypedLogWriter opens a log writer that encodes application values with codec.

Example 
package main

import (
	"context"
	"log"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	type event struct {
		ID   int64
		Kind string
	}

	codec := fgo.CodecFuncs[event]{
		EncodeFunc: func(value event) (fgo.Row, error) {
			return fgo.Row{value.ID, value.Kind}, nil
		},
		DecodeFunc: func(row fgo.Row) (event, error) {
			return event{ID: row[0].(int64), Kind: row[1].(string)}, nil
		},
	}

	ctx := context.Background()
	client, err := fgo.Open(ctx, fgo.WithSeedBrokers("coordinator.example:9123"))
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	table, err := client.OpenTable(ctx, fgo.TablePath{
		Database: "production",
		Table:    "events",
	})
	if err != nil {
		log.Fatal(err)
	}
	writer, err := fgo.NewTypedLogWriter(ctx, client, table, codec)
	if err != nil {
		log.Fatal(err)
	}
	defer writer.Close(ctx)

	result := writer.Append(ctx, event{ID: 42, Kind: "created"}).Await(ctx)
	if result.Err != nil {
		log.Fatal(result.Err)
	}
}

func (*TypedLogWriter[T]) Append 

func (w *TypedLogWriter[T]) Append(ctx context.Context, value T) *WriteFuture

Append encodes value and queues it using the wrapped writer. Encoding failures are returned through the completed future.

func (*TypedLogWriter[T]) Close 

func (w *TypedLogWriter[T]) Close(ctx context.Context) error

Close flushes accepted values and closes the wrapped writer.

func (*TypedLogWriter[T]) Flush 

func (w *TypedLogWriter[T]) Flush(ctx context.Context) error

Flush waits until all values accepted before the call have completed.

type TypedLookupClient 

type TypedLookupClient[T, K any] struct {
	// contains filtered or unexported fields
}

TypedLookupClient encodes application keys and decodes returned rows.

func NewTypedLookupClient 

func NewTypedLookupClient[T, K any](
	ctx context.Context,
	client *Client,
	table Table,
	codec Codec[T],
	keyCodec KeyCodec[K],
	options ...LookupOption,
) (*TypedLookupClient[T, K], error)

NewTypedLookupClient opens a lookup client that encodes keys and decodes rows.

func (*TypedLookupClient[T, K]) Close 

func (c *TypedLookupClient[T, K]) Close() error

Close prevents new lookups and releases wrapped client resources.

func (*TypedLookupClient[T, K]) Lookup 

func (c *TypedLookupClient[T, K]) Lookup(ctx context.Context, keys ...K) []TypedLookupResult[K, T]

Lookup returns one result per input key in input order. A failure for one key does not discard results for other keys.

func (*TypedLookupClient[T, K]) PrefixLookup 

func (c *TypedLookupClient[T, K]) PrefixLookup(
	ctx context.Context,
	prefixes ...K,
) []TypedPrefixLookupResult[K, T]

PrefixLookup returns one result per input prefix in input order. A failure for one prefix does not discard results for other prefixes.

type TypedLookupResult 

type TypedLookupResult[K, T any] struct {
	// Key is the application key associated with this result.
	Key K
	// Found reports whether Value contains a server row.
	Found bool
	// Value is valid only when Found is true and Err is nil.
	Value T
	// Err is a key-local encoding, request, or decoding failure.
	Err error
}

TypedLookupResult is one decoded point-lookup outcome.

type TypedPrefixLookupResult 

type TypedPrefixLookupResult[K, T any] struct {
	// Prefix is the application prefix associated with this result.
	Prefix K
	// Values contains decoded matches when Err is nil.
	Values []T
	// Err is a prefix-local encoding, request, or decoding failure.
	Err error
}

TypedPrefixLookupResult contains decoded rows for one requested key prefix.

type TypedScanRecord 

type TypedScanRecord[T any] struct {
	// Bucket identifies the source table bucket.
	Bucket int32
	// Value is the decoded row value.
	Value T
	// Change identifies the row mutation represented by the record.
	Change ChangeType
	// Timestamp is the server commit timestamp.
	Timestamp time.Time
	// Offset is the record offset within Bucket.
	Offset int64
}

TypedScanRecord is one decoded log record and its source metadata.

type TypedScanResult 

type TypedScanResult[T any] struct {
	// Records contains successfully decoded records.
	Records []TypedScanRecord[T]
	// BucketErrors contains failures that did not invalidate other buckets.
	BucketErrors []BucketScanError
	// HighWatermark maps bucket IDs to the observed log end offset.
	HighWatermark map[int32]int64
	// Done reports that configured row or stopping-offset bounds were reached.
	Done bool
}

TypedScanResult is a decoded log poll result.

type WriteFuture 

type WriteFuture struct {
	// contains filtered or unexported fields
}

WriteFuture represents one queued mutation. Await may be called concurrently and does not consume the result.

func (*WriteFuture) Await 

func (f *WriteFuture) Await(ctx context.Context) WriteResult

Await waits for the mutation result or ctx cancellation. Cancellation stops waiting but does not cancel a mutation already in flight.

type WriteResult 

type WriteResult struct {
	// Bucket is the target bucket, or zero when routing did not complete.
	Bucket int32
	// BaseOffset is the first assigned offset when OffsetKnown is true.
	BaseOffset int64
	// OffsetKnown reports whether the successful response included an offset.
	// A recovered duplicate-sequence acknowledgement can succeed without one.
	OffsetKnown bool
	// Records is the number of records completed by this result.
	Records int
	// Err is the terminal mutation error.
	Err error
}

WriteResult is the terminal outcome of one queued mutation.

Example 
package main

import (
	"errors"
	"fmt"

	"github.com/pletorco/fluss-go/pkg/fgo"
)

func main() {
	result := fgo.WriteResult{
		Bucket: 2,
		Err:    fmt.Errorf("%w: reconcile the previous batch", fgo.ErrWriterState),
	}
	if errors.Is(result.Err, fgo.ErrWriterState) {
		fmt.Printf("bucket %d requires reconciliation\n", result.Bucket)
	}
}

Output:
bucket 2 requires reconciliation

type WriterRetryPolicy 

type WriterRetryPolicy struct {
	// MaxAttempts includes the initial request.
	MaxAttempts int
	// Backoff returns the delay before the next attempt. The argument is the
	// failed attempt number, starting at one.
	Backoff func(attempt int) time.Duration
}

WriterRetryPolicy controls bounded retries of idempotent writer batches. The first request counts as an attempt. Retries preserve the writer ID, bucket sequence, and encoded batch bytes.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.