You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@tsfile.apache.org by "ColinLeeo (via GitHub)" <gi...@apache.org> on 2026/09/09 02:22:53 UTC
[PR] Feat/go table model api [tsfile]
ColinLeeo opened a new pull request, #949:
URL: https://github.com/apache/tsfile/pull/949
## Summary
Add a table-model-only Go API backed by the public C ABI.
The Go writer now binds and retains a single `TableSchema` during construction. Callers can write targetless Tablets or Arrow Record/Table batches, then query table data through one configurable `Reader.Query` entry point.
## Changes
- Add `NewWriter(path, schema, options...)` with schema ownership and validation.
- Change `NewTablet` to accept only columns and row capacity.
- Add Tablet and Arrow batch writes for all supported table data types.
- Add unified query options for time range, Tag filters, pagination, and Arrow batch size.
- Add typed row access with one-based ResultSet column indexes.
- Add Arrow RecordBatch/Table result consumption with `io.EOF` handling.
- Add table schema lookup and enumeration.
- Preserve BLOB length, embedded NUL bytes, empty values, and NULL values.
- Normalize table and column identifiers consistently with the native table model.
- Truncate existing files when creating a writer.
- Remove tree-model symbols and examples from the public Go API.
- Strengthen C ABI argument validation and resource ownership handling.
## Validation
- `go test ./...`
- `go test -race ./tsfile`
- `go vet ./...`
- C++ test suite: 934 passed, 3 skipped because external compatibility fixtures were not configured
- `./mvnw -P with-cpp -DskipTests -Dxml-format.skip=true package`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3974990515
##########
go/tsfile/cgo_bridge.go:
##########
@@ -412,13 +421,7 @@ func newTabletHandle(target string, columnNames []string, dataTypes []DataType,
for i, t := range dataTypes {
types[i] = C.TSDataType(t)
}
- h := C.tablet_new_with_target_name(
- targetCs,
- &names[0],
- &types[0],
- C.int(len(dataTypes)),
- C.int(maxRows),
- )
+ h := C.tablet_new(&names[0], &types[0], C.uint32_t(len(dataTypes)), C.uint32_t(maxRows))
if h == nil {
return nil, newError("new tablet", C.RET_OOM)
}
Review Comment:
The C ABI `tablet_new` now returns `nullptr` for invalid arguments/schema (e.g., duplicate names, unsupported types, oversize max rows), not only OOM. Mapping `nil` unconditionally to `RET_OOM` can misreport errors and mislead callers. Consider adding equivalent Go-side validation for all `tablet_new` rejection conditions so `nil` can be treated as OOM reliably, or extend the C ABI to return an `ERRNO` out-param for `tablet_new` so Go can surface the correct error.
##########
go/tsfile/result_set.go:
##########
@@ -111,55 +138,104 @@ func (rs *ResultSet) IsNull(column int) (value bool, err error) {
return
}
-func (rs *ResultSet) get(column int, allowed []DataType, fn func()) error {
+func (rs *ResultSet) get(column int, allowed []DataType, fn func() error) error {
return rs.locked(func() error {
+ if err := rs.requireMode(resultModeRows); err != nil {
+ return err
+ }
if err := rs.validateColumn(column, allowed...); err != nil {
return err
}
if rs.handle.isNull(column) {
return ErrNullValue
}
- fn()
- return nil
+ return fn()
})
}
// Bool returns a BOOLEAN column from the current row.
func (rs *ResultSet) Bool(column int) (value bool, err error) {
- err = rs.get(column, []DataType{DataTypeBoolean}, func() { value = rs.handle.bool(column) })
+ err = rs.get(column, []DataType{DataTypeBoolean}, func() error { value = rs.handle.bool(column); return nil })
return
}
// Int32 returns an INT32 or DATE column from the current row.
func (rs *ResultSet) Int32(column int) (value int32, err error) {
- err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() { value = rs.handle.int32(column) })
+ err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() error { value = rs.handle.int32(column); return nil })
return
}
// Int64 returns an INT64 or TIMESTAMP column from the current row.
func (rs *ResultSet) Int64(column int) (value int64, err error) {
- err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() { value = rs.handle.int64(column) })
+ err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() error { value = rs.handle.int64(column); return nil })
return
}
// Float32 returns a FLOAT column from the current row.
func (rs *ResultSet) Float32(column int) (value float32, err error) {
- err = rs.get(column, []DataType{DataTypeFloat}, func() { value = rs.handle.float32(column) })
+ err = rs.get(column, []DataType{DataTypeFloat}, func() error { value = rs.handle.float32(column); return nil })
return
}
// Float64 returns a DOUBLE column from the current row.
func (rs *ResultSet) Float64(column int) (value float64, err error) {
- err = rs.get(column, []DataType{DataTypeDouble}, func() { value = rs.handle.float64(column) })
+ err = rs.get(column, []DataType{DataTypeDouble}, func() error { value = rs.handle.float64(column); return nil })
return
}
// String returns a TEXT or STRING column from the current row.
func (rs *ResultSet) String(column int) (value string, err error) {
- err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() { value = rs.handle.string(column) })
+ err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() error { value = rs.handle.string(column); return nil })
return
}
+// Bytes returns a copy of a BLOB column from the current row.
+func (rs *ResultSet) Bytes(column int) (value []byte, err error) {
+ err = rs.get(column, []DataType{DataTypeBlob}, func() error {
+ value, err = rs.handle.bytes(column)
+ return err
+ })
+ return
+}
+
+// ReadArrowRecordBatch returns the next query batch. The caller owns the
+// returned record and must call Release. It returns io.EOF after the last
+// batch.
+func (rs *ResultSet) ReadArrowRecordBatch() (arrow.Record, error) {
+ var nativeArray cdata.CArrowArray
+ var nativeSchema cdata.CArrowSchema
+ defer cdata.ReleaseCArrowArray(&nativeArray)
+ defer cdata.ReleaseCArrowSchema(&nativeSchema)
Review Comment:
Be careful with Arrow C Data Interface ownership. `ImportCRecordBatch` commonly takes ownership of the passed `CArrowArray`/`CArrowSchema` (and releases them when the returned `arrow.Record` is released). Unconditionally deferring `ReleaseCArrowArray/Schema` risks double-free if the import function does not clear the release callbacks/structs. Consider only releasing the C structs on error, or explicitly clearing/zeroing them after a successful import if Arrow Go does not already do so.
##########
go/tsfile/result_set.go:
##########
@@ -111,55 +138,104 @@ func (rs *ResultSet) IsNull(column int) (value bool, err error) {
return
}
-func (rs *ResultSet) get(column int, allowed []DataType, fn func()) error {
+func (rs *ResultSet) get(column int, allowed []DataType, fn func() error) error {
return rs.locked(func() error {
+ if err := rs.requireMode(resultModeRows); err != nil {
+ return err
+ }
if err := rs.validateColumn(column, allowed...); err != nil {
return err
}
if rs.handle.isNull(column) {
return ErrNullValue
}
- fn()
- return nil
+ return fn()
})
}
// Bool returns a BOOLEAN column from the current row.
func (rs *ResultSet) Bool(column int) (value bool, err error) {
- err = rs.get(column, []DataType{DataTypeBoolean}, func() { value = rs.handle.bool(column) })
+ err = rs.get(column, []DataType{DataTypeBoolean}, func() error { value = rs.handle.bool(column); return nil })
return
}
// Int32 returns an INT32 or DATE column from the current row.
func (rs *ResultSet) Int32(column int) (value int32, err error) {
- err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() { value = rs.handle.int32(column) })
+ err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() error { value = rs.handle.int32(column); return nil })
return
}
// Int64 returns an INT64 or TIMESTAMP column from the current row.
func (rs *ResultSet) Int64(column int) (value int64, err error) {
- err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() { value = rs.handle.int64(column) })
+ err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() error { value = rs.handle.int64(column); return nil })
return
}
// Float32 returns a FLOAT column from the current row.
func (rs *ResultSet) Float32(column int) (value float32, err error) {
- err = rs.get(column, []DataType{DataTypeFloat}, func() { value = rs.handle.float32(column) })
+ err = rs.get(column, []DataType{DataTypeFloat}, func() error { value = rs.handle.float32(column); return nil })
return
}
// Float64 returns a DOUBLE column from the current row.
func (rs *ResultSet) Float64(column int) (value float64, err error) {
- err = rs.get(column, []DataType{DataTypeDouble}, func() { value = rs.handle.float64(column) })
+ err = rs.get(column, []DataType{DataTypeDouble}, func() error { value = rs.handle.float64(column); return nil })
return
}
// String returns a TEXT or STRING column from the current row.
func (rs *ResultSet) String(column int) (value string, err error) {
- err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() { value = rs.handle.string(column) })
+ err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() error { value = rs.handle.string(column); return nil })
return
}
+// Bytes returns a copy of a BLOB column from the current row.
+func (rs *ResultSet) Bytes(column int) (value []byte, err error) {
+ err = rs.get(column, []DataType{DataTypeBlob}, func() error {
+ value, err = rs.handle.bytes(column)
+ return err
+ })
+ return
+}
+
+// ReadArrowRecordBatch returns the next query batch. The caller owns the
+// returned record and must call Release. It returns io.EOF after the last
+// batch.
+func (rs *ResultSet) ReadArrowRecordBatch() (arrow.Record, error) {
+ var nativeArray cdata.CArrowArray
+ var nativeSchema cdata.CArrowSchema
+ defer cdata.ReleaseCArrowArray(&nativeArray)
+ defer cdata.ReleaseCArrowSchema(&nativeSchema)
+ err := rs.locked(func() error {
+ if err := rs.requireMode(resultModeBatches); err != nil {
+ return err
+ }
+ return rs.handle.nextArrow(unsafe.Pointer(&nativeArray), unsafe.Pointer(&nativeSchema))
+ })
+ if err != nil {
+ if nativeError, ok := err.(*Error); ok && nativeError.Code == 21 {
+ return nil, io.EOF
+ }
+ return nil, err
+ }
+ record, err := cdata.ImportCRecordBatch(&nativeArray, &nativeSchema)
Review Comment:
Be careful with Arrow C Data Interface ownership. `ImportCRecordBatch` commonly takes ownership of the passed `CArrowArray`/`CArrowSchema` (and releases them when the returned `arrow.Record` is released). Unconditionally deferring `ReleaseCArrowArray/Schema` risks double-free if the import function does not clear the release callbacks/structs. Consider only releasing the C structs on error, or explicitly clearing/zeroing them after a successful import if Arrow Go does not already do so.
##########
go/tsfile/result_set.go:
##########
@@ -111,55 +138,104 @@ func (rs *ResultSet) IsNull(column int) (value bool, err error) {
return
}
-func (rs *ResultSet) get(column int, allowed []DataType, fn func()) error {
+func (rs *ResultSet) get(column int, allowed []DataType, fn func() error) error {
return rs.locked(func() error {
+ if err := rs.requireMode(resultModeRows); err != nil {
+ return err
+ }
if err := rs.validateColumn(column, allowed...); err != nil {
return err
}
if rs.handle.isNull(column) {
return ErrNullValue
}
- fn()
- return nil
+ return fn()
})
}
// Bool returns a BOOLEAN column from the current row.
func (rs *ResultSet) Bool(column int) (value bool, err error) {
- err = rs.get(column, []DataType{DataTypeBoolean}, func() { value = rs.handle.bool(column) })
+ err = rs.get(column, []DataType{DataTypeBoolean}, func() error { value = rs.handle.bool(column); return nil })
return
}
// Int32 returns an INT32 or DATE column from the current row.
func (rs *ResultSet) Int32(column int) (value int32, err error) {
- err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() { value = rs.handle.int32(column) })
+ err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() error { value = rs.handle.int32(column); return nil })
return
}
// Int64 returns an INT64 or TIMESTAMP column from the current row.
func (rs *ResultSet) Int64(column int) (value int64, err error) {
- err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() { value = rs.handle.int64(column) })
+ err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() error { value = rs.handle.int64(column); return nil })
return
}
// Float32 returns a FLOAT column from the current row.
func (rs *ResultSet) Float32(column int) (value float32, err error) {
- err = rs.get(column, []DataType{DataTypeFloat}, func() { value = rs.handle.float32(column) })
+ err = rs.get(column, []DataType{DataTypeFloat}, func() error { value = rs.handle.float32(column); return nil })
return
}
// Float64 returns a DOUBLE column from the current row.
func (rs *ResultSet) Float64(column int) (value float64, err error) {
- err = rs.get(column, []DataType{DataTypeDouble}, func() { value = rs.handle.float64(column) })
+ err = rs.get(column, []DataType{DataTypeDouble}, func() error { value = rs.handle.float64(column); return nil })
return
}
// String returns a TEXT or STRING column from the current row.
func (rs *ResultSet) String(column int) (value string, err error) {
- err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() { value = rs.handle.string(column) })
+ err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() error { value = rs.handle.string(column); return nil })
return
}
+// Bytes returns a copy of a BLOB column from the current row.
+func (rs *ResultSet) Bytes(column int) (value []byte, err error) {
+ err = rs.get(column, []DataType{DataTypeBlob}, func() error {
+ value, err = rs.handle.bytes(column)
+ return err
+ })
+ return
+}
+
+// ReadArrowRecordBatch returns the next query batch. The caller owns the
+// returned record and must call Release. It returns io.EOF after the last
+// batch.
+func (rs *ResultSet) ReadArrowRecordBatch() (arrow.Record, error) {
+ var nativeArray cdata.CArrowArray
+ var nativeSchema cdata.CArrowSchema
+ defer cdata.ReleaseCArrowArray(&nativeArray)
+ defer cdata.ReleaseCArrowSchema(&nativeSchema)
+ err := rs.locked(func() error {
+ if err := rs.requireMode(resultModeBatches); err != nil {
+ return err
+ }
+ return rs.handle.nextArrow(unsafe.Pointer(&nativeArray), unsafe.Pointer(&nativeSchema))
+ })
+ if err != nil {
+ if nativeError, ok := err.(*Error); ok && nativeError.Code == 21 {
+ return nil, io.EOF
+ }
Review Comment:
The EOF mapping relies on the magic error code `21`. This is brittle and makes the Go layer depend on a numeric errno staying stable. Prefer introducing a named sentinel (e.g., `ErrNoMoreData = &Error{Code: 21}`) or an internal constant tied to the pinned C errno list, then use `errors.Is(err, ErrNoMoreData)` (or equivalent) to translate to `io.EOF`.
##########
go/tsfile/schema.go:
##########
@@ -79,32 +60,43 @@ func validDataType(value DataType) bool {
switch value {
case DataTypeBoolean, DataTypeInt32, DataTypeInt64, DataTypeFloat,
DataTypeDouble, DataTypeText, DataTypeTimestamp, DataTypeDate,
- DataTypeString:
+ DataTypeBlob, DataTypeString:
return true
default:
return false
}
}
-func validEncoding(value Encoding) bool {
- return value >= EncodingPlain && value <= EncodingCamel
-}
-
-func validCompression(value Compression) bool {
- return value >= CompressionUncompressed && value <= CompressionLZMA2
-}
-
-func validColumnCategory(value ColumnCategory) bool {
- return value >= ColumnCategoryTag && value <= ColumnCategoryTime
-}
-
-func validateTimeseriesSchema(op string, schema TimeseriesSchema) error {
- if err := validateCString(op, "timeseries name", schema.Name); err != nil {
- return err
+func validateAndCopyTableSchema(schema TableSchema) (TableSchema, error) {
+ if err := validateCString("validate table schema", "table name", schema.Table); err != nil {
+ return TableSchema{}, fmt.Errorf("%w: %v", ErrInvalidSchema, err)
}
- if !validDataType(schema.DataType) || !validEncoding(schema.Encoding) ||
- !validCompression(schema.Compression) {
- return fmt.Errorf("%w: invalid timeseries schema for %q", newError(op, 8), schema.Name)
+ if len(schema.Columns) == 0 {
+ return TableSchema{}, fmt.Errorf("%w: table requires at least one column", ErrInvalidSchema)
}
- return nil
+ copy := TableSchema{Table: normalizeIdentifier(schema.Table), Columns: append([]ColumnSchema(nil), schema.Columns...)}
+ seen := make(map[string]struct{}, len(copy.Columns))
+ for i, column := range copy.Columns {
+ if err := validateCString("validate table schema", "column name", column.Name); err != nil {
+ return TableSchema{}, fmt.Errorf("%w: column %d: %v", ErrInvalidSchema, i, err)
+ }
+ name := normalizeIdentifier(column.Name)
+ if _, ok := seen[name]; ok {
+ return TableSchema{}, fmt.Errorf("%w: duplicate column %q", ErrInvalidSchema, column.Name)
+ }
+ seen[name] = struct{}{}
+ if !validDataType(column.DataType) {
+ return TableSchema{}, fmt.Errorf("%w: unsupported data type %d for column %q", ErrInvalidSchema, column.DataType, column.Name)
+ }
+ if column.Category != ColumnCategoryTag && column.Category != ColumnCategoryField {
+ return TableSchema{}, fmt.Errorf("%w: unsupported category %d for column %q", ErrInvalidSchema, column.Category, column.Name)
+ }
+ if column.Category == ColumnCategoryTag && column.DataType != DataTypeString {
+ return TableSchema{}, fmt.Errorf("%w: TAG column %q must use STRING", ErrInvalidSchema, column.Name)
+ }
+ copy.Columns[i].Name = name
+ }
+ return copy, nil
}
+
+func normalizeIdentifier(value string) string { return strings.ToLower(value) }
Review Comment:
Identifier normalization uses `strings.ToLower`, which applies Unicode case-mapping. If the native side normalizes identifiers differently (commonly ASCII-only lowering, or byte-wise `tolower`), non-ASCII identifiers could normalize inconsistently between Go and C++ and lead to mismatches (e.g., schema lookup vs. query/write). To keep the Go API 'normalized consistently with the native table model', consider implementing the same normalization algorithm as native (often ASCII-only) and/or clearly documenting the supported identifier character set and case-folding behavior.
##########
go/go.mod:
##########
@@ -17,4 +17,21 @@
module github.com/apache/tsfile/go
-go 1.22
+go 1.22.0
Review Comment:
The `go` directive is typically specified as a major.minor version (e.g., `go 1.22`). Using `go 1.22.0` can reduce compatibility with tooling that expects the conventional format. If you want to pin the patch toolchain, prefer using `go 1.22` plus an explicit `toolchain go1.22.0` directive (if desired/appropriate for this repo).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "HTHou (via GitHub)" <gi...@apache.org>.
HTHou commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3964510224
##########
docs/superpowers/specs/2026-09-08-go-table-api-design.md:
##########
Review Comment:
superpowers?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3974953806
##########
cpp/src/cwrapper/tsfile_cwrapper.cc:
##########
@@ -302,14 +366,33 @@ ERRNO tsfile_reader_close(TsFileReader reader) {
Tablet tablet_new(char** column_name_list, TSDataType* data_types,
uint32_t column_num, uint32_t max_rows) {
- std::vector<std::string> measurement_list;
- std::vector<common::TSDataType> data_type_list;
- for (uint32_t i = 0; i < column_num; i++) {
- measurement_list.emplace_back(storage::to_lower(column_name_list[i]));
- data_type_list.push_back(
- static_cast<common::TSDataType>(*(data_types + i)));
+ if (column_num == 0 || max_rows == 0 || column_name_list == nullptr ||
+ data_types == nullptr || max_rows >= (1u << 30)) {
+ return nullptr;
+ }
+ try {
+ std::vector<std::string> measurement_list;
+ std::vector<common::TSDataType> data_type_list;
+ std::set<std::string> names;
+ for (uint32_t i = 0; i < column_num; i++) {
+ if (column_name_list[i] == nullptr ||
+ column_name_list[i][0] == '\0' ||
+ !is_supported_table_type(data_types[i])) {
+ return nullptr;
+ }
+ std::string name = storage::to_lower(column_name_list[i]);
+ if (!names.insert(name).second) {
+ return nullptr;
+ }
+ measurement_list.emplace_back(std::move(name));
+ data_type_list.push_back(
+ static_cast<common::TSDataType>(*(data_types + i)));
+ }
+ return new (std::nothrow)
+ storage::Tablet(measurement_list, data_type_list, max_rows);
+ } catch (...) {
+ return nullptr;
}
Review Comment:
Return ERRNO?
##########
go/tsfile/writer.go:
##########
@@ -79,78 +89,189 @@ func (w *Writer) withHandle(fn func(*writerHandle) error) error {
return fn(w.handle)
}
-// RegisterTable registers a table-model schema.
-func (w *Writer) RegisterTable(schema TableSchema) error {
- if err := validateCString("register table", "table name", schema.Table); err != nil {
- return err
+func (w *Writer) writeTablet(tablet *Tablet) error {
+ if tablet == nil {
+ return fmt.Errorf("%w: nil tablet", ErrInvalidArgument)
}
- if len(schema.Columns) == 0 {
- return fmt.Errorf("%w: table requires at least one column", ErrInvalidSchema)
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.handle == nil || w.handle.ptr == nil {
+ return ErrClosed
}
- for i, column := range schema.Columns {
- if err := validateCString("register table", "column name", column.Name); err != nil {
- return fmt.Errorf("column %d: %w", i, err)
+ tablet.mu.Lock()
+ defer tablet.mu.Unlock()
+ if tablet.handle == nil || tablet.handle.ptr == nil {
+ return ErrClosed
+ }
+ if len(tablet.columns) != len(w.schema.Columns) {
+ return fmt.Errorf("%w: Tablet has %d columns; writer schema has %d",
+ ErrInvalidSchema, len(tablet.columns), len(w.schema.Columns))
+ }
Review Comment:
Why would a tablet need a mutex?
##########
go/tsfile/cgo_bridge.go:
##########
@@ -476,122 +479,89 @@ func (h *tabletHandle) addString(row, column int, value string) error {
C.Tablet(h.ptr), C.uint32_t(row), C.uint32_t(column), cs, C.int(len(value)))))
}
+func (h *tabletHandle) addBytes(row, column int, value []byte) error {
+ p, n, err := copyCBytes("set tablet bytes", value)
+ if err != nil {
+ return err
+ }
+ defer freeCBytes(p)
+ return newError("set tablet bytes", cerrno(C.tablet_add_value_by_index_string_with_len(
+ C.Tablet(h.ptr), C.uint32_t(row), C.uint32_t(column), (*C.char)(p), C.int(n))))
+}
+
// ---------------------------------------------------------------------------
// Writer handle
// ---------------------------------------------------------------------------
-// writerHandle is the native half of the public Writer type. It uses the
-// public generic writer API; tsfile_generic_writer_close flushes, closes, and
-// deletes the native writer.
-type writerHandle struct{ nativeHandle }
+// writerHandle owns a table writer and the WriteFile passed to it.
+type writerHandle struct {
+ nativeHandle
+ file unsafe.Pointer
+}
// releaseWriter closes and deletes the native writer; nil is RET_OK.
func releaseWriter(ptr unsafe.Pointer) cerrno {
if ptr == nil {
return C.RET_OK
}
- return cerrno(C.tsfile_generic_writer_close(C.TsFileGenericWriter(ptr)))
+ return cerrno(C.tsfile_writer_close(C.TsFileWriter(ptr)))
+}
+
+func (h *writerHandle) close() error {
+ if err := h.nativeHandle.close(); err != nil {
+ return err
+ }
+ if h.file != nil {
+ file := C.WriteFile(h.file)
+ C.free_write_file(&file)
+ h.file = nil
+ }
+ return nil
}
// newWriterHandle opens (or creates) the TsFile at path for writing.
// memoryThresholdBytes bounds the in-memory write buffer.
-func newWriterHandle(path string, memoryThresholdBytes uint64) (*writerHandle, error) {
+func newWriterHandle(path string, schema TableSchema, memoryThresholdBytes uint64) (*writerHandle, error) {
if err := validateCString("open writer", "path", path); err != nil {
return nil, err
}
cs := cStringPtr(path)
defer freeCString(cs)
var code C.ERRNO
- h := C.tsfile_generic_writer_new(cs, C.uint64_t(memoryThresholdBytes), &code)
- if h == nil {
+ file := C.write_file_new(cs, &code)
+ if file == nil {
return nil, newError("open writer", cerrno(code))
}
- return &writerHandle{nativeHandle{ptr: unsafe.Pointer(h), release: releaseWriter}}, nil
-}
-
-func allocTimeseriesSchema(schema TimeseriesSchema) (*C.TimeseriesSchema, func(), error) {
- name := cStringPtr(schema.Name)
- if name == nil {
- return nil, func() {}, newError("marshal timeseries schema", C.RET_OOM)
- }
- p := (*C.TimeseriesSchema)(C.malloc(C.size_t(C.sizeof_TimeseriesSchema)))
- if p == nil {
- freeCString(name)
- return nil, func() {}, newError("marshal timeseries schema", C.RET_OOM)
- }
- p.timeseries_name = name
- p.data_type = C.TSDataType(schema.DataType)
- p.encoding = C.TSEncoding(schema.Encoding)
- p.compression = C.CompressionType(schema.Compression)
- return p, func() {
- freeCString(name)
- C.free(unsafe.Pointer(p))
- }, nil
-}
-
-func (h *writerHandle) registerTimeseries(device string, schema TimeseriesSchema) error {
- deviceName := cStringPtr(device)
- defer freeCString(deviceName)
- native, freeSchema, err := allocTimeseriesSchema(schema)
+ nativeSchema, freeSchema, err := allocTableSchema(schema)
if err != nil {
- return err
+ C.free_write_file(&file)
+ return nil, err
}
defer freeSchema()
- return newError("register timeseries", cerrno(C.tsfile_generic_writer_register_timeseries(
- C.TsFileGenericWriter(h.ptr), deviceName, native)))
-}
-
-func (h *writerHandle) registerDevice(schema DeviceSchema) error {
- deviceName := cStringPtr(schema.Device)
- defer freeCString(deviceName)
- names, err := marshalCStrings(timeseriesNames(schema.TimeSeries))
- if err != nil {
- return err
- }
- defer freeCStrings(names)
-
- var series *C.TimeseriesSchema
- if len(schema.TimeSeries) > 0 {
- mem := C.malloc(C.size_t(len(schema.TimeSeries)) * C.size_t(C.sizeof_TimeseriesSchema))
- if mem == nil {
- return newError("marshal device schema", C.RET_OOM)
- }
- defer C.free(mem)
- items := unsafe.Slice((*C.TimeseriesSchema)(mem), len(schema.TimeSeries))
- for i, item := range schema.TimeSeries {
- items[i].timeseries_name = names[i]
- items[i].data_type = C.TSDataType(item.DataType)
- items[i].encoding = C.TSEncoding(item.Encoding)
- items[i].compression = C.CompressionType(item.Compression)
- }
- series = (*C.TimeseriesSchema)(mem)
- }
- native := (*C.DeviceSchema)(C.malloc(C.size_t(C.sizeof_DeviceSchema)))
- if native == nil {
- return newError("marshal device schema", C.RET_OOM)
+ h := C.tsfile_writer_new_with_memory_threshold(file, nativeSchema, C.uint64_t(memoryThresholdBytes), &code)
+ if h == nil {
+ C.free_write_file(&file)
+ return nil, newError("open writer", cerrno(code))
}
- defer C.free(unsafe.Pointer(native))
- native.device_name = deviceName
- native.timeseries_schema = series
- native.timeseries_num = C.int(len(schema.TimeSeries))
- return newError("register device", cerrno(C.tsfile_generic_writer_register_device(
- C.TsFileGenericWriter(h.ptr), native)))
+ return &writerHandle{
+ nativeHandle: nativeHandle{ptr: unsafe.Pointer(h), release: releaseWriter},
+ file: unsafe.Pointer(file),
+ }, nil
}
-func (h *writerHandle) registerTable(schema TableSchema) error {
+func allocTableSchema(schema TableSchema) (*C.TableSchema, func(), error) {
Review Comment:
Add some comments to explain what the second return value is?
##########
go/tsfile/tablet.go:
##########
@@ -23,35 +23,68 @@ import (
"sync"
)
-// Tablet is a fixed-capacity batch of rows for one device or table.
+// Tablet is a fixed-capacity batch of table rows.
type Tablet struct {
mu sync.Mutex
handle *tabletHandle
columns []TabletColumn
maxRows int
+ timeSet []bool
Review Comment:
Why would we need this?
We do not allow any holes in the time column of a Tablet.
##########
go/tsfile/schema.go:
##########
@@ -79,32 +60,43 @@ func validDataType(value DataType) bool {
switch value {
case DataTypeBoolean, DataTypeInt32, DataTypeInt64, DataTypeFloat,
DataTypeDouble, DataTypeText, DataTypeTimestamp, DataTypeDate,
- DataTypeString:
+ DataTypeBlob, DataTypeString:
return true
default:
return false
}
}
-func validEncoding(value Encoding) bool {
- return value >= EncodingPlain && value <= EncodingCamel
-}
-
-func validCompression(value Compression) bool {
- return value >= CompressionUncompressed && value <= CompressionLZMA2
-}
-
-func validColumnCategory(value ColumnCategory) bool {
- return value >= ColumnCategoryTag && value <= ColumnCategoryTime
-}
-
-func validateTimeseriesSchema(op string, schema TimeseriesSchema) error {
- if err := validateCString(op, "timeseries name", schema.Name); err != nil {
- return err
+func validateAndCopyTableSchema(schema TableSchema) (TableSchema, error) {
Review Comment:
Where is the copy needed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "ColinLeeo (via GitHub)" <gi...@apache.org>.
ColinLeeo merged PR #949:
URL: https://github.com/apache/tsfile/pull/949
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "ColinLeeo (via GitHub)" <gi...@apache.org>.
ColinLeeo commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3977319097
##########
go/tsfile/writer.go:
##########
@@ -79,78 +89,189 @@ func (w *Writer) withHandle(fn func(*writerHandle) error) error {
return fn(w.handle)
}
-// RegisterTable registers a table-model schema.
-func (w *Writer) RegisterTable(schema TableSchema) error {
- if err := validateCString("register table", "table name", schema.Table); err != nil {
- return err
+func (w *Writer) writeTablet(tablet *Tablet) error {
+ if tablet == nil {
+ return fmt.Errorf("%w: nil tablet", ErrInvalidArgument)
}
- if len(schema.Columns) == 0 {
- return fmt.Errorf("%w: table requires at least one column", ErrInvalidSchema)
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.handle == nil || w.handle.ptr == nil {
+ return ErrClosed
}
- for i, column := range schema.Columns {
- if err := validateCString("register table", "column name", column.Name); err != nil {
- return fmt.Errorf("column %d: %w", i, err)
+ tablet.mu.Lock()
+ defer tablet.mu.Unlock()
+ if tablet.handle == nil || tablet.handle.ptr == nil {
+ return ErrClosed
+ }
+ if len(tablet.columns) != len(w.schema.Columns) {
+ return fmt.Errorf("%w: Tablet has %d columns; writer schema has %d",
+ ErrInvalidSchema, len(tablet.columns), len(w.schema.Columns))
+ }
Review Comment:
Removed. Tablet lifecycle is caller-managed, so `Close` must not overlap with active writes; no Go-side Tablet mutex is added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "ColinLeeo (via GitHub)" <gi...@apache.org>.
ColinLeeo commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3977317520
##########
cpp/src/cwrapper/tsfile_cwrapper.cc:
##########
@@ -302,14 +366,33 @@ ERRNO tsfile_reader_close(TsFileReader reader) {
Tablet tablet_new(char** column_name_list, TSDataType* data_types,
uint32_t column_num, uint32_t max_rows) {
- std::vector<std::string> measurement_list;
- std::vector<common::TSDataType> data_type_list;
- for (uint32_t i = 0; i < column_num; i++) {
- measurement_list.emplace_back(storage::to_lower(column_name_list[i]));
- data_type_list.push_back(
- static_cast<common::TSDataType>(*(data_types + i)));
+ if (column_num == 0 || max_rows == 0 || column_name_list == nullptr ||
+ data_types == nullptr || max_rows >= (1u << 30)) {
+ return nullptr;
+ }
+ try {
+ std::vector<std::string> measurement_list;
+ std::vector<common::TSDataType> data_type_list;
+ std::set<std::string> names;
+ for (uint32_t i = 0; i < column_num; i++) {
+ if (column_name_list[i] == nullptr ||
+ column_name_list[i][0] == '\0' ||
+ !is_supported_table_type(data_types[i])) {
+ return nullptr;
+ }
+ std::string name = storage::to_lower(column_name_list[i]);
+ if (!names.insert(name).second) {
+ return nullptr;
+ }
+ measurement_list.emplace_back(std::move(name));
+ data_type_list.push_back(
+ static_cast<common::TSDataType>(*(data_types + i)));
+ }
+ return new (std::nothrow)
+ storage::Tablet(measurement_list, data_type_list, max_rows);
+ } catch (...) {
+ return nullptr;
}
Review Comment:
Fixed in `eee4c5c93`. `tablet_new` now reports the exact failure code through an `ERRNO*` out parameter, and Go preserves it.
##########
go/tsfile/cgo_bridge.go:
##########
@@ -476,122 +479,89 @@ func (h *tabletHandle) addString(row, column int, value string) error {
C.Tablet(h.ptr), C.uint32_t(row), C.uint32_t(column), cs, C.int(len(value)))))
}
+func (h *tabletHandle) addBytes(row, column int, value []byte) error {
+ p, n, err := copyCBytes("set tablet bytes", value)
+ if err != nil {
+ return err
+ }
+ defer freeCBytes(p)
+ return newError("set tablet bytes", cerrno(C.tablet_add_value_by_index_string_with_len(
+ C.Tablet(h.ptr), C.uint32_t(row), C.uint32_t(column), (*C.char)(p), C.int(n))))
+}
+
// ---------------------------------------------------------------------------
// Writer handle
// ---------------------------------------------------------------------------
-// writerHandle is the native half of the public Writer type. It uses the
-// public generic writer API; tsfile_generic_writer_close flushes, closes, and
-// deletes the native writer.
-type writerHandle struct{ nativeHandle }
+// writerHandle owns a table writer and the WriteFile passed to it.
+type writerHandle struct {
+ nativeHandle
+ file unsafe.Pointer
+}
// releaseWriter closes and deletes the native writer; nil is RET_OK.
func releaseWriter(ptr unsafe.Pointer) cerrno {
if ptr == nil {
return C.RET_OK
}
- return cerrno(C.tsfile_generic_writer_close(C.TsFileGenericWriter(ptr)))
+ return cerrno(C.tsfile_writer_close(C.TsFileWriter(ptr)))
+}
+
+func (h *writerHandle) close() error {
+ if err := h.nativeHandle.close(); err != nil {
+ return err
+ }
+ if h.file != nil {
+ file := C.WriteFile(h.file)
+ C.free_write_file(&file)
+ h.file = nil
+ }
+ return nil
}
// newWriterHandle opens (or creates) the TsFile at path for writing.
// memoryThresholdBytes bounds the in-memory write buffer.
-func newWriterHandle(path string, memoryThresholdBytes uint64) (*writerHandle, error) {
+func newWriterHandle(path string, schema TableSchema, memoryThresholdBytes uint64) (*writerHandle, error) {
if err := validateCString("open writer", "path", path); err != nil {
return nil, err
}
cs := cStringPtr(path)
defer freeCString(cs)
var code C.ERRNO
- h := C.tsfile_generic_writer_new(cs, C.uint64_t(memoryThresholdBytes), &code)
- if h == nil {
+ file := C.write_file_new(cs, &code)
+ if file == nil {
return nil, newError("open writer", cerrno(code))
}
- return &writerHandle{nativeHandle{ptr: unsafe.Pointer(h), release: releaseWriter}}, nil
-}
-
-func allocTimeseriesSchema(schema TimeseriesSchema) (*C.TimeseriesSchema, func(), error) {
- name := cStringPtr(schema.Name)
- if name == nil {
- return nil, func() {}, newError("marshal timeseries schema", C.RET_OOM)
- }
- p := (*C.TimeseriesSchema)(C.malloc(C.size_t(C.sizeof_TimeseriesSchema)))
- if p == nil {
- freeCString(name)
- return nil, func() {}, newError("marshal timeseries schema", C.RET_OOM)
- }
- p.timeseries_name = name
- p.data_type = C.TSDataType(schema.DataType)
- p.encoding = C.TSEncoding(schema.Encoding)
- p.compression = C.CompressionType(schema.Compression)
- return p, func() {
- freeCString(name)
- C.free(unsafe.Pointer(p))
- }, nil
-}
-
-func (h *writerHandle) registerTimeseries(device string, schema TimeseriesSchema) error {
- deviceName := cStringPtr(device)
- defer freeCString(deviceName)
- native, freeSchema, err := allocTimeseriesSchema(schema)
+ nativeSchema, freeSchema, err := allocTableSchema(schema)
if err != nil {
- return err
+ C.free_write_file(&file)
+ return nil, err
}
defer freeSchema()
- return newError("register timeseries", cerrno(C.tsfile_generic_writer_register_timeseries(
- C.TsFileGenericWriter(h.ptr), deviceName, native)))
-}
-
-func (h *writerHandle) registerDevice(schema DeviceSchema) error {
- deviceName := cStringPtr(schema.Device)
- defer freeCString(deviceName)
- names, err := marshalCStrings(timeseriesNames(schema.TimeSeries))
- if err != nil {
- return err
- }
- defer freeCStrings(names)
-
- var series *C.TimeseriesSchema
- if len(schema.TimeSeries) > 0 {
- mem := C.malloc(C.size_t(len(schema.TimeSeries)) * C.size_t(C.sizeof_TimeseriesSchema))
- if mem == nil {
- return newError("marshal device schema", C.RET_OOM)
- }
- defer C.free(mem)
- items := unsafe.Slice((*C.TimeseriesSchema)(mem), len(schema.TimeSeries))
- for i, item := range schema.TimeSeries {
- items[i].timeseries_name = names[i]
- items[i].data_type = C.TSDataType(item.DataType)
- items[i].encoding = C.TSEncoding(item.Encoding)
- items[i].compression = C.CompressionType(item.Compression)
- }
- series = (*C.TimeseriesSchema)(mem)
- }
- native := (*C.DeviceSchema)(C.malloc(C.size_t(C.sizeof_DeviceSchema)))
- if native == nil {
- return newError("marshal device schema", C.RET_OOM)
+ h := C.tsfile_writer_new_with_memory_threshold(file, nativeSchema, C.uint64_t(memoryThresholdBytes), &code)
+ if h == nil {
+ C.free_write_file(&file)
+ return nil, newError("open writer", cerrno(code))
}
- defer C.free(unsafe.Pointer(native))
- native.device_name = deviceName
- native.timeseries_schema = series
- native.timeseries_num = C.int(len(schema.TimeSeries))
- return newError("register device", cerrno(C.tsfile_generic_writer_register_device(
- C.TsFileGenericWriter(h.ptr), native)))
+ return &writerHandle{
+ nativeHandle: nativeHandle{ptr: unsafe.Pointer(h), release: releaseWriter},
+ file: unsafe.Pointer(file),
+ }, nil
}
-func (h *writerHandle) registerTable(schema TableSchema) error {
+func allocTableSchema(schema TableSchema) (*C.TableSchema, func(), error) {
Review Comment:
Added. The second value is the cleanup callback for the C-owned `TableSchema` and its nested allocations.
##########
go/tsfile/schema.go:
##########
@@ -79,32 +60,43 @@ func validDataType(value DataType) bool {
switch value {
case DataTypeBoolean, DataTypeInt32, DataTypeInt64, DataTypeFloat,
DataTypeDouble, DataTypeText, DataTypeTimestamp, DataTypeDate,
- DataTypeString:
+ DataTypeBlob, DataTypeString:
return true
default:
return false
}
}
-func validEncoding(value Encoding) bool {
- return value >= EncodingPlain && value <= EncodingCamel
-}
-
-func validCompression(value Compression) bool {
- return value >= CompressionUncompressed && value <= CompressionLZMA2
-}
-
-func validColumnCategory(value ColumnCategory) bool {
- return value >= ColumnCategoryTag && value <= ColumnCategoryTime
-}
-
-func validateTimeseriesSchema(op string, schema TimeseriesSchema) error {
- if err := validateCString(op, "timeseries name", schema.Name); err != nil {
- return err
+func validateAndCopyTableSchema(schema TableSchema) (TableSchema, error) {
Review Comment:
The writer retains the normalized schema for later Tablet validation, so it needs an owned copy. Added a comment to clarify this.
##########
go/tsfile/tablet.go:
##########
@@ -23,35 +23,68 @@ import (
"sync"
)
-// Tablet is a fixed-capacity batch of rows for one device or table.
+// Tablet is a fixed-capacity batch of table rows.
type Tablet struct {
mu sync.Mutex
handle *tabletHandle
columns []TabletColumn
maxRows int
+ timeSet []bool
Review Comment:
Removed the Go-only timestamp presence tracking and hole validation. Go now follows the native Tablet row semantics.
##########
go/tsfile/writer.go:
##########
@@ -79,78 +89,189 @@ func (w *Writer) withHandle(fn func(*writerHandle) error) error {
return fn(w.handle)
}
-// RegisterTable registers a table-model schema.
-func (w *Writer) RegisterTable(schema TableSchema) error {
- if err := validateCString("register table", "table name", schema.Table); err != nil {
- return err
+func (w *Writer) writeTablet(tablet *Tablet) error {
+ if tablet == nil {
+ return fmt.Errorf("%w: nil tablet", ErrInvalidArgument)
}
- if len(schema.Columns) == 0 {
- return fmt.Errorf("%w: table requires at least one column", ErrInvalidSchema)
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.handle == nil || w.handle.ptr == nil {
+ return ErrClosed
}
- for i, column := range schema.Columns {
- if err := validateCString("register table", "column name", column.Name); err != nil {
- return fmt.Errorf("column %d: %w", i, err)
+ tablet.mu.Lock()
+ defer tablet.mu.Unlock()
+ if tablet.handle == nil || tablet.handle.ptr == nil {
+ return ErrClosed
+ }
+ if len(tablet.columns) != len(w.schema.Columns) {
+ return fmt.Errorf("%w: Tablet has %d columns; writer schema has %d",
+ ErrInvalidSchema, len(tablet.columns), len(w.schema.Columns))
+ }
Review Comment:
Removed the Tablet mutex. Tablet is documented as not concurrency-safe; only Writer serializes access to its own handle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/go table model api [tsfile]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #949:
URL: https://github.com/apache/tsfile/pull/949#discussion_r3975161345
##########
go/go.mod:
##########
@@ -17,4 +17,21 @@
module github.com/apache/tsfile/go
-go 1.22
+go 1.22.0
Review Comment:
The `go` directive is typically `major.minor` (e.g., `go 1.22`). If you intend to pin a specific toolchain patch version, prefer using a `toolchain go1.22.0` directive (supported by newer Go versions) while keeping `go 1.22` for module semantics.
##########
go/tsfile/result_set.go:
##########
@@ -111,55 +138,104 @@ func (rs *ResultSet) IsNull(column int) (value bool, err error) {
return
}
-func (rs *ResultSet) get(column int, allowed []DataType, fn func()) error {
+func (rs *ResultSet) get(column int, allowed []DataType, fn func() error) error {
return rs.locked(func() error {
+ if err := rs.requireMode(resultModeRows); err != nil {
+ return err
+ }
if err := rs.validateColumn(column, allowed...); err != nil {
return err
}
if rs.handle.isNull(column) {
return ErrNullValue
}
- fn()
- return nil
+ return fn()
})
}
// Bool returns a BOOLEAN column from the current row.
func (rs *ResultSet) Bool(column int) (value bool, err error) {
- err = rs.get(column, []DataType{DataTypeBoolean}, func() { value = rs.handle.bool(column) })
+ err = rs.get(column, []DataType{DataTypeBoolean}, func() error { value = rs.handle.bool(column); return nil })
return
}
// Int32 returns an INT32 or DATE column from the current row.
func (rs *ResultSet) Int32(column int) (value int32, err error) {
- err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() { value = rs.handle.int32(column) })
+ err = rs.get(column, []DataType{DataTypeInt32, DataTypeDate}, func() error { value = rs.handle.int32(column); return nil })
return
}
// Int64 returns an INT64 or TIMESTAMP column from the current row.
func (rs *ResultSet) Int64(column int) (value int64, err error) {
- err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() { value = rs.handle.int64(column) })
+ err = rs.get(column, []DataType{DataTypeInt64, DataTypeTimestamp}, func() error { value = rs.handle.int64(column); return nil })
return
}
// Float32 returns a FLOAT column from the current row.
func (rs *ResultSet) Float32(column int) (value float32, err error) {
- err = rs.get(column, []DataType{DataTypeFloat}, func() { value = rs.handle.float32(column) })
+ err = rs.get(column, []DataType{DataTypeFloat}, func() error { value = rs.handle.float32(column); return nil })
return
}
// Float64 returns a DOUBLE column from the current row.
func (rs *ResultSet) Float64(column int) (value float64, err error) {
- err = rs.get(column, []DataType{DataTypeDouble}, func() { value = rs.handle.float64(column) })
+ err = rs.get(column, []DataType{DataTypeDouble}, func() error { value = rs.handle.float64(column); return nil })
return
}
// String returns a TEXT or STRING column from the current row.
func (rs *ResultSet) String(column int) (value string, err error) {
- err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() { value = rs.handle.string(column) })
+ err = rs.get(column, []DataType{DataTypeText, DataTypeString}, func() error { value = rs.handle.string(column); return nil })
return
}
+// Bytes returns a copy of a BLOB column from the current row.
+func (rs *ResultSet) Bytes(column int) (value []byte, err error) {
+ err = rs.get(column, []DataType{DataTypeBlob}, func() error {
+ value, err = rs.handle.bytes(column)
+ return err
+ })
+ return
+}
+
+// ReadArrowRecordBatch returns the next query batch. The caller owns the
+// returned record and must call Release. It returns io.EOF after the last
+// batch.
+func (rs *ResultSet) ReadArrowRecordBatch() (arrow.Record, error) {
+ var nativeArray cdata.CArrowArray
+ var nativeSchema cdata.CArrowSchema
+ defer cdata.ReleaseCArrowArray(&nativeArray)
+ defer cdata.ReleaseCArrowSchema(&nativeSchema)
+ err := rs.locked(func() error {
+ if err := rs.requireMode(resultModeBatches); err != nil {
+ return err
+ }
+ return rs.handle.nextArrow(unsafe.Pointer(&nativeArray), unsafe.Pointer(&nativeSchema))
+ })
+ if err != nil {
+ if nativeError, ok := err.(*Error); ok && nativeError.Code == 21 {
+ return nil, io.EOF
+ }
Review Comment:
This hard-codes native error code `21` to map to `io.EOF`, which is brittle and obscures meaning. Introduce a named sentinel (e.g., `ErrNoMoreData = &Error{Code: 21}`) and use `errors.Is(err, ErrNoMoreData)` here, or centralize the mapping in one place so future errno changes don't require hunting magic numbers.
##########
go/tsfile/schema.go:
##########
@@ -79,32 +60,43 @@ func validDataType(value DataType) bool {
switch value {
case DataTypeBoolean, DataTypeInt32, DataTypeInt64, DataTypeFloat,
DataTypeDouble, DataTypeText, DataTypeTimestamp, DataTypeDate,
- DataTypeString:
+ DataTypeBlob, DataTypeString:
return true
default:
return false
}
}
-func validEncoding(value Encoding) bool {
- return value >= EncodingPlain && value <= EncodingCamel
-}
-
-func validCompression(value Compression) bool {
- return value >= CompressionUncompressed && value <= CompressionLZMA2
-}
-
-func validColumnCategory(value ColumnCategory) bool {
- return value >= ColumnCategoryTag && value <= ColumnCategoryTime
-}
-
-func validateTimeseriesSchema(op string, schema TimeseriesSchema) error {
- if err := validateCString(op, "timeseries name", schema.Name); err != nil {
- return err
+func validateAndCopyTableSchema(schema TableSchema) (TableSchema, error) {
+ if err := validateCString("validate table schema", "table name", schema.Table); err != nil {
+ return TableSchema{}, fmt.Errorf("%w: %v", ErrInvalidSchema, err)
}
Review Comment:
This wraps the underlying error using `%v`, which prevents callers from detecting the root sentinel via `errors.Is` (e.g., `ErrInvalidArgument` from `validateCString`). If you want to preserve the cause chain, wrap with `%w` (e.g., `fmt.Errorf(\"%w: %w\", ErrInvalidSchema, err)`) or choose a single sentinel consistently for this failure mode.
##########
go/tsfile/schema.go:
##########
@@ -79,32 +60,43 @@ func validDataType(value DataType) bool {
switch value {
case DataTypeBoolean, DataTypeInt32, DataTypeInt64, DataTypeFloat,
DataTypeDouble, DataTypeText, DataTypeTimestamp, DataTypeDate,
- DataTypeString:
+ DataTypeBlob, DataTypeString:
return true
default:
return false
}
}
-func validEncoding(value Encoding) bool {
- return value >= EncodingPlain && value <= EncodingCamel
-}
-
-func validCompression(value Compression) bool {
- return value >= CompressionUncompressed && value <= CompressionLZMA2
-}
-
-func validColumnCategory(value ColumnCategory) bool {
- return value >= ColumnCategoryTag && value <= ColumnCategoryTime
-}
-
-func validateTimeseriesSchema(op string, schema TimeseriesSchema) error {
- if err := validateCString(op, "timeseries name", schema.Name); err != nil {
- return err
+func validateAndCopyTableSchema(schema TableSchema) (TableSchema, error) {
+ if err := validateCString("validate table schema", "table name", schema.Table); err != nil {
+ return TableSchema{}, fmt.Errorf("%w: %v", ErrInvalidSchema, err)
}
- if !validDataType(schema.DataType) || !validEncoding(schema.Encoding) ||
- !validCompression(schema.Compression) {
- return fmt.Errorf("%w: invalid timeseries schema for %q", newError(op, 8), schema.Name)
+ if len(schema.Columns) == 0 {
+ return TableSchema{}, fmt.Errorf("%w: table requires at least one column", ErrInvalidSchema)
}
- return nil
+ copy := TableSchema{Table: normalizeIdentifier(schema.Table), Columns: append([]ColumnSchema(nil), schema.Columns...)}
+ seen := make(map[string]struct{}, len(copy.Columns))
+ for i, column := range copy.Columns {
+ if err := validateCString("validate table schema", "column name", column.Name); err != nil {
+ return TableSchema{}, fmt.Errorf("%w: column %d: %v", ErrInvalidSchema, i, err)
+ }
+ name := normalizeIdentifier(column.Name)
+ if _, ok := seen[name]; ok {
+ return TableSchema{}, fmt.Errorf("%w: duplicate column %q", ErrInvalidSchema, column.Name)
+ }
+ seen[name] = struct{}{}
+ if !validDataType(column.DataType) {
+ return TableSchema{}, fmt.Errorf("%w: unsupported data type %d for column %q", ErrInvalidSchema, column.DataType, column.Name)
+ }
+ if column.Category != ColumnCategoryTag && column.Category != ColumnCategoryField {
+ return TableSchema{}, fmt.Errorf("%w: unsupported category %d for column %q", ErrInvalidSchema, column.Category, column.Name)
+ }
+ if column.Category == ColumnCategoryTag && column.DataType != DataTypeString {
+ return TableSchema{}, fmt.Errorf("%w: TAG column %q must use STRING", ErrInvalidSchema, column.Name)
+ }
+ copy.Columns[i].Name = name
+ }
+ return copy, nil
}
+
+func normalizeIdentifier(value string) string { return strings.ToLower(value) }
Review Comment:
Identifier normalization must match the native implementation to avoid mismatches between Go-side schema/tablet/query normalization and C++ lookup rules. `strings.ToLower` uses Unicode case mapping; if the native `to_lower` is ASCII-only (common in C++), non-ASCII identifiers could normalize differently across the boundary. Consider restricting identifiers to ASCII at validation time, or implementing the same ASCII-only lowering logic used by the native layer.
##########
go/tsfile/reader.go:
##########
@@ -87,79 +145,60 @@ func (r *Reader) query(fn func(*readerHandle) (*resultSetHandle, error)) (*Resul
if err != nil {
return nil, err
}
- result := &ResultSet{handle: handle, reader: r}
+ result := &ResultSet{handle: handle, reader: r, mode: mode}
result.metadata = handle.metadata()
r.results[result] = struct{}{}
return result, nil
}
-// QueryTable executes a time-range table-model query.
-func (r *Reader) QueryTable(query TableQuery) (*ResultSet, error) {
- if err := validateCString("query table", "table", query.Table); err != nil {
+// Query executes a table query. Omitting options scans the full time range in
+// row mode.
+func (r *Reader) Query(table string, columns []string, options ...QueryOption) (*ResultSet, error) {
+ if err := validateCString("query table", "table", table); err != nil {
return nil, err
}
- if err := validateQueryStrings("query table", "columns", query.Columns); err != nil {
+ if err := validateQueryStrings("query table", "columns", columns); err != nil {
return nil, err
}
- if query.End < query.Start {
- return nil, fmt.Errorf("%w: end must not precede start", ErrInvalidArgument)
- }
- return r.query(func(h *readerHandle) (*resultSetHandle, error) { return h.queryTable(query) })
-}
-
-// QueryTree executes a time-range query for exact full paths.
-func (r *Reader) QueryTree(query TreeQuery) (*ResultSet, error) {
- if err := validateQueryStrings("query tree", "paths", query.Paths); err != nil {
+ settings, err := buildQueryOptions(options...)
+ if err != nil {
return nil, err
}
- if query.End < query.Start {
- return nil, fmt.Errorf("%w: end must not precede start", ErrInvalidArgument)
- }
- return r.query(func(h *readerHandle) (*resultSetHandle, error) { return h.queryTree(query) })
-}
-
-func validateRows(offset, limit int, op string) error {
- if offset < 0 {
- return fmt.Errorf("%w: offset must be nonnegative", ErrInvalidArgument)
+ mode := resultModeRows
+ if settings.batchSize > 0 {
+ mode = resultModeBatches
}
- if err := validateCInt32(op, "offset", offset); err != nil {
- return err
+ table = normalizeIdentifier(table)
+ normalizedColumns := make([]string, len(columns))
+ for i, column := range columns {
+ normalizedColumns[i] = normalizeIdentifier(column)
Review Comment:
After normalization, `columns` can become case-insensitive duplicates (e.g., [\"A\", \"a\"]). This currently passes `validateQueryStrings` but can produce ambiguous results or native-side errors. Consider adding a post-normalization duplicate check (e.g., `seen` set over `normalizedColumns`) and return `ErrInvalidArgument` if duplicates exist.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@tsfile.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org