Mailing-List: contact dev-help@phoenix.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@phoenix.apache.org
Date: Fri, 7 Jul 2017 03:07:00 +0000 (UTC)
From: "James Taylor (JIRA)" <jira@apache.org>
To: dev@phoenix.apache.org
Message-ID: <JIRA.13076670.1496355969000.190445.1499396820051@Atlassian.JIRA>
In-Reply-To: <JIRA.13076670.1496355969000@Atlassian.JIRA>
References: <JIRA.13076670.1496355969000@Atlassian.JIRA> <JIRA.13076670.1496355969584@jira-lw-us.apache.org>
Subject: [jira] [Updated] (PHOENIX-3905) Allow dynamic filtered join queries
 in UPSERT SELECT to be distributed across cluster
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Fri, 07 Jul 2017 03:07:07 -0000


     [ https://issues.apache.org/jira/browse/PHOENIX-3905?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

James Taylor updated PHOENIX-3905:
----------------------------------
    Description: 
Semi joins on the leading part of the primary key end up doing batches of point queries (as opposed to a broadcast hash join), and thus could be distributed across the cluster to improve performance when used in an UPSERT SELECT. 

Here's a set of example schemas that executes a skip scan on the inner query:
{code}
CREATE TABLE COMPLETED_BATCHES (
    BATCH_SEQUENCE_NUM BIGINT NOT NULL,
    BATCH_ID           BIGINT NOT NULL,
    CONSTRAINT PK PRIMARY KEY
    (
        BATCH_SEQUENCE_NUM,
        BATCH_ID
    )
);

CREATE TABLE ITEMS (
   BATCH_ID BIGINT NOT NULL,
   ITEM_ID BIGINT NOT NULL,
   ITEM_TYPE BIGINT,
   ITEM_VALUE VARCHAR,
   CONSTRAINT PK PRIMARY KEY
   (
        BATCH_ID,
        ITEM_ID
   )
);

CREATE TABLE COMPLETED_ITEMS (
   ITEM_TYPE          BIGINT NOT NULL,
   BATCH_SEQUENCE_NUM BIGINT NOT NULL,
   ITEM_ID            BIGINT NOT NULL,
   ITEM_VALUE         VARCHAR,
   CONSTRAINT PK PRIMARY KEY
   (
      ITEM_TYPE,
      BATCH_SEQUENCE_NUM,  
      ITEM_ID
   )
);
{code}

The explain plan of these indicate that a dynamic filter will be performed like this:
{code}
UPSERT SELECT
CLIENT PARALLEL 1-WAY FULL SCAN OVER ITEMS
    SKIP-SCAN-JOIN TABLE 0
        CLIENT PARALLEL 1-WAY RANGE SCAN OVER COMPLETED_BATCHES [1] - [2]
            SERVER FILTER BY FIRST KEY ONLY
            SERVER AGGREGATE INTO DISTINCT ROWS BY [BATCH_ID]
        CLIENT MERGE SORT
    DYNAMIC SERVER FILTER BY I.BATCH_ID IN ($8.$9)
{code}

A complete unit test looks like this:
{code}
    @Test
    public void testNestedLoopJoin() throws Exception {
        try (Connection conn = DriverManager.getConnection(getUrl())) {
            String t1="COMPLETED_BATCHES";
            String ddl1 = "CREATE TABLE " + t1 + " (\n" + 
                    "    BATCH_SEQUENCE_NUM BIGINT NOT NULL,\n" + 
                    "    BATCH_ID           BIGINT NOT NULL,\n" + 
                    "    CONSTRAINT PK PRIMARY KEY\n" + 
                    "    (\n" + 
                    "        BATCH_SEQUENCE_NUM,\n" + 
                    "        BATCH_ID\n" + 
                    "    )\n" + 
                    ")" + 
                    "";
            conn.createStatement().execute(ddl1);
            
            String t2="ITEMS";
            String ddl2 = "CREATE TABLE " + t2 + " (\n" + 
                    "   BATCH_ID BIGINT NOT NULL,\n" + 
                    "   ITEM_ID BIGINT NOT NULL,\n" + 
                    "   ITEM_TYPE BIGINT,\n" + 
                    "   ITEM_VALUE VARCHAR,\n" + 
                    "   CONSTRAINT PK PRIMARY KEY\n" + 
                    "   (\n" + 
                    "        BATCH_ID,\n" + 
                    "        ITEM_ID\n" + 
                    "   )\n" + 
                    ")";
            conn.createStatement().execute(ddl2);

            String t3="COMPLETED_ITEMS";
            String ddl3 = "CREATE TABLE " + t3 + "(\n" + 
                    "   ITEM_TYPE          BIGINT NOT NULL,\n" + 
                    "   BATCH_SEQUENCE_NUM BIGINT NOT NULL,\n" + 
                    "   ITEM_ID            BIGINT NOT NULL,\n" + 
                    "   ITEM_VALUE         VARCHAR,\n" + 
                    "   CONSTRAINT PK PRIMARY KEY\n" + 
                    "   (\n" + 
                    "      ITEM_TYPE,\n" + 
                    "      BATCH_SEQUENCE_NUM,  \n" + 
                    "      ITEM_ID\n" + 
                    "   )\n" + 
                    ")";
            conn.createStatement().execute(ddl3);

            conn.createStatement().execute("UPSERT INTO "+t1+"(BATCH_SEQUENCE_NUM, batch_id) VALUES (1,2)");
            conn.createStatement().execute("UPSERT INTO "+t1+"(BATCH_SEQUENCE_NUM, batch_id) VALUES (1,4)");

            conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (1,100, 10, 'a')");
            conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (2,200, 20, 'a')");
            conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (3,300, 10, 'a')");
            conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (4,400, 20, 'a')");
            conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (5,500, 10, 'a')");
            conn.commit();
            
            conn.setAutoCommit(true);
            String dml = "UPSERT INTO " + t3 + " (ITEM_TYPE, BATCH_SEQUENCE_NUM, ITEM_ID, ITEM_VALUE)\n" + 
                    "SELECT ITEM_TYPE, 1, ITEM_ID, ITEM_VALUE   \n" + 
                    "FROM  " + t2 + " i\n" + 
                    "WHERE EXISTS (" +
                    "  SELECT 1 FROM " + t1 + " b WHERE b.BATCH_ID = i.BATCH_ID AND " +
                    "  b.BATCH_SEQUENCE_NUM > 0 AND b.BATCH_SEQUENCE_NUM < 2)";
            conn.createStatement().execute(dml);
            ResultSet rs = conn.createStatement().executeQuery("SELECT ITEM_ID FROM " + t3);
            assertTrue(rs.next());
            assertEquals(rs.getLong(1), 200L);
            assertTrue(rs.next());
            assertEquals(rs.getLong(1), 400L);
            assertFalse(rs.next());
        }
    }
{code}

Currently, for these types of UPSERT SELECT queries, the selected data will flow back to the client and then back out to the appropriate server. It'll still be parallelized, but only on a single client as opposed to across multiple region servers in the cluster. The benefit would depend on how many regions servers would be involved in fetching the data for the select part of the query.

  was:
Joins on the leading part of the primary key end up doing batches of point queries (as opposed to a broadcast hash join), and thus could be distributed across the cluster to improve performance when used in an UPSERT SELECT. The explain plan of these indicate that a dynamic filter will be performed like this:
{code}
DYNAMIC SERVER FILTER BY (DML.PK1 DML.PK2, DML.PK3) 
IN ((COM.PK1, COM.PK2, COM.PK3))
{code}

Currently, for these types of UPSERT SELECT queries, the selected data will flow back to the client and then back out to the appropriate server. It'll still be parallelized, but only on a single client as opposed to across multiple region servers in the cluster. The benefit would depend on how many regions servers would be involved in fetching the data for the select part of the query.


> Allow dynamic filtered join queries in UPSERT SELECT to be distributed across cluster
> -------------------------------------------------------------------------------------
>
>                 Key: PHOENIX-3905
>                 URL: https://issues.apache.org/jira/browse/PHOENIX-3905
>             Project: Phoenix
>          Issue Type: Improvement
>            Reporter: James Taylor
>
> Semi joins on the leading part of the primary key end up doing batches of point queries (as opposed to a broadcast hash join), and thus could be distributed across the cluster to improve performance when used in an UPSERT SELECT. 
> Here's a set of example schemas that executes a skip scan on the inner query:
> {code}
> CREATE TABLE COMPLETED_BATCHES (
>     BATCH_SEQUENCE_NUM BIGINT NOT NULL,
>     BATCH_ID           BIGINT NOT NULL,
>     CONSTRAINT PK PRIMARY KEY
>     (
>         BATCH_SEQUENCE_NUM,
>         BATCH_ID
>     )
> );
> CREATE TABLE ITEMS (
>    BATCH_ID BIGINT NOT NULL,
>    ITEM_ID BIGINT NOT NULL,
>    ITEM_TYPE BIGINT,
>    ITEM_VALUE VARCHAR,
>    CONSTRAINT PK PRIMARY KEY
>    (
>         BATCH_ID,
>         ITEM_ID
>    )
> );
> CREATE TABLE COMPLETED_ITEMS (
>    ITEM_TYPE          BIGINT NOT NULL,
>    BATCH_SEQUENCE_NUM BIGINT NOT NULL,
>    ITEM_ID            BIGINT NOT NULL,
>    ITEM_VALUE         VARCHAR,
>    CONSTRAINT PK PRIMARY KEY
>    (
>       ITEM_TYPE,
>       BATCH_SEQUENCE_NUM,  
>       ITEM_ID
>    )
> );
> {code}
> The explain plan of these indicate that a dynamic filter will be performed like this:
> {code}
> UPSERT SELECT
> CLIENT PARALLEL 1-WAY FULL SCAN OVER ITEMS
>     SKIP-SCAN-JOIN TABLE 0
>         CLIENT PARALLEL 1-WAY RANGE SCAN OVER COMPLETED_BATCHES [1] - [2]
>             SERVER FILTER BY FIRST KEY ONLY
>             SERVER AGGREGATE INTO DISTINCT ROWS BY [BATCH_ID]
>         CLIENT MERGE SORT
>     DYNAMIC SERVER FILTER BY I.BATCH_ID IN ($8.$9)
> {code}
> A complete unit test looks like this:
> {code}
>     @Test
>     public void testNestedLoopJoin() throws Exception {
>         try (Connection conn = DriverManager.getConnection(getUrl())) {
>             String t1="COMPLETED_BATCHES";
>             String ddl1 = "CREATE TABLE " + t1 + " (\n" + 
>                     "    BATCH_SEQUENCE_NUM BIGINT NOT NULL,\n" + 
>                     "    BATCH_ID           BIGINT NOT NULL,\n" + 
>                     "    CONSTRAINT PK PRIMARY KEY\n" + 
>                     "    (\n" + 
>                     "        BATCH_SEQUENCE_NUM,\n" + 
>                     "        BATCH_ID\n" + 
>                     "    )\n" + 
>                     ")" + 
>                     "";
>             conn.createStatement().execute(ddl1);
>             
>             String t2="ITEMS";
>             String ddl2 = "CREATE TABLE " + t2 + " (\n" + 
>                     "   BATCH_ID BIGINT NOT NULL,\n" + 
>                     "   ITEM_ID BIGINT NOT NULL,\n" + 
>                     "   ITEM_TYPE BIGINT,\n" + 
>                     "   ITEM_VALUE VARCHAR,\n" + 
>                     "   CONSTRAINT PK PRIMARY KEY\n" + 
>                     "   (\n" + 
>                     "        BATCH_ID,\n" + 
>                     "        ITEM_ID\n" + 
>                     "   )\n" + 
>                     ")";
>             conn.createStatement().execute(ddl2);
>             String t3="COMPLETED_ITEMS";
>             String ddl3 = "CREATE TABLE " + t3 + "(\n" + 
>                     "   ITEM_TYPE          BIGINT NOT NULL,\n" + 
>                     "   BATCH_SEQUENCE_NUM BIGINT NOT NULL,\n" + 
>                     "   ITEM_ID            BIGINT NOT NULL,\n" + 
>                     "   ITEM_VALUE         VARCHAR,\n" + 
>                     "   CONSTRAINT PK PRIMARY KEY\n" + 
>                     "   (\n" + 
>                     "      ITEM_TYPE,\n" + 
>                     "      BATCH_SEQUENCE_NUM,  \n" + 
>                     "      ITEM_ID\n" + 
>                     "   )\n" + 
>                     ")";
>             conn.createStatement().execute(ddl3);
>             conn.createStatement().execute("UPSERT INTO "+t1+"(BATCH_SEQUENCE_NUM, batch_id) VALUES (1,2)");
>             conn.createStatement().execute("UPSERT INTO "+t1+"(BATCH_SEQUENCE_NUM, batch_id) VALUES (1,4)");
>             conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (1,100, 10, 'a')");
>             conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (2,200, 20, 'a')");
>             conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (3,300, 10, 'a')");
>             conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (4,400, 20, 'a')");
>             conn.createStatement().execute("UPSERT INTO "+t2+"(batch_id, item_id, item_type, item_value) VALUES (5,500, 10, 'a')");
>             conn.commit();
>             
>             conn.setAutoCommit(true);
>             String dml = "UPSERT INTO " + t3 + " (ITEM_TYPE, BATCH_SEQUENCE_NUM, ITEM_ID, ITEM_VALUE)\n" + 
>                     "SELECT ITEM_TYPE, 1, ITEM_ID, ITEM_VALUE   \n" + 
>                     "FROM  " + t2 + " i\n" + 
>                     "WHERE EXISTS (" +
>                     "  SELECT 1 FROM " + t1 + " b WHERE b.BATCH_ID = i.BATCH_ID AND " +
>                     "  b.BATCH_SEQUENCE_NUM > 0 AND b.BATCH_SEQUENCE_NUM < 2)";
>             conn.createStatement().execute(dml);
>             ResultSet rs = conn.createStatement().executeQuery("SELECT ITEM_ID FROM " + t3);
>             assertTrue(rs.next());
>             assertEquals(rs.getLong(1), 200L);
>             assertTrue(rs.next());
>             assertEquals(rs.getLong(1), 400L);
>             assertFalse(rs.next());
>         }
>     }
> {code}
> Currently, for these types of UPSERT SELECT queries, the selected data will flow back to the client and then back out to the appropriate server. It'll still be parallelized, but only on a single client as opposed to across multiple region servers in the cluster. The benefit would depend on how many regions servers would be involved in fetching the data for the select part of the query.


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)