Parse Time64 value without setting it during table creation (#3953)

- Adds ClickHouse Time64(6) type support for TIME/TIMETZ columns, gated
by a read-only capability flag (clickhouse_time64_enabled) detected at
flow creation time and persisted to the flow config protobuf.
- Replaces the previous InternalVersion-based approach with a
map<string, bool> flags field on FlowConnectionConfigs,
FlowConnectionConfigsCore, QRepConfig, and
SetupNormalizedTableBatchInput. This decouples connector-specific
capabilities from the global linear version ordering.
- Introduces a GetFlagsConnector interface. ClickHouse's implementation
queries system.settings to check if enable_time_time64_type is already
enabled on the server -- PeerDB never attempts to set this value.
- Flags are immutable after flow creation: if the flag was false and the
admin enables it later, only new flows will use Time64, existing flows
retain their original behavior, including resync. If the flag was true
and the admin disables the setting, queries will error out if it depends
on Time64.

Testing: e2e tests + test migration path in in CHC + test non-supported
CHC environments

Author: jgao54

.github/workflows/flow.yml

@@ -248,7 +248,7 @@ jobs:
           if [ "${{ matrix.db-version.ch }}" = "lts" ]; then
             echo "ch_version=v25.8.11.66-lts" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "stable" ]; then
-            echo "ch_version=v25.11.1.558-stable" >> $GITHUB_OUTPUT
+            echo "ch_version=v25.12.4.35-stable" >> $GITHUB_OUTPUT
           elif [ "${{ matrix.db-version.ch }}" = "latest" ]; then
             # note: latest tag does not always reflect the latest version (could be an update on an lts),
             # but that is okay as we are only using it to invalidate the cache.

flow/activities/flowable_core.go

@@ -242,7 +242,9 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 		defer dstClose(ctx)
 
 		syncState.Store(shared.Ptr("updating schema"))
