| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package org.apache.hyracks.algebricks.tests.pushruntime; |
| |
| import java.io.BufferedReader; |
| import java.io.File; |
| import java.io.FileReader; |
| import java.io.IOException; |
| |
| import org.apache.hyracks.algebricks.data.IPrinterFactory; |
| import org.apache.hyracks.algebricks.data.impl.BinaryBooleanInspectorImpl; |
| import org.apache.hyracks.algebricks.data.impl.BinaryIntegerInspectorImpl; |
| import org.apache.hyracks.algebricks.data.impl.IntegerPrinterFactory; |
| import org.apache.hyracks.algebricks.data.impl.NoopMissingWriterFactory; |
| import org.apache.hyracks.algebricks.data.impl.UTF8StringPrinterFactory; |
| import org.apache.hyracks.algebricks.runtime.aggregators.TupleCountAggregateFunctionFactory; |
| import org.apache.hyracks.algebricks.runtime.aggregators.TupleCountRunningAggregateFunctionFactory; |
| import org.apache.hyracks.algebricks.runtime.base.AlgebricksPipeline; |
| import org.apache.hyracks.algebricks.runtime.base.IAggregateEvaluatorFactory; |
| import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.base.IRunningAggregateEvaluatorFactory; |
| import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory; |
| import org.apache.hyracks.algebricks.runtime.base.IUnnestingEvaluatorFactory; |
| import org.apache.hyracks.algebricks.runtime.evaluators.TupleFieldEvaluatorFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.aggreg.AggregateRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.aggreg.NestedPlansAccumulatingAggregatorFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.group.MicroPreClusteredGroupRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.meta.AlgebricksMetaOperatorDescriptor; |
| import org.apache.hyracks.algebricks.runtime.operators.meta.SubplanRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.sort.InMemorySortRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.AssignRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.EmptyTupleSourceRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.NestedTupleSourceRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.PrinterRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.RunningAggregateRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.SinkWriterRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.StreamLimitRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.StreamProjectRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.StreamSelectRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.StringStreamingRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.operators.std.UnnestRuntimeFactory; |
| import org.apache.hyracks.algebricks.runtime.writers.PrinterBasedWriterFactory; |
| import org.apache.hyracks.algebricks.tests.util.AlgebricksHyracksIntegrationUtil; |
| import org.apache.hyracks.api.constraints.PartitionConstraintHelper; |
| import org.apache.hyracks.api.dataflow.IOperatorDescriptor; |
| import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory; |
| import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory; |
| import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer; |
| import org.apache.hyracks.api.dataflow.value.RecordDescriptor; |
| import org.apache.hyracks.api.io.FileReference; |
| import org.apache.hyracks.api.job.JobSpecification; |
| import org.apache.hyracks.data.std.accessors.PointableBinaryComparatorFactory; |
| import org.apache.hyracks.data.std.primitive.IntegerPointable; |
| import org.apache.hyracks.data.std.primitive.UTF8StringPointable; |
| import org.apache.hyracks.dataflow.common.data.marshalling.FloatSerializerDeserializer; |
| import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer; |
| import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer; |
| import org.apache.hyracks.dataflow.common.data.parsers.FloatParserFactory; |
| import org.apache.hyracks.dataflow.common.data.parsers.IValueParserFactory; |
| import org.apache.hyracks.dataflow.common.data.parsers.IntegerParserFactory; |
| import org.apache.hyracks.dataflow.common.data.parsers.UTF8StringParserFactory; |
| import org.apache.hyracks.dataflow.std.connectors.OneToOneConnectorDescriptor; |
| import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider; |
| import org.apache.hyracks.dataflow.std.file.DelimitedDataTupleParserFactory; |
| import org.apache.hyracks.dataflow.std.file.FileScanOperatorDescriptor; |
| import org.apache.hyracks.dataflow.std.file.FileSplit; |
| import org.apache.hyracks.dataflow.std.file.IFileSplitProvider; |
| import org.apache.hyracks.dataflow.std.file.LineFileWriteOperatorDescriptor; |
| import org.apache.hyracks.dataflow.std.group.preclustered.PreclusteredGroupOperatorDescriptor; |
| import org.apache.hyracks.dataflow.std.misc.SplitOperatorDescriptor; |
| import org.apache.hyracks.dataflow.std.sort.InMemorySortOperatorDescriptor; |
| import org.junit.AfterClass; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| |
| import junit.framework.Assert; |
| |
| public class PushRuntimeTest { |
| |
| private static final String SEPARATOR = System.getProperty("file.separator"); |
| private static final String PATH_ACTUAL = "rttest"; |
| private static final String PATH_BASE = "src" + SEPARATOR + "test" + SEPARATOR + "resources"; |
| private static final String PATH_EXPECTED = PATH_BASE + SEPARATOR + "results"; |
| private static final int FRAME_SIZE = 32768; |
| |
| private static final String[] DEFAULT_NODES = new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }; |
| |
| @BeforeClass |
| public static void setUp() throws Exception { |
| File outdir = new File(PATH_ACTUAL); |
| outdir.mkdirs(); |
| AlgebricksHyracksIntegrationUtil.init(); |
| } |
| |
| @AfterClass |
| public static void tearDown() throws Exception { |
| AlgebricksHyracksIntegrationUtil.deinit(); |
| File outdir = new File(PATH_ACTUAL); |
| File[] files = outdir.listFiles(); |
| if (files == null || files.length == 0) { |
| outdir.delete(); |
| } |
| } |
| |
| @Test |
| public void etsAssignPrint() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| IntegerConstantEvalFactory const1 = new IntegerConstantEvalFactory(400); |
| IntegerConstantEvalFactory const2 = new IntegerConstantEvalFactory(3); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| AssignRuntimeFactory assign = new AssignRuntimeFactory(new int[] { 0, 1 }, |
| new IScalarEvaluatorFactory[] { const1, const2 }, new int[] { 0, 1 }); |
| RecordDescriptor assignDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| PrinterRuntimeFactory printer = new PrinterRuntimeFactory(new int[] { 0, 1 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE, IntegerPrinterFactory.INSTANCE }, assignDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, assign, printer }, |
| new RecordDescriptor[] { etsDesc, assignDesc, null }); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, DEFAULT_NODES); |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| } |
| |
| @Test |
| public void etsAssignWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| IntegerConstantEvalFactory const1 = new IntegerConstantEvalFactory(400); |
| IntegerConstantEvalFactory const2 = new IntegerConstantEvalFactory(3); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| AssignRuntimeFactory assign = new AssignRuntimeFactory(new int[] { 0, 1 }, |
| new IScalarEvaluatorFactory[] { const1, const2 }, new int[] { 0, 1 }); |
| RecordDescriptor assignDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsAssignWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0, 1 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE, IntegerPrinterFactory.INSTANCE }, outFile, |
| PrinterBasedWriterFactory.INSTANCE, assignDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, assign, writer }, |
| new RecordDescriptor[] { etsDesc, assignDesc, null }); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, DEFAULT_NODES); |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("400; 3", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanSelectWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] intFileSplits = new FileSplit[1]; |
| intFileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/simple/int-part1.tbl"))); |
| IFileSplitProvider intSplitProvider = new ConstantFileSplitProvider(intFileSplits); |
| RecordDescriptor intScannerDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor intScanner = new FileScanOperatorDescriptor(spec, intSplitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), intScannerDesc); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, intScanner, DEFAULT_NODES); |
| |
| // the algebricks op. |
| IScalarEvaluatorFactory cond = new IntegerGreaterThanEvalFactory(new IntegerConstantEvalFactory(2), |
| new TupleFieldEvaluatorFactory(0)); |
| StreamSelectRuntimeFactory select = new StreamSelectRuntimeFactory(cond, new int[] { 0 }, |
| BinaryBooleanInspectorImpl.FACTORY, false, -1, null); |
| RecordDescriptor selectDesc = intScannerDesc; |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "scanSelectWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| selectDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { select, writer }, new RecordDescriptor[] { selectDesc, null }); |
| |
| PartitionConstraintHelper.addPartitionCountConstraint(spec, algebricksOp, 1); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), intScanner, 0, algebricksOp, 0); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("0", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void etsAssignProjectWrite() throws Exception { |
| |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| IntegerConstantEvalFactory const1 = new IntegerConstantEvalFactory(400); |
| IntegerConstantEvalFactory const2 = new IntegerConstantEvalFactory(3); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| AssignRuntimeFactory assign = new AssignRuntimeFactory(new int[] { 0, 1 }, |
| new IScalarEvaluatorFactory[] { const1, const2 }, new int[] { 0, 1 }); |
| RecordDescriptor assignDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| StreamProjectRuntimeFactory project = new StreamProjectRuntimeFactory(new int[] { 1 }); |
| RecordDescriptor projectDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsAssignProjectWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| projectDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, assign, project, writer }, |
| new RecordDescriptor[] { etsDesc, assignDesc, projectDesc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, DEFAULT_NODES); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("3", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanLimitWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] fileSplits = new FileSplit[1]; |
| fileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/tpch0.001/customer.tbl"))); |
| IFileSplitProvider splitProvider = new ConstantFileSplitProvider(fileSplits); |
| |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer(), |
| new UTF8StringSerializerDeserializer(), IntegerSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), FloatSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), new UTF8StringSerializerDeserializer() }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor scanner = new FileScanOperatorDescriptor(spec, splitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), scannerDesc); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanner, DEFAULT_NODES); |
| |
| // the algebricks op. |
| StreamLimitRuntimeFactory limit = new StreamLimitRuntimeFactory(new IntegerConstantEvalFactory(2), null, |
| new int[] { 0 }, BinaryIntegerInspectorImpl.FACTORY); |
| RecordDescriptor limitDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "scanLimitWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| limitDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { limit, writer }, new RecordDescriptor[] { limitDesc, null }); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanner, 0, algebricksOp, 0); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("12", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void etsUnnestWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| IUnnestingEvaluatorFactory aggregFactory = new IntArrayUnnester(new int[] { 100, 200, 300 }); |
| UnnestRuntimeFactory unnest = new UnnestRuntimeFactory(0, aggregFactory, new int[] { 0 }, false, null); |
| RecordDescriptor unnestDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsUnnestWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| unnestDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, unnest, writer }, |
| new RecordDescriptor[] { etsDesc, unnestDesc, null }); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("100200300", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanAggregateWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] fileSplits = new FileSplit[1]; |
| fileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/tpch0.001/customer-part1.tbl"))); |
| IFileSplitProvider splitProvider = new ConstantFileSplitProvider(fileSplits); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer(), |
| new UTF8StringSerializerDeserializer(), IntegerSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), FloatSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), new UTF8StringSerializerDeserializer() }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor scanner = new FileScanOperatorDescriptor(spec, splitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), scannerDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanner, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the algebricks op. |
| AggregateRuntimeFactory agg = new AggregateRuntimeFactory( |
| new IAggregateEvaluatorFactory[] { new TupleCountAggregateFunctionFactory() }); |
| RecordDescriptor aggDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "scanAggregateWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| aggDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { agg, writer }, new RecordDescriptor[] { aggDesc, null }); |
| |
| PartitionConstraintHelper.addPartitionCountConstraint(spec, algebricksOp, 1); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanner, 0, algebricksOp, 0); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("75", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanSortGbySelectWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] fileSplits = new FileSplit[1]; |
| fileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/tpch0.001/customer.tbl"))); |
| IFileSplitProvider splitProvider = new ConstantFileSplitProvider(fileSplits); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer(), |
| new UTF8StringSerializerDeserializer(), IntegerSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), FloatSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), new UTF8StringSerializerDeserializer() }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor scanner = new FileScanOperatorDescriptor(spec, splitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), scannerDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanner, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the sort (by nation id) |
| RecordDescriptor sortDesc = scannerDesc; |
| InMemorySortOperatorDescriptor sort = new InMemorySortOperatorDescriptor(spec, new int[] { 3 }, |
| new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY) }, |
| sortDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, sort, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the group-by |
| NestedTupleSourceRuntimeFactory nts = new NestedTupleSourceRuntimeFactory(); |
| RecordDescriptor ntsDesc = sortDesc; |
| AggregateRuntimeFactory agg = new AggregateRuntimeFactory( |
| new IAggregateEvaluatorFactory[] { new TupleCountAggregateFunctionFactory() }); |
| RecordDescriptor aggDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| AlgebricksPipeline pipeline = new AlgebricksPipeline(new IPushRuntimeFactory[] { nts, agg }, |
| new RecordDescriptor[] { ntsDesc, aggDesc }); |
| NestedPlansAccumulatingAggregatorFactory npaaf = new NestedPlansAccumulatingAggregatorFactory( |
| new AlgebricksPipeline[] { pipeline }, new int[] { 3 }, new int[] {}); |
| RecordDescriptor gbyDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| PreclusteredGroupOperatorDescriptor gby = new PreclusteredGroupOperatorDescriptor(spec, new int[] { 3 }, |
| new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY) }, npaaf, |
| gbyDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, gby, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the algebricks op. |
| IScalarEvaluatorFactory cond = new IntegerEqualsEvalFactory(new IntegerConstantEvalFactory(3), |
| new TupleFieldEvaluatorFactory(0)); // Canadian customers |
| StreamSelectRuntimeFactory select = new StreamSelectRuntimeFactory(cond, new int[] { 1 }, |
| BinaryBooleanInspectorImpl.FACTORY, false, -1, null); |
| RecordDescriptor selectDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "scanSortGbySelectWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| selectDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { select, writer }, new RecordDescriptor[] { selectDesc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanner, 0, sort, 0); |
| spec.connect(new OneToOneConnectorDescriptor(spec), sort, 0, gby, 0); |
| spec.connect(new OneToOneConnectorDescriptor(spec), gby, 0, algebricksOp, 0); |
| spec.addRoot(algebricksOp); |
| |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("9", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void etsUnnestRunningaggregateWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| IUnnestingEvaluatorFactory aggregFactory = new IntArrayUnnester(new int[] { 100, 200, 300 }); |
| UnnestRuntimeFactory unnest = new UnnestRuntimeFactory(0, aggregFactory, new int[] { 0 }, false, null); |
| RecordDescriptor unnestDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| RunningAggregateRuntimeFactory ragg = new RunningAggregateRuntimeFactory(new int[] { 1 }, |
| new IRunningAggregateEvaluatorFactory[] { new TupleCountRunningAggregateFunctionFactory() }, |
| new int[] { 0, 1 }); |
| RecordDescriptor raggDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsUnnestRunningaggregateWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 1 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| raggDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, unnest, ragg, writer }, |
| new RecordDescriptor[] { etsDesc, unnestDesc, raggDesc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("123", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void etsAssignScriptWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| IntegerConstantEvalFactory const1 = new IntegerConstantEvalFactory(400); |
| IntegerConstantEvalFactory const2 = new IntegerConstantEvalFactory(3); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| AssignRuntimeFactory assign = new AssignRuntimeFactory(new int[] { 0, 1 }, |
| new IScalarEvaluatorFactory[] { const1, const2 }, new int[] { 0, 1 }); |
| RecordDescriptor assignDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| IValueParserFactory[] valueParsers = { IntegerParserFactory.INSTANCE, IntegerParserFactory.INSTANCE }; |
| |
| String osname = System.getProperty("os.name"); |
| String command; |
| if (osname.equals("Linux")) { |
| command = "bash target/testscripts/idscript"; |
| } else if (osname.startsWith("Windows")) { |
| command = "target\\testscripts\\idscript.cmd"; |
| } else { |
| // don't know how to test |
| return; |
| } |
| |
| StringStreamingRuntimeFactory script = new StringStreamingRuntimeFactory(command, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE, IntegerPrinterFactory.INSTANCE }, ' ', |
| new DelimitedDataTupleParserFactory(valueParsers, ' ')); |
| RecordDescriptor scriptDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsAssignScriptWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0, 1 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE, IntegerPrinterFactory.INSTANCE }, outFile, |
| PrinterBasedWriterFactory.INSTANCE, scriptDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, assign, script, writer }, |
| new RecordDescriptor[] { etsDesc, assignDesc, scriptDesc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("400; 3", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanSplitWrite() throws Exception { |
| final int outputArity = 2; |
| |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| String inputFileName = "data/tpch0.001/customer.tbl"; |
| File inputFile = new File(inputFileName); |
| File[] outputFile = new File[outputArity]; |
| for (int i = 0; i < outputArity; i++) { |
| outputFile[i] = File.createTempFile("splitop", null); |
| } |
| |
| FileSplit[] inputSplits = new FileSplit[] { |
| new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, new FileReference(inputFile)) }; |
| |
| DelimitedDataTupleParserFactory stringParser = new DelimitedDataTupleParserFactory( |
| new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE }, '\u0000'); |
| RecordDescriptor stringRec = new RecordDescriptor( |
| new ISerializerDeserializer[] { new UTF8StringSerializerDeserializer(), }); |
| |
| FileScanOperatorDescriptor scanOp = new FileScanOperatorDescriptor(spec, |
| new ConstantFileSplitProvider(inputSplits), stringParser, stringRec); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| SplitOperatorDescriptor splitOp = new SplitOperatorDescriptor(spec, stringRec, outputArity); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, splitOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| IOperatorDescriptor outputOp[] = new IOperatorDescriptor[outputFile.length]; |
| for (int i = 0; i < outputArity; i++) { |
| outputOp[i] = new LineFileWriteOperatorDescriptor(spec, new FileSplit[] { |
| new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, new FileReference(outputFile[i])) }); |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, outputOp[i], |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| } |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanOp, 0, splitOp, 0); |
| for (int i = 0; i < outputArity; i++) { |
| spec.connect(new OneToOneConnectorDescriptor(spec), splitOp, i, outputOp[i], 0); |
| } |
| |
| for (int i = 0; i < outputArity; i++) { |
| spec.addRoot(outputOp[i]); |
| } |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| for (int i = 0; i < outputArity; i++) { |
| compareFiles(inputFileName, outputFile[i].getAbsolutePath()); |
| } |
| } |
| |
| @Test |
| public void scanMicroSortWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] fileSplits = new FileSplit[1]; |
| fileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/tpch0.001/nation.tbl"))); |
| IFileSplitProvider splitProvider = new ConstantFileSplitProvider(fileSplits); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer(), |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer() }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor scanner = new FileScanOperatorDescriptor(spec, splitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), scannerDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanner, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the algebricks op. |
| InMemorySortRuntimeFactory sort = new InMemorySortRuntimeFactory(new int[] { 1 }, null, |
| new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(UTF8StringPointable.FACTORY) }, |
| null); |
| RecordDescriptor sortDesc = scannerDesc; |
| |
| String fileName = "scanMicroSortWrite.out"; |
| String filePath = PATH_ACTUAL + SEPARATOR + fileName; |
| String resultFilePath = PATH_EXPECTED + SEPARATOR + fileName; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0, 1, 2, 3 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE, UTF8StringPrinterFactory.INSTANCE, |
| IntegerPrinterFactory.INSTANCE, UTF8StringPrinterFactory.INSTANCE }, |
| outFile, PrinterBasedWriterFactory.INSTANCE, sortDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { sort, writer }, new RecordDescriptor[] { sortDesc, null }); |
| |
| PartitionConstraintHelper.addPartitionCountConstraint(spec, algebricksOp, 1); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanner, 0, algebricksOp, 0); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| compareFiles(filePath, resultFilePath); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void etsAssignSubplanProjectWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| IntegerConstantEvalFactory const1 = new IntegerConstantEvalFactory(400); |
| IntegerConstantEvalFactory const2 = new IntegerConstantEvalFactory(3); |
| |
| EmptyTupleSourceRuntimeFactory ets = new EmptyTupleSourceRuntimeFactory(); |
| RecordDescriptor etsDesc = new RecordDescriptor(new ISerializerDeserializer[] {}); |
| |
| AssignRuntimeFactory assign1 = new AssignRuntimeFactory(new int[] { 0 }, |
| new IScalarEvaluatorFactory[] { const1 }, new int[] { 0 }); |
| RecordDescriptor assign1Desc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| NestedTupleSourceRuntimeFactory nts = new NestedTupleSourceRuntimeFactory(); |
| |
| AssignRuntimeFactory assign2 = new AssignRuntimeFactory(new int[] { 1 }, |
| new IScalarEvaluatorFactory[] { new IntegerAddEvalFactory(new TupleFieldEvaluatorFactory(0), const2) }, |
| new int[] { 0, 1 }); |
| RecordDescriptor assign2Desc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| StreamProjectRuntimeFactory project1 = new StreamProjectRuntimeFactory(new int[] { 1 }); |
| RecordDescriptor project1Desc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| AlgebricksPipeline pipeline = new AlgebricksPipeline(new IPushRuntimeFactory[] { nts, assign2, project1 }, |
| new RecordDescriptor[] { assign1Desc, assign2Desc, project1Desc }); |
| |
| SubplanRuntimeFactory subplan = new SubplanRuntimeFactory(pipeline, |
| new IMissingWriterFactory[] { NoopMissingWriterFactory.INSTANCE }, assign1Desc, null); |
| |
| RecordDescriptor subplanDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| |
| StreamProjectRuntimeFactory project2 = new StreamProjectRuntimeFactory(new int[] { 1 }); |
| RecordDescriptor project2Desc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "etsAssignSubplanProjectWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| project2Desc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 0, 0, |
| new IPushRuntimeFactory[] { ets, assign1, subplan, project2, writer }, |
| new RecordDescriptor[] { etsDesc, assign1Desc, subplanDesc, project2Desc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, DEFAULT_NODES); |
| |
| spec.addRoot(algebricksOp); |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("403", buf.toString()); |
| outFile.delete(); |
| } |
| |
| @Test |
| public void scanMicroSortGbySelectWrite() throws Exception { |
| JobSpecification spec = new JobSpecification(FRAME_SIZE); |
| |
| // the scanner |
| FileSplit[] fileSplits = new FileSplit[1]; |
| fileSplits[0] = new FileSplit(AlgebricksHyracksIntegrationUtil.NC1_ID, |
| new FileReference(new File("data/tpch0.001/customer.tbl"))); |
| IFileSplitProvider splitProvider = new ConstantFileSplitProvider(fileSplits); |
| RecordDescriptor scannerDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer(), |
| new UTF8StringSerializerDeserializer(), IntegerSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), FloatSerializerDeserializer.INSTANCE, |
| new UTF8StringSerializerDeserializer(), new UTF8StringSerializerDeserializer() }); |
| IValueParserFactory[] valueParsers = new IValueParserFactory[] { IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, IntegerParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE, FloatParserFactory.INSTANCE, UTF8StringParserFactory.INSTANCE, |
| UTF8StringParserFactory.INSTANCE }; |
| FileScanOperatorDescriptor scanner = new FileScanOperatorDescriptor(spec, splitProvider, |
| new DelimitedDataTupleParserFactory(valueParsers, '|'), scannerDesc); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, scanner, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| // the sort (by nation id) |
| RecordDescriptor sortDesc = scannerDesc; |
| InMemorySortRuntimeFactory sort = new InMemorySortRuntimeFactory(new int[] { 3 }, null, |
| new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY) }, null); |
| |
| // the group-by |
| NestedTupleSourceRuntimeFactory nts = new NestedTupleSourceRuntimeFactory(); |
| RecordDescriptor ntsDesc = sortDesc; |
| AggregateRuntimeFactory agg = new AggregateRuntimeFactory( |
| new IAggregateEvaluatorFactory[] { new TupleCountAggregateFunctionFactory() }); |
| RecordDescriptor aggDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| AlgebricksPipeline pipeline = new AlgebricksPipeline(new IPushRuntimeFactory[] { nts, agg }, |
| new RecordDescriptor[] { ntsDesc, aggDesc }); |
| NestedPlansAccumulatingAggregatorFactory npaaf = new NestedPlansAccumulatingAggregatorFactory( |
| new AlgebricksPipeline[] { pipeline }, new int[] { 3 }, new int[] {}); |
| RecordDescriptor gbyDesc = new RecordDescriptor(new ISerializerDeserializer[] { |
| IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE }); |
| MicroPreClusteredGroupRuntimeFactory gby = new MicroPreClusteredGroupRuntimeFactory(new int[] { 3 }, |
| new IBinaryComparatorFactory[] { PointableBinaryComparatorFactory.of(IntegerPointable.FACTORY) }, npaaf, |
| sortDesc, gbyDesc, null); |
| |
| // the algebricks op. |
| IScalarEvaluatorFactory cond = new IntegerEqualsEvalFactory(new IntegerConstantEvalFactory(3), |
| new TupleFieldEvaluatorFactory(0)); // Canadian customers |
| StreamSelectRuntimeFactory select = new StreamSelectRuntimeFactory(cond, new int[] { 1 }, |
| BinaryBooleanInspectorImpl.FACTORY, false, -1, null); |
| RecordDescriptor selectDesc = new RecordDescriptor( |
| new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE }); |
| |
| String filePath = PATH_ACTUAL + SEPARATOR + "scanSortGbySelectWrite.out"; |
| File outFile = new File(filePath); |
| SinkWriterRuntimeFactory writer = new SinkWriterRuntimeFactory(new int[] { 0 }, |
| new IPrinterFactory[] { IntegerPrinterFactory.INSTANCE }, outFile, PrinterBasedWriterFactory.INSTANCE, |
| selectDesc); |
| |
| AlgebricksMetaOperatorDescriptor algebricksOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 0, |
| new IPushRuntimeFactory[] { sort, gby, select, writer }, |
| new RecordDescriptor[] { sortDesc, gbyDesc, selectDesc, null }); |
| |
| PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, algebricksOp, |
| new String[] { AlgebricksHyracksIntegrationUtil.NC1_ID }); |
| |
| spec.connect(new OneToOneConnectorDescriptor(spec), scanner, 0, algebricksOp, 0); |
| spec.addRoot(algebricksOp); |
| |
| AlgebricksHyracksIntegrationUtil.runJob(spec); |
| StringBuilder buf = new StringBuilder(); |
| readFileToString(outFile, buf); |
| Assert.assertEquals("9", buf.toString()); |
| outFile.delete(); |
| } |
| |
| private static void readFileToString(File file, StringBuilder buf) throws Exception { |
| BufferedReader result = new BufferedReader(new FileReader(file)); |
| boolean first = true; |
| while (true) { |
| String s = result.readLine(); |
| if (s == null) { |
| break; |
| } else { |
| if (!first) { |
| first = false; |
| buf.append('\n'); |
| } |
| buf.append(s); |
| } |
| } |
| result.close(); |
| } |
| |
| public void compareFiles(String fileNameA, String fileNameB) throws IOException { |
| BufferedReader fileA = new BufferedReader(new FileReader(fileNameA)); |
| BufferedReader fileB = new BufferedReader(new FileReader(fileNameB)); |
| |
| String lineA, lineB; |
| while ((lineA = fileA.readLine()) != null) { |
| lineB = fileB.readLine(); |
| Assert.assertEquals(lineA, lineB); |
| } |
| Assert.assertNull(fileB.readLine()); |
| } |
| |
| } |
