From d6e365bb4235d7a7656a2f3bac7f66aa141cb7b1 Mon Sep 17 00:00:00 2001
From: Herman Schaaf <hermanschaaf@gmail.com>
Date: Mon, 10 Oct 2022 14:47:50 +0100
Subject: [PATCH 1/3] Make concurrency change backwards-compatible

---
 specs/source.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/specs/source.go b/specs/source.go
index 0aaf93f1b2..29dde59481 100644
--- a/specs/source.go
+++ b/specs/source.go
@@ -26,6 +26,7 @@ type Source struct {
 	Path string `json:"path,omitempty"`
 	// Registry can be github,local,grpc.
 	Registry            Registry `json:"registry,omitempty"`
+	Concurrency         uint64   `json:"concurrency,omitempty"` // deprecated: use TableConcurrency and ResourceConcurrency instead
 	TableConcurrency    uint64   `json:"table_concurrency,omitempty"`
 	ResourceConcurrency uint64   `json:"resource_concurrency,omitempty"`
 	// Tables to sync from the source plugin
@@ -53,6 +54,12 @@ func (s *Source) SetDefaults() {
 		s.Tables = []string{"*"}
 	}
 
+	if s.Concurrency != 0 {
+		// attempt to make a sensible backwards-compatible choice, but the CLI
+		// should raise a warning about this until concurrency is fully removed.
+		s.TableConcurrency = s.Concurrency
+		s.ResourceConcurrency = s.Concurrency
+	}
 	if s.TableConcurrency == 0 {
 		s.TableConcurrency = defaultTableConcurrency
 	}

From e8849cd8fc005a240e3bc9d33f950086d2644afa Mon Sep 17 00:00:00 2001
From: Herman Schaaf <hermanschaaf@gmail.com>
Date: Mon, 10 Oct 2022 14:53:51 +0100
Subject: [PATCH 2/3] Update

---
 specs/source.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/specs/source.go b/specs/source.go
index 29dde59481..3d39f8e3bf 100644
--- a/specs/source.go
+++ b/specs/source.go
@@ -54,9 +54,9 @@ func (s *Source) SetDefaults() {
 		s.Tables = []string{"*"}
 	}
 
-	if s.Concurrency != 0 {
+	if s.Concurrency != 0 && s.TableConcurrency == 0 && s.ResourceConcurrency == 0 {
 		// attempt to make a sensible backwards-compatible choice, but the CLI
-		// should raise a warning about this until concurrency is fully removed.
+		// should raise a warning about this until the `concurrency` option is fully removed.
 		s.TableConcurrency = s.Concurrency
 		s.ResourceConcurrency = s.Concurrency
 	}

From 1a43fb5a0bc6ec371d39a2b5b5d1f74d9da7764e Mon Sep 17 00:00:00 2001
From: Herman Schaaf <hermanschaaf@gmail.com>
Date: Mon, 10 Oct 2022 16:36:20 +0100
Subject: [PATCH 3/3] Add Validate function to get warnings

---
 clients/destination.go             |  22 ++
 clients/source.go                  |  22 ++
 internal/pb/base.pb.go             |   2 +-
 internal/pb/destination.pb.go      | 490 ++++++++++++++++++++---------
 internal/pb/destination.proto      |  11 +
 internal/pb/destination_grpc.pb.go |  40 ++-
 internal/pb/source.pb.go           | 408 +++++++++++++++++-------
 internal/pb/source.proto           |  12 +
 internal/pb/source_grpc.pb.go      |  40 ++-
 internal/servers/destinations.go   |  24 ++
 internal/servers/source.go         |  36 ++-
 plugins/destination.go             |   7 +
 plugins/source.go                  |  14 +
 specs/destination.go               |   6 +
 specs/source.go                    |  12 +
 15 files changed, 879 insertions(+), 267 deletions(-)

diff --git a/clients/destination.go b/clients/destination.go
index 80d94d4905..5410344ec4 100644
--- a/clients/destination.go
+++ b/clients/destination.go
@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"io"
 	"net"
 	"os"
@@ -193,6 +195,26 @@ func (c *DestinationClient) Initialize(ctx context.Context, spec specs.Destinati
 	return nil
 }
 
+func (c *DestinationClient) Validate(ctx context.Context, spec specs.Destination) (warnings, errors []string, err error) {
+	b, err := json.Marshal(spec)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to marshal destination spec: %w", err)
+	}
+	resp, err := c.pbClient.Validate(ctx, &pb.ValidateDestination_Request{
+		Spec: b,
+	})
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok && st.Code() == codes.Unimplemented {
+			// Backwards-compatibility with older plugin versions that don't support Validate().
+			// In this case, we only return one warning: that the plugin should be updated.
+			return []string{"the version of this plugin is outdated and should be updated"}, nil, nil
+		}
+		return nil, nil, fmt.Errorf("failed to call Validate: %w", err)
+	}
+	return resp.Warnings, resp.Errors, nil
+}
+
 func (c *DestinationClient) Migrate(ctx context.Context, tables []*schema.Table) error {
 	b, err := json.Marshal(tables)
 	if err != nil {
diff --git a/clients/source.go b/clients/source.go
index a135684d36..05efdffeab 100644
--- a/clients/source.go
+++ b/clients/source.go
@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"io"
 	"net"
 	"os"
@@ -195,6 +197,26 @@ func (c *SourceClient) GetTables(ctx context.Context) ([]*schema.Table, error) {
 	return tables, nil
 }
 
+func (c *SourceClient) Validate(ctx context.Context, spec specs.Source) (warnings, errors []string, err error) {
+	b, err := json.Marshal(spec)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to marshal source spec: %w", err)
+	}
+	resp, err := c.pbClient.Validate(ctx, &pb.ValidateSource_Request{
+		Spec: b,
+	})
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok && st.Code() == codes.Unimplemented {
+			// Backwards-compatibility with older plugin versions that don't support Validate().
+			// In this case, we only return one warning: that the plugin should be updated.
+			return []string{"the version of this plugin is outdated and should be updated"}, nil, nil
+		}
+		return nil, nil, fmt.Errorf("failed to call Validate: %w", err)
+	}
+	return resp.Warnings, resp.Errors, nil
+}
+
 // Sync start syncing for the source client per the given spec and returning the results
 // in the given channel. res is marshaled schema.Resource. We are not unmarshalling this for performance reasons
 // as usually this is sent over-the-wire anyway to a destination plugin
diff --git a/internal/pb/base.pb.go b/internal/pb/base.pb.go
index b3a77b7ac1..b716a03a42 100644
--- a/internal/pb/base.pb.go
+++ b/internal/pb/base.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.6
+// 	protoc        v3.21.5
 // source: internal/pb/base.proto
 
 package pb
diff --git a/internal/pb/destination.pb.go b/internal/pb/destination.pb.go
index d1f5e97270..b5933b7b11 100644
--- a/internal/pb/destination.pb.go
+++ b/internal/pb/destination.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.6
+// 	protoc        v3.21.5
 // source: internal/pb/destination.proto
 
 package pb
