feat(elt-pipelines): Proposal postgresql sources pipeline - #396
feat(elt-pipelines): Proposal postgresql sources pipeline#396bashanlam wants to merge 10 commits into
Conversation
Implement the elt-pipeline proposal to extract data from the PostgreSQL database source and load it into the Iceberg destination.
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughChangesAdds environment-configured PostgreSQL extraction, schema mapping to PyArrow, chunked table streaming, and pipeline resources for each configured table. Postgres extraction pipeline
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
elt-pipelines/fase/ingest/fase/utils/postgres.py (2)
18-23: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFormat multiple statements on separate lines.
Putting multiple statements on a single line violates PEP 8 guidelines and reduces readability.
🧹 Proposed fix
- 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' + 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'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@elt-pipelines/fase/ingest/fase/utils/postgres.py` around lines 18 - 23, Reformat the type-mapping conditionals in the visible utility function so each if statement and its return statement appears on separate lines, preserving the existing ordering and return values.Source: Linters/SAST tools
39-44: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFormat multiple statements on separate lines.
Placing the
if/elif/elsecondition and assignment on the same line makes the code harder to read and violates PEP 8 formatting standards.🧹 Proposed fix
- 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()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@elt-pipelines/fase/ingest/fase/utils/postgres.py` around lines 39 - 44, Reformat the type-mapping conditional in the visible PostgreSQL utility code so each if/elif/else condition and its pa_type assignment appear on separate properly indented lines, without changing the existing type mappings or fallback behavior.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@elt-pipelines/fase/ingest/fase/proposal/proposal.py`:
- Around line 31-37: Update the connection_uri method to URL-encode the stripped
username and password with urllib.parse.quote_plus before interpolating them
into the URI, while preserving the existing driver, host, port, and database
formatting.
In `@elt-pipelines/fase/ingest/fase/utils/postgres.py`:
- Around line 81-84: Update the empty-result branch around has_data so it yields
a zero-row PyArrow table using target_schema, rather than constructing dummy_row
and a one-row DataFrame of None values. Preserve the existing behavior for
sources containing data.
---
Nitpick comments:
In `@elt-pipelines/fase/ingest/fase/utils/postgres.py`:
- Around line 18-23: Reformat the type-mapping conditionals in the visible
utility function so each if statement and its return statement appears on
separate lines, preserving the existing ordering and return values.
- Around line 39-44: Reformat the type-mapping conditional in the visible
PostgreSQL utility code so each if/elif/else condition and its pa_type
assignment appear on separate properly indented lines, without changing the
existing type mappings or fallback behavior.
🪄 Autofix (Beta)
✅ Autofix completed
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: bf9112f6-fe76-4ea3-a46b-f6dde01e1c47
📒 Files selected for processing (2)
elt-pipelines/fase/ingest/fase/proposal/proposal.pyelt-pipelines/fase/ingest/fase/utils/postgres.py
bashanlam
changed the title
bashanlam
changed the title
|
Note Autofix is a beta feature. Expect some limitations and changes as we gather feedback and continue to improve it. Fixes Applied SuccessfullyFixed 2 file(s) based on 2 unresolved review comments. Files modified:
Commit: The changes have been pushed to the Time taken: |
Fixed 2 file(s) based on 2 unresolved review comments. Co-authored-by: CodeRabbit <noreply@coderabbit.ai>
This reverts commit 62abf0d.
WHTaylor
left a comment
WHTaylor
left a comment
There was a problem hiding this comment.
A lot of this was supposed to be covered by the sqldatabase package. If that didn't work for this, it'd probably be ideal if we could fix it instead of reimplementing a lot of the same functionality.
Add set up for a fase_landing warehouse in the local infrastructure Centralize loading pg and oracle by using sqlalchemy at sqldatabase/__init__.py
Add unit test
WHTaylor
left a comment
WHTaylor
left a comment
There was a problem hiding this comment.
It might be useful to split this PR into separate 'Add JSON handling for SQL extracts' and 'Add Proposal DB extract' PRs, as they're kinda separate. Not that fussed though.
| statusdisplay = [ | ||
| proposal = [ | ||
| "pyarrow>=24.0.0", | ||
| "sqlalchemy>=2.0.0", | ||
| "psycopg[binary]>=3.1.0", | ||
| ] | ||
| statusdisplay = [ | ||
| "requests>=2.34.2", | ||
| ] |
There was a problem hiding this comment.
- The
pyarrowdependency has been removed fromstatusdisplay, could it be put back?- I reckon it should eventually be a dependency of
elt-common, but we're currently getting it transitively fromdlt[parquet] - If this happened because of a manual change to this file, using
uv add <dependency> --optional <optional group>is probably a good way to avoid it
- I reckon it should eventually be a dependency of
- Would
sqlalchemy[postgresql-psycopg]orsqlalchemy[postgresql-psycopgbinary]work instead of specifying thepsycopgdependency separately? Would make sure the versions stay in sync.
| postgresql.JSON: _SQL_ROOT_TYPES[sa.String], | ||
| postgresql.JSONB: _SQL_ROOT_TYPES[sa.String], |
There was a problem hiding this comment.
| postgresql.JSON: _SQL_ROOT_TYPES[sa.String], | |
| postgresql.JSONB: _SQL_ROOT_TYPES[sa.String], | |
| postgresql.JSON: _SQL_ROOT_TYPES[sa.JSON], | |
| postgresql.JSONB: _SQL_ROOT_TYPES[sa.JSON], |
| sa.Float: pa.float64, | ||
| sa.Integer: pa.int32, | ||
| sa.Interval: lambda: pa.duration("us"), | ||
| sa.JSON: pa.string, |
There was a problem hiding this comment.
As the only one of these types that don't match up directly, I feel like an explanatory comment would help. Something like this?
| sa.JSON: pa.string, | |
| # Because iceberg doesn't support JSON values, we coerce JSON values to strings | |
| sa.JSON: pa.string, |
| conn.execute( | ||
| db_table.insert(), | ||
| [ | ||
| {"id": 1, "json_col": '{"key": "val1"}', "jsonb_col": '{"key": "val2"}'}, |
There was a problem hiding this comment.
It'd be good to add a value with some list/object types involved, as those are the most complicated parts of the conversion.
|
|
||
| class PipelinePostgresConfig(SqlDatabaseSourceConfig): | ||
| drivername: str = "postgresql+psycopg" | ||
| table: str |
There was a problem hiding this comment.
Instead of str, this could be list[str]. The tables can then be specified as a JSON string which pydantic will deserialize automatically, rather than having to do the split and strip stuff.
| def get_table_schema(self, table_name: str) -> pa.Schema: | ||
| """Autoloads the SQL table and constructs a PyArrow schema using schema.py.""" | ||
| table = sa.Table( | ||
| table_name, | ||
| self._metadata, | ||
| autoload_with=self._engine, | ||
| ) | ||
| schema = to_pyarrow_schema(table) | ||
|
|
||
| # Apply column normalization (e.g. lowercase) if overridden by subclasses | ||
| normalized_fields = [ | ||
| pa.field(self.normalize_column_name(field.name), field.type, field.nullable) | ||
| for field in schema | ||
| ] | ||
| return pa.schema(normalized_fields) | ||
|
|
||
| def normalize_column_name(self, name: str) -> str: | ||
| return name | ||
|
|
There was a problem hiding this comment.
Does this do anything beyond the to_pyarrow_schema that was used originally?
@martyngigg and I had a small discussion about #405, and came to the conclusion that leaving the DB table names as they are by default is probably what we want to do.
| if v is not None and not isinstance(v, str) | ||
| else v | ||
| for v in values | ||
| ] |
There was a problem hiding this comment.
A lot has been added to this method, but it feels like this block is the only bit we really care about? Is there a way to pare things back and make it clear that we're just converting JSON into strings, and maybe avoid some of the extra data copying that's been introduced?
There was a problem hiding this comment.
Github didn't pick up the lines this comment was supposed to be about, I'm referring to the 'convert JSON values into strings' bit.
3 participants
ref #340
Description
This PR implements the ingestion elt-pipelines to move proposal application data into the lakehouse. It establishes the end-to-end data flow by extracting data from PostgreSQL sources and loading it into the Iceberg destination.
Scope of Work
Summary by CodeRabbit