github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-19 16:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(import): honor custom title rows in Excel streaming import

Browse Source
This commit is contained in:
lxk955
2026-10-10 11:23:43 +08:00
committed by GitHub
parent 20fc49d7db
commit fa7c9ef49e
1 changed files with 178 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/data/table_import.rs
+178 -37
View File
@@ -3462,6 +3462,7 @@ fn parse_xlsx_preview_file_with_options(
let date_1904 = xlsx_workbook_uses_1904_date_system(&workbook_xml);
let empty_string_as_null = options.empty_string_as_null.unwrap_or(true);
let row_range = effective_import_row_range(options)?;
let min_import_row = row_range.title_row.unwrap_or(row_range.data_start_row);
let preview_limit = preview_limit.max(1);
let preview_end_from = |first_row: usize| {
let preview_last_row = first_row.saturating_add(preview_limit.saturating_sub(1));
@@ -3506,7 +3507,9 @@ fn parse_xlsx_preview_file_with_options(
{
break;
}
observed_max_row = observed_max_row.max(current_row);
if current_row >= min_import_row {
observed_max_row = observed_max_row.max(current_row);
}
}
Ok(Event::Empty(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => {
let position = xml_attr_value(&reader, &element, b"r")
@@ -3515,13 +3518,14 @@ fn parse_xlsx_preview_file_with_options(
.unwrap_or_else(|| (current_row.max(1), current_column.saturating_add(1).max(1)));
current_row = position.0;
current_column = position.1;
observed_min_row = observed_min_row.min(position.0);
requested_last_row = preview_end_from(row_range.data_start_row.max(observed_min_row));
observed_min_column = observed_min_column.min(position.1);
observed_max_row = observed_max_row.max(position.0);
observed_max_column = observed_max_column.max(position.1);
if current_row >= min_import_row {
observed_min_row = observed_min_row.min(position.0);
requested_last_row = preview_end_from(row_range.data_start_row.max(observed_min_row));
observed_min_column = observed_min_column.min(position.1);
observed_max_row = observed_max_row.max(position.0);
observed_max_column = observed_max_column.max(position.1);
}
}
Ok(Event::Empty(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => continue,
Ok(Event::Start(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => {
let position = xml_attr_value(&reader, &element, b"r")
.as_deref()
@@ -3570,15 +3574,17 @@ fn parse_xlsx_preview_file_with_options(
}
Ok(Event::End(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => {
if let Some((row, column)) = current_position.take() {
observed_min_row = observed_min_row.min(row);
requested_last_row = preview_end_from(row_range.data_start_row.max(observed_min_row));
observed_min_column = observed_min_column.min(column);
observed_max_column = observed_max_column.max(column);
observed_max_row = observed_max_row.max(row);
if row == row_range.title_row.unwrap_or_default()
|| (row >= row_range.data_start_row && row <= requested_last_row)
{
raw_cells.insert((row, column), std::mem::take(&mut current_cell));
if row >= min_import_row {
observed_min_row = observed_min_row.min(row);
requested_last_row = preview_end_from(row_range.data_start_row.max(observed_min_row));
observed_min_column = observed_min_column.min(column);
observed_max_column = observed_max_column.max(column);
observed_max_row = observed_max_row.max(row);
if row == row_range.title_row.unwrap_or_default()
|| (row >= row_range.data_start_row && row <= requested_last_row)
{
raw_cells.insert((row, column), std::mem::take(&mut current_cell));
}
}
}
inline_phonetic_depth = 0;
@@ -3616,8 +3622,10 @@ fn parse_xlsx_preview_file_with_options(
));
}
let dimension_end_column = dimension
.filter(|((dimension_start_row, dimension_start_column), _)| {
*dimension_start_row == start_row && *dimension_start_column == start_column
.filter(|((dimension_start_row, dimension_start_column), (dimension_end_row, _))| {
*dimension_start_row <= start_row
&& start_row <= *dimension_end_row
&& *dimension_start_column == start_column
})
.map(|(_, (_, end_column))| end_column)
.filter(|end_column| {
@@ -3877,6 +3885,7 @@ struct XlsxStreamRowsState {
start_row: Option<usize>,
start_column: usize,
declared_column_count: Option<usize>,
has_expected_columns: bool,
columns: Vec<String>,
header_sent: bool,
pending_rows: Vec<Vec<serde_json::Value>>,
@@ -3896,6 +3905,7 @@ impl XlsxStreamRowsState {
batch_size: usize,
) -> Self {
let batch_size = batch_size.max(1);
let has_expected_columns = expected_columns.is_some();
Self {
sender,
row_range,
@@ -3903,6 +3913,7 @@ impl XlsxStreamRowsState {
start_row: None,
start_column: 0,
declared_column_count: None,
has_expected_columns,
columns: expected_columns.unwrap_or_default(),
header_sent: false,
pending_rows: Vec::with_capacity(batch_size),
@@ -3922,8 +3933,10 @@ impl XlsxStreamRowsState {
let dimension = self.dimension.filter(|((start_row, start_column), (end_row, end_column))| {
let column_count = end_column.saturating_sub(*start_column).saturating_add(1);
let row_count = end_row.saturating_sub(*start_row).saturating_add(1);
*start_row == first_row
&& *start_column == first_column
*start_row <= first_row
&& first_row <= *end_row
&& *start_column <= first_column
&& first_column <= *end_column
&& column_count <= MAX_FAST_PREVIEW_CELLS
&& expected_column_count
.map_or(column_count.saturating_mul(row_count) <= MAX_FAST_PREVIEW_CELLS, |expected| {
@@ -3931,11 +3944,39 @@ impl XlsxStreamRowsState {
})
});
self.start_row = Some(first_row);
self.start_column = first_column;
self.start_column =
dimension.map(|((_, start_column), _)| start_column.min(first_column)).unwrap_or(first_column);
self.declared_column_count = dimension
.map(|((_, start_column), (_, end_column))| end_column.saturating_sub(start_column).saturating_add(1));
}
fn ensure_column_range(&mut self, absolute_row: usize, absolute_column: usize) -> Result<(), String> {
self.initialize_range(absolute_row, absolute_column);
if absolute_column < self.start_column {
if self.header_sent {
return Err(format!(
"Excel row {absolute_row} contains a cell before the confirmed import range start column"
));
}
let shift = self.start_column.saturating_sub(absolute_column);
self.start_column = absolute_column;
if shift > 0 {
let mut shifted = vec![serde_json::Value::Null; shift];
shifted.append(&mut self.current_values);
self.current_values = shifted;
if !self.has_expected_columns && !self.columns.is_empty() {
let mut shifted_columns: Vec<String> = (1..=shift).map(|index| format!("column_{index}")).collect();
shifted_columns.append(&mut self.columns);
self.columns = unique_import_headers(shifted_columns);
}
if let Some(count) = self.declared_column_count.as_mut() {
*count = count.saturating_add(shift);
}
}
}
Ok(())
}
fn selected_range_finished(&self, absolute_row: usize) -> bool {
self.row_range.last_data_row.is_some_and(|last| absolute_row > last)
}
@@ -3946,7 +3987,7 @@ impl XlsxStreamRowsState {
absolute_column: usize,
text_source_columns: &HashSet<String>,
) -> bool {
self.initialize_range(absolute_row, absolute_column);
let _ = self.ensure_column_range(absolute_row, absolute_column);
absolute_column
.checked_sub(self.start_column)
.and_then(|offset| self.columns.get(offset))
@@ -3961,15 +4002,13 @@ impl XlsxStreamRowsState {
has_content: bool,
progress: u64,
) -> Result<(), String> {
self.initialize_range(absolute_row, absolute_column);
if self.current_row != Some(absolute_row) {
self.flush_current_row(progress)?;
self.current_row = Some(absolute_row);
}
self.ensure_column_range(absolute_row, absolute_column)?;
self.current_row_has_content |= has_content;
let column_offset = absolute_column.checked_sub(self.start_column).ok_or_else(|| {
format!("Excel row {absolute_row} contains a cell before the detected import range start column")
})?;
let column_offset = absolute_column.saturating_sub(self.start_column);
if column_offset >= MAX_FAST_PREVIEW_CELLS {
return Err(format!("Excel import column {} exceeds the safety limit", column_offset + 1));
}
@@ -4160,6 +4199,7 @@ fn stream_xlsx_rows_to_channel_with_control(
&mut on_shared_progress,
)?;
let row_range = effective_import_row_range(options)?;
let min_import_row = row_range.title_row.unwrap_or(row_range.data_start_row);
let sheet = zip.by_name(&sheet_path).map_err(|error| error.to_string())?;
let uncompressed_sheet_bytes = sheet.size().max(1);
let mut reader = XmlReader::from_reader(BufReader::new(sheet));
@@ -4210,7 +4250,9 @@ fn stream_xlsx_rows_to_channel_with_control(
.unwrap_or_else(|| (current_row.max(1), current_column.saturating_add(1).max(1)));
current_row = position.0;
current_column = position.1;
rows.push_cell(position.0, position.1, serde_json::Value::Null, false, progress)?;
if current_row >= min_import_row && !rows.selected_range_finished(current_row) {
rows.push_cell(position.0, position.1, serde_json::Value::Null, false, progress)?;
}
}
Ok(Event::Start(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => {
let position = xml_attr_value(&reader, &element, b"r")
@@ -4256,16 +4298,18 @@ fn stream_xlsx_rows_to_channel_with_control(
}
Ok(Event::End(element)) if xml_local_name_eq(element.name().as_ref(), b"c") => {
if let Some((row, column)) = current_position.take() {
let format_as_text = rows.is_text_source_column(row, column, &text_source_columns);
let value = xlsx_stream_cell_value(
&current_cell,
&mut shared_strings,
&styles,
date_1904,
format_as_text,
empty_string_as_null,
)?;
rows.push_cell(row, column, value, current_cell.has_content(), progress)?;
if row >= min_import_row && !rows.selected_range_finished(row) {
let format_as_text = rows.is_text_source_column(row, column, &text_source_columns);
let value = xlsx_stream_cell_value(
&current_cell,
&mut shared_strings,
&styles,
date_1904,
format_as_text,
empty_string_as_null,
)?;
rows.push_cell(row, column, value, current_cell.has_content(), progress)?;
}
current_cell = XlsxPreviewRawCell::default();
}
inline_phonetic_depth = 0;
@@ -9046,6 +9090,103 @@ mod tests {
assert!(state.selected_range_finished(11));
}
#[test]
fn xlsx_stream_custom_title_row_with_leading_row_cells() {
// Reproduces #11525: Row 1 has cells starting at column B (e.g. B1 banner),
// title_row is 2 (starting at column A: A2, B2), and data_start_row is 3 (A3, B3).
let path = std::env::temp_dir().join(format!("dbx-excel-leading-row-cells-{}.xlsx", uuid::Uuid::new_v4()));
std::fs::write(
&path,
build_styled_test_xlsx(
false,
&[
("B1", 0, 999.0),
("A2", 0, 10.0),
("B2", 0, 20.0),
("A3", 0, 1.0),
("B3", 0, 2.0),
("A4", 0, 3.0),
("B4", 0, 4.0),
],
),
)
.unwrap();
let options = TableImportParseOptions {
title_row: Some(2),
data_start_row: Some(3),
..TableImportParseOptions::default()
};
let (preview, _) = parse_xlsx_preview_file_with_options(&path.to_string_lossy(), &options, 50).unwrap();
assert_eq!(preview.columns, vec!["10", "20"]);
assert_eq!(
preview.rows,
vec![vec![serde_json::json!(1), serde_json::json!(2)], vec![serde_json::json!(3), serde_json::json!(4)],]
);
let (sender, mut receiver) = tokio::sync::mpsc::channel(16);
let stream_result = stream_xlsx_rows_to_channel(
&path.to_string_lossy(),
&options,
500,
Some(preview.columns.clone()),
HashSet::new(),
false,
sender,
);
assert!(stream_result.is_ok(), "stream_xlsx_rows_to_channel failed: {:?}", stream_result.err());
let mut streamed_header = None;
let mut streamed_rows = Vec::new();
while let Some(message) = receiver.blocking_recv() {
match message.unwrap() {
XlsxStreamMessage::Header(header) => streamed_header = Some(header),
XlsxStreamMessage::Rows(rows) => streamed_rows.extend(rows),
_ => {}
}
}
assert_eq!(streamed_header, Some(preview.columns));
assert_eq!(streamed_rows, preview.rows);
let _ = std::fs::remove_file(path);
}
#[test]
fn xlsx_stream_title_row_with_empty_leading_cell() {
// Title row 1 only has cell B1; data row 2 has cells A2 and B2.
let path = std::env::temp_dir().join(format!("dbx-excel-empty-leading-cell-{}.xlsx", uuid::Uuid::new_v4()));
std::fs::write(&path, build_styled_test_xlsx(false, &[("B1", 0, 20.0), ("A2", 0, 1.0), ("B2", 0, 2.0)]))
.unwrap();
let options = TableImportParseOptions {
title_row: Some(1),
data_start_row: Some(2),
..TableImportParseOptions::default()
};
let (preview, _) = parse_xlsx_preview_file_with_options(&path.to_string_lossy(), &options, 50).unwrap();
assert_eq!(preview.columns, vec!["column_1", "20"]);
let (sender, mut receiver) = tokio::sync::mpsc::channel(16);
let stream_result = stream_xlsx_rows_to_channel(
&path.to_string_lossy(),
&options,
500,
Some(preview.columns.clone()),
HashSet::new(),
false,
sender,
);
assert!(stream_result.is_ok());
let mut streamed_rows = Vec::new();
while let Some(message) = receiver.blocking_recv() {
if let XlsxStreamMessage::Rows(rows) = message.unwrap() {
streamed_rows.extend(rows);
}
}
assert_eq!(streamed_rows, vec![vec![serde_json::json!(1), serde_json::json!(2)]]);
let _ = std::fs::remove_file(path);
}
fn assert_xlsx_empty_string_option(options: TableImportParseOptions, expected_row: Vec<serde_json::Value>) {
let path =
std::env::temp_dir().join(format!("dbx-table-import-empty-string-option-{}.xlsx", uuid::Uuid::new_v4()));