feat: pass in the schema to ParquetTable from bigquery #695
Conversation
WalkthroughThis update reorganizes import statements for clarity in multiple files and introduces a new utility method, Changes
Sequence Diagram(s)sequenceDiagram
participant BigQuery as BigQuery External Table
participant DelegatingCatalog as DelegatingBigQueryMetastoreCatalog
participant SparkBQUtils as SparkBQUtils
participant ParquetTable as ParquetTable
DelegatingCatalog->>BigQuery: Get ExternalTableDefinition (with schema)
DelegatingCatalog->>SparkBQUtils: toSparkSchema(bqSchema)
SparkBQUtils-->>DelegatingCatalog: StructType (Spark schema)
DelegatingCatalog->>ParquetTable: Create ParquetTable with explicit schema
Possibly related PRs
Suggested reviewers
Poem
Warning Review ran into problems🔥 ProblemsGitHub Actions and Pipeline Checks: Resource not accessible by integration - https://docs.github.com/rest/actions/workflow-runs#list-workflow-runs-for-a-repository. Please grant the required permissions to the CodeRabbit GitHub App under the organization or repository settings. 📜 Recent review detailsConfiguration used: CodeRabbit UI 📒 Files selected for processing (3)
✅ Files skipped from review due to trivial changes (1)
🚧 Files skipped from review as they are similar to previous changes (2)
⏰ Context from checks skipped due to timeout of 90000ms (19)
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/SparkBQUtils.scala (1)
28-32: New utility method for schema conversionClean implementation of BigQuery to Spark schema conversion. Consider adding error handling for malformed schemas.
def toSparkSchema(bqSchema: Schema): StructType = { val schemaConverters = SchemaConverters.from(SchemaConvertersConfiguration.createDefault()) - schemaConverters.toSpark(bqSchema) + try { + schemaConverters.toSpark(bqSchema) + } catch { + case e: Exception => + throw new RuntimeException(s"Failed to convert BigQuery schema: ${e.getMessage}", e) + } }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro (Legacy)
📒 Files selected for processing (3)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/BigQueryNative.scala(1 hunks)cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DelegatingBigQueryMetastoreCatalog.scala(2 hunks)cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/SparkBQUtils.scala(2 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (19)
- GitHub Check: service_tests
- GitHub Check: service_tests
- GitHub Check: service_commons_tests
- GitHub Check: hub_tests
- GitHub Check: hub_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: api_tests
- GitHub Check: online_tests
- GitHub Check: online_tests
- GitHub Check: aggregator_tests
- GitHub Check: flink_tests
- GitHub Check: flink_tests
- GitHub Check: aggregator_tests
- GitHub Check: scala_compile_fmt_fix
- GitHub Check: api_tests
- GitHub Check: enforce_triggered_workflows
🔇 Additional comments (6)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/SparkBQUtils.scala (2)
6-7: Added imports for BigQuery schema handlingNew imports needed for the schema conversion utility.
11-12: Added Spark StructType importRequired for the return type of the new utility method.
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/BigQueryNative.scala (1)
3-7: Reorganized imports for better readabilityImports are now more logically grouped.
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DelegatingBigQueryMetastoreCatalog.scala (3)
3-11: Reorganized importsImproved import organization.
15-15: Moved NoSuchTableException importRelocated for better organization.
116-117:✅ Verification successful
Added explicit schema conversion for ParquetTable
Now using the schema from BigQuery instead of None. This improves schema accuracy.
🏁 Script executed:
#!/bin/bash # Check if there are any other ParquetTable instances missing schema conversion rg "ParquetTable\(" -A 10 | grep -B 5 -A 5 "None" | grep -v "Option(SparkBQUtils"Length of output: 81
🏁 Script executed:
#!/bin/bash # List all ParquetTable instantiations with context to verify schema conversion consistency rg "ParquetTable\(" -n -A5 -B5Length of output: 1666
Explicit schema conversion for ParquetTable
– Now passing Option(SparkBQUtils.toSparkSchema(...)) instead of None for ParquetTable
– Verified no other ParquetTable instantiations lack schema conversion
tchow-zlai
changed the title
tchow-zlai
force-pushed
the
tchow/bigquery-catalog
branch
from
e6d9666 to
c888f74
Compare
Co-authored-by: Thomas Chow <[email protected]> Co-authored-by: Thomas Chow <[email protected]>
Co-authored-by: Thomas Chow <[email protected]> Co-authored-by: Thomas Chow <[email protected]>
tchow-zlai
force-pushed
the
tchow/bigquery-catalog
branch
from
c888f74 to
278a9bb
Compare
Summary
Checklist
Summary by CodeRabbit
New Features
Refactor
Bug Fixes