public static long calculate(PhysicalRel rel, ExpressionEvalContext evalContext) {
        GcdCalculatorVisitor visitor = new GcdCalculatorVisitor(evalContext);
        visitor.go(rel);

        if (visitor.gcd == 0) {
            // there's no window aggr in the rel, return the value for joins, which is already capped at some reasonable value
            return visitor.maximumIntervalForJoins;
        }
        // if there's window aggr, cap it with the maximumIntervalForJoins
        return Math.min(visitor.gcd, visitor.maximumIntervalForJoins);
    }
@Test
    public void when_unionAboveSlidingWindows_then_returnGcdOfWindowsSize() {
        HazelcastTable table = streamGeneratorTable("s1", 1);
        HazelcastTable table2 = streamGeneratorTable("s2", 10);
        HazelcastTable table3 = streamGeneratorTable("s3", 100);
        List<QueryDataType> parameterTypes = Collections.singletonList(INT);

        int expectedGcd = 12;

        final String query = "SELECT * FROM "
                + hop("s1", expectedGcd * 4, expectedGcd)
                + " UNION ALL "
                + hop("s2", expectedGcd * 3, expectedGcd)
                + " UNION ALL "
                + hop("s3", expectedGcd * 2, expectedGcd);

        PhysicalRel optimizedPhysicalRel = optimizePhysical(query, parameterTypes, table, table2, table3).getPhysical();

        // assert plan for better visibility of what was generated
        assertPlan(optimizedPhysicalRel, plan(
                planRow(0, UnionPhysicalRel.class),

                planRow(1, SlidingWindowAggregatePhysicalRel.class),
                planRow(2, CalcPhysicalRel.class),
                planRow(3, FullScanPhysicalRel.class),

                planRow(1, SlidingWindowAggregatePhysicalRel.class),
                planRow(2, CalcPhysicalRel.class),
                planRow(3, FullScanPhysicalRel.class),

                planRow(1, SlidingWindowAggregatePhysicalRel.class),
                planRow(2, CalcPhysicalRel.class),
                planRow(3, FullScanPhysicalRel.class)
        ));

        // GCD(48, 36, 24) = 12
        assertThat(WatermarkThrottlingFrameSizeCalculator.calculate(optimizedPhysicalRel, MOCK_EEC))
                .isEqualTo(expectedGcd);
    }