chore: clean up unused code (numaproj#370)

dseapy · Nov 19, 2022 · c3739cc · c3739cc
1 parent 7e041d8
commit c3739cc
Show file tree

Hide file tree

Showing 6 changed files with 5 additions and 140 deletions.
diff --git a/docs/APIs.md b/docs/APIs.md
@@ -3448,15 +3448,6 @@ state.
 </tr>
 </tbody>
 </table>
-<h3 id="numaflow.numaproj.io/v1alpha1.StoreType">
-StoreType (<code>string</code> alias)
-</p>
-</h3>
-<p>
-<p>
-StoreType is the PBQ store’s backend type.
-</p>
-</p>
 <h3 id="numaflow.numaproj.io/v1alpha1.TLS">
 TLS
 </h3>
@@ -4134,12 +4125,6 @@ Description
 </tr>
 </tbody>
 </table>
-<h3 id="numaflow.numaproj.io/v1alpha1.WindowType">
-WindowType (<code>string</code> alias)
-</p>
-</h3>
-<p>
-</p>
 <hr/>
 <p>
 <em> Generated with <code>gen-crd-api-reference-docs</code>. </em>

diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
@@ -130,64 +130,15 @@ const (
 	DefaultPBQReadTimeout       = 1 * time.Second // Default read timeout for pbq
 	DefaultPBQReadBatchSize     = 100             // Default read batch size for pbq
 
+	// PVC mount path for PBQ
+	PathPBQMount = "/var/numaflow/pbq"
+
 	// Default persistent store options
 	DefaultStoreSyncDuration  = 2 * time.Second        // Default sync duration for pbq
-	DefaultStoreType          = NoOpType               // Default store type
-	DefaultStoreSize          = 1000000                // Default persistent store size
 	DefaultStoreMaxBufferSize = 100000                 // Default buffer size for pbq in bytes
 	DefaultStorePath          = PathPBQMount + "/wals" // Default store path
-
-	// Default window options
-	DefaultWindowType     = FixedType
-	DefaultWindowDuration = 0
-
-	// PVC mount path for PBQ
-	PathPBQMount = "/var/numaflow/pbq"
 )
 
-// StoreType is the PBQ store's backend type.
-type StoreType string
-
-const (
-	InMemoryType   StoreType = "in-memory"
-	FileSystemType StoreType = "file-system"
-	NoOpType       StoreType = "no-op"
-)
-
-func (st StoreType) String() string {
-	switch st {
-	case InMemoryType:
-		return string(InMemoryType)
-	case FileSystemType:
-		return string(FileSystemType)
-	case NoOpType:
-		return string(NoOpType)
-	default:
-		return "unknownStoreType"
-	}
-}
-
-type WindowType string
-
-const (
-	FixedType   WindowType = "fixed"
-	SlidingType WindowType = "sliding"
-	SessionType WindowType = "session"
-)
-
-func (wt WindowType) String() string {
-	switch wt {
-	case FixedType:
-		return string(FixedType)
-	case SlidingType:
-		return string(SlidingType)
-	case SessionType:
-		return string(SessionType)
-	default:
-		return "unknownWindowType"
-	}
-}
-
 var (
 	MessageKeyDrop = fmt.Sprintf("%U__DROP__", '\\') // U+005C__DROP__
 	MessageKeyAll  = fmt.Sprintf("%U__ALL__", '\\')  // U+005C__ALL__

diff --git a/pkg/reduce/options.go b/pkg/reduce/options.go
@@ -18,23 +18,19 @@ package reduce
 
 import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
-	"github.com/numaproj/numaflow/pkg/window"
 )
 
 // Options for forwarding the message
 type Options struct {
 	// readBatchSize is the default batch size
 	readBatchSize int64
-	// windowOpts Options for window
-	windowOpts *window.Options
 }
 
 type Option func(*Options) error
 
 func DefaultOptions() *Options {
 	return &Options{
 		readBatchSize: dfv1.DefaultReadBatchSize,
-		windowOpts:    window.DefaultOptions(),
 	}
 }
 
@@ -45,16 +41,3 @@ func WithReadBatchSize(f int64) Option {
 		return nil
 	}
 }
-
-// WithWindowOptions sets different window options
-func WithWindowOptions(opts ...window.Option) Option {
-	return func(options *Options) error {
-		for _, opt := range opts {
-			err := opt(options.windowOpts)
-			if err != nil {
-				return err
-			}
-		}
-		return nil
-	}
-}
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
@@ -67,7 +67,7 @@ func NewReadLoop(ctx context.Context,
 	toBuffers map[string]isb.BufferWriter,
 	whereToDecider forward.ToWhichStepDecider,
 	pw map[string]publish.Publisher,
-	_ *window.Options) *ReadLoop {
+) *ReadLoop {
 
 	op := newOrderedForwarder(ctx)
 

diff --git a/pkg/reduce/reduce.go b/pkg/reduce/reduce.go
@@ -67,7 +67,7 @@ func NewDataForward(ctx context.Context,
 		}
 	}
 
-	rl := readloop.NewReadLoop(ctx, udf, pbqManager, windowingStrategy, toBuffers, whereToDecider, watermarkPublishers, options.windowOpts)
+	rl := readloop.NewReadLoop(ctx, udf, pbqManager, windowingStrategy, toBuffers, whereToDecider, watermarkPublishers)
 	return &DataForward{
 		fromBuffer:          fromBuffer,
 		toBuffers:           toBuffers,

diff --git a/pkg/window/options.go b/pkg/window/options.go