Mailing-List: contact notifications-help@asterixdb.incubator.apache.org;
 run by ezmlm
Precedence: bulk
Reply-To: dev@asterixdb.incubator.apache.org
Date: Tue, 1 Dec 2015 02:12:10 +0000 (UTC)
From: "Yingyi Bu (JIRA)" <jira@apache.org>
To: notifications@asterixdb.incubator.apache.org
Message-ID: <JIRA.12917125.1448935903000.227306.1448935930963@Atlassian.JIRA>
In-Reply-To: <JIRA.12917125.1448935903000@Atlassian.JIRA>
References: <JIRA.12917125.1448935903000@Atlassian.JIRA>
 <JIRA.12917125.1448935903309@arcas>
Subject: [jira] [Created] (ASTERIXDB-1206) Hyracks level job rewriting
 results in single-thread-per-partition for binary input operators
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit

Yingyi Bu created ASTERIXDB-1206:
------------------------------------

             Summary: Hyracks level job rewriting results in single-thread-per-partition for binary input operators
                 Key: ASTERIXDB-1206
                 URL: https://issues.apache.org/jira/browse/ASTERIXDB-1206
             Project: Apache AsterixDB
          Issue Type: Bug
          Components: Hyracks Core
            Reporter: Yingyi Bu
            Assignee: Yingyi Bu


The following job will be run in a single thread for one partition.
But the two initializer -- one for each input source operator should be run in two parallel threads.

@Test
    public void testScanUnion() throws Exception {
        JobSpecification spec = new JobSpecification();

        IFileSplitProvider splitProvider1 = new ConstantFileSplitProvider(new FileSplit[] {
                new FileSplit(NC1_ID, new FileReference(new File("data/words.txt"))) });

        IFileSplitProvider splitProvider2 = new ConstantFileSplitProvider(new FileSplit[] {
                new FileSplit(NC2_ID, new FileReference(new File("data/words.txt"))) });

        RecordDescriptor desc = new RecordDescriptor(
                new ISerializerDeserializer[] { new UTF8StringSerializerDeserializer() });

        FileScanOperatorDescriptor csvScanner1 = new FileScanOperatorDescriptor(
                spec,
                splitProvider1,
                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE },
                        ','),
                desc);
        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, csvScanner1, NC1_ID);

        FileScanOperatorDescriptor csvScanner2 = new FileScanOperatorDescriptor(
                spec,
                splitProvider2,
                new DelimitedDataTupleParserFactory(new IValueParserFactory[] { UTF8StringParserFactory.INSTANCE },
                        ','),
                desc);
        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, csvScanner2, NC2_ID);

        UnionAllOperatorDescriptor union = new UnionAllOperatorDescriptor(spec, 2, desc);
        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, union, NC1_ID);

        IConnectorDescriptor connScan1ToUnion = new OneToOneConnectorDescriptor(spec);
        IConnectorDescriptor connScan2ToUnion = new OneToOneConnectorDescriptor(spec);
        spec.connect(connScan1ToUnion, csvScanner1, 0, union, 0);
        spec.connect(connScan2ToUnion, csvScanner2, 0, union, 1);

        IOperatorDescriptor printer = new PrinterOperatorDescriptor(spec);

        PartitionConstraintHelper.addAbsoluteLocationConstraint(spec, printer, NC1_ID);

        IConnectorDescriptor conn = new OneToOneConnectorDescriptor(spec);
        spec.connect(conn, union, 0, printer, 0);

        spec.addRoot(printer);
        runTest(spec);
    }


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)