Update pg connector documentation

Author: ilyario

docs/en/connectors.md

@@ -364,16 +364,19 @@ sink:
 
 ## PostgreSQL
 
-The PostgreSQL connector supports reading from and writing to PostgreSQL tables. It supports custom SQL queries, periodic polling, batch inserts, auto-create tables, and UPSERT mode.
+The PostgreSQL connector supports reading from and writing to PostgreSQL tables. It supports custom SQL queries, periodic polling, batch inserts, auto-create tables, UPSERT mode, CDC-style change tracking (inserts and updates), soft delete, and SecretRef for credentials.
 
 ### Source
 
 ```yaml
 source:
   type: postgresql
   postgresql:
+    # Connection string (required, or use connectionStringSecretRef)
     connectionString: "postgres://user:password@localhost:5432/dbname?sslmode=disable"
+    # Table to read from (required if query not specified). Supports schema.table (e.g. public.products)
     table: source_table
+
     # Custom SQL query (optional)
     query: "SELECT * FROM source_table WHERE updated_at > NOW() - INTERVAL '1 hour'"
     # Poll interval in seconds (optional, default: 5)
@@ -382,14 +385,32 @@ source:
     # Raw mode (optional, default: false)
     # When true, wraps each row as JSON: {"value": <row data>, "_metadata": {table, id}}
     rawMode: true
+
+    # CDC-style options (optional)
+    readBatchSize: 1000           # Limit rows per poll to reduce DB load (0 = no limit)
+    changeTrackingColumn: updated_at  # Column to track changes (default: updated_at). Not used when query is specified
+    autoCreateTable: true         # Create table if it doesn't exist before reading
+
+    # SecretRef (optional) - use instead of direct values
+    # connectionStringSecretRef:
+    #   name: postgres-credentials
+    #   key: connectionString
+    # tableSecretRef:
+    #   name: postgres-credentials
+    #   key: table
 ```
 
-### Features
+### Source Features
 
 - **Periodic Polling**: Regularly polls the table for new data
 - **Custom Queries**: Support for complex SQL with JOIN, WHERE, etc.
-- **Metadata**: Each message contains `table` metadata
+- **Metadata**: Each message contains `table` metadata and `operation` (insert/update)
 - **Raw Mode**: When enabled, wraps each row as JSON with `value` (row data) and `_metadata` (table, id)
+- **Read Batch Size**: Limits rows per poll to reduce database load when many new records appear
+- **Change Tracking**: By default tracks changes via `updated_at` column (or `changeTrackingColumn`), captures both INSERTs and UPDATEs
+- **Auto-create Table**: Creates the table with CDC-friendly schema (id, created_at, updated_at) if it doesn't exist
+- **Schema notation**: Table name supports `schema.table` format (e.g. `public.products`)
+- **In-memory state**: Read position (lastReadChangeTime) is stored only in memory. On pod/connector restart, the table is fully re-read. For pg→pg flows, enable `upsertMode: true` in sink to update duplicates instead of inserting them again.
 
 ### Sink
 
@@ -398,22 +419,38 @@ sink:
   type: postgresql
   postgresql:
     connectionString: "postgres://user:password@localhost:5432/dbname?sslmode=disable"
+    # Table to write to. Supports schema.table (e.g. public.products_clone)
     table: target_table
+
     # Batch size (optional, default: 1). 0 = flush only by timer
     batchSize: 100
     # Flush interval in seconds (optional, default: 10). 0 = disable timer
     batchFlushIntervalSeconds: 10
     autoCreateTable: true
+
+    # Raw mode (optional, default: false)
+    # When true, expects messages in format {"value": <data>, "_metadata": {...}}
+    # Table is created with value JSONB and _metadata JSONB columns
+    rawMode: false
+
     # UPSERT mode (optional, default: false)
     upsertMode: true
     conflictKey: "id"
