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

README.md

@@ -68,7 +68,9 @@ bulk insert DBNAME/TABLENAME from PATH with (CONFIG)
         VERBOSE: Enable verbose mode. (Default: 0).
         THREADS: Define the number of threads to be used during the importation process. Defaul value: half of system available.
         COLLATE: Perform collation after import, removing data larger than MAX_RECORD bytes. (Default: 0).
-        MAX_RECORD: Maximum record size in bytes (Default value: 2048).
+        MAX_RECORD: Maximum record size in bytes (Default value: 2048). Applied at import too: a CSV record whose
+                    data exceeds MAX_RECORD is skipped (counted and logged), instead of the old fixed 1 KB limit
+                    on the whole CSV line.
         MAX_RAM_PERCENT: limit the system RAM usage during collate process. Default value: 50.
         TMP_PATH: Path to the folder used for temporary files (default: /tmp).
         LOG_PATH: Path to a custom log file (default: /var/log/scanoss/ldb/DBNAME.log).

src/collate.c

@@ -542,7 +542,7 @@ bool key_in_delete_list(struct ldb_collate_data *collate, uint8_t *key, uint8_t
 							result = true;
 						else
 						{
-							unsigned char tuple_bin[MAX_CSV_LINE_LEN];
+							unsigned char tuple_bin[LDB_MAX_REC_LN];
 							if(!decode && !ldb_decoder_lib_load())
 								return false;
 

src/import.c

@@ -587,6 +587,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(MD5_LEN, 1);
 	uint8_t *field2 = calloc(MD5_LEN, 1);
 	uint8_t *item_buf = malloc(LDB_MAX_NODE_LN);
@@ -603,6 +611,11 @@ int ldb_import_csv(ldb_importation_config_t * job)
 	uint32_t skipped = 0;
 	uint32_t skipped_invalid = 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;
 
@@ -681,6 +694,7 @@ int ldb_import_csv(ldb_importation_config_t * job)
 				free(item_buf);
 				free(item_lastid);
 				free(field2);
+				free(data_bin);
 				if (item_sector) ldb_close_unlock(item_sector);
 				return LDB_ERROR_THREAD_ABORT;
 			}
@@ -694,6 +708,7 @@ int ldb_import_csv(ldb_importation_config_t * job)
 				free(item_buf);
 				free(item_lastid);
 				free(field2);
+				free(data_bin);
 				if (item_sector) ldb_close_unlock(item_sector);
 				return LDB_ERROR_CSV_TOO_MANY_SKIPPED;
 			}
@@ -730,10 +745,12 @@ int ldb_import_csv(ldb_importation_config_t * job)
 			skipped_invalid++;
 			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_invalid++;
 			continue;
 		}
@@ -803,13 +820,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)
 					{
@@ -833,6 +856,15 @@ int ldb_import_csv(ldb_importation_config_t * job)
 				continue;
 			}
 
+			/* 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);
+				skipped++;
+				continue;
+			}
 		}
 
 		if (data || (oss_bulk.keys > 1 && job->opt.params.csv_fields < 3))
@@ -881,6 +913,7 @@ int ldb_import_csv(ldb_importation_config_t * job)
 							free(item_buf);
 							free(item_lastid);
 							free(field2);
+							free(data_bin);
 							if (item_sector) ldb_close_unlock(item_sector);
 							return error;
 						}
@@ -981,6 +1014,7 @@ int ldb_import_csv(ldb_importation_config_t * job)
 			free(item_buf);
 			free(item_lastid);
 			free(field2);
+			free(data_bin);
 			if (item_sector) ldb_close_unlock(item_sector);
 			return error;
 		}
@@ -1005,8 +1039,9 @@ 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)
 	{
 		for (int i=0; i < 256; i++)

src/ldb.h

@@ -27,7 +27,7 @@
 #include "./ldb/types.h"
 #include "./ldb/mz.h"
 
-#define LDB_VERSION "4.1.12"
+#define LDB_VERSION "4.1.13"
 
 #define LDB_TABLE_DEFINITION_UNDEFINED -1
 #define LDB_TABLE_DEFINITION_STANDARD 0

test/test_kb

@@ -144,6 +144,49 @@ test_12_collate_zero_subkey_first_record() {
     rm -rf "$tmpdir" /var/lib/ldb/test_kbzero /usr/local/etc/scanoss/ldb/test_kbzero.conf /var/log/scanoss/ldb/test_kbzero.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_13_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,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).
+    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 () {
     ../ldb -u source/mined -n test_kb
 }