Set createdon in BlobMetadata and is_iceberg in Drop Channel request (#…

…845) - Set createdOn on the blob Metadata, which can be used in the service to populate the FileRegistrationRequest - Set is_iceberg in /channels/drop API payload since it gets validated the same way as open channel (where we do have the is_iceberg API available)
snowflakedb · Sep 30, 2024 · 27cd3a7 · 27cd3a7
1 parent 7b3881b
commit 27cd3a7
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 5 deletions.
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java b/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java
@@ -142,7 +142,8 @@ static <T> Blob constructBlobAndMetadata(
           chunkMetadataBuilder
               .setMajorVersion(Constants.PARQUET_MAJOR_VERSION)
               .setMinorVersion(Constants.PARQUET_MINOR_VERSION)
-              .setCreatedOn(0L)
+              // set createdOn in seconds
+              .setCreatedOn(System.currentTimeMillis() / 1000)
               .setExtendedMetadataSize(-1L);
         }
 

diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/DropChannelRequestInternal.java b/src/main/java/net/snowflake/ingest/streaming/internal/DropChannelRequestInternal.java
@@ -33,20 +33,25 @@ class DropChannelRequestInternal implements IStreamingIngestRequest {
   @JsonProperty("client_sequencer")
   Long clientSequencer;
 
+  @JsonProperty("is_iceberg")
+  boolean isIceberg;
+
   DropChannelRequestInternal(
       String requestId,
       String role,
       String database,
       String schema,
       String table,
       String channel,
+      boolean isIceberg,
       Long clientSequencer) {
     this.requestId = requestId;
     this.role = role;
     this.database = database;
     this.schema = schema;
     this.table = table;
     this.channel = channel;
+    this.isIceberg = isIceberg;
     this.clientSequencer = clientSequencer;
   }
 
@@ -74,6 +79,10 @@ String getSchema() {
     return schema;
   }
 
+  boolean isIceberg() {
+    return isIceberg;
+  }
+
   Long getClientSequencer() {
     return clientSequencer;
   }
@@ -86,7 +95,7 @@ String getFullyQualifiedTableName() {
   public String getStringForLogging() {
     return String.format(
         "DropChannelRequest(requestId=%s, role=%s, db=%s, schema=%s, table=%s, channel=%s,"
-            + " clientSequencer=%s)",
-        requestId, role, database, schema, table, channel, clientSequencer);
+            + " isIceberg=%s, clientSequencer=%s)",
+        requestId, role, database, schema, table, channel, isIceberg, clientSequencer);
   }
 }
diff --git a/.../java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientInternal.java b/.../java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientInternal.java
@@ -422,6 +422,7 @@ public void dropChannel(DropChannelRequest request) {
               request.getSchemaName(),
               request.getTableName(),
               request.getChannelName(),
+              this.isIcebergMode,
               request instanceof DropChannelVersionRequest
                   ? ((DropChannelVersionRequest) request).getClientSequencer()
                   : null);

diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeServiceClientTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeServiceClientTest.java
@@ -9,8 +9,18 @@
 import net.snowflake.ingest.utils.Constants;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
+@RunWith(Parameterized.class)
 public class SnowflakeServiceClientTest {
+  @Parameterized.Parameters(name = "isIceberg: {0}")
+  public static Object[] isIceberg() {
+    return new Object[] {false, true};
+  }
+
+  @Parameterized.Parameter public boolean isIceberg;
+
   private SnowflakeServiceClient snowflakeServiceClient;
 
   @Before
@@ -50,7 +60,7 @@ public void testOpenChannel() throws IngestResponseException, IOException {
             "test_table",
             "test_channel",
             Constants.WriteMode.CLOUD_STORAGE,
-            false,
+            isIceberg,
             "test_offset_token");
     OpenChannelResponse openChannelResponse =
         snowflakeServiceClient.openChannel(openChannelRequest);
@@ -72,7 +82,14 @@ public void testOpenChannel() throws IngestResponseException, IOException {
   public void testDropChannel() throws IngestResponseException, IOException {
     DropChannelRequestInternal dropChannelRequest =
         new DropChannelRequestInternal(
-            "request_id", "test_role", "test_db", "test_schema", "test_table", "test_channel", 0L);
+            "request_id",
+            "test_role",
+            "test_db",
+            "test_schema",
+            "test_table",
+            "test_channel",
+            isIceberg,
+            0L);
     DropChannelResponse dropChannelResponse =
         snowflakeServiceClient.dropChannel(dropChannelRequest);
     assert dropChannelResponse.getStatusCode() == 0L;