SNOW-1512047 Introduce independent per-table flushes when interleaving is disabled #788

sfc-gh-alhuang · 2024-07-03T22:44:53Z

The current SDK behavior flushes all channels simultaneously when any buffer reaches its limit, potentially causing unnecessary small file flushes if interleaving is disabled (MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST = 1) and the ingestion throughput between tables is uneven.

As the WIP streaming to Iceberg table feature set MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST = 1. This PR introduces per-table flushing to avoid the above issue.

JIRA

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

sfc-gh-hmadan · 2024-07-08T21:12:31Z

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

+        && !tablesToFlush.isEmpty()) {
+      tablesToFlush.addAll(
+          this.channelCache.entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toSet()));
+    }


Why do we need to do this, If the previous code block already picked up the minimal set of tables needing flush?

+1, even if interleaving is enabled, I'd prefer to keep the above logic for flushing and wait until the MaxClientLag for each channel

I aimed to maintain the original interleaving behavior, where all channels are flushed if any channel needs it. With independent flushing intervals, we might miss the chance to combine multiple chunks into the same BDEC. A potential workaround is to discretize timestamps and reduce jitter on lastFlushTime in interleaving mode. This can increase the chances of combining multiple chunks into the same blob. What do you think?

sfc-gh-hmadan · 2024-07-08T21:14:11Z

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

@@ -33,6 +40,12 @@ void addChannel(SnowflakeStreamingIngestChannelInternal<T> channel) {
        this.cache.computeIfAbsent(
            channel.getFullyQualifiedTableName(), v -> new ConcurrentHashMap<>());

+    // Update the last flush time for the table, add jitter to avoid all channels flush at the same
+    // time when the blobs are not interleaved
+    this.lastFlushTime.putIfAbsent(


not sure i understand what this helps with. If someone does addChannel, and doesn't add any data for a minute, the first row that they add will trigger a flush since we'll mistakenly think its been a long time since the last flush.

Yes. Should we change the logic to following?

Don't edit lastFlushTime when creating a channel.

When calling putRow or putRows, if lastFlushTime is null, set to current time.

Whenever a table is flushed, set lastFlushTime to null, go to step 2.

as discussed pl track this with a JIRA so we don't forget about it.

Jira created.

sfc-gh-asen

LGTM, had small comments on top of Hitesh's comments

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

sfc-gh-asen · 2024-07-09T01:37:01Z

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

+        && !tablesToFlush.isEmpty()) {
+      tablesToFlush.addAll(
+          this.channelCache.entrySet().stream().map(Map.Entry::getKey).collect(Collectors.toSet()));
+    }


+1, even if interleaving is enabled, I'd prefer to keep the above logic for flushing and wait until the MaxClientLag for each channel

sfc-gh-hmadan

LGTM, please get Alkin's or Toby's signoff before merging!

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

sfc-gh-tzhang · 2024-07-12T22:30:54Z

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

-          return;
+          logFlushTask(isForce, tablesToFlush, flushStartTime);
+          distributeFlushTasks(tablesToFlush);
+          long flushEndTime = System.currentTimeMillis();


The name is confusing, this is the end time of the previous flush?

Renamed to prevFlushEndTime.

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

sfc-gh-tzhang · 2024-07-13T01:04:18Z

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

+    if (this.owningClient.getParameterProvider().getMaxChunksInBlobAndRegistrationRequest() != 1
+        && !tablesToFlush.isEmpty()) {


Can we do the check before populating tablesToFlush?

Discussed offline. Preserve the client level lastFlushTime and isNeedFlush to avoid checking table level flush info when interleaving is enabled which might cause performance change. Preserve old logging format when interleaving is enabled to avoid logging too much information.

cc: @sfc-gh-hmadan

sfc-gh-tzhang

Left some comments, otherwise LGTM

src/main/java/net/snowflake/ingest/streaming/internal/ChannelCache.java

src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java

sfc-gh-tzhang · 2024-07-26T00:53:38Z

src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestIT.java

@@ -276,7 +276,7 @@ public void testDropChannel() throws Exception {
  @Test
  public void testParameterOverrides() throws Exception {
    Map<String, Object> parameterMap = new HashMap<>();
-    parameterMap.put(ParameterProvider.MAX_CLIENT_LAG, "3 sec");


why this is changed?

The ParameterProvider does not support "sec" (ref). The old code somehow ignore the exception thrown in thread.

jdcaperon · 2024-08-03T10:55:17Z

Thanks for solving this issue :D #570

…g is disabled (snowflakedb#788)

Per table flush

be25b1e

sfc-gh-alhuang force-pushed the alhuang-table-level-flush branch from f3ef51c to be25b1e Compare July 3, 2024 22:53

sfc-gh-alhuang marked this pull request as ready for review July 3, 2024 23:55

sfc-gh-alhuang requested review from sfc-gh-tzhang and a team as code owners July 3, 2024 23:55

sfc-gh-alhuang requested a review from sfc-gh-hmadan July 3, 2024 23:55