-		if err := dstConn.ReplayTableSchemaDeltas(ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas); err != nil {
+		if err := dstConn.ReplayTableSchemaDeltas(
+			ctx, config.Env, flowName, options.TableMappings, recordBatchSync.SchemaDeltas, config.Flags,
+		); err != nil {
 			return nil, fmt.Errorf("failed to sync schema: %w", err)
 		}
 
@@ -285,6 +287,7 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 			TableNameSchemaMapping: tableNameSchemaMapping,
 			Env:                    config.Env,
 			Version:                config.Version,
+			Flags:                  config.Flags,
 		})
 		if err != nil {
 			return a.Alerter.LogFlowError(ctx, flowName, fmt.Errorf("failed to push records: %w", err))
@@ -681,6 +684,7 @@ func (a *FlowableActivity) startNormalize(
 			SyncedAtColName:        config.SyncedAtColName,
 			SyncBatchID:            batchID,
 			Version:                config.Version,
+			Flags:                  config.Flags,
 		})
 		if err != nil {
 			return a.Alerter.LogFlowError(ctx, config.FlowJobName,

flow/cmd/handler.go

@@ -61,6 +61,23 @@ func (h *FlowRequestHandler) getPeerID(ctx context.Context, peerName string) (in
 	return id.Int32, nil
 }
 
+func (h *FlowRequestHandler) determineFlags(
+	ctx context.Context,
+	env map[string]string,
+	destPeerName string,
+) ([]string, error) {
+	conn, connClose, err := connectors.GetByNameAs[connectors.GetFlagsConnector](ctx, env, h.pool, destPeerName)
+	if err != nil {
+		if errors.Is(err, errors.ErrUnsupported) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("failed to get destination connector: %w", err)
+	}
+	defer connClose(ctx)
+
+	return conn.GetFlags(ctx)
+}
+
 func (h *FlowRequestHandler) cdcJobEntryExists(ctx context.Context, flowJobName string) (bool, error) {
 	var exists bool
 	err := h.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM flows WHERE name = $1)`, flowJobName).Scan(&exists)
@@ -145,11 +162,16 @@ func (h *FlowRequestHandler) CreateCDCFlow(
 	if cfg == nil {
 		return nil, NewInvalidArgumentApiError(errors.New("connection configs cannot be nil"))
 	}
-	internalVersion, err := internal.PeerDBForceInternalVersion(ctx, req.ConnectionConfigs.Env)
-	if err != nil {
-		return nil, NewInternalApiError(fmt.Errorf("failed to get internal version: %w", err))
+	if internalVersion, err := internal.PeerDBForceInternalVersion(ctx, cfg.Env); err != nil {
+		return nil, NewInternalApiError(err)
+	} else {
+		cfg.Version = internalVersion
+	}
+	if flags, err := h.determineFlags(ctx, cfg.Env, cfg.DestinationName); err != nil {
+		return nil, NewInternalApiError(err)
+	} else {
+		cfg.Flags = flags
 	}
-	cfg.Version = internalVersion
 
 	if !req.AttachToExisting {
 		if exists, err := h.cdcJobEntryExists(ctx, cfg.FlowJobName); err != nil {
@@ -227,11 +249,16 @@ func (h *FlowRequestHandler) CreateQRepFlow(
 	ctx context.Context, req *protos.CreateQRepFlowRequest,
 ) (*protos.CreateQRepFlowResponse, APIError) {
 	cfg := req.QrepConfig
-	internalVersion, err := internal.PeerDBForceInternalVersion(ctx, req.QrepConfig.Env)
-	if err != nil {
-		return nil, NewInternalApiError(fmt.Errorf("failed to get internal version: %w", err))
+	if internalVersion, err := internal.PeerDBForceInternalVersion(ctx, cfg.Env); err != nil {
+		return nil, NewInternalApiError(err)
+	} else {
+		cfg.Version = internalVersion
+	}
+	if flags, err := h.determineFlags(ctx, cfg.Env, cfg.DestinationName); err != nil {
+		return nil, NewInternalApiError(err)
+	} else {
+		cfg.Flags = flags
 	}
-	cfg.Version = internalVersion
 
 	workflowID := fmt.Sprintf("%s-qrepflow-%s", cfg.FlowJobName, uuid.New())
 	workflowOptions := client.StartWorkflowOptions{

flow/cmd/validate_mirror.go

@@ -6,16 +6,31 @@ import (
 	"fmt"
 	"log/slog"
 	"regexp"
+	"slices"
 
 	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/generated/proto_conversions"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 var CustomColumnTypeRegex = regexp.MustCompile(`^$|^[a-zA-Z][a-zA-Z0-9(),]*$`)
 
+type flagConstraint struct {
+	ErrorMessage  string
+	AffectedTypes []types.QValueKind
+}
+
+var FlagConstraints = map[string]flagConstraint{
+	shared.Flag_ClickHouseTime64Enabled: {
+		AffectedTypes: []types.QValueKind{types.QValueKindTime, types.QValueKindTimeTZ},
+		ErrorMessage: "mirror uses time/timetz columns that require ClickHouse setting 'enable_time_time64_type';" +
+			" re-enable it or recreate the mirror",
+	},
+}
+
 func (h *FlowRequestHandler) ValidateCDCMirror(
 	ctx context.Context, req *protos.CreateCDCFlowRequest,
 ) (*protos.ValidateCDCMirrorResponse, APIError) {
@@ -55,6 +70,12 @@ func (h *FlowRequestHandler) validateCDCMirrorImpl(
 		}
 	}
 
+	if connectionConfigs.Resync {
+		if apiErr := h.checkFlagsCompatibility(ctx, connectionConfigs); apiErr != nil {
+			return nil, apiErr
+		}
+	}
+
 	if connectionConfigs == nil {
 		slog.ErrorContext(ctx, "/validatecdc connection configs is nil")
 		return nil, NewInvalidArgumentApiError(errors.New("connection configs is nil"))
@@ -125,3 +146,53 @@ func (h *FlowRequestHandler) checkIfMirrorNameExists(ctx context.Context, mirror
 
 	return nameExists, nil
 }
+
+// checkFlagsCompatibility blocks resync when a destination feature flag that was
+// enabled at mirror creation is now disabled and the stored schema contains
+// column types whose type mapping depends on that flag.
+func (h *FlowRequestHandler) checkFlagsCompatibility(
+	ctx context.Context,
+	cfg *protos.FlowConnectionConfigsCore,
+) APIError {
+	newFlags, err := h.determineFlags(ctx, cfg.Env, cfg.DestinationName)
+	if err != nil {
+		return NewInternalApiError(fmt.Errorf("failed to determine destination flags: %w", err))
+	}
+
+	schemaHasColumnTypes := func(colTypes []types.QValueKind) (bool, error) {
+		tableNames := make([]string, 0, len(cfg.TableMappings))
+		for _, tm := range cfg.TableMappings {
+			tableNames = append(tableNames, tm.DestinationTableIdentifier)
+		}
+		schemas, err := internal.LoadTableSchemasFromCatalog(ctx, h.pool, cfg.FlowJobName, tableNames)
+		if err != nil {
+			return false, err
+		}
+		for _, schema := range schemas {
+			for _, col := range schema.Columns {
+				if slices.Contains(colTypes, types.QValueKind(col.Type)) {
+					return true, nil
+				}
+			}
+		}
+		return false, nil
+	}
+
+	for _, flag := range cfg.Flags {
+		if slices.Contains(newFlags, flag) {
+			continue
+		}
+		constraint, ok := FlagConstraints[flag]
+		if !ok {
+			continue
+		}
+		affected, err := schemaHasColumnTypes(constraint.AffectedTypes)
+		if err != nil {
+			return NewInternalApiError(fmt.Errorf("failed to check schema for flag %q: %w", flag, err))
+		}
+		if affected {
+			return NewFailedPreconditionApiError(errors.New(constraint.ErrorMessage))
+		}
+	}
+	return nil
+}

flow/connectors/bigquery/bigquery.go

@@ -239,6 +239,7 @@ func (c *BigQueryConnector) ReplayTableSchemaDeltas(
 	flowJobName string,
 	_ []*protos.TableMapping,
 	schemaDeltas []*protos.TableSchemaDelta,
+	_ []string,
 ) error {
 	for _, schemaDelta := range schemaDeltas {
 		if schemaDelta == nil || len(schemaDelta.AddedColumns) == 0 {

flow/connectors/bigquery/qrep.go

@@ -87,7 +87,7 @@ func (c *BigQueryConnector) replayTableSchemaDeltasQRep(
 	}
 
 	if err := c.ReplayTableSchemaDeltas(
-		ctx, config.Env, config.FlowJobName, nil, []*protos.TableSchemaDelta{tableSchemaDelta},
+		ctx, config.Env, config.FlowJobName, nil, []*protos.TableSchemaDelta{tableSchemaDelta}, nil,
 	); err != nil {
 		return nil, fmt.Errorf("failed to add columns to destination table: %w", err)
 	}

flow/connectors/bigquery/qrep_avro_sync.go

@@ -98,7 +98,9 @@ func (s *QRepAvroSyncMethod) SyncRecords(
 		slog.String(string(shared.FlowNameKey), req.FlowJobName),
 		slog.String("dstTableName", rawTableName))
 
-	if err := s.connector.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil {
+	if err := s.connector.ReplayTableSchemaDeltas(
+		ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas, nil,
+	); err != nil {
 		return nil, fmt.Errorf("failed to sync schema changes: %w", err)
 	}
 

flow/connectors/clickhouse/avro_sync.go

@@ -14,6 +14,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/internal/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/shared"
@@ -293,11 +294,11 @@ func (s *ClickHouseAvroSyncMethod) pushS3DataToClickHouseForSnapshot(
 	}
 	numParts = max(numParts, 1)
 
-	chSettings := NewCHSettings(s.chVersion)
-	chSettings.Add(SettingThrowOnMaxPartitionsPerInsertBlock, "0")
-	chSettings.Add(SettingTypeJsonSkipDuplicatedPaths, "1")
+	chSettings := clickhouse.NewCHSettings(s.chVersion)
+	chSettings.Add(clickhouse.SettingThrowOnMaxPartitionsPerInsertBlock, "0")
+	chSettings.Add(clickhouse.SettingTypeJsonSkipDuplicatedPaths, "1")
 	if config.Version >= shared.InternalVersion_JsonEscapeDotsInKeys {
-		chSettings.Add(SettingJsonTypeEscapeDotsInKeys, "1")
+		chSettings.Add(clickhouse.SettingJsonTypeEscapeDotsInKeys, "1")
 	}
 
 	// Process each chunk file individually

flow/connectors/clickhouse/cdc.go

@@ -12,6 +12,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	chinternal "github.com/PeerDB-io/peerdb/flow/internal/clickhouse"
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/model/qvalue"
 	peerdb_clickhouse "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse"
@@ -131,7 +132,7 @@ func (c *ClickHouseConnector) syncRecordsViaAvro(
 	}
 	warnings := numericTruncator.Warnings()
 
-	if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas); err != nil {
+	if err := c.ReplayTableSchemaDeltas(ctx, req.Env, req.FlowJobName, req.TableMappings, req.Records.SchemaDeltas, req.Flags); err != nil {
 		return nil, fmt.Errorf("failed to sync schema changes: %w", err)
 	}
 
@@ -165,6 +166,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 	flowJobName string,
 	tableMappings []*protos.TableMapping,
 	schemaDeltas []*protos.TableSchemaDelta,
+	flags []string,
 ) error {
 	if len(schemaDeltas) == 0 {
 		return nil
@@ -188,7 +190,7 @@ func (c *ClickHouseConnector) ReplayTableSchemaDeltas(
 		for _, addedColumn := range schemaDelta.AddedColumns {
 			qvKind := types.QValueKind(addedColumn.Type)
 			clickHouseColType, err := qvalue.ToDWHColumnType(
-				ctx, qvKind, env, protos.DBType_CLICKHOUSE, c.chVersion, addedColumn, schemaDelta.NullableEnabled,
+				ctx, qvKind, env, protos.DBType_CLICKHOUSE, c.chVersion, addedColumn, schemaDelta.NullableEnabled, flags,
 			)
 			if err != nil {
 				return fmt.Errorf("failed to convert column type %s to ClickHouse type: %w", addedColumn.Type, err)
@@ -228,7 +230,8 @@ func (c *ClickHouseConnector) RenameTables(
 	req *protos.RenameTablesInput,
 ) (*protos.RenameTablesOutput, error) {
 	onCluster := c.onCluster()
-	dropTableSQLWithCHSetting := dropTableIfExistsSQL + NewCHSettingsString(c.chVersion, SettingMaxTableSizeToDrop, "0")
+	dropTableSQLWithCHSetting := dropTableIfExistsSQL +
+		chinternal.NewCHSettingsString(c.chVersion, chinternal.SettingMaxTableSizeToDrop, "0")
 	for _, renameRequest := range req.RenameTableOptions {
 		if renameRequest.CurrentName == renameRequest.NewName {
 			c.logger.Info("table rename is nop, probably Null table engine, skipping rename for it",
@@ -302,7 +305,8 @@ func (c *ClickHouseConnector) SyncFlowCleanup(ctx context.Context, jobName strin
 	// delete raw table if exists
 	rawTableIdentifier := c.GetRawTableName(jobName)
 	onCluster := c.onCluster()
-	dropTableSQLWithCHSetting := dropTableIfExistsSQL + NewCHSettingsString(c.chVersion, SettingMaxTableSizeToDrop, "0")
+	dropTableSQLWithCHSetting := dropTableIfExistsSQL +
+		chinternal.NewCHSettingsString(c.chVersion, chinternal.SettingMaxTableSizeToDrop, "0")
 	if err := c.execWithLogging(ctx,
 		fmt.Sprintf(dropTableSQLWithCHSetting, peerdb_clickhouse.QuoteIdentifier(rawTableIdentifier), onCluster),
 	); err != nil {

flow/connectors/clickhouse/clickhouse.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"database/sql"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -406,6 +407,25 @@ func (c *ClickHouseConnector) GetVersion(ctx context.Context) (string, error) {
 	return clickhouseVersion.Version.String(), nil
 }
 
+func (c *ClickHouseConnector) GetFlags(ctx context.Context) ([]string, error) {
+	var flags []string
+
+	var time64Setting string
+	err := c.queryRow(ctx,
+		"SELECT value FROM system.settings WHERE name = 'enable_time_time64_type'",
+	).Scan(&time64Setting)
+	if err != nil {
+		if !errors.Is(err, sql.ErrNoRows) {
+			return nil, fmt.Errorf("failed to query enable_time_time64_type setting: %w", err)
+		}
+	} else if time64Setting == "1" {
+		c.logger.Info("[clickhouse] enable_time_time64_type is enabled")
+		flags = append(flags, shared.Flag_ClickHouseTime64Enabled)
+	}
+
+	return flags, nil
+}
+
 func GetTableSchemaForTable(tm *protos.TableMapping, columns []driver.ColumnType) (*protos.TableSchema, error) {
 	colFields := make([]*protos.FieldDescription, 0, len(columns))
 	for _, column := range columns {
@@ -443,6 +463,8 @@ func GetTableSchemaForTable(tm *protos.TableMapping, columns []driver.ColumnType
 			qkind = types.QValueKindUUID
 		case "DateTime64(6)", "Nullable(DateTime64(6))", "DateTime64(9)", "Nullable(DateTime64(9))":
 			qkind = types.QValueKindTimestamp
+		case "Time64(6)", "Nullable(Time64(6))":
+			qkind = types.QValueKindTime
 		case "Date32", "Nullable(Date32)":
 			qkind = types.QValueKindDate
 		case "Float32", "Nullable(Float32)":