Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

SNOW-1764301 Iceberg parquet file configuration #874

Merged
merged 5 commits into from
Oct 29, 2024
Merged

SNOW-1764301 Iceberg parquet file configuration #874

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8c47343
done
sfc-gh-alhuang Oct 23, 2024
61656f0
address comments
sfc-gh-alhuang Oct 25, 2024
7bec169
done
sfc-gh-alhuang Oct 25, 2024
f6e0d2a
Merge branch 'master' into alhuang-parquet-config
sfc-gh-alhuang Oct 29, 2024
487d6e7
format
sfc-gh-alhuang Oct 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions src/main/java/net/snowflake/ingest/streaming/internal/ClientBufferParameters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import java.util.Optional;
import net.snowflake.ingest.utils.Constants;
import net.snowflake.ingest.utils.ParameterProvider;
import org.apache.parquet.column.ParquetProperties;

/** Channel's buffer relevant parameters that are set at the owning client level. */
public class ClientBufferParameters {
Expand All @@ -29,6 +30,8 @@ public class ClientBufferParameters {

private boolean enableValuesCount;

private boolean enableDictionaryEncoding;

/**
* Private constructor used for test methods
*
Expand All @@ -53,10 +56,13 @@ private ClientBufferParameters(
this.enableIcebergStreaming = enableIcebergStreaming;
this.enableDistinctValuesCount = enableDistinctValuesCount;
this.enableValuesCount = enableValuesCount;
this.enableDictionaryEncoding = enableIcebergStreaming;
}

/** @param clientInternal reference to the client object where the relevant parameters are set */
public ClientBufferParameters(SnowflakeStreamingIngestClientInternal clientInternal) {
public ClientBufferParameters(
SnowflakeStreamingIngestClientInternal clientInternal,
ParquetProperties.WriterVersion parquetWriterVersion) {
this.maxChunkSizeInBytes =
clientInternal != null
? clientInternal.getParameterProvider().getMaxChunkSizeInBytes()
Expand Down Expand Up @@ -89,6 +95,9 @@ public ClientBufferParameters(SnowflakeStreamingIngestClientInternal clientInter
clientInternal != null
? clientInternal.getInternalParameterProvider().isEnableValuesCount()
: InternalParameterProvider.ENABLE_VALUES_COUNT_DEFAULT;
this.enableDictionaryEncoding =
Copy link
Collaborator

@sfc-gh-hmadan sfc-gh-hmadan Oct 28, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not very satisfied by how there's now a proliferation of places that control runtime behavior :(
There's ParameterProvider, internalParameterProvider, clientBuferParameters, and some completely unnecesary defaulting for when clientInternal is null (it should never be not even in tests).

Tried to come up with a better answer but nothing that sticks and is relevant to your current PR.

As a followup whenever you have time i'd suggest first removing the null defaulting, remove the "isTestMode" thing from everywhere (fully doable after a round of changes I did for the ExternalVolume class). Lets see how things look after these two!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added jira. Will merge this after the #874

enableIcebergStreaming
&& parquetWriterVersion == ParquetProperties.WriterVersion.PARQUET_2_0;
}

/**
Expand Down Expand Up @@ -154,6 +163,6 @@ public boolean isEnableValuesCount() {
}

public boolean isEnableDictionaryEncoding() {
return enableIcebergStreaming;
return enableDictionaryEncoding;
}
}
2 changes: 1 addition & 1 deletion ...java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestChannelInternal.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ class SnowflakeStreamingIngestChannelInternal<T> implements SnowflakeStreamingIn
getFullyQualifiedName(),
this::collectRowSize,
channelState,
new ClientBufferParameters(owningClient),
new ClientBufferParameters(owningClient, parquetWriterVersion),
offsetTokenVerificationFunction,
parquetWriterVersion,
owningClient.getTelemetryService());
Expand Down
12 changes: 10 additions & 2 deletions src/main/java/net/snowflake/ingest/utils/ParameterProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ public class ParameterProvider {
public static final Constants.BdecParquetCompression BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT =
Constants.BdecParquetCompression.GZIP;

public static final Constants.BdecParquetCompression
ICEBERG_PARQUET_COMPRESSION_ALGORITHM_DEFAULT = Constants.BdecParquetCompression.ZSTD;

/* 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_ICEBERG_MODE_DEFAULT = 1;

Expand Down Expand Up @@ -253,7 +256,9 @@ private void setParameterMap(Map<String, Object> parameterOverrides, Properties

this.checkAndUpdate(
BDEC_PARQUET_COMPRESSION_ALGORITHM,
BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT,
isEnableIcebergStreaming()
? ICEBERG_PARQUET_COMPRESSION_ALGORITHM_DEFAULT
: BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT,
parameterOverrides,
props,
false /* enforceDefault */);
Expand Down Expand Up @@ -488,7 +493,10 @@ public int getMaxChunksInRegistrationRequest() {
public Constants.BdecParquetCompression getBdecParquetCompressionAlgorithm() {
Object val =
this.parameterMap.getOrDefault(
BDEC_PARQUET_COMPRESSION_ALGORITHM, BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT);
BDEC_PARQUET_COMPRESSION_ALGORITHM,
isEnableIcebergStreaming()
? ICEBERG_PARQUET_COMPRESSION_ALGORITHM_DEFAULT
: BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT);
if (val instanceof Constants.BdecParquetCompression) {
return (Constants.BdecParquetCompression) val;
}
Expand Down
4 changes: 3 additions & 1 deletion src/test/java/net/snowflake/ingest/streaming/internal/ParameterProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,7 +158,9 @@ public void withDefaultValues() {
ParameterProvider.MAX_CHANNEL_SIZE_IN_BYTES_DEFAULT,
parameterProvider.getMaxChannelSizeInBytes());
Assert.assertEquals(
ParameterProvider.BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT,
enableIcebergStreaming
? ParameterProvider.ICEBERG_PARQUET_COMPRESSION_ALGORITHM_DEFAULT
: ParameterProvider.BDEC_PARQUET_COMPRESSION_ALGORITHM_DEFAULT,
parameterProvider.getBdecParquetCompressionAlgorithm());
Assert.assertEquals(
enableIcebergStreaming
Expand Down
Loading