SQL: Rebuild clustered columnstore indexes when they require maintenance

----------------------------------------------------------------------
-- Rebuild clustered columnstore indexes when necessary
--
-- Dr Greg Low v2.4
----------------------------------------------------------------------
-- rebuild when more than supplied % of rows have been deleted
-- rebuild when any segments contain more than supplied % deleted rows
-- rebuild if more than supplied number of segments are empty
-- rebuild when segments are on average less than the supplied % of
--     maximum capacity

DECLARE @DeletedTotalRowPercentage int = 10; 
DECLARE @DeletedSegmentsRowPercentage int = 20;
DECLARE @EmptySegmentsAllowed int = 0;
DECLARE @SegmentPotentialFullnessPercentage int = 90;

----------------------------------------------------------------------

DECLARE @IndexesToRebuild TABLE 
(
    SchemaName sysname,
    TableName sysname,
    IndexName sysname,
    PartitionNumber int,
    IsPartitioned bit
);

SET NOCOUNT ON;

WITH ClusteredColumnstoreIndexes
AS
( 
    SELECT t.object_id AS ObjectID,
           SCHEMA_NAME(t.schema_id) AS SchemaName,
           t.name AS TableName,
           i.name AS IndexName
    FROM sys.indexes AS i
    INNER JOIN sys.tables AS t
    ON t.object_id = i.object_id
    WHERE i.type = 5
),
RowGroups
AS
( 
    SELECT csrg.object_id AS ObjectID,
           csrg.partition_number AS PartitionNumber,
           csrg.total_rows AS TotalRows,
           csrg.deleted_rows AS DeletedRows,
           csrg.deleted_rows * 100.0 / csrg.total_rows AS DeletedPercentage,
           CASE WHEN csrg.total_rows = csrg.deleted_rows
                THEN 1 ELSE 0
           END AS IsEmptySegment
    FROM sys.column_store_row_groups AS csrg
    WHERE csrg.state = 3 -- Compressed 
    -- (Ignoring: 0 - Hidden, 1 - Open, 2 - Closed, 4 - Tombstone)
),
IndexPartitionStats
AS
( 
    SELECT cci.ObjectID,
           cci.SchemaName,
           cci.TableName,
           cci.IndexName,
           rg.PartitionNumber,
           SUM(CAST(rg.TotalRows AS decimal(18,0))) AS TotalRows,
           SUM(CAST(rg.DeletedRows AS decimal(18,0))) AS DeletedRows,
           SUM(CAST(rg.DeletedRows AS decimal(18,0))) * 100.0
             / SUM(CAST(rg.TotalRows AS decimal(18,0)))
             AS DeletedPercentage,
           SUM(rg.IsEmptySegment) aS EmptySegments,
           COUNT(rg.TotalRows) AS TotalSegments
    FROM ClusteredColumnstoreIndexes AS cci
    INNER JOIN RowGroups AS rg
    ON rg.ObjectID = cci.ObjectID
    GROUP BY cci.ObjectID, cci.SchemaName, cci.TableName, cci.IndexName, rg.PartitionNumber
)
INSERT @IndexesToRebuild (SchemaName, TableName, IndexName, PartitionNumber, IsPartitioned)
SELECT s.SchemaName,
       s.TableName,
       s.IndexName,
       s.PartitionNumber,
       CASE WHEN EXISTS(SELECT 1
                        FROM IndexPartitionStats AS ps
                        WHERE s.SchemaName = ps.SchemaName
                        AND s.TableName = ps.TableName
                        AND s.IndexName = ps.IndexName
                        AND ps.PartitionNumber > 1)
            THEN 1
            ELSE 0
       END
FROM IndexPartitionStats AS s
WHERE s.DeletedPercentage > @DeletedTotalRowPercentage
OR s.EmptySegments > @EmptySegmentsAllowed
OR EXISTS(SELECT 1 FROM RowGroups AS rg
                   WHERE rg.ObjectID = s.ObjectID
                   AND rg.DeletedPercentage > @DeletedSegmentsRowPercentage)
OR (s.TotalRows * 100 / (s.TotalSegments * 1048576.0))
            < @SegmentPotentialFullnessPercentage;

DECLARE @SchemaName sysname;
DECLARE @TableName sysname;
DECLARE @IndexName sysname;
DECLARE @PartitionNumber int;
DECLARE @IsPartitioned bit;
DECLARE @SqlCommand nvarchar(max);
DECLARE IndexList CURSOR FAST_FORWARD READ_ONLY
FOR
  SELECT SchemaName, TableName, IndexName, PartitionNumber, IsPartitioned
  FROM @IndexesToRebuild
  ORDER BY SchemaName, TableName, IndexName, PartitionNumber;

OPEN IndexList;
FETCH NEXT FROM IndexList INTO @SchemaName, @TableName, @IndexName, @PartitionNumber, @IsPartitioned;
WHILE @@FETCH_STATUS = 0
BEGIN
  SET @SqlCommand = N'ALTER INDEX ' + QUOTENAME(@IndexName)
                  + N' ON ' + QUOTENAME(@SchemaName)
                  + N'.' + QUOTENAME(@TableName)
                  + N' REBUILD'
                  + CASE WHEN @IsPartitioned <> 0
                         THEN N' PARTITION = ' + CAST(@PartitionNumber AS nvarchar(10))
                         ELSE N''
                    END
                  + ';';
  PRINT @SqlCommand;
  EXEC(@SqlCommand);
  FETCH NEXT FROM IndexList INTO @SchemaName, @TableName, @IndexName, @PartitionNumber, @IsPartitioned;
END;
CLOSE IndexList;
DEALLOCATE IndexList;