Skip to content

Commit cdeb9e8

Browse files
authored
Check memtable against current minute when loading (#468)
1 parent fbaec27 commit cdeb9e8

File tree

1 file changed

+13
-3
lines changed

1 file changed

+13
-3
lines changed

server/src/query.rs

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919
mod filter_optimizer;
2020
mod table_provider;
2121

22-
use chrono::TimeZone;
2322
use chrono::{DateTime, Utc};
23+
use chrono::{TimeZone, Timelike};
2424
use datafusion::arrow::datatypes::Schema;
2525
use datafusion::arrow::record_batch::RecordBatch;
2626
use datafusion::datasource::file_format::parquet::ParquetFormat;
@@ -128,8 +128,18 @@ impl Query {
128128
) -> Result<(Vec<RecordBatch>, Vec<String>), ExecuteError> {
129129
let ctx = self.create_session_context();
130130
let remote_listing_table = self._remote_query(storage)?;
131-
let memtable =
132-
crate::event::STREAM_WRITERS.recordbatches_cloned(&self.stream_name, &self.schema);
131+
132+
let current_minute = Utc::now()
133+
.with_second(0)
134+
.and_then(|x| x.with_nanosecond(0))
135+
.expect("zeroed value is valid");
136+
137+
let memtable = if self.end > current_minute {
138+
crate::event::STREAM_WRITERS.recordbatches_cloned(&self.stream_name, &self.schema)
139+
} else {
140+
None
141+
};
142+
133143
let table =
134144
QueryTableProvider::try_new(memtable, remote_listing_table, self.schema.clone())?;
135145

0 commit comments

Comments
 (0)