Remove (*Table)Write and introduce (*GenericTable[T]).Write (#626)

This commit removes `(*Table).Write` in favor of `(*GenericTable[T]).Write`

`GenericTable` offers a lot of improvements from the previous API. One of the
biggest advantage is there is no longer need to define schema before hand
instead the schema is part of the struct that is written in the form of struct
tags.

Example code has been refactored to use `map[string]T` for dynamic columns,
however slice struct dynamic columns still works, this is intentional to avoid
the PR from blowing out of proportion.

`GenericTable` embeds `Table` so from user perspective only few changes are
needed to work with the new API.

Author: gernest

aggregate_test.go

@@ -256,48 +256,17 @@ func TestDurationAggregation(t *testing.T) {
 	db, err := c.DB(context.Background(), "test")
 	require.NoError(t, err)
 
-	schema := &schemapb.Schema{
-		Name: "test",
-		Columns: []*schemapb.Column{
-			{
-				Name: "timestamp",
-				StorageLayout: &schemapb.StorageLayout{
-					Type: schemapb.StorageLayout_TYPE_INT64,
-				},
-				Dynamic: false,
-			},
-			{
-				Name: "stacktrace",
-				StorageLayout: &schemapb.StorageLayout{
-					Type: schemapb.StorageLayout_TYPE_STRING,
-				},
-				Dynamic: false,
-			},
-			{
-				Name: "value",
-				StorageLayout: &schemapb.StorageLayout{
-					Type: schemapb.StorageLayout_TYPE_INT64,
-				},
-				Dynamic: false,
-			},
-		},
-		SortingColumns: []*schemapb.SortingColumn{
-			{
-				Name:      "timestamp",
-				Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
-			},
-		},
+	type Record struct {
+		Timestamp  int64 `frostdb:",asc"`
+		Stacktrace string
+		Value      int64
 	}
 
-	config := NewTableConfig(schema)
-	table, err := db.Table("test", config)
+	table, err := NewGenericTable[Record](db, "test", memory.NewGoAllocator())
 	require.NoError(t, err)
+	defer table.Release()
 
-	records := []struct {
-		Timestamp  int64
-		Stacktrace string
-		Value      int64
-	}{
+	records := []Record{
 		{
 			Timestamp:  1 * int64(time.Second),
 			Stacktrace: "stack1",
@@ -324,11 +293,8 @@ func TestDurationAggregation(t *testing.T) {
 			Value:      3,
 		},
 	}
-
-	for _, record := range records {
-		_, err := table.Write(context.Background(), record)
-		require.NoError(t, err)
-	}
+	err = table.Write(context.Background(), records...)
+	require.NoError(t, err)
 
 	engine := query.NewEngine(memory.DefaultAllocator, db.TableProvider())
 

db_test.go

@@ -1062,37 +1062,12 @@ func Test_DB_Block_Optimization(t *testing.T) {
 
 func Test_DB_TableWrite_FlatSchema(t *testing.T) {
 	ctx := context.Background()
-	flatDefinition := &schemapb.Schema{
-		Name: "test",
-		Columns: []*schemapb.Column{{
-			Name: "example_type",
-			StorageLayout: &schemapb.StorageLayout{
-				Type:     schemapb.StorageLayout_TYPE_STRING,
-				Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
-			},
-			Dynamic: false,
-		}, {
-			Name: "timestamp",
-			StorageLayout: &schemapb.StorageLayout{
-				Type: schemapb.StorageLayout_TYPE_INT64,
-			},
-			Dynamic: false,
-		}, {
-			Name: "value",
-			StorageLayout: &schemapb.StorageLayout{
-				Type: schemapb.StorageLayout_TYPE_INT64,
-			},
-			Dynamic: false,
-		}},
-		SortingColumns: []*schemapb.SortingColumn{{
-			Name:      "example_type",
-			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
-		}, {
-			Name:      "timestamp",
-			Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
-		}},
+
+	type Flat struct {
+		ExampleType string `frostdb:",rle_dict,asc(0)"`
+		Timestamp   int64  `frostdb:",asc(1)"`
+		Value       int64
 	}
-	config := NewTableConfig(flatDefinition)
 
 	c, err := New(WithLogger(newTestLogger(t)))
 	require.NoError(t, err)
@@ -1101,20 +1076,15 @@ func Test_DB_TableWrite_FlatSchema(t *testing.T) {
 	db, err := c.DB(ctx, "flatschema")
 	require.NoError(t, err)
 
-	table, err := db.Table("test", config)
+	table, err := NewGenericTable[Flat](db, "test", memory.NewGoAllocator())
 	require.NoError(t, err)
+	defer table.Release()
 
-	s := struct {
-		ExampleType string
-		Timestamp   int64
-		Value       int64
-	}{
+	err = table.Write(ctx, Flat{
 		ExampleType: "hello-world",
 		Timestamp:   7,
 		Value:       8,
-	}
-
-	_, err = table.Write(ctx, s)
+	})
 	require.NoError(t, err)
 
 	engine := query.NewEngine(
@@ -1132,9 +1102,6 @@ func Test_DB_TableWrite_FlatSchema(t *testing.T) {
 
 func Test_DB_TableWrite_DynamicSchema(t *testing.T) {
 	ctx := context.Background()
-	config := NewTableConfig(
-		dynparquet.SampleDefinition(),
-	)
 
 	c, err := New(WithLogger(newTestLogger(t)))
 	require.NoError(t, err)
@@ -1143,8 +1110,9 @@ func Test_DB_TableWrite_DynamicSchema(t *testing.T) {
 	db, err := c.DB(ctx, "sampleschema")
 	require.NoError(t, err)
 
-	table, err := db.Table("test", config)
+	table, err := NewGenericTable[dynparquet.Sample](db, "test", memory.NewGoAllocator())
 	require.NoError(t, err)
+	defer table.Release()
 
 	now := time.Now()
 	ts := now.UnixMilli()
@@ -1191,7 +1159,7 @@ func Test_DB_TableWrite_DynamicSchema(t *testing.T) {
 		},
 	}
 
-	_, err = table.Write(ctx, samples[0], samples[1], samples[2])
+	err = table.Write(ctx, samples[0], samples[1], samples[2])
 	require.NoError(t, err)
 
 	engine := query.NewEngine(

dynparquet/record_builder.go

@@ -343,6 +343,9 @@ func newMapFieldBuilder(newField func(string) fieldBuilder) *mapFieldBuilder {
 var _ fieldBuilder = (*mapFieldBuilder)(nil)
 
 func (m *mapFieldBuilder) Fields() (o []arrow.Field) {
+	if len(m.columns) == 0 {
+		return []arrow.Field{}
+	}
 	o = make([]arrow.Field, 0, len(m.columns))
 	m.keys = slices.Grow(m.keys, len(m.columns))
 	for k := range m.columns {
@@ -356,6 +359,9 @@ func (m *mapFieldBuilder) Fields() (o []arrow.Field) {
 }
 
 func (m *mapFieldBuilder) NewArray(a []arrow.Array) []arrow.Array {
+	if len(m.columns) == 0 {
+		return a
+	}
 	m.keys = m.keys[:0]
 	for k := range m.columns {
 		m.keys = append(m.keys, k)
@@ -393,6 +399,12 @@ func (m *mapFieldBuilder) Append(v reflect.Value) error {
 }
 
 func (m *mapFieldBuilder) appendMap(v reflect.Value) error {
+	if v.IsNil() || v.Len() == 0 {
+		for _, v := range m.columns {
+			v.AppendNull()
+		}
+		return nil
+	}
 	clear(m.seen)
 	keys := v.MapKeys()
 	size := m.Len()
@@ -416,6 +428,12 @@ func (m *mapFieldBuilder) appendMap(v reflect.Value) error {
 }
 
 func (m *mapFieldBuilder) appendSlice(v reflect.Value) error {
+	if v.IsNil() || v.Len() == 0 {
+		for _, v := range m.columns {
+			v.AppendNull()
+		}
+		return nil
+	}
 	clear(m.seen)
 	size := m.Len()
 	for n := 0; n < v.Len(); n++ {

examples/aggregations/aggregations.go

@@ -8,7 +8,6 @@ import (
 	"github.com/apache/arrow/go/v14/arrow/memory"
 
 	"github.com/polarsignals/frostdb"
-	schemapb "github.com/polarsignals/frostdb/gen/proto/go/frostdb/schema/v1alpha1"
 	"github.com/polarsignals/frostdb/query"
 	"github.com/polarsignals/frostdb/query/logicalplan"
 )
@@ -23,38 +22,34 @@ func main() {
 	// Open up a database in the column store
 	database, _ := columnstore.DB(context.Background(), "weather_db")
 
-	// Define our aggregation schema of labels and values
-	schema := aggregationSchema()
-
-	// Create a table named snowfall_table in our database
-	table, _ := database.Table(
-		"snowfall_table",
-		frostdb.NewTableConfig(schema),
-	)
-
 	// Create values to insert into the database. We support a dynamic structure for city to
 	// accommodate cities in different regions
 	type WeatherRecord struct {
-		City     interface{}
-		Day      string
+		City     map[string]string `frostdb:",rle_dict,asc(0)"`
+		Day      string            `frostdb:",rle_dict,asc(1)"`
 		Snowfall float64
 	}
 
-	type CityInProvince struct {
-		Name     string
-		Province string
-	}
+	// Create a table named snowfall_table in our database
+	table, _ := frostdb.NewGenericTable[WeatherRecord](
+		database, "snowfall_table", memory.DefaultAllocator,
+	)
+	defer table.Release()
 
-	type CityInState struct {
-		Name  string
-		State string
+	montreal := map[string]string{
+		"name":     "Montreal",
+		"province": "Quebec",
+	}
+	toronto := map[string]string{
+		"name":     "Toronto",
+		"province": "Ontario",
+	}
+	minneapolis := map[string]string{
+		"name":  "Minneapolis",
+		"state": "Minnesota",
 	}
 
-	montreal := CityInProvince{Name: "Montreal", Province: "Quebec"}
-	toronto := CityInProvince{Name: "Toronto", Province: "Ontario"}
-	minneapolis := CityInState{Name: "Minneapolis", State: "Minnesota"}
-
-	_, _ = table.Write(context.Background(),
+	_ = table.Write(context.Background(),
 		WeatherRecord{Day: "Mon", Snowfall: 20, City: montreal},
 		WeatherRecord{Day: "Tue", Snowfall: 00, City: montreal},
 		WeatherRecord{Day: "Wed", Snowfall: 30, City: montreal},
@@ -105,45 +100,3 @@ func main() {
 			return nil
 		})
 }
-
-func aggregationSchema() *schemapb.Schema {
-	return &schemapb.Schema{
-		Name: "snowfall_table",
-		Columns: []*schemapb.Column{
-			{
-				Name: "city",
-				StorageLayout: &schemapb.StorageLayout{
-					Type:     schemapb.StorageLayout_TYPE_STRING,
-					Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
-					Nullable: false,
-				},
-				Dynamic: true,
-			},
-			{
-				Name: "day",
-				StorageLayout: &schemapb.StorageLayout{
-					Type:     schemapb.StorageLayout_TYPE_STRING,
-					Encoding: schemapb.StorageLayout_ENCODING_RLE_DICTIONARY,
-					Nullable: false,
-				},
-			},
-			{
-				Name: "snowfall",
-				StorageLayout: &schemapb.StorageLayout{
-					Type: schemapb.StorageLayout_TYPE_DOUBLE,
-				},
-				Dynamic: false,
-			},
-		},
-		SortingColumns: []*schemapb.SortingColumn{
-			{
-				Name:      "city",
-				Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
-			},
-			{
-				Name:      "day",
-				Direction: schemapb.SortingColumn_DIRECTION_ASCENDING,
-			},
-		},
-	}
-}