mirror of
https://github.com/dolthub/dolt.git
synced 2026-04-21 19:39:04 -05:00
Merge branch 'main' into fulghum/issue-3776
This commit is contained in:
Jason Fulghum
Generated
+1248
-624
File diff suppressed because it is too large
Load Diff
@@ -35,7 +35,6 @@ import (
|
||||
"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
|
||||
"github.com/dolthub/dolt/go/libraries/doltcore/sqle/mysql_file_handler"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/config"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
)
|
||||
|
||||
// SqlEngine packages up the context necessary to run sql queries against dsqle.
|
||||
@@ -276,9 +275,7 @@ func dsqleDBsAsSqlDBs(dbs []dsqle.SqlDatabase) []sql.Database {
|
||||
|
||||
func newSqlContext(sess *dsess.DoltSession, initialDb string) func(ctx context.Context) (*sql.Context, error) {
|
||||
return func(ctx context.Context) (*sql.Context, error) {
|
||||
sqlCtx := sql.NewContext(ctx,
|
||||
sql.WithSession(sess),
|
||||
sql.WithTracer(tracing.Tracer(ctx)))
|
||||
sqlCtx := sql.NewContext(ctx, sql.WithSession(sess))
|
||||
|
||||
// If the session was already updated with a database then continue using it in the new session. Otherwise
|
||||
// use the initial one.
|
||||
|
||||
@@ -40,7 +40,6 @@ import (
|
||||
"github.com/dolthub/dolt/go/libraries/doltcore/table/editor"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/argparser"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/config"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
"github.com/dolthub/dolt/go/store/hash"
|
||||
)
|
||||
|
||||
@@ -229,9 +228,7 @@ func processFilterQuery(ctx context.Context, dEnv *env.DoltEnv, cm *doltdb.Commi
|
||||
func rebaseSqlEngine(ctx context.Context, dEnv *env.DoltEnv, cm *doltdb.Commit) (*sql.Context, *engine.SqlEngine, error) {
|
||||
sess := dsess.DefaultSession().NewDoltSession(config.NewMapConfig(make(map[string]string)))
|
||||
|
||||
sqlCtx := sql.NewContext(ctx,
|
||||
sql.WithSession(sess),
|
||||
sql.WithTracer(tracing.Tracer(ctx)))
|
||||
sqlCtx := sql.NewContext(ctx, sql.WithSession(sess))
|
||||
err := sqlCtx.SetSessionVariable(sqlCtx, sql.AutoCommitSessionVar, false)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
|
||||
+20
-10
@@ -28,10 +28,12 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/fatih/color"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
"github.com/pkg/profile"
|
||||
"github.com/uber/jaeger-client-go"
|
||||
"github.com/uber/jaeger-client-go/transport"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/exporters/jaeger"
|
||||
"go.opentelemetry.io/otel/sdk/resource"
|
||||
tracesdk "go.opentelemetry.io/otel/sdk/trace"
|
||||
semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
|
||||
|
||||
"github.com/dolthub/dolt/go/cmd/dolt/cli"
|
||||
"github.com/dolthub/dolt/go/cmd/dolt/commands"
|
||||
@@ -203,13 +205,21 @@ func runMain() int {
|
||||
// and browse to http://localhost:16686
|
||||
case jaegerFlag:
|
||||
cli.Println("running with jaeger tracing reporting to localhost")
|
||||
transport := transport.NewHTTPTransport("http://localhost:14268/api/traces?format=jaeger.thrift", transport.HTTPBatchSize(128000))
|
||||
reporter := jaeger.NewRemoteReporter(transport)
|
||||
tracer, closer := jaeger.NewTracer("dolt", jaeger.NewConstSampler(true), reporter)
|
||||
opentracing.SetGlobalTracer(tracer)
|
||||
defer closer.Close()
|
||||
args = args[1:]
|
||||
|
||||
exp, err := jaeger.New(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint("http://localhost:14268/api/traces")))
|
||||
if err != nil {
|
||||
cli.Println(color.YellowString("could not create jaeger collector: %v", err))
|
||||
} else {
|
||||
tp := tracesdk.NewTracerProvider(
|
||||
tracesdk.WithBatcher(exp),
|
||||
tracesdk.WithResource(resource.NewWithAttributes(
|
||||
semconv.SchemaURL,
|
||||
semconv.ServiceNameKey.String("dolt"),
|
||||
)),
|
||||
)
|
||||
otel.SetTracerProvider(tp)
|
||||
defer tp.Shutdown(context.Background())
|
||||
args = args[1:]
|
||||
}
|
||||
// Currently goland doesn't support running with a different working directory when using go modules.
|
||||
// This is a hack that allows a different working directory to be set after the application starts using
|
||||
// chdir=<DIR>. The syntax is not flexible and must match exactly this.
|
||||
|
||||
@@ -20,7 +20,7 @@ require (
|
||||
github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81
|
||||
github.com/dolthub/vitess v0.0.0-20220628181817-dae896d89d02
|
||||
github.com/dustin/go-humanize v1.0.0
|
||||
github.com/fatih/color v1.9.0
|
||||
github.com/fatih/color v1.13.0
|
||||
github.com/flynn-archive/go-shlex v0.0.0-20150515145356-3f9db97f8568
|
||||
github.com/go-sql-driver/mysql v1.6.0
|
||||
github.com/gocraft/dbr/v2 v2.7.2
|
||||
@@ -30,10 +30,9 @@ require (
|
||||
github.com/google/uuid v1.2.0
|
||||
github.com/jpillora/backoff v1.0.0
|
||||
github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d
|
||||
github.com/mattn/go-isatty v0.0.12
|
||||
github.com/mattn/go-isatty v0.0.14
|
||||
github.com/mattn/go-runewidth v0.0.9
|
||||
github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b
|
||||
github.com/opentracing/opentracing-go v1.2.0
|
||||
github.com/pkg/errors v0.9.1
|
||||
github.com/pkg/profile v1.5.0
|
||||
github.com/rivo/uniseg v0.1.0
|
||||
@@ -43,11 +42,9 @@ require (
|
||||
github.com/sirupsen/logrus v1.8.1
|
||||
github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966
|
||||
github.com/spf13/cobra v1.0.0
|
||||
github.com/stretchr/testify v1.7.0
|
||||
github.com/stretchr/testify v1.7.1
|
||||
github.com/tealeg/xlsx v1.0.5
|
||||
github.com/tklauser/go-sysconf v0.3.9 // indirect
|
||||
github.com/uber/jaeger-client-go v2.25.0+incompatible
|
||||
github.com/uber/jaeger-lib v2.4.0+incompatible // indirect
|
||||
go.uber.org/zap v1.15.0
|
||||
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
|
||||
golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f
|
||||
@@ -62,7 +59,7 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220707222451-29cab0137de9
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220708213239-a9724caf9408
|
||||
github.com/google/flatbuffers v2.0.6+incompatible
|
||||
github.com/gosuri/uilive v0.0.4
|
||||
github.com/kch42/buzhash v0.0.0-20160816060738-9bdec3dec7c6
|
||||
@@ -72,6 +69,10 @@ require (
|
||||
github.com/xitongsys/parquet-go-source v0.0.0-20211010230925-397910c5e371
|
||||
github.com/zeebo/blake3 v0.2.3
|
||||
github.com/zeebo/xxh3 v1.0.2
|
||||
go.opentelemetry.io/otel v1.7.0
|
||||
go.opentelemetry.io/otel/exporters/jaeger v1.7.0
|
||||
go.opentelemetry.io/otel/sdk v1.7.0
|
||||
go.opentelemetry.io/otel/trace v1.7.0
|
||||
golang.org/x/text v0.3.7
|
||||
gonum.org/v1/plot v0.11.0
|
||||
)
|
||||
@@ -90,6 +91,8 @@ require (
|
||||
github.com/go-fonts/liberation v0.2.0 // indirect
|
||||
github.com/go-kit/kit v0.10.0 // indirect
|
||||
github.com/go-latex/latex v0.0.0-20210823091927-c0d11ff05a81 // indirect
|
||||
github.com/go-logr/logr v1.2.3 // indirect
|
||||
github.com/go-logr/stdr v1.2.2 // indirect
|
||||
github.com/go-ole/go-ole v1.2.6 // indirect
|
||||
github.com/go-pdf/fpdf v0.6.0 // indirect
|
||||
github.com/golang/freetype v0.0.0-20170609003504-e2365dfdc4a0 // indirect
|
||||
@@ -103,7 +106,7 @@ require (
|
||||
github.com/klauspost/cpuid/v2 v2.0.12 // indirect
|
||||
github.com/lestrrat-go/strftime v1.0.4 // indirect
|
||||
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
|
||||
github.com/mattn/go-colorable v0.1.7 // indirect
|
||||
github.com/mattn/go-colorable v0.1.9 // indirect
|
||||
github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
|
||||
github.com/mitchellh/hashstructure v1.1.0 // indirect
|
||||
github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852 // indirect
|
||||
|
||||
@@ -175,8 +175,8 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
|
||||
github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
|
||||
github.com/dolthub/fslock v0.0.3 h1:iLMpUIvJKMKm92+N1fmHVdxJP5NdyDK5bK7z7Ba2s2U=
|
||||
github.com/dolthub/fslock v0.0.3/go.mod h1:QWql+P17oAAMLnL4HGB5tiovtDuAjdDTPbuqx7bYfa0=
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220707222451-29cab0137de9 h1:xZQNMCQ/TokT18Xu4DolmKdXHbbj0TfZc/RqPRlIEQg=
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220707222451-29cab0137de9/go.mod h1:E7d38ZJM2YloaxJl+f7P3MYwfAA+NxAxBEKzNkzlUag=
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220708213239-a9724caf9408 h1:d9S6ejy+EF2fbAOXl7K9DZCm79mfZyZY/dgtoZu21DI=
|
||||
github.com/dolthub/go-mysql-server v0.12.1-0.20220708213239-a9724caf9408/go.mod h1:fhyVDvV0K59cdk9N7TQsPjr2Hp/Qseej8+R9tVqPDCg=
|
||||
github.com/dolthub/ishell v0.0.0-20220112232610-14e753f0f371 h1:oyPHJlzumKta1vnOQqUnfdz+pk3EmnHS3Nd0cCT0I2g=
|
||||
github.com/dolthub/ishell v0.0.0-20220112232610-14e753f0f371/go.mod h1:dhGBqcCEfK5kuFmeO5+WOx3hqc1k3M29c1oS/R7N4ms=
|
||||
github.com/dolthub/jsonpath v0.0.0-20210609232853-d49537a30474 h1:xTrR+l5l+1Lfq0NvhiEsctylXinUMFhhsqaEcl414p8=
|
||||
@@ -202,8 +202,9 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
|
||||
github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
|
||||
github.com/fatih/color v1.9.0 h1:8xPHl4/q1VyqGIPif1F+1V3Y3lSmrq01EabUW3CoW5s=
|
||||
github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU=
|
||||
github.com/fatih/color v1.13.0 h1:8LOYc1KYPPmyKMuN8QV2DNRWNbLo6LZ0iLs8+mlH53w=
|
||||
github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
|
||||
github.com/flynn-archive/go-shlex v0.0.0-20150515145356-3f9db97f8568 h1:BMXYYRWTLOJKlh+lOBt6nUQgXAfB7oVIQt5cNreqSLI=
|
||||
github.com/flynn-archive/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:rZfgFAXFS/z/lEd6LJmf9HVZ1LkgYiHx5pHhV5DR16M=
|
||||
github.com/fogleman/gg v1.3.0/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
|
||||
@@ -234,6 +235,11 @@ github.com/go-latex/latex v0.0.0-20210823091927-c0d11ff05a81/go.mod h1:SX0U8uGpx
|
||||
github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
|
||||
github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
|
||||
github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
|
||||
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
|
||||
github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
|
||||
github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
|
||||
github.com/go-ole/go-ole v1.2.1/go.mod h1:7FAglXiTm7HKlQRDeOQ6ZNUHidzCWXuZWq/1dTyBNF8=
|
||||
github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
|
||||
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
|
||||
@@ -485,8 +491,9 @@ github.com/matoous/godox v0.0.0-20190911065817-5d6d842e92eb/go.mod h1:1BELzlh859
|
||||
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
|
||||
github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
|
||||
github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
|
||||
github.com/mattn/go-colorable v0.1.7 h1:bQGKb3vps/j0E9GfJQ03JyhRuxsvdAanXlT9BTw3mdw=
|
||||
github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
|
||||
github.com/mattn/go-colorable v0.1.9 h1:sqDoxXbdeALODt0DAeJCVp38ps9ZogZEAXjus69YV3U=
|
||||
github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
|
||||
github.com/mattn/go-ieproxy v0.0.1/go.mod h1:pYabZ6IHcRpFh7vIaLfK7rdcWgFEb3SFJ6/gNWuh88E=
|
||||
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
github.com/mattn/go-isatty v0.0.4/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
|
||||
@@ -494,8 +501,9 @@ github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hd
|
||||
github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/mattn/go-isatty v0.0.11/go.mod h1:PhnuNfih5lzO57/f3n+odYbM4JtupLOxQOAqxQCu2WE=
|
||||
github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
|
||||
github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
|
||||
github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y=
|
||||
github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
|
||||
github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
|
||||
github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
|
||||
github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
|
||||
@@ -557,7 +565,6 @@ github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492/go
|
||||
github.com/opentracing/basictracer-go v1.0.0/go.mod h1:QfBfYuafItcjQuMwinw9GhYKwFXS9KnPs5lxoYwgW74=
|
||||
github.com/opentracing/opentracing-go v1.0.2/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
|
||||
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
|
||||
github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+1B0VhjKrZUs=
|
||||
github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
|
||||
github.com/openzipkin-contrib/zipkin-go-opentracing v0.4.5/go.mod h1:/wsWhb9smxSfWAKL3wpBW7V8scJMt8N8gnaMCS9E/cA=
|
||||
github.com/openzipkin/zipkin-go v0.1.6/go.mod h1:QgAqvLzwWbR/WpD4A3cGpPtJrZXNIiJc5AZX7/PBEpw=
|
||||
@@ -702,8 +709,9 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
|
||||
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
|
||||
github.com/tdakkota/asciicheck v0.0.0-20200416190851-d7f85be797a2/go.mod h1:yHp0ai0Z9gUljN3o0xMhYJnH/IcvkdTBOX2fmJ93JEM=
|
||||
github.com/tealeg/xlsx v1.0.5 h1:+f8oFmvY8Gw1iUXzPk+kz+4GpbDZPK1FhPiQRd+ypgE=
|
||||
@@ -717,10 +725,6 @@ github.com/tklauser/numcpus v0.3.0/go.mod h1:yFGUr7TUHQRAhyqBcEg0Ge34zDBAsIvJJcy
|
||||
github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
|
||||
github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
|
||||
github.com/tommy-muehle/go-mnd v1.3.1-0.20200224220436-e6f9a994e8fa/go.mod h1:dSUh0FtTP8VhvkL1S+gUR1OKd9ZnSaozuI6r3m6wOig=
|
||||
github.com/uber/jaeger-client-go v2.25.0+incompatible h1:IxcNZ7WRY1Y3G4poYlx24szfsn/3LvK9QHCq9oQw8+U=
|
||||
github.com/uber/jaeger-client-go v2.25.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
|
||||
github.com/uber/jaeger-lib v2.4.0+incompatible h1:fY7QsGQWiCt8pajv4r7JEvmATdCVaWxXbjwyYwsNaLQ=
|
||||
github.com/uber/jaeger-lib v2.4.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
|
||||
github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
|
||||
github.com/ultraware/funlen v0.0.3/go.mod h1:Dp4UiAus7Wdb9KUZsYWZEWiRzGuM2kXM1lPbfaF6xhA=
|
||||
github.com/ultraware/whitespace v0.0.4/go.mod h1:aVMh/gQve5Maj9hQ/hg+F75lr/X5A89uZnzAmWSineA=
|
||||
@@ -765,6 +769,14 @@ go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
|
||||
go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
|
||||
go.opencensus.io v0.22.4 h1:LYy1Hy3MJdrCdMwwzxA/dRok4ejH+RwNGbuoD9fCjto=
|
||||
go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
|
||||
go.opentelemetry.io/otel v1.7.0 h1:Z2lA3Tdch0iDcrhJXDIlC94XE+bxok1F9B+4Lz/lGsM=
|
||||
go.opentelemetry.io/otel v1.7.0/go.mod h1:5BdUoMIz5WEs0vt0CUEMtSSaTSHBBVwrhnz7+nrD5xk=
|
||||
go.opentelemetry.io/otel/exporters/jaeger v1.7.0 h1:wXgjiRldljksZkZrldGVe6XrG9u3kYDyQmkZwmm5dI0=
|
||||
go.opentelemetry.io/otel/exporters/jaeger v1.7.0/go.mod h1:PwQAOqBgqbLQRKlj466DuD2qyMjbtcPpfPfj+AqbSBs=
|
||||
go.opentelemetry.io/otel/sdk v1.7.0 h1:4OmStpcKVOfvDOgCt7UriAPtKolwIhxpnSNI/yK+1B0=
|
||||
go.opentelemetry.io/otel/sdk v1.7.0/go.mod h1:uTEOTwaqIVuTGiJN7ii13Ibp75wJmYUDe374q6cZwUU=
|
||||
go.opentelemetry.io/otel/trace v1.7.0 h1:O37Iogk1lEkMRXewVtZ1BBTVn5JEp8GrJvP92bJqC6o=
|
||||
go.opentelemetry.io/otel/trace v1.7.0/go.mod h1:fzLSB9nqR2eXzxPXb2JW9IKE+ScyXA48yyE4TNvoHqU=
|
||||
go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
|
||||
go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
|
||||
@@ -962,7 +974,9 @@ golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7w
|
||||
golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210423185535-09eb48e85fd7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220111092808-5a964db01320 h1:0jf+tOCoZ3LyutmCOWpVni1chK4VfFLhRsDK7MhqGRY=
|
||||
golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
|
||||
@@ -83,15 +83,14 @@ func pushDataset(ctx context.Context, destDB, srcDB datas.Database, tempTableDir
|
||||
}
|
||||
|
||||
puller, err := pull.NewPuller(ctx, tempTableDir, defaultChunksPerTF, srcCS, destCS, waf, addr, nil)
|
||||
if err == pull.ErrDBUpToDate {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
if err != nil && err != pull.ErrDBUpToDate {
|
||||
return err
|
||||
}
|
||||
|
||||
err = puller.Pull(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
if err != pull.ErrDBUpToDate {
|
||||
err = puller.Pull(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
ds, err = destDB.GetDataset(ctx, rf.String())
|
||||
|
||||
@@ -83,3 +83,27 @@ func (db hooksDatabase) CommitWithWorkingSet(
|
||||
}
|
||||
return commitDS, workingSetDS, err
|
||||
}
|
||||
|
||||
func (db hooksDatabase) Commit(ctx context.Context, ds datas.Dataset, v types.Value, opts datas.CommitOptions) (datas.Dataset, error) {
|
||||
ds, err := db.Database.Commit(ctx, ds, v, opts)
|
||||
if err == nil {
|
||||
db.ExecuteCommitHooks(ctx, ds)
|
||||
}
|
||||
return ds, err
|
||||
}
|
||||
|
||||
func (db hooksDatabase) SetHead(ctx context.Context, ds datas.Dataset, newHeadAddr hash.Hash) (datas.Dataset, error) {
|
||||
ds, err := db.Database.SetHead(ctx, ds, newHeadAddr)
|
||||
if err == nil {
|
||||
db.ExecuteCommitHooks(ctx, ds)
|
||||
}
|
||||
return ds, err
|
||||
}
|
||||
|
||||
func (db hooksDatabase) FastForward(ctx context.Context, ds datas.Dataset, newHeadAddr hash.Hash) (datas.Dataset, error) {
|
||||
ds, err := db.Database.FastForward(ctx, ds, newHeadAddr)
|
||||
if err == nil {
|
||||
db.ExecuteCommitHooks(ctx, ds)
|
||||
}
|
||||
return ds, err
|
||||
}
|
||||
|
||||
@@ -31,12 +31,13 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cenkalti/backoff"
|
||||
"github.com/opentracing/opentracing-go"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
remotesapi "github.com/dolthub/dolt/go/gen/proto/dolt/services/remotesapi/v1alpha1"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/iohelp"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
"github.com/dolthub/dolt/go/store/atomicerr"
|
||||
"github.com/dolthub/dolt/go/store/chunks"
|
||||
"github.com/dolthub/dolt/go/store/hash"
|
||||
@@ -77,6 +78,8 @@ const (
|
||||
uploadRetryCount = 5
|
||||
)
|
||||
|
||||
var tracer = otel.Tracer("github.com/dolthub/dolt/go/libraries/doltcore/remotestorage")
|
||||
|
||||
var uploadRetryParams = backoff.NewExponentialBackOff()
|
||||
var downRetryParams = backoff.NewExponentialBackOff()
|
||||
|
||||
@@ -271,9 +274,7 @@ func (dcs *DoltChunkStore) GetMany(ctx context.Context, hashes hash.HashSet, fou
|
||||
atomic.AddUint64(&decompressedSize, uint64(len(c.Data())))
|
||||
found(ctx, &c)
|
||||
})
|
||||
if span := opentracing.SpanFromContext(ctx); span != nil {
|
||||
span.LogKV("decompressed_bytes", decompressedSize)
|
||||
}
|
||||
trace.SpanFromContext(ctx).SetAttributes(attribute.Int64("decompressed_bytes", int64(decompressedSize)))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -286,12 +287,12 @@ func (dcs *DoltChunkStore) GetMany(ctx context.Context, hashes hash.HashSet, fou
|
||||
// GetMany gets the Chunks with |hashes| from the store. On return, |foundChunks| will have been fully sent all chunks
|
||||
// which have been found. Any non-present chunks will silently be ignored.
|
||||
func (dcs *DoltChunkStore) GetManyCompressed(ctx context.Context, hashes hash.HashSet, found func(context.Context, nbs.CompressedChunk)) error {
|
||||
span, ctx := tracing.StartSpan(ctx, "remotestorage.GetManyCompressed")
|
||||
defer span.Finish()
|
||||
ctx, span := tracer.Start(ctx, "remotestorage.GetManyCompressed")
|
||||
defer span.End()
|
||||
|
||||
hashToChunk := dcs.cache.Get(hashes)
|
||||
|
||||
span.LogKV("num_hashes", len(hashes), "cache_hits", len(hashToChunk))
|
||||
span.SetAttributes(attribute.Int("num_hashes", len(hashes)), attribute.Int("cache_hits", len(hashToChunk)))
|
||||
atomic.AddUint32(&dcs.stats.Hits, uint32(len(hashToChunk)))
|
||||
|
||||
notCached := make([]hash.Hash, 0, len(hashes))
|
||||
@@ -517,9 +518,8 @@ func (l *dlLocations) Add(resp *remotesapi.DownloadLoc) {
|
||||
}
|
||||
|
||||
func (dcs *DoltChunkStore) getDLLocs(ctx context.Context, hashes []hash.Hash) (dlLocations, error) {
|
||||
span, ctx := tracing.StartSpan(ctx, "remotestorage.getDLLocs")
|
||||
span.LogKV("num_hashes", len(hashes))
|
||||
defer span.Finish()
|
||||
ctx, span := tracer.Start(ctx, "remotestorage.getDLLocs", trace.WithAttributes(attribute.Int("num_hashes", len(hashes))))
|
||||
defer span.End()
|
||||
|
||||
res := newDlLocations()
|
||||
|
||||
@@ -1030,7 +1030,7 @@ var defaultConcurrency ConcurrencyParams = ConcurrencyParams{
|
||||
LargeFetchSize: 2 * 1024 * 1024,
|
||||
}
|
||||
|
||||
func logDownloadStats(span opentracing.Span, originalGets map[string]*GetRange, computedGets []*GetRange) {
|
||||
func logDownloadStats(span trace.Span, originalGets map[string]*GetRange, computedGets []*GetRange) {
|
||||
chunkCount := 0
|
||||
originalBytes := uint64(0)
|
||||
for _, r := range originalGets {
|
||||
@@ -1041,14 +1041,20 @@ func logDownloadStats(span opentracing.Span, originalGets map[string]*GetRange,
|
||||
for _, r := range computedGets {
|
||||
downloadBytes += r.RangeLen()
|
||||
}
|
||||
span.LogKV("num_files", len(originalGets), "num_chunks", chunkCount, "num_batches", len(computedGets), "original_bytes", originalBytes, "download_bytes", downloadBytes)
|
||||
span.SetAttributes(
|
||||
attribute.Int("num_files", len(originalGets)),
|
||||
attribute.Int("num_chunks", chunkCount),
|
||||
attribute.Int("num_batches", len(computedGets)),
|
||||
attribute.Int64("original_bytes", int64(originalBytes)),
|
||||
attribute.Int64("download_bytes", int64(downloadBytes)),
|
||||
)
|
||||
}
|
||||
|
||||
// creates work functions for each download and executes them in parallel. The work functions write downloaded chunks
|
||||
// to chunkChan
|
||||
func (dcs *DoltChunkStore) downloadChunks(ctx context.Context, dlLocs dlLocations, chunkChan chan nbs.CompressedChunk) error {
|
||||
span, ctx := tracing.StartSpan(ctx, "remotestorage.downloadChunks")
|
||||
defer span.Finish()
|
||||
ctx, span := tracer.Start(ctx, "remotestorage.downloadChunks")
|
||||
defer span.End()
|
||||
|
||||
resourceGets := dlLocs.ranges
|
||||
|
||||
|
||||
@@ -23,16 +23,13 @@ import (
|
||||
|
||||
"github.com/dolthub/dolt/go/libraries/doltcore/env"
|
||||
"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
)
|
||||
|
||||
// These functions cannot be in the sqlfmt package as the reliance on the sqle package creates a circular reference.
|
||||
|
||||
func PrepareCreateTableStmt(ctx context.Context, sqlDb sql.Database) (*sql.Context, *sqle.Engine, *dsess.Session) {
|
||||
sess := dsess.DefaultSession()
|
||||
sqlCtx := sql.NewContext(ctx,
|
||||
sql.WithSession(sess),
|
||||
sql.WithTracer(tracing.Tracer(ctx)))
|
||||
sqlCtx := sql.NewContext(ctx, sql.WithSession(sess))
|
||||
|
||||
pro := NewDoltDatabaseProvider(env.DefaultInitBranch, nil, sqlDb)
|
||||
engine := sqle.NewDefault(pro)
|
||||
|
||||
@@ -1,46 +0,0 @@
|
||||
// Copyright 2020 Dolthub, Inc.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package tracing
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/opentracing/opentracing-go"
|
||||
)
|
||||
|
||||
// Called throughout dolt to get the tracer. Default implementation returns
|
||||
// opentracing.GlobalTracer(), but another implementation could be installed to
|
||||
// return a context-specific tracer.
|
||||
var Tracer func(ctx context.Context) opentracing.Tracer
|
||||
|
||||
func init() {
|
||||
Tracer = func(ctx context.Context) opentracing.Tracer {
|
||||
return opentracing.GlobalTracer()
|
||||
}
|
||||
}
|
||||
|
||||
// Start a new span, named `name`, as a child of the current span associated
|
||||
// with `ctx`. Starts a root span if there is no Span associated with `ctx`.
|
||||
// Returns the newly created Span and a new `ctx` associated with the Span.
|
||||
func StartSpan(ctx context.Context, name string) (opentracing.Span, context.Context) {
|
||||
parentSpan := opentracing.SpanFromContext(ctx)
|
||||
var opts []opentracing.StartSpanOption
|
||||
if parentSpan != nil {
|
||||
opts = append(opts, opentracing.ChildOf(parentSpan.Context()))
|
||||
}
|
||||
span := Tracer(ctx).StartSpan(name, opts...)
|
||||
ctx = opentracing.ContextWithSpan(ctx, span)
|
||||
return span, ctx
|
||||
}
|
||||
+11
-15
@@ -35,9 +35,11 @@ import (
|
||||
"cloud.google.com/go/storage"
|
||||
"github.com/dustin/go-humanize"
|
||||
"github.com/pkg/errors"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
"github.com/dolthub/dolt/go/store/blobstore"
|
||||
"github.com/dolthub/dolt/go/store/chunks"
|
||||
"github.com/dolthub/dolt/go/store/hash"
|
||||
@@ -72,6 +74,8 @@ var (
|
||||
globalFDCache *fdCache
|
||||
)
|
||||
|
||||
var tracer = otel.Tracer("github.com/dolthub/dolt/go/store/nbs")
|
||||
|
||||
func makeGlobalCaches() {
|
||||
globalFDCache = newFDCache(defaultMaxTables)
|
||||
|
||||
@@ -624,10 +628,8 @@ func (nbs *NomsBlockStore) addChunk(ctx context.Context, h addr, data []byte) bo
|
||||
}
|
||||
|
||||
func (nbs *NomsBlockStore) Get(ctx context.Context, h hash.Hash) (chunks.Chunk, error) {
|
||||
span, ctx := tracing.StartSpan(ctx, "nbs.Get")
|
||||
defer func() {
|
||||
span.Finish()
|
||||
}()
|
||||
ctx, span := tracer.Start(ctx, "nbs.Get")
|
||||
defer span.End()
|
||||
|
||||
t1 := time.Now()
|
||||
defer func() {
|
||||
@@ -673,22 +675,16 @@ func (nbs *NomsBlockStore) Get(ctx context.Context, h hash.Hash) (chunks.Chunk,
|
||||
}
|
||||
|
||||
func (nbs *NomsBlockStore) GetMany(ctx context.Context, hashes hash.HashSet, found func(context.Context, *chunks.Chunk)) error {
|
||||
span, ctx := tracing.StartSpan(ctx, "nbs.GetMany")
|
||||
span.LogKV("num_hashes", len(hashes))
|
||||
defer func() {
|
||||
span.Finish()
|
||||
}()
|
||||
ctx, span := tracer.Start(ctx, "nbs.GetMany", trace.WithAttributes(attribute.Int("num_hashes", len(hashes))))
|
||||
span.End()
|
||||
return nbs.getManyWithFunc(ctx, hashes, func(ctx context.Context, cr chunkReader, eg *errgroup.Group, reqs []getRecord, stats *Stats) (bool, error) {
|
||||
return cr.getMany(ctx, eg, reqs, found, nbs.stats)
|
||||
})
|
||||
}
|
||||
|
||||
func (nbs *NomsBlockStore) GetManyCompressed(ctx context.Context, hashes hash.HashSet, found func(context.Context, CompressedChunk)) error {
|
||||
span, ctx := tracing.StartSpan(ctx, "nbs.GetManyCompressed")
|
||||
span.LogKV("num_hashes", len(hashes))
|
||||
defer func() {
|
||||
span.Finish()
|
||||
}()
|
||||
ctx, span := tracer.Start(ctx, "nbs.GetManyCompressed", trace.WithAttributes(attribute.Int("num_hashes", len(hashes))))
|
||||
defer span.End()
|
||||
return nbs.getManyWithFunc(ctx, hashes, func(ctx context.Context, cr chunkReader, eg *errgroup.Group, reqs []getRecord, stats *Stats) (bool, error) {
|
||||
return cr.getManyCompressed(ctx, eg, reqs, found, nbs.stats)
|
||||
})
|
||||
|
||||
@@ -16,6 +16,9 @@ package prolly
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/dolthub/dolt/go/store/prolly/message"
|
||||
"github.com/dolthub/dolt/go/store/prolly/tree"
|
||||
@@ -134,3 +137,51 @@ func (mut MutableMap) IterRange(ctx context.Context, rng Range) (MapIter, error)
|
||||
func (mut MutableMap) HasEdits() bool {
|
||||
return mut.tuples.edits.Count() > 0
|
||||
}
|
||||
|
||||
func debugFormat(ctx context.Context, m MutableMap) (string, error) {
|
||||
kd, vd := m.keyDesc, m.valDesc
|
||||
|
||||
editIter := m.tuples.edits.IterAtStart()
|
||||
tupleIter, err := m.tuples.tree.iterAll(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var sb strings.Builder
|
||||
sb.WriteString("Mutable Map {\n")
|
||||
|
||||
c := strconv.Itoa(m.tuples.edits.Count())
|
||||
sb.WriteString("\tedits (count: " + c + ") {\n")
|
||||
for {
|
||||
k, v := editIter.Current()
|
||||
if k == nil {
|
||||
break
|
||||
}
|
||||
sb.WriteString("\t\t")
|
||||
sb.WriteString(kd.Format(k))
|
||||
sb.WriteString(": ")
|
||||
sb.WriteString(vd.Format(v))
|
||||
sb.WriteString(",\n")
|
||||
editIter.Advance()
|
||||
}
|
||||
sb.WriteString("\t},\n")
|
||||
|
||||
c = strconv.Itoa(m.tuples.tree.count())
|
||||
sb.WriteString("\ttree (count: " + c + ") {\n")
|
||||
for {
|
||||
k, v, err := tupleIter.Next(ctx)
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
sb.WriteString("\t\t")
|
||||
sb.WriteString(kd.Format(k))
|
||||
sb.WriteString(": ")
|
||||
sb.WriteString(vd.Format(v))
|
||||
sb.WriteString(",\n")
|
||||
}
|
||||
sb.WriteString("\t}\n}\n")
|
||||
return sb.String(), nil
|
||||
}
|
||||
|
||||
@@ -104,6 +104,14 @@ func TestMutableMapReads(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMutableMapFormat(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mutableMap, _ := makeMutableMap(t, 100)
|
||||
s, err := debugFormat(ctx, mutableMap.(MutableMap))
|
||||
assert.NoError(t, err)
|
||||
assert.NotEmpty(t, s)
|
||||
}
|
||||
|
||||
func makeMutableMap(t *testing.T, count int) (testMap, [][2]val.Tuple) {
|
||||
ctx := context.Background()
|
||||
ns := tree.NewTestNodeStore()
|
||||
|
||||
@@ -23,8 +23,6 @@ import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/dolthub/dolt/go/store/prolly/message"
|
||||
"github.com/dolthub/dolt/go/store/prolly/tree"
|
||||
"github.com/dolthub/dolt/go/store/val"
|
||||
)
|
||||
|
||||
@@ -71,6 +69,9 @@ func TestMutableMapWrites(t *testing.T) {
|
||||
})
|
||||
})
|
||||
}
|
||||
t.Run("test internal node splits", func(t *testing.T) {
|
||||
testInternalNodeSplits(t)
|
||||
})
|
||||
}
|
||||
|
||||
func testPointUpdates(t *testing.T, mapCount int) {
|
||||
@@ -425,32 +426,60 @@ func testBulkInserts(t *testing.T, size int) {
|
||||
}
|
||||
}
|
||||
|
||||
func testInternalNodeSplits(t *testing.T) {
|
||||
const n = 100_000
|
||||
var err error
|
||||
ctx := context.Background()
|
||||
|
||||
kd := val.NewTupleDescriptor(
|
||||
val.Type{Enc: val.Int32Enc},
|
||||
val.Type{Enc: val.Int32Enc},
|
||||
)
|
||||
vd := val.NewTupleDescriptor()
|
||||
bld := val.NewTupleBuilder(kd)
|
||||
|
||||
tuples := make([][2]val.Tuple, n)
|
||||
for i := range tuples {
|
||||
bld.PutInt32(0, int32(i))
|
||||
bld.PutInt32(1, int32(0))
|
||||
tuples[i][0] = bld.Build(sharedPool)
|
||||
tuples[i][1] = val.EmptyTuple
|
||||
}
|
||||
pm := prollyMapFromTuples(t, kd, vd, tuples).(Map)
|
||||
|
||||
// reproduces chunker panic (k = 10_600)
|
||||
repro := 20_000
|
||||
|
||||
for k := 100; k <= repro; k += 100 {
|
||||
mut := pm.Mutate()
|
||||
for j := 1; j <= k; j++ {
|
||||
bld.PutInt32(0, int32(j))
|
||||
bld.PutInt32(1, int32(j))
|
||||
key := bld.Build(sharedPool)
|
||||
value := val.EmptyTuple
|
||||
err = mut.Put(ctx, key, value)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
pm, err = mut.Map(ctx)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, n+k, pm.Count())
|
||||
}
|
||||
}
|
||||
|
||||
// utilities
|
||||
|
||||
func ascendingIntMap(t *testing.T, count int) Map {
|
||||
return ascendingIntMapWithStep(t, count, 1)
|
||||
}
|
||||
|
||||
func ascendingIntMapWithStep(t *testing.T, count, step int) Map {
|
||||
ctx := context.Background()
|
||||
ns := tree.NewTestNodeStore()
|
||||
|
||||
tuples := make([][2]val.Tuple, count)
|
||||
for i := range tuples {
|
||||
v := int64(i * step)
|
||||
tuples[i][0], tuples[i][1] = makePut(v, v)
|
||||
}
|
||||
|
||||
serializer := message.ProllyMapSerializer{Pool: ns.Pool()}
|
||||
chunker, err := tree.NewEmptyChunker(ctx, ns, serializer)
|
||||
require.NoError(t, err)
|
||||
|
||||
for _, pair := range tuples {
|
||||
err = chunker.AddPair(ctx, tree.Item(pair[0]), tree.Item(pair[1]))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
root, err := chunker.Done(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
return NewMap(root, ns, mutKeyDesc, mutKeyDesc)
|
||||
pm := prollyMapFromTuples(t, mutKeyDesc, mutValDesc, tuples)
|
||||
return pm.(Map)
|
||||
}
|
||||
|
||||
var mutKeyDesc = val.NewTupleDescriptor(
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"math"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/dolthub/dolt/go/store/hash"
|
||||
@@ -175,13 +176,17 @@ func TestWriteImmutableTree(t *testing.T) {
|
||||
return nil
|
||||
})
|
||||
|
||||
require.Equal(t, expLevel, root.Level())
|
||||
assert.Equal(t, expLevel, root.Level())
|
||||
if tt.checkSum {
|
||||
require.Equal(t, expSum, sum)
|
||||
assert.Equal(t, expSum, sum)
|
||||
}
|
||||
assert.Equal(t, tt.inputSize, byteCnt)
|
||||
assert.Equal(t, expUnfilled, unfilledCnt)
|
||||
if expLevel > 0 {
|
||||
for i := range expSubtrees {
|
||||
assert.Equal(t, expSubtrees[i], root.getSubtreeCount(i))
|
||||
}
|
||||
}
|
||||
require.Equal(t, tt.inputSize, byteCnt)
|
||||
require.Equal(t, expUnfilled, unfilledCnt)
|
||||
require.Equal(t, expSubtrees, root.getSubtreeCounts())
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -200,9 +205,9 @@ func expectedLevel(size, chunk int) int {
|
||||
return l
|
||||
}
|
||||
|
||||
func expectedSubtrees(size, chunk int) SubtreeCounts {
|
||||
func expectedSubtrees(size, chunk int) subtreeCounts {
|
||||
if size <= chunk {
|
||||
return SubtreeCounts{0}
|
||||
return subtreeCounts{0}
|
||||
}
|
||||
l := expectedLevel(size, chunk)
|
||||
|
||||
@@ -211,7 +216,7 @@ func expectedSubtrees(size, chunk int) SubtreeCounts {
|
||||
|
||||
filledSubtree := int(math.Pow(float64(intChunk), float64(l-1)))
|
||||
|
||||
subtrees := make(SubtreeCounts, 0)
|
||||
subtrees := make(subtreeCounts, 0)
|
||||
for size > filledSubtree {
|
||||
subtrees = append(subtrees, uint64(filledSubtree))
|
||||
size -= filledSubtree
|
||||
|
||||
@@ -27,10 +27,13 @@ import (
|
||||
|
||||
type Item []byte
|
||||
|
||||
type subtreeCounts []uint64
|
||||
|
||||
type Node struct {
|
||||
// keys and values contain sub-slices of |msg|,
|
||||
// allowing faster lookups by avoiding the vtable
|
||||
keys, values val.SlicedBuffer
|
||||
subtrees subtreeCounts
|
||||
count uint16
|
||||
msg message.Message
|
||||
}
|
||||
@@ -140,16 +143,24 @@ func (nd Node) getValue(i int) Item {
|
||||
return nd.values.GetSlice(i)
|
||||
}
|
||||
|
||||
func (nd *Node) getSubtreeCount(i int) uint64 {
|
||||
if nd.IsLeaf() {
|
||||
return 1
|
||||
}
|
||||
if nd.subtrees == nil {
|
||||
// deserializing subtree counts requires a
|
||||
// malloc, so we lazily load them here
|
||||
nd.subtrees = message.GetSubtrees(nd.msg)
|
||||
}
|
||||
return nd.subtrees[i]
|
||||
}
|
||||
|
||||
// getAddress returns the |ith| address of this node.
|
||||
// This method assumes values are 20-byte address hashes.
|
||||
func (nd Node) getAddress(i int) hash.Hash {
|
||||
return hash.New(nd.getValue(i))
|
||||
}
|
||||
|
||||
func (nd Node) getSubtreeCounts() SubtreeCounts {
|
||||
return message.GetSubtrees(nd.msg)
|
||||
}
|
||||
|
||||
func (nd Node) empty() bool {
|
||||
return nd.bytes() == nil || nd.count == 0
|
||||
}
|
||||
|
||||
@@ -66,7 +66,7 @@ func newNodeBuilder[S message.Serializer](serializer S, level int) (nb *nodeBuil
|
||||
type nodeBuilder[S message.Serializer] struct {
|
||||
keys, values [][]byte
|
||||
size, level int
|
||||
subtrees SubtreeCounts
|
||||
subtrees subtreeCounts
|
||||
serializer S
|
||||
}
|
||||
|
||||
|
||||
@@ -30,20 +30,10 @@ import (
|
||||
|
||||
// Cursor explores a tree of Nodes.
|
||||
type Cursor struct {
|
||||
nd Node
|
||||
idx int
|
||||
parent *Cursor
|
||||
subtrees SubtreeCounts
|
||||
nrw NodeStore
|
||||
}
|
||||
|
||||
type SubtreeCounts []uint64
|
||||
|
||||
func (sc SubtreeCounts) Sum() (s uint64) {
|
||||
for _, count := range sc {
|
||||
s += count
|
||||
}
|
||||
return
|
||||
nd Node
|
||||
idx int
|
||||
parent *Cursor
|
||||
nrw NodeStore
|
||||
}
|
||||
|
||||
type CompareFn func(left, right Item) int
|
||||
@@ -112,11 +102,8 @@ func NewCursorAtOrdinal(ctx context.Context, ns NodeStore, nd Node, ord uint64)
|
||||
return int(distance)
|
||||
}
|
||||
|
||||
// |subtrees| contains cardinalities of each child tree in |nd|
|
||||
subtrees := nd.getSubtreeCounts()
|
||||
|
||||
for idx = range subtrees {
|
||||
card := int64(subtrees[idx])
|
||||
for idx = 0; idx < nd.Count(); idx++ {
|
||||
card := int64(nd.getSubtreeCount(idx))
|
||||
if (distance - card) < 0 {
|
||||
break
|
||||
}
|
||||
@@ -230,10 +217,7 @@ func (cur *Cursor) currentSubtreeSize() uint64 {
|
||||
if cur.isLeaf() {
|
||||
return 1
|
||||
}
|
||||
if cur.subtrees == nil { // lazy load
|
||||
cur.subtrees = cur.nd.getSubtreeCounts()
|
||||
}
|
||||
return cur.subtrees[cur.idx]
|
||||
return cur.nd.getSubtreeCount(cur.idx)
|
||||
}
|
||||
|
||||
func (cur *Cursor) firstKey() Item {
|
||||
@@ -403,8 +387,6 @@ func (cur *Cursor) Advance(ctx context.Context) error {
|
||||
}
|
||||
|
||||
cur.skipToNodeStart()
|
||||
cur.subtrees = nil // lazy load
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -440,8 +422,6 @@ func (cur *Cursor) Retreat(ctx context.Context) error {
|
||||
}
|
||||
|
||||
cur.skipToNodeEnd()
|
||||
cur.subtrees = nil // lazy load
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -67,7 +67,7 @@ func TestRoundTripNodeItems(t *testing.T) {
|
||||
|
||||
func TestNodeSize(t *testing.T) {
|
||||
sz := unsafe.Sizeof(Node{})
|
||||
assert.Equal(t, 128, int(sz))
|
||||
assert.Equal(t, 152, int(sz))
|
||||
}
|
||||
|
||||
func TestNodeHashValueCompatibility(t *testing.T) {
|
||||
|
||||
@@ -25,13 +25,18 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/dolthub/dolt/go/libraries/utils/tracing"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/attribute"
|
||||
"go.opentelemetry.io/otel/trace"
|
||||
|
||||
"github.com/dolthub/dolt/go/store/d"
|
||||
"github.com/dolthub/dolt/go/store/hash"
|
||||
)
|
||||
|
||||
var emptyKey = orderedKey{}
|
||||
|
||||
var tracer = otel.Tracer("github.com/dolthub/dolt/go/store/types")
|
||||
|
||||
func newMetaTuple(ref Ref, key orderedKey, numLeaves uint64) (metaTuple, error) {
|
||||
d.PanicIfTrue(ref.buff == nil)
|
||||
w := newBinaryNomsWriter()
|
||||
@@ -416,10 +421,8 @@ func (ms metaSequence) isLeaf() bool {
|
||||
|
||||
// metaSequence interface
|
||||
func (ms metaSequence) getChildSequence(ctx context.Context, idx int) (sequence, error) {
|
||||
span, ctx := tracing.StartSpan(ctx, "metaSequence.getChildSequence")
|
||||
defer func() {
|
||||
span.Finish()
|
||||
}()
|
||||
ctx, span := tracer.Start(ctx, "metaSequence.getChildSequence")
|
||||
span.End()
|
||||
|
||||
item, err := ms.getItem(idx)
|
||||
|
||||
@@ -438,11 +441,11 @@ func (ms metaSequence) getChildSequence(ctx context.Context, idx int) (sequence,
|
||||
// Returns the sequences pointed to by all items[i], s.t. start <= i < end, and returns the
|
||||
// concatentation as one long composite sequence
|
||||
func (ms metaSequence) getCompositeChildSequence(ctx context.Context, start uint64, length uint64) (sequence, error) {
|
||||
span, ctx := tracing.StartSpan(ctx, "metaSequence.getChildSequence")
|
||||
span.LogKV("level", ms.treeLevel(), "length", length)
|
||||
defer func() {
|
||||
span.Finish()
|
||||
}()
|
||||
ctx, span := tracer.Start(ctx, "metaSequence.getChildSequence", trace.WithAttributes(
|
||||
attribute.Int64("level", int64(ms.treeLevel())),
|
||||
attribute.Int64("length", int64(length)),
|
||||
))
|
||||
defer span.End()
|
||||
|
||||
level := ms.treeLevel()
|
||||
d.PanicIfFalse(level > 0)
|
||||
|
||||
@@ -103,6 +103,64 @@ teardown() {
|
||||
[[ "$output" =~ "t1" ]] || false
|
||||
}
|
||||
|
||||
@test "replication: push on call dolt_branch(..." {
|
||||
cd repo1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_to_remote backup1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_heads main,new_branch
|
||||
dolt sql -q "create table t1 (a int primary key)"
|
||||
dolt sql -q "call dolt_commit('-am', 'commit')"
|
||||
dolt sql -q "call dolt_branch('new_branch')"
|
||||
|
||||
cd ..
|
||||
dolt clone file://./bac1 repo2
|
||||
cd repo2
|
||||
run dolt branch -av
|
||||
[ "$status" -eq 0 ]
|
||||
[ "${#lines[@]}" -eq 3 ]
|
||||
[[ "$output" =~ "remotes/origin/main" ]] || false
|
||||
[[ "$output" =~ "remotes/origin/new_branch" ]] || false
|
||||
}
|
||||
|
||||
@test "replication: push on call dolt_checkout(-b..." {
|
||||
cd repo1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_to_remote backup1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_heads main,new_branch
|
||||
dolt sql -q "create table t1 (a int primary key)"
|
||||
dolt sql -q "call dolt_commit('-am', 'commit')"
|
||||
dolt sql -q "call dolt_checkout('-b', 'new_branch')"
|
||||
|
||||
cd ..
|
||||
dolt clone file://./bac1 repo2
|
||||
cd repo2
|
||||
run dolt branch -av
|
||||
[ "$status" -eq 0 ]
|
||||
[ "${#lines[@]}" -eq 3 ]
|
||||
[[ "$output" =~ "remotes/origin/main" ]] || false
|
||||
[[ "$output" =~ "remotes/origin/new_branch" ]] || false
|
||||
}
|
||||
|
||||
@test "replication: push on call dolt_merge, fast-forward merge" {
|
||||
cd repo1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_to_remote backup1
|
||||
dolt config --local --add sqlserver.global.dolt_replicate_heads main,new_branch
|
||||
dolt sql -q "create table t1 (a int primary key)"
|
||||
dolt sql -q "call dolt_commit('-am', 'commit')"
|
||||
dolt sql -q "call dolt_checkout('-b', 'new_branch')"
|
||||
dolt sql -q "create table t2 (b int primary key)"
|
||||
dolt sql -q "call dolt_commit('-am', 'commit')"
|
||||
dolt sql -q "call dolt_checkout('main')"
|
||||
dolt sql -q "call dolt_merge('new_branch')"
|
||||
|
||||
cd ..
|
||||
dolt clone file://./bac1 repo2
|
||||
cd repo2
|
||||
run dolt ls
|
||||
[ "$status" -eq 0 ]
|
||||
[ "${#lines[@]}" -eq 3 ]
|
||||
[[ "$output" =~ "t1" ]] || false
|
||||
[[ "$output" =~ "t2" ]] || false
|
||||
}
|
||||
|
||||
@test "replication: pull non-main head" {
|
||||
dolt clone file://./rem1 repo2
|
||||
cd repo2
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
const knex = require("knex");
|
||||
const wtfnode = require("wtfnode")
|
||||
Socket = require('net').Socket;
|
||||
|
||||
const args = process.argv.slice(2);
|
||||
|
||||
const user = args[0];
|
||||
const port = args[1];
|
||||
const dbName = args[2];
|
||||
@@ -17,7 +18,7 @@ const db = knex({
|
||||
});
|
||||
|
||||
async function createTable() {
|
||||
val = await db.schema.createTable('test2', (table) => {
|
||||
let val = await db.schema.createTable('test2', (table) => {
|
||||
table.integer('id').primary()
|
||||
table.integer('foo')
|
||||
});
|
||||
@@ -25,12 +26,12 @@ async function createTable() {
|
||||
}
|
||||
|
||||
async function upsert(table, data) {
|
||||
val = await db(table).insert(data).onConflict().merge();
|
||||
let val = await db(table).insert(data).onConflict().merge();
|
||||
return val
|
||||
}
|
||||
|
||||
async function select() {
|
||||
val = await db.select('id', 'foo').from('test2');
|
||||
let val = await db.select('id', 'foo').from('test2');
|
||||
return val
|
||||
}
|
||||
|
||||
@@ -50,8 +51,34 @@ async function main() {
|
||||
throw new Error("Query failed")
|
||||
}
|
||||
|
||||
db.destroy()
|
||||
process.exit(0)
|
||||
await db.destroy();
|
||||
|
||||
// cc: https://github.com/dolthub/dolt/issues/3752
|
||||
setTimeout(async () => {
|
||||
let sockets = await getOpenSockets();
|
||||
|
||||
if (sockets.length > 0) {
|
||||
wtfnode.dump();
|
||||
process.exit(1);
|
||||
throw new Error("Database not properly destroyed. Hanging server connections");
|
||||
}
|
||||
|
||||
}, 3000);
|
||||
}
|
||||
|
||||
// cc: https://github.com/myndzi/wtfnode/blob/master/index.js#L457
|
||||
async function getOpenSockets() {
|
||||
let sockets = []
|
||||
process._getActiveHandles().forEach(function (h) {
|
||||
// handles can be null now? early exit to guard against this
|
||||
if (!h) { return; }
|
||||
|
||||
if (h instanceof Socket) {
|
||||
if ((h.fd == null) && (h.localAddress) && !(h.destroyed)) { sockets.push(h); }
|
||||
}
|
||||
});
|
||||
|
||||
return sockets
|
||||
}
|
||||
|
||||
main();
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
"dependencies": {
|
||||
"knex": "^1.0.7",
|
||||
"mysql": "^2.18.1",
|
||||
"mysql2": "^2.3.3"
|
||||
"mysql2": "^2.3.3",
|
||||
"wtfnode": "^0.9.1"
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user