Large columns

pom.xml

@@ -226,6 +226,11 @@
             <artifactId>kafka-connect-storage-common-avatica-shaded</artifactId>
             <version>${kafka.connect.storage.common.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.ibm.db2</groupId>
+            <artifactId>jcc</artifactId>
+            <version>11.5.8.0</version>
+        </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

src/main/java/io/confluent/connect/hdfs/jdbc/FilteredColumnToStructVisitor.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.connect.hdfs.jdbc;
+
+import org.apache.kafka.connect.data.Struct;
+
+import java.sql.Blob;
+import java.sql.Clob;
+import java.sql.SQLException;
+import java.sql.SQLXML;
+
+public class FilteredColumnToStructVisitor extends FilteredJdbcColumnVisitor {
+  private final Struct struct;
+
+  public FilteredColumnToStructVisitor(HashCache hashCache,
+                                       JdbcTableInfo tableInfo,
+                                       String primaryKey,
+                                       Struct struct) {
+    super(hashCache, tableInfo, primaryKey);
+    this.struct = struct;
+  }
+
+  @Override
+  public void visit(String columnName, Blob value) throws SQLException {
+    // TODO: Would be so much better if we could stream this data
+    // TODO: Write to a disk-buffer first, and then digest()? RocksDB?
+    byte[] bytes = value != null
+        ? value.getBytes(1L, (int) value.length())
+        : null;
+
+    updateCache(columnName, bytes);
+
+    struct.put(columnName, bytes);
+  }
+
+  @Override
+  public void visit(String columnName, Clob value) throws SQLException {
+    // TODO: Would be so much better if we could stream this data
+    // TODO: Write to a disk-buffer first, and then digest()? RocksDB?
+    String valueStr = value != null
+        ? value.getSubString(1L, (int) value.length())
+        : null;
+
+    updateCache(columnName, valueStr);
+
+    struct.put(columnName, valueStr);
+  }
+
+  @Override
+  public void visit(String columnName, SQLXML value) throws SQLException {
+    // TODO: Would be so much better if we could stream this data
+    // TODO: Write to a disk-buffer first, and then digest()? RocksDB?
+    String valueStr = value != null
+        ? value.getString()
+        : null;
+
+    updateCache(columnName, valueStr);
+
+    struct.put(columnName, valueStr);
+  }
+
+  @Override
+  public void visit(String columnName, String value) {
+    updateCache(columnName, value);
+
+    struct.put(columnName, value);
+  }
+}

src/main/java/io/confluent/connect/hdfs/jdbc/FilteredJdbcColumnVisitor.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.connect.hdfs.jdbc;
+
+import java.util.Optional;
+
+public abstract class FilteredJdbcColumnVisitor implements JdbcColumnVisitor {
+  private final HashCache hashCache;
+  protected final JdbcTableInfo tableInfo;
+  protected final String primaryKey;
+  private int columnsChanged = 0;
+
+  public FilteredJdbcColumnVisitor(HashCache hashCache,
+                                   JdbcTableInfo tableInfo,
+                                   String primaryKey) {
+    this.hashCache = hashCache;
+    this.tableInfo = tableInfo;
+    this.primaryKey = primaryKey;
+  }
+
+  public boolean hasChangedColumns() {
+    return columnsChanged > 0;
+  }
+
+  protected void updateCache(String columnName, byte[] value) {
+    boolean columnChanged = Optional
+        .ofNullable(hashCache)
+        .map(hashCache_ -> hashCache_.updateCache(tableInfo, primaryKey, columnName, value))
+        .orElse(true);
+
+    // If it has changed, indicate that we should write the new value to HDFS
+    if (columnChanged) {
+      columnsChanged++;
+    }
+  }
+
+  protected void updateCache(String columnName, String value) {
+    boolean columnChanged = Optional
+        .ofNullable(hashCache)
+        .map(hashCache_ -> hashCache_.updateCache(tableInfo, primaryKey, columnName, value))
+        .orElse(true);
+
+    // If it has changed, indicate that we should write the new value to HDFS
+    if (columnChanged) {
+      columnsChanged++;
+    }
+  }
+}

src/main/java/io/confluent/connect/hdfs/jdbc/HashCache.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.connect.hdfs.jdbc;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Cache of LOB hashes, for pruning unnecessary HDFS writes;
+ * NOTE: None of this is synchronized, as it expects to be called from non-parallel code.
+ * NOTE: MD5 Hashing takes about 1-2 seconds per gig, at least locally
+ */
+public class HashCache {
+  private static final byte[] EMPTY_HASH = {};
+
+  private final Map<JdbcTableInfo, LinkedHashMap<String, Map<String, byte[]>>>
+      tablePkColumnCache = new HashMap<>();
+
+  private final int maxHashSize;
+  private final MessageDigest messageDigest;
+
+  public HashCache(int maxHashSize, MessageDigest messageDigest) {
+    this.maxHashSize = maxHashSize;
+    this.messageDigest = messageDigest;
+  }
+
+  public boolean updateCache(JdbcTableInfo tableInfo,
+                             String primaryKey,
+                             String columnName,
+                             String value
+  ) {
+    byte[] bytes = Optional
+        .ofNullable(value)
+        // TODO: Should we hardcode UTF_8 here?
+        .map(value_ -> value_.getBytes(StandardCharsets.UTF_8))
+        .orElse(null);
+
+    return updateCache(tableInfo, primaryKey, columnName, bytes);
+  }
+
+  /**
+   * Update cache; and return true if hash has changed
+   */
+  public boolean updateCache(JdbcTableInfo tableInfo,
+                             String primaryKey,
+                             String columnName,
+                             byte[] value
+  ) {
+    byte[] hash = Optional
+        .ofNullable(value)
+        .map(bytes -> {
+          messageDigest.reset();
+          return messageDigest.digest(bytes);
+        })
+        .orElse(EMPTY_HASH);
+
+    // Get the Cache for the given Table + PK
+    Map<String, Map<String, byte[]>> pkColumnHashCache = getPkColumnCache(tableInfo);
+    Map<String, byte[]> columnHashCache = pkColumnHashCache.computeIfAbsent(
+        primaryKey,
+        __ -> new HashMap<>()
+    );
+
+    // Check the hash against the cache
+    byte[] oldHash = columnHashCache.put(columnName, hash);
+    return !Arrays.equals(hash, oldHash);
+  }
+
+  private LinkedHashMap<String, Map<String, byte[]>> getPkColumnCache(
+      JdbcTableInfo tableInfo
+  ) {
+    // The entire tablePkColumnCache map structure is accessed serially,
+    // so no need to make it threadsafe
+    return tablePkColumnCache.computeIfAbsent(
+        tableInfo,
+        __ -> new LinkedHashMap<String, Map<String, byte[]>>(
+            100,
+            .75f,
+            true
+        ) {
+          // NOTE: Turns LinkedHashMap into an LRU Cache
+          @Override
+          protected boolean removeEldestEntry(
+              Map.Entry<String, Map<String, byte[]>> eldest
+          ) {
+            return this.size() > maxHashSize;
+          }
+        }
+    );
+  }
+}

src/main/java/io/confluent/connect/hdfs/jdbc/JdbcColumnInfo.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.connect.hdfs.jdbc;
+
+import java.sql.JDBCType;
+import java.util.Comparator;
+import java.util.Objects;
+
+public class JdbcColumnInfo {
+  private final String name;
+  private final JDBCType jdbcType;
+  private final int ordinal;
+  private final boolean nullable;
+
+  public static Comparator<JdbcColumnInfo> byOrdinal =
+      Comparator.comparingInt(JdbcColumnInfo::getOrdinal);
+
+  public JdbcColumnInfo(String name,
+                        JDBCType jdbcType,
+                        int ordinal,
+                        boolean nullable) {
+    this.name = name;
+    this.jdbcType = jdbcType;
+    this.ordinal = ordinal;
+    this.nullable = nullable;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public JDBCType getJdbcType() {
+    return jdbcType;
+  }
+
+  public int getOrdinal() {
+    return ordinal;
+  }
+
+  public boolean isNullable() {
+    return nullable;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof JdbcColumnInfo)) {
+      return false;
+    }
+    JdbcColumnInfo column = (JdbcColumnInfo) o;
+    return Objects.equals(getName(), column.getName())
+           && getJdbcType() == column.getJdbcType()
+           && getOrdinal() == column.getOrdinal()
+           && isNullable() == column.isNullable();
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(getName(), getJdbcType(), getOrdinal(), isNullable());
+  }
+
+  @Override
+  public String toString() {
+    return getClass().getSimpleName()
+           + "{"
+           + "name='" + name + "'"
+           + ", jdbcType=" + jdbcType
+           + ", ordinal=" + ordinal
+           + ", nullable=" + nullable
+           + "}";
+  }
+}