| 164 | } |
| 165 | |
| 166 | func testPluginCustom(t *testing.T, s *spec.Spec) { |
| 167 | ctx := context.Background() |
| 168 | |
| 169 | var client plugin.Client |
| 170 | |
| 171 | p := plugin.NewPlugin("file", "development", func(ctx context.Context, logger zerolog.Logger, spec []byte, opts plugin.NewClientOptions) (plugin.Client, error) { |
| 172 | var err error |
| 173 | client, err = New(ctx, logger, spec, opts) |
| 174 | return client, err |
| 175 | }) |
| 176 | b, err := json.Marshal(s) |
| 177 | if err != nil { |
| 178 | t.Fatal(err) |
| 179 | } |
| 180 | if err := p.Init(ctx, b, plugin.NewClientOptions{}); err != nil { |
| 181 | t.Fatal(err) |
| 182 | } |
| 183 | |
| 184 | tableName := fmt.Sprintf("cq_test_custom_insert_%d", time.Now().UnixNano()) |
| 185 | table := &schema.Table{ |
| 186 | Name: tableName, |
| 187 | Columns: []schema.Column{ |
| 188 | {Name: "name", Type: arrow.BinaryTypes.String}, |
| 189 | }, |
| 190 | } |
| 191 | if err := p.WriteAll(ctx, []message.WriteMessage{ |
| 192 | &message.WriteMigrateTable{ |
| 193 | Table: table, |
| 194 | }, |
| 195 | }); err != nil { |
| 196 | t.Fatal(fmt.Errorf("failed to create table: %w", err)) |
| 197 | } |
| 198 | |
| 199 | bldr := array.NewRecordBuilder(memory.DefaultAllocator, table.ToArrowSchema()) |
| 200 | bldr.Field(0).(*array.StringBuilder).Append("foo") |
| 201 | record := bldr.NewRecordBatch() |
| 202 | |
| 203 | if err := p.WriteAll(ctx, []message.WriteMessage{ |
| 204 | &message.WriteInsert{ |
| 205 | Record: record, |
| 206 | }, |
| 207 | &message.WriteInsert{ |
| 208 | Record: record, |
| 209 | }, |
| 210 | }); err != nil { |
| 211 | t.Fatal(fmt.Errorf("failed to insert records: %w", err)) |
| 212 | } |
| 213 | |
| 214 | if err := client.Close(ctx); err != nil { |
| 215 | t.Fatal(fmt.Errorf("failed to close client: %w", err)) |
| 216 | } |
| 217 | |
| 218 | readRecords, err := readAll(ctx, client, table) |
| 219 | if err != nil { |
| 220 | t.Fatal(fmt.Errorf("failed to sync: %w", err)) |
| 221 | } |
| 222 | |
| 223 | totalItems := plugin.TotalRows(readRecords) |