update

Author: zhyass

src/query/catalog/src/plan/partition.rs

@@ -392,7 +392,6 @@ impl ReclusterParts {
  }
 }
 
-// TODO refine this
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Default)]
 pub struct ReclusterInfoSideCar {
  pub merged_blocks: Vec<Arc<BlockMeta>>,

src/query/catalog/src/table_context.rs

@@ -279,11 +279,11 @@ pub trait TableContext: Send + Sync {
  max_files: Option<usize>,
  ) -> Result<FilteredCopyFiles>;
 
- fn add_inserted_segment_location(&self, segment_loc: Location) -> Result<()>;
+ fn add_written_segment_location(&self, segment_loc: Location) -> Result<()>;
 
- fn clear_inserted_segment_locations(&self) -> Result<()>;
+ fn clear_written_segment_locations(&self) -> Result<()>;
 
- fn get_inserted_segment_locations(&self) -> Result<Vec<Location>>;
+ fn get_written_segment_locations(&self) -> Result<Vec<Location>>;
 
  fn add_selected_segment_location(&self, _segment_loc: Location) {
  unimplemented!()

src/query/service/src/interpreters/hook/compact_hook.rs

@@ -186,7 +186,7 @@ async fn compact_table(
  PipelineCompleteExecutor::from_pipelines(pipelines, executor_settings)?;
 
  // Clears previously generated segment locations to avoid duplicate data in the refresh phase
- ctx.clear_inserted_segment_locations()?;
+ ctx.clear_written_segment_locations()?;
  ctx.set_executor(complete_executor.get_inner())?;
  complete_executor.execute()?;
  drop(complete_executor);

src/query/service/src/interpreters/hook/refresh_hook.rs

@@ -209,7 +209,7 @@ async fn generate_refresh_index_plan(
  catalog: &str,
  table_id: MetaId,
 ) -> Result<Vec<Plan>> {
- let segment_locs = ctx.get_inserted_segment_locations()?;
+ let segment_locs = ctx.get_written_segment_locations()?;
  let catalog = ctx.get_catalog(catalog).await?;
  let mut plans = vec![];
  let indexes = catalog
@@ -272,7 +272,7 @@ async fn generate_refresh_inverted_index_plan(
  desc: &RefreshDesc,
  table: Arc<dyn Table>,
 ) -> Result<Vec<Plan>> {
- let segment_locs = ctx.get_inserted_segment_locations()?;
+ let segment_locs = ctx.get_written_segment_locations()?;
  let mut plans = vec![];
 
  let table_meta = &table.get_table_info().meta;
@@ -296,7 +296,7 @@ async fn generate_refresh_virtual_column_plan(
  ctx: Arc<QueryContext>,
  desc: &RefreshDesc,
 ) -> Result<Option<Plan>> {
- let segment_locs = ctx.get_inserted_segment_locations()?;
+ let segment_locs = ctx.get_written_segment_locations()?;
 
  let table_info = ctx
  .get_table(&desc.catalog, &desc.database, &desc.table)

src/query/service/src/interpreters/interpreter_table_recluster.rs

@@ -217,7 +217,7 @@ impl ReclusterTableInterpreter {
 
  let complete_executor =
  PipelineCompleteExecutor::from_pipelines(pipelines, executor_settings)?;
- self.ctx.clear_inserted_segment_locations()?;
+ self.ctx.clear_written_segment_locations()?;
  self.ctx.set_executor(complete_executor.get_inner())?;
  complete_executor.execute()?;
  // make sure the executor is dropped before the next loop.

src/query/service/src/sessions/query_ctx.rs

@@ -151,7 +151,8 @@ pub struct QueryContext {
  query_settings: Arc<Settings>,
  fragment_id: Arc<AtomicUsize>,
  // Used by synchronized generate aggregating indexes when new data written.
- inserted_segment_locs: Arc<RwLock<HashSet<Location>>>,
+ written_segment_locs: Arc<RwLock<HashSet<Location>>>,
+ // Used by hilbert clustering when do recluster.
  selected_segment_locs: Arc<RwLock<HashSet<Location>>>,
  // Temp table for materialized CTE, first string is the database_name, second string is the table_name
  // All temp tables' catalog is `CATALOG_DEFAULT`, so we don't need to store it.
@@ -179,7 +180,7 @@ impl QueryContext {
  shared,
  query_settings,
  fragment_id: Arc::new(AtomicUsize::new(0)),
- inserted_segment_locs: Default::default(),
+ written_segment_locs: Default::default(),
  block_threshold: Default::default(),
  m_cte_temp_table: Default::default(),
  selected_segment_locs: Default::default(),
@@ -1206,21 +1207,21 @@ impl TableContext for QueryContext {
  })
  }
 
- fn add_inserted_segment_location(&self, segment_loc: Location) -> Result<()> {
- let mut segment_locations = self.inserted_segment_locs.write();
+ fn add_written_segment_location(&self, segment_loc: Location) -> Result<()> {
+ let mut segment_locations = self.written_segment_locs.write();
  segment_locations.insert(segment_loc);
  Ok(())
  }
 
- fn clear_inserted_segment_locations(&self) -> Result<()> {
- let mut segment_locations = self.inserted_segment_locs.write();
+ fn clear_written_segment_locations(&self) -> Result<()> {
+ let mut segment_locations = self.written_segment_locs.write();
  segment_locations.clear();
  Ok(())
  }
 
- fn get_inserted_segment_locations(&self) -> Result<Vec<Location>> {
+ fn get_written_segment_locations(&self) -> Result<Vec<Location>> {
  Ok(self
- .inserted_segment_locs
+ .written_segment_locs
  .read()
  .iter()
  .cloned()

src/query/service/tests/it/sql/exec/get_table_bind_test.rs

@@ -845,15 +845,15 @@ impl TableContext for CtxDelegation {
  todo!()
  }
 
- fn add_inserted_segment_location(&self, _segment_loc: Location) -> Result<()> {
+ fn add_written_segment_location(&self, _segment_loc: Location) -> Result<()> {
  todo!()
  }
 
- fn clear_inserted_segment_locations(&self) -> Result<()> {
+ fn clear_written_segment_locations(&self) -> Result<()> {
  todo!()
  }
 
- fn get_inserted_segment_locations(&self) -> Result<Vec<Location>> {
+ fn get_written_segment_locations(&self) -> Result<Vec<Location>> {
  todo!()
  }
 

src/query/service/tests/it/storages/fuse/operations/commit.rs

@@ -730,15 +730,15 @@ impl TableContext for CtxDelegation {
  HashMap::new()
  }
 
- fn add_inserted_segment_location(&self, _segment_loc: Location) -> Result<()> {
+ fn add_written_segment_location(&self, _segment_loc: Location) -> Result<()> {
  todo!()
  }
 
- fn clear_inserted_segment_locations(&self) -> Result<()> {
+ fn clear_written_segment_locations(&self) -> Result<()> {
  todo!()
  }
 
- fn get_inserted_segment_locations(&self) -> Result<Vec<Location>> {
+ fn get_written_segment_locations(&self) -> Result<Vec<Location>> {
  todo!()
  }
 

src/query/storages/fuse/src/operations/common/processors/sink_commit.rs

@@ -442,7 +442,7 @@ where F: SnapshotGenerator + Send + 'static
  metrics_inc_commit_copied_files(files.file_info.len() as u64);
  }
  for segment_loc in std::mem::take(&mut self.new_segment_locs).into_iter() {
- self.ctx.add_inserted_segment_location(segment_loc)?;
+ self.ctx.add_written_segment_location(segment_loc)?;
  }
 
  let target_descriptions = {