[POC] Alternate implementation of using snapshot schema when reading snapshot #3314
Conversation
This file contains 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
wypoon
changed the title
According to Edwin Choi, in order to get the schema for a snapshot, the only safe option is to scan the metadata files to find the one where the current-snapshot-id matches target snapshot id.
The changes are mostly in spark3. They are necessitated by the catalog support introduced in apache#1783. As the spark3 IcebergSource now implements SupportsCatalogOptions, DataFrameReader#load no longer calls IcebergSource#getTable but calls SparkCatalog#loadTable directly. In order for the SparkTable returned by SparkCatalog#loadTable(Identifier) to be aware of the snapshot, the information about the snapshot needs to be present in the Identifier. For this reason, we introduce a SnapshotAwareIdentifier interface extending Identifier. As SupportsCatalogOptions does not allow a schema to be specified (requested), SparkTable no longer needs a requestedSchema field, so some dead code is removed from it.
Rebased on master. Use constants from SparkReadOptions. Implement snapshotSchema() in SparkFilesScan as it extends SparkBatchScan.
Avoid introducing new methods to BaseTable. Add helper methods to SnapshotUtil instead. Move recovery of schema from previous metadata files in the event that snapshot does not have associated schema id to new PR. Remove snapshotSchema method from SparkBatchScam and its subclasses, as it is not needed. Adjust schema in BaseTableScan when useSnapshot is called.
Use the existing CatalogAndIdentifier and swap out the Identifier for a snapshot-aware TableIdentifier if snapshotId or asOfTimestamp is set.
Fix a bug in BaseTableScan#useSnapshot. Some clean up in SnapshotUtil. Some streamlining in added unit tests. Refactor spark2 Reader to configure the TableScan on construction, and let the TableScan get the schema for the snapshot. Rename new TableIdentifier to SparkTableIdentifier to avoid confusion with existing TableIdentifier (in different package). Add convenience constructor to PathIdentifier to avoid modifying tests for it.
…ableScan. Use SnapshotUtil.snapshotIdAsOfTime in BaseTableScan#asOfTime. Move formatTimestampMillis from BaseTableScan to SnapshotUtil in order to use it there (BaseTableScan is a package-private and not a public class). Fix some error messages.
Incorporate the approach shown in apache#3269 by Ryan Blue. That defines a syntax for selecting a snapshot or timestamp through the table name. Use that instead of a SnapshotAwareIdentifier to load the SparkTable.
wypoon
force-pushed
the
schema-for-snapshot2
branch
from
f7fadf3 to
129cd44
Compare
Removed test left two unused imports.
wypoon
commented
Oct 19, 2021
| CatalogManager catalogManager = spark.sessionState().catalogManager(); | ||
|
|
||
| if (path.contains("/")) { | ||
| // contains a path. Return iceberg default catalog and a PathIdentifier | ||
| String newPath = selector.equals("") ? path : path + "#" + selector; |
There was a problem hiding this comment.
This is one area I'm not sure about. I am not too familiar with Hadoop tables and PathIdentifier. Is the only thing that can go after # the name of a metadata table (and now the snapshot/timestamp selector)?
wypoon
commented
Oct 19, 2021
Comment on lines
+293
to
+302
| // If the table is loaded using the Spark DataFrame API, and option("snapshot-id", <snapshot_id>) | ||
| // or option("as-of-timestamp", <timestamp>) is applied to the DataFrameReader, SparkTable will be | ||
| // constructed with a non-null snapshotId. Subsequently SparkTable#newScanBuilder will be called | ||
| // with the options, which will include "snapshot-id" or "as-of-timestamp". | ||
| // On the other hand, if the table is loaded using SQL, with the table suffixed with a snapshot | ||
| // or timestamp selector, then SparkTable will be constructed with a non-null snapshotId, but | ||
| // SparkTable#newScanBuilder will be called without the "snapshot-id" or "as-of-timestamp" option. | ||
| // We therefore add a "snapshot-id" option here in this latter case. | ||
| // As a consistency check, if "snapshot-id" is in the options, the id must match what we already | ||
| // have. |
There was a problem hiding this comment.
This took me some figuring out. We only need to add snapshot id if the SparkTable is loaded from SQL via the table name syntax, not when loaded using the DataFrame API.
|
Obsolete. Superseded by #3722 which is merged. |
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 is an alternate approach to #1508 based on #3269. This handles the support for Spark 3.