public void contributeRuntimeOperator()

in algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java [111:213]
98 lines of code
14 McCabe index (conditional complexity)

    public void contributeRuntimeOperator(IHyracksJobBuilder builder, JobGenContext context, ILogicalOperator op,
            IOperatorSchema propagatedSchema, IOperatorSchema[] inputSchemas, IOperatorSchema outerPlanSchema)
                    throws AlgebricksException {
        int[] keysLeft = JobGenHelper.variablesToFieldIndexes(keysLeftBranch, inputSchemas[0]);
        int[] keysRight = JobGenHelper.variablesToFieldIndexes(keysRightBranch, inputSchemas[1]);
        IVariableTypeEnvironment env = context.getTypeEnvironment(op);
        IBinaryHashFunctionFactory[] hashFunFactories = JobGenHelper
                .variablesToBinaryHashFunctionFactories(keysLeftBranch, env, context);
        IBinaryHashFunctionFamily[] hashFunFamilies = JobGenHelper.variablesToBinaryHashFunctionFamilies(keysLeftBranch,
                env, context);
        IBinaryComparatorFactory[] comparatorFactories = new IBinaryComparatorFactory[keysLeft.length];
        int i = 0;
        IBinaryComparatorFactoryProvider bcfp = context.getBinaryComparatorFactoryProvider();
        for (LogicalVariable v : keysLeftBranch) {
            Object t = env.getVarType(v);
            comparatorFactories[i++] = bcfp.getBinaryComparatorFactory(t, true);
        }

        IPredicateEvaluatorFactoryProvider predEvaluatorFactoryProvider = context
                .getPredicateEvaluatorFactoryProvider();
        IPredicateEvaluatorFactory predEvaluatorFactory = (predEvaluatorFactoryProvider == null ? null
                : predEvaluatorFactoryProvider.getPredicateEvaluatorFactory(keysLeft, keysRight));

        RecordDescriptor recDescriptor = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op),
                propagatedSchema, context);
        IOperatorDescriptorRegistry spec = builder.getJobSpec();
        IOperatorDescriptor opDesc = null;

        boolean optimizedHashJoin = true;
        for (IBinaryHashFunctionFamily family : hashFunFamilies) {
            if (family == null) {
                optimizedHashJoin = false;
                break;
            }
        }

        if (!optimizedHashJoin) {
            try {
                switch (kind) {
                    case INNER: {
                        opDesc = new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
                                maxInputBuildSizeInFrames, aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight,
                                hashFunFactories, comparatorFactories, recDescriptor, predEvaluatorFactory, false, null);
                        break;
                    }
                    case LEFT_OUTER: {
                        INullWriterFactory[] nullWriterFactories = new INullWriterFactory[inputSchemas[1].getSize()];
                        for (int j = 0; j < nullWriterFactories.length; j++) {
                            nullWriterFactories[j] = context.getNullWriterFactory();
                        }
                        opDesc = new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
                                maxInputBuildSizeInFrames, aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight,
                                hashFunFactories, comparatorFactories, recDescriptor, predEvaluatorFactory, true,
                                nullWriterFactories);
                        break;
                    }
                    default: {
                        throw new NotImplementedException();
                    }
                }
            } catch (HyracksDataException e) {
                throw new AlgebricksException(e);
            }
        } else {
            try {
                switch (kind) {
                    case INNER: {
                        opDesc = new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
                                maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
                                comparatorFactories, recDescriptor,
                                new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
                                new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft),
                                predEvaluatorFactory);
                        break;
                    }
                    case LEFT_OUTER: {
                        INullWriterFactory[] nullWriterFactories = new INullWriterFactory[inputSchemas[1].getSize()];
                        for (int j = 0; j < nullWriterFactories.length; j++) {
                            nullWriterFactories[j] = context.getNullWriterFactory();
                        }
                        opDesc = new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
                                maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
                                comparatorFactories, recDescriptor,
                                new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
                                new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft),
                                predEvaluatorFactory, true, nullWriterFactories);
                        break;
                    }
                    default: {
                        throw new NotImplementedException();
                    }
                }
            } catch (HyracksDataException e) {
                throw new AlgebricksException(e);
            }
        }
        contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);

        ILogicalOperator src1 = op.getInputs().get(0).getValue();
        builder.contributeGraphEdge(src1, 0, op, 0);
        ILogicalOperator src2 = op.getInputs().get(1).getValue();
        builder.contributeGraphEdge(src2, 0, op, 1);
    }