[clickhouse] change raw table definition to use int64 for _peerdb_bat…

…ch_id (#2248)
PeerDB-io · Nov 13, 2024 · 0fbd7f7 · 0fbd7f7
1 parent 1f969d1
commit 0fbd7f7
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git a/flow/connectors/clickhouse/cdc.go b/flow/connectors/clickhouse/cdc.go
@@ -46,13 +46,13 @@ func (c *ClickHouseConnector) CreateRawTable(ctx context.Context, req *protos.Cr
 	rawTableName := c.getRawTableName(req.FlowJobName)
 
 	createRawTableSQL := `CREATE TABLE IF NOT EXISTS %s (
-		_peerdb_uid UUID NOT NULL,
-		_peerdb_timestamp Int64 NOT NULL,
-		_peerdb_destination_table_name String NOT NULL,
-		_peerdb_data String NOT NULL,
-		_peerdb_record_type Int NOT NULL,
+		_peerdb_uid UUID,
+		_peerdb_timestamp Int64,
+		_peerdb_destination_table_name String,
+		_peerdb_data String,
+		_peerdb_record_type Int,
 		_peerdb_match_data String,
-		_peerdb_batch_id Int,
+		_peerdb_batch_id Int64,
 		_peerdb_unchanged_toast_columns String
 	) ENGINE = MergeTree() ORDER BY (_peerdb_batch_id, _peerdb_destination_table_name);`
 

diff --git a/flow/connectors/clickhouse/normalize.go b/flow/connectors/clickhouse/normalize.go
@@ -462,7 +462,7 @@ func (c *ClickHouseConnector) getDistinctTableNamesInBatch(
 	rawTbl := c.getRawTableName(flowJobName)
 
 	q := fmt.Sprintf(
-		`SELECT DISTINCT _peerdb_destination_table_name FROM %s WHERE _peerdb_batch_id > %d AND _peerdb_batch_id <= %d`,
+		`SELECT DISTINCT _peerdb_destination_table_name FROM %s WHERE _peerdb_batch_id>%d AND _peerdb_batch_id<=%d`,
 		rawTbl, normalizeBatchID, syncBatchID)
 
 	rows, err := c.query(ctx, q)