+    # Soft delete column (optional). When set, DELETE operations UPDATE this column instead of physical delete
+    softDeleteColumn: "deleted_at"
+
+    # SecretRef (optional)
+    # connectionStringSecretRef: { name: postgres-credentials, key: connectionString }
+    # tableSecretRef: { name: postgres-credentials, key: table }
 ```
 
-### Features
+### Sink Features
 
-- **Batch Inserts**: Groups messages for efficient writing
-- **Auto-create Tables**: Creates tables with JSONB field and GIN index
+- **Batch Inserts**: Groups messages for efficient writing. Flush when `batchSize` reached or on timer. Use `batchFlushIntervalSeconds: 0` for size-only flush; `batchSize: 0` for timer-only flush.
+- **Auto-create Tables**: When `rawMode: true` — creates table with `value` JSONB, `_metadata` JSONB, created_at, updated_at, deleted_at and GIN index. When `rawMode: false` — infers table structure from the first message (replicates source schema).
+- **Raw Mode**: When true, expects `{"value": <data>, "_metadata": {...}}` and creates table with value/_metadata columns. When false, uses columnar format from message.
 - **UPSERT Mode**: Updates existing records on conflict (PRIMARY KEY or `conflictKey`)
+- **Soft Delete**: When `softDeleteColumn` is set and message has `metadata.operation=delete`, performs `UPDATE ... SET deleted_at = NOW()` instead of physical DELETE
 
 ## ClickHouse
 

docs/ru/connectors.md

@@ -210,7 +210,7 @@ sink:
 
 ## PostgreSQL
 
-PostgreSQL коннектор поддерживает чтение из таблиц и запись в таблицы PostgreSQL. Поддерживает кастомные SQL запросы, периодический опрос и батч-вставки.
+PostgreSQL коннектор поддерживает чтение из таблиц и запись в таблицы PostgreSQL. Поддерживает кастомные SQL запросы, периодический опрос, батч-вставки, CDC-стиль отслеживания изменений (INSERT и UPDATE), soft delete и SecretRef для учётных данных.
 
 ### Источник (Source)
 
@@ -220,34 +220,45 @@ PostgreSQL коннектор поддерживает чтение из таб
 source:
   type: postgresql
   postgresql:
-    # Connection string (обязательно)
+    # Connection string (обязательно, или connectionStringSecretRef)
     # Формат: postgres://user:password@host:port/dbname?sslmode=mode
     connectionString: "postgres://user:password@localhost:5432/dbname?sslmode=disable"
 
-    # Таблица для чтения (обязательно, если не указан query)
+    # Таблица для чтения (обязательно, если не указан query). Поддерживает schema.table (напр. public.products)
     table: source_table
 
     # Кастомный SQL запрос (опционально)
     # Если указан, используется вместо чтения всей таблицы
     query: "SELECT * FROM source_table WHERE updated_at > NOW() - INTERVAL '1 hour'"
 
     # Интервал опроса в секундах (опционально, по умолчанию: 5)
-    # Используется для периодического чтения новых данных
     pollInterval: 60
 
     # Режим сырой записи (опционально, по умолчанию: false)
     # При true каждая строка оборачивается в JSON: {"value": <данные строки>, "_metadata": {table, id}}
     rawMode: true
+
+    # Опции CDC-стиля (опционально)
+    readBatchSize: 1000           # Ограничение строк за один опрос (0 = без лимита)
+    changeTrackingColumn: updated_at  # Колонка для отслеживания изменений (по умолчанию: updated_at). Не используется при заданном query
+    autoCreateTable: true         # Создать таблицу, если не существует, перед чтением
+
+    # SecretRef (опционально)
+    # connectionStringSecretRef: { name: postgres-credentials, key: connectionString }
+    # tableSecretRef: { name: postgres-credentials, key: table }
 ```
 
 #### Особенности PostgreSQL источника
 
 - **Периодический опрос**: Регулярно опрашивает таблицу на наличие новых данных
 - **Кастомные запросы**: Поддержка сложных SQL запросов с JOIN, WHERE, и т.д.
-- **Метаданные**: Каждое сообщение содержит метаданные:
-  - `table` - название таблицы
-- **Формат данных**: Данные преобразуются в JSON формат
-- **Режим сырой записи (rawMode)**: При включении каждая строка оборачивается в JSON с полями `value` (данные строки) и `_metadata` (table, id)
+- **Метаданные**: Каждое сообщение содержит метаданные `table` и `operation` (insert/update)
+- **Режим rawMode**: При включении каждая строка оборачивается в JSON с полями `value` (данные строки) и `_metadata` (table, id)
+- **Размер батча чтения**: Ограничивает количество строк за один опрос для снижения нагрузки на БД
+- **Отслеживание обновлений**: По умолчанию отслеживает изменения по колонке `updated_at` (или `changeTrackingColumn`), захватывает INSERT и UPDATE
+- **Автосоздание таблицы**: Создаёт таблицу с CDC-совместимой схемой (id, created_at, updated_at), если не существует
+- **Схема таблицы**: Имя таблицы поддерживает формат `schema.table` (напр. `public.products`)
+- **Состояние в памяти**: Позиция чтения (lastReadChangeTime) хранится только в памяти. При перезапуске пода/коннектора происходит полное перечитывание таблицы. Для pg→pg включите `upsertMode: true` в sink, чтобы дубликаты обновлялись, а не вставлялись повторно.
 
 #### Пример с кастомным запросом
 
@@ -276,66 +287,69 @@ source:
 sink:
   type: postgresql
   postgresql:
-    # Connection string (обязательно)
     connectionString: "postgres://user:password@localhost:5432/dbname?sslmode=disable"
-
-    # Таблица для записи (обязательно)
+    # Таблица для записи. Поддерживает schema.table (напр. public.products_clone)
     table: target_table
 
