adding tests - squash before merge

compaction.go

@@ -1479,6 +1479,7 @@ func (d *DB) runIngestFlush(c *compaction) (*manifest.VersionEdit, error) {
 		}
 
 		if ingestFlushable.exciseSpan.Valid() {
+			ve.DeletedFiles = map[manifest.DeletedFileEntry]*manifest.FileMetadata{}
 			// Iterate through all levels and find files that intersect with exciseSpan.
 			for level = range c.version.Levels {
 				overlaps := c.version.Overlaps(level, ingestFlushable.exciseSpan.Start, ingestFlushable.exciseSpan.End, true /* exclusiveEnd */)

data_test.go

@@ -1250,6 +1250,7 @@ func (d *DB) waitTableStats() {
 
 func runIngestAndExciseCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 	var exciseSpan KeyRange
+	var doFlushableIngest bool
 	paths := make([]string, 0, len(td.CmdArgs))
 	for i, arg := range td.CmdArgs {
 		switch td.CmdArgs[i].Key {
@@ -1263,12 +1264,14 @@ func runIngestAndExciseCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 			}
 			exciseSpan.Start = []byte(fields[0])
 			exciseSpan.End = []byte(fields[1])
+		case "flushable-ingest":
+			doFlushableIngest = true
 		default:
 			paths = append(paths, arg.String())
 		}
 	}
 
-	if _, err := d.IngestAndExcise(paths, nil, nil, exciseSpan, false); err != nil {
+	if _, err := d.IngestAndExcise(paths, nil, nil, exciseSpan, doFlushableIngest); err != nil {
 		return err
 	}
 	return nil

ingest.go

@@ -1355,6 +1355,7 @@ func (d *DB) handleIngestAsFlushable(
 	d.mu.mem.queue = append(d.mu.mem.queue, entry)
 	d.rotateMemtable(newLogNum, nextSeqNum, currMem)
 	d.updateReadStateLocked(d.opts.DebugCheck)
+	// TODO(aaditya): is this necessary? we call this already in rotateMemtable above
 	d.maybeScheduleFlush()
 	return nil
 }
@@ -1560,11 +1561,19 @@ func (d *DB) ingest(
 			mut.writerRef()
 			return
 		}
-		// The ingestion overlaps with some entry in the flushable queue.
-		if d.FormatMajorVersion() < FormatFlushableIngest ||
-			d.opts.Experimental.DisableIngestAsFlushable() || !doFlushableIngest ||
-			len(shared) > 0 || exciseSpan.Valid() || len(external) > 0 ||
-			(len(d.mu.mem.queue) > d.opts.MemTableStopWritesThreshold-1) {
+
+		// The ingestion overlaps with some entry in the flushable queue. If the
+		// pre-conditions are met below, we can treat this ingestion as a flushable
+		// ingest, otherwise we wait on the memtable flush before ingestion.
+		//
+		// TODO(aaditya): We should make flushableIngest compatible with remote
+		// files.
+		hasRemoteFiles := len(shared) > 0 || len(external) > 0
+		canIngestFlushable := d.FormatMajorVersion() >= FormatFlushableIngest &&
+			(len(d.mu.mem.queue) < d.opts.MemTableStopWritesThreshold) &&
+			!d.opts.Experimental.DisableIngestAsFlushable() && !hasRemoteFiles
+
+		if !canIngestFlushable || (exciseSpan.Valid() && !doFlushableIngest) {
 			// We're not able to ingest as a flushable,
 			// so we must synchronously flush.
 			//

ingest_test.go

@@ -696,6 +696,9 @@ func TestExcise(t *testing.T) {
 
 		case "ingest-and-excise":
 			flushed = false
+			d.mu.Lock()
+			prevFlushableIngests := d.mu.versions.metrics.Flush.AsIngestCount
+			d.mu.Unlock()
 			if err := runIngestAndExciseCmd(td, d, mem); err != nil {
 				return err.Error()
 			}
@@ -704,8 +707,12 @@ func TestExcise(t *testing.T) {
 			for d.mu.compact.flushing {
 				d.mu.compact.cond.Wait()
 			}
+			flushableIngests := d.mu.versions.metrics.Flush.AsIngestCount
 			d.mu.Unlock()
 			if flushed {
+				if prevFlushableIngests < flushableIngests {
+					return "flushable ingest"
+				}
 				return "memtable flushed"
 			}
 			return ""

testdata/excise

@@ -540,3 +540,137 @@ c: (something, .)
 c: (something, .)
 .
 .
+
+
+
+
+
+
+####
+# FlushableIngest
+####
+
+reset
+----
+
+batch
+set a foo
+set b bar
+----
+
+batch
+set d@6 baz
+----
+
+flush
+----
+
+lsm
+----
+0.0:
+  000005:[a#10,SET-d@6#12,SET]
+
+compact a z
+----
+
+lsm
+----
+6:
+  000005:[a#10,SET-d@6#12,SET]
+
+batch
+set d@6 something
+set g something
+----
+
+flush
+----
+
+lsm
+----
+0.0:
+  000007:[d@6#13,SET-g#14,SET]
+6:
+  000005:[a#10,SET-d@6#12,SET]
+
+batch
+set x something
+----
+
+file-only-snapshot s1
+a z
+----
+ok
+
+lsm
+----
+0.0:
+  000007:[d@6#13,SET-g#14,SET]
+6:
+  000005:[a#10,SET-d@6#12,SET]
+
+build ext7
+del d@6
+----
+
+lsm
+----
+0.0:
+  000007:[d@6#13,SET-g#14,SET]
+6:
+  000005:[a#10,SET-d@6#12,SET]
+
+ingest ext7
+----
+
+lsm
+----
+0.1:
+  000008:[d@6#16,DEL-d@6#16,DEL]
+0.0:
+  000007:[d@6#13,SET-g#14,SET]
+6:
+  000005:[a#10,SET-d@6#12,SET]
+
+compact c e
+----
+
+lsm
+----
+6:
+  000009:[a#0,SET-g#0,SET]
+
+build ext5
+set c something
+set b something
+set f something
+del b-e
+----
+
+ingest-and-excise ext5 excise=b-e flushable-ingest
+----
+flushable ingest
+
+lsm
+----
+0.0:
+  000013:[x#15,SET-x#15,SET]
+6:
+  000014(000009):[a#0,SET-a#0,SET]
+  000010:[b#17,SET-f#17,SET]
+  000015(000009):[g#0,SET-g#0,SET]
+
+iter lower=c upper=e
+last
+prev
+prev
+seek-lt dd
+prev
+prev
+----
+c: (something, .)
+.
+.
+c: (something, .)
+.
+.