SNOW-1727532 Set number of values for repeated fields #861

sfc-gh-alhuang · 2024-10-11T17:44:23Z

This PR includes following change:

Set numberOfValues for repeated fields (e.g. map, list) and used it in EP file registration.
Increment null count for all sub-columns of a null column.
Use hash for approximate NDV for string and bytes.
Fix NPE when inserting null to a structured column, throwing SF custom exception instead.
Added extendedMetadataSize and metadataSize fields for EP registration.
Address comments of SNOW-1708577 Parquet V2 support for new table format #851

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

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

sfc-gh-hmadan · 2024-10-16T03:29:39Z

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

    this.maxChunkSizeInBytes = maxChunkSizeInBytes;
    this.maxAllowedRowSizeInBytes = maxAllowedRowSizeInBytes;
    this.bdecParquetCompression = bdecParquetCompression;
    this.enableNewJsonParsingLogic = enableNewJsonParsingLogic;
    this.maxRowGroups = maxRowGroups;
    this.isIcebergMode = isIcebergMode;
+    this.enableDistinctValuesCount = enableDistinctValuesCount;
+    this.enableValuesCount = enableValuesCount;
  }

  /** @param clientInternal reference to the client object where the relevant parameters are set */


I've never understood why clientInternal can or should be allowed to be null. Looks like a testcase author leaking their convenience needs into production code.

The clientInterTest need this.

there's another ctor for testcases to pass in whatever booleans they want, so still unclear why there's a need to pass in a null clientInternal to this ctor.

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

sfc-gh-hmadan · 2024-10-16T17:00:22Z

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

+    final int magicOffset = bytes.length - ParquetFileWriter.MAGIC.length;
+    final int footerSizeOffset = magicOffset - Integer.BYTES;
+    if (footerSizeOffset < 0
+        || !ParquetFileWriter.MAGIC_STR.equals(


curious why is this MAGIC_STR comparison being added only now / why is it needed only when calculating ext metadta size?

Addressing this comment

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

sfc-gh-hmadan · 2024-10-16T17:19:41Z

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

@@ -427,7 +432,7 @@ Set<String> verifyInputColumns(
    List<String> missingCols = new ArrayList<>();
    for (String columnName : this.nonNullableFieldNames) {
      if (!inputColNamesMap.containsKey(columnName)) {
-        missingCols.add(statsMap.get(columnName).getColumnDisplayName());
+        missingCols.add(fieldIndex.get(columnName).columnMetadata.getName());


can we remove the getColumnDisplayName() method from the RowBufferStats class now?

actually, is there a need to even make this change?

If a root column is a structured data type, statsMap.get(columnName) will trow NPE as we only stores leaf column stats. Use fieldIndex instead for logging to avoid this.

please add this as a code comment.

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

sfc-gh-hmadan · 2024-10-16T17:24:58Z

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

@@ -49,6 +48,7 @@ public class ParquetRowBuffer extends AbstractRowBuffer<ParquetChunkData> {
  private final ParquetProperties.WriterVersion parquetWriterVersion;

  private MessageType schema;
+  private SubColumnFinder statsFinder;


rename to subColumnFinder

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

sfc-gh-hmadan · 2024-10-16T17:37:05Z

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

-      statsMap.get(columnName).incCurrentNullCount();
+      statsFinder
+          .getSubColumns(columnName)
+          .forEach(subColumn -> statsMap.get(subColumn).incCurrentNullCount());


nullref if statsMap.get(subColumn) returns null.
can that happen?

No as both subColumnFinder and statsMap are built from the same schema. Added check here in case.

sfc-gh-hmadan · 2024-10-17T22:39:02Z

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

+   * @return the footer size
+   */
+  public static long getParquetFooterSize(byte[] bytes) throws IOException {
+    final int magicOffset = bytes.length - ParquetFileWriter.MAGIC.length;


curious, does this handle both PARE and PAR1 ?
I guess they're both 4 bytes so the length won't change.

Yes they have the same length. Added check for PARE.

sfc-gh-hmadan · 2024-10-17T22:40:52Z

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

+              new String(bytes, magicOffset, ParquetFileWriter.MAGIC.length)));
+    }
+
+    return BytesUtils.readIntLittleEndian(bytes, footerSizeOffset);


is it always little endian irrespective of parquet V1 and parquet V2? How can we add a test that ensures this method is correct?

Yes, the V1 and V2 only affects data page. (ref). For testing, I think we can use serializedBytes.length - writer.getColumnIndexReference().getOffset() - metadataSize - ParquetFileWriter.MAGIC.length - footerSizeBytes(4) = extenedMetadataSize. What do you think?

Added test for parquet footer size & extendedMetadataSize in BlobBuilderTest

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

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

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

sfc-gh-hmadan · 2024-10-18T06:40:06Z

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

@@ -313,9 +352,27 @@ private float addRow(
          RowBufferStats.getCombinedStats(statsMap.get(columnName), forkedColStats.getValue()));


oh wow this looks like a thread unsafe operation here. RowBuffer.statsMap and tempStatsMap are both constructed as non-concurrent collections, but we don't prohibit concurrent insertRows on one channel!

(nothig to fix for now we'll followup on this separately)

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

sfc-gh-hmadan

LGTM!

sfc-gh-alhuang requested a review from sfc-gh-hmadan October 11, 2024 17:44

sfc-gh-alhuang force-pushed the alhuang/ndv branch from 8785341 to 28c8435 Compare October 11, 2024 23:14

Base automatically changed from alhuang/parquetV2 to master October 14, 2024 21:16

done

bef322d

sfc-gh-alhuang force-pushed the alhuang/ndv branch from 28c8435 to 19d66e4 Compare October 15, 2024 22:45

sfc-gh-alhuang marked this pull request as ready for review October 15, 2024 22:45

sfc-gh-alhuang requested review from sfc-gh-tzhang and a team as code owners October 15, 2024 22:45

sfc-gh-alhuang force-pushed the alhuang/ndv branch 2 times, most recently from ae10eee to 9c77f1b Compare October 15, 2024 22:50

fix npe

4d9c70b

sfc-gh-alhuang force-pushed the alhuang/ndv branch from 9c77f1b to 4d9c70b Compare October 16, 2024 00:07