-    # Размер батча для вставки (опционально, по умолчанию: 1)
-    # Увеличьте для повышения производительности. 0 — сброс только по таймеру
+    # Размер батча (опционально, по умолчанию: 1). 0 — сброс только по таймеру
     batchSize: 100
-
-    # Интервал сброса батча в секундах (опционально, по умолчанию: 10)
-    # Батч сбрасывается при достижении batchSize или по истечении интервала. 0 — отключить сброс по таймеру
+    # Интервал сброса в секундах (опционально, по умолчанию: 10). 0 — отключить сброс по таймеру
     batchFlushIntervalSeconds: 10
-
-    # Автоматическое создание таблицы (опционально, по умолчанию: false)
-    # Если true, создает таблицу с JSONB полем для гибкой схемы
     autoCreateTable: true
 
+    # Режим rawMode (опционально, по умолчанию: false)
+    # При true ожидает сообщения в формате {"value": <данные>, "_metadata": {...}}
+    # Таблица создаётся с колонками value JSONB и _metadata JSONB
+    rawMode: false
+
     # Режим UPSERT (опционально, по умолчанию: false)
-    # Если true, существующие записи будут обновляться вместо пропуска
-    # Используется ON CONFLICT DO UPDATE для обновления записей
     upsertMode: true
-
-    # Ключ конфликта для UPSERT (опционально, по умолчанию: PRIMARY KEY)
-    # Указывает колонку(ы) для определения конфликта при UPSERT
-    # Если не указан, используется PRIMARY KEY таблицы
     conflictKey: "id"
+    # Колонка для soft delete (опционально)
+    softDeleteColumn: "deleted_at"
+
+    # SecretRef (опционально)
+    # connectionStringSecretRef: { name: postgres-credentials, key: connectionString }
+    # tableSecretRef: { name: postgres-credentials, key: table }
 ```
 
 #### Особенности PostgreSQL приемника
 
-- **Батч-вставки**: Группирует сообщения для эффективной записи. По умолчанию сброс при достижении `batchSize` или по таймеру (10 с). Только по размеру: `batchFlushIntervalSeconds: 0`. Только по времени: `batchSize: 0`
-- **Автосоздание таблиц**: Автоматически создает таблицы с JSONB полем
-- **Гибкая схема**: Поддерживает как JSONB (для автосозданных таблиц), так и колоночный формат
-- **Индексы**: Автоматически создает GIN индекс на JSONB поле для быстрого поиска
-- **UPSERT режим**: Поддерживает обновление существующих записей при конфликте по PRIMARY KEY или указанному ключу
+- **Батч-вставки**: Группирует сообщения для эффективной записи. Сброс при достижении `batchSize` или по таймеру. Только по размеру: `batchFlushIntervalSeconds: 0`. Только по времени: `batchSize: 0`
+- **Автосоздание таблиц**: При `rawMode: true` — создаёт таблицу с колонками `value` JSONB, `_metadata` JSONB, created_at, updated_at, deleted_at и GIN индексом. При `rawMode: false` — выводит структуру таблицы из первого сообщения (реплицирует схему источника)
+- **Режим rawMode**: При true ожидает `{"value": <данные>, "_metadata": {...}}` и создаёт таблицу с колонками value/_metadata. При false использует колоночный формат из сообщения
+- **UPSERT режим**: Обновляет существующие записи при конфликте по PRIMARY KEY или указанному `conflictKey`
+- **Soft delete**: При заданном `softDeleteColumn` и `metadata.operation=delete` выполняет `UPDATE ... SET deleted_at = NOW()` вместо физического DELETE
 
-#### Пример с автосозданием таблицы
+#### Пример с автосозданием таблицы (rawMode)
 
 ```yaml
 sink:
   type: postgresql
   postgresql:
     connectionString: "postgres://user:password@localhost:5432/analytics"
     table: events
-    autoCreateTable: true  # Создаст таблицу с JSONB полем
+    autoCreateTable: true
+    rawMode: true  # Таблица с value и _metadata
     batchSize: 50
 ```
 
-Автоматически созданная таблица будет иметь структуру:
+При `rawMode: true` и `autoCreateTable: true` таблица создаётся со структурой:
 
 ```sql
 CREATE TABLE events (
     id SERIAL PRIMARY KEY,
-    data JSONB NOT NULL,
-    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    value JSONB NOT NULL,
+    _metadata JSONB,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    deleted_at TIMESTAMP
 );
-CREATE INDEX idx_events_data ON events USING GIN (data);
+CREATE INDEX idx_events_value ON events USING GIN (value);
 ```
 
+При `rawMode: false` структура таблицы выводится из первого сообщения (колоночный формат).
+
 #### Пример с UPSERT для обновления существующих записей
 
 ```yaml