Revert "SNOW-1437885 Disable blob interleaving when in Iceberg mode (#…

…763)"

This reverts commit c6dfbf3.

src/main/java/net/snowflake/ingest/streaming/SnowflakeStreamingIngestClientFactory.java

@@ -28,10 +28,6 @@ public static class Builder {
     // Allows client to override some default parameter values
     private Map<String, Object> parameterOverrides;
 
-    // Indicates whether it's streaming to Iceberg tables. Open channels on regular tables should
-    // fail in this mode.
-    private boolean isIcebergMode;
-
     // Indicates whether it's under test mode
     private boolean isTestMode;
 
@@ -49,11 +45,6 @@ public Builder setParameterOverrides(Map<String, Object> parameterOverrides) {
       return this;
     }
 
-    public Builder setIsIceberg(boolean isIcebergMode) {
-      this.isIcebergMode = isIcebergMode;
-      return this;
-    }
-
     public Builder setIsTestMode(boolean isTestMode) {
       this.isTestMode = isTestMode;
       return this;
@@ -67,12 +58,7 @@ public SnowflakeStreamingIngestClient build() {
       SnowflakeURL accountURL = new SnowflakeURL(prop.getProperty(Constants.ACCOUNT_URL));
 
       return new SnowflakeStreamingIngestClientInternal<>(
-          this.name,
-          accountURL,
-          prop,
-          this.parameterOverrides,
-          this.isIcebergMode,
-          this.isTestMode);
+          this.name, accountURL, prop, this.parameterOverrides, this.isTestMode);
     }
   }
 }

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

@@ -121,9 +121,6 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
   // Indicates whether the client has closed
   private volatile boolean isClosed;
 
-  // Indicates wheter the client is streaming to Iceberg tables
-  private final boolean isIcebergMode;
-
   // Indicates whether the client is under test mode
   private final boolean isTestMode;
 
@@ -155,7 +152,6 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
    * @param accountURL Snowflake account url
    * @param prop connection properties
    * @param httpClient http client for sending request
-   * @param isIcebergMode whether we're streaming to iceberg tables
    * @param isTestMode whether we're under test mode
    * @param requestBuilder http request builder
    * @param parameterOverrides parameters we override in case we want to set different values
@@ -165,15 +161,13 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
       SnowflakeURL accountURL,
       Properties prop,
       CloseableHttpClient httpClient,
