supporting Snowflake as a source for full table QRep #1368

heavycrystal · 2024-02-23T20:21:10Z

sample command for reference:

CREATE MIRROR watermark3 FROM sf_target TO catalog FOR
$$
  SELECT * FROM public.retwal1
$$
WITH (
        destination_table_name = 'public.retwal2',
        watermark_column = '', -- has to be empty
        watermark_table_name = 'public.retwal1',
        mode = 'overwrite',
        parallelism = 2, -- no effect
        refresh_interval = 10, -- no effect
        num_rows_per_partition = 500, -- no effect
        setup_watermark_table_on_destination = true,
        initial_copy_only=true
);

UI:

some refactors for QRepPullConnector, SQL Server is now not a valid QRepPullConnector

serprex · 2024-02-23T20:35:32Z

flow/activities/flowable.go


-		stream, err = recordBatch.ToQRecordStream(bufferSize)
+	go func() {


should split out a preceding PR to make this use errgroup instead of waitgroup+goroutine+goroutineErr

ui/app/mirrors/create/handlers.ts

flow/connectors/postgres/qrep_sql_sync.go

Amogh-Bharadwaj · 2024-02-27T21:17:02Z

Should SQL Server tests and QRep support be removed in a preceding PR? @heavycrystal @iskakaushik
Those tests are failing since we've removed SQL Server as a QRep pull connector here

serprex · 2024-02-27T22:17:40Z

flow/activities/flowable.go

@@ -141,6 +141,21 @@ func (a *FlowableActivity) GetTableSchema(
 	return srcConn.GetTableSchema(ctx, config)
 }

+// GetTableSchema returns the schema of a table.


Suggested change

// GetTableSchema returns the schema of a table.

// GetTableSchemaQRep returns the schema of a table.

Better off moving GetTableSchema to its own connector interface & having both workflows use same activity method

serprex · 2024-02-27T22:20:10Z

flow/cmd/peer_data.go

+
+			tables = append(tables, table.String)
+		}
+		return &protos.AllTablesResponse{Tables: tables}, nil


It's surprising, but you also need to check rows.Err() after loop. Which is why the pgx Collect methods are nice, because they handle this all

serprex · 2024-02-27T22:21:13Z

flow/connectors/core.go

