diff --git a/README.md b/README.md index 5d52bc9..9c34ad1 100644 --- a/README.md +++ b/README.md @@ -140,10 +140,12 @@ bulk insert DBNAME/TABLENAME from PATH with (CONFIG) THREADS: Number of threads used during the importation process. (Default: half of the available cores) COLLATE: Perform collation after import, removing data larger than MAX_RECORD bytes. (Default: 0) MAX_RECORD: Maximum record size in bytes. (Default: 1024) - Collate reserves one fixed slot of MAX_RECORD bytes per record, so this value drives the collate - memory footprint. Lower it to match the table's real record size on tables with many records - under a single key. If a key still exceeds the sector-size cap, its remaining records are - DISCARDED and the import reports E078. + Applied at two stages. At import, a CSV record whose data exceeds MAX_RECORD is skipped + (counted in the import summary and logged), instead of the old fixed 1 KB limit on the whole + CSV line. At collate, it reserves one fixed slot of MAX_RECORD bytes per record, so this + value also drives the collate memory footprint. Lower it to match the table's real record + size on tables with many records under a single key. If a key still exceeds the sector-size + cap, its remaining records are DISCARDED and the import reports E078. MAX_RAM_PERCENT: max % of TOTAL system RAM the collate may use to hold input sectors in memory, aggregated across all threads. Threads share a single budget: a sector is loaded in RAM only while the running total fits; otherwise it is collated in (slower) disk mode. diff --git a/src/import.c b/src/import.c index ac96702..f07307a 100644 --- a/src/import.c +++ b/src/import.c @@ -46,7 +46,6 @@ #include "ldb_error.h" #define DECODE_BASE64 8 -#define MAX_CSV_LINE_LEN 1024 #define REC_SIZE_LEN 2 pthread_mutex_t lock; @@ -570,6 +569,14 @@ int ldb_import_csv(ldb_importation_config_t * job) /* Node size is a 16-bit int */ int node_limit = 65536; + /* Maximum data record size accepted on import. Follows MAX_RECORD, keeping + import consistent with collate (which drops records with size > max_rec_ln) + instead of the old fixed 1 KB cap on the whole CSV line. Hard-capped by the + 2-byte record-size field. */ + int max_record = job->opt.params.collate_max_rec; + if (max_record <= 0 || max_record > LDB_MAX_REC_LN) + max_record = LDB_MAX_REC_LN; + uint8_t *itemid = calloc(hash_len, 1); uint8_t *field2 = NULL; /* holds all (keys-1) subkeys in binary; allocated once field2_ln is known */ uint8_t *item_buf = malloc(LDB_MAX_NODE_LN); @@ -585,6 +592,11 @@ int ldb_import_csv(ldb_importation_config_t * job) uint32_t imported = 0; uint32_t skipped = 0; + /* Decode buffer for binary (base64) records. Grows with the line buffer so a + record is never truncated to a fixed size; replaces the old 1 KB array. */ + unsigned char *data_bin = NULL; + size_t data_bin_size = 0; + uint64_t totalbytes = ldb_file_size(job->csv_path); size_t bytecounter = 0; @@ -656,10 +668,12 @@ int ldb_import_csv(ldb_importation_config_t * job) skipped++; continue; } - /* Skip records with sizes out of range */ - if (lineln > MAX_CSV_LINE_LEN || lineln < min_line_size) + /* Skip malformed lines too short to hold a key and the field separators. + The upper bound is no longer a fixed line length: it is enforced on the + record data size against MAX_RECORD, below (consistent with collate). */ + if (lineln < min_line_size) { - log_debug("%s: Line %d -- Skipped, %ld exceed MAX line size %d.\n", job->csv_path, line_number, lineln, MAX_CSV_LINE_LEN); + log_debug("%s: Line %d -- Skipped, %ld below min line size %d.\n", job->csv_path, line_number, lineln, min_line_size); skipped++; continue; } @@ -716,13 +730,19 @@ int ldb_import_csv(ldb_importation_config_t * job) /* Calculate record size */ int r_size = 0; - unsigned char data_bin[MAX_CSV_LINE_LEN]; if (data) { if (bin_mode) - { + { if (decode) { + /* Grow the decode buffer to hold the whole line if needed; the + base64 output is never larger than its input. */ + if (len > data_bin_size) + { + data_bin_size = len; + data_bin = realloc(data_bin, data_bin_size); + } r_size = decode(DECODE_BASE64,NULL, NULL, data, strlen(data), data_bin); if (r_size <= 0) { @@ -745,6 +765,15 @@ int ldb_import_csv(ldb_importation_config_t * job) skip = true; } + /* Enforce MAX_RECORD on the data size, consistent with collate's + max_rec_ln filter. Logged at info level (and counted) so an + oversized record is no longer dropped silently. */ + if (r_size > max_record) + { + log_info("%s: Line %d -- Skipped, record size %d exceeds MAX_RECORD %d.\n", job->csv_path, line_number, r_size, max_record); + skip = true; + } + if (skip) { skipped++; @@ -907,6 +936,7 @@ int ldb_import_csv(ldb_importation_config_t * job) free(item_buf); free(item_lastid); free(field2); + free(data_bin); if (job->opt.params.overwrite) diff --git a/src/ldb.h b/src/ldb.h index 7c97182..a3c73d9 100644 --- a/src/ldb.h +++ b/src/ldb.h @@ -27,7 +27,7 @@ #include "./ldb/types.h" #include "./ldb/mz.h" -#define LDB_VERSION "5.0.2-crc64" +#define LDB_VERSION "5.0.3-crc64" #define LDB_TABLE_DEFINITION_UNDEFINED -1 #define LDB_TABLE_DEFINITION_STANDARD 0 diff --git a/test/test_kb b/test/test_kb index 70dc461..9117bad 100644 --- a/test/test_kb +++ b/test/test_kb @@ -101,6 +101,51 @@ test_11_collate_keeps_records_of_adjacent_keys_with_same_data() { rm -rf "$tmpdir" /var/lib/ldb/test_kbdup /usr/local/etc/scanoss/ldb/test_kbdup.conf /var/log/scanoss/ldb/test_kbdup.log } +#Regression (#67): csv bulk import silently dropped every line longer than 1,023 +#bytes (a fixed MAX_CSV_LINE_LEN cap on the whole getline() line), ignoring +#MAX_RECORD. The limit must instead follow MAX_RECORD and apply to the record +#data size, consistent with collate, and an oversized record must be counted and +#logged rather than dropped in silence. +test_12_import_respects_max_record_not_fixed_line_len() { + rm -rf /var/lib/ldb/test_kbmax /usr/local/etc/scanoss/ldb/test_kbmax.conf /var/log/scanoss/ldb/test_kbmax.log + tmpdir=$(mktemp -d) + # key(32 hex) + "," + data -> record data size = line length - 33. + # With MAX_RECORD=2048: the first three must import (data <= 2048), the last + # one must be skipped (data 2967 > 2048). The 1100/2000-byte records are the + # ones the old fixed 1 KB cap dropped. + awk 'BEGIN { + split("1000 1100 2000 3000", N, " ") + for (i = 1; i <= 4; i++) { + key = sprintf("%032x", i) + d = ""; for (j = 0; j < N[i] - 33; j++) d = d "a" + printf "%s,%s\n", key, d + } + }' > "$tmpdir/00.csv" + + echo "bulk insert test_kbmax/t from $tmpdir/00.csv with (KEYS=1,KEY_SIZE=16,FIELDS=2,MAX_RECORD=2048,SORT=0,VALIDATE_VERSION=0,OVERWRITE=1)" | ../ldb -q + + imported=$(grep -oE "[0-9]+ records imported" /var/log/scanoss/ldb/test_kbmax.log | tail -n1 | cut -d' ' -f1) + assert_equals "3" "$imported" "import dropped records that fit within MAX_RECORD" + + # Records whose data fits MAX_RECORD must be present, including the >1 KB ones. + for i in 1 2 3; do + k=$(printf '%032x' $i) + result=$(echo "select from test_kbmax/t key $k csv hex 16" | ../ldb -q | wc -l) + assert_equals "1" "$result" "import dropped a record within MAX_RECORD (key $k)" + done + + # The record larger than MAX_RECORD must be skipped, not stored. + over=$(echo "select from test_kbmax/t key $(printf '%032x' 4) csv hex 16" | ../ldb -q | wc -l) + assert_equals "0" "$over" "import stored a record larger than MAX_RECORD" + + # The skip must be visible: counted in the summary and logged (not silent). + skipped=$(grep -oE "[0-9]+ skipped" /var/log/scanoss/ldb/test_kbmax.log | tail -n1 | cut -d' ' -f1) + assert_equals "1" "$skipped" "oversized record was not counted as skipped" + assert "grep -q 'exceeds MAX_RECORD' /var/log/scanoss/ldb/test_kbmax.log" "oversized record skip was not logged" + + rm -rf "$tmpdir" /var/lib/ldb/test_kbmax /usr/local/etc/scanoss/ldb/test_kbmax.conf /var/log/scanoss/ldb/test_kbmax.log +} + setup_suite () { # This suite covers the MD5 (16-byte key) mode. The default generated # configuration targets CRC64 (KEY_SIZE=8), so pin KEY_SIZE=16 in the db