/* Enter a unique ExecutionPlan */
@Plan:name('IsAnalytics-ExecutionPlan-AuthenticationData')

/* Enter a unique description for ExecutionPlan */
-- @Plan:description('ExecutionPlan')

/* define streams/tables and write queries here ... */


@Import('org.wso2.is.analytics.stream.OverallAuthentication:1.0.0')
define stream InStream (meta_tenantId int, contextId string, eventId string, eventType string, authenticationSuccess bool, username string, localUsername string, userStoreDomain string, tenantDomain string, remoteIp string, region string, inboundAuthType string, serviceProvider string, rememberMeEnabled bool, forceAuthEnabled bool, passiveAuthEnabled bool, rolesCommaSeparated string, authenticationStep string, identityProvider string, authStepSuccess bool, stepAuthenticator string, isFirstLogin bool, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.ProcessedOverallAuthentication:1.0.0')
define stream OverallAuthenticationProcessedStream (meta_tenantId int, contextId string, eventId string, eventType string, authenticationSuccess bool, username string, localUsername string, userStoreDomain string, tenantDomain string, remoteIp string, region string, inboundAuthType string, serviceProvider string, rememberMeEnabled bool, forceAuthEnabled bool, passiveAuthEnabled bool, rolesCommaSeparated string, authenticationStep string, identityProvider string, authStepSuccess bool, stepAuthenticator string, isFirstLogin bool, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.LocalUserAuthStatPerMinute:1.0.0')
define stream PerMinLocalUsernameStream (meta_tenantId int, year int, month int, day int, hour int, minute int, localUsername string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.SpAuthStatPerMinute:1.0.0')
define stream PerMinserviceProviderStream (meta_tenantId int, year int, month int, day int, hour int, minute int, serviceProvider string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, authFirstSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.AuthStatPerMinute:1.0.0')
define stream PerMinAuthenticationStatStream (meta_tenantId int, year int, month int, day int, hour int, minute int, username string, localUsername string, rolesCommaSeparated string, serviceProvider string, identityProvider string, remoteIp string, region string, userStoreDomain string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, authFirstSuccessCount long,facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.UserRegionStatPerMinute:1.0.0')
define stream PerMinRegionStream (meta_tenantId int, year int, month int, day int, hour int, minute int, remoteIp string, region string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.UserstoreAuthStatPerMinute:1.0.0')
define stream PerMinUserStoreStream (meta_tenantId int, year int, month int, day int, hour int, minute int, userStoreDomain string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.RoleAuthStatPerMinute:1.0.0')
define stream PerMinRoleStream (meta_tenantId int, year int, month int, day int, hour int, minute int, username string, role string, serviceProvider string, identityProvider string, remoteIp string, region string,userStoreDomain string, authSuccessCount long, authFailureCount long,authStepSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.IdpAuthStatPerMinute:1.0.0')
define stream PerMinIdentityProviderStream (meta_tenantId int, year int, month int, day int, hour int, minute int, identityProvider string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, authFirstSuccessCount long, facetStartTime string, identityProviderType string, _timestamp long);

@Export('org.wso2.is.analytics.stream.UserAuthStatPerMinute:1.0.0')
define stream PerMinUserStream (meta_tenantId int, year int, month int, day int, hour int, minute int, username string, authSuccessCount long, authFailureCount long, authStepSuccessCount long, authFirstSuccessCount long, facetStartTime string, identityProviderType string,_timestamp long);

define window countWindow (meta_tenantId int, eventId string, username string, localUsername string, userStoreDomain string, tenantDomain string, remoteIp string, region string, inboundAuthType string, serviceProvider string, rememberMeEnabled bool, forceAuthEnabled bool, passiveAuthEnabled bool, rolesCommaSeparated string, authenticationStep string, identityProvider string, authStepSuccess bool, stepAuthenticator string, isFirstLogin bool, successValue int, failureValue int, stepSuccessValue int, firstLoginValue int, identityProviderType string, batchEndTime long, _timestamp long) externalTimeBatch(batchEndTime, 1 sec, 0, 10 sec, true);

--- Pre Processing Queries -------

from InStream 
select meta_tenantId, contextId, eventId, eventType, authenticationSuccess, username, localUsername, userStoreDomain, tenantDomain, remoteIp, isAnalytics:ipToCountry(remoteIp) as region, inboundAuthType, serviceProvider, rememberMeEnabled, forceAuthEnabled, passiveAuthEnabled, rolesCommaSeparated, authenticationStep, identityProvider, authStepSuccess, stepAuthenticator, isFirstLogin, identityProviderType, _timestamp
insert into OverallAuthenticationProcessedStream;

from OverallAuthenticationProcessedStream
select meta_tenantId, eventId, username, localUsername, userStoreDomain, tenantDomain, remoteIp, region, inboundAuthType, serviceProvider, rememberMeEnabled, forceAuthEnabled, passiveAuthEnabled, rolesCommaSeparated, authenticationStep, identityProvider, authStepSuccess, stepAuthenticator, isFirstLogin,ifThenElse(authenticationSuccess == true, 1, 0) as successValue,ifThenElse((authStepSuccess == false and eventType == 'step'), 1, 0) as failureValue,ifThenElse((authStepSuccess == true and eventType == 'step'), 1, 0) as stepSuccessValue,ifThenElse(authenticationSuccess == true AND isFirstLogin == true, 1, 0) as firstLoginValue,identityProviderType, _timestamp as eventTime, _timestamp
insert into countStream;


-- Per Second -------------

from countStream
select meta_tenantId, eventId, username, localUsername, userStoreDomain, tenantDomain, remoteIp, region, inboundAuthType, serviceProvider, rememberMeEnabled, forceAuthEnabled, passiveAuthEnabled, rolesCommaSeparated, authenticationStep, identityProvider, authStepSuccess, stepAuthenticator, isFirstLogin, successValue, failureValue, stepSuccessValue, firstLoginValue,identityProviderType, eventTime as batchEndTime, _timestamp
insert into countWindow;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, username, localUsername, rolesCommaSeparated,serviceProvider,identityProvider,remoteIp,region,userStoreDomain,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount,sum(stepSuccessValue) as authStepSuccessCount, sum(firstLoginValue) as  authFirstSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username, localUsername,rolesCommaSeparated,serviceProvider,identityProvider,remoteIp,userStoreDomain,identityProviderType
insert into perSecondAuthenticationStatStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,  username,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount,sum(stepSuccessValue) as authStepSuccessCount, sum(firstLoginValue) as  authFirstSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username,identityProviderType
insert into perSecondUserStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,  localUsername,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount,sum(stepSuccessValue) as authStepSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username,identityProviderType
insert into perSecondLocalUsernameStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, identityProvider,sum(stepSuccessValue) as authSuccessCount, sum(failureValue) as authFailureCount, sum(stepSuccessValue) as authStepSuccessCount, sum(firstLoginValue) as  authFirstSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, identityProvider,identityProviderType
insert into perSecondIdentityProviderStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, userStoreDomain,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount, sum(stepSuccessValue) as authStepSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, userStoreDomain,identityProviderType
insert into perSecondUserStoreStream;

from countWindow#isAnalytics:duplicator(rolesCommaSeparated)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, username,role,serviceProvider,identityProvider,remoteIp,region,userStoreDomain, sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount,sum(stepSuccessValue) as authStepSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username,role,serviceProvider,identityProvider,remoteIp,userStoreDomain,identityProviderType
insert into perSecondRoleStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, serviceProvider,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount, sum(stepSuccessValue) as authStepSuccessCount, sum(firstLoginValue) as  authFirstSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime,identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, serviceProvider,identityProviderType
insert into perSecondserviceProviderStream;

from countWindow
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute, remoteIp,region,sum(successValue) as authSuccessCount, sum(failureValue) as authFailureCount, sum(stepSuccessValue) as authStepSuccessCount, convert(isAnalytics:getSecondStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getSecondStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, remoteIp,identityProviderType
insert into perSecondRegionStream;


-- Per Minute ---------------

from perSecondAuthenticationStatStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , username, '_', localUsername, '_', rolesCommaSeparated, '_', serviceProvider, '_', identityProvider, '_', remoteIp , '_' , userStoreDomain , '_' , identityProviderType) as uniqueId, year, month, day, hour, minute, username, localUsername, rolesCommaSeparated,serviceProvider,identityProvider,remoteIp,region,userStoreDomain, authSuccessCount, authFailureCount, authStepSuccessCount, authFirstSuccessCount, facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondAuthenticationStatStreamWithUniqueId;

from perSecondAuthenticationStatStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,username,localUsername,rolesCommaSeparated,serviceProvider,identityProvider,remoteIp,region,userStoreDomain,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, sum(authFirstSuccessCount) as  authFirstSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username, localUsername, rolesCommaSeparated,serviceProvider,identityProvider,remoteIp,userStoreDomain,identityProviderType
insert into PerMinAuthenticationStatStream;

---------

from perSecondUserStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , username, '_', identityProviderType) as uniqueId, year, month, day, hour, minute, username, authSuccessCount, authFailureCount, authStepSuccessCount, authFirstSuccessCount,  facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondUserStreamWithUniqueId;

from perSecondUserStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,username,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, sum(authFirstSuccessCount) as  authFirstSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username,identityProviderType
insert into PerMinUserStream;

---------

from perSecondLocalUsernameStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , localUsername, '_', identityProviderType) as uniqueId, year, month, day, hour, minute, localUsername, authSuccessCount, authFailureCount, authStepSuccessCount,  facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondLocalUsernameStreamWithUniqueId;

from perSecondLocalUsernameStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,localUsername,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, localUsername,identityProviderType
insert into PerMinLocalUsernameStream;

----------

from perSecondIdentityProviderStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , identityProvider, '_', identityProviderType) as uniqueId, year, month, day, hour, minute,identityProvider, authSuccessCount, authFailureCount, authStepSuccessCount, authFirstSuccessCount, facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondIdentityProviderStreamWithUniqueId;

from perSecondIdentityProviderStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,identityProvider,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, sum(authFirstSuccessCount) as  authFirstSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, identityProvider,identityProviderType
insert into PerMinIdentityProviderStream;

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

from perSecondUserStoreStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , userStoreDomain, '_', identityProviderType) as uniqueId, year, month, day, hour, minute,userStoreDomain, authSuccessCount, authFailureCount, authStepSuccessCount, facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondUserStoreStreamWithUniqueId;


from perSecondUserStoreStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,userStoreDomain,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, userStoreDomain,identityProviderType
insert into PerMinUserStoreStream;

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

from perSecondRoleStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , username, '_', role, '_', serviceProvider, '_', identityProvider, '_', remoteIp , '_', userStoreDomain, '_', identityProviderType) as uniqueId, year, month, day, hour, minute,username,role,serviceProvider,identityProvider,remoteIp,region,userStoreDomain, authSuccessCount, authFailureCount, authStepSuccessCount, facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondRoleStreamWithUniqueId;

from perSecondRoleStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,username,role,serviceProvider,identityProvider,remoteIp,region,userStoreDomain, sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount,sum(authStepSuccessCount) as authStepSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, username,role,serviceProvider,identityProvider,remoteIp,userStoreDomain,identityProviderType
insert into PerMinRoleStream;

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

from perSecondserviceProviderStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , serviceProvider, '_', identityProviderType) as uniqueId, year, month, day, hour, minute, serviceProvider, authSuccessCount, authFailureCount, authStepSuccessCount, authFirstSuccessCount, facetStartTime,identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondserviceProviderStreamWithUniqueId;

from perSecondserviceProviderStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,serviceProvider,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount ,sum(authStepSuccessCount) as authStepSuccessCount,sum(authFirstSuccessCount) as  authFirstSuccessCount , convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime,identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, serviceProvider,identityProviderType
insert into PerMinserviceProviderStream;

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

from perSecondRegionStream
select meta_tenantId, str:concat(meta_tenantId, '_' , facetStartTime, '_' , remoteIp, '_', identityProviderType) as uniqueId, year, month, day, hour, minute, remoteIp,region, authSuccessCount, authFailureCount, authStepSuccessCount, facetStartTime, identityProviderType, _timestamp as batchEndTime, _timestamp
insert into perSecondRegionStreamWithUniqueId;

from perSecondRegionStreamWithUniqueId#window.uniqueExternalTimeBatch(uniqueId, batchEndTime, 60 sec, 59999 millisec, 10 sec, true)
select meta_tenantId, time:extract(batchEndTime, 'year') as year, time:extract(batchEndTime, 'month') as month, time:extract(batchEndTime, 'day') as day, time:extract(batchEndTime, 'hour') as hour, time:extract(batchEndTime, 'minute') as minute,remoteIp,region,sum(authSuccessCount) as authSuccessCount, sum(authFailureCount) as authFailureCount, sum(authStepSuccessCount) as authStepSuccessCount, convert(isAnalytics:getMinuteStartingTime(batchEndTime), 'string') as facetStartTime, identityProviderType, isAnalytics:getMinuteStartingTime(batchEndTime) as _timestamp
group by meta_tenantId, remoteIp,identityProviderType
insert into PerMinRegionStream;