@@ -21,6 +21,44 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type ValidateDestination struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *ValidateDestination) Reset() {
+	*x = ValidateDestination{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_destination_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateDestination) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateDestination) ProtoMessage() {}
+
+func (x *ValidateDestination) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_destination_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateDestination.ProtoReflect.Descriptor instead.
+func (*ValidateDestination) Descriptor() ([]byte, []int) {
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0}
+}
+
 type Migrate struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -30,7 +68,7 @@ type Migrate struct {
 func (x *Migrate) Reset() {
 	*x = Migrate{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[0]
+		mi := &file_internal_pb_destination_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -43,7 +81,7 @@ func (x *Migrate) String() string {
 func (*Migrate) ProtoMessage() {}
 
 func (x *Migrate) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[0]
+	mi := &file_internal_pb_destination_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -56,7 +94,7 @@ func (x *Migrate) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Migrate.ProtoReflect.Descriptor instead.
 func (*Migrate) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1}
 }
 
 type Write struct {
@@ -68,7 +106,7 @@ type Write struct {
 func (x *Write) Reset() {
 	*x = Write{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[1]
+		mi := &file_internal_pb_destination_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -81,7 +119,7 @@ func (x *Write) String() string {
 func (*Write) ProtoMessage() {}
 
 func (x *Write) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[1]
+	mi := &file_internal_pb_destination_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -94,7 +132,7 @@ func (x *Write) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Write.ProtoReflect.Descriptor instead.
 func (*Write) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2}
 }
 
 type Close struct {
@@ -106,7 +144,7 @@ type Close struct {
 func (x *Close) Reset() {
 	*x = Close{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[2]
+		mi := &file_internal_pb_destination_proto_msgTypes[3]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -119,7 +157,7 @@ func (x *Close) String() string {
 func (*Close) ProtoMessage() {}
 
 func (x *Close) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[2]
+	mi := &file_internal_pb_destination_proto_msgTypes[3]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -132,7 +170,7 @@ func (x *Close) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Close.ProtoReflect.Descriptor instead.
 func (*Close) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3}
 }
 
 type DeleteStale struct {
@@ -144,7 +182,7 @@ type DeleteStale struct {
 func (x *DeleteStale) Reset() {
 	*x = DeleteStale{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[3]
+		mi := &file_internal_pb_destination_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -157,7 +195,7 @@ func (x *DeleteStale) String() string {
 func (*DeleteStale) ProtoMessage() {}
 
 func (x *DeleteStale) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[3]
+	mi := &file_internal_pb_destination_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -170,7 +208,109 @@ func (x *DeleteStale) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use DeleteStale.ProtoReflect.Descriptor instead.
 func (*DeleteStale) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{4}
+}
+
+type ValidateDestination_Request struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Spec []byte `protobuf:"bytes,1,opt,name=spec,proto3" json:"spec,omitempty"`
+}
+
+func (x *ValidateDestination_Request) Reset() {
+	*x = ValidateDestination_Request{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_destination_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateDestination_Request) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateDestination_Request) ProtoMessage() {}
+
+func (x *ValidateDestination_Request) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_destination_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateDestination_Request.ProtoReflect.Descriptor instead.
+func (*ValidateDestination_Request) Descriptor() ([]byte, []int) {
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0, 0}
+}
+
+func (x *ValidateDestination_Request) GetSpec() []byte {
+	if x != nil {
+		return x.Spec
+	}
+	return nil
+}
+
+type ValidateDestination_Response struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Warnings []string `protobuf:"bytes,1,rep,name=warnings,proto3" json:"warnings,omitempty"`
+	Errors   []string `protobuf:"bytes,2,rep,name=errors,proto3" json:"errors,omitempty"`
+}
+
+func (x *ValidateDestination_Response) Reset() {
+	*x = ValidateDestination_Response{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_destination_proto_msgTypes[6]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateDestination_Response) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateDestination_Response) ProtoMessage() {}
+
+func (x *ValidateDestination_Response) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_destination_proto_msgTypes[6]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateDestination_Response.ProtoReflect.Descriptor instead.
+func (*ValidateDestination_Response) Descriptor() ([]byte, []int) {
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0, 1}
+}
+
+func (x *ValidateDestination_Response) GetWarnings() []string {
+	if x != nil {
+		return x.Warnings
+	}
+	return nil
+}
+
+func (x *ValidateDestination_Response) GetErrors() []string {
+	if x != nil {
+		return x.Errors
+	}
+	return nil
 }
 
 type Migrate_Request struct {
@@ -186,7 +326,7 @@ type Migrate_Request struct {
 func (x *Migrate_Request) Reset() {
 	*x = Migrate_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[4]
+		mi := &file_internal_pb_destination_proto_msgTypes[7]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -199,7 +339,7 @@ func (x *Migrate_Request) String() string {
 func (*Migrate_Request) ProtoMessage() {}
 
 func (x *Migrate_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[4]
+	mi := &file_internal_pb_destination_proto_msgTypes[7]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -212,7 +352,7 @@ func (x *Migrate_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Migrate_Request.ProtoReflect.Descriptor instead.
 func (*Migrate_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0, 0}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1, 0}
 }
 
 func (x *Migrate_Request) GetName() string {
@@ -245,7 +385,7 @@ type Migrate_Response struct {
 func (x *Migrate_Response) Reset() {
 	*x = Migrate_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[5]
+		mi := &file_internal_pb_destination_proto_msgTypes[8]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -258,7 +398,7 @@ func (x *Migrate_Response) String() string {
 func (*Migrate_Response) ProtoMessage() {}
 
 func (x *Migrate_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[5]
+	mi := &file_internal_pb_destination_proto_msgTypes[8]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -271,7 +411,7 @@ func (x *Migrate_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Migrate_Response.ProtoReflect.Descriptor instead.
 func (*Migrate_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{0, 1}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1, 1}
 }
 
 type Write_Request struct {
@@ -288,7 +428,7 @@ type Write_Request struct {
 func (x *Write_Request) Reset() {
 	*x = Write_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[6]
+		mi := &file_internal_pb_destination_proto_msgTypes[9]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -301,7 +441,7 @@ func (x *Write_Request) String() string {
 func (*Write_Request) ProtoMessage() {}
 
 func (x *Write_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[6]
+	mi := &file_internal_pb_destination_proto_msgTypes[9]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -314,7 +454,7 @@ func (x *Write_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Write_Request.ProtoReflect.Descriptor instead.
 func (*Write_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1, 0}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2, 0}
 }
 
 func (x *Write_Request) GetResource() []byte {
@@ -349,7 +489,7 @@ type Write_Response struct {
 func (x *Write_Response) Reset() {
 	*x = Write_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[7]
+		mi := &file_internal_pb_destination_proto_msgTypes[10]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -362,7 +502,7 @@ func (x *Write_Response) String() string {
 func (*Write_Response) ProtoMessage() {}
 
 func (x *Write_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[7]
+	mi := &file_internal_pb_destination_proto_msgTypes[10]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -375,7 +515,7 @@ func (x *Write_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Write_Response.ProtoReflect.Descriptor instead.
 func (*Write_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{1, 1}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2, 1}
 }
 
 func (x *Write_Response) GetFailedWrites() uint64 {
@@ -394,7 +534,7 @@ type Close_Request struct {
 func (x *Close_Request) Reset() {
 	*x = Close_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[8]
+		mi := &file_internal_pb_destination_proto_msgTypes[11]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -407,7 +547,7 @@ func (x *Close_Request) String() string {
 func (*Close_Request) ProtoMessage() {}
 
 func (x *Close_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[8]
+	mi := &file_internal_pb_destination_proto_msgTypes[11]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -420,7 +560,7 @@ func (x *Close_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Close_Request.ProtoReflect.Descriptor instead.
 func (*Close_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2, 0}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3, 0}
 }
 
 type Close_Response struct {
@@ -432,7 +572,7 @@ type Close_Response struct {
 func (x *Close_Response) Reset() {
 	*x = Close_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[9]
+		mi := &file_internal_pb_destination_proto_msgTypes[12]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -445,7 +585,7 @@ func (x *Close_Response) String() string {
 func (*Close_Response) ProtoMessage() {}
 
 func (x *Close_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[9]
+	mi := &file_internal_pb_destination_proto_msgTypes[12]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -458,7 +598,7 @@ func (x *Close_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Close_Response.ProtoReflect.Descriptor instead.
 func (*Close_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{2, 1}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3, 1}
 }
 
 type DeleteStale_Request struct {
@@ -474,7 +614,7 @@ type DeleteStale_Request struct {
 func (x *DeleteStale_Request) Reset() {
 	*x = DeleteStale_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[10]
+		mi := &file_internal_pb_destination_proto_msgTypes[13]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -487,7 +627,7 @@ func (x *DeleteStale_Request) String() string {
 func (*DeleteStale_Request) ProtoMessage() {}
 
 func (x *DeleteStale_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[10]
+	mi := &file_internal_pb_destination_proto_msgTypes[13]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -500,7 +640,7 @@ func (x *DeleteStale_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use DeleteStale_Request.ProtoReflect.Descriptor instead.
 func (*DeleteStale_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3, 0}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{4, 0}
 }
 
 func (x *DeleteStale_Request) GetSource() string {
@@ -535,7 +675,7 @@ type DeleteStale_Response struct {
 func (x *DeleteStale_Response) Reset() {
 	*x = DeleteStale_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_destination_proto_msgTypes[11]
+		mi := &file_internal_pb_destination_proto_msgTypes[14]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -548,7 +688,7 @@ func (x *DeleteStale_Response) String() string {
 func (*DeleteStale_Response) ProtoMessage() {}
 
 func (x *DeleteStale_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_destination_proto_msgTypes[11]
+	mi := &file_internal_pb_destination_proto_msgTypes[14]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -561,7 +701,7 @@ func (x *DeleteStale_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use DeleteStale_Response.ProtoReflect.Descriptor instead.
 func (*DeleteStale_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_destination_proto_rawDescGZIP(), []int{3, 1}
+	return file_internal_pb_destination_proto_rawDescGZIP(), []int{4, 1}
 }
 
 func (x *DeleteStale_Response) GetFailedDeletes() uint64 {
@@ -580,67 +720,80 @@ var file_internal_pb_destination_proto_rawDesc = []byte{
 	0x2f, 0x70, 0x62, 0x2f, 0x62, 0x61, 0x73, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f,
 	0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f,
 	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22,
-	0x66, 0x0a, 0x07, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74, 0x65, 0x1a, 0x4f, 0x0a, 0x07, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72,
-	0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73,
-	0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x18, 0x03, 0x20,
-	0x01, 0x28, 0x0c, 0x52, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x1a, 0x0a, 0x0a, 0x08, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xb1, 0x01, 0x0a, 0x05, 0x57, 0x72, 0x69, 0x74,
-	0x65, 0x1a, 0x77, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72,
-	0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x03, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52,
-	0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x1a, 0x2f, 0x0a, 0x08, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64,
-	0x5f, 0x77, 0x72, 0x69, 0x74, 0x65, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0c, 0x66,
-	0x61, 0x69, 0x6c, 0x65, 0x64, 0x57, 0x72, 0x69, 0x74, 0x65, 0x73, 0x22, 0x1e, 0x0a, 0x05, 0x43,
-	0x6c, 0x6f, 0x73, 0x65, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x0a, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xb5, 0x01, 0x0a, 0x0b,
-	0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x1a, 0x73, 0x0a, 0x07, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x38,
-	0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74,
-	0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x61, 0x62, 0x6c,
-	0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x73,
-	0x1a, 0x31, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x0e,
-	0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x73, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x44, 0x65, 0x6c, 0x65,
-	0x74, 0x65, 0x73, 0x32, 0xc2, 0x03, 0x0a, 0x0b, 0x44, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x61, 0x74,
-	0x69, 0x6f, 0x6e, 0x12, 0x3a, 0x0a, 0x07, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x16,
-	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52,
-	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47,
-	0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12,
-	0x43, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e,
-	0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x73, 0x70,
-	0x6f, 0x6e, 0x73, 0x65, 0x12, 0x40, 0x0a, 0x09, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72,
-	0x65, 0x12, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
-	0x75, 0x72, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x2e, 0x52, 0x65,
-	0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3a, 0x0a, 0x07, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74,
-	0x65, 0x12, 0x16, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74,
-	0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x2e, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
-	0x73, 0x65, 0x12, 0x36, 0x0a, 0x05, 0x57, 0x72, 0x69, 0x74, 0x65, 0x12, 0x14, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x15, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x2e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x28, 0x01, 0x12, 0x34, 0x0a, 0x05, 0x43, 0x6c,
-	0x6f, 0x73, 0x65, 0x12, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6c, 0x6f, 0x73,
-	0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x70, 0x72, 0x6f, 0x74,
-	0x6f, 0x2e, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x12, 0x46, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x12,
-	0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74,
-	0x61, 0x6c, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x2e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x42, 0x05, 0x5a, 0x03, 0x2f, 0x70, 0x62, 0x62,
-	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x74, 0x0a, 0x13, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x44, 0x65, 0x73, 0x74, 0x69,
+	0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x1a, 0x1d, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x12, 0x12, 0x0a, 0x04, 0x73, 0x70, 0x65, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52,
+	0x04, 0x73, 0x70, 0x65, 0x63, 0x1a, 0x3e, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x12, 0x1a, 0x0a, 0x08, 0x77, 0x61, 0x72, 0x6e, 0x69, 0x6e, 0x67, 0x73, 0x18, 0x01, 0x20,
+	0x03, 0x28, 0x09, 0x52, 0x08, 0x77, 0x61, 0x72, 0x6e, 0x69, 0x6e, 0x67, 0x73, 0x12, 0x16, 0x0a,
+	0x06, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x65,
+	0x72, 0x72, 0x6f, 0x72, 0x73, 0x22, 0x66, 0x0a, 0x07, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74, 0x65,
+	0x1a, 0x4f, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e,
+	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12,
+	0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x61, 0x62,
+	0x6c, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65,
+	0x73, 0x1a, 0x0a, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xb1, 0x01,
+	0x0a, 0x05, 0x57, 0x72, 0x69, 0x74, 0x65, 0x1a, 0x77, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x16,
+	0x0a, 0x06, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
+	0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65,
+	0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x1a, 0x2f, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x23, 0x0a, 0x0d,
+	0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x77, 0x72, 0x69, 0x74, 0x65, 0x73, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x04, 0x52, 0x0c, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x57, 0x72, 0x69, 0x74, 0x65,
+	0x73, 0x22, 0x1e, 0x0a, 0x05, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x0a, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x22, 0xb5, 0x01, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c,
+	0x65, 0x1a, 0x73, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a, 0x06,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x6f,
+	0x75, 0x72, 0x63, 0x65, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
+	0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74,
+	0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x16,
+	0x0a, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06,
+	0x74, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x1a, 0x31, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x64, 0x65, 0x6c,
+	0x65, 0x74, 0x65, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0d, 0x66, 0x61, 0x69, 0x6c,
+	0x65, 0x64, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x73, 0x32, 0x97, 0x04, 0x0a, 0x0b, 0x44, 0x65,
+	0x73, 0x74, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3a, 0x0a, 0x07, 0x47, 0x65, 0x74,
+	0x4e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74,
+	0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52, 0x65, 0x73,
+	0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73,
+	0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x56,
+	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f,
+	0x6e, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x40, 0x0a, 0x09, 0x43, 0x6f,
+	0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x12, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e,
+	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x75, 0x72, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
+	0x74, 0x1a, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
+	0x75, 0x72, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x53, 0x0a, 0x08,
+	0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x12, 0x22, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x44, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x61,
+	0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x23, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x44, 0x65, 0x73,
+	0x74, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x12, 0x3a, 0x0a, 0x07, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74, 0x65, 0x12, 0x16, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4d, 0x69, 0x67, 0x72, 0x61, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x4d, 0x69, 0x67,
+	0x72, 0x61, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x36, 0x0a,
+	0x05, 0x57, 0x72, 0x69, 0x74, 0x65, 0x12, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x57,
+	0x72, 0x69, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x57, 0x72, 0x69, 0x74, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x28, 0x01, 0x12, 0x34, 0x0a, 0x05, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x12, 0x14,
+	0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6c, 0x6f, 0x73, 0x65, 0x2e, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x43, 0x6c, 0x6f,
+	0x73, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x46, 0x0a, 0x0b, 0x44,
+	0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x12, 0x1a, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x2e, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x44,
+	0x65, 0x6c, 0x65, 0x74, 0x65, 0x53, 0x74, 0x61, 0x6c, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f,
+	0x6e, 0x73, 0x65, 0x42, 0x05, 0x5a, 0x03, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x33,
 }
 
 var (
@@ -655,47 +808,52 @@ func file_internal_pb_destination_proto_rawDescGZIP() []byte {
 	return file_internal_pb_destination_proto_rawDescData
 }
 
-var file_internal_pb_destination_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
+var file_internal_pb_destination_proto_msgTypes = make([]protoimpl.MessageInfo, 15)
 var file_internal_pb_destination_proto_goTypes = []interface{}{
-	(*Migrate)(nil),               // 0: proto.Migrate
-	(*Write)(nil),                 // 1: proto.Write
-	(*Close)(nil),                 // 2: proto.Close
-	(*DeleteStale)(nil),           // 3: proto.DeleteStale
-	(*Migrate_Request)(nil),       // 4: proto.Migrate.Request
-	(*Migrate_Response)(nil),      // 5: proto.Migrate.Response
-	(*Write_Request)(nil),         // 6: proto.Write.Request
-	(*Write_Response)(nil),        // 7: proto.Write.Response
-	(*Close_Request)(nil),         // 8: proto.Close.Request
-	(*Close_Response)(nil),        // 9: proto.Close.Response
-	(*DeleteStale_Request)(nil),   // 10: proto.DeleteStale.Request
-	(*DeleteStale_Response)(nil),  // 11: proto.DeleteStale.Response
-	(*timestamppb.Timestamp)(nil), // 12: google.protobuf.Timestamp
-	(*GetName_Request)(nil),       // 13: proto.GetName.Request
-	(*GetVersion_Request)(nil),    // 14: proto.GetVersion.Request
-	(*Configure_Request)(nil),     // 15: proto.Configure.Request
-	(*GetName_Response)(nil),      // 16: proto.GetName.Response
-	(*GetVersion_Response)(nil),   // 17: proto.GetVersion.Response
-	(*Configure_Response)(nil),    // 18: proto.Configure.Response
+	(*ValidateDestination)(nil),          // 0: proto.ValidateDestination
+	(*Migrate)(nil),                      // 1: proto.Migrate
+	(*Write)(nil),                        // 2: proto.Write
+	(*Close)(nil),                        // 3: proto.Close
+	(*DeleteStale)(nil),                  // 4: proto.DeleteStale
+	(*ValidateDestination_Request)(nil),  // 5: proto.ValidateDestination.Request
+	(*ValidateDestination_Response)(nil), // 6: proto.ValidateDestination.Response
+	(*Migrate_Request)(nil),              // 7: proto.Migrate.Request
+	(*Migrate_Response)(nil),             // 8: proto.Migrate.Response
+	(*Write_Request)(nil),                // 9: proto.Write.Request
+	(*Write_Response)(nil),               // 10: proto.Write.Response
+	(*Close_Request)(nil),                // 11: proto.Close.Request
+	(*Close_Response)(nil),               // 12: proto.Close.Response
+	(*DeleteStale_Request)(nil),          // 13: proto.DeleteStale.Request
+	(*DeleteStale_Response)(nil),         // 14: proto.DeleteStale.Response
+	(*timestamppb.Timestamp)(nil),        // 15: google.protobuf.Timestamp
+	(*GetName_Request)(nil),              // 16: proto.GetName.Request
+	(*GetVersion_Request)(nil),           // 17: proto.GetVersion.Request
+	(*Configure_Request)(nil),            // 18: proto.Configure.Request
+	(*GetName_Response)(nil),             // 19: proto.GetName.Response
+	(*GetVersion_Response)(nil),          // 20: proto.GetVersion.Response
+	(*Configure_Response)(nil),           // 21: proto.Configure.Response
 }
 var file_internal_pb_destination_proto_depIdxs = []int32{
-	12, // 0: proto.Write.Request.timestamp:type_name -> google.protobuf.Timestamp
-	12, // 1: proto.DeleteStale.Request.timestamp:type_name -> google.protobuf.Timestamp
-	13, // 2: proto.Destination.GetName:input_type -> proto.GetName.Request
-	14, // 3: proto.Destination.GetVersion:input_type -> proto.GetVersion.Request
-	15, // 4: proto.Destination.Configure:input_type -> proto.Configure.Request
-	4,  // 5: proto.Destination.Migrate:input_type -> proto.Migrate.Request
-	6,  // 6: proto.Destination.Write:input_type -> proto.Write.Request
-	8,  // 7: proto.Destination.Close:input_type -> proto.Close.Request
-	10, // 8: proto.Destination.DeleteStale:input_type -> proto.DeleteStale.Request
-	16, // 9: proto.Destination.GetName:output_type -> proto.GetName.Response
-	17, // 10: proto.Destination.GetVersion:output_type -> proto.GetVersion.Response
-	18, // 11: proto.Destination.Configure:output_type -> proto.Configure.Response
-	5,  // 12: proto.Destination.Migrate:output_type -> proto.Migrate.Response
-	7,  // 13: proto.Destination.Write:output_type -> proto.Write.Response
-	9,  // 14: proto.Destination.Close:output_type -> proto.Close.Response
-	11, // 15: proto.Destination.DeleteStale:output_type -> proto.DeleteStale.Response
-	9,  // [9:16] is the sub-list for method output_type
-	2,  // [2:9] is the sub-list for method input_type
+	15, // 0: proto.Write.Request.timestamp:type_name -> google.protobuf.Timestamp
+	15, // 1: proto.DeleteStale.Request.timestamp:type_name -> google.protobuf.Timestamp
+	16, // 2: proto.Destination.GetName:input_type -> proto.GetName.Request
+	17, // 3: proto.Destination.GetVersion:input_type -> proto.GetVersion.Request
+	18, // 4: proto.Destination.Configure:input_type -> proto.Configure.Request
+	5,  // 5: proto.Destination.Validate:input_type -> proto.ValidateDestination.Request
+	7,  // 6: proto.Destination.Migrate:input_type -> proto.Migrate.Request
+	9,  // 7: proto.Destination.Write:input_type -> proto.Write.Request
+	11, // 8: proto.Destination.Close:input_type -> proto.Close.Request
+	13, // 9: proto.Destination.DeleteStale:input_type -> proto.DeleteStale.Request
+	19, // 10: proto.Destination.GetName:output_type -> proto.GetName.Response
+	20, // 11: proto.Destination.GetVersion:output_type -> proto.GetVersion.Response
+	21, // 12: proto.Destination.Configure:output_type -> proto.Configure.Response
+	6,  // 13: proto.Destination.Validate:output_type -> proto.ValidateDestination.Response
+	8,  // 14: proto.Destination.Migrate:output_type -> proto.Migrate.Response
+	10, // 15: proto.Destination.Write:output_type -> proto.Write.Response
+	12, // 16: proto.Destination.Close:output_type -> proto.Close.Response
+	14, // 17: proto.Destination.DeleteStale:output_type -> proto.DeleteStale.Response
+	10, // [10:18] is the sub-list for method output_type
+	2,  // [2:10] is the sub-list for method input_type
 	2,  // [2:2] is the sub-list for extension type_name
 	2,  // [2:2] is the sub-list for extension extendee
 	0,  // [0:2] is the sub-list for field type_name
@@ -709,7 +867,7 @@ func file_internal_pb_destination_proto_init() {
 	file_internal_pb_base_proto_init()
 	if !protoimpl.UnsafeEnabled {
 		file_internal_pb_destination_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Migrate); i {
+			switch v := v.(*ValidateDestination); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -721,7 +879,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Write); i {
+			switch v := v.(*Migrate); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -733,7 +891,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Close); i {
+			switch v := v.(*Write); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -745,7 +903,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DeleteStale); i {
+			switch v := v.(*Close); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -757,7 +915,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Migrate_Request); i {
+			switch v := v.(*DeleteStale); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -769,7 +927,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Migrate_Response); i {
+			switch v := v.(*ValidateDestination_Request); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -781,7 +939,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Write_Request); i {
+			switch v := v.(*ValidateDestination_Response); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -793,7 +951,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Write_Response); i {
+			switch v := v.(*Migrate_Request); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -805,7 +963,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Close_Request); i {
+			switch v := v.(*Migrate_Response); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -817,7 +975,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Close_Response); i {
+			switch v := v.(*Write_Request); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -829,7 +987,7 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*DeleteStale_Request); i {
+			switch v := v.(*Write_Response); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -841,6 +999,42 @@ func file_internal_pb_destination_proto_init() {
 			}
 		}
 		file_internal_pb_destination_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Close_Request); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_destination_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Close_Response); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_destination_proto_msgTypes[13].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*DeleteStale_Request); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_destination_proto_msgTypes[14].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*DeleteStale_Response); i {
 			case 0:
 				return &v.state
@@ -859,7 +1053,7 @@ func file_internal_pb_destination_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_internal_pb_destination_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   12,
+			NumMessages:   15,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/internal/pb/destination.proto b/internal/pb/destination.proto
index f03b8f7913..5ef880062a 100644
--- a/internal/pb/destination.proto
+++ b/internal/pb/destination.proto
@@ -11,6 +11,8 @@ service Destination {
   rpc GetVersion(GetVersion.Request) returns (GetVersion.Response);
   // Configure the plugin with the given credentials and mode
   rpc Configure(Configure.Request) returns (Configure.Response);
+  // Validate returns warnings and errors about the spec.
+  rpc Validate(ValidateDestination.Request) returns (ValidateDestination.Response);
   // Migrate tables to the given plugin version
   rpc Migrate(Migrate.Request) returns (Migrate.Response);
   // Write resources
@@ -22,6 +24,15 @@ service Destination {
   rpc DeleteStale(DeleteStale.Request) returns (DeleteStale.Response);
 }
 
+message ValidateDestination {
+  message Request {
+    bytes spec = 1;
+  }
+  message Response {
+    repeated string warnings = 1;
+    repeated string errors = 2;
+  }
+}
 
 message Migrate {
   message Request {
diff --git a/internal/pb/destination_grpc.pb.go b/internal/pb/destination_grpc.pb.go
index b2e68dd670..dfb8537425 100644
--- a/internal/pb/destination_grpc.pb.go
+++ b/internal/pb/destination_grpc.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.2.0
-// - protoc             v3.21.6
+// - protoc             v3.21.5
 // source: internal/pb/destination.proto
 
 package pb
@@ -28,6 +28,8 @@ type DestinationClient interface {
 	GetVersion(ctx context.Context, in *GetVersion_Request, opts ...grpc.CallOption) (*GetVersion_Response, error)
 	// Configure the plugin with the given credentials and mode
 	Configure(ctx context.Context, in *Configure_Request, opts ...grpc.CallOption) (*Configure_Response, error)
+	// Validate returns warnings and errors about the spec.
+	Validate(ctx context.Context, in *ValidateDestination_Request, opts ...grpc.CallOption) (*ValidateDestination_Response, error)
 	// Migrate tables to the given plugin version
 	Migrate(ctx context.Context, in *Migrate_Request, opts ...grpc.CallOption) (*Migrate_Response, error)
 	// Write resources
@@ -74,6 +76,15 @@ func (c *destinationClient) Configure(ctx context.Context, in *Configure_Request
 	return out, nil
 }
 
+func (c *destinationClient) Validate(ctx context.Context, in *ValidateDestination_Request, opts ...grpc.CallOption) (*ValidateDestination_Response, error) {
+	out := new(ValidateDestination_Response)
+	err := c.cc.Invoke(ctx, "/proto.Destination/Validate", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *destinationClient) Migrate(ctx context.Context, in *Migrate_Request, opts ...grpc.CallOption) (*Migrate_Response, error) {
 	out := new(Migrate_Response)
 	err := c.cc.Invoke(ctx, "/proto.Destination/Migrate", in, out, opts...)
@@ -145,6 +156,8 @@ type DestinationServer interface {
 	GetVersion(context.Context, *GetVersion_Request) (*GetVersion_Response, error)
 	// Configure the plugin with the given credentials and mode
 	Configure(context.Context, *Configure_Request) (*Configure_Response, error)
+	// Validate returns warnings and errors about the spec.
+	Validate(context.Context, *ValidateDestination_Request) (*ValidateDestination_Response, error)
 	// Migrate tables to the given plugin version
 	Migrate(context.Context, *Migrate_Request) (*Migrate_Response, error)
 	// Write resources
@@ -170,6 +183,9 @@ func (UnimplementedDestinationServer) GetVersion(context.Context, *GetVersion_Re
 func (UnimplementedDestinationServer) Configure(context.Context, *Configure_Request) (*Configure_Response, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Configure not implemented")
 }
+func (UnimplementedDestinationServer) Validate(context.Context, *ValidateDestination_Request) (*ValidateDestination_Response, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Validate not implemented")
+}
 func (UnimplementedDestinationServer) Migrate(context.Context, *Migrate_Request) (*Migrate_Response, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Migrate not implemented")
 }
@@ -249,6 +265,24 @@ func _Destination_Configure_Handler(srv interface{}, ctx context.Context, dec fu
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Destination_Validate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ValidateDestination_Request)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(DestinationServer).Validate(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/proto.Destination/Validate",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(DestinationServer).Validate(ctx, req.(*ValidateDestination_Request))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _Destination_Migrate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(Migrate_Request)
 	if err := dec(in); err != nil {
@@ -348,6 +382,10 @@ var Destination_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "Configure",
 			Handler:    _Destination_Configure_Handler,
 		},
+		{
+			MethodName: "Validate",
+			Handler:    _Destination_Validate_Handler,
+		},
 		{
 			MethodName: "Migrate",
 			Handler:    _Destination_Migrate_Handler,
diff --git a/internal/pb/source.pb.go b/internal/pb/source.pb.go
index 7673fab13c..8dfb3bbb0f 100644
--- a/internal/pb/source.pb.go
+++ b/internal/pb/source.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.28.1
-// 	protoc        v3.21.6
+// 	protoc        v3.21.5
 // source: internal/pb/source.proto
 
 package pb
@@ -21,6 +21,44 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type ValidateSource struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *ValidateSource) Reset() {
+	*x = ValidateSource{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_source_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateSource) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateSource) ProtoMessage() {}
+
+func (x *ValidateSource) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_source_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateSource.ProtoReflect.Descriptor instead.
+func (*ValidateSource) Descriptor() ([]byte, []int) {
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{0}
+}
+
 type Sync struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -30,7 +68,7 @@ type Sync struct {
 func (x *Sync) Reset() {
 	*x = Sync{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[0]
+		mi := &file_internal_pb_source_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -43,7 +81,7 @@ func (x *Sync) String() string {
 func (*Sync) ProtoMessage() {}
 
 func (x *Sync) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[0]
+	mi := &file_internal_pb_source_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -56,7 +94,7 @@ func (x *Sync) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Sync.ProtoReflect.Descriptor instead.
 func (*Sync) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{0}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{1}
 }
 
 type GetSyncSummary struct {
@@ -68,7 +106,7 @@ type GetSyncSummary struct {
 func (x *GetSyncSummary) Reset() {
 	*x = GetSyncSummary{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[1]
+		mi := &file_internal_pb_source_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -81,7 +119,7 @@ func (x *GetSyncSummary) String() string {
 func (*GetSyncSummary) ProtoMessage() {}
 
 func (x *GetSyncSummary) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[1]
+	mi := &file_internal_pb_source_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -94,7 +132,7 @@ func (x *GetSyncSummary) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetSyncSummary.ProtoReflect.Descriptor instead.
 func (*GetSyncSummary) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{1}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{2}
 }
 
 type GetTables struct {
@@ -106,7 +144,7 @@ type GetTables struct {
 func (x *GetTables) Reset() {
 	*x = GetTables{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[2]
+		mi := &file_internal_pb_source_proto_msgTypes[3]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -119,7 +157,7 @@ func (x *GetTables) String() string {
 func (*GetTables) ProtoMessage() {}
 
 func (x *GetTables) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[2]
+	mi := &file_internal_pb_source_proto_msgTypes[3]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -132,7 +170,109 @@ func (x *GetTables) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetTables.ProtoReflect.Descriptor instead.
 func (*GetTables) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{2}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{3}
+}
+
+type ValidateSource_Request struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Spec []byte `protobuf:"bytes,1,opt,name=spec,proto3" json:"spec,omitempty"`
+}
+
+func (x *ValidateSource_Request) Reset() {
+	*x = ValidateSource_Request{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_source_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateSource_Request) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateSource_Request) ProtoMessage() {}
+
+func (x *ValidateSource_Request) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_source_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateSource_Request.ProtoReflect.Descriptor instead.
+func (*ValidateSource_Request) Descriptor() ([]byte, []int) {
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{0, 0}
+}
+
+func (x *ValidateSource_Request) GetSpec() []byte {
+	if x != nil {
+		return x.Spec
+	}
+	return nil
+}
+
+type ValidateSource_Response struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Warnings []string `protobuf:"bytes,1,rep,name=warnings,proto3" json:"warnings,omitempty"`
+	Errors   []string `protobuf:"bytes,2,rep,name=errors,proto3" json:"errors,omitempty"`
+}
+
+func (x *ValidateSource_Response) Reset() {
+	*x = ValidateSource_Response{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_pb_source_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ValidateSource_Response) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ValidateSource_Response) ProtoMessage() {}
+
+func (x *ValidateSource_Response) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_pb_source_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ValidateSource_Response.ProtoReflect.Descriptor instead.
+func (*ValidateSource_Response) Descriptor() ([]byte, []int) {
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{0, 1}
+}
+
+func (x *ValidateSource_Response) GetWarnings() []string {
+	if x != nil {
+		return x.Warnings
+	}
+	return nil
+}
+
+func (x *ValidateSource_Response) GetErrors() []string {
+	if x != nil {
+		return x.Errors
+	}
+	return nil
 }
 
 type Sync_Request struct {
@@ -148,7 +288,7 @@ type Sync_Request struct {
 func (x *Sync_Request) Reset() {
 	*x = Sync_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[3]
+		mi := &file_internal_pb_source_proto_msgTypes[6]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -161,7 +301,7 @@ func (x *Sync_Request) String() string {
 func (*Sync_Request) ProtoMessage() {}
 
 func (x *Sync_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[3]
+	mi := &file_internal_pb_source_proto_msgTypes[6]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -174,7 +314,7 @@ func (x *Sync_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Sync_Request.ProtoReflect.Descriptor instead.
 func (*Sync_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{0, 0}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{1, 0}
 }
 
 func (x *Sync_Request) GetSpec() []byte {
@@ -203,7 +343,7 @@ type Sync_Response struct {
 func (x *Sync_Response) Reset() {
 	*x = Sync_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[4]
+		mi := &file_internal_pb_source_proto_msgTypes[7]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -216,7 +356,7 @@ func (x *Sync_Response) String() string {
 func (*Sync_Response) ProtoMessage() {}
 
 func (x *Sync_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[4]
+	mi := &file_internal_pb_source_proto_msgTypes[7]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -229,7 +369,7 @@ func (x *Sync_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Sync_Response.ProtoReflect.Descriptor instead.
 func (*Sync_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{0, 1}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{1, 1}
 }
 
 func (x *Sync_Response) GetResource() []byte {
@@ -248,7 +388,7 @@ type GetSyncSummary_Request struct {
 func (x *GetSyncSummary_Request) Reset() {
 	*x = GetSyncSummary_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[5]
+		mi := &file_internal_pb_source_proto_msgTypes[8]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -261,7 +401,7 @@ func (x *GetSyncSummary_Request) String() string {
 func (*GetSyncSummary_Request) ProtoMessage() {}
 
 func (x *GetSyncSummary_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[5]
+	mi := &file_internal_pb_source_proto_msgTypes[8]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -274,7 +414,7 @@ func (x *GetSyncSummary_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetSyncSummary_Request.ProtoReflect.Descriptor instead.
 func (*GetSyncSummary_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{1, 0}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{2, 0}
 }
 
 type GetSyncSummary_Response struct {
@@ -289,7 +429,7 @@ type GetSyncSummary_Response struct {
 func (x *GetSyncSummary_Response) Reset() {
 	*x = GetSyncSummary_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[6]
+		mi := &file_internal_pb_source_proto_msgTypes[9]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -302,7 +442,7 @@ func (x *GetSyncSummary_Response) String() string {
 func (*GetSyncSummary_Response) ProtoMessage() {}
 
 func (x *GetSyncSummary_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[6]
+	mi := &file_internal_pb_source_proto_msgTypes[9]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -315,7 +455,7 @@ func (x *GetSyncSummary_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetSyncSummary_Response.ProtoReflect.Descriptor instead.
 func (*GetSyncSummary_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{1, 1}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{2, 1}
 }
 
 func (x *GetSyncSummary_Response) GetSummary() []byte {
@@ -334,7 +474,7 @@ type GetTables_Request struct {
 func (x *GetTables_Request) Reset() {
 	*x = GetTables_Request{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[7]
+		mi := &file_internal_pb_source_proto_msgTypes[10]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -347,7 +487,7 @@ func (x *GetTables_Request) String() string {
 func (*GetTables_Request) ProtoMessage() {}
 
 func (x *GetTables_Request) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[7]
+	mi := &file_internal_pb_source_proto_msgTypes[10]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -360,7 +500,7 @@ func (x *GetTables_Request) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetTables_Request.ProtoReflect.Descriptor instead.
 func (*GetTables_Request) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{2, 0}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{3, 0}
 }
 
 type GetTables_Response struct {
@@ -377,7 +517,7 @@ type GetTables_Response struct {
 func (x *GetTables_Response) Reset() {
 	*x = GetTables_Response{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_internal_pb_source_proto_msgTypes[8]
+		mi := &file_internal_pb_source_proto_msgTypes[11]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -390,7 +530,7 @@ func (x *GetTables_Response) String() string {
 func (*GetTables_Response) ProtoMessage() {}
 
 func (x *GetTables_Response) ProtoReflect() protoreflect.Message {
-	mi := &file_internal_pb_source_proto_msgTypes[8]
+	mi := &file_internal_pb_source_proto_msgTypes[11]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -403,7 +543,7 @@ func (x *GetTables_Response) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use GetTables_Response.ProtoReflect.Descriptor instead.
 func (*GetTables_Response) Descriptor() ([]byte, []int) {
-	return file_internal_pb_source_proto_rawDescGZIP(), []int{2, 1}
+	return file_internal_pb_source_proto_rawDescGZIP(), []int{3, 1}
 }
 
 func (x *GetTables_Response) GetName() string {
@@ -435,49 +575,60 @@ var file_internal_pb_source_proto_rawDesc = []byte{
 	0x6f, 0x1a, 0x16, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x62, 0x2f, 0x62,
 	0x61, 0x73, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
 	0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73,
-	0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x87, 0x01, 0x0a, 0x04, 0x53,
-	0x79, 0x6e, 0x63, 0x1a, 0x57, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x12,
-	0x0a, 0x04, 0x73, 0x70, 0x65, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x73, 0x70,
-	0x65, 0x63, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
-	0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x1a, 0x26, 0x0a, 0x08,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x22, 0x41, 0x0a, 0x0e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63, 0x53,
-	0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x24, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x18, 0x0a,
-	0x07, 0x73, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07,
-	0x73, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x22, 0x68, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x54, 0x61,
-	0x62, 0x6c, 0x65, 0x73, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x50, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e,
-	0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12,
-	0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x61, 0x62,
-	0x6c, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x74, 0x61, 0x62, 0x6c, 0x65,
-	0x73, 0x32, 0xd1, 0x02, 0x0a, 0x06, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x3a, 0x0a, 0x07,
-	0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e,
-	0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e,
-	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a, 0x0a, 0x47, 0x65, 0x74, 0x56,
-	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47,
-	0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73,
-	0x74, 0x1a, 0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72,
-	0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x40, 0x0a,
-	0x09, 0x47, 0x65, 0x74, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x12, 0x18, 0x2e, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x2e, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74,
-	0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12,
-	0x4f, 0x0a, 0x0e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72,
-	0x79, 0x12, 0x1d, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e,
-	0x63, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x1a, 0x1e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63,
-	0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x12, 0x33, 0x0a, 0x04, 0x53, 0x79, 0x6e, 0x63, 0x12, 0x13, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
-	0x2e, 0x53, 0x79, 0x6e, 0x63, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e,
-	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x53, 0x79, 0x6e, 0x63, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f,
-	0x6e, 0x73, 0x65, 0x30, 0x01, 0x42, 0x05, 0x5a, 0x03, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x6f, 0x0a, 0x0e, 0x56, 0x61,
+	0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x1a, 0x1d, 0x0a, 0x07,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x73, 0x70, 0x65, 0x63, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x73, 0x70, 0x65, 0x63, 0x1a, 0x3e, 0x0a, 0x08, 0x52,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x77, 0x61, 0x72, 0x6e, 0x69,
+	0x6e, 0x67, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x77, 0x61, 0x72, 0x6e, 0x69,
+	0x6e, 0x67, 0x73, 0x12, 0x16, 0x0a, 0x06, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x73, 0x18, 0x02, 0x20,
+	0x03, 0x28, 0x09, 0x52, 0x06, 0x65, 0x72, 0x72, 0x6f, 0x72, 0x73, 0x22, 0x87, 0x01, 0x0a, 0x04,
+	0x53, 0x79, 0x6e, 0x63, 0x1a, 0x57, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
+	0x12, 0x0a, 0x04, 0x73, 0x70, 0x65, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x73,
+	0x70, 0x65, 0x63, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
+	0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x1a, 0x26, 0x0a,
+	0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x22, 0x41, 0x0a, 0x0e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63,
+	0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x1a, 0x24, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x18,
+	0x0a, 0x07, 0x73, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52,
+	0x07, 0x73, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x22, 0x68, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x54,
+	0x61, 0x62, 0x6c, 0x65, 0x73, 0x1a, 0x09, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x50, 0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
+	0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x61,
+	0x62, 0x6c, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x74, 0x61, 0x62, 0x6c,
+	0x65, 0x73, 0x32, 0x9c, 0x03, 0x0a, 0x06, 0x53, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x3a, 0x0a,
+	0x07, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x16, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x2e, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
+	0x1a, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x4e, 0x61, 0x6d, 0x65,
+	0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x43, 0x0a, 0x0a, 0x47, 0x65, 0x74,
+	0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e,
+	0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65,
+	0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65,
+	0x72, 0x73, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x40,
+	0x0a, 0x09, 0x47, 0x65, 0x74, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x12, 0x18, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x2e, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x19, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65,
+	0x74, 0x54, 0x61, 0x62, 0x6c, 0x65, 0x73, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x12, 0x49, 0x0a, 0x08, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x12, 0x1d, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x53, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x2e, 0x56, 0x61, 0x6c, 0x69, 0x64, 0x61, 0x74, 0x65, 0x53, 0x6f, 0x75, 0x72,
+	0x63, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x4f, 0x0a, 0x0e, 0x47,
+	0x65, 0x74, 0x53, 0x79, 0x6e, 0x63, 0x53, 0x75, 0x6d, 0x6d, 0x61, 0x72, 0x79, 0x12, 0x1d, 0x2e,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63, 0x53, 0x75, 0x6d,
+	0x6d, 0x61, 0x72, 0x79, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x79, 0x6e, 0x63, 0x53, 0x75, 0x6d, 0x6d,
+	0x61, 0x72, 0x79, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x33, 0x0a, 0x04,
+	0x53, 0x79, 0x6e, 0x63, 0x12, 0x13, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x53, 0x79, 0x6e,
+	0x63, 0x2e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x2e, 0x53, 0x79, 0x6e, 0x63, 0x2e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x30,
+	0x01, 0x42, 0x05, 0x5a, 0x03, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -492,37 +643,42 @@ func file_internal_pb_source_proto_rawDescGZIP() []byte {
 	return file_internal_pb_source_proto_rawDescData
 }
 
-var file_internal_pb_source_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
+var file_internal_pb_source_proto_msgTypes = make([]protoimpl.MessageInfo, 12)
 var file_internal_pb_source_proto_goTypes = []interface{}{
-	(*Sync)(nil),                    // 0: proto.Sync
-	(*GetSyncSummary)(nil),          // 1: proto.GetSyncSummary
-	(*GetTables)(nil),               // 2: proto.GetTables
-	(*Sync_Request)(nil),            // 3: proto.Sync.Request
-	(*Sync_Response)(nil),           // 4: proto.Sync.Response
-	(*GetSyncSummary_Request)(nil),  // 5: proto.GetSyncSummary.Request
-	(*GetSyncSummary_Response)(nil), // 6: proto.GetSyncSummary.Response
-	(*GetTables_Request)(nil),       // 7: proto.GetTables.Request
-	(*GetTables_Response)(nil),      // 8: proto.GetTables.Response
-	(*timestamppb.Timestamp)(nil),   // 9: google.protobuf.Timestamp
-	(*GetName_Request)(nil),         // 10: proto.GetName.Request
-	(*GetVersion_Request)(nil),      // 11: proto.GetVersion.Request
-	(*GetName_Response)(nil),        // 12: proto.GetName.Response
-	(*GetVersion_Response)(nil),     // 13: proto.GetVersion.Response
+	(*ValidateSource)(nil),          // 0: proto.ValidateSource
+	(*Sync)(nil),                    // 1: proto.Sync
+	(*GetSyncSummary)(nil),          // 2: proto.GetSyncSummary
+	(*GetTables)(nil),               // 3: proto.GetTables
+	(*ValidateSource_Request)(nil),  // 4: proto.ValidateSource.Request
+	(*ValidateSource_Response)(nil), // 5: proto.ValidateSource.Response
+	(*Sync_Request)(nil),            // 6: proto.Sync.Request
+	(*Sync_Response)(nil),           // 7: proto.Sync.Response
+	(*GetSyncSummary_Request)(nil),  // 8: proto.GetSyncSummary.Request
+	(*GetSyncSummary_Response)(nil), // 9: proto.GetSyncSummary.Response
+	(*GetTables_Request)(nil),       // 10: proto.GetTables.Request
+	(*GetTables_Response)(nil),      // 11: proto.GetTables.Response
+	(*timestamppb.Timestamp)(nil),   // 12: google.protobuf.Timestamp
+	(*GetName_Request)(nil),         // 13: proto.GetName.Request
+	(*GetVersion_Request)(nil),      // 14: proto.GetVersion.Request
+	(*GetName_Response)(nil),        // 15: proto.GetName.Response
+	(*GetVersion_Response)(nil),     // 16: proto.GetVersion.Response
 }
 var file_internal_pb_source_proto_depIdxs = []int32{
-	9,  // 0: proto.Sync.Request.timestamp:type_name -> google.protobuf.Timestamp
-	10, // 1: proto.Source.GetName:input_type -> proto.GetName.Request
-	11, // 2: proto.Source.GetVersion:input_type -> proto.GetVersion.Request
-	7,  // 3: proto.Source.GetTables:input_type -> proto.GetTables.Request
-	5,  // 4: proto.Source.GetSyncSummary:input_type -> proto.GetSyncSummary.Request
-	3,  // 5: proto.Source.Sync:input_type -> proto.Sync.Request
-	12, // 6: proto.Source.GetName:output_type -> proto.GetName.Response
-	13, // 7: proto.Source.GetVersion:output_type -> proto.GetVersion.Response
-	8,  // 8: proto.Source.GetTables:output_type -> proto.GetTables.Response
-	6,  // 9: proto.Source.GetSyncSummary:output_type -> proto.GetSyncSummary.Response
-	4,  // 10: proto.Source.Sync:output_type -> proto.Sync.Response
-	6,  // [6:11] is the sub-list for method output_type
-	1,  // [1:6] is the sub-list for method input_type
+	12, // 0: proto.Sync.Request.timestamp:type_name -> google.protobuf.Timestamp
+	13, // 1: proto.Source.GetName:input_type -> proto.GetName.Request
+	14, // 2: proto.Source.GetVersion:input_type -> proto.GetVersion.Request
+	10, // 3: proto.Source.GetTables:input_type -> proto.GetTables.Request
+	4,  // 4: proto.Source.Validate:input_type -> proto.ValidateSource.Request
+	8,  // 5: proto.Source.GetSyncSummary:input_type -> proto.GetSyncSummary.Request
+	6,  // 6: proto.Source.Sync:input_type -> proto.Sync.Request
+	15, // 7: proto.Source.GetName:output_type -> proto.GetName.Response
+	16, // 8: proto.Source.GetVersion:output_type -> proto.GetVersion.Response
+	11, // 9: proto.Source.GetTables:output_type -> proto.GetTables.Response
+	5,  // 10: proto.Source.Validate:output_type -> proto.ValidateSource.Response
+	9,  // 11: proto.Source.GetSyncSummary:output_type -> proto.GetSyncSummary.Response
+	7,  // 12: proto.Source.Sync:output_type -> proto.Sync.Response
+	7,  // [7:13] is the sub-list for method output_type
+	1,  // [1:7] is the sub-list for method input_type
 	1,  // [1:1] is the sub-list for extension type_name
 	1,  // [1:1] is the sub-list for extension extendee
 	0,  // [0:1] is the sub-list for field type_name
@@ -536,7 +692,7 @@ func file_internal_pb_source_proto_init() {
 	file_internal_pb_base_proto_init()
 	if !protoimpl.UnsafeEnabled {
 		file_internal_pb_source_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Sync); i {
+			switch v := v.(*ValidateSource); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -548,7 +704,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetSyncSummary); i {
+			switch v := v.(*Sync); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -560,7 +716,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetTables); i {
+			switch v := v.(*GetSyncSummary); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -572,7 +728,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Sync_Request); i {
+			switch v := v.(*GetTables); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -584,7 +740,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Sync_Response); i {
+			switch v := v.(*ValidateSource_Request); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -596,7 +752,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetSyncSummary_Request); i {
+			switch v := v.(*ValidateSource_Response); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -608,7 +764,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetSyncSummary_Response); i {
+			switch v := v.(*Sync_Request); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -620,7 +776,7 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*GetTables_Request); i {
+			switch v := v.(*Sync_Response); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -632,6 +788,42 @@ func file_internal_pb_source_proto_init() {
 			}
 		}
 		file_internal_pb_source_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GetSyncSummary_Request); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_source_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GetSyncSummary_Response); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_source_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*GetTables_Request); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_internal_pb_source_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*GetTables_Response); i {
 			case 0:
 				return &v.state
@@ -650,7 +842,7 @@ func file_internal_pb_source_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_internal_pb_source_proto_rawDesc,
 			NumEnums:      0,
-			NumMessages:   9,
+			NumMessages:   12,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/internal/pb/source.proto b/internal/pb/source.proto
index 658231ec97..28288d115d 100644
--- a/internal/pb/source.proto
+++ b/internal/pb/source.proto
@@ -11,6 +11,8 @@ service Source {
   rpc GetVersion(GetVersion.Request) returns (GetVersion.Response);
   // Get all tables the source plugin supports
   rpc GetTables(GetTables.Request) returns (GetTables.Response);
+  // Validate returns warnings and errors about the spec.
+  rpc Validate(ValidateSource.Request) returns (ValidateSource.Response);
   // GetSyncSummary returns the latest sync summary of the source plugin. we don't want to send the summary on
   // every sync request.
   rpc GetSyncSummary(GetSyncSummary.Request) returns (GetSyncSummary.Response);
@@ -18,6 +20,16 @@ service Source {
   rpc Sync(Sync.Request) returns (stream Sync.Response);
 }
 
+message ValidateSource {
+  message Request {
+    bytes spec = 1;
+  }
+  message Response {
+    repeated string warnings = 1;
+    repeated string errors = 2;
+  }
+}
+
 message Sync {
   message Request {
     bytes spec = 1;
diff --git a/internal/pb/source_grpc.pb.go b/internal/pb/source_grpc.pb.go
index 72bddd378b..3978e800cd 100644
--- a/internal/pb/source_grpc.pb.go
+++ b/internal/pb/source_grpc.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go-grpc. DO NOT EDIT.
 // versions:
 // - protoc-gen-go-grpc v1.2.0
-// - protoc             v3.21.6
+// - protoc             v3.21.5
 // source: internal/pb/source.proto
 
 package pb
@@ -28,6 +28,8 @@ type SourceClient interface {
 	GetVersion(ctx context.Context, in *GetVersion_Request, opts ...grpc.CallOption) (*GetVersion_Response, error)
 	// Get all tables the source plugin supports
 	GetTables(ctx context.Context, in *GetTables_Request, opts ...grpc.CallOption) (*GetTables_Response, error)
+	// Validate returns warnings and errors about the spec.
+	Validate(ctx context.Context, in *ValidateSource_Request, opts ...grpc.CallOption) (*ValidateSource_Response, error)
 	// GetSyncSummary returns the latest sync summary of the source plugin. we don't want to send the summary on
 	// every sync request.
 	GetSyncSummary(ctx context.Context, in *GetSyncSummary_Request, opts ...grpc.CallOption) (*GetSyncSummary_Response, error)
@@ -70,6 +72,15 @@ func (c *sourceClient) GetTables(ctx context.Context, in *GetTables_Request, opt
 	return out, nil
 }
 
+func (c *sourceClient) Validate(ctx context.Context, in *ValidateSource_Request, opts ...grpc.CallOption) (*ValidateSource_Response, error) {
+	out := new(ValidateSource_Response)
+	err := c.cc.Invoke(ctx, "/proto.Source/Validate", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *sourceClient) GetSyncSummary(ctx context.Context, in *GetSyncSummary_Request, opts ...grpc.CallOption) (*GetSyncSummary_Response, error) {
 	out := new(GetSyncSummary_Response)
 	err := c.cc.Invoke(ctx, "/proto.Source/GetSyncSummary", in, out, opts...)
@@ -121,6 +132,8 @@ type SourceServer interface {
 	GetVersion(context.Context, *GetVersion_Request) (*GetVersion_Response, error)
 	// Get all tables the source plugin supports
 	GetTables(context.Context, *GetTables_Request) (*GetTables_Response, error)
+	// Validate returns warnings and errors about the spec.
+	Validate(context.Context, *ValidateSource_Request) (*ValidateSource_Response, error)
 	// GetSyncSummary returns the latest sync summary of the source plugin. we don't want to send the summary on
 	// every sync request.
 	GetSyncSummary(context.Context, *GetSyncSummary_Request) (*GetSyncSummary_Response, error)
@@ -142,6 +155,9 @@ func (UnimplementedSourceServer) GetVersion(context.Context, *GetVersion_Request
 func (UnimplementedSourceServer) GetTables(context.Context, *GetTables_Request) (*GetTables_Response, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetTables not implemented")
 }
+func (UnimplementedSourceServer) Validate(context.Context, *ValidateSource_Request) (*ValidateSource_Response, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Validate not implemented")
+}
 func (UnimplementedSourceServer) GetSyncSummary(context.Context, *GetSyncSummary_Request) (*GetSyncSummary_Response, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetSyncSummary not implemented")
 }
@@ -215,6 +231,24 @@ func _Source_GetTables_Handler(srv interface{}, ctx context.Context, dec func(in
 	return interceptor(ctx, in, info, handler)
 }
 
+func _Source_Validate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ValidateSource_Request)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SourceServer).Validate(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/proto.Source/Validate",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SourceServer).Validate(ctx, req.(*ValidateSource_Request))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _Source_GetSyncSummary_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(GetSyncSummary_Request)
 	if err := dec(in); err != nil {
@@ -273,6 +307,10 @@ var Source_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetTables",
 			Handler:    _Source_GetTables_Handler,
 		},
+		{
+			MethodName: "Validate",
+			Handler:    _Source_Validate_Handler,
+		},
 		{
 			MethodName: "GetSyncSummary",
 			Handler:    _Source_GetSyncSummary_Handler,
diff --git a/internal/servers/destinations.go b/internal/servers/destinations.go
index ebccdfdd90..2b32182154 100644
--- a/internal/servers/destinations.go
+++ b/internal/servers/destinations.go
@@ -1,6 +1,7 @@
 package servers
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -42,6 +43,18 @@ func (s *DestinationServer) GetVersion(context.Context, *pb.GetVersion_Request)
 	}, nil
 }
 
+func (s *DestinationServer) Validate(ctx context.Context, req *pb.ValidateDestination_Request) (*pb.ValidateDestination_Response, error) {
+	spec, err := decodeDestinationSpec(req.Spec)
+	if err != nil {
+		return nil, err
+	}
+	warns, errs := s.Plugin.Validate(spec)
+	return &pb.ValidateDestination_Response{
+		Warnings: warns,
+		Errors:   errs,
+	}, nil
+}
+
 func (s *DestinationServer) Migrate(ctx context.Context, req *pb.Migrate_Request) (*pb.Migrate_Response, error) {
 	var tables []*schema.Table
 	if err := json.Unmarshal(req.Tables, &tables); err != nil {
@@ -130,3 +143,14 @@ func (s *DestinationServer) DeleteStale(ctx context.Context, req *pb.DeleteStale
 func (s *DestinationServer) Close(ctx context.Context, _ *pb.Close_Request) (*pb.Close_Response, error) {
 	return &pb.Close_Response{}, s.Plugin.Close(ctx)
 }
+
+func decodeDestinationSpec(b []byte) (specs.Destination, error) {
+	var spec specs.Destination
+	dec := json.NewDecoder(bytes.NewReader(b))
+	dec.UseNumber()
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&spec); err != nil {
+		return specs.Destination{}, status.Errorf(codes.InvalidArgument, "failed to decode spec: %v", err)
+	}
+	return spec, nil
+}
diff --git a/internal/servers/source.go b/internal/servers/source.go
index 2aec192ab4..f3b657211f 100644
--- a/internal/servers/source.go
+++ b/internal/servers/source.go
@@ -54,18 +54,27 @@ func (s *SourceServer) GetSyncSummary(context.Context, *pb.GetSyncSummary_Reques
 	}, nil
 }
 
+func (s *SourceServer) Validate(ctx context.Context, req *pb.ValidateSource_Request) (*pb.ValidateSource_Response, error) {
+	spec, err := decodeSourceSpec(req.Spec)
+	if err != nil {
+		return nil, err
+	}
+	warns, errs := s.Plugin.Validate(spec)
+	return &pb.ValidateSource_Response{
+		Warnings: warns,
+		Errors:   errs,
+	}, nil
+}
+
 func (s *SourceServer) Sync(req *pb.Sync_Request, stream pb.Source_SyncServer) error {
+	spec, err := decodeSourceSpec(req.Spec)
+	if err != nil {
+		return err
+	}
+
 	resources := make(chan *schema.Resource)
 	var syncErr error
 
-	var spec specs.Source
-	dec := json.NewDecoder(bytes.NewReader(req.Spec))
-	dec.UseNumber()
-	dec.DisallowUnknownFields()
-	if err := dec.Decode(&spec); err != nil {
-		return status.Errorf(codes.InvalidArgument, "failed to decode spec: %v", err)
-	}
-
 	go func() {
 		defer close(resources)
 		var err error
@@ -92,3 +101,14 @@ func (s *SourceServer) Sync(req *pb.Sync_Request, stream pb.Source_SyncServer) e
 
 	return nil
 }
+
+func decodeSourceSpec(b []byte) (specs.Source, error) {
+	var spec specs.Source
+	dec := json.NewDecoder(bytes.NewReader(b))
+	dec.UseNumber()
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&spec); err != nil {
+		return specs.Source{}, status.Errorf(codes.InvalidArgument, "failed to decode spec: %v", err)
+	}
+	return spec, nil
+}
diff --git a/plugins/destination.go b/plugins/destination.go
index c076ffefab..c4e70d6ef4 100644
--- a/plugins/destination.go
+++ b/plugins/destination.go
@@ -71,6 +71,13 @@ func (p *DestinationPlugin) Init(ctx context.Context, logger zerolog.Logger, spe
 	return nil
 }
 
+// Validate checks a destination spec for config issues. It is not currently used, but is
+// reserved for future use.
+func (p *DestinationPlugin) Validate(spec specs.Destination) (warnings, errors []string) {
+	spec.Validate()
+	return warnings, errors
+}
+
 // we implement all DestinationClient functions so we can hook into pre-post behavior
 func (p *DestinationPlugin) Migrate(ctx context.Context, tables schema.Tables) error {
 	SetDestinationManagedCqColumns(tables)
diff --git a/plugins/source.go b/plugins/source.go
index 5d50e9029f..1890c99a50 100644
--- a/plugins/source.go
+++ b/plugins/source.go
@@ -100,6 +100,20 @@ func (p *SourcePlugin) Version() string {
 	return p.version
 }
 
+// Validate checks a source spec for config issues.
+func (p *SourcePlugin) Validate(spec specs.Source) (warnings, errors []string) {
+	warnings = spec.Warnings()
+
+	if err := spec.Validate(); err != nil {
+		errors = append(errors, err.Error())
+	}
+	_, err := p.listAndValidateTables(spec.Tables, spec.SkipTables)
+	if err != nil {
+		errors = append(errors, err.Error())
+	}
+	return warnings, errors
+}
+
 // Sync is syncing data from the requested tables in spec to the given channel
 func (p *SourcePlugin) Sync(ctx context.Context, logger zerolog.Logger, spec specs.Source, res chan<- *schema.Resource) (*schema.SyncSummary, error) {
 	spec.SetDefaults()
diff --git a/specs/destination.go b/specs/destination.go
index 20b01072ea..bcf40e5439 100644
--- a/specs/destination.go
+++ b/specs/destination.go
@@ -51,6 +51,12 @@ func (d *Destination) UnmarshalSpec(out interface{}) error {
 	return dec.Decode(out)
 }
 
+// Warnings returns non-critical validation errors, such as deprecation notices.
+func (d *Destination) Warnings() (warnings []string) {
+	// add future deprecation notices here
+	return warnings
+}
+
 func (d *Destination) Validate() error {
 	if d.Name == "" {
 		return fmt.Errorf("name is required")
diff --git a/specs/source.go b/specs/source.go
index 0aaf93f1b2..9794777c5f 100644
--- a/specs/source.go
+++ b/specs/source.go
@@ -73,6 +73,18 @@ func (s *Source) UnmarshalSpec(out interface{}) error {
 	return dec.Decode(out)
 }
 
+// Warnings returns non-critical validation errors, such as deprecation notices.
+func (s *Source) Warnings() (warnings []string) {
+	if s.Concurrency != 0 {
+		warnings = append(warnings, `"concurrency" is deprecated and will be removed in a future version: use "table_concurrency" and "resource_concurrency" instead.`)
+	}
+	if s.Path == "" {
+		warnings = append(warnings, `"path" will become a required parameter in a future version`)
+	}
+	return warnings
+}
+
+// Validate returns critical validation errors that will prevent syncs from running.
 func (s *Source) Validate() error {
 	if s.Name == "" {
 		return fmt.Errorf("name is required")