materialize-motherduck: use materialize-sql-v2

Updates the materialize-motherduck connector to use materialize-sql-v2.

This is the first of many of these conversions, which are relatively simple.
Starting with this one single connector makes sense right now since it gets
enough usage to hopefully flush out any bugs before converting some of the more
extensively used materializations.

Author: williamhbaker

materialize-motherduck/.snapshots/TestSQLGeneration

@@ -1,6 +1,6 @@
---- Begin ""."a-schema".key_value createTargetTable ---
+--- Begin db."a-schema".key_value createTargetTable ---
 
-CREATE TABLE IF NOT EXISTS ""."a-schema".key_value (
+CREATE TABLE IF NOT EXISTS db."a-schema".key_value (
 	key1 BIGINT NOT NULL,
 	key2 BOOLEAN NOT NULL,
 	"key!binary" VARCHAR NOT NULL,
@@ -22,16 +22,16 @@ CREATE TABLE IF NOT EXISTS ""."a-schema".key_value (
 	"stringNumber" DOUBLE,
 	flow_document JSON NOT NULL
 );
---- End ""."a-schema".key_value createTargetTable ---
+--- End db."a-schema".key_value createTargetTable ---
 
---- Begin ""."a-schema".delta_updates createTargetTable ---
+--- Begin db."a-schema".delta_updates createTargetTable ---
 
-CREATE TABLE IF NOT EXISTS ""."a-schema".delta_updates (
+CREATE TABLE IF NOT EXISTS db."a-schema".delta_updates (
 	"theKey" VARCHAR NOT NULL,
 	"aValue" BIGINT,
 	flow_published_at TIMESTAMP WITH TIME ZONE NOT NULL
 );
---- End ""."a-schema".delta_updates createTargetTable ---
+--- End db."a-schema".delta_updates createTargetTable ---
 
 --- Begin Fence Update ---
 UPDATE path."to".checkpoints
@@ -42,9 +42,9 @@ UPDATE path."to".checkpoints
 	AND   fence     = 123;
 --- End Fence Update ---
 
---- Begin ""."a-schema".key_value loadQuery ---
+--- Begin db."a-schema".key_value loadQuery ---
 SELECT 0 AS binding, l.flow_document AS doc
-FROM ""."a-schema".key_value AS l
+FROM db."a-schema".key_value AS l
 JOIN read_json(
 	['s3://bucket/file1', 's3://bucket/file2'],
 	format='newline_delimited',
@@ -58,10 +58,10 @@ JOIN read_json(
 	 ON  l.key1 = r.key1 AND l.key1 >= 10 AND l.key1 <= 100
 	 AND l.key2 = r.key2
 	 AND l."key!binary" = r."key!binary" AND l."key!binary" >= 'aGVsbG8K' AND l."key!binary" <= 'Z29vZGJ5ZQo='
---- End ""."a-schema".key_value loadQuery ---
+--- End db."a-schema".key_value loadQuery ---
 
---- Begin ""."a-schema".key_value storeDeleteQuery ---
-DELETE FROM ""."a-schema".key_value AS l
+--- Begin db."a-schema".key_value storeDeleteQuery ---
+DELETE FROM db."a-schema".key_value AS l
 USING read_json(
 	['s3://bucket/file1', 's3://bucket/file2'],
 	format='newline_delimited',
@@ -92,10 +92,10 @@ USING read_json(
 	 WHERE l.key1 = r.key1 AND l.key1 >= 10 AND l.key1 <= 100
 	 AND l.key2 = r.key2
 	 AND l."key!binary" = r."key!binary" AND l."key!binary" >= 'aGVsbG8K' AND l."key!binary" <= 'Z29vZGJ5ZQo=';
---- End ""."a-schema".key_value storeDeleteQuery ---
+--- End db."a-schema".key_value storeDeleteQuery ---
 
---- Begin ""."a-schema".key_value storeQuery ---
-INSERT INTO ""."a-schema".key_value BY NAME
+--- Begin db."a-schema".key_value storeQuery ---
+INSERT INTO db."a-schema".key_value BY NAME
 SELECT * FROM read_json(
 	['s3://bucket/file1', 's3://bucket/file2'],
 	format='newline_delimited',
@@ -123,10 +123,10 @@ SELECT * FROM read_json(
 		flow_document: 'JSON NOT NULL'
 	}
 ) WHERE flow_document != '"delete"';
---- End ""."a-schema".key_value storeQuery ---
+--- End db."a-schema".key_value storeQuery ---
 
---- Begin ""."a-schema".delta_updates storeQuery ---
-INSERT INTO ""."a-schema".delta_updates BY NAME
+--- Begin db."a-schema".delta_updates storeQuery ---
+INSERT INTO db."a-schema".delta_updates BY NAME
 SELECT * FROM read_json(
 	['s3://bucket/file1', 's3://bucket/file2'],
 	format='newline_delimited',
@@ -137,6 +137,6 @@ SELECT * FROM read_json(
 		flow_published_at: 'TIMESTAMP WITH TIME ZONE NOT NULL'
 	}
 );
---- End ""."a-schema".delta_updates storeQuery ---
+--- End db."a-schema".delta_updates storeQuery ---
 
 

materialize-motherduck/Dockerfile

@@ -17,10 +17,10 @@ COPY flow-bin/flowctl-go /usr/local/bin/flowctl-go
 COPY go                      ./go
 COPY materialize-boilerplate ./materialize-boilerplate
 COPY materialize-motherduck  ./materialize-motherduck
-COPY materialize-sql         ./materialize-sql
+COPY materialize-sql-v2      ./materialize-sql-v2
 
 # Test and build the connector.
-RUN go test  -tags nozstd -v ./materialize-sql/...
+RUN go test  -tags nozstd -v ./materialize-sql-v2/...
 RUN go test  -tags nozstd -v ./materialize-motherduck/...
 RUN go build -tags nozstd -v -o ./connector ./materialize-motherduck/...
 

materialize-motherduck/client.go

@@ -10,36 +10,32 @@ import (
 	"github.com/estuary/connectors/go/blob"
 	cerrors "github.com/estuary/connectors/go/connector-errors"
 	boilerplate "github.com/estuary/connectors/materialize-boilerplate"
-	sql "github.com/estuary/connectors/materialize-sql"
+	sql "github.com/estuary/connectors/materialize-sql-v2"
 
 	_ "github.com/marcboeker/go-duckdb/v2"
 )
 
 var _ sql.SchemaManager = (*client)(nil)
 
 type client struct {
-	db  *stdsql.DB
-	cfg *config
-	ep  *sql.Endpoint
+	db *stdsql.DB
+	ep *sql.Endpoint[config]
 }
 
-func newClient(ctx context.Context, ep *sql.Endpoint) (sql.Client, error) {
-	cfg := ep.Config.(*config)
-
-	db, err := cfg.db(ctx)
+func newClient(ctx context.Context, ep *sql.Endpoint[config]) (sql.Client, error) {
+	db, err := ep.Config.db(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	return &client{
-		db:  db,
-		cfg: cfg,
-		ep:  ep,
+		db: db,
+		ep: ep,
 	}, nil
 }
 
-func (c *client) InfoSchema(ctx context.Context, resourcePaths [][]string) (*boilerplate.InfoSchema, error) {
-	return sql.StdFetchInfoSchema(ctx, c.db, c.ep.Dialect, c.cfg.Database, resourcePaths)
+func (c *client) PopulateInfoSchema(ctx context.Context, is *boilerplate.InfoSchema, resourcePaths [][]string) error {
+	return sql.StdPopulateInfoSchema(ctx, is, c.db, c.ep.Dialect, c.ep.Config.Database, resourcePaths)
 }
 
 func (c *client) CreateTable(ctx context.Context, tc sql.TableCreate) error {
@@ -99,7 +95,7 @@ func (c *client) ListSchemas(ctx context.Context) ([]string, error) {
 	// StdListSchemasFn won't work if there are schemas with the same name in other databases.
 	rows, err := c.db.QueryContext(ctx, fmt.Sprintf(
 		"select schema_name from information_schema.schemata where catalog_name = %s",
-		duckDialect.Literal(c.cfg.Database),
+		duckDialect.Literal(c.ep.Config.Database),
 	))
 	if err != nil {
 		return nil, fmt.Errorf("querying schemata: %w", err)
@@ -119,15 +115,13 @@ func (c *client) ListSchemas(ctx context.Context) ([]string, error) {
 	return out, nil
 }
 
-func (c *client) CreateSchema(ctx context.Context, schemaName string) error {
+func (c *client) CreateSchema(ctx context.Context, schemaName string) (string, error) {
 	return sql.StdCreateSchema(ctx, c.db, duckDialect, schemaName)
 }
 
-func preReqs(ctx context.Context, conf any, tenant string) *cerrors.PrereqErr {
+func preReqs(ctx context.Context, cfg config, tenant string) *cerrors.PrereqErr {
 	errs := &cerrors.PrereqErr{}
 
-	cfg := conf.(*config)
-
 	db, err := cfg.db(ctx)
 	if err != nil {
 		errs.Err(err)

materialize-motherduck/config.go

@@ -12,7 +12,6 @@ import (
 	"github.com/estuary/connectors/go/blob"
 	schemagen "github.com/estuary/connectors/go/schema-gen"
 	boilerplate "github.com/estuary/connectors/materialize-boilerplate"
-	sql "github.com/estuary/connectors/materialize-sql"
 	"github.com/invopop/jsonschema"
 	"google.golang.org/api/option"
 )
@@ -34,7 +33,7 @@ type advancedConfig struct {
 	FeatureFlags string `json:"feature_flags,omitempty" jsonschema:"title=Feature Flags,description=This property is intended for Estuary internal use. You should only modify this field as directed by Estuary support."`
 }
 
-func (c *config) Validate() error {
+func (c config) Validate() error {
 	var requiredProperties = [][]string{
 		{"token", c.Token},
 		{"database", c.Database},
@@ -107,6 +106,14 @@ func (c *config) Validate() error {
 	return nil
 }
 
+func (c config) DefaultNamespace() string {
+	return c.Schema
+}
+
+func (c config) FeatureFlags() (string, map[string]bool) {
+	return c.Advanced.FeatureFlags, featureFlagDefaults
+}
+
 type stagingBucketType string
 
 const (
@@ -225,16 +232,6 @@ type tableConfig struct {
 	database string
 }
 
-func newTableConfig(ep *sql.Endpoint) sql.Resource {
-	return &tableConfig{
-		// Default to the endpoint schema. This will be over-written by a present `schema` property
-		// within `raw`.
-		Schema:   ep.Config.(*config).Schema,
-		database: ep.Config.(*config).Database,
-	}
-}
-
-// Validate the resource configuration.
 func (r tableConfig) Validate() error {
 	if r.Table == "" {
 		return fmt.Errorf("missing table")
@@ -243,10 +240,15 @@ func (r tableConfig) Validate() error {
 	return nil
 }
 
-func (c tableConfig) Path() sql.TablePath {
-	return []string{c.database, c.Schema, c.Table}
+func (r tableConfig) WithDefaults(cfg config) tableConfig {
+	if r.Schema == "" {
+		r.Schema = cfg.Schema
+	}
+	r.database = cfg.Database
+
+	return r
 }
 
-func (c tableConfig) DeltaUpdates() bool {
-	return c.Delta
+func (r tableConfig) Parameters() ([]string, bool, error) {
+	return []string{r.database, r.Schema, r.Table}, r.Delta, nil
 }

materialize-motherduck/driver.go

@@ -9,11 +9,10 @@ import (
 	"strings"
 
 	"github.com/estuary/connectors/go/blob"
-	"github.com/estuary/connectors/go/common"
 	m "github.com/estuary/connectors/go/protocols/materialize"
 	boilerplate "github.com/estuary/connectors/materialize-boilerplate"
 	enc "github.com/estuary/connectors/materialize-boilerplate/stream-encode"
-	sql "github.com/estuary/connectors/materialize-sql"
+	sql "github.com/estuary/connectors/materialize-sql-v2"
 	pf "github.com/estuary/flow/go/protocols/flow"
 	pm "github.com/estuary/flow/go/protocols/materialize"
 	"github.com/google/uuid"
@@ -25,46 +24,39 @@ import (
 	_ "github.com/marcboeker/go-duckdb/v2"
 )
 
-func newDuckDriver() *sql.Driver {
-	return &sql.Driver{
+func newDuckDriver() *sql.Driver[config, tableConfig] {
+	return &sql.Driver[config, tableConfig]{
 		DocumentationURL: "https://go.estuary.dev/materialize-motherduck",
-		EndpointSpecType: new(config),
-		ResourceSpecType: new(tableConfig),
-		StartTunnel:      func(ctx context.Context, conf any) error { return nil },
-		NewEndpoint: func(ctx context.Context, raw json.RawMessage, tenant string) (*sql.Endpoint, error) {
-			var cfg = new(config)
-			if err := pf.UnmarshalStrict(raw, cfg); err != nil {
-				return nil, fmt.Errorf("could not parse endpoint configuration: %w", err)
-			}
-
+		StartTunnel:      func(ctx context.Context, cfg config) error { return nil },
+		NewEndpoint: func(ctx context.Context, cfg config, tenant string, featureFlags map[string]bool) (*sql.Endpoint[config], error) {
 			log.WithFields(log.Fields{
 				"database": cfg.Database,
 			}).Info("opening database")
 
-			var featureFlags = common.ParseFeatureFlags(cfg.Advanced.FeatureFlags, featureFlagDefaults)
-			if cfg.Advanced.FeatureFlags != "" {
-				log.WithField("flags", featureFlags).Info("parsed feature flags")
-			}
-
-			return &sql.Endpoint{
+			return &sql.Endpoint[config]{
 				Config:              cfg,
 				Dialect:             duckDialect,
 				MetaCheckpoints:     sql.FlowCheckpointsTable([]string{cfg.Database, cfg.Schema}),
 				NewClient:           newClient,
 				CreateTableTemplate: tplCreateTargetTable,
-				NewResource:         newTableConfig,
 				NewTransactor:       newTransactor,
 				Tenant:              tenant,
 				ConcurrentApply:     false,
-				FeatureFlags:        featureFlags,
+				Options: boilerplate.MaterializeOptions{
+					ExtendedLogging: true,
+					AckSchedule: &boilerplate.AckScheduleOption{
+						Config: cfg.Schedule,
+						Jitter: []byte(cfg.Token),
+					},
+				},
 			}, nil
 		},
 		PreReqs: preReqs,
 	}
 }
 
 type transactor struct {
-	cfg *config
+	cfg config
 
 	fence      sql.Fence
 	conn       *stdsql.Conn
@@ -79,28 +71,28 @@ type transactor struct {
 
 func newTransactor(
 	ctx context.Context,
-	ep *sql.Endpoint,
+	ep *sql.Endpoint[config],
 	fence sql.Fence,
 	bindings []sql.Table,
 	open pm.Request_Open,
 	is *boilerplate.InfoSchema,
 	be *boilerplate.BindingEvents,
-) (_ m.Transactor, _ *boilerplate.MaterializeOptions, err error) {
-	cfg := ep.Config.(*config)
+) (_ m.Transactor, err error) {
+	var cfg = ep.Config
 
 	db, err := cfg.db(ctx)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	conn, err := db.Conn(ctx)
 	if err != nil {
-		return nil, nil, fmt.Errorf("creating connection: %w", err)
+		return nil, fmt.Errorf("creating connection: %w", err)
 	}
 
 	bucket, bucketPath, err := cfg.toBucketAndPath(ctx)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	t := &transactor{
@@ -124,15 +116,7 @@ func newTransactor(
 		})
 	}
 
-	opts := &boilerplate.MaterializeOptions{
-		ExtendedLogging: true,
-		AckSchedule: &boilerplate.AckScheduleOption{
-			Config: cfg.Schedule,
-			Jitter: []byte(cfg.Token),
-		},
-	}
-
-	return t, opts, nil
+	return t, nil
 }
 
 type binding struct {