Station partitioning

db/db.go

@@ -342,6 +342,8 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		ALTER TYPE enum_retention_type ADD VALUE 'ack_based';
 		ALTER TABLE stations ADD COLUMN IF NOT EXISTS tenant_name VARCHAR NOT NULL DEFAULT '$memphis';
 		ALTER TABLE stations ADD COLUMN IF NOT EXISTS resend_disabled BOOL NOT NULL DEFAULT false;
+		ALTER TABLE stations ADD COLUMN IF NOT EXISTS partitions INTEGER[];
+		ALTER TABLE stations ADD COLUMN IF NOT EXISTS version INTEGER NOT NULL DEFAULT 0;
 		DROP INDEX IF EXISTS unique_station_name_deleted;
 		CREATE UNIQUE INDEX unique_station_name_deleted ON stations(name, is_deleted, tenant_name) WHERE is_deleted = false;
 		END IF;
@@ -371,6 +373,8 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		tiered_storage_enabled BOOL NOT NULL,
 		tenant_name VARCHAR NOT NULL DEFAULT '$memphis',
 		resend_disabled BOOL NOT NULL DEFAULT false,
+		partitions INTEGER[],
+		version INTEGER NOT NULL DEFAULT 0,
 		PRIMARY KEY (id),
 		CONSTRAINT fk_tenant_name_stations
 			FOREIGN KEY(tenant_name)
@@ -1157,7 +1161,9 @@ func InsertNewStation(
 	isNative bool,
 	dlsConfiguration models.DlsConfiguration,
 	tieredStorageEnabled bool,
-	tenantName string) (models.Station, int64, error) {
+	tenantName string,
+	partitionsList []int,
+	version int) (models.Station, int64, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 
@@ -1185,9 +1191,11 @@ func InsertNewStation(
 		dls_configuration_poison, 
 		dls_configuration_schemaverse,
 		tiered_storage_enabled,
-		tenant_name
+		tenant_name,
+		partitions,
+		version
 		) 
-    VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) RETURNING id`
+    VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) RETURNING id`
 
 	stmt, err := conn.Conn().Prepare(ctx, "insert_new_station", query)
 	if err != nil {
@@ -1202,7 +1210,7 @@ func InsertNewStation(
 	}
 	rows, err := conn.Conn().Query(ctx, stmt.Name,
 		stationName, retentionType, retentionValue, storageType, replicas, userId, username, createAt, updatedAt,
-		false, schemaName, schemaVersionUpdate, idempotencyWindow, isNative, dlsConfiguration.Poison, dlsConfiguration.Schemaverse, tieredStorageEnabled, tenantName)
+		false, schemaName, schemaVersionUpdate, idempotencyWindow, isNative, dlsConfiguration.Poison, dlsConfiguration.Schemaverse, tieredStorageEnabled, tenantName, partitionsList, version)
 	if err != nil {
 		return models.Station{}, 0, err
 	}
@@ -1254,6 +1262,8 @@ func InsertNewStation(
 		DlsConfigurationSchemaverse: dlsConfiguration.Schemaverse,
 		TieredStorageEnabled:        tieredStorageEnabled,
 		TenantName:                  tenantName,
+		PartitionsList:              partitionsList,
+		Version:                     version,
 	}
 
 	rowsAffected := rows.CommandTag().RowsAffected()
