sj1.setJoiner(SIRJoiner.createUniformRR(sj1, new JIntLiteral(O)));
JExpression[] weights = createWeights(sj, reps, O); sj1.setJoiner(SIRJoiner.createWeightedRR(sj1, weights));
    private SIRPipeline doOneTransformation(SIRSplitJoin sj) {        // get parameters from the sketch        int N = sj.size();        SIRFilter filter = (SIRFilter)sj.get(0);        int O = filter.getPopInt();        int U = filter.getPushInt();        SIRSplitter origSplit = sj.getSplitter();        SIRJoiner origJoin = sj.getJoiner();        // make new first splitjoin        SIRSplitJoin sj1 = new SIRSplitJoin(null, "CollapsedDataParallel_1");        sj1.setInit(SIRStream.makeEmptyInit());        for (int i=0; i<N; i++) {            sj1.add(new SIRIdentity(sj.getInputType()));        }        sj1.setSplitter(origSplit);        sj1.setJoiner(SIRJoiner.createUniformRR(sj1, new JIntLiteral(O)));        sj1.rescale();        // make the second splitjoin        SIRSplitJoin sj2 = new SIRSplitJoin(null, "CollapsedDataParallel_2");        sj2.setInit(SIRStream.makeEmptyInit());        for (int i=0; i<N; i++) {            sj2.add(new SIRIdentity(sj.getInputType()));        }        sj2.setSplitter(SIRSplitter.createUniformRR(sj2, new JIntLiteral(U)));        sj2.setJoiner(origJoin);        sj2.rescale();        // make the overall pipeline        SIRPipeline result = new SIRPipeline(null, "CollapsedDataParallel");        result.setInit(SIRStream.makeEmptyInit());        result.add(sj1);        result.add(filter);        result.add(sj2);        // fuse the splitjoins down into filters (it replaces them in parent)        SIRStream fused1 = FuseSimpleSplit.fuse(sj1);        SIRStream fused2 = FuseSimpleSplit.fuse(sj2);        assert !(fused1 instanceof SIRSplitJoin) : "Expected simple fusion to apply; it didn't.";        assert !(fused2 instanceof SIRSplitJoin) : "Expected simple fusion to apply; it didn't.";                return result;    }