PHOENIX-7054 Shallow grammar support for DROP CDC and ALTER CDC (apa…

…che#1694)

phoenix-core/src/main/antlr3/PhoenixSQL.g

@@ -439,8 +439,10 @@ oneStatement returns [BindableStatement ret]
     |   s=declare_cursor_node
     |   s=drop_table_node
     |   s=drop_index_node
+    |   s=drop_cdc_node
     |   s=alter_index_node
     |   s=alter_table_node
+    |   s=alter_cdc_node
     |   s=show_node
     |   s=show_create_table_node
     |   s=trace_node
@@ -687,13 +689,26 @@ drop_index_node returns [DropIndexStatement ret]
       {ret = factory.dropIndex(i, t, ex!=null); }
     ;
 
+// Parse a drop CDC statement
+drop_cdc_node returns [DropCDCStatement ret]
+   : DROP CDC (IF ex=EXISTS)? o=cdc_name ON t=from_table_name
+     {ret = factory.dropCDC(o, t, ex!=null); }
+   ;
+
 // Parse a alter index statement
 alter_index_node returns [AlterIndexStatement ret]
     : ALTER INDEX (IF ex=EXISTS)? i=index_name ON t=from_table_name
       ((s=(USABLE | UNUSABLE | REBUILD (isRebuildAll=ALL)? | DISABLE | ACTIVE)) (async=ASYNC)? ((SET?)p=fam_properties)?)
       {ret = factory.alterIndex(factory.namedTable(null, TableName.create(t.getSchemaName(), i.getName())), t.getTableName(), ex!=null, PIndexState.valueOf(SchemaUtil.normalizeIdentifier(s.getText())), isRebuildAll!=null, async!=null, p); }
     ;
 
+// Parse a alter CDC statement
+alter_cdc_node returns [AlterCDCStatement ret]
+   : ALTER CDC (IF ex=EXISTS)? o=cdc_name ON t=from_table_name
+     ((SET?) p=fam_properties)?
+     {ret = factory.alterCDC(factory.namedTable(null, TableName.create(t.getSchemaName(), o.getName())), t.getTableName(), ex!=null, p); }
+   ;
+
 // Parse a trace statement.
 trace_node returns [TraceStatement ret]
     :   TRACE ((flag = ON  ( WITH SAMPLING s = sampling_rate)?) | flag = OFF)

phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java

@@ -121,6 +121,7 @@
 import org.apache.phoenix.log.QueryStatus;
 import org.apache.phoenix.monitoring.TableMetricsManager;
 import org.apache.phoenix.optimize.Cost;
+import org.apache.phoenix.optimize.Cost;
 import org.apache.phoenix.parse.AddColumnStatement;
 import org.apache.phoenix.parse.AddJarsStatement;
 import org.apache.phoenix.parse.AliasedNode;
@@ -182,6 +183,8 @@
 import org.apache.phoenix.parse.UpdateStatisticsStatement;
 import org.apache.phoenix.parse.UpsertStatement;
 import org.apache.phoenix.parse.UseSchemaStatement;
+import org.apache.phoenix.parse.AlterCDCStatement;
+import org.apache.phoenix.parse.DropCDCStatement;
 import org.apache.phoenix.query.HBaseFactoryProvider;
 import org.apache.phoenix.query.KeyRange;
 import org.apache.phoenix.query.QueryConstants;
@@ -1569,6 +1572,19 @@ public MutationState execute() throws SQLException {
         }
     }
 
+    private static class ExecutableDropCDCStatement extends DropCDCStatement implements CompilableStatement {
+
+        public ExecutableDropCDCStatement(NamedNode cdcObjName, TableName tableName, boolean ifExists) {
+            super(cdcObjName, tableName, ifExists);
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException {
+            return null;
+        }
+    }
+
     private static class ExecutableAlterIndexStatement extends AlterIndexStatement implements CompilableStatement {
 
         public ExecutableAlterIndexStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async, ListMultimap<String,Pair<String,Object>> props) {
@@ -1593,6 +1609,19 @@ public MutationState execute() throws SQLException {
             };
         }
     }
