lightning: refine write and ingest logic more like producer-consumer

br/pkg/lightning/backend/local/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "local_unix_generic.go",
         "local_windows.go",
         "localhelper.go",
+        "region_job.go",
     ],
     importpath = "github.com/pingcap/tidb/br/pkg/lightning/backend/local",
     visibility = ["//visibility:public"],
@@ -28,7 +29,6 @@ go_library(
         "//br/pkg/lightning/manual",
         "//br/pkg/lightning/metric",
         "//br/pkg/lightning/tikv",
-        "//br/pkg/lightning/worker",
         "//br/pkg/logutil",
         "//br/pkg/membuf",
         "//br/pkg/pdutil",
@@ -40,6 +40,7 @@ go_library(
         "//kv",
         "//parser/model",
         "//parser/mysql",
+        "//store/pdtypes",
         "//table",
         "//tablecodec",
         "//types",
@@ -93,19 +94,18 @@ go_test(
         "local_check_test.go",
         "local_test.go",
         "localhelper_test.go",
+        "region_job_test.go",
     ],
     embed = [":local"],
     flaky = True,
     shard_count = 45,
     deps = [
-        "//br/pkg/errors",
         "//br/pkg/lightning/backend",
         "//br/pkg/lightning/backend/kv",
         "//br/pkg/lightning/common",
         "//br/pkg/lightning/glue",
         "//br/pkg/lightning/log",
         "//br/pkg/lightning/mydump",
-        "//br/pkg/lightning/worker",
         "//br/pkg/membuf",
         "//br/pkg/mock",
         "//br/pkg/pdutil",

br/pkg/lightning/backend/local/engine.go

@@ -988,7 +988,14 @@ func (e *Engine) newKVIter(ctx context.Context, opts *pebble.IterOptions) Iter {
 	return newDupDetectIter(ctx, e.db, e.keyAdapter, opts, e.duplicateDB, logger, e.dupDetectOpt)
 }
 
+// getFirstAndLastKey reads the first and last key in range [lowerBound, upperBound)
+// in the engine. Empty upperBound means unbounded.
 func (e *Engine) getFirstAndLastKey(lowerBound, upperBound []byte) ([]byte, []byte, error) {
+	if len(upperBound) == 0 {
+		// we use empty slice for unbounded upper bound, but it means max value in pebble
+		// so reset to nil
+		upperBound = nil
+	}
 	opt := &pebble.IterOptions{
 		LowerBound: lowerBound,
 		UpperBound: upperBound,

br/pkg/lightning/backend/local/engine_test.go

@@ -122,4 +122,9 @@ func TestGetFirstAndLastKey(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, first)
 	require.Nil(t, last)
+
+	first, last, err = f.getFirstAndLastKey([]byte("e"), []byte(""))
+	require.NoError(t, err)
+	require.Equal(t, []byte("e"), first)
+	require.Equal(t, []byte("e"), last)
 }