Implement source ingest filter (#439)

* Implement source ingest filter - this allows us to reject rows from the source based on a WHERE clause

* Fixed some issues with error propagation and a race in schema deletion

Author: purplefox

.github/workflows/ci.yml

@@ -32,4 +32,4 @@ jobs:
       - name: Proto lint
         run: make -C protos lint
       - name: Test
-        run: go test -v -race -count=1 -timeout 1h -failfast -p 1 -tags integration ./...
+        run: go test -race -count=1 -timeout 1h -failfast -p 1 -tags integration ./...

api/server.go

@@ -7,7 +7,6 @@ import (
 	"github.com/squareup/pranadb/pull/exec"
 	"net"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/squareup/pranadb/meta"
@@ -157,13 +156,7 @@ func (s *Server) doExecuteStatement(executor exec.PullExecutor, batchSize int, e
 		if errors.As(err, &perr) {
 			return perr
 		}
-		// For internal errors we don't return internal error messages to the CLI as this would leak
-		// server implementation details. Instead, we generate a sequence number and add that to the message
-		// and log the internal error in the server logs with the sequence number so it can be looked up
-		seq := atomic.AddInt64(&s.errorSequence, 1)
-		perr = errors.NewInternalError(seq)
-		log.Errorf("internal error occurred with sequence number %d\n%v", seq, err)
-		return perr
+		return common.LogInternalError(err)
 	}
 
 	// First send column definitions.

cfg/example.conf

@@ -58,6 +58,6 @@ remoting-heartbeat-interval       = "10s" // Amount of time between remoting hea
 remoting-heartbeat-timeout        = "5s" // Timeout for a remoting heartbeat
 enable-api-server                 = true // Set to true to enable the API server - needed for CLI access
 global-ingest-limit-rows-per-sec  = 1000 // The maximum number of rows per second that can be ingested in the broker - ingest will be throttled to this rate. -1 represents no throttling
-raft-rtt-ms                       = 100 // The size of a Raft RTT unit in ms
+raft-rtt-ms                       = 50 // The size of a Raft RTT unit in ms
 raft-heartbeat-rtt                = 30 // The Raft heartbeat period in units of raft-rtt-ms
 raft-election-rtt                 = 300 // The Raft election period in units of raft-rtt-ms

command/create_source_command.go

@@ -222,6 +222,7 @@ func (c *CreateSourceCommand) getSourceInfo(ast *parser.CreateSource) (*common.S
 
 	var (
 		headerEncoding, keyEncoding, valueEncoding common.KafkaEncoding
+		ingestFilter                               string
 		propsMap                                   map[string]string
 		colSelectors                               []selector.ColumnSelector
 		brokerName, topicName                      string
@@ -243,6 +244,8 @@ func (c *CreateSourceCommand) getSourceInfo(ast *parser.CreateSource) (*common.S
 			if valueEncoding.Encoding == common.EncodingUnknown {
 				return nil, errors.NewPranaErrorf(errors.UnknownTopicEncoding, "Unknown topic encoding %s", opt.ValueEncoding)
 			}
+		case opt.IngestFilter != "":
+			ingestFilter = opt.IngestFilter
 		case opt.Properties != nil:
 			propsMap = make(map[string]string, len(opt.Properties))
 			for _, prop := range opt.Properties {
@@ -287,6 +290,7 @@ func (c *CreateSourceCommand) getSourceInfo(ast *parser.CreateSource) (*common.S
 		HeaderEncoding: headerEncoding,
 		KeyEncoding:    keyEncoding,
 		ValueEncoding:  valueEncoding,
+		IngestFilter:   ingestFilter,
 		ColSelectors:   colSelectors,
 		Properties:     propsMap,
 	}

command/ddl_runner.go

@@ -117,7 +117,7 @@ func (d *DDLCommandRunner) HandleNotification(notification remoting.ClusterMessa
 		return errors.Errorf("cannot find command with id %d:%d", ddlInfo.GetOriginatingNodeId(), ddlInfo.GetCommandId())
 	}
 	err := com.OnPhase(phase)
-	if phase == int32(com.NumPhases()-1) {
+	if phase == int32(com.NumPhases()-1) || err != nil {
 		// Final phase so delete the command
 		delete(d.commands, skey)
 	}

command/parser/ast.go

@@ -105,6 +105,7 @@ type TopicInformation struct {
 	HeaderEncoding string                        `|"HeaderEncoding" "=" @String`
 	KeyEncoding    string                        `|"KeyEncoding" "=" @String`
 	ValueEncoding  string                        `|"ValueEncoding" "=" @String`
+	IngestFilter   string                        `|"IngestFilter" "=" @String`
 	ColSelectors   []*selector.ColumnSelectorAST `|"ColumnSelectors" "=" "(" (@@ ("," @@)*)? ")"`
 	Properties     []*TopicInfoProperty          `|"Properties" "=" "(" (@@ ("," @@)*)? ")"`
 }

command/parser/ast_test.go

@@ -50,6 +50,7 @@ func TestParse(t *testing.T) {
 			headerencoding = "json",
 			keyencoding = "json",
 			valueencoding = "json",
+            ingestfilter = "where sensor_id=1",
 			columnselectors = (
 				meta("key").k0,
 				v1,
@@ -75,6 +76,7 @@ func TestParse(t *testing.T) {
 					{HeaderEncoding: "json"},
 					{KeyEncoding: "json"},
 					{ValueEncoding: "json"},
+					{IngestFilter: "where sensor_id=1"},
 					{ColSelectors: []*selector.ColumnSelectorAST{
 						{MetaKey: stringRef("key"), Next: &selector.SelectorAST{Field: "k0"}},
 						{Field: stringRef("v1")},

common/internal_error.go

@@ -0,0 +1,24 @@
+package common
+
+import (
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+	"github.com/squareup/pranadb/errors"
+)
+
+func LogInternalError(err error) errors.PranaError {
+	id, err := uuid.NewRandom()
+	var errRef string
+	if err != nil {
+		log.Errorf("failed to generate uuid %v", err)
+		errRef = ""
+	} else {
+		errRef = id.String()
+	}
+	// For internal errors we don't return internal error messages to the CLI as this would leak
+	// server implementation details. Instead, we generate a random UUID and add that to the message
+	// and log the internal error in the server logs with the UUID so it can be looked up
+	perr := errors.NewInternalError(errRef)
+	log.Errorf("internal error occurred with reference %s\n%v", errRef, err)
+	return perr
+}

common/meta.go

@@ -186,11 +186,10 @@ func (i *IndexInfo) ContainsColIndex(colIndex int) bool {
 
 type Schema struct {
 	// Schema can be mutated from different goroutines so we need to lock to protect access to it's maps
-	lock    sync.RWMutex
-	Name    string
-	tables  map[string]Table
-	sinks   map[string]*SinkInfo
-	deleted bool
+	lock   sync.RWMutex
+	Name   string
+	tables map[string]Table
+	sinks  map[string]*SinkInfo
 }
 
 func NewSchema(name string) *Schema {
@@ -280,20 +279,6 @@ func (s *Schema) Equal(other *Schema) bool {
 	return reflect.DeepEqual(s, other)
 }
 
-func (s *Schema) SetDeleted() {
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	s.deleted = true
-}
-
-func (s *Schema) IsDeleted() bool {
-	// Schema cna become deleted if all tables are removed, but sessions might still have it cached
-	// checking isDeleted() allows sessions to refresh the schema if necessary
-	s.lock.RLock()
-	defer s.lock.RUnlock()
-	return s.deleted
-}
-
 type SourceInfo struct {
 	*TableInfo
 	TopicInfo *TopicInfo
@@ -324,6 +309,7 @@ type TopicInfo struct {
 	HeaderEncoding KafkaEncoding
 	ColSelectors   []selector.ColumnSelector
 	Properties     map[string]string
+	IngestFilter   string
 }
 
 type KafkaEncoding struct {

conf/conf.go

@@ -17,7 +17,7 @@ const (
 	DefaultRemotingHeartbeatInterval   = 10 * time.Second
 	DefaultRemotingHeartbeatTimeout    = 5 * time.Second
 	DefaultGlobalIngestLimitRowsPerSec = 1000
-	DefaultRaftRTTMs                   = 100
+	DefaultRaftRTTMs                   = 50
 	DefaultRaftHeartbeatRTT            = 30
 	DefaultRaftElectionRTT             = 300
 )