feat: Allow homogeneous data types to be configured (#1533)

For OpenSearch (and presumably ElasticSearch) arrays (even in JSON objects) are limited to the same data types. This adds a configuration option to the writer test suite to only use the same data type in the JSON example data - allowing the test suite to be used for the OpenSearch plugin. Currently JSON objects in the ElasticSearch and OpenSearch plugins are mapped to strings.

Author: mnorbury

plugin/testing_write.go

@@ -24,6 +24,9 @@ type WriterTestSuite struct {
 	// Destination setups that don't support nulls in lists should set this to true.
 	ignoreNullsInLists bool
 
+	// useHomogeneousTypes use the same type for all items within an array - useful for destinations that don't support mixed types.
+	useHomogeneousTypes bool
+
 	// genDataOptions define how to generate test data and which data types to skip
 	genDatOptions schema.TestSourceOptions
 
@@ -92,6 +95,12 @@ func WithRandomSeed(seed int64) func(o *WriterTestSuite) {
 	}
 }
 
+func WithHomogeneousTypes() func(o *WriterTestSuite) {
+	return func(o *WriterTestSuite) {
+		o.useHomogeneousTypes = true
+	}
+}
+
 func TestWriterSuiteRunner(t *testing.T, p *Plugin, tests WriterTestSuiteTests, opts ...func(o *WriterTestSuite)) {
 	suite := &WriterTestSuite{
 		tests:  tests,

plugin/testing_write_delete.go

@@ -93,10 +93,11 @@ func (s *WriterTestSuite) testDeleteStaleAll(ctx context.Context, t *testing.T)
 
 	tg := schema.NewTestDataGenerator(0)
 	normalRecord := tg.Generate(table, schema.GenTestDataOptions{
-		MaxRows:       rowsPerRecord,
-		TimePrecision: s.genDatOptions.TimePrecision,
-		SourceName:    "test",
-		SyncTime:      syncTime, // Generate call may truncate the value further based on the options
+		MaxRows:            rowsPerRecord,
+		TimePrecision:      s.genDatOptions.TimePrecision,
+		SourceName:         "test",
+		SyncTime:           syncTime, // Generate call may truncate the value further based on the options
+		UseHomogeneousType: s.useHomogeneousTypes,
 	})
 	r.NoErrorf(s.plugin.writeOne(ctx, &message.WriteInsert{Record: normalRecord}), "failed to insert record")
 	normalRecord = s.handleNulls(normalRecord) // we process nulls after writing
@@ -118,11 +119,12 @@ func (s *WriterTestSuite) testDeleteStaleAll(ctx context.Context, t *testing.T)
 	// https://github.com/golang/go/issues/41087
 	syncTime = time.Now().UTC().Truncate(time.Microsecond)
 	nullRecord := tg.Generate(table, schema.GenTestDataOptions{
-		MaxRows:       rowsPerRecord,
-		TimePrecision: s.genDatOptions.TimePrecision,
-		NullRows:      true,
-		SourceName:    "test",
-		SyncTime:      syncTime, // Generate call may truncate the value further based on the options
+		MaxRows:            rowsPerRecord,
+		TimePrecision:      s.genDatOptions.TimePrecision,
+		NullRows:           true,
+		SourceName:         "test",
+		SyncTime:           syncTime, // Generate call may truncate the value further based on the options
+		UseHomogeneousType: s.useHomogeneousTypes,
 	})
 	r.NoErrorf(s.plugin.writeOne(ctx, &message.WriteInsert{Record: nullRecord}), "failed to insert record second time")
 	nullRecord = s.handleNulls(nullRecord) // we process nulls after writing

plugin/testing_write_insert.go

@@ -91,8 +91,9 @@ func (s *WriterTestSuite) testInsertAll(ctx context.Context) error {
 	}
 	tg := schema.NewTestDataGenerator(0)
 	normalRecord := tg.Generate(table, schema.GenTestDataOptions{
-		MaxRows:       rowsPerRecord,
-		TimePrecision: s.genDatOptions.TimePrecision,
+		MaxRows:            rowsPerRecord,
+		TimePrecision:      s.genDatOptions.TimePrecision,
+		UseHomogeneousType: s.useHomogeneousTypes,
 	})
 	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
 		Record: normalRecord,

plugin/testing_write_migrate.go

@@ -38,10 +38,11 @@ func (s *WriterTestSuite) migrate(ctx context.Context, target *schema.Table, sou
 	sourceName := target.Name
 	syncTime := time.Now().UTC().Round(1 * time.Second)
 	opts := schema.GenTestDataOptions{
-		SourceName:    sourceName,
-		SyncTime:      syncTime,
-		MaxRows:       rowsPerRecord,
-		TimePrecision: s.genDatOptions.TimePrecision,
+		SourceName:         sourceName,
+		SyncTime:           syncTime,
+		MaxRows:            rowsPerRecord,
+		TimePrecision:      s.genDatOptions.TimePrecision,
+		UseHomogeneousType: s.useHomogeneousTypes,
 	}
 	// Test Generator should be initialized with the current number of items in the destination
 	// this allows us to have multi-pass tests that ensure the migrations are stable

plugin/testing_write_upsert.go

@@ -80,8 +80,9 @@ func (s *WriterTestSuite) testUpsertAll(ctx context.Context) error {
 
 	tg := schema.NewTestDataGenerator(0)
 	normalRecord := tg.Generate(table, schema.GenTestDataOptions{
-		MaxRows:       rowsPerRecord,
-		TimePrecision: s.genDatOptions.TimePrecision,
+		MaxRows:            rowsPerRecord,
+		TimePrecision:      s.genDatOptions.TimePrecision,
+		UseHomogeneousType: s.useHomogeneousTypes,
 	})
 	if err := s.plugin.writeOne(ctx, &message.WriteInsert{
 		Record: normalRecord,

schema/testdata.go

@@ -196,6 +196,8 @@ type GenTestDataOptions struct {
 	TimePrecision time.Duration
 	// NullRows indicates whether to generate rows with all null values.
 	NullRows bool
+	// UseHomogeneousType indicates whether to use a single type for JSON arrays.
+	UseHomogeneousType bool
 }
 
 type TestDataGenerator struct {
@@ -297,8 +299,14 @@ func (tg TestDataGenerator) getExampleJSON(colName string, dataType arrow.DataTy
 	}
 	if arrow.TypeEqual(dataType, types.ExtensionTypes.JSON) {
 		if strings.HasSuffix(colName, "_array") {
+			if opts.UseHomogeneousType {
+				return `[{"test1":"test1"},{"test2":"test2"},{"test3":"test3"}]`
+			}
 			return `[{"test":"test"},123,{"test_number":456}]`
 		}
+		if opts.UseHomogeneousType {
+			return `{"test":["a", "b", "c"]}`
+		}
 		return `{"test":["a","b",3]}`
 	}
 	if arrow.TypeEqual(dataType, types.ExtensionTypes.Inet) {