-
Notifications
You must be signed in to change notification settings - Fork 70
/
exclude.rs
268 lines (237 loc) · 9.46 KB
/
exclude.rs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
static USAGE: &str = r#"
Removes a set of CSV data from another set based on the specified columns.
Also can compute the intersection of two CSV sets with the -v flag.
Matching is always done by ignoring leading and trailing whitespace. By default,
matching is done case sensitively, but this can be disabled with the --ignore-case
flag.
The columns arguments specify the columns to match for each input. Columns can
be referenced by name or index, starting at 1. Specify multiple columns by
separating them with a comma. Specify a range of columns with `-`. Both
columns1 and columns2 must specify exactly the same number of columns.
(See 'qsv select --help' for the full syntax.)
Examples:
qsv exclude id records.csv id previously-processed.csv
qsv exclude col1,col2 records.csv col1,col2 previously-processed.csv
qsv exclude col1-col5 records.csv col1-col5 previously-processed.csv
qsv exclude id records.csv id previously-processed.csv > new-records.csv
qsv exclude id records.csv id previously-processed.csv --output new-records.csv
qsv exclude -v id records.csv id previously-processed.csv -o intersection.csv
qsv exclude --ignore-case id records.csv id previously-processed.csv
qsv exclude id records.csv id previously-processed.csv |
qsv sort > new-sorted-records.csv
qsv exclude id records.csv id previously-processed.csv | qsv sort |
qsv --sorted dedup > new-sorted-deduped-records.csv
For more examples, see https://github.com/jqnatividad/qsv/blob/master/tests/test_exclude.rs.
Usage:
qsv exclude [options] <columns1> <input1> <columns2> <input2>
qsv exclude --help
input arguments:
<input1> is the file from which data will be removed.
<input2> is the file containing the data to be removed from <input1>
e.g. 'qsv exclude id records.csv id previously-processed.csv'
exclude options:
-i, --ignore-case When set, matching is done case insensitively.
-v When set, matching rows will be the only ones included,
forming set intersection, instead of the ones discarded.
Common options:
-h, --help Display this message
-o, --output <file> Write output to <file> instead of stdout.
-n, --no-headers When set, the first row will not be interpreted
as headers. (i.e., They are not searched, analyzed,
sliced, etc.)
-d, --delimiter <arg> The field delimiter for reading CSV data.
Must be a single character. (default: ,)
"#;
use std::{collections::hash_map::Entry, fs, io, str};
use ahash::AHashMap;
use byteorder::{BigEndian, WriteBytesExt};
use serde::Deserialize;
use crate::{
config::{Config, Delimiter},
index::Indexed,
select::{SelectColumns, Selection},
util,
util::ByteString,
CliResult,
};
#[derive(Deserialize)]
struct Args {
arg_columns1: SelectColumns,
arg_input1: String,
arg_columns2: SelectColumns,
arg_input2: String,
flag_v: bool,
flag_output: Option<String>,
flag_no_headers: bool,
flag_ignore_case: bool,
flag_delimiter: Option<Delimiter>,
}
pub fn run(argv: &[&str]) -> CliResult<()> {
let args: Args = util::get_args(USAGE, argv)?;
let mut state = args.new_io_state()?;
state.write_headers()?;
state.exclude(args.flag_v)
}
struct IoState<R, W: io::Write> {
wtr: csv::Writer<W>,
rdr1: csv::Reader<R>,
sel1: Selection,
rdr2: csv::Reader<R>,
sel2: Selection,
no_headers: bool,
casei: bool,
}
impl<R: io::Read + io::Seek, W: io::Write> IoState<R, W> {
fn write_headers(&mut self) -> CliResult<()> {
if !self.no_headers {
let headers = self.rdr1.byte_headers()?.clone();
self.wtr.write_record(&headers)?;
}
Ok(())
}
fn exclude(mut self, invert: bool) -> CliResult<()> {
// amortize allocations
#[allow(unused_assignments)]
let mut curr_row = csv::ByteRecord::new();
let validx = ValueIndex::new(self.rdr2, &self.sel2, self.casei)?;
for row in self.rdr1.byte_records() {
curr_row = row?;
let key = get_row_key(&self.sel1, &curr_row, self.casei);
if let Some(_rows) = validx.values.get(&key) {
if invert {
self.wtr.write_record(curr_row.iter())?;
}
} else if !invert {
self.wtr.write_record(curr_row.iter())?;
}
}
Ok(())
}
}
impl Args {
fn new_io_state(&self) -> CliResult<IoState<fs::File, Box<dyn io::Write + 'static>>> {
let rconf1 = Config::new(&Some(self.arg_input1.clone()))
.delimiter(self.flag_delimiter)
.no_headers(self.flag_no_headers)
.select(self.arg_columns1.clone());
let rconf2 = Config::new(&Some(self.arg_input2.clone()))
.delimiter(self.flag_delimiter)
.no_headers(self.flag_no_headers)
.select(self.arg_columns2.clone());
let mut rdr1 = rconf1.reader_file()?;
let mut rdr2 = rconf2.reader_file()?;
let (sel1, sel2) = self.get_selections(&rconf1, &mut rdr1, &rconf2, &mut rdr2)?;
Ok(IoState {
wtr: Config::new(&self.flag_output).writer()?,
rdr1,
sel1,
rdr2,
sel2,
no_headers: rconf1.no_headers,
casei: self.flag_ignore_case,
})
}
#[allow(clippy::unused_self)]
fn get_selections<R: io::Read>(
&self,
rconf1: &Config,
rdr1: &mut csv::Reader<R>,
rconf2: &Config,
rdr2: &mut csv::Reader<R>,
) -> CliResult<(Selection, Selection)> {
let headers1 = rdr1.byte_headers()?;
let headers2 = rdr2.byte_headers()?;
let select1 = rconf1.selection(headers1)?;
let select2 = rconf2.selection(headers2)?;
if select1.len() != select2.len() {
return fail_incorrectusage_clierror!(
"Column selections must have the same number of columns, but found column \
selections with {} and {} columns.",
select1.len(),
select2.len()
);
}
Ok((select1, select2))
}
}
#[allow(dead_code)]
struct ValueIndex<R> {
// This maps tuples of values to corresponding rows.
values: AHashMap<Vec<ByteString>, Vec<usize>>,
idx: Indexed<R, io::Cursor<Vec<u8>>>,
num_rows: usize,
}
impl<R: io::Read + io::Seek> ValueIndex<R> {
fn new(mut rdr: csv::Reader<R>, sel: &Selection, casei: bool) -> CliResult<ValueIndex<R>> {
let mut val_idx = AHashMap::with_capacity(10000);
let mut row_idx = io::Cursor::new(Vec::with_capacity(8 * 10000));
let (mut rowi, mut count) = (0_usize, 0_usize);
// This logic is kind of tricky. Basically, we want to include
// the header row in the line index (because that's what csv::index
// does), but we don't want to include header values in the ValueIndex.
if rdr.has_headers() {
// ... so if there are headers, we make sure that we've parsed
// them, and write the offset of the header row to the index.
rdr.byte_headers()?;
row_idx.write_u64::<BigEndian>(0)?;
count += 1;
} else {
// ... and if there are no headers, we seek to the beginning and
// index everything.
let mut pos = csv::Position::new();
pos.set_byte(0);
rdr.seek(pos)?;
}
let mut row = csv::ByteRecord::new();
while rdr.read_byte_record(&mut row)? {
// This is a bit hokey. We're doing this manually instead of using
// the `csv-index` crate directly so that we can create both
// indexes in one pass.
row_idx.write_u64::<BigEndian>(row.position().unwrap().byte())?;
let fields: Vec<_> = sel
.select(&row)
.map(|v| util::transform(v, casei))
.collect();
match val_idx.entry(fields) {
Entry::Vacant(v) => {
let mut rows = Vec::with_capacity(4);
rows.push(rowi);
v.insert(rows);
},
Entry::Occupied(mut v) => {
v.get_mut().push(rowi);
},
}
rowi += 1;
count += 1;
}
row_idx.write_u64::<BigEndian>(count as u64)?;
let idx = Indexed::open(rdr, io::Cursor::new(row_idx.into_inner()))?;
Ok(ValueIndex {
values: val_idx,
idx,
num_rows: rowi,
})
}
}
use std::fmt;
impl<R> fmt::Debug for ValueIndex<R> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
// Sort the values by order of first appearance.
let mut kvs = self.values.iter().collect::<Vec<_>>();
kvs.sort_by(|&(_, v1), &(_, v2)| v1[0].cmp(&v2[0]));
for (keys, rows) in kvs {
// This is just for debugging, so assume Unicode for now.
let keys = keys
.iter()
.map(|k| String::from_utf8(k.clone()).unwrap())
.collect::<Vec<_>>();
writeln!(f, "({}) => {rows:?}", keys.join(", "))?;
}
Ok(())
}
}
#[inline]
fn get_row_key(sel: &Selection, row: &csv::ByteRecord, casei: bool) -> Vec<ByteString> {
sel.select(row).map(|v| util::transform(v, casei)).collect()
}