Iterator-based JDBC Source (and Redshift bugfix) #1887

cgardens · 2021-01-29T00:08:45Z

What

There are 2 things happening in this PR.

Removes usage of streams from the AbstractJdbcSource in favor of iterators.
Implements JdbcStandardTest for the Redshift source.

Why?

A user (@roshan) this morning report that the Redshift source was failing. The first issue was very obviously that there were . characters in the table names, which was a known issue that we actually had a PR up for, we just hadn't merged it because we didn't want to risk adding unknown bugs right before the HN launch. I pushed a version of this PR to docker hub and gave him access. This fixed the initial issue but then we found the redshift source was hanging...
It seems like the redshift source was hanging because we no longer lazily query the database. We made the change to stop lazily querying the db when we ran into the issue with flatMap a few weeks ago (PR)). This seemed relatively safe since we fetch data from the dbs in chunks. So the first chunks would be pulled eagerly but the rest would wait until we started consuming the output. This worked with the other jdbc databases. It did not work for Redshift. I believe this is due to how connection pooling is happening in Redshift where. Empirically the database hangs up after there are multiple (>9) queries queued up. I still need to dig a little deeper into the exact limitation here.
That all being said, it was clear that we were eating a bad side effect of our eager querying. We already knew this wasn't good and were struggling with other stream issues in the JdbcSource, so I went ahead switched the internals of the JdbcSource to use iterators. This allows us to get exactly the behavior we want when consuming data from the sources (lazy, element-wize, closeable, etc). Making the querying lazy and making sure we close connections allows the Redshift source to work properly.
This took a long time to debug because Redshift was the only database that didn't implement the JdbcStandardTest suite. This is because there are no test containers for Redshift so it was a really big pain to set it up in this test suite. That said, the pain was warranted today, because this was the most reasonable way to dig in and identify where the source was hanging.

How

Add MoreIterators to provide us "stream-like" syntactic sugar when dealing with iterators.
Migrate AbstractJdbcSource to use iterators
Implement JdbcStandardTest for redshift.
Modify JdbcStandardTest to handle its own clean up and not assume that there will be a new database available for every test (a constraint of working with redshift without test containers)
Modify JdbcStandardTest to handle schemas namespaces more clearly / robustly.

Pre-merge Checklist

Unit tests for MoreIterators
Verify all integration tests still pass
Verify all jdbc standard test pass
Bump versions

Other

We still have an outstanding issue from a user about the redshift destination hanging. I do not yet know if that is at all related.

airbyte-commons/src/main/java/io/airbyte/commons/concurrency/SafeVoidCallable.java

airbyte-commons/src/main/java/io/airbyte/commons/util/MoreIterators.java

michel-tricot · 2021-01-29T00:22:11Z

airbyte-commons/src/main/java/io/airbyte/commons/util/MoreIterators.java

+   * @param <T> type
+   * @return auto closing iterator
+   */
+  public static <T> Iterator<T> autoCloseIterator(Iterator<T> iterator, SafeVoidCallable onClose) {


This is a VERY dangerous function. It should not return an Iterator, it should return an AutoCloseIterator. Otherwise there is absolutely no way to release resources but to consume the iterator to the end.

michel-tricot · 2021-01-29T00:23:53Z

airbyte-commons/src/main/java/io/airbyte/commons/util/MoreIterators.java

+   * @param <T> type
+   * @return vanilla iterator
+   */
+  public static <T> Iterator<T> toCloseableIterator(CloseableIterator<T> iterator) {


you need to kill this function, it cannot exist. Way too dangerous for the reason I mentioned earlier

airbyte-commons/src/main/java/io/airbyte/commons/concurrency/VoidCallable.java

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/Source.java

...ectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java

...s/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/StateDecoratingIterator.java

cgardens · 2021-01-30T00:57:44Z

@michel-tricot i think i hit your feedback. the AutoClosingIterator now implements the ResourceIterator interface so it can be closed at any time. I think this removes the danger you were pointing out in your previous review. Let me know what you think.

I'm glad we are removing the headache of streams. The iterator approach is unfortunately taking a ton of boilerplate to recreate some of the features of streams that we need though. Let me know if you feel that there's opportunity to reduce this boilerplate.

Still need to do unit tests.

airbyte-commons/src/main/java/io/airbyte/commons/util/ResourceIterator.java

airbyte-commons/src/main/java/io/airbyte/commons/util/AutoCloseIterator.java

michel-tricot · 2021-01-30T01:55:11Z

airbyte-commons/src/main/java/io/airbyte/commons/util/AutoCloseIterator.java

+
+  @Override
+  public void close() throws Exception {
+    internalIterator.close();


can you do the boolean management here and just not do anything if close has already been called? You probably want to use an AtomicBoolean with a get&set if you do so

I'm trying to push for composition here the way one would thing about a stream. Each of these iterators is designed to do one thing. They can be then composed to get whatever behavior you want.

So if you want close to be called once, then you compose this iterator with the default iterator. I think of this as good default behavior which is why it's called the DefaultResourceIterator but if we think that needs to named more explicitly we can do that to.

Anyway, this attempt at composition is why I am not doing an AutoClosingLazySingleCloseIterator and not combining the behavior you're describing here. Are you not a fan of this approach?

airbyte-commons/src/main/java/io/airbyte/commons/util/DefaultResourceIterator.java

airbyte-commons/src/main/java/io/airbyte/commons/util/ResourceIterator.java

michel-tricot · 2021-01-30T02:03:14Z