bugfix to ensure resync cannot fail due to table size too large

Author: jgao54

flow/connectors/clickhouse/normalize.go

@@ -270,6 +270,13 @@ func (c *ClickHouseConnector) generateCreateTableSQLForNormalizedTable(
 		if allowNullableKey {
 			stmtBuilder.WriteString(chinternal.NewCHSettingsString(chVersion, chinternal.SettingAllowNullableKey, "1"))
 		}
+		if config.IsResync {
+			// CREATE OR REPLACE TABLE (used by resync) internally would create a table with name
+			// <_tmp_replace_*>, atomic swap with the existing table, and then drop the existing table.
+			// Setting max_table_size_to_drop = 0 ensure the implicit drop does not fail; it can be
+			// non-empty if a resync is triggered on top of another resync that is still snapshotting.
+			stmtBuilder.WriteString(chinternal.NewCHSettingsString(chVersion, chinternal.SettingMaxTableSizeToDrop, "0"))
+		}
 
 		if c.Config.Cluster != "" {
 			fmt.Fprintf(&stmtBuilderDistributed, " ENGINE = Distributed(%s,%s,%s",