@@ -136,11 +136,15 @@ type CDCNormalizeConnector interface {
 type QRepPullConnector interface {
 	Connector

+	// GetTableSchema returns the schema of a table.
+	GetTableSchema(ctx context.Context, req *protos.GetTableSchemaBatchInput) (*protos.GetTableSchemaBatchOutput, error)
+
 	// GetQRepPartitions returns the partitions for a given table that haven't been synced yet.
 	GetQRepPartitions(ctx context.Context, config *protos.QRepConfig, last *protos.QRepPartition) ([]*protos.QRepPartition, error)

 	// PullQRepRecords returns the records for a given partition.


Suggested change

// PullQRepRecords returns the records for a given partition.

// PullQRepRecordStream returns the records for a given partition.

Need to update rest of comment too

serprex · 2024-02-27T22:22:09Z

flow/connectors/core.go

@@ -255,7 +259,8 @@ var (
 	_ NormalizedTablesConnector = &connclickhouse.ClickhouseConnector{}

 	_ QRepPullConnector = &connpostgres.PostgresConnector{}
-	_ QRepPullConnector = &connsqlserver.SQLServerConnector{}
+	// _ QRepPullConnector = &connsqlserver.SQLServerConnector{}


Need to implement GetTableSchema for SQLServerConnector

Or push to have support for it removed

serprex · 2024-02-27T22:23:27Z

flow/connectors/postgres/qrep_query_executor.go

+	defer func() {
+		rows.Close()
+
+		// description of .Close() says it should only be called once rows are closed


Suggested change

// description of .Close() says it should only be called once rows are closed

// description of .Err() says it should only be called once rows are closed

Seems like it'd be more straightforward to explicitly call rows.Close before the rows.Err check before. Fine to close multiple times

Replaces a usage waitgroup with errgroup in replicateQRepPartitions in flowable.go This is a preceding PR to #1368

serprex · 2024-02-29T19:12:42Z

flow/connectors/core.go

 	// GetQRepPartitions returns the partitions for a given table that haven't been synced yet.
 	GetQRepPartitions(ctx context.Context, config *protos.QRepConfig, last *protos.QRepPartition) ([]*protos.QRepPartition, error)

 	// PullQRepRecords returns the records for a given partition.
-	PullQRepRecords(ctx context.Context, config *protos.QRepConfig, partition *protos.QRepPartition) (*model.QRecordBatch, error)
+	PullQRepRecordStream(ctx context.Context, config *protos.QRepConfig,


can create something like a QRepPullStreamConnector interface instead, prefer that, fallback to QRepPullConnector when ErrUnsupportedFunctionality

heavycrystal · 2024-03-18T14:52:41Z

not mainlining this for now, will reraise PR and address issues then

heavycrystal requested review from serprex and iskakaushik February 23, 2024 20:21

heavycrystal marked this pull request as draft February 23, 2024 20:21

serprex reviewed Feb 23, 2024

View reviewed changes

serprex reviewed Feb 26, 2024

View reviewed changes

ui/app/mirrors/create/handlers.ts Outdated Show resolved Hide resolved

serprex reviewed Feb 26, 2024

View reviewed changes

flow/connectors/postgres/qrep_sql_sync.go Outdated Show resolved Hide resolved

Amogh-Bharadwaj marked this pull request as ready for review February 27, 2024 20:42

Amogh-Bharadwaj force-pushed the sf-full-table-qrep branch from e78a04c to ec32b34 Compare February 27, 2024 20:45

Amogh-Bharadwaj mentioned this pull request Feb 27, 2024

Flowable: Use errgroup in replicateQRep #1390

Merged

Amogh-Bharadwaj force-pushed the sf-full-table-qrep branch from 4aa074a to 1d8f164 Compare February 27, 2024 21:15

serprex reviewed Feb 27, 2024

View reviewed changes

Amogh-Bharadwaj added a commit that referenced this pull request Feb 28, 2024

Flowable: Use errgroup in replicateQRep (#1390)

96ca628

Replaces a usage waitgroup with errgroup in replicateQRepPartitions in flowable.go This is a preceding PR to #1368

serprex reviewed Feb 29, 2024

View reviewed changes

heavycrystal and others added 11 commits March 7, 2024 01:11

supporting Snowflake as a source for full table QRep

9de1501

fixed log

665561c

patch create qrep UI

188727b

ui: make overwrite only option

13b15d6

support overwrite with drop and recreate

b518e36

ui: dropdown for table select

153ec47

changes

beed89c

minor change

9c5bcf9

lint

bc811a6

lint 2

e436b0c

fixed some comments

10adad0

heavycrystal force-pushed the sf-full-table-qrep branch from 1d8f164 to 10adad0 Compare March 6, 2024 20:30

heavycrystal and others added 2 commits March 7, 2024 02:00

Merge branch 'main' into sf-full-table-qrep

7267370

fixed minor bug

a5099c3

heavycrystal closed this Mar 18, 2024

serprex deleted the sf-full-table-qrep branch July 19, 2024 15:27

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

supporting Snowflake as a source for full table QRep #1368

supporting Snowflake as a source for full table QRep #1368

heavycrystal commented Feb 23, 2024 •

edited by Amogh-Bharadwaj

Loading

serprex Feb 23, 2024

serprex Feb 27, 2024

Amogh-Bharadwaj commented Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024

serprex Feb 27, 2024

serprex Feb 27, 2024 •

edited

Loading

serprex Feb 29, 2024

heavycrystal commented Mar 18, 2024


		stream, err = recordBatch.ToQRecordStream(bufferSize)
		go func() {

	// GetTableSchema returns the schema of a table.
	// GetTableSchemaQRep returns the schema of a table.

	// PullQRepRecords returns the records for a given partition.
	// PullQRepRecordStream returns the records for a given partition.

	// description of .Close() says it should only be called once rows are closed
	// description of .Err() says it should only be called once rows are closed

supporting Snowflake as a source for full table QRep #1368

supporting Snowflake as a source for full table QRep #1368

Conversation

heavycrystal commented Feb 23, 2024 • edited by Amogh-Bharadwaj Loading

serprex Feb 23, 2024

Choose a reason for hiding this comment

serprex Feb 27, 2024

Choose a reason for hiding this comment

Amogh-Bharadwaj commented Feb 27, 2024 • edited Loading

serprex Feb 27, 2024 • edited Loading

Choose a reason for hiding this comment

serprex Feb 27, 2024 • edited Loading

Choose a reason for hiding this comment

serprex Feb 27, 2024

Choose a reason for hiding this comment

serprex Feb 27, 2024

Choose a reason for hiding this comment

serprex Feb 27, 2024 • edited Loading

Choose a reason for hiding this comment

serprex Feb 29, 2024

Choose a reason for hiding this comment

heavycrystal commented Mar 18, 2024

heavycrystal commented Feb 23, 2024 •

edited by Amogh-Bharadwaj

Loading

Amogh-Bharadwaj commented Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024 •

edited

Loading

serprex Feb 27, 2024 •

edited

Loading