fix: csv import honors MAX_RECORD instead of a fixed 1KB line cap (#67)

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.

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)

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

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