diff --git a/pom.xml b/pom.xml
index cb4b2ed0b..b7763631a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -352,6 +352,30 @@
         <version>1.14.9</version>
         <scope>test</scope>
       </dependency>
+      <dependency>
+        <groupId>org.apache.hadoop</groupId>
+        <artifactId>hadoop-mapreduce-client-core</artifactId>
+        <version>${hadoop.version}</version>
+        <scope>test</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>jakarta.xml.bind</groupId>
+            <artifactId>jakarta.xml.bind-api</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>javax.xml.bind</groupId>
+            <artifactId>jaxb-api</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-reload4j</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-core</artifactId>
@@ -478,6 +502,10 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+    </dependency>
     <!-- https://mvnrepository.com/artifact/org.apache.parquet/parquet-hadoop -->
     <dependency>
       <groupId>org.apache.parquet</groupId>
@@ -756,8 +784,8 @@
               <ignoreNonCompile>true</ignoreNonCompile>
               <ignoredDependencies>
                 <!-- We defined these as direct dependencies (as opposed to just declaring it in dependencyManagement)
-                                                                                                                                                                                                to workaround https://issues.apache.org/jira/browse/MNG-7982. Now the dependency analyzer complains that
-                                                                                                                                                                                                the dependency is unused, so we ignore it here-->
+                                                                                                                                                                                                                                to workaround https://issues.apache.org/jira/browse/MNG-7982. Now the dependency analyzer complains that
+                                                                                                                                                                                                                                the dependency is unused, so we ignore it here-->
                 <ignoredDependency>org.apache.commons:commons-compress</ignoredDependency>
                 <ignoredDependency>org.apache.commons:commons-configuration2</ignoredDependency>
               </ignoredDependencies>
@@ -852,9 +880,9 @@
         <configuration>
           <errorRemedy>failFast</errorRemedy>
           <!--
-                                                                                                                                                                 The list of allowed licenses. If you see the build failing due to "There are some forbidden licenses used, please
-                                                                                                                                                                check your dependencies", verify the conditions of the license and add the reference to it here.
-                                                                                                                                                              -->
+                                                                                                                                                                                     The list of allowed licenses. If you see the build failing due to "There are some forbidden licenses used, please
+                                                                                                                                                                                    check your dependencies", verify the conditions of the license and add the reference to it here.
+                                                                                                                                                                                  -->
           <includedLicenses>
             <includedLicense>Apache License 2.0</includedLicense>
             <includedLicense>BSD 2-Clause License</includedLicense>
@@ -1167,9 +1195,9 @@
             </executions>
           </plugin>
           <!--
