Station partitioning

db/db.go

@@ -340,9 +340,11 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		IF EXISTS (
 			SELECT 1 FROM information_schema.tables WHERE table_name = 'stations' AND table_schema = 'public'
 		) THEN
-		ALTER TYPE enum_retention_type ADD VALUE 'ack_based';
+		ALTER TYPE enum_retention_type ADD VALUE IF NOT EXISTS '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;
@@ -372,6 +374,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)
@@ -436,6 +440,7 @@ func createTables(MetadataDbClient MetadataStorage) error {
 			CREATE INDEX IF NOT EXISTS producer_name ON producers(name);
 			CREATE INDEX IF NOT EXISTS producer_tenant_name ON producers(tenant_name);
 			CREATE INDEX IF NOT EXISTS producer_connection_id ON producers(connection_id);
+			ALTER TABLE producers ADD COLUMN IF NOT EXISTS partitions INTEGER[];
 			IF EXISTS (
 				SELECT 1
 				FROM information_schema.columns
@@ -458,6 +463,7 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		is_active BOOL NOT NULL DEFAULT true,
 		updated_at TIMESTAMPTZ NOT NULL,
 		tenant_name VARCHAR NOT NULL DEFAULT '$memphis',
+		partitions INTEGER[],
 		PRIMARY KEY (id),
 		CONSTRAINT fk_station_id
 			FOREIGN KEY(station_id)
@@ -478,6 +484,7 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		) THEN
 			ALTER TABLE dls_messages ADD COLUMN IF NOT EXISTS tenant_name VARCHAR NOT NULL DEFAULT '$memphis';
 			ALTER TABLE dls_messages ADD COLUMN IF NOT EXISTS producer_name VARCHAR NOT NULL DEFAULT '';
+			ALTER TABLE dls_messages ADD COLUMN IF NOT EXISTS partition_number INTEGER NOT NULL DEFAULT -1;
 			DROP INDEX IF EXISTS dls_producer_id;
 			IF EXISTS (
 				SELECT 1 FROM information_schema.columns WHERE table_name = 'dls_messages' AND column_name = 'producer_id'
@@ -509,6 +516,7 @@ func createTables(MetadataDbClient MetadataStorage) error {
 		validation_error VARCHAR DEFAULT '',
 		tenant_name VARCHAR NOT NULL DEFAULT '$memphis',
 		producer_name VARCHAR NOT NULL,
+		partition_number INTEGER NOT NULL DEFAULT -1,
 		PRIMARY KEY (id),
 		CONSTRAINT fk_station_id
 			FOREIGN KEY(station_id)
@@ -1168,7 +1176,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()
 
@@ -1196,9 +1206,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 {
@@ -1213,7 +1225,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
 	}
@@ -1265,6 +1277,8 @@ func InsertNewStation(
 		DlsConfigurationSchemaverse: dlsConfiguration.Schemaverse,
 		TieredStorageEnabled:        tieredStorageEnabled,
 		TenantName:                  tenantName,
+		PartitionsList:              partitionsList,
+		Version:                     version,
 	}
 
 	rowsAffected := rows.CommandTag().RowsAffected()
@@ -1498,6 +1512,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
@@ -2332,7 +2348,7 @@ func GetActiveProducerByStationID(producerName string, stationId int) (bool, mod
 	return true, producers[0], nil
 }
 
-func InsertNewProducer(name string, stationId int, producerType string, connectionIdObj string, tenantName string) (models.Producer, error) {
+func InsertNewProducer(name string, stationId int, producerType string, connectionIdObj string, tenantName string, partitionsList []int) (models.Producer, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 
@@ -2349,8 +2365,9 @@ func InsertNewProducer(name string, stationId int, producerType string, connecti
 		is_active, 
 		updated_at, 
 		type,
-		tenant_name) 
-    VALUES($1, $2, $3, $4, $5, $6, $7) RETURNING id`
+		tenant_name,
+		partitions) 
+    VALUES($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id`
 
 	stmt, err := conn.Conn().Prepare(ctx, "insert_new_producer", query)
 	if err != nil {
@@ -2363,7 +2380,7 @@ func InsertNewProducer(name string, stationId int, producerType string, connecti
 	if tenantName != conf.GlobalAccount {
 		tenantName = strings.ToLower(tenantName)
 	}
-	rows, err := conn.Conn().Query(ctx, stmt.Name, name, stationId, connectionIdObj, isActive, updatedAt, producerType, tenantName)
+	rows, err := conn.Conn().Query(ctx, stmt.Name, name, stationId, connectionIdObj, isActive, updatedAt, producerType, tenantName, partitionsList)
 	if err != nil {
 		return models.Producer{}, err
 	}
@@ -2393,13 +2410,14 @@ func InsertNewProducer(name string, stationId int, producerType string, connecti
 	}
 
 	newProducer := models.Producer{
-		ID:           producerId,
-		Name:         name,
-		StationId:    stationId,
-		Type:         producerType,
-		ConnectionId: connectionIdObj,
-		IsActive:     isActive,
-		UpdatedAt:    time.Now(),
+		ID:             producerId,
+		Name:           name,
+		StationId:      stationId,
+		Type:           producerType,
+		ConnectionId:   connectionIdObj,
+		IsActive:       isActive,
+		UpdatedAt:      time.Now(),
+		PartitionsList: partitionsList,
 	}
 	return newProducer, nil
 }
@@ -4622,7 +4640,7 @@ func GetAllActiveUsersStations(tenantName string) ([]models.FilteredUser, error)
 	FROM users AS u
 	JOIN stations AS s ON u.id = s.created_by
 	WHERE s.tenant_name=$1 AND username NOT LIKE '$%'` // filter memphis internal users
-	
+
 	stmt, err := conn.Conn().Prepare(ctx, "get_all_active_users_stations", query)
 	if err != nil {
 		return []models.FilteredUser{}, err
@@ -4658,7 +4676,7 @@ func GetAllActiveUsersSchemaVersions(tenantName string) ([]models.FilteredUser,
 	FROM users AS u
 	JOIN schema_versions AS s ON u.id = s.created_by
 	WHERE s.tenant_name=$1 AND username NOT LIKE '$%'` // filter memphis internal users
-	
+
 	stmt, err := conn.Conn().Prepare(ctx, "get_all_active_users_schema_versions", query)
 	if err != nil {
 		return []models.FilteredUser{}, err
@@ -5146,7 +5164,7 @@ func GetImage(name string, tenantName string) (bool, models.Image, error) {
 }
 
 // dls Functions
-func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, validationError string, tenantName string) (models.DlsMessage, error) {
+func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, validationError string, tenantName string, partitionNumber int) (models.DlsMessage, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 	connection, err := MetadataDbClient.Client.Acquire(ctx)
@@ -5164,8 +5182,9 @@ func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string,
 			message_type,
 			validation_error,
 			tenant_name,
-			producer_name) 
-		VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9)
+			producer_name,
+			partition_number) 
+		VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
 		RETURNING id`
 
 	stmt, err := connection.Conn().Prepare(ctx, "insert_dls_messages", query)
@@ -5176,7 +5195,7 @@ func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string,
 	if tenantName != conf.GlobalAccount {
 		tenantName = strings.ToLower(tenantName)
 	}
-	rows, err := connection.Conn().Query(ctx, stmt.Name, stationId, messageSeq, poisonedCgs, messageDetails, updatedAt, "schema", validationError, tenantName, producerName)
+	rows, err := connection.Conn().Query(ctx, stmt.Name, stationId, messageSeq, poisonedCgs, messageDetails, updatedAt, "schema", validationError, tenantName, producerName, partitionNumber)
 	if err != nil {
 		return models.DlsMessage{}, err
 	}
@@ -5207,8 +5226,9 @@ func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string,
 			Data:     messageDetails.Data,
 			Headers:  messageDetails.Headers,
 		},
-		UpdatedAt:  updatedAt,
-		TenantName: tenantName,
+		UpdatedAt:       updatedAt,
+		TenantName:      tenantName,
+		PartitionNumber: partitionNumber,
 	}
 
 	if err := rows.Err(); err != nil {
@@ -5231,7 +5251,7 @@ func InsertSchemaverseDlsMsg(stationId int, messageSeq int, producerName string,
 	return deadLetterPayload, nil
 }
 
-func GetMsgByStationIdAndMsgSeq(stationId, messageSeq int) (bool, models.DlsMessage, error) {
+func GetMsgByStationIdAndMsgSeq(stationId, messageSeq, partitionNumber int) (bool, models.DlsMessage, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 
@@ -5241,14 +5261,14 @@ func GetMsgByStationIdAndMsgSeq(stationId, messageSeq int) (bool, models.DlsMess
 	}
 	defer connection.Release()
 
-	query := `SELECT * FROM dls_messages WHERE station_id = $1 AND message_seq = $2 LIMIT 1`
+	query := `SELECT * FROM dls_messages WHERE station_id = $1 AND message_seq = $2 AND partition_number = $3 LIMIT 1`
 
 	stmt, err := connection.Conn().Prepare(ctx, "get_dls_messages_by_station_id_and_message_seq", query)
 	if err != nil {
 		return false, models.DlsMessage{}, err
 	}
 
-	rows, err := connection.Conn().Query(ctx, stmt.Name, stationId, messageSeq)
+	rows, err := connection.Conn().Query(ctx, stmt.Name, stationId, messageSeq, partitionNumber)
 	if err != nil {
 		return false, models.DlsMessage{}, err
 	}
@@ -5265,7 +5285,7 @@ func GetMsgByStationIdAndMsgSeq(stationId, messageSeq int) (bool, models.DlsMess
 	return true, message[0], nil
 }
 
-func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, tenantName string) (int, error) {
+func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName string, poisonedCgs []string, messageDetails models.MessagePayload, tenantName string, partitionNumber int) (int, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
 
@@ -5302,12 +5322,12 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin
 	}
 	checkRows.Close()
 
-	query = `SELECT * FROM dls_messages WHERE station_id = $1 AND message_seq = $2 AND tenant_name =$3 LIMIT 1 FOR UPDATE`
+	query = `SELECT * FROM dls_messages WHERE station_id = $1 AND message_seq = $2 AND tenant_name =$3 AND partition_number = $4 LIMIT 1 FOR UPDATE`
 	stmt, err = tx.Prepare(ctx, "handle_insert_dls_message", query)
 	if err != nil {
 		return 0, err
 	}
-	rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, tenantName)
+	rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, tenantName, partitionNumber)
 	if err != nil {
 		return 0, err
 	}
@@ -5329,7 +5349,8 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin
 			updated_at,
 			message_type,
 			validation_error,
-			tenant_name
+			tenant_name,
+			partition_number
 			) 
 		VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9)
 		RETURNING id`
@@ -5342,7 +5363,7 @@ func StorePoisonMsg(stationId, messageSeq int, cgName string, producerName strin
 		if tenantName != conf.GlobalAccount {
 			tenantName = strings.ToLower(tenantName)
 		}
-		rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, producerName, poisonedCgs, messageDetails, updatedAt, "poison", "", tenantName)
+		rows, err := tx.Query(ctx, stmt.Name, stationId, messageSeq, producerName, poisonedCgs, messageDetails, updatedAt, "poison", "", tenantName, partitionNumber)
 		if err != nil {
 			return 0, err
 		}
@@ -5497,6 +5518,28 @@ func PurgeDlsMsgsFromStation(station_id int) error {
 	return nil
 }
 
+func PurgeDlsMsgsFromPartition(station_id, partitionNumber int) error {
+	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
+	defer cancelfunc()
+	conn, err := MetadataDbClient.Client.Acquire(ctx)
+	if err != nil {
+		return errors.New("PurgeDlsMsgsFromPartition: " + err.Error())
+	}
+	defer conn.Release()
+
+	query := `DELETE FROM dls_messages where station_id=$1 AND partition_number = $2`
+	stmt, err := conn.Conn().Prepare(ctx, "purge_dls_messages", query)
+	if err != nil {
+		return err
+	}
+
+	_, err = conn.Conn().Exec(ctx, stmt.Name, station_id, partitionNumber)
+	if err != nil {
+		return errors.New("PurgeDlsMsgsFromPartition: " + err.Error())
+	}
+	return nil
+}
+
 func RemoveCgFromDlsMsg(msgId int, cgName string, tenantName string) error {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()
@@ -5550,6 +5593,35 @@ func GetDlsMsgsByStationId(stationId int) ([]models.DlsMessage, error) {
 	return dlsMsgs, nil
 }
 
+func GetDlsMsgsByStationAndPartition(stationId, partitionNumber int) ([]models.DlsMessage, error) {
+	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
+	defer cancelfunc()
+	conn, err := MetadataDbClient.Client.Acquire(ctx)
+	if err != nil {
+		return []models.DlsMessage{}, err
+	}
+	defer conn.Release()
+	query := `SELECT * from dls_messages where station_id=$1 AND partition_number = $2 ORDER BY updated_at DESC limit 1000`
+	stmt, err := conn.Conn().Prepare(ctx, "get_dls_msg_by_station_and_partition", query)
+	if err != nil {
+		return []models.DlsMessage{}, err
+	}
+	rows, err := conn.Conn().Query(ctx, stmt.Name, stationId, partitionNumber)
+	if err != nil {
+		return []models.DlsMessage{}, err
+	}
+	defer rows.Close()
+	dlsMsgs, err := pgx.CollectRows(rows, pgx.RowToStructByPos[models.DlsMessage])
+	if err != nil {
+		return []models.DlsMessage{}, err
+	}
+	if len(dlsMsgs) == 0 {
+		return []models.DlsMessage{}, nil
+	}
+
+	return dlsMsgs, nil
+}
+
 func GetDlsMessageById(messageId int) (bool, models.DlsMessage, error) {
 	ctx, cancelfunc := context.WithTimeout(context.Background(), DbOperationTimeout*time.Second)
 	defer cancelfunc()

models/dead_letter_station.go

@@ -61,6 +61,7 @@ type SchemaVerseDlsMessageSdk struct {
 	Producer        ProducerDetails `json:"producer"`
 	Message         MessagePayload  `json:"message"`
 	ValidationError string          `json:"validation_error"`
+	PartitionNumber int             `json:"partition_number"`
 }
 
 type DlsMessage struct {
@@ -74,6 +75,7 @@ type DlsMessage struct {
 	ValidationError string         `json:"validation_error"`
 	TenantName      string         `json:"tenant_name"`
 	ProducerName    string         `json:"producer_name"`
+	PartitionNumber int            `json:"partition_number"`
 }
 
 type DlsMsgResendAll struct {

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 {