+
+    private static class ExecutableAlterCDCStatement extends AlterCDCStatement implements CompilableStatement {
+
+        public ExecutableAlterCDCStatement(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, ListMultimap<String,Pair<String,Object>> props) {
+            super(indexTableNode, dataTableName, ifExists, props);
+        }
+
+        @SuppressWarnings("unchecked")
+        @Override
+        public MutationPlan compilePlan(final PhoenixStatement stmt, Sequence.ValueOp seqAction) throws SQLException {
+            return null;
+        }
+    }
 
     private static class ExecutableTraceStatement extends TraceStatement implements CompilableStatement {
 
@@ -1952,11 +1981,21 @@ public DropIndexStatement dropIndex(NamedNode indexName, TableName tableName, bo
             return new ExecutableDropIndexStatement(indexName, tableName, ifExists);
         }
 
+        @Override
+        public DropCDCStatement dropCDC(NamedNode cdcObjName, TableName tableName, boolean ifExists) {
+            return new ExecutableDropCDCStatement(cdcObjName, tableName, ifExists);
+        }
+
         @Override
         public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async, ListMultimap<String,Pair<String,Object>> props) {
             return new ExecutableAlterIndexStatement(indexTableNode, dataTableName, ifExists, state, isRebuildAll, async, props);
         }
 
