Support Postgres partitioned table; confluentinc/kafka-connect-jdbc#1092

.gitignore

@@ -6,6 +6,12 @@ target/
 .idea/
 .ipr
 
+# Eclipse IDE
+.project
+.classpath
+.checkstyle
+.settings/
+
 # Documentation build output
 /docs/_build
 

src/main/java/io/confluent/connect/jdbc/sink/JdbcSinkConfig.java

@@ -249,8 +249,9 @@ public enum PrimaryKeyMode {
   private static final String TABLE_TYPES_DOC =
       "The comma-separated types of database tables to which the sink connector can write. "
       + "By default this is ``" + TableType.TABLE + "``, but any combination of ``"
-      + TableType.TABLE + "`` and ``" + TableType.VIEW + "`` is allowed. Not all databases "
-      + "support writing to views, and when they do the the sink connector will fail if the "
+      + TableType.TABLE + "``, ``" + TableType.PARTITIONED_TABLE + "`` and ``"
+      + TableType.VIEW + "`` is allowed. Not all databases support writing to views, "
+      + "and when they do the sink connector will fail if the "
       + "view definition does not match the records' schemas (regardless of ``"
       + AUTO_EVOLVE + "``).";
 

src/main/java/io/confluent/connect/jdbc/util/TableType.java

@@ -21,7 +21,9 @@
 
 public enum TableType {
 
-  TABLE("TABLE", "Table"), VIEW("VIEW", "View");
+  TABLE("TABLE", "Table"),
+  PARTITIONED_TABLE("PARTITIONED TABLE", "Partitioned Table"),
+  VIEW("VIEW", "View");
 
   private final String value;
   private final String capitalCase;

src/test/java/io/confluent/connect/jdbc/sink/JdbcSinkConfigTest.java

@@ -86,6 +86,13 @@ public void shouldCreateConfigWithTableOnly() {
     assertTableTypes(TableType.TABLE);
   }
 
+  @Test
+  public void shouldCreateConfigWithPartitionedTableOnly() {
+    props.put("table.types", "partitioned table");
+    createConfig();
+    assertTableTypes(TableType.PARTITIONED_TABLE);
+  }
+
   @Test
   public void shouldCreateConfigWithViewAndTable() {
     props.put("table.types", "view,table");

src/test/java/io/confluent/connect/jdbc/util/TableTypeTest.java

@@ -25,6 +25,7 @@
 public class TableTypeTest {
 
   private static EnumSet<TableType> TABLE_ONLY = types(TableType.TABLE);
+  private static EnumSet<TableType> PARTITIONED_TABLE_ONLY = types(TableType.PARTITIONED_TABLE);
   private static EnumSet<TableType> VIEW_ONLY = types(TableType.VIEW);
   private static EnumSet<TableType> TABLE_AND_VIEW = types(TableType.TABLE, TableType.VIEW);
 
@@ -56,6 +57,7 @@ public void shouldParseTypeStringWithWhitespace() {
   public void shouldComputeJdbcTypeArray() {
     assertArrayEquals(array("TABLE"), TableType.asJdbcTableTypeArray(TABLE_ONLY));
     assertArrayEquals(array("VIEW"), TableType.asJdbcTableTypeArray(VIEW_ONLY));
+    assertArrayEquals(array("PARTITIONED TABLE"), TableType.asJdbcTableTypeArray(PARTITIONED_TABLE_ONLY));
     assertArrayEquals(array("TABLE", "VIEW"), TableType.asJdbcTableTypeArray(TABLE_AND_VIEW));
   }