-      boolean isIcebergMode,
       boolean isTestMode,
       RequestBuilder requestBuilder,
       Map<String, Object> parameterOverrides) {
-    this.parameterProvider = new ParameterProvider(parameterOverrides, prop, isIcebergMode);
+    this.parameterProvider = new ParameterProvider(parameterOverrides, prop);
 
     this.name = name;
     String accountName = accountURL == null ? null : accountURL.getAccount();
-    this.isIcebergMode = isIcebergMode;
     this.isTestMode = isTestMode;
     this.httpClient = httpClient == null ? HttpUtil.getHttpClient(accountName) : httpClient;
     this.channelCache = new ChannelCache<>();
@@ -257,25 +251,23 @@ public class SnowflakeStreamingIngestClientInternal<T> implements SnowflakeStrea
    * @param accountURL Snowflake account url
    * @param prop connection properties
    * @param parameterOverrides map of parameters to override for this client
-   * @param isIcebergMode whether we're streaming to iceberg tables
    * @param isTestMode indicates whether it's under test mode
    */
   public SnowflakeStreamingIngestClientInternal(
       String name,
       SnowflakeURL accountURL,
       Properties prop,
       Map<String, Object> parameterOverrides,
-      boolean isIcebergMode,
       boolean isTestMode) {
-    this(name, accountURL, prop, null, isIcebergMode, isTestMode, null, parameterOverrides);
+    this(name, accountURL, prop, null, isTestMode, null, parameterOverrides);
   }
 
   /*** Constructor for TEST ONLY
    *
    * @param name the name of the client
    */
-  SnowflakeStreamingIngestClientInternal(String name, boolean isIcebergMode) {
-    this(name, null, null, null, isIcebergMode, true, null, new HashMap<>());
+  SnowflakeStreamingIngestClientInternal(String name) {
+    this(name, null, null, null, true, null, new HashMap<>());
   }
 
   // TESTING ONLY - inject the request builder

src/main/java/net/snowflake/ingest/utils/ParameterProvider.java

@@ -1,7 +1,5 @@
 package net.snowflake.ingest.utils;
 
-import static net.snowflake.ingest.utils.ErrorCode.INVALID_CONFIG_PARAMETER;
-
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -66,10 +64,6 @@ public class ParameterProvider {
   public static final Constants.BdecParquetCompression BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT =
       Constants.BdecParquetCompression.GZIP;
 
-  /* Iceberg mode parameters: When streaming to Iceberg mode, different default parameters are required because it generates Parquet files instead of BDEC files. */
-  public static final int MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_ICEBERG_MODE_DEFAULT =
-      1; // 1 parquet file per blob
-
   /* Parameter that enables using internal Parquet buffers for buffering of rows before serializing.
   It reduces memory consumption compared to using Java Objects for buffering.*/
   public static final boolean ENABLE_PARQUET_INTERNAL_BUFFERING_DEFAULT = false;
@@ -86,17 +80,14 @@ public class ParameterProvider {
    *
    * @param parameterOverrides Map of parameter name to value
    * @param props Properties from profile file
-   * @param isIcebergMode If the provided parameters need to be verified and modified to meet
-   *     Iceberg mode
    */
-  public ParameterProvider(
-      Map<String, Object> parameterOverrides, Properties props, boolean isIcebergMode) {
-    this.setParameterMap(parameterOverrides, props, isIcebergMode);
+  public ParameterProvider(Map<String, Object> parameterOverrides, Properties props) {
+    this.setParameterMap(parameterOverrides, props);
   }
 
   /** Empty constructor for tests */
-  public ParameterProvider(boolean isIcebergMode) {
-    this(null, null, isIcebergMode);
+  public ParameterProvider() {
+    this(null, null);
   }
 
   private void updateValue(
@@ -108,8 +99,6 @@ private void updateValue(
       this.parameterMap.put(key, parameterOverrides.getOrDefault(key, defaultValue));
     } else if (props != null) {
       this.parameterMap.put(key, props.getOrDefault(key, defaultValue));
-    } else {
-      this.parameterMap.put(key, defaultValue);
     }
   }
 
@@ -118,11 +107,8 @@ private void updateValue(
    *
    * @param parameterOverrides Map<String, Object> of parameter name -> value
    * @param props Properties file provided to client constructor
-   * @param isIcebergMode If the provided parameters need to be verified and modified to meet
-   *     Iceberg mode
    */
-  private void setParameterMap(
-      Map<String, Object> parameterOverrides, Properties props, boolean isIcebergMode) {
+  private void setParameterMap(Map<String, Object> parameterOverrides, Properties props) {
     // BUFFER_FLUSH_INTERVAL_IN_MILLIS is deprecated and disallowed
     if ((parameterOverrides != null
             && parameterOverrides.containsKey(BUFFER_FLUSH_INTERVAL_IN_MILLIS))
@@ -193,9 +179,7 @@ private void setParameterMap(
 
     this.updateValue(
         MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST,
-        isIcebergMode
-            ? MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_ICEBERG_MODE_DEFAULT
-            : MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_DEFAULT,
+        MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_DEFAULT,
         parameterOverrides,
         props);
 
@@ -204,13 +188,6 @@ private void setParameterMap(
         BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT,
         parameterOverrides,
         props);
-
-    // Required parameter override for Iceberg mode
-    if (isIcebergMode) {
-      icebergModeValidation(
-          MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST,
-          MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_ICEBERG_MODE_DEFAULT);
-    }
   }
 
   /** @return Longest interval in milliseconds between buffer flushes */
@@ -434,14 +411,4 @@ public Constants.BdecParquetCompression getBdecParquetCompressionAlgorithm() {
   public String toString() {
     return "ParameterProvider{" + "parameterMap=" + parameterMap + '}';
   }
-
-  private void icebergModeValidation(String key, Object expected) {
-    Object val = this.parameterMap.get(key);
-    if (!val.equals(expected)) {
-      throw new SFException(
-          INVALID_CONFIG_PARAMETER,
-          String.format(
-              "The value %s for %s is invalid in Iceberg mode, should be %s.", val, key, expected));
-    }
-  }
 }

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

@@ -24,7 +24,7 @@ public class ChannelCacheTest {
   @Before
   public void setup() {
     cache = new ChannelCache<>();
-    client = new SnowflakeStreamingIngestClientInternal<>("client", false);
+    client = new SnowflakeStreamingIngestClientInternal<>("client");
     channel1 =
         new SnowflakeStreamingIngestChannelInternal<>(
             "channel1",

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

@@ -7,7 +7,6 @@
 import static net.snowflake.ingest.utils.Constants.BLOB_NO_HEADER;
 import static net.snowflake.ingest.utils.Constants.BLOB_TAG_SIZE_IN_BYTES;
 import static net.snowflake.ingest.utils.Constants.BLOB_VERSION_SIZE_IN_BYTES;
-import static net.snowflake.ingest.utils.ParameterProvider.MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_ICEBERG_MODE_DEFAULT;
 import static net.snowflake.ingest.utils.ParameterProvider.MAX_CHUNK_SIZE_IN_BYTES_DEFAULT;
 
 import com.codahale.metrics.Histogram;
@@ -49,22 +48,11 @@
 import net.snowflake.ingest.utils.SFException;
 import org.junit.Assert;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
 
-@RunWith(Parameterized.class)
 public class FlushServiceTest {
-
-  @Parameterized.Parameters(name = "isIcebergMode: {0}")
-  public static Object[] isIcebergMode() {
-    return new Object[] {false, true};
-  }
-
-  @Parameterized.Parameter public static boolean isIcebergMode;
-
   public FlushServiceTest() {
     this.testContextFactory = ParquetTestContext.createFactory();
   }
@@ -98,7 +86,7 @@ private abstract static class TestContext<T> implements AutoCloseable {
     TestContext() {
       stage = Mockito.mock(StreamingIngestStage.class);
       Mockito.when(stage.getClientPrefix()).thenReturn("client_prefix");
-      parameterProvider = new ParameterProvider(isIcebergMode);
+      parameterProvider = new ParameterProvider();
       client = Mockito.mock(SnowflakeStreamingIngestClientInternal.class);
       Mockito.when(client.getParameterProvider()).thenReturn(parameterProvider);
       channelCache = new ChannelCache<>();
@@ -598,9 +586,7 @@ public void runTestBlobSplitDueToNumberOfChunks(int numberOfRows) throws Excepti
             Math.ceil(
                 (double) numberOfRows
                     / channelsPerTable
-                    / (isIcebergMode
-                        ? MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_ICEBERG_MODE_DEFAULT
-                        : ParameterProvider.MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_DEFAULT));
+                    / ParameterProvider.MAX_CHUNKS_IN_BLOB_AND_REGISTRATION_REQUEST_DEFAULT);
 
     final TestContext<List<List<Object>>> testContext = testContextFactory.create();
 
@@ -875,7 +861,7 @@ public void testInvalidateChannels() {
     // Create a new Client in order to not interfere with other tests
     SnowflakeStreamingIngestClientInternal<StubChunkData> client =
         Mockito.mock(SnowflakeStreamingIngestClientInternal.class);
-    ParameterProvider parameterProvider = new ParameterProvider(isIcebergMode);
+    ParameterProvider parameterProvider = new ParameterProvider();
     ChannelCache<StubChunkData> channelCache = new ChannelCache<>();
     Mockito.when(client.getChannelCache()).thenReturn(channelCache);
     Mockito.when(client.getParameterProvider()).thenReturn(parameterProvider);

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

@@ -114,7 +114,7 @@ public void testCreateOAuthClient() throws Exception {
   @Test
   public void testSetRefreshToken() throws Exception {
     SnowflakeStreamingIngestClientInternal<StubChunkData> client =
-        new SnowflakeStreamingIngestClientInternal<>("TEST_CLIENT", false);
+        new SnowflakeStreamingIngestClientInternal<>("TEST_CLIENT");
     MockOAuthClient mockOAuthClient = new MockOAuthClient();
 
     OAuthManager oAuthManager =