Skip to content

Commit 669b6ca

Browse files
parmesantparmesant
committed
introduce flag for migration
1 parent 6587254 commit 669b6ca

9 files changed

Lines changed: 21 additions & 19 deletions

File tree

‎src/catalog/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ pub async fn update_snapshot(
117117
let mut meta: ObjectStoreFormat = serde_json::from_slice(
118118
&PARSEABLE
119119
.metastore
120-
.get_stream_json(stream_name, false, tenant_id)
120+
.get_stream_json(stream_name, false, tenant_id, false)
121121
.await
122122
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
123123
)?;
@@ -507,7 +507,7 @@ pub async fn remove_manifest_from_snapshot(
507507
let mut meta: ObjectStoreFormat = serde_json::from_slice(
508508
&PARSEABLE
509509
.metastore
510-
.get_stream_json(stream_name, false, tenant_id)
510+
.get_stream_json(stream_name, false, tenant_id, false)
511511
.await
512512
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
513513
)?;

‎src/enterprise/utils.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ pub async fn fetch_parquet_file_paths(
8484
let object_store_format: ObjectStoreFormat = serde_json::from_slice(
8585
&PARSEABLE
8686
.metastore
87-
.get_stream_json(stream, false, tenant_id)
87+
.get_stream_json(stream, false, tenant_id, false)
8888
.await
8989
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
9090
)?;

‎src/handlers/http/logstream.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -466,7 +466,7 @@ pub async fn put_stream_hot_tier(
466466
let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
467467
&PARSEABLE
468468
.metastore
469-
.get_stream_json(&stream_name, false, &tenant_id)
469+
.get_stream_json(&stream_name, false, &tenant_id, false)
470470
.await?,
471471
)?;
472472
stream_metadata.hot_tier_enabled = true;
@@ -572,7 +572,7 @@ pub async fn delete_stream_hot_tier(
572572
let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
573573
&PARSEABLE
574574
.metastore
575-
.get_stream_json(&stream_name, false, &tenant_id)
575+
.get_stream_json(&stream_name, false, &tenant_id, false)
576576
.await?,
577577
)?;
578578
stream_metadata.hot_tier_enabled = false;

‎src/metastore/metastore_traits.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -229,6 +229,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
229229
stream_name: &str,
230230
get_base: bool,
231231
tenant_id: &Option<String>,
232+
is_migration: bool
232233
) -> Result<Bytes, MetastoreError>;
233234
async fn put_stream_json(
234235
&self,

‎src/metastore/metastores/object_store_metastore.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -870,6 +870,7 @@ impl Metastore for ObjectStoreMetastore {
870870
stream_name: &str,
871871
get_base: bool,
872872
tenant_id: &Option<String>,
873+
_is_migration: bool,
873874
) -> Result<Bytes, MetastoreError> {
874875
let tenant = tenant_id.as_deref().unwrap_or("");
875876
let path = if get_base {

‎src/migration/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -307,7 +307,7 @@ async fn fetch_or_create_stream_metadata(
307307
) -> anyhow::Result<Bytes> {
308308
if let Ok(stream_metadata) = PARSEABLE
309309
.metastore
310-
.get_stream_json(stream, false, tenant_id)
310+
.get_stream_json(stream, false, tenant_id, true)
311311
.await
312312
{
313313
Ok(stream_metadata)

‎src/query/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -758,7 +758,7 @@ pub async fn get_manifest_list(
758758
let object_store_format: ObjectStoreFormat = serde_json::from_slice(
759759
&PARSEABLE
760760
.metastore
761-
.get_stream_json(stream_name, false, tenant_id)
761+
.get_stream_json(stream_name, false, tenant_id, false)
762762
.await?,
763763
)?;
764764

‎src/query/stream_schema_provider.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -543,7 +543,7 @@ impl TableProvider for StandardTableProvider {
543543
let object_store_format: ObjectStoreFormat = serde_json::from_slice(
544544
&PARSEABLE
545545
.metastore
546-
.get_stream_json(&self.stream, false, &self.tenant_id)
546+
.get_stream_json(&self.stream, false, &self.tenant_id, false)
547547
.await
548548
.map_err(|e| DataFusionError::Plan(e.to_string()))?,
549549
)

‎src/storage/object_storage.rs‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -467,7 +467,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
467467
let mut format: ObjectStoreFormat = serde_json::from_slice(
468468
&PARSEABLE
469469
.metastore
470-
.get_stream_json(stream_name, false, tenant_id)
470+
.get_stream_json(stream_name, false, tenant_id, false)
471471
.await
472472
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
473473
)?;
@@ -490,7 +490,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
490490
let mut format: ObjectStoreFormat = serde_json::from_slice(
491491
&PARSEABLE
492492
.metastore
493-
.get_stream_json(stream_name, false, tenant_id)
493+
.get_stream_json(stream_name, false, tenant_id, false)
494494
.await
495495
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
496496
)?;
@@ -513,7 +513,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
513513
let mut format: ObjectStoreFormat = serde_json::from_slice(
514514
&PARSEABLE
515515
.metastore
516-
.get_stream_json(stream_name, false, tenant_id)
516+
.get_stream_json(stream_name, false, tenant_id, false)
517517
.await
518518
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
519519
)?;
@@ -537,7 +537,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
537537
let mut format: ObjectStoreFormat = serde_json::from_slice(
538538
&PARSEABLE
539539
.metastore
540-
.get_stream_json(stream_name, false, tenant_id)
540+
.get_stream_json(stream_name, false, tenant_id, false)
541541
.await
542542
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
543543
)?;
@@ -583,7 +583,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
583583
let mut format: ObjectStoreFormat = serde_json::from_slice(
584584
&PARSEABLE
585585
.metastore
586-
.get_stream_json(stream_name, false, tenant_id)
586+
.get_stream_json(stream_name, false, tenant_id, false)
587587
.await
588588
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
589589
)?;
@@ -606,7 +606,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
606606
let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
607607
&PARSEABLE
608608
.metastore
609-
.get_stream_json(stream_name, false, tenant_id)
609+
.get_stream_json(stream_name, false, tenant_id, false)
610610
.await
611611
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
612612
)?;
@@ -629,7 +629,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
629629
let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
630630
&PARSEABLE
631631
.metastore
632-
.get_stream_json(stream_name, false, tenant_id)
632+
.get_stream_json(stream_name, false, tenant_id, false)
633633
.await
634634
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
635635
)?;
@@ -649,15 +649,15 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
649649
) -> Result<ObjectStoreFormat, ObjectStorageError> {
650650
let stream_metadata = match PARSEABLE
651651
.metastore
652-
.get_stream_json(stream_name, false, tenant_id)
652+
.get_stream_json(stream_name, false, tenant_id, false)
653653
.await
654654
{
655655
Ok(data) => data,
656656
Err(_) => {
657657
// get the base stream metadata
658658
let bytes = PARSEABLE
659659
.metastore
660-
.get_stream_json(stream_name, true, tenant_id)
660+
.get_stream_json(stream_name, true, tenant_id, false)
661661
.await
662662
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
663663

@@ -705,7 +705,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
705705
) -> Result<Bytes, ObjectStorageError> {
706706
if let Ok(querier_stream_json_bytes) = PARSEABLE
707707
.metastore
708-
.get_stream_json(stream_name, true, tenant_id)
708+
.get_stream_json(stream_name, true, tenant_id, false)
709709
.await
710710
{
711711
let querier_stream_metadata =
@@ -737,7 +737,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
737737
if PARSEABLE.get_stream(stream_name, tenant_id).is_ok() {
738738
let stream_metadata_bytes = PARSEABLE
739739
.metastore
740-
.get_stream_json(stream_name, false, tenant_id)
740+
.get_stream_json(stream_name, false, tenant_id, false)
741741
.await
742742
.map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
743743
return Ok(stream_metadata_bytes);

0 commit comments

Comments
 (0)