From 0449562ec02c6dc69281966b45f9c598abb74826 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 12:43:20 +0530 Subject: [PATCH 01/15] feat(go): add row format null bitmap and alignment utils --- ci/tasks/go.py | 1 + go/fory/row/bitmap.go | 55 ++++++++++++++++++++++++ go/fory/row/bitmap_test.go | 87 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 143 insertions(+) create mode 100644 go/fory/row/bitmap.go create mode 100644 go/fory/row/bitmap_test.go diff --git a/ci/tasks/go.py b/ci/tasks/go.py index 3df17f81f2..109013cf42 100644 --- a/ci/tasks/go.py +++ b/ci/tasks/go.py @@ -24,4 +24,5 @@ def run(): logging.info("Executing fory go tests") common.cd_project_subdir("go/fory") common.exec_cmd("go test -v") + common.exec_cmd("go test -v ./row/...") logging.info("Executing fory go tests succeeds") diff --git a/go/fory/row/bitmap.go b/go/fory/row/bitmap.go new file mode 100644 index 0000000000..4f406283a8 --- /dev/null +++ b/go/fory/row/bitmap.go @@ -0,0 +1,55 @@ +// 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 row implements the Fory standard row format defined in +// docs/specification/row_format_spec.md: a random-access binary format +// where each record is laid out as a null bitmap, fixed 8-byte field +// slots, and an 8-byte-aligned variable data region. +package row + +// The null bitmap tracks one bit per field or array element. Unlike the +// Arrow validity bitmap, a set bit means the value is NULL. Bits are +// LSB-first: bit 0 of byte 0 corresponds to index 0. + +// bitmapWidthInBytes returns the null bitmap size for n fields or +// elements, rounded up to a whole 8-byte word per the spec: +// ((n + 63) / 64) * 8. +func bitmapWidthInBytes(n int) int { + return ((n + 63) / 64) * 8 +} + +// setBit marks index i as null. +func setBit(bitmap []byte, i int) { + bitmap[i>>3] |= 1 << (uint(i) & 7) +} + +// clearBit marks index i as not null. +func clearBit(bitmap []byte, i int) { + bitmap[i>>3] &^= 1 << (uint(i) & 7) +} + +// getBit reports whether index i is null. +func getBit(bitmap []byte, i int) bool { + return bitmap[i>>3]&(1<<(uint(i)&7)) != 0 +} + +// roundToWord rounds n up to the nearest multiple of 8. The spec requires +// every variable-length value and data region to be zero-padded to an +// 8-byte boundary. +func roundToWord(n int) int { + return (n + 7) &^ 7 +} diff --git a/go/fory/row/bitmap_test.go b/go/fory/row/bitmap_test.go new file mode 100644 index 0000000000..37130b6870 --- /dev/null +++ b/go/fory/row/bitmap_test.go @@ -0,0 +1,87 @@ +// 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 row + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestBitmapWidthInBytes(t *testing.T) { + cases := []struct{ n, want int }{ + {0, 0}, + {1, 8}, + {10, 8}, + {63, 8}, + {64, 8}, + {65, 16}, + {128, 16}, + {129, 24}, + } + for _, c := range cases { + require.Equal(t, c.want, bitmapWidthInBytes(c.n), "n=%d", c.n) + } +} + +func TestBitOperations(t *testing.T) { + bitmap := make([]byte, 16) + + // Bit 0 of byte 0 is index 0 (LSB-first). + setBit(bitmap, 0) + require.Equal(t, byte(0b1), bitmap[0]) + require.True(t, getBit(bitmap, 0)) + + setBit(bitmap, 2) + require.Equal(t, byte(0b101), bitmap[0]) + + // Index 7 stays in byte 0; index 8 moves to byte 1. + setBit(bitmap, 7) + require.Equal(t, byte(0b10000101), bitmap[0]) + setBit(bitmap, 8) + require.Equal(t, byte(0b1), bitmap[1]) + + // Word boundary: index 63 is the last bit of the first word, + // index 64 the first bit of the second. + setBit(bitmap, 63) + require.Equal(t, byte(0b10000000), bitmap[7]) + setBit(bitmap, 64) + require.Equal(t, byte(0b1), bitmap[8]) + + // Clearing affects only the targeted bit. + clearBit(bitmap, 2) + require.False(t, getBit(bitmap, 2)) + require.True(t, getBit(bitmap, 0)) + require.True(t, getBit(bitmap, 7)) + + require.False(t, getBit(bitmap, 1)) +} + +func TestRoundToWord(t *testing.T) { + cases := []struct{ n, want int }{ + {0, 0}, + {1, 8}, + {7, 8}, + {8, 8}, + {9, 16}, + {16, 16}, + } + for _, c := range cases { + require.Equal(t, c.want, roundToWord(c.n), "n=%d", c.n) + } +} From efdb6e65eeeb08f3ec6794f076d5b9e145759c1c Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 12:50:00 +0530 Subject: [PATCH 02/15] fix: refactor in-line comments --- go/fory/row/bitmap.go | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/go/fory/row/bitmap.go b/go/fory/row/bitmap.go index 4f406283a8..4e77b0cc9b 100644 --- a/go/fory/row/bitmap.go +++ b/go/fory/row/bitmap.go @@ -15,19 +15,13 @@ // specific language governing permissions and limitations // under the License. -// Package row implements the Fory standard row format defined in -// docs/specification/row_format_spec.md: a random-access binary format -// where each record is laid out as a null bitmap, fixed 8-byte field -// slots, and an 8-byte-aligned variable data region. +// Package row implements the Fory standard row format. package row -// The null bitmap tracks one bit per field or array element. Unlike the -// Arrow validity bitmap, a set bit means the value is NULL. Bits are -// LSB-first: bit 0 of byte 0 corresponds to index 0. +// The null bitmap tracks one bit per field or array element. // bitmapWidthInBytes returns the null bitmap size for n fields or -// elements, rounded up to a whole 8-byte word per the spec: -// ((n + 63) / 64) * 8. +// elements, rounded up to a whole 8-byte word. func bitmapWidthInBytes(n int) int { return ((n + 63) / 64) * 8 } @@ -47,7 +41,7 @@ func getBit(bitmap []byte, i int) bool { return bitmap[i>>3]&(1<<(uint(i)&7)) != 0 } -// roundToWord rounds n up to the nearest multiple of 8. The spec requires +// roundToWord rounds n up to the nearest multiple of 8. Row Format requires // every variable-length value and data region to be zero-padded to an // 8-byte boundary. func roundToWord(n int) int { From 86774ce4b48d5969a74510eab9b3f21a18c889b3 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 13:11:27 +0530 Subject: [PATCH 03/15] feat(go): define fixed-width and variable-width values --- go/fory/row/datatype.go | 313 +++++++++++++++++++++++++++++++++++++ go/fory/row/schema_test.go | 136 ++++++++++++++++ 2 files changed, 449 insertions(+) create mode 100644 go/fory/row/datatype.go create mode 100644 go/fory/row/schema_test.go diff --git a/go/fory/row/datatype.go b/go/fory/row/datatype.go new file mode 100644 index 0000000000..89cc7e70cc --- /dev/null +++ b/go/fory/row/datatype.go @@ -0,0 +1,313 @@ +// 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 row + +import ( + "fmt" + "strings" + + fory "github.com/apache/fory/go/fory" +) + +// DataType describes the row-format type of a field. +// +// Type ids are the cross-language Fory type ids shared with the Java, +// C++, and Python row-format implementations; they appear verbatim in +// the serialized schema wire format. +type DataType interface { + TypeID() fory.TypeId + // ByteWidth returns the fixed storage width in bytes, or -1 for + // variable-width types stored through an offset+size slot. + ByteWidth() int + String() string +} + +// Cross-language child field names for matching Java DataTypes. +const ( + listItemName = "item" + mapKeyName = "key" + mapValueName = "value" +) + +type BoolType struct{} + +func (BoolType) TypeID() fory.TypeId { return fory.BOOL } +func (BoolType) ByteWidth() int { return 1 } +func (BoolType) String() string { return "bool" } + +type Int8Type struct{} + +func (Int8Type) TypeID() fory.TypeId { return fory.INT8 } +func (Int8Type) ByteWidth() int { return 1 } +func (Int8Type) String() string { return "int8" } + +type Int16Type struct{} + +func (Int16Type) TypeID() fory.TypeId { return fory.INT16 } +func (Int16Type) ByteWidth() int { return 2 } +func (Int16Type) String() string { return "int16" } + +type Int32Type struct{} + +func (Int32Type) TypeID() fory.TypeId { return fory.INT32 } +func (Int32Type) ByteWidth() int { return 4 } +func (Int32Type) String() string { return "int32" } + +type Int64Type struct{} + +func (Int64Type) TypeID() fory.TypeId { return fory.INT64 } +func (Int64Type) ByteWidth() int { return 8 } +func (Int64Type) String() string { return "int64" } + +// Float16Type exists so schemas received from other languages parse, +// the Go writer, reader, and encoder do not support it yet. +type Float16Type struct{} + +func (Float16Type) TypeID() fory.TypeId { return fory.FLOAT16 } +func (Float16Type) ByteWidth() int { return 2 } +func (Float16Type) String() string { return "float16" } + +type Float32Type struct{} + +func (Float32Type) TypeID() fory.TypeId { return fory.FLOAT32 } +func (Float32Type) ByteWidth() int { return 4 } +func (Float32Type) String() string { return "float32" } + +type Float64Type struct{} + +func (Float64Type) TypeID() fory.TypeId { return fory.FLOAT64 } +func (Float64Type) ByteWidth() int { return 8 } +func (Float64Type) String() string { return "float64" } + +// StringType values are UTF-8 bytes in the variable data region, +// stored as (offset + size) in the fixed slot region. +type StringType struct{} + +func (StringType) TypeID() fory.TypeId { return fory.STRING } +func (StringType) ByteWidth() int { return -1 } +func (StringType) String() string { return "string" } + +type BinaryType struct{} + +func (BinaryType) TypeID() fory.TypeId { return fory.BINARY } +func (BinaryType) ByteWidth() int { return -1 } +func (BinaryType) String() string { return "binary" } + +// Date32Type values are days since the Unix epoch (int32). +type Date32Type struct{} + +func (Date32Type) TypeID() fory.TypeId { return fory.DATE } +func (Date32Type) ByteWidth() int { return 4 } +func (Date32Type) String() string { return "date32" } + +// TimestampType values are microseconds since the Unix epoch (int64). +type TimestampType struct{} + +func (TimestampType) TypeID() fory.TypeId { return fory.TIMESTAMP } +func (TimestampType) ByteWidth() int { return 8 } +func (TimestampType) String() string { return "timestamp" } + +// DurationType values are microseconds (int64). +type DurationType struct{} + +func (DurationType) TypeID() fory.TypeId { return fory.DURATION } +func (DurationType) ByteWidth() int { return 8 } +func (DurationType) String() string { return "duration" } + +// DecimalType exists so schemas received from other languages parse; the +// Go writer, reader, and encoder do not support it yet. Use it as a +// value, not a pointer, so schema equality compares precision and scale. +type DecimalType struct { + Precision int + Scale int +} + +func (DecimalType) TypeID() fory.TypeId { return fory.DECIMAL } +func (DecimalType) ByteWidth() int { return -1 } +func (t DecimalType) String() string { + return fmt.Sprintf("decimal(%d, %d)", t.Precision, t.Scale) +} + +// ListType is a variable-length sequence of Elem values. +type ListType struct { + Elem Field +} + +func (*ListType) TypeID() fory.TypeId { return fory.LIST } +func (*ListType) ByteWidth() int { return -1 } +func (t *ListType) String() string { return "list<" + t.Elem.Type.String() + ">" } + +// MapType stores keys and values as two adjacent arrays. +type MapType struct { + Key Field + Value Field +} + +func (*MapType) TypeID() fory.TypeId { return fory.MAP } +func (*MapType) ByteWidth() int { return -1 } +func (t *MapType) String() string { + return "map<" + t.Key.Type.String() + ", " + t.Value.Type.String() + ">" +} + +// StructType is a nested row with its own bitmap, slots, and variable +// data region. +type StructType struct { + Fields []Field +} + +func (*StructType) TypeID() fory.TypeId { return fory.STRUCT } +func (*StructType) ByteWidth() int { return -1 } +func (t *StructType) String() string { + var sb strings.Builder + sb.WriteString("struct<") + for i, f := range t.Fields { + if i > 0 { + sb.WriteString(", ") + } + sb.WriteString(f.Name) + sb.WriteString(": ") + sb.WriteString(f.Type.String()) + } + sb.WriteString(">") + return sb.String() +} + +// Field is a named, optionally nullable slot in a schema or composite type. +type Field struct { + Name string + Type DataType + Nullable bool +} + +func NewField(name string, dataType DataType, nullable bool) Field { + return Field{Name: name, Type: dataType, Nullable: nullable} +} + +func (f Field) Equal(o Field) bool { + return f.Name == o.Name && f.Nullable == o.Nullable && dataTypeEqual(f.Type, o.Type) +} + +// List returns a list type with the cross-language element field name +// "item" and nullable elements. Build ListType directly for +// non-nullable elements. +func List(elem DataType) *ListType { + return &ListType{Elem: Field{Name: listItemName, Type: elem, Nullable: true}} +} + +// Map returns a map type with the cross-language field names +// "key"/"value". Keys are always non-nullable. +func Map(key, value DataType) *MapType { + return &MapType{ + Key: Field{Name: mapKeyName, Type: key, Nullable: false}, + Value: Field{Name: mapValueName, Type: value, Nullable: true}, + } +} + +func Struct(fields []Field) *StructType { + return &StructType{Fields: fields} +} + +func dataTypeEqual(a, b DataType) bool { + switch at := a.(type) { + case *ListType: + bt, ok := b.(*ListType) + return ok && at.Elem.Equal(bt.Elem) + case *MapType: + bt, ok := b.(*MapType) + return ok && at.Key.Equal(bt.Key) && at.Value.Equal(bt.Value) + case *StructType: + bt, ok := b.(*StructType) + if !ok || len(at.Fields) != len(bt.Fields) { + return false + } + for i := range at.Fields { + if !at.Fields[i].Equal(bt.Fields[i]) { + return false + } + } + return true + default: + // Primitive types are empty structs and DecimalType is a + // comparable value, so interface equality is exact. + return a == b + } +} + +// Schema describes the fields of a top-level row. +type Schema struct { + fields []Field + byName map[string]int +} + +// NewSchema builds a schema from fields in their schema-declared order, +// which fixes the field slot layout. For duplicate names, FieldIndex +// resolves to the first occurrence. +func NewSchema(fields []Field) *Schema { + byName := make(map[string]int, len(fields)) + for i, f := range fields { + if _, ok := byName[f.Name]; !ok { + byName[f.Name] = i + } + } + return &Schema{fields: fields, byName: byName} +} + +func (s *Schema) NumFields() int { return len(s.fields) } + +func (s *Schema) Field(i int) Field { return s.fields[i] } + +// Fields returns the backing field slice; callers must not modify it. +func (s *Schema) Fields() []Field { return s.fields } + +// FieldIndex returns the ordinal of the named field, or -1 if absent. +func (s *Schema) FieldIndex(name string) int { + if i, ok := s.byName[name]; ok { + return i + } + return -1 +} + +func (s *Schema) Equal(o *Schema) bool { + if s == o { + return true + } + if o == nil || len(s.fields) != len(o.fields) { + return false + } + for i := range s.fields { + if !s.fields[i].Equal(o.fields[i]) { + return false + } + } + return true +} + +func (s *Schema) String() string { + var sb strings.Builder + sb.WriteString("schema<") + for i, f := range s.fields { + if i > 0 { + sb.WriteString(", ") + } + sb.WriteString(f.Name) + sb.WriteString(": ") + sb.WriteString(f.Type.String()) + } + sb.WriteString(">") + return sb.String() +} diff --git a/go/fory/row/schema_test.go b/go/fory/row/schema_test.go new file mode 100644 index 0000000000..2fc8691579 --- /dev/null +++ b/go/fory/row/schema_test.go @@ -0,0 +1,136 @@ +// 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 row + +import ( + "testing" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +// Type ids are serialized into the cross-language schema bytes, so they +// are pinned to their literal values here. +func TestTypeIDs(t *testing.T) { + cases := []struct { + dataType DataType + id fory.TypeId + }{ + {BoolType{}, 1}, + {Int8Type{}, 2}, + {Int16Type{}, 3}, + {Int32Type{}, 4}, + {Int64Type{}, 6}, + {Float16Type{}, 17}, + {Float32Type{}, 19}, + {Float64Type{}, 20}, + {StringType{}, 21}, + {List(Int32Type{}), 22}, + {Map(StringType{}, Int64Type{}), 24}, + {Struct(nil), 27}, + {DurationType{}, 37}, + {TimestampType{}, 38}, + {Date32Type{}, 39}, + {DecimalType{Precision: 10, Scale: 2}, 40}, + {BinaryType{}, 41}, + } + for _, c := range cases { + require.Equal(t, c.id, c.dataType.TypeID(), "%s", c.dataType) + } +} + +func TestByteWidths(t *testing.T) { + cases := []struct { + dataType DataType + width int + }{ + {BoolType{}, 1}, + {Int8Type{}, 1}, + {Int16Type{}, 2}, + {Int32Type{}, 4}, + {Int64Type{}, 8}, + {Float16Type{}, 2}, + {Float32Type{}, 4}, + {Float64Type{}, 8}, + {Date32Type{}, 4}, + {TimestampType{}, 8}, + {DurationType{}, 8}, + {StringType{}, -1}, + {BinaryType{}, -1}, + {DecimalType{}, -1}, + {List(Int32Type{}), -1}, + {Map(StringType{}, Int64Type{}), -1}, + {Struct(nil), -1}, + } + for _, c := range cases { + require.Equal(t, c.width, c.dataType.ByteWidth(), "%s", c.dataType) + } +} + +func TestCompositeFactories(t *testing.T) { + list := List(StringType{}) + require.Equal(t, "item", list.Elem.Name) + require.True(t, list.Elem.Nullable) + + m := Map(StringType{}, Int32Type{}) + require.Equal(t, "key", m.Key.Name) + require.False(t, m.Key.Nullable, "map keys must be non-nullable") + require.Equal(t, "value", m.Value.Name) + require.True(t, m.Value.Nullable) + + st := Struct([]Field{NewField("a", Int32Type{}, false)}) + require.Len(t, st.Fields, 1) + require.Equal(t, "a", st.Fields[0].Name) +} + +func TestSchemaLookup(t *testing.T) { + s := NewSchema([]Field{ + NewField("id", Int64Type{}, false), + NewField("name", StringType{}, true), + }) + require.Equal(t, 2, s.NumFields()) + require.Equal(t, "id", s.Field(0).Name) + require.Equal(t, 1, s.FieldIndex("name")) + require.Equal(t, -1, s.FieldIndex("missing")) +} + +func TestSchemaEqual(t *testing.T) { + nested := func() *Schema { + return NewSchema([]Field{ + NewField("id", Int64Type{}, false), + NewField("tags", List(StringType{}), true), + NewField("attrs", Map(StringType{}, Int32Type{}), true), + NewField("inner", Struct([]Field{NewField("x", Float64Type{}, false)}), true), + NewField("price", DecimalType{Precision: 10, Scale: 2}, true), + }) + } + require.True(t, nested().Equal(nested())) + + base := NewSchema([]Field{NewField("a", Int32Type{}, false)}) + require.False(t, base.Equal(nil)) + require.False(t, base.Equal(NewSchema([]Field{NewField("b", Int32Type{}, false)}))) + require.False(t, base.Equal(NewSchema([]Field{NewField("a", Int64Type{}, false)}))) + require.False(t, base.Equal(NewSchema([]Field{NewField("a", Int32Type{}, true)}))) + require.False(t, base.Equal(NewSchema(nil))) + + // Composite mismatches must compare structurally, not by pointer. + require.False(t, NewSchema([]Field{NewField("l", List(Int32Type{}), true)}). + Equal(NewSchema([]Field{NewField("l", List(Int64Type{}), true)}))) + require.False(t, NewSchema([]Field{NewField("d", DecimalType{Precision: 10, Scale: 2}, true)}). + Equal(NewSchema([]Field{NewField("d", DecimalType{Precision: 12, Scale: 2}, true)}))) +} From 0c700628067a287001db101043bf2401be6eac67 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 17:50:13 +0530 Subject: [PATCH 04/15] feat(go): add writer to write data byte blob to the buffer --- go/fory/row/writer.go | 370 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 370 insertions(+) create mode 100644 go/fory/row/writer.go diff --git a/go/fory/row/writer.go b/go/fory/row/writer.go new file mode 100644 index 0000000000..9b612674c9 --- /dev/null +++ b/go/fory/row/writer.go @@ -0,0 +1,370 @@ +// 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 row + +import ( + "encoding/binary" + "fmt" + "math" + "time" + + fory "github.com/apache/fory/go/fory" +) + +const maxArrayDataBytes = math.MaxInt32 - 15 + +// RowWriter writes one row into a ByteBuffer: null bitmap, fixed 8-byte +// field slots, then the variable data region. All slot and bitmap +// offsets are relative to the row base, so a nested RowWriter can share +// the parent's buffer and produce a self-contained row. +// +// Reset rebases the writer at the buffer's current writer index and must +// be called before writing each row. Writers are not goroutine-safe. +type RowWriter struct { + schema *Schema + buf *fory.ByteBuffer + base int + numFields int + bitmapWidth int + fixedSize int +} + +func NewRowWriter(schema *Schema) *RowWriter { + return NewRowWriterWithBuffer(schema, fory.NewByteBuffer(nil)) +} + +// NewRowWriterWithBuffer shares an existing buffer, e.g. to write a +// nested struct row into its parent's variable data region. +func NewRowWriterWithBuffer(schema *Schema, buf *fory.ByteBuffer) *RowWriter { + n := schema.NumFields() + bitmapWidth := bitmapWidthInBytes(n) + return &RowWriter{ + schema: schema, + buf: buf, + numFields: n, + bitmapWidth: bitmapWidth, + fixedSize: bitmapWidth + n*8, + } +} + +func (w *RowWriter) Schema() *Schema { return w.schema } +func (w *RowWriter) Buffer() *fory.ByteBuffer { return w.buf } + +// Reset rebases the writer at the current writer index and zeroes the +// whole fixed region. The buffer reuses dirty capacity after shrinking, +// so zeroing here is what makes null slots and padding deterministic; +// it also lets narrow fixed-width writes store just the value. +func (w *RowWriter) Reset() { + w.base = w.buf.WriterIndex() + w.buf.Reserve(w.fixedSize) + data := w.buf.GetData() + clear(data[w.base : w.base+w.fixedSize]) + w.buf.SetWriterIndex(w.base + w.fixedSize) +} + +// Size returns the bytes written for this row so far. +func (w *RowWriter) Size() int { return w.buf.WriterIndex() - w.base } + +// ToBytes returns a view of the row bytes; it stays valid only until the +// buffer is written to or reset again. +func (w *RowWriter) ToBytes() []byte { + return w.buf.GetByteSlice(w.base, w.buf.WriterIndex()) +} + +func (w *RowWriter) slot(i int) int { + if uint(i) >= uint(w.numFields) { + panic(fmt.Sprintf("row: field index %d out of range [0, %d)", i, w.numFields)) + } + return w.base + w.bitmapWidth + i*8 +} + +// SetNullAt marks field i null and zeroes its slot. +func (w *RowWriter) SetNullAt(i int) { + slot := w.slot(i) + data := w.buf.GetData() + setBit(data[w.base:w.base+w.bitmapWidth], i) + binary.LittleEndian.PutUint64(data[slot:], 0) +} + +func (w *RowWriter) SetNotNullAt(i int) { + w.slot(i) // bounds check + clearBit(w.buf.GetData()[w.base:w.base+w.bitmapWidth], i) +} + +func (w *RowWriter) WriteBool(i int, v bool) { + slot := w.slot(i) + if v { + w.buf.GetData()[slot] = 1 + } else { + w.buf.GetData()[slot] = 0 + } +} + +func (w *RowWriter) WriteInt8(i int, v int8) { + w.buf.GetData()[w.slot(i)] = byte(v) +} + +func (w *RowWriter) WriteInt16(i int, v int16) { + binary.LittleEndian.PutUint16(w.buf.GetData()[w.slot(i):], uint16(v)) +} + +func (w *RowWriter) WriteInt32(i int, v int32) { + binary.LittleEndian.PutUint32(w.buf.GetData()[w.slot(i):], uint32(v)) +} + +func (w *RowWriter) WriteInt64(i int, v int64) { + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(v)) +} + +func (w *RowWriter) WriteFloat32(i int, v float32) { + binary.LittleEndian.PutUint32(w.buf.GetData()[w.slot(i):], math.Float32bits(v)) +} + +func (w *RowWriter) WriteFloat64(i int, v float64) { + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], math.Float64bits(v)) +} + +func (w *RowWriter) WriteDate(i int, d fory.Date) error { + days, err := fory.DateToEpochDay(d) + if err != nil { + return err + } + if days < math.MinInt32 || days > math.MaxInt32 { + return fmt.Errorf("row: date %v out of date32 range", d) + } + w.WriteInt32(i, int32(days)) + return nil +} + +func (w *RowWriter) WriteTimestamp(i int, t time.Time) { + w.WriteInt64(i, t.UnixMicro()) +} + +func (w *RowWriter) WriteDuration(i int, d time.Duration) { + w.WriteInt64(i, d.Microseconds()) +} + +func (w *RowWriter) WriteString(i int, s string) { + start := appendStringRegion(w.buf, s) + w.SetOffsetAndSize(i, start, len(s)) +} + +func (w *RowWriter) WriteBytes(i int, b []byte) { + start := appendBytesRegion(w.buf, b) + w.SetOffsetAndSize(i, start, len(b)) +} + +// SetOffsetAndSize patches field i's slot with the row-relative offset +// and byte size of a value already appended to the variable data region. +// Use it after writing a nested struct, array, or map at absStart. +func (w *RowWriter) SetOffsetAndSize(i, absStart, size int) { + rel := absStart - w.base + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(uint32(size))) +} + +// ArrayWriter writes one array: an 8-byte element count, a null bitmap, +// then elements at their natural width (variable-width elements use +// 8-byte offset+size slots). Offsets are relative to the array base. +type ArrayWriter struct { + elem Field + buf *fory.ByteBuffer + base int + elemSize int + numElements int + headerBytes int +} + +func NewArrayWriter(elem Field, buf *fory.ByteBuffer) *ArrayWriter { + elemSize := elem.Type.ByteWidth() + if elemSize < 0 { + elemSize = 8 + } + return &ArrayWriter{elem: elem, buf: buf, elemSize: elemSize} +} + +func (w *ArrayWriter) Buffer() *fory.ByteBuffer { return w.buf } + +// Reset rebases the writer at the current writer index and writes the +// array header plus a zeroed element region for numElements elements. +func (w *ArrayWriter) Reset(numElements int) error { + if numElements < 0 { + return fmt.Errorf("row: negative array length %d", numElements) + } + dataBytes := int64(numElements) * int64(w.elemSize) + if dataBytes > maxArrayDataBytes { + return fmt.Errorf("row: array of %d elements exceeds maximum size", numElements) + } + headerBytes := 8 + bitmapWidthInBytes(numElements) + total := headerBytes + roundToWord(int(dataBytes)) + base := w.buf.WriterIndex() + w.buf.Reserve(total) + data := w.buf.GetData() + binary.LittleEndian.PutUint64(data[base:], uint64(numElements)) + clear(data[base+8 : base+total]) + w.buf.SetWriterIndex(base + total) + w.base, w.numElements, w.headerBytes = base, numElements, headerBytes + return nil +} + +func (w *ArrayWriter) Size() int { return w.buf.WriterIndex() - w.base } + +func (w *ArrayWriter) ToBytes() []byte { + return w.buf.GetByteSlice(w.base, w.buf.WriterIndex()) +} + +func (w *ArrayWriter) slot(i int) int { + if uint(i) >= uint(w.numElements) { + panic(fmt.Sprintf("row: array index %d out of range [0, %d)", i, w.numElements)) + } + return w.base + w.headerBytes + i*w.elemSize +} + +// SetNullAt marks element i null and re-zeroes its element bytes. +func (w *ArrayWriter) SetNullAt(i int) { + slot := w.slot(i) + data := w.buf.GetData() + setBit(data[w.base+8:w.base+w.headerBytes], i) + clear(data[slot : slot+w.elemSize]) +} + +func (w *ArrayWriter) WriteBool(i int, v bool) { + slot := w.slot(i) + if v { + w.buf.GetData()[slot] = 1 + } else { + w.buf.GetData()[slot] = 0 + } +} + +func (w *ArrayWriter) WriteInt8(i int, v int8) { + w.buf.GetData()[w.slot(i)] = byte(v) +} + +func (w *ArrayWriter) WriteInt16(i int, v int16) { + binary.LittleEndian.PutUint16(w.buf.GetData()[w.slot(i):], uint16(v)) +} + +func (w *ArrayWriter) WriteInt32(i int, v int32) { + binary.LittleEndian.PutUint32(w.buf.GetData()[w.slot(i):], uint32(v)) +} + +func (w *ArrayWriter) WriteInt64(i int, v int64) { + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(v)) +} + +func (w *ArrayWriter) WriteFloat32(i int, v float32) { + binary.LittleEndian.PutUint32(w.buf.GetData()[w.slot(i):], math.Float32bits(v)) +} + +func (w *ArrayWriter) WriteFloat64(i int, v float64) { + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], math.Float64bits(v)) +} + +func (w *ArrayWriter) WriteDate(i int, d fory.Date) error { + days, err := fory.DateToEpochDay(d) + if err != nil { + return err + } + if days < math.MinInt32 || days > math.MaxInt32 { + return fmt.Errorf("row: date %v out of date32 range", d) + } + w.WriteInt32(i, int32(days)) + return nil +} + +func (w *ArrayWriter) WriteTimestamp(i int, t time.Time) { + w.WriteInt64(i, t.UnixMicro()) +} + +func (w *ArrayWriter) WriteDuration(i int, d time.Duration) { + w.WriteInt64(i, d.Microseconds()) +} + +func (w *ArrayWriter) WriteString(i int, s string) { + start := appendStringRegion(w.buf, s) + w.SetOffsetAndSize(i, start, len(s)) +} + +func (w *ArrayWriter) WriteBytes(i int, b []byte) { + start := appendBytesRegion(w.buf, b) + w.SetOffsetAndSize(i, start, len(b)) +} + +// SetOffsetAndSize patches element i's slot with the array-relative +// offset and byte size of a value already appended after the array. +func (w *ArrayWriter) SetOffsetAndSize(i, absStart, size int) { + rel := absStart - w.base + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(uint32(size))) +} + +// MapWriter writes one map: an 8-byte keys-array size, the keys array, +// then the values array. Write flow: Reset, write the keys with an +// ArrayWriter, FinishKeys, write the values with another ArrayWriter. +type MapWriter struct { + buf *fory.ByteBuffer + base int +} + +func NewMapWriter(buf *fory.ByteBuffer) *MapWriter { + return &MapWriter{buf: buf} +} + +// Reset rebases the writer at the current writer index and reserves the +// keys-array size word, which FinishKeys patches later. +func (w *MapWriter) Reset() { + w.base = w.buf.WriterIndex() + w.buf.Reserve(8) + data := w.buf.GetData() + clear(data[w.base : w.base+8]) + w.buf.SetWriterIndex(w.base + 8) +} + +// FinishKeys records the keys array size; call it after the keys array +// is fully written and before starting the values array. +func (w *MapWriter) FinishKeys() { + keysSize := w.buf.WriterIndex() - w.base - 8 + binary.LittleEndian.PutUint64(w.buf.GetData()[w.base:], uint64(keysSize)) +} + +func (w *MapWriter) Size() int { return w.buf.WriterIndex() - w.base } + +// appendBytesRegion appends b to the variable data region, zero-padded +// to an 8-byte boundary, and returns its buffer offset. +func appendBytesRegion(buf *fory.ByteBuffer, b []byte) int { + n := len(b) + rounded := roundToWord(n) + start := buf.WriterIndex() + buf.Reserve(rounded) + data := buf.GetData() + clear(data[start+n : start+rounded]) + copy(data[start:], b) + buf.SetWriterIndex(start + rounded) + return start +} + +func appendStringRegion(buf *fory.ByteBuffer, s string) int { + n := len(s) + rounded := roundToWord(n) + start := buf.WriterIndex() + buf.Reserve(rounded) + data := buf.GetData() + clear(data[start+n : start+rounded]) + copy(data[start:], s) + buf.SetWriterIndex(start + rounded) + return start +} From 9680d573a9966eecedefe07ba8b0ca39ae1a94c0 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 18:12:44 +0530 Subject: [PATCH 05/15] feat(go): add row reader --- go/fory/row/row.go | 338 ++++++++++++++++++++++++++++++++++++++++ go/fory/row/row_test.go | 264 +++++++++++++++++++++++++++++++ 2 files changed, 602 insertions(+) create mode 100644 go/fory/row/row.go create mode 100644 go/fory/row/row_test.go diff --git a/go/fory/row/row.go b/go/fory/row/row.go new file mode 100644 index 0000000000..e2780b95c9 --- /dev/null +++ b/go/fory/row/row.go @@ -0,0 +1,338 @@ +// 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 row + +import ( + "encoding/binary" + "fmt" + "math" + "time" + + fory "github.com/apache/fory/go/fory" +) + +// Row is a zero-copy reader over one row's bytes. Getters read directly +// from the underlying data without deserializing other fields; nested +// values are sub-slice views, never copies (except String). +// +// Fixed-width getters return the zero value for null fields; use +// IsNullAt to distinguish. Variable-width getters return nil (or "") +// for null fields. An out-of-range field index panics; offsets in +// corrupt or untrusted data may panic on slice bounds, so wrap +// untrusted decoding with recover. +type Row struct { + fields []Field + data []byte + bitmapWidth int +} + +func NewRow(schema *Schema, data []byte) *Row { + return newRow(schema.Fields(), data) +} + +func newRow(fields []Field, data []byte) *Row { + return &Row{fields: fields, data: data, bitmapWidth: bitmapWidthInBytes(len(fields))} +} + +func (r *Row) NumFields() int { return len(r.fields) } +func (r *Row) SizeBytes() int { return len(r.data) } + +// Data returns the underlying row bytes; callers must not modify them. +func (r *Row) Data() []byte { return r.data } + +func (r *Row) IsNullAt(i int) bool { + if uint(i) >= uint(len(r.fields)) { + panic(fmt.Sprintf("row: field index %d out of range [0, %d)", i, len(r.fields))) + } + return getBit(r.data, i) +} + +func (r *Row) slot(i int) int { return r.bitmapWidth + i*8 } + +func (r *Row) Bool(i int) bool { + return !r.IsNullAt(i) && r.data[r.slot(i)] != 0 +} + +func (r *Row) Int8(i int) int8 { + if r.IsNullAt(i) { + return 0 + } + return int8(r.data[r.slot(i)]) +} + +func (r *Row) Int16(i int) int16 { + if r.IsNullAt(i) { + return 0 + } + return int16(binary.LittleEndian.Uint16(r.data[r.slot(i):])) +} + +func (r *Row) Int32(i int) int32 { + if r.IsNullAt(i) { + return 0 + } + return int32(binary.LittleEndian.Uint32(r.data[r.slot(i):])) +} + +func (r *Row) Int64(i int) int64 { + if r.IsNullAt(i) { + return 0 + } + return int64(binary.LittleEndian.Uint64(r.data[r.slot(i):])) +} + +func (r *Row) Float32(i int) float32 { + if r.IsNullAt(i) { + return 0 + } + return math.Float32frombits(binary.LittleEndian.Uint32(r.data[r.slot(i):])) +} + +func (r *Row) Float64(i int) float64 { + if r.IsNullAt(i) { + return 0 + } + return math.Float64frombits(binary.LittleEndian.Uint64(r.data[r.slot(i):])) +} + +func (r *Row) Date(i int) fory.Date { + if r.IsNullAt(i) { + return fory.Date{} + } + return dateFromDays(int32(binary.LittleEndian.Uint32(r.data[r.slot(i):]))) +} + +func (r *Row) Timestamp(i int) time.Time { + if r.IsNullAt(i) { + return time.Time{} + } + return time.UnixMicro(r.Int64(i)) +} + +func (r *Row) Duration(i int) time.Duration { + return time.Duration(r.Int64(i)) * time.Microsecond +} + +// varData returns the value bytes of variable-width field i, or nil if +// the field is null. An empty value returns an empty non-nil slice. +func (r *Row) varData(i int) []byte { + if r.IsNullAt(i) { + return nil + } + offset, size := decodeOffsetAndSize(binary.LittleEndian.Uint64(r.data[r.slot(i):])) + return r.data[offset : offset+size] +} + +// Binary returns a zero-copy view of field i's bytes. +func (r *Row) Binary(i int) []byte { return r.varData(i) } + +func (r *Row) String(i int) string { return string(r.varData(i)) } + +func (r *Row) Struct(i int) *Row { + data := r.varData(i) + if data == nil { + return nil + } + structType := r.fields[i].Type.(*StructType) + return newRow(structType.Fields, data) +} + +func (r *Row) Array(i int) *ArrayData { + data := r.varData(i) + if data == nil { + return nil + } + listType := r.fields[i].Type.(*ListType) + return NewArrayData(listType.Elem, data) +} + +func (r *Row) Map(i int) *MapData { + data := r.varData(i) + if data == nil { + return nil + } + return NewMapData(r.fields[i].Type.(*MapType), data) +} + +// ArrayData is a zero-copy reader over one array's bytes, with the same +// null and bounds semantics as Row. +type ArrayData struct { + elem Field + data []byte + numElements int + elemSize int + headerBytes int +} + +func NewArrayData(elem Field, data []byte) *ArrayData { + numElements := int(binary.LittleEndian.Uint64(data)) + elemSize := elem.Type.ByteWidth() + if elemSize < 0 { + elemSize = 8 + } + return &ArrayData{ + elem: elem, + data: data, + numElements: numElements, + elemSize: elemSize, + headerBytes: 8 + bitmapWidthInBytes(numElements), + } +} + +func (a *ArrayData) NumElements() int { return a.numElements } +func (a *ArrayData) SizeBytes() int { return len(a.data) } + +func (a *ArrayData) IsNullAt(i int) bool { + if uint(i) >= uint(a.numElements) { + panic(fmt.Sprintf("row: array index %d out of range [0, %d)", i, a.numElements)) + } + return getBit(a.data[8:a.headerBytes], i) +} + +func (a *ArrayData) slot(i int) int { return a.headerBytes + i*a.elemSize } + +func (a *ArrayData) Bool(i int) bool { + return !a.IsNullAt(i) && a.data[a.slot(i)] != 0 +} + +func (a *ArrayData) Int8(i int) int8 { + if a.IsNullAt(i) { + return 0 + } + return int8(a.data[a.slot(i)]) +} + +func (a *ArrayData) Int16(i int) int16 { + if a.IsNullAt(i) { + return 0 + } + return int16(binary.LittleEndian.Uint16(a.data[a.slot(i):])) +} + +func (a *ArrayData) Int32(i int) int32 { + if a.IsNullAt(i) { + return 0 + } + return int32(binary.LittleEndian.Uint32(a.data[a.slot(i):])) +} + +func (a *ArrayData) Int64(i int) int64 { + if a.IsNullAt(i) { + return 0 + } + return int64(binary.LittleEndian.Uint64(a.data[a.slot(i):])) +} + +func (a *ArrayData) Float32(i int) float32 { + if a.IsNullAt(i) { + return 0 + } + return math.Float32frombits(binary.LittleEndian.Uint32(a.data[a.slot(i):])) +} + +func (a *ArrayData) Float64(i int) float64 { + if a.IsNullAt(i) { + return 0 + } + return math.Float64frombits(binary.LittleEndian.Uint64(a.data[a.slot(i):])) +} + +func (a *ArrayData) Date(i int) fory.Date { + if a.IsNullAt(i) { + return fory.Date{} + } + return dateFromDays(int32(binary.LittleEndian.Uint32(a.data[a.slot(i):]))) +} + +func (a *ArrayData) Timestamp(i int) time.Time { + if a.IsNullAt(i) { + return time.Time{} + } + return time.UnixMicro(a.Int64(i)) +} + +func (a *ArrayData) Duration(i int) time.Duration { + return time.Duration(a.Int64(i)) * time.Microsecond +} + +func (a *ArrayData) varData(i int) []byte { + if a.IsNullAt(i) { + return nil + } + offset, size := decodeOffsetAndSize(binary.LittleEndian.Uint64(a.data[a.slot(i):])) + return a.data[offset : offset+size] +} + +// Binary returns a zero-copy view of element i's bytes. +func (a *ArrayData) Binary(i int) []byte { return a.varData(i) } + +func (a *ArrayData) String(i int) string { return string(a.varData(i)) } + +func (a *ArrayData) Struct(i int) *Row { + data := a.varData(i) + if data == nil { + return nil + } + return newRow(a.elem.Type.(*StructType).Fields, data) +} + +func (a *ArrayData) Array(i int) *ArrayData { + data := a.varData(i) + if data == nil { + return nil + } + return NewArrayData(a.elem.Type.(*ListType).Elem, data) +} + +func (a *ArrayData) Map(i int) *MapData { + data := a.varData(i) + if data == nil { + return nil + } + return NewMapData(a.elem.Type.(*MapType), data) +} + +// MapData is a zero-copy reader over one map's bytes: the keys array +// and values array views share the map's underlying data. +type MapData struct { + keys *ArrayData + values *ArrayData +} + +func NewMapData(mapType *MapType, data []byte) *MapData { + keysSize := int(binary.LittleEndian.Uint64(data)) + return &MapData{ + keys: NewArrayData(mapType.Key, data[8:8+keysSize]), + values: NewArrayData(mapType.Value, data[8+keysSize:]), + } +} + +func (m *MapData) NumElements() int { return m.keys.numElements } +func (m *MapData) Keys() *ArrayData { return m.keys } +func (m *MapData) Values() *ArrayData { return m.values } + +func decodeOffsetAndSize(slotValue uint64) (offset, size int) { + return int(slotValue >> 32), int(uint32(slotValue)) +} + +// dateFromDays converts int32 epoch days to a Date. The int32 range +// always yields an in-range year, so the conversion cannot fail. +func dateFromDays(days int32) fory.Date { + d, _ := fory.DateFromEpochDay(int64(days)) + return d +} diff --git a/go/fory/row/row_test.go b/go/fory/row/row_test.go new file mode 100644 index 0000000000..55f124a104 --- /dev/null +++ b/go/fory/row/row_test.go @@ -0,0 +1,264 @@ +// 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 row + +import ( + "strings" + "sync" + "testing" + "time" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +func int64StringSchema() *Schema { + return NewSchema([]Field{ + NewField("f1", Int64Type{}, false), + NewField("f2", StringType{}, true), + }) +} + +// The full byte image is fixed by the spec: 8-byte bitmap, one 8-byte +// slot per field (variable-width slots hold offset<<32|size relative to +// the row base), then the variable region zero-padded to 8 bytes. +func TestRowExactLayout(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + w.WriteInt64(0, 0x0102030405060708) + w.WriteString(1, "ab") + require.Equal(t, []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, // null bitmap + 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01, // f1 slot + 0x02, 0x00, 0x00, 0x00, 0x18, 0x00, 0x00, 0x00, // f2 slot: size=2, offset=24 + 'a', 'b', 0, 0, 0, 0, 0, 0, // variable region, zero-padded + }, w.ToBytes()) + + r := NewRow(w.Schema(), w.ToBytes()) + require.Equal(t, int64(0x0102030405060708), r.Int64(0)) + require.Equal(t, "ab", r.String(1)) +} + +func TestNullAndEmptyString(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + w.WriteInt64(0, 42) + w.SetNullAt(1) + require.Equal(t, []byte{ + 0x02, 0, 0, 0, 0, 0, 0, 0, // bit 1 set: f2 is null + 42, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, // null slot zeroed + }, w.ToBytes()) + r := NewRow(w.Schema(), w.ToBytes()) + require.True(t, r.IsNullAt(1)) + require.Nil(t, r.Binary(1)) + require.Equal(t, "", r.String(1)) + + // An empty string is not null: bit clear, size 0, no variable bytes. + w.Buffer().SetWriterIndex(0) + w.Reset() + w.WriteInt64(0, 42) + w.WriteString(1, "") + r = NewRow(w.Schema(), w.ToBytes()) + require.Equal(t, 24, r.SizeBytes()) + require.False(t, r.IsNullAt(1)) + require.Equal(t, "", r.String(1)) + require.NotNil(t, r.Binary(1)) + require.Empty(t, r.Binary(1)) +} + +func TestPrimitiveRoundTrip(t *testing.T) { + s := NewSchema([]Field{ + NewField("b", BoolType{}, false), + NewField("i8", Int8Type{}, false), + NewField("i16", Int16Type{}, false), + NewField("i32", Int32Type{}, false), + NewField("i64", Int64Type{}, false), + NewField("f32", Float32Type{}, false), + NewField("f64", Float64Type{}, false), + NewField("bin", BinaryType{}, true), + }) + w := NewRowWriter(s) + w.Reset() + w.WriteBool(0, true) + w.WriteInt8(1, -5) + w.WriteInt16(2, -1000) + w.WriteInt32(3, -100000) + w.WriteInt64(4, -5_000_000_000_000) + w.WriteFloat32(5, 3.5) + w.WriteFloat64(6, -2.25) + w.WriteBytes(7, []byte{1, 2, 3}) + + r := NewRow(s, w.ToBytes()) + require.True(t, r.Bool(0)) + require.Equal(t, int8(-5), r.Int8(1)) + require.Equal(t, int16(-1000), r.Int16(2)) + require.Equal(t, int32(-100000), r.Int32(3)) + require.Equal(t, int64(-5_000_000_000_000), r.Int64(4)) + require.Equal(t, float32(3.5), r.Float32(5)) + require.Equal(t, -2.25, r.Float64(6)) + require.Equal(t, []byte{1, 2, 3}, r.Binary(7)) +} + +func TestTemporalRoundTrip(t *testing.T) { + s := NewSchema([]Field{ + NewField("d", Date32Type{}, false), + NewField("ts", TimestampType{}, false), + NewField("dur", DurationType{}, false), + }) + w := NewRowWriter(s) + w.Reset() + date := fory.Date{Year: 2023, Month: time.March, Day: 15} + require.NoError(t, w.WriteDate(0, date)) + w.WriteTimestamp(1, time.UnixMicro(1_234_567_890_123_456)) + w.WriteDuration(2, 90*time.Second) + + r := NewRow(s, w.ToBytes()) + require.Equal(t, date, r.Date(0)) + require.Equal(t, int64(1_234_567_890_123_456), r.Timestamp(1).UnixMicro()) + require.Equal(t, 90*time.Second, r.Duration(2)) +} + +// A 65-field schema needs a 16-byte bitmap, shifting every slot by 16. +func TestMultiWordBitmap(t *testing.T) { + fields := make([]Field, 65) + for i := range fields { + fields[i] = NewField("f"+string(rune('a'+i%26))+string(rune('0'+i/26)), Int64Type{}, true) + } + s := NewSchema(fields) + w := NewRowWriter(s) + w.Reset() + w.WriteInt64(0, 100) + w.WriteInt64(64, 200) + w.SetNullAt(63) + + data := w.ToBytes() + require.Equal(t, 16+65*8, len(data)) + require.Equal(t, byte(0x80), data[7], "bit 63 is the top bit of bitmap byte 7") + + r := NewRow(s, data) + require.Equal(t, int64(100), r.Int64(0)) + require.Equal(t, int64(200), r.Int64(64)) + require.True(t, r.IsNullAt(63)) + require.Equal(t, int64(0), r.Int64(63)) +} + +// The buffer reuses dirty capacity, so a rewritten smaller row must +// still produce byte-identical output with zeroed padding and slots. +func TestWriterReuseZeroesDirtyBuffer(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + w.WriteInt64(0, 1) + w.WriteString(1, strings.Repeat("Z", 64)) + + w.Buffer().SetWriterIndex(0) + w.Reset() + w.WriteInt64(0, 0x0102030405060708) + w.WriteString(1, "ab") + require.Equal(t, []byte{ + 0, 0, 0, 0, 0, 0, 0, 0, + 0x08, 0x07, 0x06, 0x05, 0x04, 0x03, 0x02, 0x01, + 0x02, 0x00, 0x00, 0x00, 0x18, 0x00, 0x00, 0x00, + 'a', 'b', 0, 0, 0, 0, 0, 0, + }, w.ToBytes()) + + w.Buffer().SetWriterIndex(0) + w.Reset() + w.WriteInt64(0, 42) + w.SetNullAt(1) + require.Equal(t, []byte{ + 0x02, 0, 0, 0, 0, 0, 0, 0, + 42, 0, 0, 0, 0, 0, 0, 0, + 0, 0, 0, 0, 0, 0, 0, 0, + }, w.ToBytes()) +} + +func TestNestedStruct(t *testing.T) { + inner := Struct([]Field{ + NewField("x", Float64Type{}, false), + NewField("s", StringType{}, true), + }) + s := NewSchema([]Field{ + NewField("id", Int64Type{}, false), + NewField("inner", inner, true), + }) + w := NewRowWriter(s) + w.Reset() + w.WriteInt64(0, 7) + child := NewRowWriterWithBuffer(NewSchema(inner.Fields), w.Buffer()) + start := w.Buffer().WriterIndex() + child.Reset() + child.WriteFloat64(0, 3.5) + child.WriteString(1, "hi") + w.SetOffsetAndSize(1, start, w.Buffer().WriterIndex()-start) + + r := NewRow(s, w.ToBytes()) + require.Equal(t, int64(7), r.Int64(0)) + nested := r.Struct(1) + require.NotNil(t, nested) + require.Equal(t, 3.5, nested.Float64(0)) + require.Equal(t, "hi", nested.String(1)) +} + +// A value larger than the initial buffer forces mid-row growth; slots +// patched after reallocation must still land in the right place. +func TestLargeStringGrowth(t *testing.T) { + s := NewSchema([]Field{NewField("s", StringType{}, true)}) + w := NewRowWriter(s) + w.Reset() + big := strings.Repeat("x", 5000) + w.WriteString(0, big) + + r := NewRow(s, w.ToBytes()) + require.Equal(t, big, r.String(0)) + require.Equal(t, 8+8+roundToWord(5000), r.SizeBytes()) +} + +// One-past-the-end indices must be rejected, matching the other +// row format implementations. +func TestOutOfRangeIndexPanics(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + require.Panics(t, func() { w.WriteInt64(2, 1) }) + require.Panics(t, func() { w.SetNullAt(-1) }) + + r := NewRow(w.Schema(), w.ToBytes()) + require.Panics(t, func() { r.Int64(2) }) + require.Panics(t, func() { r.IsNullAt(-1) }) +} + +func TestConcurrentReads(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + w.WriteInt64(0, 99) + w.WriteString(1, "shared") + r := NewRow(w.Schema(), w.ToBytes()) + + var wg sync.WaitGroup + for g := 0; g < 4; g++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 200; j++ { + require.Equal(t, int64(99), r.Int64(0)) + require.Equal(t, "shared", r.String(1)) + } + }() + } + wg.Wait() +} From 81ae126d85918d7d8ae553f6c58cd89670bf2a22 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 18:13:16 +0530 Subject: [PATCH 06/15] feat(test): add array and map reader tests --- go/fory/row/array_test.go | 108 ++++++++++++++++++++++++++++++++++++++ go/fory/row/map_test.go | 90 +++++++++++++++++++++++++++++++ 2 files changed, 198 insertions(+) create mode 100644 go/fory/row/array_test.go create mode 100644 go/fory/row/map_test.go diff --git a/go/fory/row/array_test.go b/go/fory/row/array_test.go new file mode 100644 index 0000000000..9ede434760 --- /dev/null +++ b/go/fory/row/array_test.go @@ -0,0 +1,108 @@ +// 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 row + +import ( + "testing" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +// Arrays store an 8-byte count, a null bitmap rounded to 8-byte words, +// then elements at their natural width, padded to an 8-byte boundary. +func TestArrayExactLayout(t *testing.T) { + buf := fory.NewByteBuffer(nil) + w := NewArrayWriter(List(Int32Type{}).Elem, buf) + require.NoError(t, w.Reset(3)) + w.WriteInt32(0, 1) + w.WriteInt32(1, 2) + w.WriteInt32(2, 3) + require.Equal(t, []byte{ + 3, 0, 0, 0, 0, 0, 0, 0, // element count + 0, 0, 0, 0, 0, 0, 0, 0, // null bitmap + 1, 0, 0, 0, 2, 0, 0, 0, // int32 elements at natural width + 3, 0, 0, 0, 0, 0, 0, 0, // last element + alignment padding + }, w.ToBytes()) + + a := NewArrayData(List(Int32Type{}).Elem, w.ToBytes()) + require.Equal(t, 3, a.NumElements()) + require.Equal(t, int32(2), a.Int32(1)) +} + +func TestStringArrayWithNullElement(t *testing.T) { + buf := fory.NewByteBuffer(nil) + elem := List(StringType{}).Elem + w := NewArrayWriter(elem, buf) + require.NoError(t, w.Reset(3)) + w.WriteString(0, "str1") + w.SetNullAt(1) + w.WriteString(2, "str2") + + a := NewArrayData(elem, w.ToBytes()) + require.Equal(t, 3, a.NumElements()) + require.Equal(t, "str1", a.String(0)) + require.True(t, a.IsNullAt(1)) + require.Equal(t, "", a.String(1)) + require.Equal(t, "str2", a.String(2)) +} + +func TestEmptyArray(t *testing.T) { + buf := fory.NewByteBuffer(nil) + elem := List(Int64Type{}).Elem + w := NewArrayWriter(elem, buf) + require.NoError(t, w.Reset(0)) + require.Equal(t, []byte{0, 0, 0, 0, 0, 0, 0, 0}, w.ToBytes()) + + a := NewArrayData(elem, w.ToBytes()) + require.Equal(t, 0, a.NumElements()) + require.Panics(t, func() { a.Int64(0) }) +} + +func TestNestedArray(t *testing.T) { + buf := fory.NewByteBuffer(nil) + outerElem := List(List(Int32Type{})).Elem // item: list + innerElem := List(Int32Type{}).Elem // item: int32 + + outer := NewArrayWriter(outerElem, buf) + require.NoError(t, outer.Reset(2)) + inner := NewArrayWriter(innerElem, buf) + for i, values := range [][]int32{{1, 2}, {3, 4, 5}} { + start := buf.WriterIndex() + require.NoError(t, inner.Reset(len(values))) + for j, v := range values { + inner.WriteInt32(j, v) + } + outer.SetOffsetAndSize(i, start, buf.WriterIndex()-start) + } + + a := NewArrayData(outerElem, outer.ToBytes()) + require.Equal(t, 2, a.NumElements()) + first := a.Array(0) + require.Equal(t, 2, first.NumElements()) + require.Equal(t, int32(2), first.Int32(1)) + second := a.Array(1) + require.Equal(t, 3, second.NumElements()) + require.Equal(t, int32(5), second.Int32(2)) +} + +func TestArrayResetRejectsInvalidLength(t *testing.T) { + w := NewArrayWriter(List(Int64Type{}).Elem, fory.NewByteBuffer(nil)) + require.Error(t, w.Reset(-1)) + require.Error(t, w.Reset(1<<40)) +} diff --git a/go/fory/row/map_test.go b/go/fory/row/map_test.go new file mode 100644 index 0000000000..e1f1fa4f93 --- /dev/null +++ b/go/fory/row/map_test.go @@ -0,0 +1,90 @@ +// 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 row + +import ( + "encoding/binary" + "testing" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +// Maps are an 8-byte keys-array size followed by two complete arrays; +// both halves must decode independently. +func TestMapLayout(t *testing.T) { + buf := fory.NewByteBuffer(nil) + mapType := Map(StringType{}, Int64Type{}) + m := NewMapWriter(buf) + m.Reset() + keys := NewArrayWriter(mapType.Key, buf) + require.NoError(t, keys.Reset(2)) + keys.WriteString(0, "k1") + keys.WriteString(1, "k2") + m.FinishKeys() + values := NewArrayWriter(mapType.Value, buf) + require.NoError(t, values.Reset(2)) + values.WriteInt64(0, 10) + values.WriteInt64(1, 20) + + data := buf.GetByteSlice(0, buf.WriterIndex()) + // keys array: 8 count + 8 bitmap + 2*8 slots + 2*8 padded strings. + require.Equal(t, uint64(48), binary.LittleEndian.Uint64(data)) + require.Equal(t, 8+48+32, m.Size()) + + md := NewMapData(mapType, data) + require.Equal(t, 2, md.NumElements()) + require.Equal(t, "k1", md.Keys().String(0)) + require.Equal(t, "k2", md.Keys().String(1)) + require.Equal(t, int64(10), md.Values().Int64(0)) + require.Equal(t, int64(20), md.Values().Int64(1)) +} + +func TestMapFieldInRow(t *testing.T) { + mapType := Map(StringType{}, Int32Type{}) + s := NewSchema([]Field{ + NewField("id", Int64Type{}, false), + NewField("attrs", mapType, true), + }) + w := NewRowWriter(s) + w.Reset() + w.WriteInt64(0, 1) + + buf := w.Buffer() + start := buf.WriterIndex() + m := NewMapWriter(buf) + m.Reset() + keys := NewArrayWriter(mapType.Key, buf) + require.NoError(t, keys.Reset(2)) + keys.WriteString(0, "a") + keys.WriteString(1, "b") + m.FinishKeys() + values := NewArrayWriter(mapType.Value, buf) + require.NoError(t, values.Reset(2)) + values.WriteInt32(0, 1) + values.SetNullAt(1) + w.SetOffsetAndSize(1, start, buf.WriterIndex()-start) + + r := NewRow(s, w.ToBytes()) + md := r.Map(1) + require.NotNil(t, md) + require.Equal(t, 2, md.NumElements()) + require.Equal(t, "a", md.Keys().String(0)) + require.Equal(t, int32(1), md.Values().Int32(0)) + require.True(t, md.Values().IsNullAt(1)) +} From 6005b886d2a767e3856f4467f7237893fda7bf20 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 18:43:18 +0530 Subject: [PATCH 07/15] feat(schema): add schema bytes methods to interop with Java --- go/fory/row/schema_bytes.go | 343 +++++++++++++++++++++++++++++++ go/fory/row/schema_bytes_test.go | 161 +++++++++++++++ 2 files changed, 504 insertions(+) create mode 100644 go/fory/row/schema_bytes.go create mode 100644 go/fory/row/schema_bytes_test.go diff --git a/go/fory/row/schema_bytes.go b/go/fory/row/schema_bytes.go new file mode 100644 index 0000000000..0150789c7f --- /dev/null +++ b/go/fory/row/schema_bytes.go @@ -0,0 +1,343 @@ +// 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 row + +import ( + "fmt" + "math" + + fory "github.com/apache/fory/go/fory" + "github.com/apache/fory/go/fory/meta" +) + +// Schema wire format, byte-compatible with Java SchemaEncoder: +// +// | version (1 byte) | num_fields (var uint32 small7) | fields... | +// +// Field: | header (1 byte) | name size (varint, only if large) | name +// bytes | type info |. Header bits 0-1 hold the index into +// fieldNameEncodings, bits 2-5 hold nameSize-1 (15 means a varint with +// nameSize-15 follows), bit 6 is the nullable flag. Type info is the +// type id byte, plus precision+scale for DECIMAL, a recursive element +// field for LIST, recursive key and value fields for MAP, and a field +// count plus recursive fields for STRUCT. +const ( + schemaVersion = 1 + fieldNameSizeThreshold = 15 +) + +// Header bits 0-1 store the INDEX into this table (matching Java +// SchemaEncoder.FIELD_NAME_ENCODINGS order), not the meta.Encoding +// wire values, which differ. +var fieldNameEncodings = [3]meta.Encoding{ + meta.UTF_8, + meta.ALL_TO_LOWER_SPECIAL, + meta.LOWER_UPPER_DIGIT_SPECIAL, +} + +// Stateless after construction, safe for concurrent use. +var ( + fieldNameEncoder = meta.NewEncoder('$', '_') + fieldNameDecoder = meta.NewDecoder('$', '_') +) + +func encodingIndexOf(encoding meta.Encoding) int { + for i, e := range fieldNameEncodings { + if e == encoding { + return i + } + } + return -1 +} + +// SchemaToBytes serializes a schema to the cross-language wire format. +func SchemaToBytes(s *Schema) ([]byte, error) { + buf := fory.NewByteBuffer(nil) + if err := SchemaToBuffer(s, buf); err != nil { + return nil, err + } + return buf.GetByteSlice(0, buf.WriterIndex()), nil +} + +// SchemaToBuffer serializes a schema into an existing buffer. +func SchemaToBuffer(s *Schema, buf *fory.ByteBuffer) error { + buf.WriteByte_(schemaVersion) + buf.WriteVarUint32Small7(uint32(s.NumFields())) + for _, f := range s.Fields() { + if err := writeSchemaField(buf, f); err != nil { + return err + } + } + return nil +} + +func writeSchemaField(buf *fory.ByteBuffer, f Field) error { + if f.Name == "" { + return fmt.Errorf("row: schema field with empty name") + } + encoding := fieldNameEncoder.ComputeEncodingWith(f.Name, fieldNameEncodings[:]) + metaString, err := fieldNameEncoder.EncodeWithEncoding(f.Name, encoding) + if err != nil { + return fmt.Errorf("row: encoding field name %q: %w", f.Name, err) + } + nameBytes := metaString.GetEncodedBytes() + nameSize := len(nameBytes) + encodingIndex := encodingIndexOf(metaString.GetEncoding()) + if encodingIndex < 0 { + return fmt.Errorf("row: field name %q got unsupported encoding %d", f.Name, metaString.GetEncoding()) + } + + header := encodingIndex & 0x03 + bigSize := nameSize > fieldNameSizeThreshold + if bigSize { + header |= fieldNameSizeThreshold << 2 + } else { + header |= (nameSize - 1) << 2 + } + if f.Nullable { + header |= 0x40 + } + buf.WriteByte_(byte(header)) + if bigSize { + buf.WriteVarUint32Small7(uint32(nameSize - fieldNameSizeThreshold)) + } + buf.WriteBinary(nameBytes) + return writeSchemaType(buf, f.Type) +} + +func writeSchemaType(buf *fory.ByteBuffer, dataType DataType) error { + buf.WriteByte_(byte(dataType.TypeID())) + switch t := dataType.(type) { + case DecimalType: + buf.WriteByte_(byte(t.Precision)) + buf.WriteByte_(byte(t.Scale)) + case *ListType: + return writeSchemaField(buf, t.Elem) + case *MapType: + if err := writeSchemaField(buf, t.Key); err != nil { + return err + } + return writeSchemaField(buf, t.Value) + case *StructType: + buf.WriteVarUint32Small7(uint32(len(t.Fields))) + for _, f := range t.Fields { + if err := writeSchemaField(buf, f); err != nil { + return err + } + } + } + return nil +} + +// SchemaFromBytes deserializes a schema from the cross-language wire +// format. It is safe on untrusted input: declared sizes are checked +// against remaining bytes before any allocation. +func SchemaFromBytes(data []byte) (*Schema, error) { + r := &schemaReader{buf: fory.NewByteBuffer(data), size: len(data)} + version := r.buf.ReadUint8(&r.err) + if err := r.err.CheckError(); err != nil { + return nil, err + } + if version != schemaVersion { + return nil, fmt.Errorf("row: unsupported schema version %d, expected %d", version, schemaVersion) + } + numFields := int(r.buf.ReadVarUint32Small7(&r.err)) + if err := r.err.CheckError(); err != nil { + return nil, err + } + // Every field costs at least one byte, so a declared count larger + // than the remaining input is corrupt; checking before the + // allocation below keeps attacker-declared counts harmless. + if numFields > r.remaining() { + return nil, fmt.Errorf("row: schema declares %d fields but only %d bytes remain", numFields, r.remaining()) + } + fields := make([]Field, 0, numFields) + for i := 0; i < numFields; i++ { + f, err := r.readField() + if err != nil { + return nil, err + } + fields = append(fields, f) + } + return NewSchema(fields), nil +} + +type schemaReader struct { + buf *fory.ByteBuffer + size int + err fory.Error +} + +func (r *schemaReader) remaining() int { return r.size - r.buf.ReaderIndex() } + +func (r *schemaReader) readField() (Field, error) { + header := int(r.buf.ReadUint8(&r.err)) + if err := r.err.CheckError(); err != nil { + return Field{}, err + } + encodingIndex := header & 0x03 + if encodingIndex >= len(fieldNameEncodings) { + return Field{}, fmt.Errorf("row: invalid field name encoding index %d", encodingIndex) + } + nameSizeMinus1 := (header >> 2) & 0x0F + nullable := header&0x40 != 0 + + var nameSize int + if nameSizeMinus1 == fieldNameSizeThreshold { + nameSize = int(r.buf.ReadVarUint32Small7(&r.err)) + fieldNameSizeThreshold + } else { + nameSize = nameSizeMinus1 + 1 + } + if err := r.err.CheckError(); err != nil { + return Field{}, err + } + if nameSize > r.remaining() { + return Field{}, fmt.Errorf("row: field name of %d bytes exceeds %d remaining", nameSize, r.remaining()) + } + nameBytes := r.buf.ReadBinary(nameSize, &r.err) + if err := r.err.CheckError(); err != nil { + return Field{}, err + } + name, err := fieldNameDecoder.Decode(nameBytes, fieldNameEncodings[encodingIndex]) + if err != nil { + return Field{}, fmt.Errorf("row: decoding field name: %w", err) + } + dataType, err := r.readType() + if err != nil { + return Field{}, err + } + return Field{Name: name, Type: dataType, Nullable: nullable}, nil +} + +func (r *schemaReader) readType() (DataType, error) { + typeID := fory.TypeId(r.buf.ReadUint8(&r.err)) + if err := r.err.CheckError(); err != nil { + return nil, err + } + switch typeID { + case fory.BOOL: + return BoolType{}, nil + case fory.INT8: + return Int8Type{}, nil + case fory.INT16: + return Int16Type{}, nil + case fory.INT32: + return Int32Type{}, nil + case fory.INT64: + return Int64Type{}, nil + case fory.FLOAT16: + return Float16Type{}, nil + case fory.FLOAT32: + return Float32Type{}, nil + case fory.FLOAT64: + return Float64Type{}, nil + case fory.STRING: + return StringType{}, nil + case fory.BINARY: + return BinaryType{}, nil + case fory.DURATION: + return DurationType{}, nil + case fory.TIMESTAMP: + return TimestampType{}, nil + case fory.DATE: + return Date32Type{}, nil + case fory.DECIMAL: + precision := int(r.buf.ReadUint8(&r.err)) + scale := int(r.buf.ReadUint8(&r.err)) + if err := r.err.CheckError(); err != nil { + return nil, err + } + return DecimalType{Precision: precision, Scale: scale}, nil + case fory.LIST: + elem, err := r.readField() + if err != nil { + return nil, err + } + return &ListType{Elem: elem}, nil + case fory.MAP: + keyField, err := r.readField() + if err != nil { + return nil, err + } + valueField, err := r.readField() + if err != nil { + return nil, err + } + // Java rebuilds maps from the child types only, restoring the + // canonical key/value names and nullability. + return Map(keyField.Type, valueField.Type), nil + case fory.STRUCT: + numFields := int(r.buf.ReadVarUint32Small7(&r.err)) + if err := r.err.CheckError(); err != nil { + return nil, err + } + if numFields > r.remaining() { + return nil, fmt.Errorf("row: struct declares %d fields but only %d bytes remain", numFields, r.remaining()) + } + fields := make([]Field, 0, numFields) + for i := 0; i < numFields; i++ { + f, err := r.readField() + if err != nil { + return nil, err + } + fields = append(fields, f) + } + return &StructType{Fields: fields}, nil + default: + return nil, fmt.Errorf("row: unknown type id %d in schema", typeID) + } +} + +// ComputeSchemaHash computes the cross-language schema hash, matching +// Java DataTypes.computeSchemaHash and the Python equivalent: fold +// hash*31 + typeId over fields, recursing into list elements, map keys +// and values, and struct fields; on signed-64 overflow, arithmetic- +// shift the hash right by 2 and retry. +func ComputeSchemaHash(s *Schema) int64 { + hash := int64(17) + for _, f := range s.Fields() { + hash = computeFieldHash(hash, f) + } + return hash +} + +func computeFieldHash(hash int64, f Field) int64 { + typeID := int64(f.Type.TypeID()) + for { + if hash <= math.MaxInt64/31 && hash >= math.MinInt64/31 { + product := hash * 31 + if product <= math.MaxInt64-typeID { + hash = product + typeID + break + } + } + hash >>= 2 + } + switch t := f.Type.(type) { + case *ListType: + hash = computeFieldHash(hash, t.Elem) + case *MapType: + hash = computeFieldHash(hash, t.Key) + hash = computeFieldHash(hash, t.Value) + case *StructType: + for _, child := range t.Fields { + hash = computeFieldHash(hash, child) + } + } + return hash +} diff --git a/go/fory/row/schema_bytes_test.go b/go/fory/row/schema_bytes_test.go new file mode 100644 index 0000000000..3c790f5128 --- /dev/null +++ b/go/fory/row/schema_bytes_test.go @@ -0,0 +1,161 @@ +// 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 row + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +// Golden bytes and hashes generated by Java +// org.apache.fory.format.type.SchemaEncoder (SCHEMA_VERSION=1) and +// DataTypes.computeSchemaHash. Regenerate with those classes if the +// Java wire format ever changes; the Java<->Go cross-language test is +// the live guard. +var schemaGoldens = []struct { + name string + schema func() *Schema + bytes []byte + hash int64 +}{ + { + name: "simple", + schema: func() *Schema { + return NewSchema([]Field{ + NewField("id", Int32Type{}, true), + NewField("name", StringType{}, true), + NewField("score", Float64Type{}, true), + NewField("active", BoolType{}, true), + }) + }, + bytes: []byte{ + 0x01, 0x04, 0x45, 0xa0, 0x60, 0x04, 0x49, 0x34, 0x0c, 0x20, 0x15, 0x4d, + 0xc8, 0x4e, 0x89, 0x00, 0x14, 0x4d, 0x00, 0x53, 0x45, 0x48, 0x01, + }, + hash: 15839823, + }, + { + name: "nested", + schema: func() *Schema { + return NewSchema([]Field{ + NewField("person", Struct([]Field{ + NewField("name", StringType{}, true), + NewField("age", Int32Type{}, false), + }), true), + NewField("tags", List(StringType{}), true), + NewField("attrs", Map(StringType{}, Int64Type{}), true), + }) + }, + bytes: []byte{ + 0x01, 0x03, 0x4d, 0x3c, 0x91, 0x93, 0x9a, 0x1b, 0x02, 0x49, 0x34, 0x0c, + 0x20, 0x15, 0x05, 0x00, 0xc4, 0x04, 0x49, 0x4c, 0x06, 0x90, 0x16, 0x49, + 0x22, 0x64, 0x60, 0x15, 0x4d, 0x82, 0x73, 0x8c, 0x80, 0x18, 0x05, 0x28, + 0x98, 0x15, 0x4d, 0xd4, 0x0b, 0xa1, 0x00, 0x06, + }, + hash: 15260761278193, + }, + { + name: "special", + schema: func() *Schema { + return NewSchema([]Field{ + NewField("a_very_long_field_name_exceeding_limit", Int64Type{}, false), + NewField("field2", StringType{}, true), + NewField("$special_name", Date32Type{}, true), + NewField("CamelCase", TimestampType{}, false), + NewField("price", DecimalType{Precision: 10, Scale: 2}, true), + }) + }, + bytes: []byte{ + 0x01, 0x05, 0x3d, 0x09, 0x03, 0x75, 0x24, 0x71, 0xb5, 0xb9, 0xa6, 0xd9, + 0x50, 0x45, 0x8f, 0x6d, 0x03, 0x09, 0xb2, 0x5c, 0x44, 0x20, 0xd0, 0xd3, + 0x6d, 0x68, 0x62, 0x26, 0x06, 0x52, 0x0a, 0x40, 0x85, 0x87, 0xb0, 0x15, + 0x61, 0xf2, 0x4f, 0x20, 0x90, 0x05, 0xed, 0xa0, 0x61, 0x00, 0x27, 0x1a, + 0x38, 0x01, 0x82, 0x16, 0xe0, 0x09, 0x08, 0x26, 0x4d, 0xbe, 0x28, 0x11, + 0x00, 0x28, 0x0a, 0x02, + }, + hash: 492901001, + }, +} + +func TestSchemaBytesGolden(t *testing.T) { + for _, g := range schemaGoldens { + schema := g.schema() + + got, err := SchemaToBytes(schema) + require.NoError(t, err, g.name) + require.Equal(t, g.bytes, got, "%s: bytes must match Java SchemaEncoder output", g.name) + + parsed, err := SchemaFromBytes(g.bytes) + require.NoError(t, err, g.name) + require.True(t, parsed.Equal(schema), "%s: parsed schema %v != %v", g.name, parsed, schema) + } +} + +func TestSchemaHashGolden(t *testing.T) { + for _, g := range schemaGoldens { + require.Equal(t, g.hash, ComputeSchemaHash(g.schema()), g.name) + } + + // 20 int64 fields overflow the int64 hash, exercising Java's + // shift-right-2-and-retry path. + fields := make([]Field, 20) + for i := range fields { + fields[i] = NewField(fmt.Sprintf("f%d", i), Int64Type{}, true) + } + require.Equal(t, int64(2627256684647412568), ComputeSchemaHash(NewSchema(fields))) +} + +func requireErrorContains(t *testing.T, err error, substr string) { + t.Helper() + require.Error(t, err) + require.Contains(t, err.Error(), substr) +} + +func TestSchemaFromBytesErrors(t *testing.T) { + // Unsupported version. + _, err := SchemaFromBytes([]byte{0x02, 0x00}) + requireErrorContains(t, err, "schema version") + + // Invalid field name encoding index (bits 0-1 == 3). + _, err = SchemaFromBytes([]byte{0x01, 0x01, 0x07}) + requireErrorContains(t, err, "encoding index") + + // Unknown type id. + corrupt := append([]byte(nil), schemaGoldens[0].bytes...) + corrupt[len(corrupt)-1] = 0x63 + _, err = SchemaFromBytes(corrupt) + requireErrorContains(t, err, "unknown type id") + + // Field count larger than the remaining input. + _, err = SchemaFromBytes([]byte{0x01, 0x7f}) + requireErrorContains(t, err, "fields") + + // Every strict prefix of a valid encoding must fail, never panic. + golden := schemaGoldens[1].bytes + for i := 0; i < len(golden); i++ { + _, err := SchemaFromBytes(golden[:i]) + require.Error(t, err, "prefix of length %d", i) + } +} + +func TestSchemaToBytesRejectsEmptyName(t *testing.T) { + _, err := SchemaToBytes(NewSchema([]Field{NewField("", Int32Type{}, true)})) + requireErrorContains(t, err, "empty name") +} From cc915029c429399ded63164687b29ae311a9251b Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:03:50 +0530 Subject: [PATCH 08/15] fix: reject untrusted data payloads --- go/fory/row/row.go | 33 +++++++++++++++++++++++++++++---- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/go/fory/row/row.go b/go/fory/row/row.go index e2780b95c9..1fedf6ad38 100644 --- a/go/fory/row/row.go +++ b/go/fory/row/row.go @@ -135,7 +135,7 @@ func (r *Row) varData(i int) []byte { return nil } offset, size := decodeOffsetAndSize(binary.LittleEndian.Uint64(r.data[r.slot(i):])) - return r.data[offset : offset+size] + return boundedSlice(r.data, offset, size) } // Binary returns a zero-copy view of field i's bytes. @@ -197,6 +197,17 @@ func NewArrayData(elem Field, data []byte) *ArrayData { func (a *ArrayData) NumElements() int { return a.numElements } func (a *ArrayData) SizeBytes() int { return len(a.data) } +// validateBounds rejects arrays whose declared element count is not +// covered by the available bytes, so decoders can check before +// allocating from an attacker-declared count. +func (a *ArrayData) validateBounds() error { + need := int64(a.headerBytes) + int64(a.numElements)*int64(a.elemSize) + if a.numElements < 0 || need > int64(len(a.data)) { + return fmt.Errorf("row: array declares %d elements but holds only %d bytes", a.numElements, len(a.data)) + } + return nil +} + func (a *ArrayData) IsNullAt(i int) bool { if uint(i) >= uint(a.numElements) { panic(fmt.Sprintf("row: array index %d out of range [0, %d)", i, a.numElements)) @@ -275,7 +286,7 @@ func (a *ArrayData) varData(i int) []byte { return nil } offset, size := decodeOffsetAndSize(binary.LittleEndian.Uint64(a.data[a.slot(i):])) - return a.data[offset : offset+size] + return boundedSlice(a.data, offset, size) } // Binary returns a zero-copy view of element i's bytes. @@ -316,10 +327,24 @@ type MapData struct { func NewMapData(mapType *MapType, data []byte) *MapData { keysSize := int(binary.LittleEndian.Uint64(data)) + keysData := boundedSlice(data, 8, keysSize) + valuesData := boundedSlice(data, 8+keysSize, len(data)-8-keysSize) return &MapData{ - keys: NewArrayData(mapType.Key, data[8:8+keysSize]), - values: NewArrayData(mapType.Value, data[8+keysSize:]), + keys: NewArrayData(mapType.Key, keysData), + values: NewArrayData(mapType.Value, valuesData), + } +} + +// boundedSlice checks declared bounds against the data LENGTH +// and caps the view so nested +// readers cannot reach outside it either. Out-of-bounds panics are +// converted to errors by the encoder's decode entry points. +func boundedSlice(data []byte, offset, size int) []byte { + end := offset + size + if offset < 0 || size < 0 || end > len(data) { + panic(fmt.Sprintf("row: value bytes [%d:%d] exceed the enclosing %d-byte region", offset, end, len(data))) } + return data[offset:end:end] } func (m *MapData) NumElements() int { return m.keys.numElements } From c3a57d809eaed820a2b91a7177f8d41b6c329717 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:05:20 +0530 Subject: [PATCH 09/15] feat: add InferSchema validating struct against the schema --- go/fory/row/infer.go | 226 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 go/fory/row/infer.go diff --git a/go/fory/row/infer.go b/go/fory/row/infer.go new file mode 100644 index 0000000000..a2daad1cb6 --- /dev/null +++ b/go/fory/row/infer.go @@ -0,0 +1,226 @@ +// 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 row + +import ( + "fmt" + "reflect" + "sort" + "strings" + "time" + "unicode" + "unicode/utf8" + + fory "github.com/apache/fory/go/fory" +) + +var ( + goDateType = reflect.TypeOf(fory.Date{}) + goTimeType = reflect.TypeOf(time.Time{}) + goDurationType = reflect.TypeOf(time.Duration(0)) +) + +// InferSchema infers the row schema for a struct type or pointer to +// struct, including every exported field not tagged `fory:"ignore"`. +// +// Fields are sorted by their lowerCamel name and named by its +// snake_case form (UserName -> user_name), matching Java's schema +// inference so both languages derive identical schemas. +// +// Type mapping: +// - bool, int8/16/32/64, float32/64: same-width row types; int maps to int64 +// - string, []byte: string and binary, nullable +// - slices, maps, nested structs: list, map, and struct, nullable +// - *T: the row type of T, nullable +// - fory.Date, time.Time, time.Duration: date32, timestamp, duration +// +// Unsigned integers, fixed-size arrays, nested pointers, and pointer +// map keys are unsupported and return an error. +func InferSchema(t reflect.Type) (*Schema, error) { + layout, err := inferStructLayout(t, nil) + if err != nil { + return nil, err + } + return layout.schema, nil +} + +// structLayout maps schema ordinals back to Go struct field indexes. +type structLayout struct { + schema *Schema + indexes []int // schema ordinal -> reflect struct field index +} + +func inferStructLayout(t reflect.Type, path []reflect.Type) (*structLayout, error) { + if t.Kind() == reflect.Ptr { + t = t.Elem() + } + if t.Kind() != reflect.Struct || t == goDateType || t == goTimeType { + return nil, fmt.Errorf("row: schema inference expects a struct type, got %v", t) + } + for _, seen := range path { + if seen == t { + return nil, fmt.Errorf("row: circular reference through type %v", t) + } + } + path = append(path, t) + + type member struct { + lowerCamel string + goIndex int + fieldType reflect.Type + } + var members []member + for i := 0; i < t.NumField(); i++ { + f := t.Field(i) + if f.PkgPath != "" || hasIgnoreTag(f.Tag.Get("fory")) { + continue + } + members = append(members, member{lowerFirst(f.Name), i, f.Type}) + } + // Java sorts by the lowerCamel member name before converting names + // to snake_case; matching that order is required for schema and + // row-layout compatibility. + sort.Slice(members, func(a, b int) bool { return members[a].lowerCamel < members[b].lowerCamel }) + + layout := &structLayout{indexes: make([]int, 0, len(members))} + fields := make([]Field, 0, len(members)) + for _, m := range members { + f, err := inferField(lowerCamelToLowerUnderscore(m.lowerCamel), m.fieldType, path) + if err != nil { + return nil, fmt.Errorf("%w (field %s of %v)", err, t.Field(m.goIndex).Name, t) + } + fields = append(fields, f) + layout.indexes = append(layout.indexes, m.goIndex) + } + layout.schema = NewSchema(fields) + return layout, nil +} + +func inferField(name string, t reflect.Type, path []reflect.Type) (Field, error) { + if t.Kind() == reflect.Ptr { + if t.Elem().Kind() == reflect.Ptr { + return Field{}, fmt.Errorf("row: nested pointer type %v is unsupported", t) + } + inner, err := inferField(name, t.Elem(), path) + if err != nil { + return Field{}, err + } + inner.Nullable = true + return inner, nil + } + switch t { + case goDateType: + return Field{Name: name, Type: Date32Type{}}, nil + case goTimeType: + return Field{Name: name, Type: TimestampType{}}, nil + case goDurationType: + return Field{Name: name, Type: DurationType{}}, nil + } + switch t.Kind() { + case reflect.Bool: + return Field{Name: name, Type: BoolType{}}, nil + case reflect.Int8: + return Field{Name: name, Type: Int8Type{}}, nil + case reflect.Int16: + return Field{Name: name, Type: Int16Type{}}, nil + case reflect.Int32: + return Field{Name: name, Type: Int32Type{}}, nil + case reflect.Int64, reflect.Int: + // `int` maps to int64 so the width never depends on the platform. + return Field{Name: name, Type: Int64Type{}}, nil + case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr: + return Field{}, fmt.Errorf("row: unsigned type %v is unsupported, use a signed type", t) + case reflect.Float32: + return Field{Name: name, Type: Float32Type{}}, nil + case reflect.Float64: + return Field{Name: name, Type: Float64Type{}}, nil + case reflect.String: + return Field{Name: name, Type: StringType{}, Nullable: true}, nil + case reflect.Slice: + if t.Elem().Kind() == reflect.Uint8 { + return Field{Name: name, Type: BinaryType{}, Nullable: true}, nil + } + elem, err := inferField(listItemName, t.Elem(), path) + if err != nil { + return Field{}, err + } + return Field{Name: name, Type: &ListType{Elem: elem}, Nullable: true}, nil + case reflect.Map: + if t.Key().Kind() == reflect.Ptr { + return Field{}, fmt.Errorf("row: pointer map key type %v is unsupported", t) + } + key, err := inferField(mapKeyName, t.Key(), path) + if err != nil { + return Field{}, err + } + key.Nullable = false + value, err := inferField(mapValueName, t.Elem(), path) + if err != nil { + return Field{}, err + } + return Field{Name: name, Type: &MapType{Key: key, Value: value}, Nullable: true}, nil + case reflect.Struct: + layout, err := inferStructLayout(t, path) + if err != nil { + return Field{}, err + } + return Field{Name: name, Type: &StructType{Fields: layout.schema.Fields()}, Nullable: true}, nil + default: + return Field{}, fmt.Errorf("row: type %v is unsupported in row format", t) + } +} + +func hasIgnoreTag(tag string) bool { + for _, part := range strings.Split(tag, ",") { + if strings.TrimSpace(part) == "ignore" { + return true + } + } + return false +} + +func lowerFirst(s string) string { + r, size := utf8.DecodeRuneInString(s) + if !unicode.IsUpper(r) { + return s + } + return string(unicode.ToLower(r)) + s[size:] +} + +// lowerCamelToLowerUnderscore ports Java StringUtils: every uppercase +// letter becomes '_' plus its lowercase, so userID becomes user_i_d. +func lowerCamelToLowerUnderscore(s string) string { + var b strings.Builder + from := 0 + for i := 0; i < len(s); i++ { + c := s[i] + if c >= 'A' && c <= 'Z' { + b.WriteString(s[from:i]) + b.WriteByte('_') + b.WriteByte(c + ('a' - 'A')) + from = i + 1 + } + } + if from == 0 { + return s + } + if from < len(s) { + b.WriteString(s[from:]) + } + return b.String() +} From f674b0bdf63657599d333519c23a5da3101f448f Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:07:03 +0530 Subject: [PATCH 10/15] feat: encode go structs to fory row format --- go/fory/row/encoder.go | 494 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 494 insertions(+) create mode 100644 go/fory/row/encoder.go diff --git a/go/fory/row/encoder.go b/go/fory/row/encoder.go new file mode 100644 index 0000000000..fd5703ae49 --- /dev/null +++ b/go/fory/row/encoder.go @@ -0,0 +1,494 @@ +// 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 row + +import ( + "encoding/binary" + "fmt" + "reflect" + "time" + + fory "github.com/apache/fory/go/fory" +) + +// Encoder converts values of struct type T to and from row bytes using +// a codec tree compiled once with reflection at construction time. The +// schema is inferred per InferSchema. +// +// An Encoder owns a reusable write buffer and is NOT goroutine-safe; +// create one Encoder per goroutine. Decoding is safe on corrupt or +// untrusted input. +type Encoder[T any] struct { + codec *structCodec + hash int64 + buf *fory.ByteBuffer +} + +func NewEncoder[T any]() (*Encoder[T], error) { + t := reflect.TypeOf((*T)(nil)).Elem() + if t.Kind() != reflect.Struct { + return nil, fmt.Errorf("row: Encoder requires a struct type, got %v", t) + } + buf := fory.NewByteBuffer(nil) + codec, err := newStructCodec(t, buf) + if err != nil { + return nil, err + } + return &Encoder[T]{codec: codec, hash: ComputeSchemaHash(codec.schema), buf: buf}, nil +} + +func (e *Encoder[T]) Schema() *Schema { return e.codec.schema } + +// SchemaHash returns the cross-language schema hash used by Encode and +// Decode to detect out-of-sync struct definitions. +func (e *Encoder[T]) SchemaHash() int64 { return e.hash } + +// ToRow serializes v to row bytes (no framing, just the row). +func (e *Encoder[T]) ToRow(v *T) ([]byte, error) { + e.buf.SetWriterIndex(0) + if err := e.codec.writeStruct(reflect.ValueOf(v).Elem()); err != nil { + return nil, err + } + out := make([]byte, e.buf.WriterIndex()) + copy(out, e.buf.GetData()[:e.buf.WriterIndex()]) + return out, nil +} + +// FromRow deserializes row bytes produced with this schema. +func (e *Encoder[T]) FromRow(data []byte) (T, error) { + var out T + err := e.FromRowInto(data, &out) + return out, err +} + +func (e *Encoder[T]) FromRowInto(data []byte, out *T) (err error) { + defer func() { + if p := recover(); p != nil { + err = fmt.Errorf("row: corrupt row data: %v", p) + } + }() + return e.codec.readStruct(NewRow(e.codec.schema, data), reflect.ValueOf(out).Elem()) +} + +// Encode frames the row for cross-language exchange: an int64 +// little-endian schema hash followed by the row bytes, matching the +// Java and Python row encoders. +func (e *Encoder[T]) Encode(v *T) ([]byte, error) { + rowBytes, err := e.ToRow(v) + if err != nil { + return nil, err + } + out := make([]byte, 8+len(rowBytes)) + binary.LittleEndian.PutUint64(out, uint64(e.hash)) + copy(out[8:], rowBytes) + return out, nil +} + +// Decode verifies the schema hash and deserializes the row. +func (e *Encoder[T]) Decode(data []byte) (T, error) { + var out T + if len(data) < 8 { + return out, fmt.Errorf("row: encoded data of %d bytes is too short for the schema hash", len(data)) + } + hash := int64(binary.LittleEndian.Uint64(data)) + if hash != e.hash { + return out, fmt.Errorf("row: schema hash mismatch: data has %d, encoder expects %d; writer and reader struct definitions are out of sync", hash, e.hash) + } + err := e.FromRowInto(data[8:], &out) + return out, err +} + +// slotWriter is the shared write surface of RowWriter and ArrayWriter, +// letting one value codec serve struct fields and array elements. +type slotWriter interface { + SetNullAt(i int) + WriteBool(i int, v bool) + WriteInt8(i int, v int8) + WriteInt16(i int, v int16) + WriteInt32(i int, v int32) + WriteInt64(i int, v int64) + WriteFloat32(i int, v float32) + WriteFloat64(i int, v float64) + WriteDate(i int, d fory.Date) error + WriteTimestamp(i int, t time.Time) + WriteDuration(i int, d time.Duration) + WriteString(i int, s string) + WriteBytes(i int, b []byte) + SetOffsetAndSize(i, absStart, size int) +} + +// valueReader is the shared read surface of Row and ArrayData. +type valueReader interface { + IsNullAt(i int) bool + Bool(i int) bool + Int8(i int) int8 + Int16(i int) int16 + Int32(i int) int32 + Int64(i int) int64 + Float32(i int) float32 + Float64(i int) float64 + Date(i int) fory.Date + Timestamp(i int) time.Time + Duration(i int) time.Duration + String(i int) string + Binary(i int) []byte + Struct(i int) *Row + Array(i int) *ArrayData + Map(i int) *MapData +} + +// valueCodec reads or writes one value at slot/element i. Write +// functions handle nil for nilable Go types; read functions overwrite +// the target even when the field is null, so reused targets stay clean. +type valueCodec struct { + write func(w slotWriter, i int, v reflect.Value) error + read func(g valueReader, i int, v reflect.Value) error +} + +// structCodec writes and reads one struct level; nested structs hold +// their own child codec with a child RowWriter sharing the buffer. +type structCodec struct { + schema *Schema + writer *RowWriter + fields []fieldCodec +} + +type fieldCodec struct { + goIndex int + codec valueCodec +} + +func newStructCodec(t reflect.Type, buf *fory.ByteBuffer) (*structCodec, error) { + layout, err := inferStructLayout(t, nil) + if err != nil { + return nil, err + } + sc := &structCodec{ + schema: layout.schema, + writer: NewRowWriterWithBuffer(layout.schema, buf), + fields: make([]fieldCodec, 0, layout.schema.NumFields()), + } + for ordinal, goIndex := range layout.indexes { + codec, err := newValueCodec(t.Field(goIndex).Type, layout.schema.Field(ordinal).Type, buf) + if err != nil { + return nil, err + } + sc.fields = append(sc.fields, fieldCodec{goIndex: goIndex, codec: codec}) + } + return sc, nil +} + +func (sc *structCodec) writeStruct(v reflect.Value) error { + sc.writer.Reset() + for ordinal := range sc.fields { + fc := &sc.fields[ordinal] + if err := fc.codec.write(sc.writer, ordinal, v.Field(fc.goIndex)); err != nil { + return err + } + } + return nil +} + +func (sc *structCodec) readStruct(r *Row, v reflect.Value) error { + for ordinal := range sc.fields { + fc := &sc.fields[ordinal] + if err := fc.codec.read(r, ordinal, v.Field(fc.goIndex)); err != nil { + return err + } + } + return nil +} + +func newValueCodec(goType reflect.Type, dataType DataType, buf *fory.ByteBuffer) (valueCodec, error) { + if goType.Kind() == reflect.Ptr { + elemType := goType.Elem() + inner, err := newValueCodec(elemType, dataType, buf) + if err != nil { + return valueCodec{}, err + } + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + if v.IsNil() { + w.SetNullAt(i) + return nil + } + return inner.write(w, i, v.Elem()) + }, + read: func(g valueReader, i int, v reflect.Value) error { + if g.IsNullAt(i) { + v.Set(reflect.Zero(goType)) + return nil + } + p := reflect.New(elemType) + if err := inner.read(g, i, p.Elem()); err != nil { + return err + } + v.Set(p) + return nil + }, + }, nil + } + + switch dt := dataType.(type) { + case BoolType: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteBool(i, v.Bool()); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetBool(g.Bool(i)); return nil }, + }, nil + case Int8Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteInt8(i, int8(v.Int())); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetInt(int64(g.Int8(i))); return nil }, + }, nil + case Int16Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteInt16(i, int16(v.Int())); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetInt(int64(g.Int16(i))); return nil }, + }, nil + case Int32Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteInt32(i, int32(v.Int())); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetInt(int64(g.Int32(i))); return nil }, + }, nil + case Int64Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteInt64(i, v.Int()); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetInt(g.Int64(i)); return nil }, + }, nil + case Float32Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteFloat32(i, float32(v.Float())); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetFloat(float64(g.Float32(i))); return nil }, + }, nil + case Float64Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteFloat64(i, v.Float()); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetFloat(g.Float64(i)); return nil }, + }, nil + case Date32Type: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { return w.WriteDate(i, v.Interface().(fory.Date)) }, + read: func(g valueReader, i int, v reflect.Value) error { v.Set(reflect.ValueOf(g.Date(i))); return nil }, + }, nil + case TimestampType: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + w.WriteTimestamp(i, v.Interface().(time.Time)) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { v.Set(reflect.ValueOf(g.Timestamp(i))); return nil }, + }, nil + case DurationType: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + w.WriteDuration(i, time.Duration(v.Int())) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetInt(int64(g.Duration(i))); return nil }, + }, nil + case StringType: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { w.WriteString(i, v.String()); return nil }, + read: func(g valueReader, i int, v reflect.Value) error { v.SetString(g.String(i)); return nil }, + }, nil + case BinaryType: + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + if v.IsNil() { + w.SetNullAt(i) + return nil + } + w.WriteBytes(i, v.Bytes()) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { + b := g.Binary(i) + if b == nil { + v.SetBytes(nil) + return nil + } + // Copy so the decoded struct never aliases the input. + owned := make([]byte, len(b)) + copy(owned, b) + v.SetBytes(owned) + return nil + }, + }, nil + case *ListType: + return newListCodec(goType, dt, buf) + case *MapType: + return newMapCodec(goType, dt, buf) + case *StructType: + child, err := newStructCodec(goType, buf) + if err != nil { + return valueCodec{}, err + } + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + start := child.writer.buf.WriterIndex() + if err := child.writeStruct(v); err != nil { + return err + } + w.SetOffsetAndSize(i, start, child.writer.buf.WriterIndex()-start) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { + nested := g.Struct(i) + if nested == nil { + v.Set(reflect.Zero(goType)) + return nil + } + return child.readStruct(nested, v) + }, + }, nil + default: + return valueCodec{}, fmt.Errorf("row: data type %s is not supported by the encoder", dataType) + } +} + +func newListCodec(goType reflect.Type, listType *ListType, buf *fory.ByteBuffer) (valueCodec, error) { + elemCodec, err := newValueCodec(goType.Elem(), listType.Elem.Type, buf) + if err != nil { + return valueCodec{}, err + } + writer := NewArrayWriter(listType.Elem, buf) + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + if v.IsNil() { + w.SetNullAt(i) + return nil + } + n := v.Len() + start := buf.WriterIndex() + if err := writer.Reset(n); err != nil { + return err + } + for j := 0; j < n; j++ { + if err := elemCodec.write(writer, j, v.Index(j)); err != nil { + return err + } + } + w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { + arr := g.Array(i) + if arr == nil { + v.Set(reflect.Zero(goType)) + return nil + } + if err := arr.validateBounds(); err != nil { + return err + } + n := arr.NumElements() + s := reflect.MakeSlice(goType, n, n) + for j := 0; j < n; j++ { + if err := elemCodec.read(arr, j, s.Index(j)); err != nil { + return err + } + } + v.Set(s) + return nil + }, + }, nil +} + +func newMapCodec(goType reflect.Type, mapType *MapType, buf *fory.ByteBuffer) (valueCodec, error) { + keyCodec, err := newValueCodec(goType.Key(), mapType.Key.Type, buf) + if err != nil { + return valueCodec{}, err + } + valueCodecImpl, err := newValueCodec(goType.Elem(), mapType.Value.Type, buf) + if err != nil { + return valueCodec{}, err + } + mapWriter := NewMapWriter(buf) + keysWriter := NewArrayWriter(mapType.Key, buf) + valuesWriter := NewArrayWriter(mapType.Value, buf) + return valueCodec{ + write: func(w slotWriter, i int, v reflect.Value) error { + if v.IsNil() { + w.SetNullAt(i) + return nil + } + n := v.Len() + start := buf.WriterIndex() + mapWriter.Reset() + // Go map iteration order changes between iterations, so + // snapshot entries once to keep keys and values aligned. + keys := make([]reflect.Value, 0, n) + values := make([]reflect.Value, 0, n) + iter := v.MapRange() + for iter.Next() { + keys = append(keys, iter.Key()) + values = append(values, iter.Value()) + } + if err := keysWriter.Reset(len(keys)); err != nil { + return err + } + for j, k := range keys { + if err := keyCodec.write(keysWriter, j, k); err != nil { + return err + } + } + mapWriter.FinishKeys() + if err := valuesWriter.Reset(len(values)); err != nil { + return err + } + for j, val := range values { + if err := valueCodecImpl.write(valuesWriter, j, val); err != nil { + return err + } + } + w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) + return nil + }, + read: func(g valueReader, i int, v reflect.Value) error { + md := g.Map(i) + if md == nil { + v.Set(reflect.Zero(goType)) + return nil + } + keys, values := md.Keys(), md.Values() + if err := keys.validateBounds(); err != nil { + return err + } + if err := values.validateBounds(); err != nil { + return err + } + n := keys.NumElements() + if values.NumElements() != n { + return fmt.Errorf("row: map has %d keys but %d values", n, values.NumElements()) + } + m := reflect.MakeMapWithSize(goType, n) + for j := 0; j < n; j++ { + key := reflect.New(goType.Key()).Elem() + if err := keyCodec.read(keys, j, key); err != nil { + return err + } + value := reflect.New(goType.Elem()).Elem() + if err := valueCodecImpl.read(values, j, value); err != nil { + return err + } + m.SetMapIndex(key, value) + } + v.Set(m) + return nil + }, + }, nil +} From 20b31f31ce73d321264278b7bf42c3aa4542e425 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:07:43 +0530 Subject: [PATCH 11/15] feat(test): add bytes <-> schema tests with Java interop --- go/fory/row/encoder_test.go | 177 ++++++++++++++++++++++++++++++++++++ go/fory/row/infer_test.go | 154 +++++++++++++++++++++++++++++++ 2 files changed, 331 insertions(+) create mode 100644 go/fory/row/encoder_test.go create mode 100644 go/fory/row/infer_test.go diff --git a/go/fory/row/encoder_test.go b/go/fory/row/encoder_test.go new file mode 100644 index 0000000000..ec000ef979 --- /dev/null +++ b/go/fory/row/encoder_test.go @@ -0,0 +1,177 @@ +// 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 row + +import ( + "encoding/binary" + "testing" + "time" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +type address struct { + City string + Zip int32 +} + +type person struct { + Id int64 + Name *string + Score float64 + Tags []string + Attrs map[string]int32 + Addr *address + Data []byte + Born fory.Date + At time.Time + Dur time.Duration +} + +func samplePerson() person { + name := "ayush" + return person{ + Id: 42, + Name: &name, + Score: 99.5, + Tags: []string{"go", "fory"}, + Attrs: map[string]int32{"a": 1, "b": 2}, + Addr: &address{City: "Delhi", Zip: 110001}, + Data: []byte{1, 2, 3}, + Born: fory.Date{Year: 2007, Month: time.June, Day: 1}, + At: time.UnixMicro(1_234_567_890_123_456), + Dur: 90 * time.Second, + } +} + +func TestEncoderRoundTrip(t *testing.T) { + enc, err := NewEncoder[person]() + require.NoError(t, err) + + original := samplePerson() + rowBytes, err := enc.ToRow(&original) + require.NoError(t, err) + + decoded, err := enc.FromRow(rowBytes) + require.NoError(t, err) + require.Equal(t, original, decoded) +} + +func TestEncoderNullsAndEmptiness(t *testing.T) { + enc, err := NewEncoder[person]() + require.NoError(t, err) + + // All nilable fields nil: they round-trip as nil, not zero values. + // Temporal fields stay valid; a zero fory.Date is rejected by + // WriteDate just like in the object-graph serializer. + original := samplePerson() + original.Name, original.Tags, original.Attrs, original.Addr, original.Data = nil, nil, nil, nil, nil + rowBytes, err := enc.ToRow(&original) + require.NoError(t, err) + decoded, err := enc.FromRow(rowBytes) + require.NoError(t, err) + require.Equal(t, original, decoded) + require.Nil(t, decoded.Name) + require.Nil(t, decoded.Tags) + require.Nil(t, decoded.Attrs) + + // Empty is distinct from nil. + original.Tags, original.Attrs, original.Data = []string{}, map[string]int32{}, []byte{} + rowBytes, err = enc.ToRow(&original) + require.NoError(t, err) + decoded, err = enc.FromRow(rowBytes) + require.NoError(t, err) + require.NotNil(t, decoded.Tags) + require.Empty(t, decoded.Tags) + require.NotNil(t, decoded.Attrs) + require.NotNil(t, decoded.Data) +} + +func TestEncoderDeepNesting(t *testing.T) { + type inner struct { + Vals []int64 + KV map[string][]int32 + } + type outer struct { + Rows []inner + Ptrs []*int32 + } + enc, err := NewEncoder[outer]() + require.NoError(t, err) + + three := int32(3) + original := outer{ + Rows: []inner{ + {Vals: []int64{1, 2}, KV: map[string][]int32{"x": {7, 8}}}, + {Vals: nil, KV: nil}, + }, + Ptrs: []*int32{&three, nil}, + } + rowBytes, err := enc.ToRow(&original) + require.NoError(t, err) + decoded, err := enc.FromRow(rowBytes) + require.NoError(t, err) + require.Equal(t, original, decoded) + require.Nil(t, decoded.Ptrs[1]) +} + +func TestEncodeDecodeFraming(t *testing.T) { + enc, err := NewEncoder[person]() + require.NoError(t, err) + original := samplePerson() + + encoded, err := enc.Encode(&original) + require.NoError(t, err) + require.Equal(t, uint64(enc.SchemaHash()), binary.LittleEndian.Uint64(encoded)) + require.Equal(t, ComputeSchemaHash(enc.Schema()), enc.SchemaHash()) + + decoded, err := enc.Decode(encoded) + require.NoError(t, err) + require.Equal(t, original, decoded) + + // Tampered hash is rejected with a descriptive error. + tampered := append([]byte(nil), encoded...) + tampered[0] ^= 0xFF + _, err = enc.Decode(tampered) + requireErrorContains(t, err, "hash mismatch") + + _, err = enc.Decode(encoded[:4]) + requireErrorContains(t, err, "too short") +} + +// Corrupt row bytes must produce errors, never panics. +func TestEncoderCorruptRowData(t *testing.T) { + enc, err := NewEncoder[person]() + require.NoError(t, err) + original := samplePerson() + rowBytes, err := enc.ToRow(&original) + require.NoError(t, err) + + for _, cut := range []int{1, 8, len(rowBytes) / 2, len(rowBytes) - 1} { + _, err := enc.FromRow(rowBytes[:cut]) + require.Error(t, err, "truncated to %d bytes", cut) + } +} + +func TestNewEncoderRejectsNonStruct(t *testing.T) { + _, err := NewEncoder[int]() + require.Error(t, err) + _, err = NewEncoder[map[string]int]() + require.Error(t, err) +} diff --git a/go/fory/row/infer_test.go b/go/fory/row/infer_test.go new file mode 100644 index 0000000000..cb7c44510f --- /dev/null +++ b/go/fory/row/infer_test.go @@ -0,0 +1,154 @@ +// 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 row + +import ( + "reflect" + "testing" + "time" + + fory "github.com/apache/fory/go/fory" + "github.com/stretchr/testify/require" +) + +func TestLowerCamelToLowerUnderscore(t *testing.T) { + cases := map[string]string{ + "id": "id", + "userName": "user_name", + "userID": "user_i_d", // Java quirk, kept for parity + "f2": "f2", + "aVeryLong": "a_very_long", + } + for in, want := range cases { + require.Equal(t, want, lowerCamelToLowerUnderscore(in), in) + } +} + +// Fields sort by lowerCamel name before snake_case conversion, +// matching Java's Descriptor ordering. +func TestInferFieldOrder(t *testing.T) { + type ordered struct { + ZItem int64 + AlphaBeta string + M2 int32 + } + s, err := InferSchema(reflect.TypeOf(ordered{})) + require.NoError(t, err) + require.Equal(t, 3, s.NumFields()) + require.Equal(t, "alpha_beta", s.Field(0).Name) + require.Equal(t, "m2", s.Field(1).Name) + require.Equal(t, "z_item", s.Field(2).Name) +} + +func TestInferTypeMapping(t *testing.T) { + type inner struct { + X float64 + } + type sample struct { + B bool + I8 int8 + I16 int16 + I32 int32 + I64 int64 + I int + F32 float32 + F64 float64 + S string + Bin []byte + L []int32 + M map[string]int64 + In inner + PI32 *int32 + PIn *inner + D fory.Date + T time.Time + Dur time.Duration + } + s, err := InferSchema(reflect.TypeOf(sample{})) + require.NoError(t, err) + + byName := func(name string) Field { return s.Field(s.FieldIndex(name)) } + + require.Equal(t, BoolType{}, byName("b").Type) + require.False(t, byName("b").Nullable) + require.Equal(t, Int8Type{}, byName("i8").Type) + require.Equal(t, Int16Type{}, byName("i16").Type) + require.Equal(t, Int32Type{}, byName("i32").Type) + require.Equal(t, Int64Type{}, byName("i64").Type) + require.Equal(t, Int64Type{}, byName("i").Type, "int maps to int64") + require.Equal(t, Float32Type{}, byName("f32").Type) + require.Equal(t, Float64Type{}, byName("f64").Type) + + require.Equal(t, StringType{}, byName("s").Type) + require.True(t, byName("s").Nullable) + require.Equal(t, BinaryType{}, byName("bin").Type) + + list := byName("l").Type.(*ListType) + require.Equal(t, Int32Type{}, list.Elem.Type) + require.False(t, list.Elem.Nullable, "value-type slice elements cannot be null") + require.True(t, byName("l").Nullable) + + m := byName("m").Type.(*MapType) + require.Equal(t, StringType{}, m.Key.Type) + require.False(t, m.Key.Nullable) + require.Equal(t, Int64Type{}, m.Value.Type) + + in := byName("in").Type.(*StructType) + require.Equal(t, "x", in.Fields[0].Name) + require.True(t, byName("in").Nullable) + + require.Equal(t, Int32Type{}, byName("p_i32").Type) + require.True(t, byName("p_i32").Nullable, "pointer fields are nullable") + require.True(t, byName("p_in").Nullable) + + require.Equal(t, Date32Type{}, byName("d").Type) + require.Equal(t, TimestampType{}, byName("t").Type) + require.Equal(t, DurationType{}, byName("dur").Type) +} + +func TestInferSkipsIgnoredAndUnexported(t *testing.T) { + type tagged struct { + A int64 + hidden int64 + Skipped string `fory:"ignore"` + } + s, err := InferSchema(reflect.TypeOf(tagged{})) + require.NoError(t, err) + require.Equal(t, 1, s.NumFields()) + require.Equal(t, "a", s.Field(0).Name) +} + +func TestInferRejectsUnsupportedTypes(t *testing.T) { + type node struct { + Next *node + } + cases := []any{ + struct{ U uint32 }{}, + struct{ C chan int }{}, + struct{ A [3]int32 }{}, + struct{ PP **int32 }{}, + struct{ M map[*string]int32 }{}, + node{}, + } + for _, c := range cases { + _, err := InferSchema(reflect.TypeOf(c)) + require.Error(t, err, "%T", c) + } + _, err := InferSchema(reflect.TypeOf(42)) + require.Error(t, err) +} From 56cfb7c3978de1c83f138add83ea7ac5b7b924e3 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:25:26 +0530 Subject: [PATCH 12/15] feat: add xlang tests to wrap with Java --- go/fory/tests/row_xlang/row_xlang_main.go | 158 ++++++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 go/fory/tests/row_xlang/row_xlang_main.go diff --git a/go/fory/tests/row_xlang/row_xlang_main.go b/go/fory/tests/row_xlang/row_xlang_main.go new file mode 100644 index 0000000000..f8491cdc17 --- /dev/null +++ b/go/fory/tests/row_xlang/row_xlang_main.go @@ -0,0 +1,158 @@ +// 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. + +// Row format cross-language peer driven by Java's +// org.apache.fory.format.GoCrossLanguageTest: each case reads bytes +// written by Java, verifies them field by field, re-encodes the same +// value, and overwrites the file for Java to check. +package main + +import ( + "bytes" + "fmt" + "os" + "reflect" + + "github.com/apache/fory/go/fory/row" +) + +// Mirrors org.apache.fory.format.CrossLanguageTest.A: pointer and +// value types are chosen so the inferred schema matches Java's +// (boxed Java types are nullable, so they map to Go pointers; string +// fields are nullable in both languages). +type a struct { + F1 *int32 + F2 map[string]string +} + +// Mirrors CrossLanguageTest.Bar. +type bar struct { + F1 *int32 + F2 string +} + +// Mirrors CrossLanguageTest.Foo. F3 uses []*string because the Java +// fixture contains a null list element. +type foo struct { + F1 *int32 + F2 string + F3 []*string + F4 map[string]*int32 + F5 *bar +} + +func main() { + if len(os.Args) < 3 { + fail("usage: row_xlang_bin ...") + } + switch caseName := os.Args[1]; caseName { + case "test_map_encoder": + testMapEncoder(os.Args[2]) + case "test_serialization_without_schema": + testSerializationWithoutSchema(os.Args[2]) + case "test_serialization_with_schema": + if len(os.Args) < 4 { + fail("test_serialization_with_schema needs ") + } + testSerializationWithSchema(os.Args[2], os.Args[3]) + default: + fail("unknown test case %q", caseName) + } +} + +func testMapEncoder(dataFile string) { + encoder, err := row.NewEncoder[a]() + must(err) + data, err := os.ReadFile(dataFile) + must(err) + + decoded, err := encoder.Decode(data) + must(err) + expected := a{F1: int32Ptr(1), F2: map[string]string{"pid": "12345", "ip": "0.0.0.0", "k1": "v1"}} + check(reflect.DeepEqual(decoded, expected), "decoded %+v, expected %+v", decoded, expected) + + encoded, err := encoder.Encode(&expected) + must(err) + must(os.WriteFile(dataFile, encoded, 0o644)) +} + +func testSerializationWithoutSchema(dataFile string) { + encoder, err := row.NewEncoder[foo]() + must(err) + data, err := os.ReadFile(dataFile) + must(err) + + decoded, err := encoder.FromRow(data) + must(err) + expected := expectedFoo() + check(reflect.DeepEqual(decoded, expected), "decoded %+v, expected %+v", decoded, expected) + + rowBytes, err := encoder.ToRow(&expected) + must(err) + must(os.WriteFile(dataFile, rowBytes, 0o644)) +} + +func testSerializationWithSchema(schemaFile, dataFile string) { + encoder, err := row.NewEncoder[foo]() + must(err) + schemaBytes, err := os.ReadFile(schemaFile) + must(err) + + parsed, err := row.SchemaFromBytes(schemaBytes) + must(err) + check(parsed.Equal(encoder.Schema()), "Java schema %v, inferred schema %v", parsed, encoder.Schema()) + check(row.ComputeSchemaHash(parsed) == encoder.SchemaHash(), + "schema hash %d, encoder hash %d", row.ComputeSchemaHash(parsed), encoder.SchemaHash()) + reencoded, err := row.SchemaToBytes(encoder.Schema()) + must(err) + check(bytes.Equal(reencoded, schemaBytes), "re-encoded schema bytes differ from Java's") + + testSerializationWithoutSchema(dataFile) +} + +func expectedFoo() foo { + return foo{ + F1: int32Ptr(1), + F2: "str", + F3: []*string{stringPtr("str1"), nil, stringPtr("str2")}, + F4: map[string]*int32{ + "k1": int32Ptr(1), "k2": int32Ptr(2), "k3": int32Ptr(3), + "k4": int32Ptr(4), "k5": int32Ptr(5), "k6": int32Ptr(6), + }, + F5: &bar{F1: int32Ptr(1), F2: "str"}, + } +} + +func int32Ptr(v int32) *int32 { return &v } +func stringPtr(s string) *string { return &s } + +func check(ok bool, format string, args ...any) { + if !ok { + fail(format, args...) + } +} + +func must(err error) { + if err != nil { + fail("%v", err) + } +} + +func fail(format string, args ...any) { + fmt.Fprintf(os.Stderr, "row_xlang: "+format+"\n", args...) + os.Exit(1) +} From f4c064adb6c4a9648517aae47eb2cdbc898acc36 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:26:30 +0530 Subject: [PATCH 13/15] feat: add cross-language tests with java --- .../fory/format/GoCrossLanguageTest.java | 142 ++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100644 java/fory-format/src/test/java/org/apache/fory/format/GoCrossLanguageTest.java diff --git a/java/fory-format/src/test/java/org/apache/fory/format/GoCrossLanguageTest.java b/java/fory-format/src/test/java/org/apache/fory/format/GoCrossLanguageTest.java new file mode 100644 index 0000000000..006a02a977 --- /dev/null +++ b/java/fory-format/src/test/java/org/apache/fory/format/GoCrossLanguageTest.java @@ -0,0 +1,142 @@ +/* + * 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 org.apache.fory.format; + +import com.google.common.collect.ImmutableMap; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.fory.format.encoder.Encoders; +import org.apache.fory.format.encoder.RowEncoder; +import org.apache.fory.format.row.binary.BinaryRow; +import org.apache.fory.format.type.DataTypes; +import org.apache.fory.memory.MemoryBuffer; +import org.apache.fory.memory.MemoryUtils; +import org.apache.fory.test.TestUtils; +import org.testng.Assert; +import org.testng.SkipException; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Row format cross-language tests against a Go peer built from {@code go/fory/tests/row_xlang}. + * Data shapes are shared with {@link CrossLanguageTest} so the Java, Python, and Go peers exercise + * the same schemas. + */ +@Test +public class GoCrossLanguageTest { + private static final boolean IS_WINDOWS = + System.getProperty("os.name").toLowerCase().contains("windows"); + private static final String GO_BINARY = IS_WINDOWS ? "row_xlang_bin.exe" : "row_xlang_bin"; + + @BeforeClass + public void ensureGoPeerReady() { + String enabled = System.getenv("FORY_GO_JAVA_CI"); + if (!"1".equals(enabled)) { + throw new SkipException("Skipping GoCrossLanguageTest: FORY_GO_JAVA_CI not set to 1"); + } + boolean goInstalled = true; + try { + Process process = new ProcessBuilder("go", "version").start(); + if (process.waitFor() != 0) { + goInstalled = false; + } + } catch (IOException | InterruptedException e) { + goInstalled = false; + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + } + if (!goInstalled) { + throw new SkipException("Skipping GoCrossLanguageTest: go not installed"); + } + List buildCommand = + Arrays.asList("go", "build", "-o", "tests/" + GO_BINARY, "./tests/row_xlang"); + boolean buildSuccess = + TestUtils.executeCommand( + buildCommand, 120, Collections.emptyMap(), new File("../../go/fory")); + if (!buildSuccess || !new File("../../go/fory/tests/" + GO_BINARY).exists()) { + throw new SkipException("Skipping GoCrossLanguageTest: failed to build " + GO_BINARY); + } + } + + public void testMapEncoder() throws IOException { + CrossLanguageTest.A a = CrossLanguageTest.A.create(); + RowEncoder encoder = Encoders.bean(CrossLanguageTest.A.class); + Path dataFile = createTempFile("row_go_map"); + Files.write(dataFile, encoder.encode(a)); + Assert.assertTrue(runGoPeer("test_map_encoder", dataFile)); + Assert.assertEquals(encoder.decode(Files.readAllBytes(dataFile)), a); + } + + public void testSerializationWithoutSchema() throws IOException { + CrossLanguageTest.Foo foo = CrossLanguageTest.Foo.create(); + RowEncoder encoder = Encoders.bean(CrossLanguageTest.Foo.class); + Path dataFile = createTempFile("row_go_foo"); + Files.write(dataFile, encoder.toRow(foo).toBytes()); + Assert.assertTrue(runGoPeer("test_serialization_without_schema", dataFile)); + Assert.assertEquals(readFoo(encoder, dataFile), foo); + } + + public void testSerializationWithSchema() throws IOException { + CrossLanguageTest.Foo foo = CrossLanguageTest.Foo.create(); + RowEncoder encoder = Encoders.bean(CrossLanguageTest.Foo.class); + Path dataFile = createTempFile("row_go_foo"); + Path schemaFile = createTempFile("row_go_foo_schema"); + BinaryRow row = encoder.toRow(foo); + Files.write(dataFile, row.toBytes()); + Files.write(schemaFile, DataTypes.serializeSchema(row.getSchema())); + Assert.assertTrue(runGoPeer("test_serialization_with_schema", schemaFile, dataFile)); + Assert.assertEquals(readFoo(encoder, dataFile), foo); + } + + private static CrossLanguageTest.Foo readFoo( + RowEncoder encoder, Path dataFile) throws IOException { + MemoryBuffer buffer = MemoryUtils.wrap(Files.readAllBytes(dataFile)); + BinaryRow row = new BinaryRow(encoder.schema()); + row.pointTo(buffer, 0, buffer.size()); + return encoder.fromRow(row); + } + + private static Path createTempFile(String prefix) throws IOException { + Path file = Files.createTempFile(prefix, "data"); + file.toFile().deleteOnExit(); + return file; + } + + private boolean runGoPeer(String caseName, Path... files) { + List command = new ArrayList<>(); + command.add(IS_WINDOWS ? GO_BINARY : "./" + GO_BINARY); + command.add(caseName); + for (Path file : files) { + command.add(file.toAbsolutePath().toString()); + } + return TestUtils.executeCommand( + command, + 30, + ImmutableMap.of("ENABLE_CROSS_LANGUAGE_TESTS", "true"), + new File("../../go/fory/tests")); + } +} From 62321e9c8622cc4a07fb70f11aa265598cb99d4f Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 19:27:28 +0530 Subject: [PATCH 14/15] feat: wire ci and update gitignore --- .github/workflows/ci.yml | 2 ++ .gitignore | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 450bd1a732..55efa183fb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1607,6 +1607,8 @@ jobs: mvn -T16 --no-transfer-progress clean install -DskipTests -Dmaven.javadoc.skip=true -Dmaven.source.skip=true cd fory-core mvn -T16 --no-transfer-progress test -Dtest=org.apache.fory.xlang.GoXlangTest + cd ../fory-format + mvn -T16 --no-transfer-progress test -Dtest=org.apache.fory.format.GoCrossLanguageTest - name: Run Go IDL Tests run: ./integration_tests/idl_tests/run_go_tests.sh diff --git a/.gitignore b/.gitignore index c543b379a8..013f108177 100644 --- a/.gitignore +++ b/.gitignore @@ -59,6 +59,8 @@ go/fory/tests/xlang_test_main go/fory/tests/xlang/xlang_test_main go/fory/xlang_test_main go/fory/xlang +go/fory/tests/row_xlang_bin +go/fory/tests/row_xlang_bin.exe # Build directories for all languages # Java From 658b794c4bd36c6eb58560ca02c254f666dc3853 Mon Sep 17 00:00:00 2001 From: ayush00git Date: Thu, 30 Jul 2026 20:02:10 +0530 Subject: [PATCH 15/15] fix: trucation of large dt and add ignore fields --- go/fory/row/array_test.go | 18 ++++++++++++++++ go/fory/row/encoder.go | 20 +++++++---------- go/fory/row/infer.go | 36 ++++++++++++++++++++++++++----- go/fory/row/infer_test.go | 14 +++++++++++- go/fory/row/row.go | 11 ++++++++-- go/fory/row/row_test.go | 12 +++++++++++ go/fory/row/schema_bytes.go | 14 +++++++++--- go/fory/row/schema_bytes_test.go | 13 +++++++++++ go/fory/row/writer.go | 37 +++++++++++++++++++++----------- 9 files changed, 139 insertions(+), 36 deletions(-) diff --git a/go/fory/row/array_test.go b/go/fory/row/array_test.go index 9ede434760..8808a1bdb8 100644 --- a/go/fory/row/array_test.go +++ b/go/fory/row/array_test.go @@ -18,6 +18,8 @@ package row import ( + "encoding/binary" + "math" "testing" fory "github.com/apache/fory/go/fory" @@ -106,3 +108,19 @@ func TestArrayResetRejectsInvalidLength(t *testing.T) { require.Error(t, w.Reset(-1)) require.Error(t, w.Reset(1<<40)) } + +// Forged element counts must fail validation even when the arithmetic +// would overflow int64 and wrap the required size negative. +func TestArrayValidateBoundsRejectsForgedCounts(t *testing.T) { + for _, count := range []uint64{ + math.MaxUint64, // negative after int conversion + math.MaxInt64, // bitmap width and product overflow + uint64(math.MaxInt64/8) + 1, // product overflow with 8-byte elements + 1 << 32, // plausible-looking but far beyond the data + } { + data := make([]byte, 16) + binary.LittleEndian.PutUint64(data, count) + a := NewArrayData(List(Int64Type{}).Elem, data) + require.Error(t, a.validateBounds(), "count %d", count) + } +} diff --git a/go/fory/row/encoder.go b/go/fory/row/encoder.go index fd5703ae49..5e0f793a17 100644 --- a/go/fory/row/encoder.go +++ b/go/fory/row/encoder.go @@ -127,9 +127,9 @@ type slotWriter interface { WriteDate(i int, d fory.Date) error WriteTimestamp(i int, t time.Time) WriteDuration(i int, d time.Duration) - WriteString(i int, s string) - WriteBytes(i int, b []byte) - SetOffsetAndSize(i, absStart, size int) + WriteString(i int, s string) error + WriteBytes(i int, b []byte) error + SetOffsetAndSize(i, absStart, size int) error } // valueReader is the shared read surface of Row and ArrayData. @@ -303,7 +303,7 @@ func newValueCodec(goType reflect.Type, dataType DataType, buf *fory.ByteBuffer) }, nil case StringType: return valueCodec{ - write: func(w slotWriter, i int, v reflect.Value) error { w.WriteString(i, v.String()); return nil }, + write: func(w slotWriter, i int, v reflect.Value) error { return w.WriteString(i, v.String()) }, read: func(g valueReader, i int, v reflect.Value) error { v.SetString(g.String(i)); return nil }, }, nil case BinaryType: @@ -313,8 +313,7 @@ func newValueCodec(goType reflect.Type, dataType DataType, buf *fory.ByteBuffer) w.SetNullAt(i) return nil } - w.WriteBytes(i, v.Bytes()) - return nil + return w.WriteBytes(i, v.Bytes()) }, read: func(g valueReader, i int, v reflect.Value) error { b := g.Binary(i) @@ -344,8 +343,7 @@ func newValueCodec(goType reflect.Type, dataType DataType, buf *fory.ByteBuffer) if err := child.writeStruct(v); err != nil { return err } - w.SetOffsetAndSize(i, start, child.writer.buf.WriterIndex()-start) - return nil + return w.SetOffsetAndSize(i, start, child.writer.buf.WriterIndex()-start) }, read: func(g valueReader, i int, v reflect.Value) error { nested := g.Struct(i) @@ -383,8 +381,7 @@ func newListCodec(goType reflect.Type, listType *ListType, buf *fory.ByteBuffer) return err } } - w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) - return nil + return w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) }, read: func(g valueReader, i int, v reflect.Value) error { arr := g.Array(i) @@ -455,8 +452,7 @@ func newMapCodec(goType reflect.Type, mapType *MapType, buf *fory.ByteBuffer) (v return err } } - w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) - return nil + return w.SetOffsetAndSize(i, start, buf.WriterIndex()-start) }, read: func(g valueReader, i int, v reflect.Value) error { md := g.Map(i) diff --git a/go/fory/row/infer.go b/go/fory/row/infer.go index a2daad1cb6..109ae74129 100644 --- a/go/fory/row/infer.go +++ b/go/fory/row/infer.go @@ -87,7 +87,14 @@ func inferStructLayout(t reflect.Type, path []reflect.Type) (*structLayout, erro var members []member for i := 0; i < t.NumField(); i++ { f := t.Field(i) - if f.PkgPath != "" || hasIgnoreTag(f.Tag.Get("fory")) { + if f.PkgPath != "" { + continue + } + ignored, err := hasIgnoreTag(f.Tag.Get("fory")) + if err != nil { + return nil, fmt.Errorf("%w (field %s of %v)", err, f.Name, t) + } + if ignored { continue } members = append(members, member{lowerFirst(f.Name), i, f.Type}) @@ -185,13 +192,32 @@ func inferField(name string, t reflect.Type, path []reflect.Type) (Field, error) } } -func hasIgnoreTag(tag string) bool { +// hasIgnoreTag mirrors the ignore semantics of the core fory tag +// parser (parseFieldTag in field_spec.go): a whole tag of "-" or an +// ignore/ignore=true part skips the field, ignore=false keeps it, and +// any other ignore value is an error. Other tag keys are not used by +// the row format. +func hasIgnoreTag(tag string) (bool, error) { + if tag == "-" { + return true, nil + } + ignore := false for _, part := range strings.Split(tag, ",") { - if strings.TrimSpace(part) == "ignore" { - return true + part = strings.TrimSpace(part) + if part == "ignore" { + ignore = true + } else if strings.HasPrefix(part, "ignore=") { + switch strings.TrimPrefix(part, "ignore=") { + case "true": + ignore = true + case "false": + ignore = false + default: + return false, fmt.Errorf("row: invalid ignore value in fory tag %q", tag) + } } } - return false + return ignore, nil } func lowerFirst(s string) string { diff --git a/go/fory/row/infer_test.go b/go/fory/row/infer_test.go index cb7c44510f..d62513eb07 100644 --- a/go/fory/row/infer_test.go +++ b/go/fory/row/infer_test.go @@ -121,16 +121,28 @@ func TestInferTypeMapping(t *testing.T) { require.Equal(t, DurationType{}, byName("dur").Type) } +// Ignore semantics must match the core fory tag parser: "-", +// "ignore", and "ignore=true" skip; "ignore=false" keeps. func TestInferSkipsIgnoredAndUnexported(t *testing.T) { type tagged struct { A int64 hidden int64 Skipped string `fory:"ignore"` + Dash string `fory:"-"` + True string `fory:"ignore=true"` + Kept int32 `fory:"ignore=false"` } s, err := InferSchema(reflect.TypeOf(tagged{})) require.NoError(t, err) - require.Equal(t, 1, s.NumFields()) + require.Equal(t, 2, s.NumFields()) require.Equal(t, "a", s.Field(0).Name) + require.Equal(t, "kept", s.Field(1).Name) + + type badTag struct { + A int64 `fory:"ignore=yes"` + } + _, err = InferSchema(reflect.TypeOf(badTag{})) + requireErrorContains(t, err, "invalid ignore value") } func TestInferRejectsUnsupportedTypes(t *testing.T) { diff --git a/go/fory/row/row.go b/go/fory/row/row.go index 1fedf6ad38..77e03d819a 100644 --- a/go/fory/row/row.go +++ b/go/fory/row/row.go @@ -201,10 +201,17 @@ func (a *ArrayData) SizeBytes() int { return len(a.data) } // covered by the available bytes, so decoders can check before // allocating from an attacker-declared count. func (a *ArrayData) validateBounds() error { - need := int64(a.headerBytes) + int64(a.numElements)*int64(a.elemSize) - if a.numElements < 0 || need > int64(len(a.data)) { + // The coarse count check involves no arithmetic, so it cannot be + // defeated by overflow: every element occupies at least one data + // byte. It also bounds numElements low enough that headerBytes + // (computed in NewArrayData) and the product below are exact. + if a.numElements < 0 || a.numElements > len(a.data) { return fmt.Errorf("row: array declares %d elements but holds only %d bytes", a.numElements, len(a.data)) } + need := int64(a.headerBytes) + int64(a.numElements)*int64(a.elemSize) + if need > int64(len(a.data)) { + return fmt.Errorf("row: array declares %d elements needing %d bytes but holds only %d", a.numElements, need, len(a.data)) + } return nil } diff --git a/go/fory/row/row_test.go b/go/fory/row/row_test.go index 55f124a104..428b69449f 100644 --- a/go/fory/row/row_test.go +++ b/go/fory/row/row_test.go @@ -242,6 +242,18 @@ func TestOutOfRangeIndexPanics(t *testing.T) { require.Panics(t, func() { r.IsNullAt(-1) }) } +// The wire format packs offset and size into 32 bits each; larger +// values must be rejected, never silently truncated. +func TestOffsetAndSizeRejectWireLimit(t *testing.T) { + w := NewRowWriter(int64StringSchema()) + w.Reset() + require.Error(t, w.SetOffsetAndSize(1, w.Buffer().WriterIndex(), 1<<33)) + + aw := NewArrayWriter(List(StringType{}).Elem, w.Buffer()) + require.NoError(t, aw.Reset(1)) + require.Error(t, aw.SetOffsetAndSize(0, w.Buffer().WriterIndex(), 1<<33)) +} + func TestConcurrentReads(t *testing.T) { w := NewRowWriter(int64StringSchema()) w.Reset() diff --git a/go/fory/row/schema_bytes.go b/go/fory/row/schema_bytes.go index 0150789c7f..0eda14b191 100644 --- a/go/fory/row/schema_bytes.go +++ b/go/fory/row/schema_bytes.go @@ -39,6 +39,8 @@ import ( const ( schemaVersion = 1 fieldNameSizeThreshold = 15 + // Caps readField/readType recursion. + maxSchemaNestingDepth = 64 ) // Header bits 0-1 store the INDEX into this table (matching Java @@ -178,14 +180,20 @@ func SchemaFromBytes(data []byte) (*Schema, error) { } type schemaReader struct { - buf *fory.ByteBuffer - size int - err fory.Error + buf *fory.ByteBuffer + size int + depth int + err fory.Error } func (r *schemaReader) remaining() int { return r.size - r.buf.ReaderIndex() } func (r *schemaReader) readField() (Field, error) { + if r.depth >= maxSchemaNestingDepth { + return Field{}, fmt.Errorf("row: schema nesting exceeds %d levels", maxSchemaNestingDepth) + } + r.depth++ + defer func() { r.depth-- }() header := int(r.buf.ReadUint8(&r.err)) if err := r.err.CheckError(); err != nil { return Field{}, err diff --git a/go/fory/row/schema_bytes_test.go b/go/fory/row/schema_bytes_test.go index 3c790f5128..61da4a3a14 100644 --- a/go/fory/row/schema_bytes_test.go +++ b/go/fory/row/schema_bytes_test.go @@ -159,3 +159,16 @@ func TestSchemaToBytesRejectsEmptyName(t *testing.T) { _, err := SchemaToBytes(NewSchema([]Field{NewField("", Int32Type{}, true)})) requireErrorContains(t, err, "empty name") } + +// A crafted chain of nested LIST type-infos costs ~3 bytes per level; +// without a depth limit it would fatally overflow the goroutine stack. +func TestSchemaFromBytesRejectsDeepNesting(t *testing.T) { + data := []byte{0x01, 0x01} + for i := 0; i < 10*maxSchemaNestingDepth; i++ { + // field: header (UTF_8 encoding, 1-byte name, non-null), + // name "a", type LIST -> recurses into the next field. + data = append(data, 0x00, 'a', 0x16) + } + _, err := SchemaFromBytes(data) + requireErrorContains(t, err, "nesting") +} diff --git a/go/fory/row/writer.go b/go/fory/row/writer.go index 9b612674c9..dd898644de 100644 --- a/go/fory/row/writer.go +++ b/go/fory/row/writer.go @@ -159,22 +159,28 @@ func (w *RowWriter) WriteDuration(i int, d time.Duration) { w.WriteInt64(i, d.Microseconds()) } -func (w *RowWriter) WriteString(i int, s string) { +func (w *RowWriter) WriteString(i int, s string) error { start := appendStringRegion(w.buf, s) - w.SetOffsetAndSize(i, start, len(s)) + return w.SetOffsetAndSize(i, start, len(s)) } -func (w *RowWriter) WriteBytes(i int, b []byte) { +func (w *RowWriter) WriteBytes(i int, b []byte) error { start := appendBytesRegion(w.buf, b) - w.SetOffsetAndSize(i, start, len(b)) + return w.SetOffsetAndSize(i, start, len(b)) } // SetOffsetAndSize patches field i's slot with the row-relative offset // and byte size of a value already appended to the variable data region. // Use it after writing a nested struct, array, or map at absStart. -func (w *RowWriter) SetOffsetAndSize(i, absStart, size int) { +// Offsets and sizes beyond 32 bits are rejected: the wire format packs +// both into one slot, and truncating would corrupt the row silently. +func (w *RowWriter) SetOffsetAndSize(i, absStart, size int) error { rel := absStart - w.base - binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(uint32(size))) + if uint64(rel) > math.MaxUint32 || uint64(size) > math.MaxUint32 { + return fmt.Errorf("row: value at offset %d with %d bytes exceeds the 32-bit wire format limit", rel, size) + } + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(size)) + return nil } // ArrayWriter writes one array: an 8-byte element count, a null bitmap, @@ -295,21 +301,26 @@ func (w *ArrayWriter) WriteDuration(i int, d time.Duration) { w.WriteInt64(i, d.Microseconds()) } -func (w *ArrayWriter) WriteString(i int, s string) { +func (w *ArrayWriter) WriteString(i int, s string) error { start := appendStringRegion(w.buf, s) - w.SetOffsetAndSize(i, start, len(s)) + return w.SetOffsetAndSize(i, start, len(s)) } -func (w *ArrayWriter) WriteBytes(i int, b []byte) { +func (w *ArrayWriter) WriteBytes(i int, b []byte) error { start := appendBytesRegion(w.buf, b) - w.SetOffsetAndSize(i, start, len(b)) + return w.SetOffsetAndSize(i, start, len(b)) } // SetOffsetAndSize patches element i's slot with the array-relative -// offset and byte size of a value already appended after the array. -func (w *ArrayWriter) SetOffsetAndSize(i, absStart, size int) { +// offset and byte size of a value already appended after the array, +// with the same 32-bit wire limit as RowWriter.SetOffsetAndSize. +func (w *ArrayWriter) SetOffsetAndSize(i, absStart, size int) error { rel := absStart - w.base - binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(uint32(size))) + if uint64(rel) > math.MaxUint32 || uint64(size) > math.MaxUint32 { + return fmt.Errorf("row: value at offset %d with %d bytes exceeds the 32-bit wire format limit", rel, size) + } + binary.LittleEndian.PutUint64(w.buf.GetData()[w.slot(i):], uint64(rel)<<32|uint64(size)) + return nil } // MapWriter writes one map: an 8-byte keys-array size, the keys array,