So in the LimitPushDown optimizer, the limit context represents the max number of rows that given operator needs to produce, which can default to Long.MAX_VALUE if there is no limit. With that in mind, two things:

1) You should only add a DistinctLimitNode if the limit is not equal to Long.MAX_VALUE
2) Assuming that we can add a DisitincLimitNode, the limit being pushed down should be absorbed at this point and not further propagated. This means that when you rewrite the source, you should not pass "limit" as the context, but rather Long.MAX_VALUE. This also means that when you pass the limit to the DistinctLimitNode, the value should not just be node.getCount(), but rather Math.min(node.getCount(), limit).

Two things:

1) You should only introduce a 

You shouldn't rewrite the source.getSource with "limit" here. Otherwise, this will have the effect of creating a LimitDistinctNode and propagating the limit even further down the tree (which is not semantically equivalent). Instead, because the LimitDistinctNode is as far as the current limit