@@ -1487,6 +1497,8 @@ func GetAllStationsDetailsLight(tenantName string) ([]models.ExtendedStationLigh
 			&stationRes.TieredStorageEnabled,
 			&stationRes.TenantName,
 			&stationRes.ResendDisabled,
+			&stationRes.PartitionsList,
+			&stationRes.Version,
 			&stationRes.Activity,
 		); err != nil {
 			return []models.ExtendedStationLight{}, err

models/monitoring.go

@@ -53,7 +53,8 @@ type SystemComponentsStatus struct {
 }
 
 type GetStationOverviewDataSchema struct {
-	StationName string `form:"station_name" json:"station_name"  binding:"required"`
+	StationName     string `form:"station_name" json:"station_name"  binding:"required"`
+	PartitionNumber int    `form:"partition_number" json:"partition_number"  binding:"required"`
 }
 
 type SystemLogsRequest struct {

models/stations.go

@@ -37,6 +37,7 @@ type MessageDetails struct {
 	ConnectionId string            `json:"connection_id" `
 	Size         int               `json:"size"`
 	Headers      map[string]string `json:"headers"`
+	Partition    int               `json:"partition"`
 }
 
 type Station struct {
@@ -60,6 +61,8 @@ type Station struct {
 	TieredStorageEnabled        bool      `json:"tiered_storage_enabled"`
 	TenantName                  string    `json:"tenant_name"`
 	ResendDisabled              bool      `json:"resend_disabled"`
+	PartitionsList              []int     `json:"partitions_list"`
+	Version                     int       `json:"version"`
 }
 
 type GetStationResponseSchema struct {
@@ -80,6 +83,8 @@ type GetStationResponseSchema struct {
 	DlsConfiguration     DlsConfiguration `json:"dls_configuration"`
 	TieredStorageEnabled bool             `json:"tiered_storage_enabled"`
 	ResendDisabled       bool             `json:"resend_disabled"`
+	PartitionsList       []int            `json:"partitions_list"`
+	PartitionsNumber     int              `json:"partitions_number"`
 }
 
 type ExtendedStation struct {
@@ -133,6 +138,8 @@ type ExtendedStationLight struct {
 	TieredStorageEnabled        bool        `json:"tiered_storage_enabled,omitempty"`
 	TenantName                  string      `json:"tenant_name"`
 	ResendDisabled              bool        `json:"resend_disabled"`
+	PartitionsList              []int       `json:"partitions_list"`
+	Version                     int         `json"version"`
 }
 
 type ActiveProducersConsumersDetails struct {
@@ -165,6 +172,7 @@ type CreateStationSchema struct {
 	IdempotencyWindow    int64            `json:"idempotency_window_in_ms"`
 	DlsConfiguration     DlsConfiguration `json:"dls_configuration"`
 	TieredStorageEnabled bool             `json:"tiered_storage_enabled"`
+	PartitionsNumber     int              `json:"partitions_number"`
 }
 
 type DlsConfiguration struct {
@@ -209,11 +217,12 @@ type GetPoisonMessageJourneySchema struct {
 }
 
 type GetMessageDetailsSchema struct {
-	IsDls       bool   `form:"is_dls" json:"is_dls"`
-	DlsType     string `form:"dls_type" json:"dls_type"`
-	MessageId   int    `form:"message_id" json:"message_id"`
-	MessageSeq  int    `form:"message_seq" json:"message_seq"`
-	StationName string `form:"station_name" json:"station_name" binding:"required"`
+	IsDls           bool   `form:"is_dls" json:"is_dls"`
+	DlsType         string `form:"dls_type" json:"dls_type"`
+	MessageId       int    `form:"message_id" json:"message_id"`
+	MessageSeq      int    `form:"message_seq" json:"message_seq"`
+	StationName     string `form:"station_name" json:"station_name" binding:"required"`
+	PartitionNumber int    `form:"partition_number" json:"partition_number" binding:"required"`
 }
 
 type UseSchema struct {

server/memphis_cloud.go

@@ -46,6 +46,7 @@ import (
 
 const shouldCreateRootUserforGlobalAcc = true
 const TENANT_SEQUENCE_START_ID = 2
+const MAX_PARTITIONS = 5
 
 type BillingHandler struct{ S *Server }
 type TenantHandler struct{ S *Server }

server/memphis_handlers.go

@@ -78,15 +78,15 @@ func getUserDetailsFromMiddleware(c *gin.Context) (models.User, error) {
 func CreateDefaultStation(tenantName string, s *Server, sn StationName, userId int, username string) (models.Station, bool, error) {
 	stationName := sn.Ext()
 	replicas := getDefaultReplicas()
-	err := s.CreateStream(tenantName, sn, "message_age_sec", 604800, "file", 120000, replicas, false)
+	err := s.CreateStream(tenantName, sn, "message_age_sec", 604800, "file", 120000, replicas, false, 1)
 	if err != nil {
 		return models.Station{}, false, err
 	}
 
 	schemaName := ""
 	schemaVersionNumber := 0
 
-	newStation, rowsUpdated, err := db.InsertNewStation(stationName, userId, username, "message_age_sec", 604800, "file", replicas, schemaName, schemaVersionNumber, 120000, true, models.DlsConfiguration{Poison: true, Schemaverse: true}, false, tenantName)
+	newStation, rowsUpdated, err := db.InsertNewStation(stationName, userId, username, "message_age_sec", 604800, "file", replicas, schemaName, schemaVersionNumber, 120000, true, models.DlsConfiguration{Poison: true, Schemaverse: true}, false, tenantName, []int{}, 1)
 	if err != nil {
 		return models.Station{}, false, err
 	}

server/memphis_handlers_monitoring.go

@@ -525,40 +525,88 @@ func (mh MonitoringHandler) GetStationOverviewData(c *gin.Context) {
 		c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
 		return
 	}
-	totalMessages, err := stationsHandler.GetTotalMessages(station.TenantName, station.Name)
-	if err != nil {
-		if IsNatsErr(err, JSStreamNotFoundErr) {
-			serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAuditLogsByStation: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
-			c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
-		} else {
-			serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
-			c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+
+	var totalMessages int
+	if body.PartitionNumber == 0 {
+		totalMessages, err = stationsHandler.GetTotalMessages(station.TenantName, station.Name, station.PartitionsList)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAuditLogsByStation: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
 		}
-		return
+	} else {
+		totalMessages, err = stationsHandler.GetTotalPartitionMessages(station.TenantName, station.Name, body.PartitionNumber)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAuditLogsByStation: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
+		}
+
 	}
-	avgMsgSize, err := stationsHandler.GetAvgMsgSize(station)
-	if err != nil {
-		if IsNatsErr(err, JSStreamNotFoundErr) {
-			serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: does not exist", user.TenantName, user.Username, body.StationName)
-			c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
-		} else {
-			serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
-			c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+
+	var avgMsgSize int64
+	if body.PartitionNumber == 0 {
+		avgMsgSize, err = stationsHandler.GetAvgMsgSize(station)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
+		}
+	} else {
+		avgMsgSize, err = stationsHandler.GetPartitionAvgMsgSize(station.TenantName, fmt.Sprintf("%v$%v", stationName.Intern(), body.PartitionNumber))
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("[tenant: %v][user: %v]GetStationOverviewData at GetAvgMsgSize: At station %v: %v", user.TenantName, user.Username, body.StationName, err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
 		}
-		return
 	}
 
 	messagesToFetch := 1000
-	messages, err := stationsHandler.GetMessages(station, messagesToFetch)
-	if err != nil {
-		if IsNatsErr(err, JSStreamNotFoundErr) {
-			serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetMessages: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
-			c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
-		} else {
-			serv.Errorf("GetStationOverviewData at GetMessages: At station " + body.StationName + ": " + err.Error())
-			c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+	messages := make([]models.MessageDetails, 0)
+	if body.PartitionNumber == 0 {
+		messages, err = stationsHandler.GetMessages(station, messagesToFetch)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetMessages: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("GetStationOverviewData at GetMessages: At station " + body.StationName + ": " + err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
+		}
+	} else {
+		messages, err = stationsHandler.GetMessagesFromPartition(station, fmt.Sprintf("%v$%v", stationName.Intern(), body.PartitionNumber), messagesToFetch, body.PartitionNumber)
+		if err != nil {
+			if IsNatsErr(err, JSStreamNotFoundErr) {
+				serv.Warnf("[tenant: %v][user: %v]GetStationOverviewData at GetMessages: nats error At station %v: does not exist", user.TenantName, user.Username, body.StationName)
+				c.AbortWithStatusJSON(SHOWABLE_ERROR_STATUS_CODE, gin.H{"message": "Station " + body.StationName + " does not exist"})
+			} else {
+				serv.Errorf("GetStationOverviewData at GetMessages: At station " + body.StationName + ": " + err.Error())
+				c.AbortWithStatusJSON(500, gin.H{"message": "Server error"})
+			}
+			return
 		}
-		return
 	}
 
 	poisonMessages, schemaFailedMessages, totalDlsAmount, err := poisonMsgsHandler.GetDlsMsgsByStationLight(station)