From 4a196a32690e53960236607ac7f7adf37cc42c29 Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Mon, 9 Dec 2024 15:49:05 +0100 Subject: [PATCH 1/6] feat(id): wip! add formats to ingress descriptor --- internal/app/id/create.go | 2 +- internal/infra/id/table_storage.go | 2 +- internal/infra/id/yaml_storage.go | 19 +++++++++++--- pkg/id/driver.go | 18 +++++++------- pkg/id/driver_test.go | 40 +++++++++++++++--------------- pkg/id/model.go | 12 +++++++++ pkg/id/model_ad.go | 12 +++++---- pkg/id/model_formats.go | 29 ++++++++++++++++++++++ 8 files changed, 94 insertions(+), 40 deletions(-) create mode 100644 pkg/id/model_formats.go diff --git a/internal/app/id/create.go b/internal/app/id/create.go index 6d2b4dee..9ba2c097 100755 --- a/internal/app/id/create.go +++ b/internal/app/id/create.go @@ -45,7 +45,7 @@ func newCreateCommand(fullName string, err *os.File, out *os.File, in *os.File) reader := infra.NewRelationReader(relations) - e := id.Create(table, []string{}, reader, idStorageFactory(ingressDescriptor)) + e := id.Create(table, []string{}, nil, reader, idStorageFactory(ingressDescriptor)) if e != nil { fmt.Fprintln(err, e.Description) os.Exit(1) diff --git a/internal/infra/id/table_storage.go b/internal/infra/id/table_storage.go index 3170013a..e21a15e8 100644 --- a/internal/infra/id/table_storage.go +++ b/internal/infra/id/table_storage.go @@ -38,5 +38,5 @@ func (s *TableStorage) Store(adef id.IngressDescriptor) *id.Error { // Read create new Ingress Descriptor with table as start table without relations func (s *TableStorage) Read() (id.IngressDescriptor, *id.Error) { - return id.NewIngressDescriptor(s.table, []string{}, id.NewIngressRelationList([]id.IngressRelation{})), nil + return id.NewIngressDescriptor(s.table, []string{}, id.NewIngressColumnFormatList(map[string]id.IngressColumnFormat{}), id.NewIngressRelationList([]id.IngressRelation{})), nil } diff --git a/internal/infra/id/yaml_storage.go b/internal/infra/id/yaml_storage.go index 0443ffb3..088f9bdb 100755 --- a/internal/infra/id/yaml_storage.go +++ b/internal/infra/id/yaml_storage.go @@ -33,9 +33,15 @@ type YAMLStructure struct { // YAMLIngressDescriptor defines how to store an ingress descriptor in YAML format. type YAMLIngressDescriptor struct { - StartTable string `yaml:"startTable"` - Select []string `yaml:"select"` - Relations []YAMLRelation `yaml:"relations"` + StartTable string `yaml:"startTable"` + Select []string `yaml:"select"` + Formats map[string]YAMLFormat `yaml:"formats"` + Relations []YAMLRelation `yaml:"relations"` +} + +type YAMLFormat struct { + imp string `yaml:"import"` + exp string `yaml:"export"` } // YAMLRelation defines how to store a relation in YAML format. @@ -100,6 +106,11 @@ func (s *YAMLStorage) Read() (id.IngressDescriptor, *id.Error) { return nil, err } + formats := map[string]id.IngressColumnFormat{} + for name, format := range structure.IngressDescriptor.Formats { + formats[name] = id.NewIngressColumnFormat(format.imp, format.exp) + } + relations := []id.IngressRelation{} for _, relation := range structure.IngressDescriptor.Relations { relations = append(relations, @@ -116,7 +127,7 @@ func (s *YAMLStorage) Read() (id.IngressDescriptor, *id.Error) { ) } - return id.NewIngressDescriptor(id.NewTable(structure.IngressDescriptor.StartTable), structure.IngressDescriptor.Select, id.NewIngressRelationList(relations)), nil + return id.NewIngressDescriptor(id.NewTable(structure.IngressDescriptor.StartTable), structure.IngressDescriptor.Select, id.NewIngressColumnFormatList(formats), id.NewIngressRelationList(relations)), nil } func writeFile(structure *YAMLStructure, filename string) *id.Error { diff --git a/pkg/id/driver.go b/pkg/id/driver.go index 99ff230f..90960380 100755 --- a/pkg/id/driver.go +++ b/pkg/id/driver.go @@ -24,7 +24,7 @@ import ( ) // Create and store ingress descriptor for the given start table and relation set. -func Create(startTable string, selectColumns []string, relReader RelationReader, storage Storage) *Error { +func Create(startTable string, selectColumns []string, formats IngressColumnFormatList, relReader RelationReader, storage Storage) *Error { relations, err := relReader.Read() if err != nil { return err @@ -63,7 +63,7 @@ func Create(startTable string, selectColumns []string, relReader RelationReader, } } - id := NewIngressDescriptor(NewTable(startTable), selectColumns, NewIngressRelationList(adrelations)) + id := NewIngressDescriptor(NewTable(startTable), selectColumns, formats, NewIngressRelationList(adrelations)) err = storage.Store(id) if err != nil { @@ -89,7 +89,7 @@ func SetStartTable(table Table, storage Storage) *Error { return &Error{Description: fmt.Sprintf("Table %s doesn't exist", table.Name())} } - updatedID := NewIngressDescriptor(table, id.Select(), id.Relations()) + updatedID := NewIngressDescriptor(table, id.Select(), id.Formats(), id.Relations()) err = storage.Store(updatedID) if err != nil { @@ -119,7 +119,7 @@ func SetChildLookup(relation string, flag bool, storage Storage) *Error { relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { @@ -149,7 +149,7 @@ func SetParentLookup(relation string, flag bool, storage Storage) *Error { relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { @@ -179,7 +179,7 @@ func SetChildWhere(relation string, where string, storage Storage) *Error { relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { @@ -209,7 +209,7 @@ func SetChildSelect(relation string, columns []string, storage Storage) *Error { relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { @@ -239,7 +239,7 @@ func SetParentWhere(relation string, where string, storage Storage) *Error { relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { @@ -269,7 +269,7 @@ func SetParentSelect(relation string, columns []string, storage Storage) *Error relations[i] = rel } - updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), NewIngressRelationList(relations)) + updatedID := NewIngressDescriptor(id.StartTable(), id.Select(), id.Formats(), NewIngressRelationList(relations)) err = storage.Store(updatedID) if err != nil { diff --git a/pkg/id/driver_test.go b/pkg/id/driver_test.go index 1528eed5..f30c3012 100755 --- a/pkg/id/driver_test.go +++ b/pkg/id/driver_test.go @@ -80,7 +80,7 @@ var adCreateTests = []struct { }), id.NewIngressDescriptor( id.NewTable("A"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("B->A", false, false), }), @@ -99,7 +99,7 @@ var adCreateTests = []struct { }), id.NewIngressDescriptor( id.NewTable("A"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, true), adRelationString("B->C", false, true), @@ -123,7 +123,7 @@ var adCreateTests = []struct { }), id.NewIngressDescriptor( id.NewTable("A"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, true), adRelationString("B->C", false, true), @@ -144,7 +144,7 @@ func TestCreate(t *testing.T) { } storage := &MemoryStorage{} - err := id.Create(startTable, []string{}, relReader, storage) + err := id.Create(startTable, []string{}, nil, relReader, storage) assert.Nil(t, err) @@ -203,7 +203,7 @@ var adShowTests = []struct { { id.NewIngressDescriptor( id.NewTable("A"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, true), }), @@ -217,7 +217,7 @@ var adShowTests = []struct { { // example 1 id.NewIngressDescriptor( id.NewTable("I"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -232,7 +232,7 @@ var adShowTests = []struct { { // example 1 id.NewIngressDescriptor( id.NewTable("I"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -247,7 +247,7 @@ var adShowTests = []struct { { // example 1 (table C) id.NewIngressDescriptor( id.NewTable("C"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -263,7 +263,7 @@ var adShowTests = []struct { { // example 1 (table D) id.NewIngressDescriptor( id.NewTable("D"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -277,7 +277,7 @@ var adShowTests = []struct { { // example 1 (table O) id.NewIngressDescriptor( id.NewTable("O"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -292,7 +292,7 @@ var adShowTests = []struct { { // example 2 id.NewIngressDescriptor( id.NewTable("C"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", false, true), adRelationString("O->D", false, true), @@ -347,7 +347,7 @@ var adShowTests = []struct { { // example 3 Variant bis id.NewIngressDescriptor( id.NewTable("O"), - []string{}, + []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("C->O", true, false), }), @@ -397,49 +397,49 @@ var adShowTests = []struct { } func TestUpdateStartTable(t *testing.T) { - storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("old"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("old"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("old->new", false, true), }))} err := id.SetStartTable(id.NewTable("new"), storage) assert.Nil(t, err) - assert.Equal(t, id.NewIngressDescriptor(id.NewTable("new"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + assert.Equal(t, id.NewIngressDescriptor(id.NewTable("new"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("old->new", false, true), })), storage.id) } func TestUpdateStartTableCheckTable(t *testing.T) { - storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("old"), []string{}, id.NewIngressRelationList([]id.IngressRelation{}))} + storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("old"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{}))} err := id.SetStartTable(id.NewTable("new"), storage) assert.EqualError(t, err, "Table new doesn't exist") - assert.Equal(t, id.NewIngressDescriptor(id.NewTable("old"), []string{}, id.NewIngressRelationList([]id.IngressRelation{})), storage.id) + assert.Equal(t, id.NewIngressDescriptor(id.NewTable("old"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{})), storage.id) } func TestUpdateParentLookup(t *testing.T) { - storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("A"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("A"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, true), }))} err := id.SetParentLookup("A_B", true, storage) assert.Nil(t, err) - assert.Equal(t, id.NewIngressDescriptor(id.NewTable("A"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + assert.Equal(t, id.NewIngressDescriptor(id.NewTable("A"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", true, true), })), storage.id) } func TestUpdateChildLookup(t *testing.T) { - storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("A"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + storage := &MemoryStorage{id: id.NewIngressDescriptor(id.NewTable("A"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, true), }))} err := id.SetChildLookup("A_B", false, storage) assert.Nil(t, err) - assert.Equal(t, id.NewIngressDescriptor(id.NewTable("A"), []string{}, id.NewIngressRelationList([]id.IngressRelation{ + assert.Equal(t, id.NewIngressDescriptor(id.NewTable("A"), []string{}, nil, id.NewIngressRelationList([]id.IngressRelation{ adRelationString("A->B", false, false), })), storage.id) } diff --git a/pkg/id/model.go b/pkg/id/model.go index 84bde93b..604ad7f7 100755 --- a/pkg/id/model.go +++ b/pkg/id/model.go @@ -66,10 +66,22 @@ type IngressRelationList interface { String() string } +type IngressColumnFormat interface { + Import() string + Export() string +} + +type IngressColumnFormatList interface { + Len() uint + Get(column string) IngressColumnFormat + // Set(column string, format IngressColumnFormat) +} + // IngressDescriptor from which the puller plan will be computed. type IngressDescriptor interface { StartTable() Table Select() []string + Formats() IngressColumnFormatList Relations() IngressRelationList String() string } diff --git a/pkg/id/model_ad.go b/pkg/id/model_ad.go index 38f3f985..ffb87505 100755 --- a/pkg/id/model_ad.go +++ b/pkg/id/model_ad.go @@ -109,19 +109,21 @@ func (l idrelationList) String() string { } // NewIngressDescriptor initialize a new IngressDescriptor object -func NewIngressDescriptor(start Table, selectColumns []string, relations IngressRelationList) IngressDescriptor { - return id{startTable: table{name: start.Name()}, selectColumns: selectColumns, relations: relations} +func NewIngressDescriptor(start Table, selectColumns []string, formats IngressColumnFormatList, relations IngressRelationList) IngressDescriptor { + return id{startTable: table{name: start.Name()}, selectColumns: selectColumns, formats: formats, relations: relations} } type id struct { startTable table selectColumns []string + formats IngressColumnFormatList relations IngressRelationList } -func (id id) StartTable() Table { return id.startTable } -func (id id) Select() []string { return id.selectColumns } -func (id id) Relations() IngressRelationList { return id.relations } +func (id id) StartTable() Table { return id.startTable } +func (id id) Select() []string { return id.selectColumns } +func (id id) Formats() IngressColumnFormatList { return id.formats } +func (id id) Relations() IngressRelationList { return id.relations } func (id id) String() string { return fmt.Sprintf("%v [%v] (%v)", id.startTable, id.selectColumns, id.relations) } diff --git a/pkg/id/model_formats.go b/pkg/id/model_formats.go new file mode 100644 index 00000000..3610134e --- /dev/null +++ b/pkg/id/model_formats.go @@ -0,0 +1,29 @@ +package id + +type idcolumnformat struct { + imp string + exp string +} + +func NewIngressColumnFormat(imp string, exp string) IngressColumnFormat { + return idcolumnformat{ + imp: imp, + exp: exp, + } +} + +func (f idcolumnformat) Import() string { return f.imp } +func (f idcolumnformat) Export() string { return f.exp } + +type idcolumnformatlist struct { + formats map[string]IngressColumnFormat +} + +func NewIngressColumnFormatList(formats map[string]IngressColumnFormat) IngressColumnFormatList { + return idcolumnformatlist{ + formats: formats, + } +} + +func (l idcolumnformatlist) Len() uint { return uint(len(l.formats)) } +func (l idcolumnformatlist) Get(column string) IngressColumnFormat { return l.formats[column] } From 60ca068a522406abf8640c107691090074fe66d5 Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Mon, 9 Dec 2024 17:28:34 +0100 Subject: [PATCH 2/6] feat(id): wip! add formats to start table --- internal/app/pull/cli.go | 25 ++++++++++++++++--------- internal/app/pull/http.go | 4 ++-- pkg/id/driver.go | 2 +- pkg/id/model.go | 2 ++ pkg/id/model_formats.go | 7 +++++++ pkg/id/model_step.go | 16 +++++++++------- pkg/pull/driver.go | 5 +++-- pkg/pull/driver_parallel.go | 3 ++- pkg/pull/driver_test.go | 4 ++-- pkg/pull/model.go | 1 + pkg/pull/model_table.go | 4 ++++ 11 files changed, 49 insertions(+), 24 deletions(-) diff --git a/internal/app/pull/cli.go b/internal/app/pull/cli.go index f295133e..0f0fa0b1 100755 --- a/internal/app/pull/cli.go +++ b/internal/app/pull/cli.go @@ -117,7 +117,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra os.Exit(1) } - plan, start, startSelect, e2 := getPullerPlan(idStorageFactory(table, ingressDescriptor)) + plan, start, startSelect, formats, e2 := getPullerPlan(idStorageFactory(table, ingressDescriptor)) if e2 != nil { fmt.Fprintln(err, e2.Error()) os.Exit(1) @@ -175,7 +175,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra } puller := pull.NewPullerParallel(plan, datasource, pullExporterFactory(out), tracer, parallel) - if e3 := puller.Pull(start, filter, startSelect, filters, filtersEx); e3 != nil { + if e3 := puller.Pull(start, filter, startSelect, formats, filters, filtersEx); e3 != nil { log.Fatal().AnErr("error", e3).Msg("Fatal error stop the pull command") os.Exit(1) } @@ -222,26 +222,26 @@ func getDataSource(dataconnectorName string, out io.Writer) (pull.DataSource, er return datasourceFactory.New(u.URL.String(), alias.Schema), nil } -func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, error) { +func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, map[string]string, error) { pp, err1 := id.GetPullerPlan(idStorage) if err1 != nil { - return pull.Plan{}, pull.Table{}, []string{}, err1 + return pull.Plan{}, pull.Table{}, []string{}, nil, err1 } relations, err2 := relStorage.List() if err2 != nil { - return pull.Plan{}, pull.Table{}, []string{}, err2 + return pull.Plan{}, pull.Table{}, []string{}, nil, err2 } tables, err3 := tabStorage.List() if err3 != nil { - return pull.Plan{}, pull.Table{}, []string{}, err3 + return pull.Plan{}, pull.Table{}, []string{}, nil, err3 } builder := newBuilder(pp, relations, tables) plan, startTable, err4 := builder.plan() if err4 != nil { - return pull.Plan{}, pull.Table{}, []string{}, err4 + return pull.Plan{}, pull.Table{}, []string{}, nil, err4 } // Check startTable existe in table.yaml @@ -256,8 +256,15 @@ func getPullerPlan(idStorage id.Storage) (pull.Plan, pull.Table, []string, error if !tableExiste { err5 := fmt.Errorf("Table '%s' does not exist in table.yaml", string(startTable.Name)) - return pull.Plan{}, pull.Table{}, []string{}, err5 + return pull.Plan{}, pull.Table{}, []string{}, nil, err5 } - return plan, startTable, pp.Select(), nil + formats := map[string]string{} + if pp.Formats() != nil { + for _, column := range pp.Formats().Columns() { + formats[column] = pp.Formats().Get(column).Export() + } + } + + return plan, startTable, pp.Select(), formats, nil } diff --git a/internal/app/pull/http.go b/internal/app/pull/http.go index 7912b788..6b8b2b94 100644 --- a/internal/app/pull/http.go +++ b/internal/app/pull/http.go @@ -126,7 +126,7 @@ func HandlerFactory(ingressDescriptor string) func(w http.ResponseWriter, r *htt return } - plan, start, startSelect, e2 := getPullerPlan(idStorageFactory(query.Get("table"), ingressDescriptor)) + plan, start, startSelect, formats, e2 := getPullerPlan(idStorageFactory(query.Get("table"), ingressDescriptor)) if e2 != nil { log.Error().Err(e2).Msg("") w.WriteHeader(http.StatusInternalServerError) @@ -141,7 +141,7 @@ func HandlerFactory(ingressDescriptor string) func(w http.ResponseWriter, r *htt pullExporter := pullExporterFactory(w) puller := pull.NewPuller(plan, datasource, pullExporter, pull.NoTraceListener{}) - e3 := puller.Pull(start, pull.Filter{Limit: limit, Values: pull.Row{}, Where: where, Distinct: distinct}, startSelect, nil, nil) + e3 := puller.Pull(start, pull.Filter{Limit: limit, Values: pull.Row{}, Where: where, Distinct: distinct}, startSelect, formats, nil, nil) if e3 != nil { log.Error().Err(e3).Msg("") w.WriteHeader(http.StatusInternalServerError) diff --git a/pkg/id/driver.go b/pkg/id/driver.go index 90960380..2b8b2e56 100755 --- a/pkg/id/driver.go +++ b/pkg/id/driver.go @@ -325,7 +325,7 @@ func GetPullerPlan(storage Storage) (PullerPlan, *Error) { log.Warn().Msg(err.Error()) } - return NewPullerPlan(steps, g.relations, g.tables, id.Select()), nil + return NewPullerPlan(steps, g.relations, g.tables, id.Select(), id.Formats()), nil } // Export the puller plan. diff --git a/pkg/id/model.go b/pkg/id/model.go index 604ad7f7..713c6b43 100755 --- a/pkg/id/model.go +++ b/pkg/id/model.go @@ -73,6 +73,7 @@ type IngressColumnFormat interface { type IngressColumnFormatList interface { Len() uint + Columns() []string Get(column string) IngressColumnFormat // Set(column string, format IngressColumnFormat) } @@ -118,6 +119,7 @@ type PullerPlan interface { Tables() TableList String() string Select() []string + Formats() IngressColumnFormatList } // Error is the error type returned by the domain diff --git a/pkg/id/model_formats.go b/pkg/id/model_formats.go index 3610134e..65cf43cd 100644 --- a/pkg/id/model_formats.go +++ b/pkg/id/model_formats.go @@ -27,3 +27,10 @@ func NewIngressColumnFormatList(formats map[string]IngressColumnFormat) IngressC func (l idcolumnformatlist) Len() uint { return uint(len(l.formats)) } func (l idcolumnformatlist) Get(column string) IngressColumnFormat { return l.formats[column] } +func (l idcolumnformatlist) Columns() []string { + columns := []string{} + for col := range l.formats { + columns = append(columns, col) + } + return columns +} diff --git a/pkg/id/model_step.go b/pkg/id/model_step.go index 2c9f2e42..381575c3 100755 --- a/pkg/id/model_step.go +++ b/pkg/id/model_step.go @@ -66,18 +66,20 @@ type pullerPlan struct { relations IngressRelationList tables TableList startSelect []string + formats IngressColumnFormatList } // NewPullerPlan initialize a new PullerPlan object -func NewPullerPlan(steps []Step, relations IngressRelationList, tables TableList, startSelect []string) PullerPlan { - return pullerPlan{uint(len(steps)), steps, relations, tables, startSelect} +func NewPullerPlan(steps []Step, relations IngressRelationList, tables TableList, startSelect []string, formats IngressColumnFormatList) PullerPlan { + return pullerPlan{uint(len(steps)), steps, relations, tables, startSelect, formats} } -func (l pullerPlan) Len() uint { return l.len } -func (l pullerPlan) Step(idx uint) Step { return l.slice[idx] } -func (l pullerPlan) Relations() IngressRelationList { return l.relations } -func (l pullerPlan) Tables() TableList { return l.tables } -func (l pullerPlan) Select() []string { return l.startSelect } +func (l pullerPlan) Len() uint { return l.len } +func (l pullerPlan) Step(idx uint) Step { return l.slice[idx] } +func (l pullerPlan) Relations() IngressRelationList { return l.relations } +func (l pullerPlan) Tables() TableList { return l.tables } +func (l pullerPlan) Select() []string { return l.startSelect } +func (l pullerPlan) Formats() IngressColumnFormatList { return l.formats } func (l pullerPlan) String() string { switch l.len { case 0: diff --git a/pkg/pull/driver.go b/pkg/pull/driver.go index 1efcf428..712df5f7 100755 --- a/pkg/pull/driver.go +++ b/pkg/pull/driver.go @@ -46,7 +46,7 @@ func NewStep(puller *puller, out ExportedRow, entry Relation) *Step { } type Puller interface { - Pull(start Table, filter Filter, selectColumns []string, filterCohort RowReader, excluded KeyStore) error + Pull(start Table, filter Filter, selectColumns []string, formats map[string]string, filterCohort RowReader, excluded KeyStore) error } type puller struct { @@ -65,8 +65,9 @@ func NewPuller(plan Plan, datasource DataSource, exporter RowExporter, diagnosti } } -func (p *puller) Pull(start Table, filter Filter, selectColumns []string, filterCohort RowReader, excluded KeyStore) error { +func (p *puller) Pull(start Table, filter Filter, selectColumns []string, formats map[string]string, filterCohort RowReader, excluded KeyStore) error { start.selectColumns(selectColumns...) + start.applyFormats(formats) start = p.graph.addMissingColumns(start) if err := p.datasource.Open(); err != nil { diff --git a/pkg/pull/driver_parallel.go b/pkg/pull/driver_parallel.go index 23a71b1c..19a63b0d 100644 --- a/pkg/pull/driver_parallel.go +++ b/pkg/pull/driver_parallel.go @@ -61,8 +61,9 @@ func NewPullerParallel(plan Plan, datasource DataSource, exporter RowExporter, d return puller } -func (p *pullerParallel) Pull(start Table, filter Filter, selectColumns []string, filterCohort RowReader, excluded KeyStore) error { +func (p *pullerParallel) Pull(start Table, filter Filter, selectColumns []string, formats map[string]string, filterCohort RowReader, excluded KeyStore) error { start.selectColumns(selectColumns...) + start.applyFormats(formats) start = p.graph.addMissingColumns(start) if err := p.datasource.Open(); err != nil { diff --git a/pkg/pull/driver_test.go b/pkg/pull/driver_test.go index a9bfaae9..9abf9282 100644 --- a/pkg/pull/driver_test.go +++ b/pkg/pull/driver_test.go @@ -74,7 +74,7 @@ func RunTest(t *testing.T, test *Test) { for _, execution := range test.Executions { collector.Reset() - assert.NoError(t, puller.Pull(execution.Start, execution.Filter, execution.Select, nil, nil)) + assert.NoError(t, puller.Pull(execution.Start, execution.Filter, execution.Select, nil, nil, nil)) assert.Len(t, collector.Result, len(execution.Result)) for i := 0; i < len(execution.Result); i++ { @@ -94,7 +94,7 @@ func RunBench(b *testing.B, test *Test) { for _, execution := range test.Executions { collector.Reset() - assert.NoError(b, puller.Pull(execution.Start, execution.Filter, execution.Select, nil, nil)) + assert.NoError(b, puller.Pull(execution.Start, execution.Filter, execution.Select, nil, nil, nil)) assert.Len(b, collector.Result, len(execution.Result)) } } diff --git a/pkg/pull/model.go b/pkg/pull/model.go index 8f6a80c4..174f74c5 100755 --- a/pkg/pull/model.go +++ b/pkg/pull/model.go @@ -73,6 +73,7 @@ type Relation struct { Foreign RelationTip Where string Select []string + Export map[string]string // optional export mode for each column } type RelationSet []Relation diff --git a/pkg/pull/model_table.go b/pkg/pull/model_table.go index e9628369..0f31ddeb 100755 --- a/pkg/pull/model_table.go +++ b/pkg/pull/model_table.go @@ -178,3 +178,7 @@ func (t *Table) selectColumns(columnNames ...string) { Interface("table", t.Name). Msg("select only columns defined") } + +func (t *Table) applyFormats(formats map[string]string) { + // TODO +} From c487ffcbe91670cb73e31affa4c79a3620f50d34 Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Mon, 9 Dec 2024 17:40:10 +0100 Subject: [PATCH 3/6] feat(id): add formats to start table --- pkg/pull/model_table.go | 32 +++++++++++++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/pkg/pull/model_table.go b/pkg/pull/model_table.go index 0f31ddeb..42c28fb6 100755 --- a/pkg/pull/model_table.go +++ b/pkg/pull/model_table.go @@ -180,5 +180,35 @@ func (t *Table) selectColumns(columnNames ...string) { } func (t *Table) applyFormats(formats map[string]string) { - // TODO + t.template = jsonline.NewTemplate() + + if len(t.Columns) > 0 { + for _, column := range t.Columns { + key := column.Name + + exp := column.Export + if override, ok := formats[column.Name]; ok { + exp = override + } + + switch exp { + case "string": + t.template.WithString(key) + case "numeric": + t.template.WithNumeric(key) + case "base64": + t.template.WithBinary(key) + case "datetime": + t.template.WithDateTime(key) + case "timestamp": + t.template.WithTimestamp(key) + case "no": + t.template.WithHidden(key) + case "presence": + t.template.WithBoolean(key) + default: + t.template.WithAuto(key) + } + } + } } From 49ead490c86be0fa30d56d51b32a7fb20d23c78b Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Wed, 11 Dec 2024 14:43:55 +0100 Subject: [PATCH 4/6] feat(id): fix unit tests --- pkg/pull/driver.go | 2 +- pkg/pull/driver_parallel.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/pull/driver.go b/pkg/pull/driver.go index 712df5f7..fbe26105 100755 --- a/pkg/pull/driver.go +++ b/pkg/pull/driver.go @@ -67,8 +67,8 @@ func NewPuller(plan Plan, datasource DataSource, exporter RowExporter, diagnosti func (p *puller) Pull(start Table, filter Filter, selectColumns []string, formats map[string]string, filterCohort RowReader, excluded KeyStore) error { start.selectColumns(selectColumns...) - start.applyFormats(formats) start = p.graph.addMissingColumns(start) + start.applyFormats(formats) if err := p.datasource.Open(); err != nil { return fmt.Errorf("%w", err) diff --git a/pkg/pull/driver_parallel.go b/pkg/pull/driver_parallel.go index 19a63b0d..5addcd27 100644 --- a/pkg/pull/driver_parallel.go +++ b/pkg/pull/driver_parallel.go @@ -63,8 +63,8 @@ func NewPullerParallel(plan Plan, datasource DataSource, exporter RowExporter, d func (p *pullerParallel) Pull(start Table, filter Filter, selectColumns []string, formats map[string]string, filterCohort RowReader, excluded KeyStore) error { start.selectColumns(selectColumns...) - start.applyFormats(formats) start = p.graph.addMissingColumns(start) + start.applyFormats(formats) if err := p.datasource.Open(); err != nil { return fmt.Errorf("%w", err) From 398a140e3258572c27ac7991333416a08568b815 Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Wed, 11 Dec 2024 14:51:27 +0100 Subject: [PATCH 5/6] feat(id): add format unit test --- pkg/pull/driver_test.go | 17 ++++++++++++----- pkg/pull/testdata/formats.yaml | 29 +++++++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 5 deletions(-) create mode 100644 pkg/pull/testdata/formats.yaml diff --git a/pkg/pull/driver_test.go b/pkg/pull/driver_test.go index 9abf9282..824280e3 100644 --- a/pkg/pull/driver_test.go +++ b/pkg/pull/driver_test.go @@ -30,10 +30,11 @@ import ( ) type Execution struct { - Start pull.Table - Filter pull.Filter - Select []string - Result []string + Start pull.Table + Filter pull.Filter + Select []string + Formats map[string]string + Result []string } type Test struct { @@ -74,7 +75,7 @@ func RunTest(t *testing.T, test *Test) { for _, execution := range test.Executions { collector.Reset() - assert.NoError(t, puller.Pull(execution.Start, execution.Filter, execution.Select, nil, nil, nil)) + assert.NoError(t, puller.Pull(execution.Start, execution.Filter, execution.Select, execution.Formats, nil, nil)) assert.Len(t, collector.Result, len(execution.Result)) for i := 0; i < len(execution.Result); i++ { @@ -174,6 +175,12 @@ func TestBug1(t *testing.T) { LoadAndRunTest(t, "bug1.yaml") } +func TestFormats(t *testing.T) { + t.Parallel() + + LoadAndRunTest(t, "formats.yaml") +} + func BenchmarkSimpleWithComponents(b *testing.B) { test, _ := LoadTest("simple.yaml") diff --git a/pkg/pull/testdata/formats.yaml b/pkg/pull/testdata/formats.yaml new file mode 100644 index 00000000..d4a34a18 --- /dev/null +++ b/pkg/pull/testdata/formats.yaml @@ -0,0 +1,29 @@ +tables: + film: &film + name: film + keys: [film_id] + columns: + - name: title + - name: release_year +dataset: + film: + - { "film_id": 1, "title": "ACADEMY DINOSAUR 1", "release_year": 2006 } + - { "film_id": 2, "title": "ACADEMY DINOSAUR 2" } + - { "film_id": 3, "title": "ACADEMY DINOSAUR 3", "release_year": 2021 } +plan: + components: + film: 0 + relations: [] +executions: + - start: *film + filter: + limit: 0 + values: {} + where: "" + formats: + release_year: "presence" + title: "no" + result: + - '{"release_year":true}' + - '{"release_year":null}' + - '{"release_year":true}' From 593b1f30c4be834e6617997a5bc2e07972cdb655 Mon Sep 17 00:00:00 2001 From: Adrien Aury <44274230+adrienaury@users.noreply.github.com> Date: Wed, 11 Dec 2024 15:10:03 +0100 Subject: [PATCH 6/6] feat(id): add formats to push command --- internal/app/push/cli.go | 21 ++++++++++++++------- internal/app/push/http.go | 4 ++-- pkg/push/driver.go | 6 +++++- pkg/push/driver_test.go | 8 ++++---- pkg/push/model_table.go | 21 +++++++++++++++++++++ 5 files changed, 46 insertions(+), 14 deletions(-) diff --git a/internal/app/push/cli.go b/internal/app/push/cli.go index 5fa345e4..e653e145 100755 --- a/internal/app/push/cli.go +++ b/internal/app/push/cli.go @@ -132,7 +132,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra os.Exit(1) } - plan, e2 := getPlan(idStorageFactory(table, ingressDescriptor), autoTruncate) + plan, formats, e2 := getPlan(idStorageFactory(table, ingressDescriptor), autoTruncate) if e2 != nil { fmt.Fprintln(err, e2.Error()) os.Exit(2) @@ -161,7 +161,7 @@ func NewCommand(fullName string, err *os.File, out *os.File, in *os.File) *cobra observers = append(observers, observer) } - e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, observers...) + e3 := push.Push(rowIteratorFactory(in), datadestination, plan, mode, commitSize, disableConstraints, rowExporter, translator, whereField, savepoint, autoTruncate, formats, observers...) if e3 != nil { log.Fatal().AnErr("error", e3).Msg("Fatal error stop the push command") os.Exit(1) @@ -253,20 +253,20 @@ func getDataDestination(dataconnectorName string) (push.DataDestination, *push.E return datadestinationFactory.New(u.URL.String(), alias.Schema), nil } -func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, *push.Error) { +func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, map[string]string, *push.Error) { id, err1 := idStorage.Read() if err1 != nil { - return nil, &push.Error{Description: err1.Error()} + return nil, nil, &push.Error{Description: err1.Error()} } relations, err2 := relStorage.List() if err2 != nil { - return nil, &push.Error{Description: err2.Error()} + return nil, nil, &push.Error{Description: err2.Error()} } tables, err3 := tabStorage.List() if err3 != nil { - return nil, &push.Error{Description: err3.Error()} + return nil, nil, &push.Error{Description: err3.Error()} } rmap := map[string]relation.Relation{} @@ -286,7 +286,14 @@ func getPlan(idStorage id.Storage, autoTruncate bool) (push.Plan, *push.Error) { pushtmap: map[string]push.Table{}, } - return converter.getPlan(id, autoTruncate), nil + formats := map[string]string{} + if id.Formats() != nil { + for _, column := range id.Formats().Columns() { + formats[column] = id.Formats().Get(column).Export() + } + } + + return converter.getPlan(id, autoTruncate), formats, nil } type idToPushConverter struct { diff --git a/internal/app/push/http.go b/internal/app/push/http.go index 92b732f3..42933a56 100644 --- a/internal/app/push/http.go +++ b/internal/app/push/http.go @@ -97,7 +97,7 @@ func Handler(w http.ResponseWriter, r *http.Request, mode push.Mode, ingressDesc } } - plan, e2 := getPlan(idStorageFactory(query.Get("table"), ingressDescriptor), autoTruncate) + plan, formats, e2 := getPlan(idStorageFactory(query.Get("table"), ingressDescriptor), autoTruncate) if e2 != nil { log.Error().Err(e2).Msg("") w.WriteHeader(http.StatusNotFound) @@ -146,7 +146,7 @@ func Handler(w http.ResponseWriter, r *http.Request, mode push.Mode, ingressDesc log.Debug().Msg(fmt.Sprintf("call Push with mode %s", mode)) - e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", false) + e3 := push.Push(rowIteratorFactory(r.Body), datadestination, plan, mode, commitSize, disableConstraints, push.NoErrorCaptureRowWriter{}, nil, query.Get("using-pk-field"), "", false, formats) if e3 != nil { log.Error().Err(e3).Msg("") w.WriteHeader(http.StatusNotFound) diff --git a/pkg/push/driver.go b/pkg/push/driver.go index 96800ee4..d3b0d6a6 100755 --- a/pkg/push/driver.go +++ b/pkg/push/driver.go @@ -26,7 +26,7 @@ import ( ) // Push write rows to target table -func Push(ri RowIterator, destination DataDestination, plan Plan, mode Mode, commitSize uint, disableConstraints bool, catchError RowWriter, translator Translator, whereField string, savepointPath string, autotruncate bool, observers ...Observer) (err *Error) { +func Push(ri RowIterator, destination DataDestination, plan Plan, mode Mode, commitSize uint, disableConstraints bool, catchError RowWriter, translator Translator, whereField string, savepointPath string, autotruncate bool, formats map[string]string, observers ...Observer) (err *Error) { //nolint:gocyclo defer func() { for _, observer := range observers { if observer != nil { @@ -35,6 +35,10 @@ func Push(ri RowIterator, destination DataDestination, plan Plan, mode Mode, com } }() + if table, ok := plan.FirstTable().(table); ok { + table.applyFormats(formats) + } + err1 := destination.Open(plan, mode, disableConstraints) if err1 != nil { return err1 diff --git a/pkg/push/driver_test.go b/pkg/push/driver_test.go index 0aaef714..332878db 100755 --- a/pkg/push/driver_test.go +++ b/pkg/push/driver_test.go @@ -49,7 +49,7 @@ func TestSimplePush(t *testing.T) { } dest := memoryDataDestination{tables, false, false, false} - err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false) + err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false, nil) assert.Nil(t, err) assert.Equal(t, true, dest.closed) @@ -88,7 +88,7 @@ func TestRelationPush(t *testing.T) { } dest := memoryDataDestination{tables, false, false, false} - err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false) + err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false, nil) // no error assert.Nil(t, err) @@ -137,7 +137,7 @@ func TestRelationPushWithEmptyRelation(t *testing.T) { } dest := memoryDataDestination{tables, false, false, false} - err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false) + err := push.Push(&ri, &dest, plan, push.Insert, 2, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false, nil) // no error assert.Nil(t, err) @@ -188,7 +188,7 @@ func TestInversseRelationPush(t *testing.T) { } dest := memoryDataDestination{tables, false, false, false} - err := push.Push(&ri, &dest, plan, push.Insert, 5, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false) + err := push.Push(&ri, &dest, plan, push.Insert, 5, true, push.NoErrorCaptureRowWriter{}, nil, "", "", false, nil) // no error assert.Nil(t, err) diff --git a/pkg/push/model_table.go b/pkg/push/model_table.go index 3e37217a..08ea0df9 100755 --- a/pkg/push/model_table.go +++ b/pkg/push/model_table.go @@ -114,6 +114,27 @@ type ImportedRow struct { jsonline.Row } +func (t *table) applyFormats(formats map[string]string) { + if t.columns == nil { + return + } + + if l := int(t.columns.Len()); l > 0 { + columns := []Column{} + for idx := 0; idx < l; idx++ { + col := t.columns.Column(uint(idx)) + key := col.Name() + + if format, exist := formats[key]; exist { + columns = append(columns, NewColumn(col.Name(), col.Export(), format, col.Length(), col.LengthInBytes(), col.Truncate())) + } else { + columns = append(columns, NewColumn(col.Name(), col.Export(), col.Import(), col.Length(), col.LengthInBytes(), col.Truncate())) + } + } + t.columns = NewColumnList(columns) + } +} + func (t *table) initTemplate() { t.template = jsonline.NewTemplate()