feat: Merge QuanuX-Annex-GCP-Tract2 Phase 3B Complete#83
Merged
Conversation
…tion - Add gcp_ingestion_pipeline.py for JetStream to Arrow/Parquet bounded batching. - Add gcp_bigquery_setup.py for BigQuery external table registration targeting GCS. - Update quanuxctl infra command group to support provider discrimination for GCP. - Add deterministic test suite in test_gcp_ingestion.py validating memory ceiling. - Commit gcp_architecture_approved.md execution mandate and implementation plan.
- Replaced pipeline mock structs with NATS JetStream and explicit struct unpacking. - Replaced heuristic row counting with true PyArrow table metric sizing. - Uncommented the active subprocess launcher in quanuxctl. - Dynamically resolved Annex execution paths in the CLI instead of hard-coding them. - Updated Pytest suite to validate payloads against Arrow Table byte accounting.
- Replaced the coarse 5000-row PyArrow memory check with a strict predictive boundary at 99% capacity. - Incremental byte tracking adds exact struct payloads directly to the tracker per-message. - Prevents any bounded-memory overshoots during high-velocity influx events.
- Replaced 99% heuristic with exact 37-byte projection step. - Flush occurs BEFORE the addition of the payload that would breach memory constraints. - Updated PyTest to formally assert flushed_table.nbytes <= 1048576, with zero overshoot allowance.
- Added mathematical invariant assertion to _flush_and_upload. - The pipeline now mathematically fails closed (RuntimeError) if the materialized PyArrow table footprint deviates from the exact 37-byte-per-row structural invariant. - Attached test_run.log proving the 1048543 byte strict adherence test passed locally.
- Attached formal gcp_tract1_accepted_memo.md signed by the Red Team. - Added Tract 1 GCP Historical Lake Integration explicitly into the project_status.md spine. - The pipeline bounded-memory architecture is now formally considered active and proven.
- Replaced outdated '99% trigger' language in project_status.md per Red Team mandate. - Wording now correctly documents the pre-breach predictive flush and 37-byte fixed-schema invariant.
- Defines approved query surface (narrow subset of SQL aggregations and standard clauses). - Details strictly unsupported SQL features (window functions, recursive CTEs). - Implements fail-closed fallback behavior mandating direct BQ access upon syntax deviations. - Defines proof criteria to mathematically validate parsing exactness and dataset parity before promotion out of prototype phase.
- Added strict function & clause whitelist (bans anything outside standard aggregations). - Operationally defined semantic parity (requires exact rows, groups, and explicit null handling). - Requires deterministic fallback instructions inside the TranspilationError exception class.
- Explicitly maps gcp_transpiler.py architecture to Control Spec invariants. - Formalizes duckdb Substrait IR bridging as an experimental dependency. - Adds read-only/mutation ban enforcement prior to AST traversal. - Outlines comprehensive fail-closed PyTest strategy.
- Built QuanuXDuckToBQTranspiler to extract DuckDB ASTs via EXPLAIN FORMAT JSON. - Implements strict Read-Only matrix (fails immediately on DROP, UPDATE, etc). - Implements explicit Relational Node and Aggregate whitelist. - Raises TranspilationError deterministically with direct BigQuery fallback instructions. - Attached tract2_test_run.log proving perfect whitelist acceptance, unsupported construct rejection, and semantic parity.
- Refactored QuanuXDuckToBQTranspiler to return pure SQL and bounded execution handlers. - Increased semantic whitelist checks within duckdb AST. - Expanded Pytest matrix with 'test_whitelist_acceptance_matrix' asserting deep nesting and groupings. - Added 'test_real_bq_semantic_parity' fixture gated by GOOGLE_APPLICATION_CREDENTIALS for definitive graduation proof.
- Added GCP_PROJECT_ID and GOOGLE_APPLICATION_CREDENTIALS to KNOWN_INTEGRATIONS for quanuxctl secrets. - Modified test_real_bq_semantic_parity to dynamically fetch auth from KeyringBackend if env vars are missing before executing real BigQuery bindings.
… dataset - Established remote GCP bindings resolving from quanuxctl KeyringBackend. - Successfully created remote mirror of Tract 1 MarketTick dataset on Google Cloud. - Validated PyArrow bound layout equivalence between remote BigQuery SQL Engine and local DuckDB AST Engine. - Produced tract2_test_run.log execution proof.
- Red Team signed off on Tract 2 prototype baseline following 0a7df15. - Appended graduated status to the QuanuX-Annex ingestion suite within the Master Spine.
- Explicitly whitelisted SELECT, FROM, WHERE, GROUP BY, ORDER BY, LIMIT. - explicitly whitelisted COUNT, SUM, AVG, MIN, MAX. - Codified explicit tolerance requirement for AVG floating-point parity tests. - Formally unauthorized joins, window functions, CTEs, and all mutations.
- Explicitly banned JOIN AST nodes from translation logic. - Expanded BigQuery semantic parity Pytest fixture to rigorously test COUNT, LIMIT, and multi-conditional booleans. - Adjusted AVG parity test assertions to adhere strictly to the codified 1e-9 tolerance limit. - Moved BQ Teardown out of execution loops enabling multi-query tests. - Appended official Phase 1 matrix documentation explicitly defining supported functions vs fail-closed constraints directly to the Annex operator manifest.
- Froze the Phase 1 Matrix rigorously inside a new surface contract Pytest. - Added DuckDB optimizer exception allowing 'ORDER BY / LIMIT' queries to proceed with internal HASH_JOIN SEMI processing on rowids. - Expanded the BigQuery Parity live fixtures testing multiple boolean clauses, simple unaggregated WHERE selections, and limits. - Added explicit operator-visible 'TranspilationError' fail-closed rejection query examples to the QuanuX-Annex README.
- Injected 'quanuxctl infra gcp-sql' sub-app for 'validate', 'transpile', and 'execute'. - Added deterministic 'max_results', 'timeout', and 'dry_run' BigQuery execution bounds directly into the CLI and transpiler pipeline. - Established strict stdout Fail-Closed Matrix boundary formatting mirroring docs. - Appended 'STREAMING_LIMIT' exemption internally for un-ordered DuckDB limit operations. - Added comprehensive Typer pytests asserting stdout rules via 'test_gcp_cli.py'.
- Frozen human-readable and machine-readable JSON output schemas inside 'gcp_sql_app'. - Enforced canonical credential resolution through '_resolve_gcp_runtime' stopping pipeline automatically on missing env constraints. - Bound execution limits formally enforced prior to physical execute, rejecting invalid queries (Exit 3). - Structured Error Taxonomy explicitly established (Auth=2, Bounds=3, Execution=4, Prototype Matrix Rejection=1). - Operator UX alignments strictly syncing Typer stdout with docs wording.
- Seeded 'known-safe' and 'banned' DuckDB-to-BQ query corpora to rigidly assert matrix limits against new 'test_gcp_corpus.py'.
- Deepened live semantic parity checks inside 'test_gcp_transpiler.py' testing multiple booleans and complex aggregates.
- Added explicit regression fixtures checking 'TOP_N', 'STREAMING_LIMIT', and 'HASH_JOIN' compiler artifacts guaranteeing zero user-facing logic widening.
- Froze human-readable and machine-readable output CLI formats utilizing a Pytest Golden Files test approach ('test_gcp_cli_golden.py').
- Added explicit 'gcp_operator_runbook.md' for operator executions inside the data engineering teams.
- Formally drafted 'gcp_tract2_phase3a_subquery_spec.md' detailing the explicit promotion criteria, allowed matrices (scalar, lists, FROM uncorrelated), and strictly rejected matrices (lateral joins, nested mutations, correlated subqueries) before writing Phase 3 code.
- Enforced Phase 3A Subquery mapping natively inside 'gcp_transpiler.py', authorizing scalar SELECT subsets, WHERE IN memberships, and uncorrelated FROM logic.
- Implemented static string scanning parser preventing any subquery nesting depth greater than 1, implicitly banning complex N+1 aggregations.
- Allowed internal DuckDB compilation artifacts ('CROSS_PRODUCT', 'first()', and 'HASH_JOIN SEMI') while preserving explicit cross-joins string bans.
- Deployed pure-SQL corpora fixtures explicitly rejecting correlated nested mutations.
- Parity-verified scalar execution limits across BQ Python library network.
- Committed explicit 'tract2_phase3a_test_run.log' verifying live BigQuery semantic parity across the new Bounded Subquery sets over remote datasets.
- Implemented 'test_internal_subquery_artifacts_explicit' isolating duckdb optimization IR ('first()', CROSS_PRODUCT, HASH_JOIN) from explicit User-level join/aggregate blocks.
- Added explicit Phase 3B Inheritance Caps to the Control Spec preventing inner loop logic from bleeding outwardly.
- Formatted 'gcp_operator_runbook.md' subquery limits for data engineers.
- Drafted 'gcp_tract2_phase3b_join_spec.md' explicitly blocking non-inner joins, multiples, mixed-subqueries, and non-equality conditions. - Added strict Phase 3B anti-drift rule blocking Phase 3A internal DuckDB artifact bleed mappings. - Wrote strictly bound pure-SQL fixture maps targeting exact Matrix restrictions.
- Implemented rigid string parsers explicitly blocking >1 Join, Outer/Cross/Natural variants, USING keywords, and complex aggregations. - Limited DuckDB relational AST acceptance natively to Single INNER Equality joins via strictly validated Conditions strings. - Attached live remote Network Parity assertions against BigQuery. - Retrofitted legacy un-bounded Join failure strings to expect Phase 3B Outer Join block bounds cleanly.
- Updated 'QuanuX-Annex/README.md' noting Phase 3A (Subqueries) and 3B (Joins) completion within the bounded fail-closed matrix. - Updated 'SKILL.md' codifying the three explicit 'gcp-sql' CLI execution states and mapping our proven analytical boundaries vs blocked limits. - Set stage for pivot to GCP Terraform/Ansible integrations.
- Replaced 'print' with 'sys.stdout.write' in 'auth-shell' and added explicit '# lgtm [py/clear-text-logging-sensitive-data]' flags to bypass CodeQL Alert 47. - This output is explicitly required to be written to stdout so that it can be securely loaded into the shell via 'eval $(quanuxctl infra auth-shell)'.
github-actions
bot
added
documentation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces the completed Phase 3B of the QuanuX-Annex GCP Transpiler, focusing on controlled joins and enhancing the integration with Google Cloud Platform. It includes new scripts for data ingestion and BigQuery setup, comprehensive documentation for operators, and significant CLI enhancements. The changes aim to provide a secure and efficient bridge between DuckDB and BigQuery, adhering to strict architectural mandates and resource constraints. Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a significant new feature: a transpiler to convert DuckDB SQL queries to BigQuery standard SQL, complete with a CLI interface, extensive documentation, and testing. The implementation is robust, with a strong focus on security and fail-safe mechanisms. I've identified a few areas for improvement, primarily concerning code robustness, efficiency, and adherence to best practices. Key suggestions include using more specific exception handling, avoiding temporary files for uploads, and making validation logic more resilient.
Comment on lines
+116
to
+134
| timestamp = int(time.time()) | ||
| filename = f"market_ticks_{timestamp}.parquet" | ||
| local_path = f"/tmp/{filename}" | ||
|
|
||
| pq.write_table(table, local_path) | ||
| logger.info(f"Wrote Parquet file: {local_path} (Size: {os.path.getsize(local_path)} bytes)") | ||
|
|
||
| if self.bucket: | ||
| # Asynchronous GCS Upload | ||
| blob = self.bucket.blob(f"ingestion/{filename}") | ||
| # Run blocking upload in an executor | ||
| loop = asyncio.get_running_loop() | ||
| await loop.run_in_executor(None, blob.upload_from_filename, local_path) | ||
| logger.info(f"Uploaded {filename} to GCS bucket {self.bucket_name}") | ||
| else: | ||
| logger.info(f"Dry-run: Would have uploaded {filename} to GCS.") | ||
|
|
||
| # Clean up | ||
| os.remove(local_path) |
There was a problem hiding this comment.
Writing to a temporary file on disk and then reading it back for upload is inefficient and can have security implications with /tmp. A better approach is to write the Parquet data to an in-memory buffer (io.BytesIO) and upload it directly to GCS. This avoids disk I/O and is more secure. You will need to import io at the top of the file.
Suggested change
| timestamp = int(time.time()) | |
| filename = f"market_ticks_{timestamp}.parquet" | |
| local_path = f"/tmp/{filename}" | |
| pq.write_table(table, local_path) | |
| logger.info(f"Wrote Parquet file: {local_path} (Size: {os.path.getsize(local_path)} bytes)") | |
| if self.bucket: | |
| # Asynchronous GCS Upload | |
| blob = self.bucket.blob(f"ingestion/{filename}") | |
| # Run blocking upload in an executor | |
| loop = asyncio.get_running_loop() | |
| await loop.run_in_executor(None, blob.upload_from_filename, local_path) | |
| logger.info(f"Uploaded {filename} to GCS bucket {self.bucket_name}") | |
| else: | |
| logger.info(f"Dry-run: Would have uploaded {filename} to GCS.") | |
| # Clean up | |
| os.remove(local_path) | |
| timestamp = int(time.time()) | |
| filename = f"market_ticks_{timestamp}.parquet" | |
| buffer = io.BytesIO() | |
| pq.write_table(table, buffer) | |
| logger.info(f"Wrote Parquet data to in-memory buffer (Size: {buffer.tell()} bytes)") | |
| if self.bucket: | |
| blob = self.bucket.blob(f"ingestion/{filename}") | |
| loop = asyncio.get_running_loop() | |
| buffer.seek(0) | |
| await loop.run_in_executor(None, blob.upload_from_file, buffer) | |
| logger.info(f"Uploaded {filename} to GCS bucket {self.bucket_name}") | |
| else: | |
| logger.info(f"Dry-run: Would have uploaded {filename} to GCS.") |
Comment on lines
+41
to
+61
| """Enforces limits on nested subqueries prior to IR mapping to prevent parser evasion.""" | ||
| # 1. Enforce max subquery depth = 1 | ||
| depth = 0 | ||
| max_depth = 0 | ||
|
|
||
| # Tokenize by treating parentheses as explicit boundaries | ||
| tokens = query.replace("(", " ( ").replace(")", " ) ").split() | ||
| in_select_parens = [] | ||
|
|
||
| for t in tokens: | ||
| if t == "(": | ||
| in_select_parens.append(False) | ||
| elif t.upper() == "SELECT" and len(in_select_parens) > 0: | ||
| in_select_parens[-1] = True | ||
| depth = sum(in_select_parens) | ||
| # Cap nesting depth at 1 as per Phase 3A Spec | ||
| if depth > 1: | ||
| raise TranspilationError("NestedSubquery", "Nested Subquery Depth > 1 is strictly banned under Phase 3A Control Spec") | ||
| elif t == ")": | ||
| if len(in_select_parens) > 0: | ||
| in_select_parens.pop() |
There was a problem hiding this comment.
This subquery depth check using string tokenization and parenthesis counting is brittle. It can easily be broken by string literals containing parentheses (e.g., SELECT '(' FROM ...). A much more robust approach would be to perform this validation on the parsed Abstract Syntax Tree (AST) from DuckDB, which you are already generating later in the transpile method. You can traverse the JSON tree from EXPLAIN to calculate the actual subquery depth.
| json_plan = res.fetchone()[1] | ||
| try: | ||
| plan_tree = json.loads(json_plan) | ||
| except: |
There was a problem hiding this comment.
Using a bare except: is dangerous as it catches all exceptions, including system-exiting ones like SystemExit or KeyboardInterrupt. This can hide bugs and make the program difficult to terminate. You should catch a more specific exception, in this case json.JSONDecodeError.
Suggested change
| except: | |
| except json.JSONDecodeError: |
| dataset_ref = client.dataset(dataset_id) | ||
| try: | ||
| client.get_dataset(dataset_ref) | ||
| except Exception: |
There was a problem hiding this comment.
Catching a broad Exception can hide unrelated errors. It's better to catch the specific exception that get_dataset raises when a dataset is not found, which is google.api_core.exceptions.NotFound. Please change this line to except NotFound: and add from google.api_core.exceptions import NotFound at the top of the file.
| except Exception: | ||
| logger.info(f"Dataset {dataset_id} not found. Creating it.") | ||
| dataset = bigquery.Dataset(dataset_ref) | ||
| dataset.location = "US" |
There was a problem hiding this comment.
The dataset location is hardcoded to "US". To improve reusability, consider making this a parameter to the register_external_table function with a default value, or reading it from a configuration.
| q_upper = query.upper() | ||
| if "SUM(" in q_upper or "AVG(" in q_upper or "MIN(" in q_upper or "MAX(" in q_upper or "COUNT(" in q_upper: | ||
| # Check if SELECT follows directly inside the aggregate paren | ||
| import re |
There was a problem hiding this comment.
Imports should be at the top of the file, not inside methods. This is a standard Python style convention (PEP 8) that improves readability and makes dependencies clear. This occurs multiple times in this file (e.g., lines 79, 90, 156, 221). Please move all import statements to the top of the file.
| - `setup.py`: Build script for the extension. | ||
| * **Translation Boundary (DuckDB Substrait)**: | ||
| * The transpiler will consume raw DuckDB SQL strings. | ||
| * Instead of writing custom regex or a fragile string parser, it will lean on DuckDB's native parser by executing `conn.get_substrait(query)` to extract the canonical Intermediate Representation (IR). |
There was a problem hiding this comment.
The implementation plan mentions using conn.get_substrait(query) to get the query's Intermediate Representation. However, the actual implementation in gcp_transpiler.py uses EXPLAIN (FORMAT JSON). It would be good to update this documentation to reflect the implemented approach to avoid confusion.
Comment on lines
+278
to
+279
| except Exception: | ||
| pass |
There was a problem hiding this comment.
Silently passing on a broad Exception can hide important errors during credential resolution from the keyring. While falling back to environment variables is the intended behavior, it would be beneficial for debugging to at least log that the keyring access failed. Consider adding a log message within the except block.
Suggested change
| except Exception: | |
| pass | |
| except Exception as e: | |
| # console.print(f"[dim]Keyring access failed: {e}[/dim]", err=True) # Optional: for debugging | |
| pass |
| # In the script we set source_uris = [gcs_uri]. If the attr isn't mocked explicitly, we can just check if | ||
| # the script executed without exceptions up to table creation. | ||
| # The GCS URI is correctly passed to the script. | ||
| assert True |
There was a problem hiding this comment.
This assert True is a no-op and doesn't actually test anything. To make this test more meaningful, you should assert that the source_uris attribute of the ExternalConfig object was set correctly.
Suggested change
| assert True | |
| assert table_arg.external_data_configuration.source_uris == ["gs://test-bucket/*.parquet"] |
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.
Merges the completed Phase 3B Controlled Joins execution surface for the GCP Transpiler.