Append index for CSV

pkg/appendable/csv_handler.go

@@ -17,12 +17,18 @@ type CSVHandler struct {
 	io.ReadSeeker
 }
 
-func (c CSVHandler) Synchronize(f *IndexFile) error {
-
+func (c CSVHandler) Synchronize(f *IndexFile, headerData []string) error {
+	fmt.Println("===SYNCHRONIZE===")
 	var headers []string
 	var err error
 
-	isHeader := true
+	isHeader := false
+	if headerData == nil {
+		isHeader = true
+	} else {
+		headers = headerData
+	}
+
 	scanner := bufio.NewScanner(f.data)
 
 	for i := 0; scanner.Scan(); i++ {
@@ -40,7 +46,7 @@ func (c CSVHandler) Synchronize(f *IndexFile) error {
 
 		f.Checksums = append(f.Checksums, xxhash.Sum64(line))
 
-		if i == 0 {
+		if i == 0 && isHeader {
 			fmt.Printf("Header %d - StartOffset: %d, EndOffset: %d, Checksum: %d\n\n", i, start, start+uint64(len(line))+1, xxhash.Sum64(line))
 
 		} else {
@@ -58,10 +64,10 @@ func (c CSVHandler) Synchronize(f *IndexFile) error {
 		}
 
 		dec := csv.NewReader(bytes.NewReader(line))
-		f.handleCSVLine(dec, headers, []string{}, uint64(existingCount), start)
+		f.handleCSVLine(dec, headers, []string{}, uint64(existingCount)-1, start)
 	}
 
-	fmt.Printf("%v\n\n", f.Checksums)
+	fmt.Printf("%v\n\n===END===", f.Checksums)
 	return nil
 }
 
@@ -113,16 +119,20 @@ func inferCSVField(fieldValue string) (interface{}, protocol.FieldType) {
 }
 
 func (i *IndexFile) handleCSVLine(dec *csv.Reader, headers []string, path []string, dataIndex, dataOffset uint64) error {
+	fmt.Println("\t===reached handleCsv line===")
 
 	record, err := dec.Read()
 
 	if err != nil {
 		return fmt.Errorf("failed to read CSV record at index %d: %w", dataIndex, err)
 	}
 
+	fmt.Printf("\tRecord looks like %v", record)
+
 	cumulativeLength := uint64(0)
 
 	for fieldIndex, fieldValue := range record {
+		fmt.Printf("\n\theaders: %v\n", headers)
 		if fieldIndex >= len(headers) {
 			return fmt.Errorf("field index %d is out of bounds with header", fieldIndex)
 		}
@@ -135,7 +145,7 @@ func (i *IndexFile) handleCSVLine(dec *csv.Reader, headers []string, path []stri
 
 		value, fieldType := inferCSVField(fieldValue)
 
-		fmt.Printf("Field '%s' - Offset: %d, Length: %d, Value: %v, Type: %v\n", fieldName, fieldOffset, fieldLength, value, fieldType)
+		fmt.Printf("\tField '%s' - Offset: %d, Length: %d, Value: %v, Type: %v\n", fieldName, fieldOffset, fieldLength, value, fieldType)
 
 		switch fieldType {
 		case protocol.FieldTypeBoolean, protocol.FieldTypeString, protocol.FieldTypeNumber:
@@ -161,7 +171,7 @@ func (i *IndexFile) handleCSVLine(dec *csv.Reader, headers []string, path []stri
 		cumulativeLength += fieldLength
 	}
 
-	fmt.Printf("\n")
+	fmt.Printf("\t===end===\n")
 
 	return nil
 }

pkg/appendable/index_file_csv_test.go

@@ -8,8 +8,8 @@ import (
 
 func TestAppendDataRowCSV(t *testing.T) {
 
-	var mockCsv string = "test\ntest1\n"
-	var mockCsv2 string = "test\ntest1\ntest3\n"
+	var mockCsv string = "header1\ntest1\n"
+	var mockCsv2 string = "header1\ntest1\ntest3\n"
 
 	t.Run("no schema changes", func(t *testing.T) {
 
@@ -49,63 +49,61 @@ func TestAppendDataRowCSV(t *testing.T) {
 		}
 	})
 
-	/*
-		t.Run("correctly sets field offset", func(t *testing.T) {
-			i, err := NewIndexFile(CSVHandler{ReadSeeker: strings.NewReader(mockCsv)})
-			if err != nil {
-				t.Fatal(err)
-			}
+	t.Run("append index to existing", func(t *testing.T) {
+		i, err := NewIndexFile(CSVHandler{ReadSeeker: strings.NewReader(mockCsv)})
+		if err != nil {
+			t.Fatal(err)
+		}
 
-			buf := &bytes.Buffer{}
+		buf := &bytes.Buffer{}
 
-			if err := i.Serialize(buf); err != nil {
-				t.Fatal(err)
-			}
+		if err := i.Serialize(buf); err != nil {
+			t.Fatal(err)
+		}
 
-			j, err := ReadIndexFile(buf, CSVHandler{ReadSeeker: strings.NewReader(mockCsv2)})
-			if err != nil {
-				t.Fatal(err)
-			}
+		j, err := ReadIndexFile(buf, CSVHandler{ReadSeeker: strings.NewReader(mockCsv2)})
+		if err != nil {
+			t.Fatal(err)
+		}
 
-			// check that the index file now has the additional data ranges but same number of indices
-			if len(j.Indexes) != 1 {
-				t.Errorf("got len(j.Indexes) = %d, want 1", len(j.Indexes))
-			}
+		// check that the index file now has the additional data ranges but same number of indices
+		if len(j.Indexes) != 1 {
+			t.Errorf("got len(j.Indexes) = %d, want 1", len(j.Indexes))
+		}
 
-			t.Logf("len: %v", j.Indexes[0].IndexRecords)
+		if len(j.Indexes[0].IndexRecords) != 2 {
+			t.Errorf("got len(j.Indexes[0].IndexRecords) = %d, want 2", len(j.Indexes[0].IndexRecords))
+		}
 
-			if len(j.Indexes[0].IndexRecords) != 2 {
-				t.Errorf("got len(j.Indexes[0].IndexRecords) = %d, want 2", len(j.Indexes[0].IndexRecords))
+		if len(j.Indexes[0].IndexRecords["test1"]) != 1 {
+			t.Errorf("got len(j.Indexes[0].IndexRecords[\"test1\"]) = %d, want 1", len(j.Indexes[0].IndexRecords["test1"]))
+		}
+		if len(j.Indexes[0].IndexRecords["test3"]) != 1 {
+			for key, records := range j.Indexes[0].IndexRecords {
+				t.Errorf("\n\n\nKey: %v, Records: %+v", key, records)
 			}
+			t.Errorf("got len(j.Indexes[0].IndexRecords[\"test3\"]) = %d, want 1", len(j.Indexes[0].IndexRecords["test3"]))
+		}
 
-			if len(j.Indexes[0].IndexRecords["test1"]) != 1 {
-				t.Errorf("got len(j.Indexes[0].IndexRecords[\"test1\"]) = %d, want 1", len(j.Indexes[0].IndexRecords["test1"]))
-			}
-			if len(j.Indexes[0].IndexRecords["test3"]) != 1 {
-				for key, records := range j.Indexes[0].IndexRecords {
-					t.Errorf("\n\n\nKey: %v, Records: %+v", key, records)
-				}
-				t.Errorf("got len(j.Indexes[0].IndexRecords[\"test3\"]) = %d, want 1", len(j.Indexes[0].IndexRecords["test3"]))
-			}
+		if j.Indexes[0].IndexRecords["test1"][0].DataNumber != 0 {
+			t.Errorf("got i.Indexes[0].IndexRecords[\"test1\"][0].DataNumber = %d, want 0", j.Indexes[0].IndexRecords["test1"][0].DataNumber)
+		}
+		if j.Indexes[0].IndexRecords["test1"][0].FieldStartByteOffset != uint64(len("{\"test\":")) {
+			t.Errorf("got i.Indexes[0].IndexRecords[\"test1\"][0].FieldStartByteOffset = %d, want 10", j.Indexes[0].IndexRecords["test1"][0].FieldStartByteOffset)
+		}
 
-			if j.Indexes[0].IndexRecords["test1"][0].DataNumber != 0 {
-				t.Errorf("got i.Indexes[0].IndexRecords[\"test1\"][0].DataNumber = %d, want 0", j.Indexes[0].IndexRecords["test1"][0].DataNumber)
-			}
-			if j.Indexes[0].IndexRecords["test1"][0].FieldStartByteOffset != uint64(len("{\"test\":")) {
-				t.Errorf("got i.Indexes[0].IndexRecords[\"test1\"][0].FieldStartByteOffset = %d, want 10", j.Indexes[0].IndexRecords["test1"][0].FieldStartByteOffset)
-			}
+		if j.Indexes[0].IndexRecords["test3"][0].DataNumber != 0 {
+			t.Errorf("got i.Indexes[0].IndexRecords[\"test3\"][0].DataNumber = %d, want 1", j.Indexes[0].IndexRecords["test3"][0].DataNumber)
+		}
 
-			if j.Indexes[0].IndexRecords["test3"][0].DataNumber != 1 {
-				t.Errorf("got i.Indexes[0].IndexRecords[\"test3\"][1].DataNumber = %d, want 1", j.Indexes[0].IndexRecords["test3"][1].DataNumber)
-			}
-			if j.Indexes[0].IndexRecords["test3"][0].FieldStartByteOffset != uint64(len("test\ntest1\n")) {
-				t.Errorf("got i.Indexes[0].IndexRecords[\"test3\"][1].FieldStartByteOffset = %d, want 10", j.Indexes[0].IndexRecords["test3"][1].FieldStartByteOffset)
-			}
-		})
-	*/
+		// verify byte offset calculation
+		if j.Indexes[0].IndexRecords["test3"][0].FieldStartByteOffset != uint64(len("header\ntest1\n")+1) {
+			t.Errorf("got i.Indexes[0].IndexRecords[\"test3\"][0].FieldStartByteOffset = %d, want 14", j.Indexes[0].IndexRecords["test3"][0].FieldStartByteOffset)
+		}
+	})
 
 	t.Run("generate index file", func(t *testing.T) {
-		i, err := NewIndexFile(CSVHandler{ReadSeeker: strings.NewReader(mockCsv2)})
+		i, err := NewIndexFile(CSVHandler{ReadSeeker: strings.NewReader("")})
 
 		if err != nil {
 			t.Fatal(err)

pkg/appendable/io.go

@@ -1,8 +1,10 @@
 package appendable
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/binary"
+	"encoding/csv"
 	"fmt"
 	"io"
 	"sort"
@@ -15,7 +17,7 @@ import (
 
 type DataHandler interface {
 	io.ReadSeeker
-	Synchronize(f *IndexFile) error
+	Synchronize(f *IndexFile, headerData []string) error
 }
 
 func NewIndexFile(data DataHandler) (*IndexFile, error) {
@@ -24,7 +26,7 @@ func NewIndexFile(data DataHandler) (*IndexFile, error) {
 		Indexes: []Index{},
 		data:    data,
 	}
-	return f, data.Synchronize(f)
+	return f, data.Synchronize(f, nil)
 }
 
 func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
@@ -33,8 +35,6 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 
 	f.data = data
 
-	fmt.Printf("data looks like %v\n", data)
-
 	// read the version
 	version, err := encoding.ReadByte(r)
 	if err != nil {
@@ -80,6 +80,8 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 			return nil, fmt.Errorf("expected to read %d bytes, read %d bytes", ifh.IndexLength, br)
 		}
 
+		fmt.Printf("headers look like: %v\n", f.Indexes)
+
 		// read the index records
 		for i, index := range f.Indexes {
 
@@ -111,6 +113,7 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 
 				switch value.(type) {
 				case nil, bool, int, int8, int16, int32, int64, float32, float64, string:
+					fmt.Printf("appending: %v", value)
 					index.IndexRecords[value] = append(index.IndexRecords[value], ir)
 				default:
 					return nil, fmt.Errorf("unsupported type: %T", value)
@@ -140,9 +143,10 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 			start = f.EndByteOffsets[0]
 			startIndex = 1
 		}
+		fmt.Printf("\ndatacount: %v\n", ifh.DataCount)
 		for i := startIndex; i < int(ifh.DataCount); i++ {
 
-			fmt.Printf("Current start idx: %d, position: %d\n", i, start)
+			fmt.Printf("\n\nCurrent start idx: %d, position: %d\n", i, start)
 
 			if _, isCsv := data.(CSVHandler); isCsv {
 				if i > 1 {
@@ -159,7 +163,7 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 			if _, err := io.CopyN(buf, data, int64(f.EndByteOffsets[i]-start-1)); err != nil {
 				return nil, fmt.Errorf("failed to read data file: %w", err)
 			}
-			fmt.Printf("string: %v and \n bytes look like: %v\n", buf.String(), buf.Bytes())
+			fmt.Printf("string: %v and bytes look like: %v\n", buf.String(), buf.Bytes())
 
 			if xxhash.Sum64(buf.Bytes()) != f.Checksums[i] {
 				return nil, fmt.Errorf("checksum mismatch a %d, b %d", xxhash.Sum64(buf.Bytes()), f.Checksums[i])
@@ -169,7 +173,25 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 	default:
 		return nil, fmt.Errorf("unsupported version: %d", version)
 	}
+	var headers []string
+	if _, isCSV := data.(CSVHandler); isCSV {
+
+		if _, err := data.Seek(0, io.SeekStart); err != nil {
+			return nil, fmt.Errorf("failed to seek data file: %w", err)
+		}
+
+		buf := bufio.NewReader(data)
+		headerLine, err := buf.ReadString('\n')
+		if err != nil {
+			return nil, fmt.Errorf("failed to read header line: %w", err)
+		}
 
+		dec := csv.NewReader(strings.NewReader(headerLine))
+		headers, err = dec.Read()
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse CSV header: %w", err)
+		}
+	}
 	fmt.Printf("endbyteoffsets: %v\n", f.EndByteOffsets)
 
 	// we've deserialized the underlying file, seek to the end of the last data range to prepare for appending
@@ -184,8 +206,10 @@ func ReadIndexFile(r io.Reader, data DataHandler) (*IndexFile, error) {
 		}
 	}
 
-	fmt.Println("End of ReadIndexFile, calling Synchronize")
-	return f, data.Synchronize(f)
+	// extract headers from 0 -> endByteOffsets[0]
+
+	fmt.Printf("\n===End of ReadIndexFile, calling Synchronize===\n")
+	return f, data.Synchronize(f, headers)
 }
 
 func (f *IndexFile) Serialize(w io.Writer) error {

pkg/appendable/jsonl_handler.go

@@ -16,7 +16,7 @@ type JSONLHandler struct {
 	io.ReadSeeker
 }
 
-func (j JSONLHandler) Synchronize(f *IndexFile) error {
+func (j JSONLHandler) Synchronize(f *IndexFile, headerData []string) error {
 
 	// read until the next newline
 	scanner := bufio.NewScanner(f.data)

pkg/protocol/protocol.go

@@ -136,8 +136,6 @@ func (i IndexRecord) CSVField(r io.ReadSeeker) (any, error) {
 		return nil, fmt.Errorf("failed to decode field: %w", err)
 	}
 
-	fmt.Printf("Fields read at offset %d: %v\n", i.FieldStartByteOffset, fields)
-
 	if _, err := r.Seek(offset, io.SeekStart); err != nil {
 		return nil, fmt.Errorf("failed to seek to original offset: %w", err)
 	}