feat(test): Test writing to a child table (#878)

Closes #877

Author: candiduslynx

internal/memdb/memdb.go

@@ -95,7 +95,7 @@ func (c *client) overwrite(table *schema.Table, data arrow.Record) {
 }
 
 func (c *client) Migrate(_ context.Context, tables schema.Tables) error {
-	for _, table := range tables {
+	for _, table := range tables.FlattenTables() {
 		tableName := table.Name
 		memTable := c.memoryDB[tableName]
 		if memTable == nil {

plugins/destination/plugin_testing_overwrite.go

@@ -20,11 +20,10 @@ func (*PluginTestSuite) destinationPluginTestWriteOverwrite(ctx context.Context,
 	}
 	tableName := fmt.Sprintf("cq_%s_%d", spec.Name, time.Now().Unix())
 	table := schema.TestTable(tableName, testSourceOptions...)
+	parent := schema.TestTable(tableName+"_parent", testSourceOptions...)
+	parent.Relations = schema.Tables{table}
 	syncTime := time.Now().UTC().Round(1 * time.Second)
-	tables := schema.Tables{
-		table,
-	}
-	if err := p.Migrate(ctx, tables); err != nil {
+	if err := p.Migrate(ctx, schema.Tables{parent}); err != nil {
 		return fmt.Errorf("failed to migrate tables: %w", err)
 	}
 

plugins/destination/plugin_testing_overwrite_delete_stale.go

@@ -22,12 +22,10 @@ func (*PluginTestSuite) destinationPluginTestWriteOverwriteDeleteStale(ctx conte
 	table := schema.TestTable(tableName, testSourceOptions...)
 	incTable := schema.TestTable(tableName+"_incremental", testSourceOptions...)
 	incTable.IsIncremental = true
+	parent := schema.TestTable(tableName+"_parent", testSourceOptions...)
+	parent.Relations = schema.Tables{table, incTable}
 	syncTime := time.Now().UTC().Round(1 * time.Second)
-	tables := schema.Tables{
-		table,
-		incTable,
-	}
-	if err := p.Migrate(ctx, tables); err != nil {
+	if err := p.Migrate(ctx, schema.Tables{parent}); err != nil {
 		return fmt.Errorf("failed to migrate tables: %w", err)
 	}
 
@@ -108,7 +106,7 @@ func (*PluginTestSuite) destinationPluginTestWriteOverwriteDeleteStale(ctx conte
 		return fmt.Errorf("after overwrite expected first resource to be different. diff: %s", diff)
 	}
 
-	resourcesRead, err = p.readAll(ctx, tables[0], sourceName)
+	resourcesRead, err = p.readAll(ctx, table, sourceName)
 	if err != nil {
 		return fmt.Errorf("failed to read all second time: %w", err)
 	}
@@ -124,7 +122,7 @@ func (*PluginTestSuite) destinationPluginTestWriteOverwriteDeleteStale(ctx conte
 
 	// we expect the incremental table to still have 2 resources, because delete-stale should
 	// not apply there
-	resourcesRead, err = p.readAll(ctx, tables[1], sourceName)
+	resourcesRead, err = p.readAll(ctx, incTable, sourceName)
 	if err != nil {
 		return fmt.Errorf("failed to read all from incremental table: %w", err)
 	}

plugins/destination/plugin_testing_write_append.go

@@ -19,11 +19,10 @@ func (s *PluginTestSuite) destinationPluginTestWriteAppend(ctx context.Context,
 	}
 	tableName := fmt.Sprintf("cq_%s_%d", spec.Name, time.Now().Unix())
 	table := schema.TestTable(tableName, testSourceOptions...)
+	parent := schema.TestTable(tableName+"_parent", testSourceOptions...)
+	parent.Relations = schema.Tables{table}
 	syncTime := time.Now().UTC().Round(1 * time.Second)
-	tables := schema.Tables{
-		table,
-	}
-	if err := p.Migrate(ctx, tables); err != nil {
+	if err := p.Migrate(ctx, schema.Tables{parent}); err != nil {
 		return fmt.Errorf("failed to migrate tables: %w", err)
 	}
 
@@ -53,7 +52,7 @@ func (s *PluginTestSuite) destinationPluginTestWriteAppend(ctx context.Context,
 		}
 	}
 
-	resourcesRead, err := p.readAll(ctx, tables[0], sourceName)
+	resourcesRead, err := p.readAll(ctx, table, sourceName)
 	if err != nil {
 		return fmt.Errorf("failed to read all second time: %w", err)
 	}