From be2ad195c31e1aa3d28e09dd9f652a386ba0d8f8 Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Thu, 16 Jul 2026 17:29:22 +0100 Subject: [PATCH 1/9] feat: extract Postgres source and load to Iceberg Implement the elt-pipeline proposal to extract data from the PostgreSQL database source and load it into the Iceberg destination. --- .../fase/ingest/fase/proposal/proposal.py | 57 +++++++++++++ .../fase/ingest/fase/utils/postgres.py | 84 +++++++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 elt-pipelines/fase/ingest/fase/proposal/proposal.py create mode 100644 elt-pipelines/fase/ingest/fase/utils/postgres.py diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py new file mode 100644 index 00000000..b856a6ae --- /dev/null +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -0,0 +1,57 @@ +import logging +from pydantic_settings import BaseSettings, SettingsConfigDict + +from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties +from fase.utils.postgres import PostgresExtractor + +LOGGER = logging.getLogger(__name__) + +class PipelinePostgresConfig(BaseSettings): + model_config = SettingsConfigDict( + env_prefix="proposal__", + env_nested_delimiter="__", + extra="ignore", + protected_namespaces=() + ) + + drivername: str = "postgresql+psycopg2" + host: str + port: int + username: str + password: str + database: str + table: str + + @property + def target_tables(self) -> list[str]: + """Splits the raw table string by commas and strips accidental whitespaces.""" + return [t.strip() for t in self.table.split(",") if t.strip()] + + @property + def connection_uri(self) -> str: + driver = self.drivername.strip() + user = self.username.strip() + pwd = self.password.strip() + h = self.host.strip() + db = self.database.strip() + return f"{driver}://{user}:{pwd}@{h}:{self.port}/{db}" + + +class Extract(BaseExtract): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + # Instantiate pipeline config and pass it directly to the generic utility extractor + self.postgres_config = PipelinePostgresConfig() + self.extractor = PostgresExtractor(self.postgres_config) + + def extract_resource_properties(self): + for table_name in self.postgres_config.target_tables: + # Encapsulate extraction stream generator mapping + yield ( + table_name, + ResourceProperties( + extractor=lambda _, t=table_name: self.extractor.fetch_as_arrow(t), + write_properties=ResourceWriteProperties(write_mode="replace"), + watermark_column=None + ) + ) \ No newline at end of file diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py new file mode 100644 index 00000000..26bedfbf --- /dev/null +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -0,0 +1,84 @@ +import json +import logging +import pandas as pd +import pyarrow as pa +from sqlalchemy import create_engine, inspect, select, Table, MetaData + +LOGGER = logging.getLogger(__name__) + +class PostgresExtractor: + def __init__(self, config): + """Accepts any valid configuration object exposing connection_uri.""" + self.config = config + self.engine = create_engine(config.connection_uri) + self.metadata = MetaData() + + def map_pg_to_pq_type(self, pg_type) -> str: + t = str(pg_type).lower() + if 'int' in t: return 'bigint' + if 'bool' in t: return 'bool' + if 'json' in t or 'uuid' in t: return 'text' + if 'float' in t or 'numeric' in t or 'double' in t: return 'double' + if 'timestamp' in t: return 'timestamp' + if 'date' in t: return 'date' + return 'text' + + def get_table_schema(self, table_name: str) -> dict: + inspector = inspect(self.engine) + columns = inspector.get_columns(table_name) + return {col['name']: self.map_pg_to_pq_type(col['type']) for col in columns} + + def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): + """Streams database data into structured PyArrow tables safely, supporting empty tables.""" + table = Table(table_name, self.metadata, autoload_with=self.engine) + col_hints = self.get_table_schema(table_name) + + arrow_fields = [] + for col_name in col_hints.keys(): + pq_type_str = col_hints[col_name] + if pq_type_str == 'bigint': pa_type = pa.int64() + elif pq_type_str == 'bool': pa_type = pa.bool_() + elif pq_type_str == 'double': pa_type = pa.float64() + elif pq_type_str == 'timestamp': pa_type = pa.timestamp('us') + elif pq_type_str == 'date': pa_type = pa.date32() + else: pa_type = pa.string() + arrow_fields.append(pa.field(col_name, pa_type, nullable=True)) + + target_schema = pa.schema(arrow_fields) + + with self.engine.connect() as conn: + result_proxy = conn.execution_options(stream_results=True).execute(select(table)) + + has_data = False + while True: + chunk = result_proxy.fetchmany(chunk_size) + if not chunk: + break + + has_data = True + df = pd.DataFrame(chunk, columns=result_proxy.keys()) + + for col in df.columns: + if df[col].dtype == 'object': + df[col] = df[col].apply( + lambda x: json.dumps(x) if isinstance(x, (dict, list)) + else str(x) if pd.notnull(x) and not isinstance(x, str) \ + else x + ) + + arrow_table = pa.Table.from_pandas(df) + + aligned_columns = [] + for field in target_schema: + if field.name in arrow_table.column_names: + aligned_columns.append(arrow_table.column(field.name).cast(field.type)) + else: + # Fallback for missing columns + aligned_columns.append(pa.array([None] * len(arrow_table), type=field.type)) + + yield pa.Table.from_arrays(aligned_columns, schema=target_schema) + + if not has_data: + dummy_row = {col_key: [None] for col_key in result_proxy.keys()} + empty_df = pd.DataFrame(dummy_row) + yield pa.Table.from_pandas(empty_df, schema=target_schema) \ No newline at end of file From 62abf0d9322a685b52826a077284ef6ed2aab4f2 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:50:12 +0000 Subject: [PATCH 2/9] fix: apply CodeRabbit auto-fixes Fixed 2 file(s) based on 2 unresolved review comments. Co-authored-by: CodeRabbit --- elt-pipelines/fase/ingest/fase/proposal/proposal.py | 5 +++-- elt-pipelines/fase/ingest/fase/utils/postgres.py | 4 +--- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py index b856a6ae..f56cd2f9 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -1,4 +1,5 @@ import logging +from urllib.parse import quote_plus from pydantic_settings import BaseSettings, SettingsConfigDict from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties @@ -30,8 +31,8 @@ def target_tables(self) -> list[str]: @property def connection_uri(self) -> str: driver = self.drivername.strip() - user = self.username.strip() - pwd = self.password.strip() + user = quote_plus(self.username.strip()) + pwd = quote_plus(self.password.strip()) h = self.host.strip() db = self.database.strip() return f"{driver}://{user}:{pwd}@{h}:{self.port}/{db}" diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py index 26bedfbf..eb60588a 100644 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -79,6 +79,4 @@ def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): yield pa.Table.from_arrays(aligned_columns, schema=target_schema) if not has_data: - dummy_row = {col_key: [None] for col_key in result_proxy.keys()} - empty_df = pd.DataFrame(dummy_row) - yield pa.Table.from_pandas(empty_df, schema=target_schema) \ No newline at end of file + yield pa.table({}, schema=target_schema) \ No newline at end of file From d85e678138df60acda110c5541bcb0f0f61b8801 Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Fri, 17 Jul 2026 18:59:23 +0100 Subject: [PATCH 3/9] Revert "fix: apply CodeRabbit auto-fixes" This reverts commit 62abf0d9322a685b52826a077284ef6ed2aab4f2. --- elt-pipelines/fase/ingest/fase/proposal/proposal.py | 5 ++--- elt-pipelines/fase/ingest/fase/utils/postgres.py | 4 +++- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py index f56cd2f9..b856a6ae 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -1,5 +1,4 @@ import logging -from urllib.parse import quote_plus from pydantic_settings import BaseSettings, SettingsConfigDict from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties @@ -31,8 +30,8 @@ def target_tables(self) -> list[str]: @property def connection_uri(self) -> str: driver = self.drivername.strip() - user = quote_plus(self.username.strip()) - pwd = quote_plus(self.password.strip()) + user = self.username.strip() + pwd = self.password.strip() h = self.host.strip() db = self.database.strip() return f"{driver}://{user}:{pwd}@{h}:{self.port}/{db}" diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py index eb60588a..26bedfbf 100644 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -79,4 +79,6 @@ def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): yield pa.Table.from_arrays(aligned_columns, schema=target_schema) if not has_data: - yield pa.table({}, schema=target_schema) \ No newline at end of file + dummy_row = {col_key: [None] for col_key in result_proxy.keys()} + empty_df = pd.DataFrame(dummy_row) + yield pa.Table.from_pandas(empty_df, schema=target_schema) \ No newline at end of file From 92d9eefa8e2793e0014799cfdd99f150cc5f2851 Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Fri, 17 Jul 2026 19:12:17 +0100 Subject: [PATCH 4/9] style: fix trailing whitespace and ruff E701 lint errors --- .../fase/ingest/fase/proposal/proposal.py | 13 +-- .../fase/ingest/fase/utils/postgres.py | 83 ++++++++++++------- 2 files changed, 59 insertions(+), 37 deletions(-) diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py index b856a6ae..bd72b8be 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -6,12 +6,13 @@ LOGGER = logging.getLogger(__name__) + class PipelinePostgresConfig(BaseSettings): model_config = SettingsConfigDict( - env_prefix="proposal__", + env_prefix="proposal__", env_nested_delimiter="__", extra="ignore", - protected_namespaces=() + protected_namespaces=(), ) drivername: str = "postgresql+psycopg2" @@ -20,7 +21,7 @@ class PipelinePostgresConfig(BaseSettings): username: str password: str database: str - table: str + table: str @property def target_tables(self) -> list[str]: @@ -52,6 +53,6 @@ def extract_resource_properties(self): ResourceProperties( extractor=lambda _, t=table_name: self.extractor.fetch_as_arrow(t), write_properties=ResourceWriteProperties(write_mode="replace"), - watermark_column=None - ) - ) \ No newline at end of file + watermark_column=None, + ), + ) diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py index 26bedfbf..f526d699 100644 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -6,79 +6,100 @@ LOGGER = logging.getLogger(__name__) + class PostgresExtractor: def __init__(self, config): """Accepts any valid configuration object exposing connection_uri.""" self.config = config self.engine = create_engine(config.connection_uri) - self.metadata = MetaData() - + self.metadata = MetaData() + def map_pg_to_pq_type(self, pg_type) -> str: t = str(pg_type).lower() - if 'int' in t: return 'bigint' - if 'bool' in t: return 'bool' - if 'json' in t or 'uuid' in t: return 'text' - if 'float' in t or 'numeric' in t or 'double' in t: return 'double' - if 'timestamp' in t: return 'timestamp' - if 'date' in t: return 'date' - return 'text' + if "int" in t: + return "bigint" + if "bool" in t: + return "bool" + if "json" in t or "uuid" in t: + return "text" + if "float" in t or "numeric" in t or "double" in t: + return "double" + if "timestamp" in t: + return "timestamp" + if "date" in t: + return "date" + return "text" def get_table_schema(self, table_name: str) -> dict: inspector = inspect(self.engine) columns = inspector.get_columns(table_name) - return {col['name']: self.map_pg_to_pq_type(col['type']) for col in columns} + return {col["name"]: self.map_pg_to_pq_type(col["type"]) for col in columns} def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): """Streams database data into structured PyArrow tables safely, supporting empty tables.""" table = Table(table_name, self.metadata, autoload_with=self.engine) col_hints = self.get_table_schema(table_name) - + arrow_fields = [] for col_name in col_hints.keys(): pq_type_str = col_hints[col_name] - if pq_type_str == 'bigint': pa_type = pa.int64() - elif pq_type_str == 'bool': pa_type = pa.bool_() - elif pq_type_str == 'double': pa_type = pa.float64() - elif pq_type_str == 'timestamp': pa_type = pa.timestamp('us') - elif pq_type_str == 'date': pa_type = pa.date32() - else: pa_type = pa.string() + if pq_type_str == "bigint": + pa_type = pa.int64() + elif pq_type_str == "bool": + pa_type = pa.bool_() + elif pq_type_str == "double": + pa_type = pa.float64() + elif pq_type_str == "timestamp": + pa_type = pa.timestamp("us") + elif pq_type_str == "date": + pa_type = pa.date32() + else: + pa_type = pa.string() arrow_fields.append(pa.field(col_name, pa_type, nullable=True)) - + target_schema = pa.schema(arrow_fields) - + with self.engine.connect() as conn: - result_proxy = conn.execution_options(stream_results=True).execute(select(table)) - + result_proxy = conn.execution_options(stream_results=True).execute( + select(table) + ) + has_data = False while True: chunk = result_proxy.fetchmany(chunk_size) if not chunk: break - + has_data = True df = pd.DataFrame(chunk, columns=result_proxy.keys()) - + for col in df.columns: - if df[col].dtype == 'object': + if df[col].dtype == "object": df[col] = df[col].apply( - lambda x: json.dumps(x) if isinstance(x, (dict, list)) - else str(x) if pd.notnull(x) and not isinstance(x, str) \ + lambda x: json.dumps(x) + if isinstance(x, (dict, list)) + else str(x) + if pd.notnull(x) and not isinstance(x, str) else x ) arrow_table = pa.Table.from_pandas(df) - + aligned_columns = [] for field in target_schema: if field.name in arrow_table.column_names: - aligned_columns.append(arrow_table.column(field.name).cast(field.type)) + aligned_columns.append( + arrow_table.column(field.name).cast(field.type) + ) else: # Fallback for missing columns - aligned_columns.append(pa.array([None] * len(arrow_table), type=field.type)) - + aligned_columns.append( + pa.array([None] * len(arrow_table), type=field.type) + ) + yield pa.Table.from_arrays(aligned_columns, schema=target_schema) if not has_data: dummy_row = {col_key: [None] for col_key in result_proxy.keys()} empty_df = pd.DataFrame(dummy_row) - yield pa.Table.from_pandas(empty_df, schema=target_schema) \ No newline at end of file + yield pa.Table.from_pandas(empty_df, schema=target_schema) From 46e74cb5bcacd598787a9e9e34ce2d05a05daea2 Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Tue, 21 Jul 2026 11:54:22 +0100 Subject: [PATCH 5/9] Refactoring with sqldatabase package --- .../fase/ingest/fase/proposal/proposal.py | 64 +++------ .../fase/ingest/fase/utils/postgres.py | 126 ++++++++++-------- 2 files changed, 88 insertions(+), 102 deletions(-) diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py index bd72b8be..f6c70027 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -1,58 +1,32 @@ -import logging -from pydantic_settings import BaseSettings, SettingsConfigDict +from elt_common.extract import ResourceWriteProperties +from elt_common.sources.sqldatabase import SqlDatabaseSourceConfig, TableInfo +from fase.utils.postgres import PostgresExtract -from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties -from fase.utils.postgres import PostgresExtractor -LOGGER = logging.getLogger(__name__) - - -class PipelinePostgresConfig(BaseSettings): - model_config = SettingsConfigDict( - env_prefix="proposal__", - env_nested_delimiter="__", - extra="ignore", - protected_namespaces=(), - ) +class PipelinePostgresConfig(SqlDatabaseSourceConfig): + model_config = { + "env_prefix": "proposal__", + "env_nested_delimiter": "__", + "extra": "ignore", + "protected_namespaces": (), + } drivername: str = "postgresql+psycopg2" - host: str - port: int - username: str - password: str - database: str table: str @property def target_tables(self) -> list[str]: - """Splits the raw table string by commas and strips accidental whitespaces.""" return [t.strip() for t in self.table.split(",") if t.strip()] - @property - def connection_uri(self) -> str: - driver = self.drivername.strip() - user = self.username.strip() - pwd = self.password.strip() - h = self.host.strip() - db = self.database.strip() - return f"{driver}://{user}:{pwd}@{h}:{self.port}/{db}" - -class Extract(BaseExtract): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - # Instantiate pipeline config and pass it directly to the generic utility extractor - self.postgres_config = PipelinePostgresConfig() - self.extractor = PostgresExtractor(self.postgres_config) +class Extract(PostgresExtract): + config_cls = PipelinePostgresConfig - def extract_resource_properties(self): - for table_name in self.postgres_config.target_tables: - # Encapsulate extraction stream generator mapping - yield ( - table_name, - ResourceProperties( - extractor=lambda _, t=table_name: self.extractor.fetch_as_arrow(t), - write_properties=ResourceWriteProperties(write_mode="replace"), - watermark_column=None, - ), + def table_info(self) -> dict[str, TableInfo]: + """Defines the target tables and their ingestion strategy.""" + return { + table_name: TableInfo( + write_properties=ResourceWriteProperties(write_mode="replace") ) + for table_name in self.config.target_tables + } diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py index f526d699..4027187b 100644 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -1,19 +1,17 @@ import json import logging +from typing import Iterator + import pandas as pd import pyarrow as pa -from sqlalchemy import create_engine, inspect, select, Table, MetaData +import sqlalchemy as sa +from elt_common.extract import Watermark +from elt_common.sources.sqldatabase import SqlDatabaseExtract LOGGER = logging.getLogger(__name__) -class PostgresExtractor: - def __init__(self, config): - """Accepts any valid configuration object exposing connection_uri.""" - self.config = config - self.engine = create_engine(config.connection_uri) - self.metadata = MetaData() - +class PostgresExtract(SqlDatabaseExtract): def map_pg_to_pq_type(self, pg_type) -> str: t = str(pg_type).lower() if "int" in t: @@ -30,19 +28,13 @@ def map_pg_to_pq_type(self, pg_type) -> str: return "date" return "text" - def get_table_schema(self, table_name: str) -> dict: - inspector = inspect(self.engine) + def get_table_schema(self, table_name: str) -> pa.Schema: + inspector = sa.inspect(self._engine) columns = inspector.get_columns(table_name) - return {col["name"]: self.map_pg_to_pq_type(col["type"]) for col in columns} - - def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): - """Streams database data into structured PyArrow tables safely, supporting empty tables.""" - table = Table(table_name, self.metadata, autoload_with=self.engine) - col_hints = self.get_table_schema(table_name) arrow_fields = [] - for col_name in col_hints.keys(): - pq_type_str = col_hints[col_name] + for col in columns: + pq_type_str = self.map_pg_to_pq_type(col["type"]) if pq_type_str == "bigint": pa_type = pa.int64() elif pq_type_str == "bool": @@ -55,51 +47,71 @@ def fetch_as_arrow(self, table_name: str, chunk_size: int = 50000): pa_type = pa.date32() else: pa_type = pa.string() - arrow_fields.append(pa.field(col_name, pa_type, nullable=True)) - - target_schema = pa.schema(arrow_fields) - - with self.engine.connect() as conn: - result_proxy = conn.execution_options(stream_results=True).execute( - select(table) + arrow_fields.append(pa.field(col["name"], pa_type, nullable=True)) + + return pa.schema(arrow_fields) + + def _extract_table( + self, + name: str, + *, + conn: sa.Connection, + watermark: Watermark | None = None, + ) -> Iterator[pa.Table]: + LOGGER.debug( + f"Extracting Postgres table {name} in chunks of {self._chunk_size} rows." + ) + + target_schema = self.get_table_schema(name) + table = sa.Table(name, self._metadata, autoload_with=self._engine) + + query = sa.select(table) + if watermark is not None: + column, max_value = watermark.column, watermark.value + LOGGER.debug( + f"Cursor value detected. Limiting query to {column} > {max_value}" ) + query = query.where(sa.column(column) > max_value) + + result = conn.execution_options(yield_per=self._chunk_size).execute(query) - has_data = False - while True: - chunk = result_proxy.fetchmany(chunk_size) - if not chunk: - break + has_data = False + while True: + chunk = result.fetchmany(self._chunk_size) + if not chunk: + break - has_data = True - df = pd.DataFrame(chunk, columns=result_proxy.keys()) + has_data = True + df = pd.DataFrame(chunk, columns=result.keys()) - for col in df.columns: - if df[col].dtype == "object": - df[col] = df[col].apply( - lambda x: json.dumps(x) + for col in df.columns: + if df[col].dtype == "object": + df[col] = df[col].apply( + lambda x: ( + json.dumps(x) if isinstance(x, (dict, list)) else str(x) if pd.notnull(x) and not isinstance(x, str) else x ) - - arrow_table = pa.Table.from_pandas(df) - - aligned_columns = [] - for field in target_schema: - if field.name in arrow_table.column_names: - aligned_columns.append( - arrow_table.column(field.name).cast(field.type) - ) - else: - # Fallback for missing columns - aligned_columns.append( - pa.array([None] * len(arrow_table), type=field.type) - ) - - yield pa.Table.from_arrays(aligned_columns, schema=target_schema) - - if not has_data: - dummy_row = {col_key: [None] for col_key in result_proxy.keys()} - empty_df = pd.DataFrame(dummy_row) - yield pa.Table.from_pandas(empty_df, schema=target_schema) + ) + + arrow_table = pa.Table.from_pandas(df) + + aligned_columns = [] + for field in target_schema: + if field.name in arrow_table.column_names: + aligned_columns.append( + arrow_table.column(field.name).cast(field.type) + ) + else: + aligned_columns.append( + pa.array([None] * len(arrow_table), type=field.type) + ) + + yield pa.Table.from_arrays(aligned_columns, schema=target_schema) + + if not has_data: + dummy_row = {col_key: [None] for col_key in result.keys()} + empty_df = pd.DataFrame(dummy_row) + yield pa.Table.from_pandas(empty_df, schema=target_schema) From 364213fb0cd456ac514c8a1ade5a39121ffe3a7d Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Tue, 21 Jul 2026 12:23:14 +0100 Subject: [PATCH 6/9] Refactoring for Pydantic settings supports --- .../fase/ingest/fase/utils/postgres.py | 79 ++++++++++++------- 1 file changed, 49 insertions(+), 30 deletions(-) diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py index 4027187b..5bdf5b30 100644 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ b/elt-pipelines/fase/ingest/fase/utils/postgres.py @@ -1,53 +1,72 @@ import json import logging -from typing import Iterator +from typing import Dict, Iterator import pandas as pd import pyarrow as pa import sqlalchemy as sa from elt_common.extract import Watermark -from elt_common.sources.sqldatabase import SqlDatabaseExtract +from elt_common.sources.sqldatabase import SqlDatabaseExtract, SqlDatabaseSourceConfig +from pydantic_settings import SettingsConfigDict LOGGER = logging.getLogger(__name__) +# Default PyArrow type lookup dictionary +DEFAULT_TYPE_MAP: Dict[str, str] = { + "int": "bigint", + "bool": "bool", + "json": "text", + "uuid": "text", + "float": "double", + "numeric": "double", + "double": "double", + "timestamp": "timestamp", + "date": "date", +} + +# Mapping string representations to actual PyArrow types +PA_TYPE_MAPPING = { + "bigint": pa.int64(), + "bool": pa.bool_(), + "double": pa.float64(), + "timestamp": pa.timestamp("us"), + "date": pa.date32(), + "text": pa.string(), +} + + +class PostgresConfig(SqlDatabaseSourceConfig): + model_config = SettingsConfigDict( + extra="ignore", + protected_namespaces=(), + ) + + type_map: Dict[str, str] = DEFAULT_TYPE_MAP + class PostgresExtract(SqlDatabaseExtract): + config_cls = PostgresConfig + def map_pg_to_pq_type(self, pg_type) -> str: t = str(pg_type).lower() - if "int" in t: - return "bigint" - if "bool" in t: - return "bool" - if "json" in t or "uuid" in t: - return "text" - if "float" in t or "numeric" in t or "double" in t: - return "double" - if "timestamp" in t: - return "timestamp" - if "date" in t: - return "date" + type_map = getattr(self.config, "type_map", DEFAULT_TYPE_MAP) + for keyword, mapped_type in type_map.items(): + if keyword in t: + return mapped_type return "text" def get_table_schema(self, table_name: str) -> pa.Schema: inspector = sa.inspect(self._engine) columns = inspector.get_columns(table_name) - arrow_fields = [] - for col in columns: - pq_type_str = self.map_pg_to_pq_type(col["type"]) - if pq_type_str == "bigint": - pa_type = pa.int64() - elif pq_type_str == "bool": - pa_type = pa.bool_() - elif pq_type_str == "double": - pa_type = pa.float64() - elif pq_type_str == "timestamp": - pa_type = pa.timestamp("us") - elif pq_type_str == "date": - pa_type = pa.date32() - else: - pa_type = pa.string() - arrow_fields.append(pa.field(col["name"], pa_type, nullable=True)) + arrow_fields = [ + pa.field( + col["name"], + PA_TYPE_MAPPING.get(self.map_pg_to_pq_type(col["type"]), pa.string()), + nullable=True, + ) + for col in columns + ] return pa.schema(arrow_fields) From 42ac2a2f97e603eb4ba6cf00189aba196287eb8b Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Fri, 24 Jul 2026 15:19:50 +0100 Subject: [PATCH 7/9] Add dependency on psycopg (using psycopg3) Add set up for a fase_landing warehouse in the local infrastructure Centralize loading pg and oracle by using sqlalchemy at sqldatabase/__init__.py --- .../sources/sqldatabase/__init__.py | 67 ++++++++- .../fase/ingest/fase/proposal/proposal.py | 21 ++- .../fase/ingest/fase/utils/postgres.py | 136 ------------------ elt-pipelines/pyproject.toml | 6 +- .../warehouses/lakekeeper/fase_landing.json | 27 ++++ 5 files changed, 106 insertions(+), 151 deletions(-) delete mode 100644 elt-pipelines/fase/ingest/fase/utils/postgres.py create mode 100644 infra/local/warehouses/lakekeeper/fase_landing.json diff --git a/elt-common/src/elt_common/sources/sqldatabase/__init__.py b/elt-common/src/elt_common/sources/sqldatabase/__init__.py index ead963e5..73dc808d 100644 --- a/elt-common/src/elt_common/sources/sqldatabase/__init__.py +++ b/elt-common/src/elt_common/sources/sqldatabase/__init__.py @@ -1,5 +1,6 @@ """Support for ingesting data from an SQL database.""" +import json import logging from abc import abstractmethod from typing import Generator, Iterator, NamedTuple, Optional @@ -9,7 +10,7 @@ from pydantic import SecretStr from pydantic_settings import BaseSettings -from elt_common.extract import ResourceProperties, ResourceWriteProperties, Watermark, BaseExtract +from elt_common.extract import BaseExtract, ResourceProperties, ResourceWriteProperties, Watermark LOGGER = logging.getLogger(__name__) @@ -160,5 +161,65 @@ def _extract_table( query = query.where(sa.column(column) > max_value) result = conn.execution_options(yield_per=self._chunk_size).execute(query) - for partition in result.mappings().partitions(): - yield pa.Table.from_pylist(partition) + + target_schema = self.get_table_schema(name) + + column_names = list(result.keys()) + + has_data = False + while True: + rows = result.fetchmany(self._chunk_size) + + if not rows: + break + + has_data = True + + # Convert SQLAlchemy Row objects to column arrays + columns = {} + + for idx, column_name in enumerate(column_names): + columns[column_name] = [row[idx] for row in rows] + + arrow_arrays = [] + + for field in target_schema: + values = columns.get( + field.name, + [None] * len(rows), + ) + + # JSON / JSONB -> string for Iceberg + if pa.types.is_string(field.type): + values = [ + json.dumps(v) + if isinstance(v, (dict, list)) + else str(v) + if v is not None and not isinstance(v, str) + else v + for v in values + ] + elif pa.types.is_integer(field.type): + values = [int(v) if v is not None else None for v in values] + elif pa.types.is_floating(field.type): + values = [float(v) if v is not None else None for v in values] + + array = pa.array( + values, + type=field.type, + ) + arrow_arrays.append(array) + + yield pa.Table.from_arrays( + arrow_arrays, + schema=target_schema, + ) + + # Return empty table with schema when no rows + if not has_data: + empty_arrays = [pa.array([], type=field.type) for field in target_schema] + + yield pa.Table.from_arrays( + empty_arrays, + schema=target_schema, + ) diff --git a/elt-pipelines/fase/ingest/fase/proposal/proposal.py b/elt-pipelines/fase/ingest/fase/proposal/proposal.py index f6c70027..1e3857d0 100644 --- a/elt-pipelines/fase/ingest/fase/proposal/proposal.py +++ b/elt-pipelines/fase/ingest/fase/proposal/proposal.py @@ -1,17 +1,16 @@ -from elt_common.extract import ResourceWriteProperties -from elt_common.sources.sqldatabase import SqlDatabaseSourceConfig, TableInfo -from fase.utils.postgres import PostgresExtract +from elt_common.extract import ( + ResourceWriteProperties, + Watermark, # noqa: F401 +) +from elt_common.sources.sqldatabase import ( + SqlDatabaseSourceConfig, + TableInfo, +) +from elt_common.sources.sqldatabase.postgres import PostgresExtract class PipelinePostgresConfig(SqlDatabaseSourceConfig): - model_config = { - "env_prefix": "proposal__", - "env_nested_delimiter": "__", - "extra": "ignore", - "protected_namespaces": (), - } - - drivername: str = "postgresql+psycopg2" + drivername: str = "postgresql+psycopg" table: str @property diff --git a/elt-pipelines/fase/ingest/fase/utils/postgres.py b/elt-pipelines/fase/ingest/fase/utils/postgres.py deleted file mode 100644 index 5bdf5b30..00000000 --- a/elt-pipelines/fase/ingest/fase/utils/postgres.py +++ /dev/null @@ -1,136 +0,0 @@ -import json -import logging -from typing import Dict, Iterator - -import pandas as pd -import pyarrow as pa -import sqlalchemy as sa -from elt_common.extract import Watermark -from elt_common.sources.sqldatabase import SqlDatabaseExtract, SqlDatabaseSourceConfig -from pydantic_settings import SettingsConfigDict - -LOGGER = logging.getLogger(__name__) - -# Default PyArrow type lookup dictionary -DEFAULT_TYPE_MAP: Dict[str, str] = { - "int": "bigint", - "bool": "bool", - "json": "text", - "uuid": "text", - "float": "double", - "numeric": "double", - "double": "double", - "timestamp": "timestamp", - "date": "date", -} - -# Mapping string representations to actual PyArrow types -PA_TYPE_MAPPING = { - "bigint": pa.int64(), - "bool": pa.bool_(), - "double": pa.float64(), - "timestamp": pa.timestamp("us"), - "date": pa.date32(), - "text": pa.string(), -} - - -class PostgresConfig(SqlDatabaseSourceConfig): - model_config = SettingsConfigDict( - extra="ignore", - protected_namespaces=(), - ) - - type_map: Dict[str, str] = DEFAULT_TYPE_MAP - - -class PostgresExtract(SqlDatabaseExtract): - config_cls = PostgresConfig - - def map_pg_to_pq_type(self, pg_type) -> str: - t = str(pg_type).lower() - type_map = getattr(self.config, "type_map", DEFAULT_TYPE_MAP) - for keyword, mapped_type in type_map.items(): - if keyword in t: - return mapped_type - return "text" - - def get_table_schema(self, table_name: str) -> pa.Schema: - inspector = sa.inspect(self._engine) - columns = inspector.get_columns(table_name) - - arrow_fields = [ - pa.field( - col["name"], - PA_TYPE_MAPPING.get(self.map_pg_to_pq_type(col["type"]), pa.string()), - nullable=True, - ) - for col in columns - ] - - return pa.schema(arrow_fields) - - def _extract_table( - self, - name: str, - *, - conn: sa.Connection, - watermark: Watermark | None = None, - ) -> Iterator[pa.Table]: - LOGGER.debug( - f"Extracting Postgres table {name} in chunks of {self._chunk_size} rows." - ) - - target_schema = self.get_table_schema(name) - table = sa.Table(name, self._metadata, autoload_with=self._engine) - - query = sa.select(table) - if watermark is not None: - column, max_value = watermark.column, watermark.value - LOGGER.debug( - f"Cursor value detected. Limiting query to {column} > {max_value}" - ) - query = query.where(sa.column(column) > max_value) - - result = conn.execution_options(yield_per=self._chunk_size).execute(query) - - has_data = False - while True: - chunk = result.fetchmany(self._chunk_size) - if not chunk: - break - - has_data = True - df = pd.DataFrame(chunk, columns=result.keys()) - - for col in df.columns: - if df[col].dtype == "object": - df[col] = df[col].apply( - lambda x: ( - json.dumps(x) - if isinstance(x, (dict, list)) - else str(x) - if pd.notnull(x) and not isinstance(x, str) - else x - ) - ) - - arrow_table = pa.Table.from_pandas(df) - - aligned_columns = [] - for field in target_schema: - if field.name in arrow_table.column_names: - aligned_columns.append( - arrow_table.column(field.name).cast(field.type) - ) - else: - aligned_columns.append( - pa.array([None] * len(arrow_table), type=field.type) - ) - - yield pa.Table.from_arrays(aligned_columns, schema=target_schema) - - if not has_data: - dummy_row = {col_key: [None] for col_key in result.keys()} - empty_df = pd.DataFrame(dummy_row) - yield pa.Table.from_pandas(empty_df, schema=target_schema) diff --git a/elt-pipelines/pyproject.toml b/elt-pipelines/pyproject.toml index 2e9fbb1d..fbc0c58e 100644 --- a/elt-pipelines/pyproject.toml +++ b/elt-pipelines/pyproject.toml @@ -10,8 +10,12 @@ dependencies = [ ] [project.optional-dependencies] -statusdisplay = [ +proposal = [ "pyarrow>=24.0.0", + "sqlalchemy>=2.0.0", + "psycopg[binary]>=3.1.0", +] +statusdisplay = [ "requests>=2.34.2", ] diff --git a/infra/local/warehouses/lakekeeper/fase_landing.json b/infra/local/warehouses/lakekeeper/fase_landing.json new file mode 100644 index 00000000..daa217e8 --- /dev/null +++ b/infra/local/warehouses/lakekeeper/fase_landing.json @@ -0,0 +1,27 @@ +{ + "warehouse-name": "fase_landing", + "storage-credential": { + "type": "s3", + "aws-access-key-id": "adpsuperuser", + "aws-secret-access-key": "adppassword", + "credential-type": "access-key" + }, + "storage-profile": { + "type": "s3", + "bucket": "fase-landing", + "key-prefix": "iceberg", + "endpoint": "http://adp-router:59000", + "region": "local-01", + "path-style-access": true, + "sts-enabled": false, + "flavor": "s3-compat" + }, + "delete-profile": { + "type": "hard" + }, + "permissions": { + "service-account-trino": [ + "select" + ] + } +} From 10a26e0dae0ee1c8605e457891479241cdb182c1 Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Mon, 27 Jul 2026 10:38:55 +0100 Subject: [PATCH 8/9] Fix for unit test --- .../sources/sqldatabase/__init__.py | 46 +++++++++++++-- .../sources/sqldatabase/postgres.py | 56 +++++++++++++++++++ 2 files changed, 98 insertions(+), 4 deletions(-) create mode 100644 elt-common/src/elt_common/sources/sqldatabase/postgres.py diff --git a/elt-common/src/elt_common/sources/sqldatabase/__init__.py b/elt-common/src/elt_common/sources/sqldatabase/__init__.py index 73dc808d..9140bf32 100644 --- a/elt-common/src/elt_common/sources/sqldatabase/__init__.py +++ b/elt-common/src/elt_common/sources/sqldatabase/__init__.py @@ -14,11 +14,17 @@ LOGGER = logging.getLogger(__name__) +DEFAULT_PA_TYPE_MAPPING = { + "bigint": pa.int64(), + "bool": pa.bool_(), + "double": pa.float64(), + "timestamp": pa.timestamp("us"), + "date": pa.date32(), + "text": pa.string(), +} -class SqlDatabaseSourceConfig(BaseSettings): - """Configuration required to connect to a database""" - # connection +class SqlDatabaseSourceConfig(BaseSettings): drivername: str database: str database_schema: Optional[str] = None @@ -27,7 +33,6 @@ class SqlDatabaseSourceConfig(BaseSettings): username: Optional[str] = None password: Optional[SecretStr] = None - # loading behaviour chunk_size: int = 5000 @property @@ -91,6 +96,39 @@ def __init__(self, config: SqlDatabaseSourceConfig): self._engine = sa.create_engine(config.connection_url) self._metadata = sa.MetaData(schema=config.database_schema) + def map_sql_to_pq_type(self, sql_type: Any) -> pa.DataType: # noqa: F821 + t = str(sql_type).lower() + if "int" in t: + return pa.int64() + if "double" in t or "float" in t or "numeric" in t: + return pa.float64() + if "bool" in t: + return pa.bool_() + if "timestamp" in t: + return pa.timestamp("us") + if "date" in t: + return pa.date32() + return pa.string() + + def get_table_schema(self, table_name: str) -> pa.Schema: + inspector = sa.inspect(self._engine) + schema = getattr(self.config, "database_schema", None) + columns = inspector.get_columns(table_name, schema=schema) + + arrow_fields = [ + pa.field( + self.normalize_column_name(col["name"]), + self.map_sql_to_pq_type(col["type"]), + nullable=True, + ) + for col in columns + ] + + return pa.schema(arrow_fields) + + def normalize_column_name(self, name: str) -> str: + return name + @abstractmethod def table_info(self) -> dict[str, Optional[TableInfo]]: """Define the tables to be extracted from the DB. diff --git a/elt-common/src/elt_common/sources/sqldatabase/postgres.py b/elt-common/src/elt_common/sources/sqldatabase/postgres.py new file mode 100644 index 00000000..64a42809 --- /dev/null +++ b/elt-common/src/elt_common/sources/sqldatabase/postgres.py @@ -0,0 +1,56 @@ +import logging +from typing import Dict + +import pyarrow as pa +from pydantic_settings import SettingsConfigDict + +from elt_common.sources.sqldatabase import SqlDatabaseExtract, SqlDatabaseSourceConfig + +LOGGER = logging.getLogger(__name__) + +# Type mapping between Postgres SQLAlchemy types and PyArrow +DEFAULT_TYPE_MAP: Dict[str, str] = { + "int": "bigint", + "bool": "bool", + "json": "text", + "uuid": "text", + "float": "double", + "numeric": "double", + "double": "double", + "timestamp": "timestamp", + "date": "date", +} + +PA_TYPE_MAPPING = { + "bigint": pa.int64(), + "bool": pa.bool_(), + "double": pa.float64(), + "timestamp": pa.timestamp("us", tz="UTC"), # Must match Iceberg timestamptz + "date": pa.date32(), + "text": pa.string(), +} + + +class PostgresConfig(SqlDatabaseSourceConfig): + model_config = SettingsConfigDict( + extra="ignore", + protected_namespaces=(), + ) + + type_map: Dict[str, str] = DEFAULT_TYPE_MAP + + +class PostgresExtract(SqlDatabaseExtract): + config_cls = PostgresConfig + + def map_sql_to_pq_type(self, sql_type: Any) -> pa.DataType: # noqa: F821 + t = str(sql_type).lower() + if "timestamp" in t: + # Postgres needs UTC timestamp for Iceberg compatibility + return pa.timestamp("us", tz="UTC") + + type_map = getattr(self.config, "type_map", DEFAULT_TYPE_MAP) + for keyword, mapped_type in type_map.items(): + if keyword in t: + return PA_TYPE_MAPPING.get(mapped_type, pa.string()) + return pa.string() From c778a0e369155c7592f81778621fcf2854c695ec Mon Sep 17 00:00:00 2001 From: Chi Kai Lam Date: Mon, 27 Jul 2026 10:58:18 +0100 Subject: [PATCH 9/9] Fix for Any is not defined --- elt-common/src/elt_common/sources/sqldatabase/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/elt-common/src/elt_common/sources/sqldatabase/__init__.py b/elt-common/src/elt_common/sources/sqldatabase/__init__.py index 9140bf32..d43c961c 100644 --- a/elt-common/src/elt_common/sources/sqldatabase/__init__.py +++ b/elt-common/src/elt_common/sources/sqldatabase/__init__.py @@ -3,7 +3,7 @@ import json import logging from abc import abstractmethod -from typing import Generator, Iterator, NamedTuple, Optional +from typing import Any, Generator, Iterator, NamedTuple, Optional import pyarrow as pa import sqlalchemy as sa @@ -96,7 +96,7 @@ def __init__(self, config: SqlDatabaseSourceConfig): self._engine = sa.create_engine(config.connection_url) self._metadata = sa.MetaData(schema=config.database_schema) - def map_sql_to_pq_type(self, sql_type: Any) -> pa.DataType: # noqa: F821 + def map_sql_to_pq_type(self, sql_type: Any) -> pa.DataType: t = str(sql_type).lower() if "int" in t: return pa.int64()