fix(clickhouse): wrap raw table TTL expression in toDateTime for backward compatibility (#4412) (#4425)

### Description

This PR resolves a raw staging table creation crash
(`BAD_TTL_EXPRESSION`) on older ClickHouse versions/clusters
(specifically ClickHouse version <= 24.2).

#### The Problem
In older ClickHouse clusters, the database engine expects `DateTime` or
`Date` columns/expressions inside a table's `TTL` definition. PeerDB's
current staging table implementation uses
`fromUnixTimestamp64Nano(_peerdb_timestamp)` which yields a
high-precision `DateTime64(9)`. This triggers a `BAD_TTL_EXPRESSION`
error, blocking schema replaying and staging.

#### The Fix
This PR wraps `fromUnixTimestamp64Nano(_peerdb_timestamp)` in a
`toDateTime()` cast, which truncates the nanosecond precision to
seconds. Since staging tables are built to retain CDC logs over a course
of days (defaulting to 90 days), this second-level precision has zero
functional impact on data eviction accuracy but ensures compatibility
with older ClickHouse clusters while continuing to work perfectly on
newer versions.

### Changes
* **CDC DDL Generation**: Modified raw table creation in
`flow/connectors/clickhouse/cdc.go` to use
`toDateTime(fromUnixTimestamp64Nano(_peerdb_timestamp))`.
* **CDC Tests**: Updated assertions in
`flow/connectors/clickhouse/cdc_test.go` to match the new `toDateTime`
format.
* **Normalize Query Tests**: Modified
`flow/connectors/clickhouse/normalize_query_test.go` to dynamically
fetch the ClickHouse native test port and host from the environment
helper instead of hardcoding `localhost:9000`.
* **Dev Environment**: Simplified the `env_file` formatting in
`docker-compose-dev.yml` to be backward-compatible with Docker Compose
versions <= v2.13.

### Verification Results
All tests in the ClickHouse connector package pass successfully locally
against the test catalog and ClickHouse instance:
```bash
cd flow
PEERDB_CATALOG_HOST=localhost CI_CLICKHOUSE_HOST=localhost go test -v ./connectors/clickhouse/...

---------

Co-authored-by: Ilia Demianenko <belk94@gmail.com>

Author: jagannalla

flow/connectors/clickhouse/cdc.go

@@ -77,7 +77,7 @@ func (c *ClickHouseConnector) CreateRawTable(ctx context.Context, req *protos.Cr
 		return nil, fmt.Errorf("failed to load raw table TTL days: %w", err)
 	}
 	createRawTableSQL := `CREATE TABLE IF NOT EXISTS %s%s %s ENGINE = %s ORDER BY (_peerdb_batch_id, _peerdb_destination_table_name)` +
-		` TTL fromUnixTimestamp64Nano(_peerdb_timestamp) + INTERVAL ` + strconv.FormatUint(uint64(ttlDays), 10) + ` DAY`
+		` TTL toDateTime(fromUnixTimestamp64Nano(_peerdb_timestamp)) + INTERVAL ` + strconv.FormatUint(uint64(ttlDays), 10) + ` DAY`
 	if err := c.execWithLogging(ctx,
 		fmt.Sprintf(createRawTableSQL, peerdb_clickhouse.QuoteIdentifier(rawTableName), onCluster, rawColumns, engine),
 	); err != nil {

flow/connectors/clickhouse/cdc_test.go

@@ -132,6 +132,7 @@ func TestCreateRawTableHasTTL(t *testing.T) {
 			)).Scan(&engineFull))
 
 			require.Contains(t, engineFull, "TTL", "engine_full should declare a TTL: %s", engineFull)
+			require.Contains(t, engineFull, "toDateTime", "TTL should contain toDateTime cast: %s", engineFull)
 			require.Contains(t, engineFull, "fromUnixTimestamp64Nano(_peerdb_timestamp)",
 				"TTL should reference _peerdb_timestamp: %s", engineFull)
 			require.Contains(t, engineFull, fmt.Sprintf("toIntervalDay(%d)", tc.expectedDays),

flow/connectors/clickhouse/normalize_query_test.go

@@ -9,12 +9,14 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/stretchr/testify/require"
 
+	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
 func TestExtendedTimeToDateTime(t *testing.T) {
 	ctx := context.Background()
-	ch, err := clickhouse.Open(&clickhouse.Options{Addr: []string{"localhost:9000"}})
+	addr := fmt.Sprintf("%s:%d", internal.ClickHouseTestHost(), internal.ClickHouseTestPort())
+	ch, err := clickhouse.Open(&clickhouse.Options{Addr: []string{addr}})
 	if err != nil {
 		t.Skipf("ClickHouse not available: %v", err)
 	}