feat(schema): Embed column creation options (#869)

Author: candiduslynx

internal/servers/destination/v0/destinations.go

@@ -164,7 +164,7 @@ func (s *Server) Write2(msg pb.Destination_Write2Server) error {
 func setCQIDAsPrimaryKeysForTables(tables schema.Tables) {
 	for _, table := range tables {
 		for i, col := range table.Columns {
-			table.Columns[i].CreationOptions.PrimaryKey = col.Name == schema.CqIDColumn.Name
+			table.Columns[i].PrimaryKey = col.Name == schema.CqIDColumn.Name
 		}
 		setCQIDAsPrimaryKeysForTables(table.Relations)
 	}
@@ -175,8 +175,8 @@ func SetDestinationManagedCqColumns(tables []*schema.Table) {
 	for _, table := range tables {
 		for i := range table.Columns {
 			if table.Columns[i].Name == schema.CqIDColumn.Name {
-				table.Columns[i].CreationOptions.Unique = true
-				table.Columns[i].CreationOptions.NotNull = true
+				table.Columns[i].Unique = true
+				table.Columns[i].NotNull = true
 			}
 		}
 		table.OverwriteOrAddColumn(&schema.CqSyncTimeColumn)

internal/servers/destination/v0/schemav2tov3.go

@@ -44,15 +44,14 @@ func ColumnsV2ToV3(columns []schemav2.Column) []schema.Column {
 
 func ColumnV2ToV3(column schemav2.Column) schema.Column {
 	return schema.Column{
-		Name:        column.Name,
-		Description: column.Description,
-		Type:        TypeV2ToV3(column.Type),
-		CreationOptions: schema.ColumnCreationOptions{
-			NotNull:    column.CreationOptions.NotNull,
-			Unique:     column.CreationOptions.Unique,
-			PrimaryKey: column.CreationOptions.PrimaryKey,
-		},
-		IgnoreInTests: column.IgnoreInTests,
+		Name:           column.Name,
+		Description:    column.Description,
+		Type:           TypeV2ToV3(column.Type),
+		NotNull:        column.CreationOptions.NotNull,
+		Unique:         column.CreationOptions.Unique,
+		PrimaryKey:     column.CreationOptions.PrimaryKey,
+		IncrementalKey: column.CreationOptions.IncrementalKey,
+		IgnoreInTests:  column.IgnoreInTests,
 	}
 }
 

plugins/destination/plugin.go

@@ -293,18 +293,18 @@ func (p *Plugin) Close(ctx context.Context) error {
 
 func checkDestinationColumns(tables schema.Tables) error {
 	for _, table := range tables {
-		if table.Columns.Index(schema.CqSourceNameField.Name) == -1 {
-			return fmt.Errorf("table %s is missing column %s. please consider upgrading source plugin", table.Name, schema.CqSourceNameField.Name)
+		if table.Columns.Index(schema.CqSourceNameColumn.Name) == -1 {
+			return fmt.Errorf("table %s is missing column %s. please consider upgrading source plugin", table.Name, schema.CqSourceNameColumn.Name)
 		}
 		if table.Columns.Index(schema.CqSyncTimeColumn.Name) == -1 {
-			return fmt.Errorf("table %s is missing column %s. please consider upgrading source plugin", table.Name, schema.CqSourceNameField.Name)
+			return fmt.Errorf("table %s is missing column %s. please consider upgrading source plugin", table.Name, schema.CqSourceNameColumn.Name)
 		}
-		field := table.Columns.Get(schema.CqIDColumn.Name)
-		if field != nil {
-			if !field.CreationOptions.NotNull {
+		column := table.Columns.Get(schema.CqIDColumn.Name)
+		if column != nil {
+			if !column.NotNull {
 				return fmt.Errorf("column %s.%s cannot be nullable. please consider upgrading source plugin", table.Name, schema.CqIDColumn.Name)
 			}
-			if !field.CreationOptions.Unique {
+			if !column.Unique {
 				return fmt.Errorf("column %s.%s must be unique. please consider upgrading source plugin", table.Name, schema.CqIDColumn.Name)
 			}
 		}

plugins/destination/plugin_testing_migrate.go

@@ -147,7 +147,7 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
 				{Name: "id", Type: types.ExtensionTypes.UUID},
-				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, NotNull: true},
 			}}
 		p := newPlugin()
 		if err := testMigration(ctx, t, p, logger, spec, target, source, strategy.AddColumnNotNull); err != nil {
@@ -170,8 +170,8 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSourceNameColumn,
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
-				{Name: "id", Type: types.ExtensionTypes.UUID, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
-				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "id", Type: types.ExtensionTypes.UUID, NotNull: true},
+				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, NotNull: true},
 			}}
 		target := &schema.Table{
 			Columns: schema.ColumnList{
@@ -203,7 +203,7 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
 				{Name: "id", Type: types.ExtensionTypes.UUID},
-				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, NotNull: true},
 			},
 		}
 		target := &schema.Table{
@@ -212,7 +212,7 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSourceNameColumn,
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
-				{Name: "id", Type: types.ExtensionTypes.UUID, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "id", Type: types.ExtensionTypes.UUID, NotNull: true},
 			}}
 
 		p := newPlugin()
@@ -237,7 +237,7 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
 				{Name: "id", Type: types.ExtensionTypes.UUID},
-				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "bool", Type: arrow.FixedWidthTypes.Boolean, NotNull: true},
 			}}
 		target := &schema.Table{
 			Name: tableName,
@@ -246,7 +246,7 @@ func (*PluginTestSuite) destinationPluginTestMigrate(
 				schema.CqSyncTimeColumn,
 				schema.CqIDColumn,
 				{Name: "id", Type: types.ExtensionTypes.UUID},
-				{Name: "bool", Type: arrow.BinaryTypes.String, CreationOptions: schema.ColumnCreationOptions{NotNull: true}},
+				{Name: "bool", Type: arrow.BinaryTypes.String, NotNull: true},
 			}}
 
 		p := newPlugin()

schema/column.go

@@ -14,17 +14,6 @@ type ColumnList []Column
 // resource holds the current row we are resolving the column for.
 type ColumnResolver func(ctx context.Context, meta ClientMeta, resource *Resource, c Column) error
 
-// ColumnCreationOptions allow modification of how column is defined when table is created
-type ColumnCreationOptions struct {
-	PrimaryKey bool
-	NotNull    bool
-	// IncrementalKey is a flag that indicates if the column is used as part of an incremental key.
-	// It is mainly used for documentation purposes, but may also be used as part of ensuring that
-	// migrations are done correctly.
-	IncrementalKey bool
-	Unique         bool
-}
-
 // Column definition for Table
 type Column struct {
 	// Name of column
@@ -35,62 +24,68 @@ type Column struct {
 	Description string
 	// Column Resolver allows to set your own data for a column; this can be an API call, setting multiple embedded values, etc
 	Resolver ColumnResolver
-	// Creation options allow modifying how column is defined when table is created
-	CreationOptions ColumnCreationOptions
+
 	// IgnoreInTests is used to skip verifying the column is non-nil in integration tests.
 	// By default, integration tests perform a fetch for all resources in cloudquery's test account, and
 	// verify all columns are non-nil.
 	// If IgnoreInTests is true, verification is skipped for this column.
 	// Used when it is hard to create a reproducible environment with this column being non-nil (e.g. various error columns).
 	IgnoreInTests bool
+
+	// PrimaryKey requires the destinations supporting this to include this column into the primary key
+	PrimaryKey bool
+	// NotNull requires the destinations supporting this to mark this column as non-nullable
+	NotNull bool
+	// IncrementalKey is a flag that indicates if the column is used as part of an incremental key.
+	// It is mainly used for documentation purposes, but may also be used as part of ensuring that
+	// migrations are done correctly.
+	IncrementalKey bool
+	// Unique requires the destinations supporting this to mark this column as unique
+	Unique bool
 }
 
 // NewColumnFromArrowField creates a new Column from an arrow.Field
 // arrow.Field is a low-level representation of a CloudQuery column
 // that can be sent over the wire in a cross-language way.
 func NewColumnFromArrowField(f arrow.Field) Column {
-	creationOptions := ColumnCreationOptions{
+	column := Column{
+		Name:    f.Name,
+		Type:    f.Type,
 		NotNull: !f.Nullable,
 	}
-	if v, ok := f.Metadata.GetValue(MetadataPrimaryKey); ok {
-		if v == MetadataTrue {
-			creationOptions.PrimaryKey = true
-		} else {
-			creationOptions.PrimaryKey = false
-		}
-	}
 
-	if v, ok := f.Metadata.GetValue(MetadataUnique); ok {
-		if v == MetadataTrue {
-			creationOptions.Unique = true
-		} else {
-			creationOptions.Unique = false
-		}
-	}
-	return Column{
-		Name:            f.Name,
-		Type:            f.Type,
-		CreationOptions: creationOptions,
-	}
+	v, ok := f.Metadata.GetValue(MetadataPrimaryKey)
+	column.PrimaryKey = ok && v == MetadataTrue
+
+	v, ok = f.Metadata.GetValue(MetadataUnique)
+	column.Unique = ok && v == MetadataTrue
+
+	v, ok = f.Metadata.GetValue(MetadataIncremental)
+	column.IncrementalKey = ok && v == MetadataTrue
+
+	return column
 }
 
 func (c Column) ToArrowField() arrow.Field {
-	mdKV := map[string]string{}
-	if c.CreationOptions.PrimaryKey {
+	mdKV := map[string]string{
+		MetadataPrimaryKey:  MetadataFalse,
+		MetadataUnique:      MetadataFalse,
+		MetadataIncremental: MetadataFalse,
+	}
+	if c.PrimaryKey {
 		mdKV[MetadataPrimaryKey] = MetadataTrue
-	} else {
-		mdKV[MetadataPrimaryKey] = MetadataFalse
 	}
-	if c.CreationOptions.Unique {
+	if c.Unique {
 		mdKV[MetadataUnique] = MetadataTrue
-	} else {
-		mdKV[MetadataUnique] = MetadataFalse
+	}
+	if c.IncrementalKey {
+		mdKV[MetadataIncremental] = MetadataTrue
 	}
 
 	return arrow.Field{
 		Name:     c.Name,
 		Type:     c.Type,
-		Nullable: !c.CreationOptions.NotNull,
+		Nullable: !c.NotNull,
 		Metadata: arrow.MetadataFrom(mdKV),
 	}
 }
@@ -100,12 +95,18 @@ func (c Column) String() string {
 	sb.WriteString(c.Name)
 	sb.WriteString(":")
 	sb.WriteString(c.Type.String())
-	if c.CreationOptions.PrimaryKey {
+	if c.PrimaryKey {
 		sb.WriteString(":PK")
 	}
-	if c.CreationOptions.NotNull {
+	if c.NotNull {
 		sb.WriteString(":NotNull")
 	}
+	if c.Unique {
+		sb.WriteString(":Unique")
+	}
+	if c.IncrementalKey {
+		sb.WriteString(":IncrementalKey")
+	}
 	return sb.String()
 }
 

schema/meta.go

@@ -17,10 +17,8 @@ var CqIDColumn = Column{
 	Name:        "_cq_id",
 	Type:        types.ExtensionTypes.UUID,
 	Description: "Internal CQ ID of the row",
-	CreationOptions: ColumnCreationOptions{
-		NotNull: true,
-		Unique:  true,
-	},
+	NotNull:     true,
+	Unique:      true,
 }
 var CqParentIDColumn = Column{
 	Name:          "_cq_parent_id",
@@ -42,22 +40,6 @@ var CqSourceNameColumn = Column{
 	Description: "Internal CQ row that references the source plugin name data was retrieved",
 }
 
-var CqIDField = arrow.Field{
-	Name: "_cq_id",
-	Type: types.ExtensionTypes.UUID,
-	Metadata: arrow.MetadataFrom(map[string]string{
-		MetadataUnique: MetadataTrue,
-	}),
-}
-var CqSyncTimeField = arrow.Field{
-	Name: "_cq_sync_time",
-	Type: arrow.FixedWidthTypes.Timestamp_us,
-}
-var CqSourceNameField = arrow.Field{
-	Name: "_cq_source_name",
-	Type: arrow.BinaryTypes.String,
-}
-
 func parentCqUUIDResolver() ColumnResolver {
 	return func(_ context.Context, _ ClientMeta, r *Resource, c Column) error {
 		if r.Parent == nil {

schema/table.go

@@ -333,7 +333,7 @@ func (t *Table) ValidateName() error {
 func (t *Table) PrimaryKeysIndexes() []int {
 	var primaryKeys []int
 	for i, c := range t.Columns {
-		if c.CreationOptions.PrimaryKey {
+		if c.PrimaryKey {
 			primaryKeys = append(primaryKeys, i)
 		}
 	}
@@ -347,23 +347,7 @@ func (t *Table) ToArrowSchema() *arrow.Schema {
 		MetadataTableName: t.Name,
 	})
 	for i, c := range t.Columns {
-		fieldMdKv := map[string]string{}
-		if c.CreationOptions.PrimaryKey {
-			fieldMdKv[MetadataPrimaryKey] = MetadataTrue
-		} else {
-			fieldMdKv[MetadataPrimaryKey] = MetadataFalse
-		}
-		if c.CreationOptions.Unique {
-			fieldMdKv[MetadataUnique] = MetadataTrue
-		} else {
-			fieldMdKv[MetadataUnique] = MetadataFalse
-		}
-		fields[i] = arrow.Field{
-			Name:     c.Name,
-			Type:     c.Type,
-			Nullable: !c.CreationOptions.NotNull,
-			Metadata: arrow.MetadataFrom(fieldMdKv),
-		}
+		fields[i] = c.ToArrowField()
 	}
 	return arrow.NewSchema(fields, &schemaMd)
 }
@@ -383,7 +367,7 @@ func (t *Table) GetChanges(old *Table) []TableColumnChange {
 			continue
 		}
 		// Column type or options (e.g. PK, Not Null) changed in the new table definition
-		if c.Type != otherColumn.Type || c.CreationOptions.NotNull != otherColumn.CreationOptions.NotNull || c.CreationOptions.PrimaryKey != otherColumn.CreationOptions.PrimaryKey {
+		if c.Type != otherColumn.Type || c.NotNull != otherColumn.NotNull || c.PrimaryKey != otherColumn.PrimaryKey {
 			changes = append(changes, TableColumnChange{
 				Type:       TableColumnChangeTypeUpdate,
 				ColumnName: c.Name,
@@ -454,7 +438,7 @@ func (t *Table) OverwriteOrAddColumn(column *Column) {
 func (t *Table) PrimaryKeys() []string {
 	var primaryKeys []string
 	for _, c := range t.Columns {
-		if c.CreationOptions.PrimaryKey {
+		if c.PrimaryKey {
 			primaryKeys = append(primaryKeys, c.Name)
 		}
 	}
@@ -465,7 +449,7 @@ func (t *Table) PrimaryKeys() []string {
 func (t *Table) IncrementalKeys() []string {
 	var incrementalKeys []string
 	for _, c := range t.Columns {
-		if c.CreationOptions.IncrementalKey {
+		if c.IncrementalKey {
 			incrementalKeys = append(incrementalKeys, c.Name)
 		}
 	}

schema/testdata.go

@@ -21,14 +21,14 @@ func TestSourceTable(name string) *Table {
 			CqIDColumn,
 			CqParentIDColumn,
 			{
-				Name:            "uuid_pk",
-				Type:            types.ExtensionTypes.UUID,
-				CreationOptions: ColumnCreationOptions{PrimaryKey: true},
+				Name:       "uuid_pk",
+				Type:       types.ExtensionTypes.UUID,
+				PrimaryKey: true,
 			},
 			{
-				Name:            "string_pk",
-				Type:            arrow.BinaryTypes.String,
-				CreationOptions: ColumnCreationOptions{PrimaryKey: true},
+				Name:       "string_pk",
+				Type:       arrow.BinaryTypes.String,
+				PrimaryKey: true,
 			},
 			{
 				Name: "bool",
@@ -152,7 +152,7 @@ func GenTestData(table *Table, opts GenTestDataOptions) []arrow.Record {
 		nullRow := j%2 == 1
 		bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc)
 		for i, c := range table.Columns {
-			if nullRow && !c.CreationOptions.NotNull && !c.CreationOptions.PrimaryKey &&
+			if nullRow && !c.NotNull && !c.PrimaryKey &&
 				c.Name != CqSourceNameColumn.Name &&
 				c.Name != CqSyncTimeColumn.Name &&
 				c.Name != CqIDColumn.Name &&

transformers/struct.go

@@ -278,7 +278,7 @@ func (t *structTransformer) addColumnFromField(field reflect.StructField, parent
 			// use path to allow the following
 			// 1. Don't duplicate the PK fields if the unwrapped struct contains a fields with the same name
 			// 2. Allow specifying the nested unwrapped field as part of the PK.
-			column.CreationOptions.PrimaryKey = true
+			column.PrimaryKey = true
 			t.pkFieldsFound = append(t.pkFieldsFound, pk)
 		}
 	}