+        @Override
+        public AlterCDCStatement alterCDC(NamedTableNode cdcTableNode, String dataTableName, boolean ifExist, ListMultimap<String,Pair<String,Object>> props) {
+            return new ExecutableAlterCDCStatement(cdcTableNode, dataTableName, ifExist, props);
+        }
+
         @Override
         public TraceStatement trace(boolean isTraceOn, double samplingRate) {
             return new ExecutableTraceStatement(isTraceOn, samplingRate);

phoenix-core/src/main/java/org/apache/phoenix/parse/AlterCDCStatement.java

@@ -0,0 +1,58 @@
+/*
+ * 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.phoenix.parse;
+
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ImmutableListMultimap;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
+
+public class AlterCDCStatement extends SingleTableStatement {
+    private final String dataTableName;
+    private final boolean ifExists;
+    private ListMultimap<String, Pair<String,Object>> props;
+    private static final PTableType tableType=PTableType.CDC;
+
+    public AlterCDCStatement(NamedTableNode cdcTableNode, String dataTableName, boolean ifExists) {
+        this(cdcTableNode,dataTableName,ifExists,null);
+    }
+
+    public AlterCDCStatement(NamedTableNode cdcTableNode, String dataTableName, boolean ifExists, ListMultimap<String,Pair<String,Object>> props) {
+        super(cdcTableNode,0);
+        this.dataTableName = dataTableName;
+        this.ifExists = ifExists;
+        this.props= props==null ? ImmutableListMultimap.<String,Pair<String,Object>>of() : props;
+    }
+
+    public String getTableName() {
+        return dataTableName;
+    }
+
+    @Override
+    public int getBindCount() {
+        return 0;
+    }
+
+    public boolean ifExists() {
+        return ifExists;
+    }
+
+    public ListMultimap<String,Pair<String,Object>> getProps() { return props; }
+
+    public PTableType getTableType(){ return tableType; }
+}

phoenix-core/src/main/java/org/apache/phoenix/parse/DropCDCStatement.java

@@ -0,0 +1,54 @@
+/*
+ * 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.phoenix.parse;
+
+import org.apache.phoenix.jdbc.PhoenixStatement;
+
+public class DropCDCStatement extends MutableStatement {
+    private final TableName tableName;
+    private final NamedNode cdcObjName;
+    private final boolean ifExists;
+
+    public DropCDCStatement(NamedNode cdcObjName, TableName tableName, boolean ifExists) {
+        this.cdcObjName = cdcObjName;
+        this.tableName = tableName;
+        this.ifExists = ifExists;
+    }
+
+    public TableName getTableName() {
+        return tableName;
+    }
+
+    public NamedNode getCdcObjName() {
+        return cdcObjName;
+    }
+
+    @Override
+    public int getBindCount() {
+        return 0;
+    }
+
+    public boolean ifExists() {
+        return ifExists;
+    }
+
+    @Override
+    public PhoenixStatement.Operation getOperation() {
+        return PhoenixStatement.Operation.DELETE;
+    }
+}

phoenix-core/src/main/java/org/apache/phoenix/parse/ParseNodeFactory.java

@@ -435,6 +435,10 @@ public DropIndexStatement dropIndex(NamedNode indexName, TableName tableName, bo
         return new DropIndexStatement(indexName, tableName, ifExists);
     }
 
+    public DropCDCStatement dropCDC(NamedNode cdcObjName, TableName tableName, boolean ifExists) {
+        return new DropCDCStatement(cdcObjName, tableName, ifExists);
+    }
+
     public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String dataTableName, boolean ifExists, PIndexState state, boolean isRebuildAll, boolean async, ListMultimap<String,Pair<String,Object>> props) {
         return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, isRebuildAll, async, props);
     }
@@ -443,6 +447,14 @@ public AlterIndexStatement alterIndex(NamedTableNode indexTableNode, String data
         return new AlterIndexStatement(indexTableNode, dataTableName, ifExists, state, false, false);
     }
 
+    public AlterCDCStatement alterCDC(NamedTableNode cdcTableNode, String dataTableName, boolean ifExist) {
+        return new AlterCDCStatement(cdcTableNode, dataTableName, ifExist);
+    }
+
+    public AlterCDCStatement alterCDC(NamedTableNode cdcTableNode, String dataTableName, boolean ifExist, ListMultimap<String,Pair<String,Object>> props) {
+        return new AlterCDCStatement(cdcTableNode, dataTableName, ifExist, props);
+    }
+
     public TraceStatement trace(boolean isTraceOn, double samplingRate) {
         return new TraceStatement(isTraceOn, samplingRate);
     }

phoenix-core/src/main/java/org/apache/phoenix/schema/PTableType.java

@@ -28,6 +28,7 @@ public enum PTableType {
     VIEW("v", "VIEW"),
     INDEX("i", "INDEX"),
     PROJECTED("p", "PROJECTED"),
+    CDC("c", "CDC"),
     SUBQUERY("q", "SUBQUERY"); 
 
     private final PName value;

phoenix-core/src/test/java/org/apache/phoenix/parse/QueryParserTest.java

@@ -571,7 +571,71 @@ public void testCreateSequence() throws Exception {
                         "increment by 1\n"));
         parseQuery(sql);
     }
-
+
+    private DropCDCStatement parseDropCDCSimple(String sql, boolean ifNotExists) throws Exception {
+        DropCDCStatement stmt = parseQuery(sql, DropCDCStatement.class);
+        assertEquals("FOO", stmt.getCdcObjName().getName());
+        assertEquals("BAR", stmt.getTableName().getTableName());
+        assertEquals(ifNotExists, stmt.ifExists());
+        return stmt;
+    }
+    @Test
+    public void testDropCDCSimple() throws Exception {
+        DropCDCStatement stmt = null;
+        parseDropCDCSimple("drop cdc foo on bar", false);
+        parseDropCDCSimple("drop cdc if exists foo on bar", true);
+        parseDropCDCSimple("drop cdc if exists foo on s.bar", true);
+    }
+
+    private void parseInvalidDropCDC(String sql, int expRrrorCode) throws IOException {
+        try {
+            parseQuery(sql);
+            fail();
+        }
+        catch (SQLException e) {
+            assertEquals(expRrrorCode, e.getErrorCode());
+        }
+    }
+
+    @Test
+    public void testInvalidDropCDC() throws Exception {
+        parseInvalidDropCDC("drop cdc foo bar", SQLExceptionCode.MISSING_TOKEN.getErrorCode());
+        parseInvalidDropCDC("drop cdc s.foo on bar", SQLExceptionCode.MISMATCHED_TOKEN.getErrorCode());
+        parseInvalidDropCDC("drop cdc foo on bar(ts)", SQLExceptionCode.MISSING_TOKEN.getErrorCode());
+    }
+
+    private AlterCDCStatement parseAlterCDCSimple(String sql, boolean ifNotExists) throws Exception {
+        AlterCDCStatement stmt = parseQuery(sql, AlterCDCStatement.class);
+        assertEquals("FOO", stmt.getTable().getName().getTableName());
+        assertEquals("BAR", stmt.getTableName());
+        assertEquals(ifNotExists, stmt.ifExists());
+        return stmt;
+    }
+    @Test
+    public void testAlterCDCSimple() throws Exception {
+        AlterCDCStatement stmt = null;
+        parseAlterCDCSimple("alter cdc foo on bar SET max_look_back=2", false);
+        parseAlterCDCSimple("alter cdc if exists foo on bar SET max_look_back=2", true);
+        parseAlterCDCSimple("alter cdc if exists foo on s.bar SET max_look_back=2", true);
+    }
+
+    private void parseInvalidAlterCDC(String sql, int expRrrorCode) throws IOException {
+        try {
+            parseQuery(sql);
+            fail();
+        }
+        catch (SQLException e) {
+            assertEquals(expRrrorCode, e.getErrorCode());
+        }
+    }
+
+    @Test
+    public void testInvalidAlterCDC() throws Exception {
+        parseInvalidAlterCDC("alter cdc foo bar SET max_look_back=2", SQLExceptionCode.MISSING_TOKEN.getErrorCode());
+        parseInvalidAlterCDC("alter cdc s.foo on bar SET max_look_back=2", SQLExceptionCode.MISMATCHED_TOKEN.getErrorCode());
+        parseInvalidAlterCDC("alter cdc foo on bar(ts) SET max_look_back=2", SQLExceptionCode.MISSING_TOKEN.getErrorCode());
+    }
+
     @Test
     public void testNextValueForSelect() throws Exception {
         String sql = ((