Skip to content

Commit

Permalink
fix(query): make datasource had attach snapshot loc hint
Browse files Browse the repository at this point in the history
  • Loading branch information
sundy-li committed Nov 27, 2024
1 parent 519f914 commit 559dc3a
Showing 1 changed file with 50 additions and 26 deletions.
76 changes: 50 additions & 26 deletions src/query/storages/fuse/src/fuse_table.rs
Original file line number Diff line number Diff line change
Expand Up @@ -372,32 +372,41 @@ impl FuseTable {
DatabaseType::NormalDB => {
let options = self.table_info.options();

if let Some(storage_prefix) = options.get(OPT_KEY_STORAGE_PREFIX) {
// If the table is attaching to someone else,
// parse the snapshot location from the hint file.
//
// The snapshot location is allowed
// to be fetched from the table level instance cache.
let snapshot_location = self
.attached_table_location
.get_or_try_init(|| async {
let hint =
format!("{}/{}", storage_prefix, FUSE_TBL_LAST_SNAPSHOT_HINT);
let hint_content = self.operator.read(&hint).await?.to_vec();
let snapshot_full_path = String::from_utf8(hint_content)?;
let operator_info = self.operator.info();
Ok::<_, ErrorCode>(
snapshot_full_path[operator_info.root().len()..].to_string(),
)
})
.await?;
Ok(Some(snapshot_location.to_owned()))
} else {
Ok(options
.get(OPT_KEY_SNAPSHOT_LOCATION)
// for backward compatibility, we check the legacy table option
.or_else(|| options.get(OPT_KEY_LEGACY_SNAPSHOT_LOC))
.cloned())
match options
.get(OPT_KEY_SNAPSHOT_LOCATION)
// for backward compatibility, we check the legacy table option
.or_else(|| options.get(OPT_KEY_LEGACY_SNAPSHOT_LOC))
.cloned()
{
Some(v) => Ok(Some(v)),
None => {
if let Some(storage_prefix) = options.get(OPT_KEY_STORAGE_PREFIX) {
// If the table is attaching to someone else,
// parse the snapshot location from the hint file.
//
// The snapshot location is allowed
// to be fetched from the table level instance cache.
let snapshot_location = self
.attached_table_location
.get_or_try_init(|| async {
let hint = format!(
"{}/{}",
storage_prefix, FUSE_TBL_LAST_SNAPSHOT_HINT
);
let hint_content = self.operator.read(&hint).await?.to_vec();
let snapshot_full_path = String::from_utf8(hint_content)?;
let operator_info = self.operator.info();
Ok::<_, ErrorCode>(
snapshot_full_path[operator_info.root().len()..]
.to_string(),
)
})
.await?;
Ok(Some(snapshot_location.to_owned()))
} else {
Ok(None)
}
}
}
}
}
Expand Down Expand Up @@ -503,6 +512,21 @@ impl Table for FuseTable {
&self.table_info
}

fn get_data_source_info(&self) -> DataSourceInfo {
let table_info = match self.attached_table_location.get() {
Some(snapshot_location) => {
let mut table_info = self.table_info.clone();
table_info
.meta
.options
.insert(OPT_KEY_SNAPSHOT_LOCATION.to_string(), snapshot_location);
table_info
}
None => self.table_info.clone(),
};
DataSourceInfo::TableSource(table_info)
}

fn get_data_metrics(&self) -> Option<Arc<StorageMetrics>> {
Some(self.data_metrics.clone())
}
Expand Down

0 comments on commit 559dc3a

Please sign in to comment.