blob: d39b6840a5ee2846529e3aa7dad8c98e7d20de57 [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package paimon
import (
"context"
"fmt"
"math"
"runtime"
"sync"
"unsafe"
"github.com/jupiterrider/ffi"
)
// Datum type tags (must match paimon-c datum_from_c).
const (
datumTagBool int32 = 0
datumTagTinyInt int32 = 1
datumTagSmallInt int32 = 2
datumTagInt int32 = 3
datumTagLong int32 = 4
datumTagFloat int32 = 5
datumTagDouble int32 = 6
datumTagString int32 = 7
datumTagDate int32 = 8
datumTagTime int32 = 9
datumTagTimestamp int32 = 10
datumTagLocalZonedTimestamp int32 = 11
datumTagDecimal int32 = 12
datumTagBytes int32 = 13
)
// Datum is a typed literal value for predicate comparison.
// The internal representation is hidden to allow future changes
// (e.g. switching to opaque handles) without breaking callers.
type Datum struct {
inner paimonDatumC
}
// BoolDatum creates a boolean datum.
func BoolDatum(v bool) Datum {
var iv int64
if v {
iv = 1
}
return Datum{inner: paimonDatumC{tag: datumTagBool, intVal: iv}}
}
// TinyIntDatum creates a tinyint datum.
func TinyIntDatum(v int8) Datum {
return Datum{inner: paimonDatumC{tag: datumTagTinyInt, intVal: int64(v)}}
}
// SmallIntDatum creates a smallint datum.
func SmallIntDatum(v int16) Datum {
return Datum{inner: paimonDatumC{tag: datumTagSmallInt, intVal: int64(v)}}
}
// IntDatum creates an int datum.
func IntDatum(v int32) Datum {
return Datum{inner: paimonDatumC{tag: datumTagInt, intVal: int64(v)}}
}
// LongDatum creates a long (bigint) datum.
func LongDatum(v int64) Datum {
return Datum{inner: paimonDatumC{tag: datumTagLong, intVal: v}}
}
// FloatDatum creates a float datum.
func FloatDatum(v float32) Datum {
return Datum{inner: paimonDatumC{tag: datumTagFloat, dblVal: float64(v)}}
}
// DoubleDatum creates a double datum.
func DoubleDatum(v float64) Datum {
return Datum{inner: paimonDatumC{tag: datumTagDouble, dblVal: v}}
}
// StringDatum creates a string datum.
func StringDatum(v string) Datum {
b := []byte(v)
d := paimonDatumC{tag: datumTagString, strLen: uintptr(len(b))}
if len(b) > 0 {
d.strData = &b[0]
}
return Datum{inner: d}
}
// Date represents a date value as epoch days since 1970-01-01.
// Usage: table.PredicateEqual("dt", paimon.Date(19000))
type Date int32
// Time represents a time-of-day value as milliseconds since midnight.
// Usage: table.PredicateEqual("t", paimon.Time(3600000))
type Time int32
// Timestamp represents a timestamp without timezone (millis + sub-millis nanos).
// Usage: table.PredicateEqual("ts", paimon.Timestamp{Millis: 1700000000000, Nanos: 0})
type Timestamp struct {
Millis int64
Nanos int32
}
// LocalZonedTimestamp represents a timestamp with local timezone semantics.
// Usage: table.PredicateEqual("lzts", paimon.LocalZonedTimestamp{Millis: 1700000000000, Nanos: 0})
type LocalZonedTimestamp struct {
Millis int64
Nanos int32
}
// Decimal represents a fixed-precision decimal value up to DECIMAL(38, s).
//
// The unscaled value is stored as a little-endian i128 split into two int64
// halves: Lo (low 64 bits, unsigned interpretation) and Hi (high 64 bits,
// sign-extended). For values that fit in int64, use [NewDecimal].
//
// Usage:
//
// paimon.NewDecimal(12345, 10, 2) // 123.45 as DECIMAL(10,2)
// paimon.Decimal{Lo: lo, Hi: hi, ...} // full i128
type Decimal struct {
Lo int64 // low 64 bits of unscaled i128 (unsigned interpretation)
Hi int64 // high 64 bits of unscaled i128 (sign extension)
Precision uint32
Scale uint32
}
// NewDecimal creates a Decimal from an int64 unscaled value.
// For unscaled values that exceed int64 range, construct Decimal directly
// with Lo/Hi fields.
func NewDecimal(unscaled int64, precision, scale uint32) Decimal {
hi := int64(0)
if unscaled < 0 {
hi = -1
}
return Decimal{Lo: unscaled, Hi: hi, Precision: precision, Scale: scale}
}
// Bytes represents a binary value.
// Usage: table.PredicateEqual("data", paimon.Bytes(someSlice))
type Bytes []byte
// toDatum converts a Go value to a Datum for predicate comparison.
//
// Supported Go types and their Paimon mappings:
// - bool → Bool
// - int8 → TinyInt
// - int16 → SmallInt
// - int32 → Int
// - int → Int (if fits int32) or Long
// - int64 → Long
// - float32 → Float
// - float64 → Double
// - string → String
// - Date → Date
// - Time → Time
// - Timestamp → Timestamp
// - LocalZonedTimestamp → LocalZonedTimestamp
// - Decimal → Decimal
// - Bytes → Bytes
// - Datum → passed through
func toDatum(v any) (Datum, error) {
switch val := v.(type) {
case bool:
return BoolDatum(val), nil
case int8:
return TinyIntDatum(val), nil
case int16:
return SmallIntDatum(val), nil
case int32:
return IntDatum(val), nil
case int:
if val >= math.MinInt32 && val <= math.MaxInt32 {
return IntDatum(int32(val)), nil
}
return LongDatum(int64(val)), nil
case int64:
return LongDatum(val), nil
case float32:
return FloatDatum(val), nil
case float64:
return DoubleDatum(val), nil
case string:
return StringDatum(val), nil
case Date:
return Datum{inner: paimonDatumC{tag: datumTagDate, intVal: int64(val)}}, nil
case Time:
return Datum{inner: paimonDatumC{tag: datumTagTime, intVal: int64(val)}}, nil
case Timestamp:
return Datum{inner: paimonDatumC{tag: datumTagTimestamp, intVal: val.Millis, intVal2: int64(val.Nanos)}}, nil
case LocalZonedTimestamp:
return Datum{inner: paimonDatumC{tag: datumTagLocalZonedTimestamp, intVal: val.Millis, intVal2: int64(val.Nanos)}}, nil
case Decimal:
return Datum{inner: paimonDatumC{
tag: datumTagDecimal, intVal: val.Lo, intVal2: val.Hi,
uintVal: val.Precision, uintVal2: val.Scale,
}}, nil
case Bytes:
d := paimonDatumC{tag: datumTagBytes, strLen: uintptr(len(val))}
if len(val) > 0 {
d.strData = &val[0]
}
return Datum{inner: d}, nil
case Datum:
return val, nil
default:
return Datum{}, fmt.Errorf("unsupported datum type: %T", v)
}
}
// Predicate is an opaque filter predicate for scan planning.
type Predicate struct {
ctx context.Context
lib *libRef
inner *paimonPredicate
closeOnce sync.Once
}
// Close releases the predicate resources. Safe to call multiple times.
// Note: predicates passed to WithFilter or combinators (And/Or/Not) are consumed
// and should NOT be closed by the caller.
func (p *Predicate) Close() {
p.closeOnce.Do(func() {
if p.inner != nil {
ffiPredicateFree.symbol(p.ctx)(p.inner)
p.inner = nil
p.lib.release()
}
})
}
// errConsumedPredicate is returned when a consumed or nil predicate is reused.
var errConsumedPredicate = fmt.Errorf("paimon: predicate already consumed or nil")
// PredicateBuilder creates filter predicates for a table.
// It holds a Go-level reference to the Table and does not own any C resources,
// so there is no Close() method.
type PredicateBuilder struct {
table *Table
}
// Eq creates an equality predicate: column = value.
func (pb *PredicateBuilder) Eq(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateEqual, column, datum)
}
// NotEq creates a not-equal predicate: column != value.
func (pb *PredicateBuilder) NotEq(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateNotEqual, column, datum)
}
// Lt creates a less-than predicate: column < value.
func (pb *PredicateBuilder) Lt(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateLessThan, column, datum)
}
// Le creates a less-or-equal predicate: column <= value.
func (pb *PredicateBuilder) Le(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateLessOrEqual, column, datum)
}
// Gt creates a greater-than predicate: column > value.
func (pb *PredicateBuilder) Gt(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateGreaterThan, column, datum)
}
// Ge creates a greater-or-equal predicate: column >= value.
func (pb *PredicateBuilder) Ge(column string, value any) (*Predicate, error) {
datum, err := toDatum(value)
if err != nil {
return nil, err
}
return pb.buildLeafPredicate(ffiPredicateGreaterOrEqual, column, datum)
}
// IsNull creates an IS NULL predicate.
func (pb *PredicateBuilder) IsNull(column string) (*Predicate, error) {
return pb.buildNullPredicate(ffiPredicateIsNull, column)
}
// IsNotNull creates an IS NOT NULL predicate.
func (pb *PredicateBuilder) IsNotNull(column string) (*Predicate, error) {
return pb.buildNullPredicate(ffiPredicateIsNotNull, column)
}
// In creates an IN predicate: column IN (values...).
func (pb *PredicateBuilder) In(column string, values ...any) (*Predicate, error) {
return pb.buildInPredicate(ffiPredicateIsIn, column, values)
}
// NotIn creates a NOT IN predicate: column NOT IN (values...).
func (pb *PredicateBuilder) NotIn(column string, values ...any) (*Predicate, error) {
return pb.buildInPredicate(ffiPredicateIsNotIn, column, values)
}
// buildLeafPredicate is a helper for comparison predicates that take (table, column, datum).
func (pb *PredicateBuilder) buildLeafPredicate(
ffiVar *FFI[func(*paimonTable, *byte, paimonDatumC) (*paimonPredicate, error)],
column string, datum Datum,
) (*Predicate, error) {
t := pb.table
if t.inner == nil {
return nil, ErrClosed
}
createFn := ffiVar.symbol(t.ctx)
cCol := append([]byte(column), 0)
inner, err := createFn(t.inner, &cCol[0], datum.inner)
runtime.KeepAlive(cCol)
runtime.KeepAlive(datum)
if err != nil {
return nil, err
}
t.lib.acquire()
return &Predicate{ctx: t.ctx, lib: t.lib, inner: inner}, nil
}
// buildNullPredicate is a helper for IS NULL / IS NOT NULL predicates.
func (pb *PredicateBuilder) buildNullPredicate(
ffiVar *FFI[func(*paimonTable, *byte) (*paimonPredicate, error)],
column string,
) (*Predicate, error) {
t := pb.table
if t.inner == nil {
return nil, ErrClosed
}
createFn := ffiVar.symbol(t.ctx)
cCol := append([]byte(column), 0)
inner, err := createFn(t.inner, &cCol[0])
runtime.KeepAlive(cCol)
if err != nil {
return nil, err
}
t.lib.acquire()
return &Predicate{ctx: t.ctx, lib: t.lib, inner: inner}, nil
}
// buildInPredicate is a helper for IS IN / IS NOT IN predicates.
func (pb *PredicateBuilder) buildInPredicate(
ffiVar *FFI[func(*paimonTable, *byte, unsafe.Pointer, uintptr) (*paimonPredicate, error)],
column string, values []any,
) (*Predicate, error) {
t := pb.table
if t.inner == nil {
return nil, ErrClosed
}
datums := make([]paimonDatumC, len(values))
for i, v := range values {
d, err := toDatum(v)
if err != nil {
return nil, err
}
datums[i] = d.inner
}
createFn := ffiVar.symbol(t.ctx)
cCol := append([]byte(column), 0)
var datumsPtr unsafe.Pointer
if len(datums) > 0 {
datumsPtr = unsafe.Pointer(&datums[0])
}
inner, err := createFn(t.inner, &cCol[0], datumsPtr, uintptr(len(datums)))
runtime.KeepAlive(cCol)
runtime.KeepAlive(datums)
runtime.KeepAlive(values)
if err != nil {
return nil, err
}
t.lib.acquire()
return &Predicate{ctx: t.ctx, lib: t.lib, inner: inner}, nil
}
// combinePredicate is a shared helper for And/Or.
func (p *Predicate) combinePredicate(
other *Predicate,
ffiVar *FFI[func(*paimonPredicate, *paimonPredicate) *paimonPredicate],
) (*Predicate, error) {
if p == nil || p.inner == nil {
return nil, errConsumedPredicate
}
if other == nil || other.inner == nil {
return nil, errConsumedPredicate
}
if p == other {
return nil, fmt.Errorf("paimon: cannot combine a predicate with itself")
}
combineFn := ffiVar.symbol(p.ctx)
p.inner = combineFn(p.inner, other.inner)
other.inner = nil
other.lib.release()
return p, nil
}
// And combines this predicate with another using AND. Consumes both predicates
// (callers must NOT close either after this call).
func (p *Predicate) And(other *Predicate) (*Predicate, error) {
return p.combinePredicate(other, ffiPredicateAnd)
}
// Or combines this predicate with another using OR. Consumes both predicates
// (callers must NOT close either after this call).
func (p *Predicate) Or(other *Predicate) (*Predicate, error) {
return p.combinePredicate(other, ffiPredicateOr)
}
// Not negates this predicate. Consumes the input
// (caller must NOT close it after this call).
func (p *Predicate) Not() (*Predicate, error) {
if p == nil || p.inner == nil {
return nil, errConsumedPredicate
}
negateFn := ffiPredicateNot.symbol(p.ctx)
p.inner = negateFn(p.inner)
return p, nil
}
// FFI wrappers for predicate functions.
var ffiPredicateFree = newFFI(ffiOpts{
sym: "paimon_predicate_free",
rType: &ffi.TypeVoid,
aTypes: []*ffi.Type{&ffi.TypePointer},
}, func(_ context.Context, ffiCall ffiCall) func(p *paimonPredicate) {
return func(p *paimonPredicate) {
ffiCall(nil, unsafe.Pointer(&p))
}
})
var ffiPredicateEqual = newPredicateLeafFFI("paimon_predicate_equal")
var ffiPredicateNotEqual = newPredicateLeafFFI("paimon_predicate_not_equal")
var ffiPredicateLessThan = newPredicateLeafFFI("paimon_predicate_less_than")
var ffiPredicateLessOrEqual = newPredicateLeafFFI("paimon_predicate_less_or_equal")
var ffiPredicateGreaterThan = newPredicateLeafFFI("paimon_predicate_greater_than")
var ffiPredicateGreaterOrEqual = newPredicateLeafFFI("paimon_predicate_greater_or_equal")
// newPredicateLeafFFI creates an FFI wrapper for comparison predicate functions
// with signature: (table, column, datum) -> result_predicate.
func newPredicateLeafFFI(sym string) *FFI[func(*paimonTable, *byte, paimonDatumC) (*paimonPredicate, error)] {
return newFFI(ffiOpts{
sym: contextKey(sym),
rType: &typeResultPredicate,
aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer, &typePaimonDatum},
}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable, *byte, paimonDatumC) (*paimonPredicate, error) {
return func(table *paimonTable, column *byte, datum paimonDatumC) (*paimonPredicate, error) {
var result resultPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&table),
unsafe.Pointer(&column),
unsafe.Pointer(&datum),
)
if result.error != nil {
return nil, parseError(ctx, result.error)
}
return result.predicate, nil
}
})
}
var ffiPredicateIsNull = newPredicateNullFFI("paimon_predicate_is_null")
var ffiPredicateIsNotNull = newPredicateNullFFI("paimon_predicate_is_not_null")
// newPredicateNullFFI creates an FFI wrapper for null-check predicate functions
// with signature: (table, column) -> result_predicate.
func newPredicateNullFFI(sym string) *FFI[func(*paimonTable, *byte) (*paimonPredicate, error)] {
return newFFI(ffiOpts{
sym: contextKey(sym),
rType: &typeResultPredicate,
aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer},
}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable, *byte) (*paimonPredicate, error) {
return func(table *paimonTable, column *byte) (*paimonPredicate, error) {
var result resultPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&table),
unsafe.Pointer(&column),
)
if result.error != nil {
return nil, parseError(ctx, result.error)
}
return result.predicate, nil
}
})
}
var ffiPredicateIsIn = newPredicateInFFI("paimon_predicate_is_in")
var ffiPredicateIsNotIn = newPredicateInFFI("paimon_predicate_is_not_in")
// newPredicateInFFI creates an FFI wrapper for IN/NOT IN predicate functions
// with signature: (table, column, datums, datums_len) -> result_predicate.
func newPredicateInFFI(sym string) *FFI[func(*paimonTable, *byte, unsafe.Pointer, uintptr) (*paimonPredicate, error)] {
return newFFI(ffiOpts{
sym: contextKey(sym),
rType: &typeResultPredicate,
aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer, &ffi.TypePointer, &ffi.TypePointer},
}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable, *byte, unsafe.Pointer, uintptr) (*paimonPredicate, error) {
return func(table *paimonTable, column *byte, datums unsafe.Pointer, datumsLen uintptr) (*paimonPredicate, error) {
var result resultPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&table),
unsafe.Pointer(&column),
unsafe.Pointer(&datums),
unsafe.Pointer(&datumsLen),
)
if result.error != nil {
return nil, parseError(ctx, result.error)
}
return result.predicate, nil
}
})
}
var ffiPredicateAnd = newFFI(ffiOpts{
sym: "paimon_predicate_and",
rType: &ffi.TypePointer,
aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer},
}, func(_ context.Context, ffiCall ffiCall) func(*paimonPredicate, *paimonPredicate) *paimonPredicate {
return func(a, b *paimonPredicate) *paimonPredicate {
var result *paimonPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&a),
unsafe.Pointer(&b),
)
return result
}
})
var ffiPredicateOr = newFFI(ffiOpts{
sym: "paimon_predicate_or",
rType: &ffi.TypePointer,
aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer},
}, func(_ context.Context, ffiCall ffiCall) func(*paimonPredicate, *paimonPredicate) *paimonPredicate {
return func(a, b *paimonPredicate) *paimonPredicate {
var result *paimonPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&a),
unsafe.Pointer(&b),
)
return result
}
})
var ffiPredicateNot = newFFI(ffiOpts{
sym: "paimon_predicate_not",
rType: &ffi.TypePointer,
aTypes: []*ffi.Type{&ffi.TypePointer},
}, func(_ context.Context, ffiCall ffiCall) func(*paimonPredicate) *paimonPredicate {
return func(p *paimonPredicate) *paimonPredicate {
var result *paimonPredicate
ffiCall(
unsafe.Pointer(&result),
unsafe.Pointer(&p),
)
return result
}
})