[jira] [Created] (PHOENIX-3999) Optimize inner joins as SKIP-SCAN-JOIN when possible

[James Taylor (JIRA)]

James Taylor created PHOENIX-3999:
-------------------------------------

             Summary: Optimize inner joins as SKIP-SCAN-JOIN when possible
                 Key: PHOENIX-3999
                 URL: https://issues.apache.org/jira/browse/PHOENIX-3999
             Project: Phoenix
          Issue Type: Bug
            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), however inner joins do not.

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}

We should also be able to leverage this optimization when an inner join is used 
such as this:
{code}
UPSERT INTO COMPLETED_ITEMS (ITEM_TYPE, BATCH_SEQUENCE_NUM, ITEM_ID, ITEM_VALUE)
   SELECT i.ITEM_TYPE, b.BATCH_SEQUENCE_NUM, i.ITEM_ID, i.ITEM_VALUE   
   FROM  ITEMS i, COMPLETED_BATCHES b
   WHERE b.BATCH_ID = i.BATCH_ID AND          
   b.BATCH_SEQUENCE_NUM > 1000 AND b.BATCH_SEQUENCE_NUM < 2000;
{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}





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