Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix: Remove namespace from DmlWrite and DmlDelete constructors

pull/24376/head
Carol (Nichols || Goulding) 2022-11-14 16:34:13 -05:00
parent f78195f7c7
commit 3943faf998
No known key found for this signature in database
GPG Key ID: E907EE5A736F87D4
15 changed files with 4 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dml/src/lib.rs
View File

@ -199,7 +199,6 @@ impl DmlWrite {
/// - a MutableBatch is empty /// - a MutableBatch is empty
/// - a MutableBatch lacks an i64 "time" column /// - a MutableBatch lacks an i64 "time" column
pub fn new( pub fn new(
_namespace: impl Into<String>,
namespace_id: NamespaceId, namespace_id: NamespaceId,
tables: HashMap<String, MutableBatch>, tables: HashMap<String, MutableBatch>,
table_ids: HashMap<String, TableId>, table_ids: HashMap<String, TableId>,
@ -337,7 +336,6 @@ pub struct DmlDelete {
impl DmlDelete { impl DmlDelete {
/// Create a new [`DmlDelete`] /// Create a new [`DmlDelete`]
pub fn new( pub fn new(
_namespace: impl Into<String>,
namespace_id: NamespaceId, namespace_id: NamespaceId,
predicate: DeletePredicate, predicate: DeletePredicate,
table_name: Option<NonEmptyString>, table_name: Option<NonEmptyString>,

12
ingester/src/data.rs
View File

@ -771,7 +771,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -815,7 +814,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w2 = DmlWrite::new( let w2 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(&mut *catalog.repositories().await, namespace.id, &batch).await, build_id_map(&mut *catalog.repositories().await, namespace.id, &batch).await,
@ -876,7 +874,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10\nmem foo=1 11", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10\nmem foo=1 11", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1004,7 +1001,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1023,7 +1019,6 @@ mod tests {
let batch = lines_to_batches("cpu foo=1 10", 1).unwrap(); let batch = lines_to_batches("cpu foo=1 10", 1).unwrap();
let w2 = DmlWrite::new( let w2 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1044,7 +1039,6 @@ mod tests {
std::mem::drop(repos); std::mem::drop(repos);
let batch = lines_to_batches("mem foo=1 30", 2).unwrap(); let batch = lines_to_batches("mem foo=1 30", 2).unwrap();
let w3 = DmlWrite::new( let w3 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(&mut *catalog.repositories().await, namespace.id, &batch).await, build_id_map(&mut *catalog.repositories().await, namespace.id, &batch).await,
@ -1297,7 +1291,6 @@ mod tests {
// write with sequence number 1 // write with sequence number 1
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1317,7 +1310,6 @@ mod tests {
// write with sequence number 2 // write with sequence number 2
let batch = lines_to_batches("mem foo=1 30\ncpu bar=1 20", 0).unwrap(); let batch = lines_to_batches("mem foo=1 30\ncpu bar=1 20", 0).unwrap();
let w2 = DmlWrite::new( let w2 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1418,7 +1410,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1432,7 +1423,6 @@ mod tests {
); );
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w2 = DmlWrite::new( let w2 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1618,7 +1608,6 @@ mod tests {
let batch = lines_to_batches("mem foo=1 10", 0).unwrap(); let batch = lines_to_batches("mem foo=1 10", 0).unwrap();
let w1 = DmlWrite::new( let w1 = DmlWrite::new(
"foo",
namespace.id, namespace.id,
batch.clone(), batch.clone(),
build_id_map(repos.deref_mut(), namespace.id, &batch).await, build_id_map(repos.deref_mut(), namespace.id, &batch).await,
@ -1663,7 +1652,6 @@ mod tests {
exprs: vec![], exprs: vec![],
}; };
let d1 = DmlDelete::new( let d1 = DmlDelete::new(
"foo",
NamespaceId::new(42), NamespaceId::new(42),
predicate, predicate,
Some(NonEmptyString::new("mem").unwrap()), Some(NonEmptyString::new("mem").unwrap()),

3
ingester/src/handler.rs
View File

@ -551,7 +551,6 @@ mod tests {
let ingest_ts1 = Time::from_timestamp_millis(42).unwrap(); let ingest_ts1 = Time::from_timestamp_millis(42).unwrap();
let write_operations = vec![DmlWrite::new( let write_operations = vec![DmlWrite::new(
"foo",
NamespaceId::new(1), NamespaceId::new(1),
lines_to_batches("cpu bar=2 20", 0).unwrap(), lines_to_batches("cpu bar=2 20", 0).unwrap(),
[("cpu".to_string(), TableId::new(1))].into_iter().collect(), [("cpu".to_string(), TableId::new(1))].into_iter().collect(),
@ -579,7 +578,6 @@ mod tests {
let ingest_ts1 = Time::from_timestamp_millis(42).unwrap(); let ingest_ts1 = Time::from_timestamp_millis(42).unwrap();
let write_operations = vec![DmlWrite::new( let write_operations = vec![DmlWrite::new(
"foo",
NamespaceId::new(1), NamespaceId::new(1),
lines_to_batches("cpu bar=2 20", 0).unwrap(), lines_to_batches("cpu bar=2 20", 0).unwrap(),
[("cpu".to_string(), TableId::new(1))].into_iter().collect(), [("cpu".to_string(), TableId::new(1))].into_iter().collect(),
@ -607,7 +605,6 @@ mod tests {
let ingest_ts1 = Time::from_timestamp_millis(42).unwrap(); let ingest_ts1 = Time::from_timestamp_millis(42).unwrap();
let write_operations = vec![DmlWrite::new( let write_operations = vec![DmlWrite::new(
"foo",
NamespaceId::new(1), NamespaceId::new(1),
lines_to_batches("cpu bar=2 20", 0).unwrap(), lines_to_batches("cpu bar=2 20", 0).unwrap(),
[("cpu".to_string(), TableId::new(1))].into_iter().collect(), [("cpu".to_string(), TableId::new(1))].into_iter().collect(),

2
ingester/src/stream_handler/handler.rs
View File

@ -553,7 +553,6 @@ mod tests {
42, 42,
); );
DmlWrite::new( DmlWrite::new(
"deprecated",
NamespaceId::new(namespace_id), NamespaceId::new(namespace_id),
tables, tables,
ids, ids,
@ -577,7 +576,6 @@ mod tests {
42, 42,
); );
DmlDelete::new( DmlDelete::new(
"deprecated",
NamespaceId::new(namespace_id), NamespaceId::new(namespace_id),
pred, pred,
None, None,

1
ingester/src/stream_handler/sink_instrumentation.rs
View File

@ -278,7 +278,6 @@ mod tests {
.map(|(i, v)| (v.clone(), TableId::new(i as _))) .map(|(i, v)| (v.clone(), TableId::new(i as _)))
.collect(); .collect();
DmlWrite::new( DmlWrite::new(
"bananas",
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
ids, ids,

3
ingester/src/test_util.rs
View File

@ -588,7 +588,7 @@ pub(crate) fn make_partitions(
pub(crate) fn make_write_op( pub(crate) fn make_write_op(
partition_key: &PartitionKey, partition_key: &PartitionKey,
shard_index: ShardIndex, shard_index: ShardIndex,
namespace: &str, _namespace: &str,
namespace_id: NamespaceId, namespace_id: NamespaceId,
table_id: TableId, table_id: TableId,
sequence_number: i64, sequence_number: i64,
@ -600,7 +600,6 @@ pub(crate) fn make_write_op(
let ids = [(TEST_TABLE.into(), table_id)].into_iter().collect(); let ids = [(TEST_TABLE.into(), table_id)].into_iter().collect();
DmlWrite::new( DmlWrite::new(
namespace.to_string(),
namespace_id, namespace_id,
tables, tables,
ids, ids,

1
ingester/tests/common/mod.rs
View File

@ -294,7 +294,6 @@ impl TestContext {
.await; .await;
self.enqueue_write(DmlWrite::new( self.enqueue_write(DmlWrite::new(
namespace,
namespace_id, namespace_id,
lines_to_batches(lp, 0).unwrap(), lines_to_batches(lp, 0).unwrap(),
ids.clone(), ids.clone(),

1
mutable_batch_tests/benches/write_pb.rs
View File

@ -20,7 +20,6 @@ fn generate_pbdata_bytes() -> Vec<(String, (usize, Bytes))> {
.collect(); .collect();
let write = DmlWrite::new( let write = DmlWrite::new(
"test_db",
NamespaceId::new(42), NamespaceId::new(42),
batches, batches,
ids, ids,

2
query_tests/src/scenarios/util.rs
View File

@ -835,7 +835,6 @@ impl MockIngester {
0, 0,
); );
let op = DmlOperation::Write(DmlWrite::new( let op = DmlOperation::Write(DmlWrite::new(
self.ns.namespace.name.clone(),
self.ns.namespace.id, self.ns.namespace.id,
mutable_batches, mutable_batches,
ids, ids,
@ -862,7 +861,6 @@ impl MockIngester {
0, 0,
); );
DmlOperation::Delete(DmlDelete::new( DmlOperation::Delete(DmlDelete::new(
self.ns.namespace.name.clone(),
self.ns.namespace.id, self.ns.namespace.id,
predicate, predicate,
Some(NonEmptyString::new(delete_table_name).unwrap()), Some(NonEmptyString::new(delete_table_name).unwrap()),

2
router/src/dml_handlers/sharded_write_buffer.rs
View File

@ -131,7 +131,6 @@ where
let iter = collated.into_iter().map(|(shard, batch)| { let iter = collated.into_iter().map(|(shard, batch)| {
let dml = DmlWrite::new( let dml = DmlWrite::new(
namespace,
namespace_id, namespace_id,
batch, batch,
table_ids.remove(&shard).unwrap(), table_ids.remove(&shard).unwrap(),
@ -168,7 +167,6 @@ where
let shards = self.sharder.shard(table_name, namespace, &predicate); let shards = self.sharder.shard(table_name, namespace, &predicate);
let dml = DmlDelete::new( let dml = DmlDelete::new(
namespace,
namespace_id, namespace_id,
predicate, predicate,
NonEmptyString::new(table_name), NonEmptyString::new(table_name),

3
write_buffer/src/codec.rs
View File

@ -178,7 +178,6 @@ pub fn decode(
}; };
Ok(DmlOperation::Write(DmlWrite::new( Ok(DmlOperation::Write(DmlWrite::new(
"deprecated",
NamespaceId::new(write.database_id), NamespaceId::new(write.database_id),
tables, tables,
ids.into_iter().map(|(k, v)| (k, TableId::new(v))).collect(), ids.into_iter().map(|(k, v)| (k, TableId::new(v))).collect(),
@ -193,7 +192,6 @@ pub fn decode(
.map_err(WriteBufferError::invalid_data)?; .map_err(WriteBufferError::invalid_data)?;
Ok(DmlOperation::Delete(DmlDelete::new( Ok(DmlOperation::Delete(DmlDelete::new(
"deprecated",
NamespaceId::new(delete.database_id), NamespaceId::new(delete.database_id),
predicate, predicate,
NonEmptyString::new(delete.table_name), NonEmptyString::new(delete.table_name),
@ -309,7 +307,6 @@ mod tests {
let (data, ids) = lp_to_batches("platanos great=42 100\nbananas greatness=1000 100"); let (data, ids) = lp_to_batches("platanos great=42 100\nbananas greatness=1000 100");
let w = DmlWrite::new( let w = DmlWrite::new(
"bananas",
NamespaceId::new(42), NamespaceId::new(42),
data, data,
ids, ids,

4
write_buffer/src/core.rs
View File

@ -392,7 +392,7 @@ pub mod test_utils {
/// Writes line protocol and returns the [`DmlWrite`] that was written. /// Writes line protocol and returns the [`DmlWrite`] that was written.
pub async fn write( pub async fn write(
namespace: &str, _namespace: &str,
writer: &impl WriteBufferWriting, writer: &impl WriteBufferWriting,
lp: &str, lp: &str,
shard_index: ShardIndex, shard_index: ShardIndex,
@ -402,7 +402,6 @@ pub mod test_utils {
let (tables, names) = lp_to_batches(lp); let (tables, names) = lp_to_batches(lp);
let write = DmlWrite::new( let write = DmlWrite::new(
namespace,
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
names, names,
@ -1251,7 +1250,6 @@ pub mod test_utils {
let (tables, names) = lp_to_batches("upc user=1 100"); let (tables, names) = lp_to_batches("upc user=1 100");
let write = DmlWrite::new( let write = DmlWrite::new(
"foo",
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
names, names,

6
write_buffer/src/kafka/mod.rs
View File

@ -829,7 +829,7 @@ mod tests {
} }
async fn write( async fn write(
namespace: &str, _namespace: &str,
producer: &RSKafkaProducer, producer: &RSKafkaProducer,
trace_collector: &Arc<RingBufferTraceCollector>, trace_collector: &Arc<RingBufferTraceCollector>,
shard_index: ShardIndex, shard_index: ShardIndex,
@ -838,7 +838,6 @@ mod tests {
let span_ctx = SpanContext::new(Arc::clone(trace_collector) as Arc<_>); let span_ctx = SpanContext::new(Arc::clone(trace_collector) as Arc<_>);
let (tables, names) = lp_to_batches("table foo=1"); let (tables, names) = lp_to_batches("table foo=1");
let write = DmlWrite::new( let write = DmlWrite::new(
namespace,
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
names, names,
@ -850,14 +849,13 @@ mod tests {
} }
async fn delete( async fn delete(
namespace: &str, _namespace: &str,
producer: &RSKafkaProducer, producer: &RSKafkaProducer,
trace_collector: &Arc<RingBufferTraceCollector>, trace_collector: &Arc<RingBufferTraceCollector>,
shard_index: ShardIndex, shard_index: ShardIndex,
) -> DmlMeta { ) -> DmlMeta {
let span_ctx = SpanContext::new(Arc::clone(trace_collector) as Arc<_>); let span_ctx = SpanContext::new(Arc::clone(trace_collector) as Arc<_>);
let op = DmlOperation::Delete(DmlDelete::new( let op = DmlOperation::Delete(DmlDelete::new(
namespace,
NamespaceId::new(42), NamespaceId::new(42),
DeletePredicate { DeletePredicate {
range: TimestampRange::new(0, 1), range: TimestampRange::new(0, 1),

2
write_buffer/src/kafka/record_aggregator.rs
View File

@ -207,7 +207,6 @@ mod tests {
use super::*; use super::*;
const NAMESPACE: &str = "bananas";
const SHARD_INDEX: ShardIndex = ShardIndex::new(42); const SHARD_INDEX: ShardIndex = ShardIndex::new(42);
const TIMESTAMP_MILLIS: i64 = 1659990497000; const TIMESTAMP_MILLIS: i64 = 1659990497000;
@ -233,7 +232,6 @@ mod tests {
.collect(); .collect();
DmlOperation::Write(DmlWrite::new( DmlOperation::Write(DmlWrite::new(
NAMESPACE.to_string(),
NamespaceId::new(42), NamespaceId::new(42),
m, m,
ids, ids,

2
write_buffer/src/mock.rs
View File

@ -178,7 +178,6 @@ impl MockBufferSharedState {
let (tables, names) = lp_to_batches(lp); let (tables, names) = lp_to_batches(lp);
let meta = DmlMeta::sequenced(sequence, iox_time::Time::from_timestamp_nanos(0), None, 0); let meta = DmlMeta::sequenced(sequence, iox_time::Time::from_timestamp_nanos(0), None, 0);
self.push_write(DmlWrite::new( self.push_write(DmlWrite::new(
"foo",
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
names, names,
@ -905,7 +904,6 @@ mod tests {
let (tables, names) = lp_to_batches("upc user=1 100"); let (tables, names) = lp_to_batches("upc user=1 100");
let operation = DmlOperation::Write(DmlWrite::new( let operation = DmlOperation::Write(DmlWrite::new(
"test_db",
NamespaceId::new(42), NamespaceId::new(42),
tables, tables,
names, names,