feat(elt-pipelines): ISISUSERDB oracle sources pipeline - #398
Open
bashanlam wants to merge 3 commits into
Open
Conversation
…eline isisuserdb to extract data from the Oracle database source and load it into the Iceberg destination.
Contributor
|
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 Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughChangesFASE Oracle ingestion
Possibly related issues
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 9
🧹 Nitpick comments (2)
elt-pipelines/fase/ingest/fase/utils/oracle.py (2)
27-39: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winPrefer the public
creatorkwarg over monkey-patchingpool._creator.
create_engine()already supports passing acreatorcallable directly, which achieves the same goal without reaching into the pool's private_creatorattribute (undocumented, version-dependent). Buildingconfig.connection_uriat all becomes largely redundant once the custom creator is installed, sincecreatorbypasses URL-derived connection args.♻️ Proposed fix
- self.engine = create_engine(config.connection_uri) def custom_connection_creator(): return cx_Oracle.connect( config.username, config.password, f"{config.host}:{config.port}/{config.database}", ) - self.engine.pool._creator = custom_connection_creator + self.engine = create_engine(config.connection_uri, creator=custom_connection_creator)🤖 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/oracle.py` around lines 27 - 39, Update OracleExtractor.__init__ to define the custom_connection_creator before constructing the engine, then pass it to create_engine via the public creator argument. Remove the post-construction assignment to self.engine.pool._creator and avoid relying on config.connection_uri when the creator supplies the connection details.
83-90: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRedundant metadata inspection call.
_get_lob_columnsre-runsinspector.get_columns(...)even thoughget_table_schemaalready fetched the same column metadata moments earlier infetch_as_arrow. Consider deriving LOB columns from the already-computedcol_hints(e.g. by checking the raw Oracle type string once) to avoid a second round-trip to the data dictionary per table.🤖 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/oracle.py` around lines 83 - 90, The _get_lob_columns method redundantly calls inspector.get_columns after fetch_as_arrow has already obtained the metadata. Update fetch_as_arrow to derive LOB column names from its existing col_hints or raw Oracle type information, and pass or reuse that result instead of invoking _get_lob_columns metadata inspection; preserve detection of both CLOB and BLOB columns.
🤖 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/isisuserdb/isisuserdb.py`:
- Line 22: Remove trailing spaces from the affected lines in the isisuserdb
module and clean any whitespace-only blank lines so the file passes the
trailing-whitespace and end-of-file-fixer hooks.
- Around line 61-74: Wire the watermark through extract_resource_properties and
its nested extractor closure instead of ignoring it: configure the corresponding
watermark_column from the table schema/configuration and pass the received
watermark to fetch_as_arrow. Update fetch_as_arrow as needed so a watermark
produces a bounded query while preserving the existing full-table behavior when
no watermark is provided.
- Line 25: Update the credential handling around
PipelineOracleConfig.connection_uri to construct the database URL with
SQLAlchemy URL.create (or properly quote username and password) instead of
directly interpolating them. Ensure reserved characters remain valid and, if
password is SecretStr, pass password.get_secret_value() only when creating the
URL while preserving secret protection in logs and reprs.
- Around line 21-29: The settings model’s schema field conflicts with Pydantic’s
BaseModel.schema member. In the configuration class containing drivername,
username, and table, rename the Python attribute to schema_ and preserve the
external configuration key as schema through the model’s field aliasing
mechanism.
In `@elt-pipelines/fase/ingest/fase/utils/oracle.py`:
- Around line 133-136: Update the LOB-column transformation in the ingest
utility so BLOB values are preserved as binary data or encoded with a reversible
representation such as base64, rather than converting raw bytes with str(). Keep
CLOB values on the existing text path and ensure null values remain null.
- Around line 143-156: Handle cast failures in the target_schema alignment loop
before yielding the table: wrap each field’s column cast in error handling,
catch pyarrow.ArrowInvalid, and provide diagnostic context including the field
name and failing data/row where available. Preserve successful casts and
missing-column null arrays, while applying the existing extraction’s
partial-failure behavior instead of aborting the entire table.
- Around line 116-121: Update the Oracle loading method around
self.engine.connect() to validate schema and table identifiers against
inspector.get_table_names() or the established allow-list before constructing
SQL, rejecting unknown objects; then replace the unbounded pd.read_sql call with
chunked or paginated reads and combine/process the chunks without loading the
entire table at once.
- Around line 53-66: The Oracle type mapping and Arrow field-building logic use
inconsistent classifications: map_oracle_to_pq_type never produces “bool” or
“date”, so those branches cannot run. Update map_oracle_to_pq_type and the
corresponding Arrow conversion loop to use one consistent classification,
ensuring Oracle DATE values follow the intended pa.date32() path and removing or
supporting the unreachable bool handling as appropriate.
- Around line 157-160: Update the empty-table branch around dummy_row and
empty_df so the constructed pandas DataFrame has zero rows while retaining all
expected columns and their schema-compatible types. Ensure pa.Table.from_pandas
yields an empty Arrow table without introducing an all-NULL record.
---
Nitpick comments:
In `@elt-pipelines/fase/ingest/fase/utils/oracle.py`:
- Around line 27-39: Update OracleExtractor.__init__ to define the
custom_connection_creator before constructing the engine, then pass it to
create_engine via the public creator argument. Remove the post-construction
assignment to self.engine.pool._creator and avoid relying on
config.connection_uri when the creator supplies the connection details.
- Around line 83-90: The _get_lob_columns method redundantly calls
inspector.get_columns after fetch_as_arrow has already obtained the metadata.
Update fetch_as_arrow to derive LOB column names from its existing col_hints or
raw Oracle type information, and pass or reuse that result instead of invoking
_get_lob_columns metadata inspection; preserve detection of both CLOB and BLOB
columns.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 2e42f987-e520-47d1-8f9d-551ebf0e15f2
📒 Files selected for processing (2)
elt-pipelines/fase/ingest/fase/isisuserdb/isisuserdb.pyelt-pipelines/fase/ingest/fase/utils/oracle.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
ref #340
Description
This PR implements the ingestion elt-pipelines to move ISISUSERDB data into the lakehouse. It establishes the end-to-end data flow by extracting data from Oracle sources and loading it into the Iceberg destination.
Scope of Work
Summary by CodeRabbit