ref(schema): Move schema into the schema store and remove schema cache #231
Conversation
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
iambriccardo
commented
Aug 5, 2025
Comment on lines
28
to
36
| fn generate_sequence_number(start_lsn: PgLsn, commit_lsn: PgLsn) -> String { | ||
| let start_lsn = u64::from(start_lsn); | ||
| let commit_lsn = u64::from(commit_lsn); | ||
|
|
||
| format!("{commit_lsn:016x}/{start_lsn:016x}") | ||
| } | ||
|
|
||
| /// Returns the BigQuery table id as [`String`] for a supplied [`TableName`]. | ||
| pub fn table_name_to_bigquery_table_id(table_name: &TableName) -> String { |
Contributor
Author
There was a problem hiding this comment.
Lifted both functions out of the BigQueryDestination for more ergonomics.
iambriccardo
commented
Aug 5, 2025
Comment on lines
-76
to
-80
| // We load the table schemas and check that they are correctly fetched. | ||
| let mut table_schemas = destination.load_table_schemas().await.unwrap(); | ||
| table_schemas.sort(); | ||
| assert_eq!(table_schemas[0], database_schema.orders_schema()); | ||
| assert_eq!(table_schemas[1], database_schema.users_schema()); |
Contributor
Author
There was a problem hiding this comment.
No need to check schemas since now they are stored in the state store, which is tested already in pipeline_test.
iambriccardo
commented
Aug 5, 2025
| inner | ||
| .table_schemas | ||
| .iter() | ||
| .map(|(id, schema)| (*id, Arc::as_ref(schema).clone())) |
Contributor
Author
There was a problem hiding this comment.
Decided to just expose TableSchema instead of Arc<TableSchema> since this is used in tests and it will make them more ergonomic.
iambriccardo
commented
Aug 5, 2025
| fn get_table_schema( | ||
| &self, | ||
| table_id: &TableId, | ||
| ) -> impl Future<Output = EtlResult<Option<Arc<TableSchema>>>> + Send; |
Contributor
Author
There was a problem hiding this comment.
Since TableSchema is immutable, we can just Arc it to avoid unnecessary clones.
imor
approved these changes
Aug 5, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
This PR implements a revised schema handling mechanism with the following changes:
TableSchemais not anymore stored in the destination, which makes the implementation of a destination much easier and gives us more control over schema management.Destinationinterface is much cleaner without the need for the uglyinjectmethod.SchemaStoretrait has been added which exposes schema storage capabilities. This trait is not used in conjunction withStateStore. The rationale for splitting them up is to follow the single responsibility principle and also allow components to explicitly highlight their interest in theSchemaStoreonly.BigQueryDestinationhas been cleaned, simplified and slightly improved (additional improvements will follow).SchemaStorefor Postgres has been added which stores the table schema data into the same instance as the one used by the PostgresStateStore.PostgresStoreto make sure there are no regressions.