-                                                                                                                                                                          Plugin executes license processing Python script, which copies third party license files into the directory
-                                                                                                                                                                          target/generated-licenses-info/META-INF/third-party-licenses, which is then included in the shaded JAR.
-                                                                                                                                                                          -->
+                                                                                                                                                                                              Plugin executes license processing Python script, which copies third party license files into the directory
+                                                                                                                                                                                              target/generated-licenses-info/META-INF/third-party-licenses, which is then included in the shaded JAR.
+                                                                                                                                                                                              -->
           <plugin>
             <groupId>org.codehaus.mojo</groupId>
             <artifactId>exec-maven-plugin</artifactId>
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java
index 71a9d501e..9172c4328 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java
@@ -496,11 +496,10 @@ public InsertValidationResponse insertRows(
    * Flush the data in the row buffer by taking the ownership of the old vectors and pass all the
    * required info back to the flush service to build the blob
    *
-   * @param filePath the name of the file the data will be written in
    * @return A ChannelData object that contains the info needed by the flush service to build a blob
    */
   @Override
-  public ChannelData<T> flush(final String filePath) {
+  public ChannelData<T> flush() {
     logger.logDebug("Start get data for channel={}", channelFullyQualifiedName);
     if (this.bufferedRowCount > 0) {
       Optional<T> oldData = Optional.empty();
@@ -518,7 +517,7 @@ public ChannelData<T> flush(final String filePath) {
       try {
         if (this.bufferedRowCount > 0) {
           // Transfer the ownership of the vectors
-          oldData = getSnapshot(filePath);
+          oldData = getSnapshot();
           oldRowCount = this.bufferedRowCount;
           oldBufferSize = this.bufferSize;
           oldRowSequencer = this.channelState.incrementAndGetRowSequencer();
@@ -615,12 +614,8 @@ void reset() {
     this.statsMap.replaceAll((key, value) -> value.forkEmpty());
   }
 
-  /**
-   * Get buffered data snapshot for later flushing.
-   *
-   * @param filePath the name of the file the data will be written in
-   */
-  abstract Optional<T> getSnapshot(final String filePath);
+  /** Get buffered data snapshot for later flushing. */
+  abstract Optional<T> getSnapshot();
 
   @VisibleForTesting
   abstract Object getVectorValueAt(String column, int index);
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java b/src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java
index 6f5296209..2e7c6507c 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/FlushService.java
@@ -430,7 +430,7 @@ void distributeFlushTasks(Set<String> tablesToFlush) {
               .forEach(
                   channel -> {
                     if (channel.isValid()) {
-                      ChannelData<T> data = channel.getData(blobPath);
+                      ChannelData<T> data = channel.getData();
                       if (data != null) {
                         channelsDataPerTable.add(data);
                       }
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetFlusher.java b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetFlusher.java
index 3ad3db5f4..d338a6a7b 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetFlusher.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetFlusher.java
@@ -212,6 +212,10 @@ private SerializationResult serializeFromJavaObjects(
     }
 
     Map<String, String> metadata = channelsDataPerTable.get(0).getVectors().metadata;
+    // We insert the filename in the file itself as metadata so that streams can work on replicated
+    // mixed tables. For a more detailed discussion on the topic see SNOW-561447 and
+    // http://go/streams-on-replicated-mixed-tables
+    metadata.put(Constants.PRIMARY_FILE_ID_KEY, StreamingIngestUtils.getShortname(filePath));
     parquetWriter =
         new BdecParquetWriter(
             mergedData,
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java
index 627478bca..30851c274 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java
@@ -22,7 +22,6 @@
 import net.snowflake.ingest.connection.TelemetryService;
 import net.snowflake.ingest.streaming.OffsetTokenVerificationFunction;
 import net.snowflake.ingest.streaming.OpenChannelRequest;
-import net.snowflake.ingest.utils.Constants;
 import net.snowflake.ingest.utils.ErrorCode;
 import net.snowflake.ingest.utils.SFException;
 import org.apache.parquet.hadoop.BdecParquetWriter;
@@ -262,12 +261,7 @@ boolean hasColumns() {
   }
 
   @Override
-  Optional<ParquetChunkData> getSnapshot(final String filePath) {
-    // We insert the filename in the file itself as metadata so that streams can work on replicated
-    // mixed tables. For a more detailed discussion on the topic see SNOW-561447 and
-    // http://go/streams-on-replicated-mixed-tables
-    metadata.put(Constants.PRIMARY_FILE_ID_KEY, StreamingIngestUtils.getShortname(filePath));
-
+  Optional<ParquetChunkData> getSnapshot() {
     List<List<Object>> oldData = new ArrayList<>();
     if (!clientBufferParameters.getEnableParquetInternalBuffering()) {
       data.forEach(r -> oldData.add(new ArrayList<>(r)));
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/RowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/RowBuffer.java
index 02905c02e..6bb2f43b9 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/RowBuffer.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/RowBuffer.java
@@ -37,10 +37,9 @@ InsertValidationResponse insertRows(
    * Flush the data in the row buffer by taking the ownership of the old vectors and pass all the
    * required info back to the flush service to build the blob
    *
-   * @param filePath the name of the file the data will be written in
    * @return A ChannelData object that contains the info needed by the flush service to build a blob
    */
-  ChannelData<T> flush(final String filePath);
+  ChannelData<T> flush();
 
   /**
    * Close the row buffer and release resources. Note that the caller needs to handle
diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java b/src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java
index ca0bbe782..4e884387b 100644
--- a/src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java
+++ b/src/main/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java
@@ -220,11 +220,10 @@ public String getFullyQualifiedTableName() {
   /**
    * Get all the data needed to build the blob during flush
    *
-   * @param filePath the name of the file the data will be written in
    * @return a ChannelData object
    */
-  ChannelData<T> getData(final String filePath) {
-    ChannelData<T> data = this.rowBuffer.flush(filePath);
+  ChannelData<T> getData() {
+    ChannelData<T> data = this.rowBuffer.flush();
     if (data != null) {
       data.setChannelContext(channelFlushContext);
     }
diff --git a/src/main/java/org/apache/parquet/hadoop/BdecParquetReader.java b/src/main/java/org/apache/parquet/hadoop/BdecParquetReader.java
index 1a92a8cd4..ef95fab14 100644
--- a/src/main/java/org/apache/parquet/hadoop/BdecParquetReader.java
+++ b/src/main/java/org/apache/parquet/hadoop/BdecParquetReader.java
@@ -34,6 +34,7 @@
  */
 public class BdecParquetReader implements AutoCloseable {
   private final InternalParquetRecordReader<List<Object>> reader;
+  private final ParquetFileReader fileReader;
 
   /**
    * @param data buffer where the data that has to be read resides.
@@ -41,7 +42,7 @@ public class BdecParquetReader implements AutoCloseable {
    */
   public BdecParquetReader(byte[] data) throws IOException {
     ParquetReadOptions options = ParquetReadOptions.builder().build();
-    ParquetFileReader fileReader = ParquetFileReader.open(new BdecInputFile(data), options);
+    fileReader = ParquetFileReader.open(new BdecInputFile(data), options);
     reader = new InternalParquetRecordReader<>(new BdecReadSupport(), options.getRecordFilter());
     reader.initialize(fileReader, options);
   }
@@ -60,6 +61,11 @@ public List<Object> read() throws IOException {
     }
   }
 
+  /** Get the key value metadata in the file */
+  public Map<String, String> getKeyValueMetadata() {
+    return fileReader.getFileMetaData().getKeyValueMetaData();
+  }
+
   /**
    * Close the reader.
    *
diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java
index 4cc5cbe30..303893bf0 100644
--- a/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java
+++ b/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java
@@ -131,7 +131,7 @@ void setParameterOverride(Map<String, Object> parameterOverride) {
 
     ChannelData<T> flushChannel(String name) {
       SnowflakeStreamingIngestChannelInternal<T> channel = channels.get(name);
-      ChannelData<T> channelData = channel.getRowBuffer().flush(name + "_snowpipe_streaming.bdec");
+      ChannelData<T> channelData = channel.getRowBuffer().flush();
       channelData.setChannelContext(channel.getChannelContext());
       this.channelData.add(channelData);
       return channelData;
diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java
index 41739c267..3f8e927a4 100644
--- a/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java
+++ b/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java
@@ -6,6 +6,7 @@
 import static net.snowflake.ingest.utils.ParameterProvider.MAX_CHUNK_SIZE_IN_BYTES_DEFAULT;
 import static org.junit.Assert.fail;
 
+import java.io.IOException;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
@@ -24,6 +25,7 @@
 import net.snowflake.ingest.utils.ErrorCode;
 import net.snowflake.ingest.utils.SFException;
 import org.apache.commons.codec.binary.Hex;
+import org.apache.parquet.hadoop.BdecParquetReader;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -499,17 +501,12 @@ private void testFlushHelper(AbstractRowBuffer<?> rowBuffer) {
     float bufferSize = rowBuffer.getSize();
 
     final String filename = "2022/7/13/16/56/testFlushHelper_streaming.bdec";
-    ChannelData<?> data = rowBuffer.flush(filename);
+    ChannelData<?> data = rowBuffer.flush();
     Assert.assertEquals(2, data.getRowCount());
     Assert.assertEquals((Long) 1L, data.getRowSequencer());
     Assert.assertEquals(startOffsetToken, data.getStartOffsetToken());
     Assert.assertEquals(endOffsetToken, data.getEndOffsetToken());
     Assert.assertEquals(bufferSize, data.getBufferSize(), 0);
-
-    final ParquetChunkData chunkData = (ParquetChunkData) data.getVectors();
-    Assert.assertEquals(
-        StreamingIngestUtils.getShortname(filename),
-        chunkData.metadata.get(Constants.PRIMARY_FILE_ID_KEY));
   }
 
   @Test
@@ -739,7 +736,7 @@ private void testStatsE2EHelper(AbstractRowBuffer<?> rowBuffer) {
     final String filename = "testStatsE2EHelper_streaming.bdec";
     InsertValidationResponse response = rowBuffer.insertRows(Arrays.asList(row1, row2), null, null);
     Assert.assertFalse(response.hasErrors());
-    ChannelData<?> result = rowBuffer.flush(filename);
+    ChannelData<?> result = rowBuffer.flush();
     Map<String, RowBufferStats> columnEpStats = result.getColumnEps();
 
     Assert.assertEquals(
@@ -783,11 +780,8 @@ private void testStatsE2EHelper(AbstractRowBuffer<?> rowBuffer) {
     Assert.assertEquals(0, columnEpStats.get("COLCHAR").getCurrentNullCount());
     Assert.assertEquals(-1, columnEpStats.get("COLCHAR").getDistinctValues());
 
-    final ParquetChunkData chunkData = (ParquetChunkData) result.getVectors();
-    Assert.assertEquals(filename, chunkData.metadata.get(Constants.PRIMARY_FILE_ID_KEY));
-
     // Confirm we reset
-    ChannelData<?> resetResults = rowBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> resetResults = rowBuffer.flush();
     Assert.assertNull(resetResults);
   }
 
@@ -846,7 +840,7 @@ private void testStatsE2ETimestampHelper(OpenChannelRequest.OnErrorOption onErro
     InsertValidationResponse response =
         innerBuffer.insertRows(Arrays.asList(row1, row2, row3), null, null);
     Assert.assertFalse(response.hasErrors());
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(3, result.getRowCount());
 
     Assert.assertEquals(
@@ -915,7 +909,7 @@ private void testE2EDateHelper(OpenChannelRequest.OnErrorOption onErrorOption) {
     Assert.assertNull(innerBuffer.getVectorValueAt("COLDATE", 2));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(3, result.getRowCount());
 
     Assert.assertEquals(
@@ -981,7 +975,7 @@ private void testE2ETimeHelper(OpenChannelRequest.OnErrorOption onErrorOption) {
     Assert.assertNull(innerBuffer.getVectorValueAt("COLTIMESB8", 2));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(3, result.getRowCount());
 
     Assert.assertEquals(
@@ -1212,7 +1206,7 @@ private void doTestFailureHalfwayThroughColumnProcessing(
       }
     }
 
-    ChannelData<?> channelData = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> channelData = innerBuffer.flush();
     RowBufferStats statsCol1 = channelData.getColumnEps().get("COLVARCHAR1");
     RowBufferStats statsCol2 = channelData.getColumnEps().get("COLVARCHAR2");
     RowBufferStats statsCol3 = channelData.getColumnEps().get("COLBOOLEAN1");
@@ -1272,7 +1266,7 @@ private void testE2EBooleanHelper(OpenChannelRequest.OnErrorOption onErrorOption
     Assert.assertNull(innerBuffer.getVectorValueAt("COLBOOLEAN", 2));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(3, result.getRowCount());
 
     Assert.assertEquals(
@@ -1327,7 +1321,7 @@ private void testE2EBinaryHelper(OpenChannelRequest.OnErrorOption onErrorOption)
     Assert.assertNull(innerBuffer.getVectorValueAt("COLBINARY", 2));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
 
     Assert.assertEquals(3, result.getRowCount());
     Assert.assertEquals(11L, result.getColumnEps().get("COLBINARY").getCurrentMaxLength());
@@ -1379,7 +1373,7 @@ private void testE2ERealHelper(OpenChannelRequest.OnErrorOption onErrorOption) {
     Assert.assertNull(innerBuffer.getVectorValueAt("COLREAL", 2));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
 
     Assert.assertEquals(3, result.getRowCount());
     Assert.assertEquals(
@@ -1462,7 +1456,7 @@ public void testOnErrorAbortFailures() {
     Assert.assertNull(innerBuffer.tempStatsMap.get("COLDECIMAL").getCurrentMaxIntValue());
     Assert.assertNull(innerBuffer.tempStatsMap.get("COLDECIMAL").getCurrentMinIntValue());
 
-    ChannelData<?> data = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> data = innerBuffer.flush();
     Assert.assertEquals(3, data.getRowCount());
     Assert.assertEquals(0, innerBuffer.bufferedRowCount);
   }
@@ -1536,7 +1530,7 @@ public void testOnErrorAbortSkipBatch() {
     Assert.assertNull(innerBuffer.tempStatsMap.get("COLDECIMAL").getCurrentMaxIntValue());
     Assert.assertNull(innerBuffer.tempStatsMap.get("COLDECIMAL").getCurrentMinIntValue());
 
-    ChannelData<?> data = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> data = innerBuffer.flush();
     Assert.assertEquals(3, data.getRowCount());
     Assert.assertEquals(0, innerBuffer.bufferedRowCount);
   }
@@ -1587,7 +1581,7 @@ private void testE2EVariantHelper(OpenChannelRequest.OnErrorOption onErrorOption
     Assert.assertEquals("3", innerBuffer.getVectorValueAt("COLVARIANT", 4));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(5, result.getRowCount());
     Assert.assertEquals(2, result.getColumnEps().get("COLVARIANT").getCurrentNullCount());
   }
@@ -1621,7 +1615,7 @@ private void testE2EObjectHelper(OpenChannelRequest.OnErrorOption onErrorOption)
     Assert.assertEquals("{\"key\":1}", innerBuffer.getVectorValueAt("COLOBJECT", 0));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(1, result.getRowCount());
   }
 
@@ -1671,7 +1665,7 @@ private void testE2EArrayHelper(OpenChannelRequest.OnErrorOption onErrorOption)
     Assert.assertEquals("[1,2,3]", innerBuffer.getVectorValueAt("COLARRAY", 4));
 
     // Check stats generation
-    ChannelData<?> result = innerBuffer.flush("my_snowpipe_streaming.bdec");
+    ChannelData<?> result = innerBuffer.flush();
     Assert.assertEquals(5, result.getRowCount());
   }
 
@@ -1718,14 +1712,14 @@ public void testOnErrorAbortRowsWithError() {
         SFException.class, () -> innerBufferOnErrorAbort.insertRows(mixedRows, "1", "3"));
 
     List<List<Object>> snapshotContinueParquet =
-        ((ParquetChunkData) innerBufferOnErrorContinue.getSnapshot("fake/filePath").get()).rows;
+        ((ParquetChunkData) innerBufferOnErrorContinue.getSnapshot().get()).rows;
     // validRows and only the good row from mixedRows are in the buffer
     Assert.assertEquals(2, snapshotContinueParquet.size());
     Assert.assertEquals(Arrays.asList("a"), snapshotContinueParquet.get(0));
     Assert.assertEquals(Arrays.asList("b"), snapshotContinueParquet.get(1));
 
     List<List<Object>> snapshotAbortParquet =
-        ((ParquetChunkData) innerBufferOnErrorAbort.getSnapshot("fake/filePath").get()).rows;
+        ((ParquetChunkData) innerBufferOnErrorAbort.getSnapshot().get()).rows;
     // only validRows and none of the mixedRows are in the buffer
     Assert.assertEquals(1, snapshotAbortParquet.size());
     Assert.assertEquals(Arrays.asList("a"), snapshotAbortParquet.get(0));
@@ -1733,9 +1727,55 @@ public void testOnErrorAbortRowsWithError() {
 
   @Test
   public void testParquetChunkMetadataCreationIsThreadSafe() throws InterruptedException {
-    final String testFileA = "testFileA";
-    final String testFileB = "testFileB";
+    final ParquetRowBuffer bufferUnderTest =
+        (ParquetRowBuffer) createTestBuffer(OpenChannelRequest.OnErrorOption.CONTINUE);
+
+    final int columnOrdinal = 1;
+    final ColumnMetadata colChar1 = new ColumnMetadata();
+    colChar1.setOrdinal(columnOrdinal);
+    colChar1.setName("COLCHAR");
+    colChar1.setPhysicalType("LOB");
+    colChar1.setNullable(true);
+    colChar1.setLogicalType("TEXT");
+    colChar1.setByteLength(14);
+    colChar1.setLength(11);
+    colChar1.setScale(0);
+
+    final ColumnMetadata colChar2 = new ColumnMetadata();
+    colChar2.setOrdinal(columnOrdinal);
+    colChar2.setName("COLCHAR");
+    colChar2.setPhysicalType("SB1");
+    colChar2.setNullable(true);
+    colChar2.setLogicalType("TEXT");
+    colChar2.setByteLength(14);
+    colChar2.setLength(11);
+    colChar2.setScale(0);
+
+    bufferUnderTest.setupSchema(Collections.singletonList(colChar1));
+
+    loadData(bufferUnderTest, Collections.singletonMap("colChar", "a"));
 
+    final CountDownLatch latch = new CountDownLatch(1);
+    final AtomicReference<ChannelData<ParquetChunkData>> firstFlushResult = new AtomicReference<>();
+    final Thread t =
+        getThreadThatWaitsForLockReleaseAndFlushes(bufferUnderTest, latch, firstFlushResult);
+    t.start();
+
+    final ChannelData<ParquetChunkData> secondFlushResult = bufferUnderTest.flush();
+    bufferUnderTest.setupSchema(Collections.singletonList(colChar2));
+
+    latch.countDown();
+    t.join();
+
+    // The logical and physical types should be different
+    Assert.assertNotEquals(
+        getColumnType(firstFlushResult.get(), columnOrdinal),
+        getColumnType(secondFlushResult, columnOrdinal));
+  }
+
+  @Test
+  public void testParquetFileNameMetadata() throws IOException {
+    String filePath = "testParquetFileNameMetadata.bdec";
     final ParquetRowBuffer bufferUnderTest =
         (ParquetRowBuffer) createTestBuffer(OpenChannelRequest.OnErrorOption.CONTINUE);
 
@@ -1750,30 +1790,20 @@ public void testParquetChunkMetadataCreationIsThreadSafe() throws InterruptedExc
     colChar.setScale(0);
 
     bufferUnderTest.setupSchema(Collections.singletonList(colChar));
-
     loadData(bufferUnderTest, Collections.singletonMap("colChar", "a"));
+    ChannelData<ParquetChunkData> data = bufferUnderTest.flush();
+    data.setChannelContext(new ChannelFlushContext("name", "db", "schema", "table", 1L, "key", 0L));
 
-    final CountDownLatch latch = new CountDownLatch(1);
-    final AtomicReference<ChannelData<ParquetChunkData>> firstFlushResult = new AtomicReference<>();
-    final Thread t =
-        getThreadThatWaitsForLockReleaseAndFlushes(
-            bufferUnderTest, testFileA, latch, firstFlushResult);
-    t.start();
-
-    final ChannelData<ParquetChunkData> secondFlushResult = bufferUnderTest.flush(testFileB);
-    Assert.assertEquals(testFileB, getPrimaryFileId(secondFlushResult));
-
-    latch.countDown();
-    t.join();
+    ParquetFlusher flusher = (ParquetFlusher) bufferUnderTest.createFlusher();
+    Flusher.SerializationResult result =
+        flusher.serialize(Collections.singletonList(data), filePath);
 
-    Assert.assertNotNull(firstFlushResult.get());
-    Assert.assertEquals(testFileA, getPrimaryFileId(firstFlushResult.get()));
-    Assert.assertEquals(testFileB, getPrimaryFileId(secondFlushResult));
+    BdecParquetReader reader = new BdecParquetReader(result.chunkData.toByteArray());
+    Assert.assertEquals(filePath, reader.getKeyValueMetadata().get(Constants.PRIMARY_FILE_ID_KEY));
   }
 
   private static Thread getThreadThatWaitsForLockReleaseAndFlushes(
       final ParquetRowBuffer bufferUnderTest,
-      final String filenameToFlush,
       final CountDownLatch latch,
       final AtomicReference<ChannelData<ParquetChunkData>> flushResult) {
     return new Thread(
@@ -1785,8 +1815,7 @@ private static Thread getThreadThatWaitsForLockReleaseAndFlushes(
           }
 
           final ChannelData<ParquetChunkData> flush =
-              loadData(bufferUnderTest, Collections.singletonMap("colChar", "b"))
-                  .flush(filenameToFlush);
+              loadData(bufferUnderTest, Collections.singletonMap("colChar", "b")).flush();
           flushResult.set(flush);
         });
   }
@@ -1801,7 +1830,7 @@ private static ParquetRowBuffer loadData(
     return bufferToLoad;
   }
 
-  private static String getPrimaryFileId(final ChannelData<ParquetChunkData> chunkData) {
-    return chunkData.getVectors().metadata.get(Constants.PRIMARY_FILE_ID_KEY);
+  private static String getColumnType(final ChannelData<ParquetChunkData> chunkData, int columnId) {
+    return chunkData.getVectors().metadata.get(Integer.toString(columnId));
   }
 }
diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelTest.java
index dc3285df8..9495b133e 100644
--- a/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelTest.java
+++ b/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelTest.java
@@ -592,7 +592,7 @@ public void testInsertRow() {
     row.put("col", 1);
 
     // Get data before insert to verify that there is no row (data should be null)
-    ChannelData<?> data = channel.getData("my_snowpipe_streaming.bdec");
+    ChannelData<?> data = channel.getData();
     Assert.assertNull(data);
 
     long insertStartTimeInMs = System.currentTimeMillis();
@@ -605,7 +605,7 @@ public void testInsertRow() {
     long insertEndTimeInMs = System.currentTimeMillis();
 
     // Get data again to verify the row is inserted
-    data = channel.getData("my_snowpipe_streaming.bdec");
+    data = channel.getData();
     Assert.assertEquals(3, data.getRowCount());
     Assert.assertEquals((Long) 1L, data.getRowSequencer());
     Assert.assertEquals(1, ((ChannelData<ParquetChunkData>) data).getVectors().rows.get(0).size());
diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtilsIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtilsIT.java
index b40cdad82..67a70f6fc 100644
--- a/src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtilsIT.java
+++ b/src/test/java/net/snowflake/ingest/streaming/internal/StreamingIngestUtilsIT.java
@@ -27,7 +27,10 @@
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest({JWTManager.class})
-@PowerMockIgnore({"javax.net.ssl.*"}) /* Avoid ssl related exception from power mockito*/
+@PowerMockIgnore({
+  "javax.net.ssl.*",
+  "javax.security.*"
+}) /* Avoid ssl related exception from power mockito*/
 public class StreamingIngestUtilsIT {
   @Test
   public void testJWTRetries() throws Exception {