SNOW-1465503 Check row count in Parquet footer before committing #784

sfc-gh-lsembera · 2024-06-25T14:19:17Z

This PR implements an additional safety check if the number of rows in the Parquet footer matches the number of metadata rows we've collected. If not, an internal exception is thrown.

sfc-gh-psaha · 2024-06-27T21:55:00Z

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

+
+      // We check if the number of rows collectively written for channels encountered so far matches
+      // the number of rows in metadata
+      if (mergedChannelWriter.getRowsWritten() != rowCount) {


Can this be potentially racy i.e. other threads writing data via the writer concurrently? Or is the mergedChannelWriter only used from this current thread?

This cannot be racy, serialization happens in a single BG thread and no other thread can modify the buffer anymore.

sfc-gh-psaha · 2024-06-27T21:57:09Z

src/main/java/org/apache/parquet/hadoop/BdecParquetWriter.java

@@ -35,6 +36,7 @@
 public class BdecParquetWriter implements AutoCloseable {
  private final InternalParquetRecordWriter<List<Object>> writer;
  private final CodecFactory codecFactory;
+  private final AtomicLong rowsWritten = new AtomicLong(0);


Can it be just a volatile? This goes back to my previous question about races. If there is no concurrent access (but non-overlapping access from different threads), then volatile would be sufficient.

In fact, in this case, we don't even need volatile. The field is accessed either within a lock, or in a BG thread with happens-before relationship with the spawning thread.

sfc-gh-gdoci · 2024-06-28T09:09:42Z

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

+  /**
+   * Safety check to verify whether the number of rows in the parquet footer matches the number of
+   * rows in metadata
+   */
+  static <T> void verifyRowCounts(
+      Flusher.SerializationResult serializationResult,
+      List<ChannelData<T>> channelsDataPerTable,
+      byte[] paddedChunkData,
+      int chunkLength) {


Another idea would be to add the check in the Parquet flusher inside serializeFromX methods right before we return the serialization result. We can get the per block row counts from the parquetWriter (through a new method) directly since the writer has access to the footer. So we wouldn't need to read and create a new BdecInputFile.

Thanks, this simplifies the code pretty significantly. I updated the PR.

sfc-gh-tjones · 2024-06-28T21:10:51Z

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

+   * @param serializationType Serialization type, used for logging purposes only
+   * @param writer Parquet writer writing the data
+   * @param channelsDataPerTable Channel data
+   * @param javaSerializationTotalRowCount Total row count when java object serialization is used.


What do you mean by java object serialization is used - looks like above this is just the total row count?

The SDK implements two types of row buffering. The default one, which buffers rows as lists of java objects in memory and during flush it iterates over all of them and writes them to Parquet. There is also an alternative implementation called internal buffering implemented by @sfc-gh-azagrebin, which doesn't collect any Java values, but which instead directly serializes rows to Parquet. So, for the default buffering method (java objects), we have 4 sources of row counts that all have to match:

size of the buffered List<List<Object>>

number of rows written by the Parquet writer

number of rows in Parquet footer

number of rows collected for metadata

For the internal buffering optimization, we only have 2nd, 3rd and 4th.

sfc-gh-gdoci

Thank you, lgtm.

src/main/java/org/apache/parquet/hadoop/BdecParquetWriter.java

sfc-gh-azagrebin

Thanks Lukas!

sfc-gh-azagrebin · 2024-07-01T11:55:58Z

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

+    if (parquetTotalRowsInFooter != totalRowsInMetadata
+        || parquetTotalRowsWritten != totalRowsInMetadata) {
+
+      final String perChannelRowCountsInMetadata =


would it make sense to dump the whole List<ChannelData<ParquetChunkData>>, if ChannelData/ParquetChunkData get the proper toString methods? because it will be not clear what channel misbehaved.

I added a log line with channel names. We should not log full channel data because it contains user data.

sfc-gh-azagrebin · 2024-07-01T11:57:45Z

src/main/java/org/apache/parquet/hadoop/BdecParquetWriter.java

@@ -35,6 +37,7 @@
 public class BdecParquetWriter implements AutoCloseable {
  private final InternalParquetRecordWriter<List<Object>> writer;
  private final CodecFactory codecFactory;
+  private long rowsWritten = 0;


rowsWritten = 0;

also in close?

Closing the writer doesn't clean up the written rows, so se should reset the counter. For example, the Parquet footer is only accessible after closing the writer.

sfc-gh-azagrebin · 2024-07-01T12:08:22Z

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

@@ -216,6 +219,9 @@ private SerializationResult serializeFromJavaObjects(
    rows.forEach(parquetWriter::writeRow);
    parquetWriter.close();

+    this.verifyRowCounts(
+        "serializeFromJavaObjects", parquetWriter, channelsDataPerTable, rows.size());


it would be good to use rowCount calculated here outside of verifyRowCounts for the check because this is what we eventually send to GS.

sfc-gh-tzhang

LGTM, please make sure the error is being sent to SF to avoid the case where customer doesn't have the log, thanks!

sfc-gh-tzhang · 2024-07-02T01:46:19Z

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

+
+      final long channelsCountInMetadata = channelsDataPerTable.size();
+
+      throw new SFException(


Could you make sure this is being sent to Snowflake?

What do you mean by sent to Snowflake? Is it possible via telemetry?

sfc-gh-tzhang · 2024-07-02T01:48:56Z

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

+    for (ChannelData<ParquetChunkData> channelData : channelsDataPerTable)
+      totalRowsInMetadata += channelData.getRowCount();


IIRC the best practice recommends to always add curly brackets

Suggested change

for (ChannelData<ParquetChunkData> channelData : channelsDataPerTable)

totalRowsInMetadata += channelData.getRowCount();

for (ChannelData<ParquetChunkData> channelData : channelsDataPerTable)

{

totalRowsInMetadata += channelData.getRowCount();

}

…wflakedb#784)

sfc-gh-lsembera force-pushed the lsembera/row-count-check branch from 02de7dc to 0d7d47c Compare June 26, 2024 11:41

sfc-gh-lsembera marked this pull request as ready for review June 26, 2024 11:48

sfc-gh-lsembera requested review from sfc-gh-tzhang and a team as code owners June 26, 2024 11:48

sfc-gh-psaha reviewed Jun 27, 2024

View reviewed changes

sfc-gh-gdoci reviewed Jun 28, 2024

View reviewed changes

sfc-gh-lsembera force-pushed the lsembera/row-count-check branch 2 times, most recently from 75e6f66 to c4ae63c Compare June 28, 2024 20:44

sfc-gh-tjones reviewed Jun 28, 2024

View reviewed changes

sfc-gh-gdoci approved these changes Jul 1, 2024

View reviewed changes

src/main/java/org/apache/parquet/hadoop/BdecParquetWriter.java Outdated Show resolved Hide resolved

sfc-gh-lsembera added 2 commits July 1, 2024 10:32

SNOW-1465503 Check row count in Parquet footer before committing

d84cb68

parquetFooterRowsPerBlock

d39ca44

sfc-gh-lsembera force-pushed the lsembera/row-count-check branch from 1b240e6 to d39ca44 Compare July 1, 2024 10:32

sfc-gh-azagrebin approved these changes Jul 1, 2024

View reviewed changes

sfc-gh-azagrebin reviewed Jul 1, 2024

View reviewed changes

sfc-gh-tzhang approved these changes Jul 2, 2024

View reviewed changes

sfc-gh-lsembera added 2 commits July 2, 2024 09:51

code review comments

b77a00c

Fix test

a15eb41

sfc-gh-lsembera merged commit 9254771 into master Jul 8, 2024
48 checks passed

sfc-gh-lsembera deleted the lsembera/row-count-check branch July 8, 2024 09:20

sfc-gh-kgaputis pushed a commit to sfc-gh-kgaputis/snowflake-ingest-java that referenced this pull request Sep 12, 2024

SNOW-1465503 Check row count in Parquet footer before committing (sno…

b838014

…wflakedb#784)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SNOW-1465503 Check row count in Parquet footer before committing #784

SNOW-1465503 Check row count in Parquet footer before committing #784

sfc-gh-lsembera commented Jun 25, 2024 •

edited

Loading

sfc-gh-psaha Jun 27, 2024

sfc-gh-lsembera Jun 28, 2024

sfc-gh-psaha Jun 27, 2024

sfc-gh-lsembera Jun 28, 2024

sfc-gh-gdoci Jun 28, 2024

sfc-gh-lsembera Jun 28, 2024

sfc-gh-tjones Jun 28, 2024

sfc-gh-lsembera Jun 28, 2024

sfc-gh-gdoci left a comment

sfc-gh-azagrebin left a comment

sfc-gh-azagrebin Jul 1, 2024

sfc-gh-lsembera Jul 2, 2024

sfc-gh-azagrebin Jul 1, 2024

sfc-gh-lsembera Jul 2, 2024

sfc-gh-azagrebin Jul 1, 2024 •

edited

Loading

sfc-gh-tzhang left a comment

sfc-gh-tzhang Jul 2, 2024

sfc-gh-lsembera Jul 2, 2024

sfc-gh-tzhang Jul 2, 2024


		final long channelsCountInMetadata = channelsDataPerTable.size();

		throw new SFException(

		for (ChannelData<ParquetChunkData> channelData : channelsDataPerTable)
		totalRowsInMetadata += channelData.getRowCount();

SNOW-1465503 Check row count in Parquet footer before committing #784

SNOW-1465503 Check row count in Parquet footer before committing #784

Conversation

sfc-gh-lsembera commented Jun 25, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-gdoci left a comment

Choose a reason for hiding this comment

sfc-gh-azagrebin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-azagrebin Jul 1, 2024 • edited Loading

Choose a reason for hiding this comment

sfc-gh-tzhang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sfc-gh-lsembera commented Jun 25, 2024 •

edited

Loading

sfc-gh-azagrebin Jul 1, 2024 •

edited

Loading