SNOW-1748333 Fix Iceberg decimal type schema parser (#864)

snowflakedb · Oct 21, 2024 · 1fe05a5 · 1fe05a5
1 parent 513b589
commit 1fe05a5
Show file tree

Hide file tree

Showing 3 changed files with 77 additions and 6 deletions.
diff --git a/pom.xml b/pom.xml
@@ -154,11 +154,6 @@
  <artifactId>commons-lang3</artifactId>
  <version>${commonslang3.version}</version>
  </dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-text</artifactId>
- <version>${commonstext.version}</version>
- </dependency>
  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>
@@ -383,6 +378,12 @@
  <version>1.14.9</version>
  <scope>test</scope>
  </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-text</artifactId>
+ <version>${commonstext.version}</version>
+ <scope>test</scope>
+ </dependency>
  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-mapreduce-client-core</artifactId>
@@ -543,6 +544,10 @@
  <groupId>org.apache.commons</groupId>
  <artifactId>commons-configuration2</artifactId>
  </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-text</artifactId>
+ </dependency>
  <dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-common</artifactId>

diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java
@@ -86,7 +86,9 @@ public class ParquetRowBuffer extends AbstractRowBuffer<ParquetChunkData> {
  public void setupSchema(List<ColumnMetadata> columns) {
  fieldIndex.clear();
  metadata.clear();
- metadata.put("sfVer", "1,1");
+ if (!clientBufferParameters.getIsIcebergMode()) {
+ metadata.put("sfVer", "1,1");
+ }
  List<Type> parquetTypes = new ArrayList<>();
  int id = 1;
 

diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergNumericTypesIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergNumericTypesIT.java
@@ -5,15 +5,21 @@
 package net.snowflake.ingest.streaming.internal.datatypes;
 
 import java.math.BigDecimal;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
 import net.snowflake.ingest.utils.Constants;
 import net.snowflake.ingest.utils.ErrorCode;
 import net.snowflake.ingest.utils.SFException;
+import org.apache.commons.text.RandomStringGenerator;
 import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Ignore("This test can be enabled after server side Iceberg EP support is released")
 public class IcebergNumericTypesIT extends AbstractDataTypeTest {
@@ -30,9 +36,21 @@ public static Object[][] parameters() {
  @Parameterized.Parameter(1)
  public static Constants.IcebergSerializationPolicy icebergSerializationPolicy;
 
+ private static final Logger logger = LoggerFactory.getLogger(IcebergNumericTypesIT.class);
+ private static Random generator;
+ private static RandomStringGenerator randomStringGenerator;
+
  @Before
  public void before() throws Exception {
  super.beforeIceberg(compressionAlgorithm, icebergSerializationPolicy);
+ long seed = System.currentTimeMillis();
+ logger.info("Random seed: {}", seed);
+ generator = new Random(seed);
+ randomStringGenerator =
+ new RandomStringGenerator.Builder()
+ .usingRandom(generator::nextInt)
+ .withinRange('0', '9')
+ .build();
  }
 
  @Test
@@ -310,6 +328,7 @@ public void testDecimal() throws Exception {
  testIcebergIngestion("decimal(3, 1)", 12.5f, new FloatProvider());
  testIcebergIngestion("decimal(3, 1)", -99, new IntProvider());
  testIcebergIngestion("decimal(38, 0)", Long.MAX_VALUE, new LongProvider());
+ testIcebergIngestion("decimal(21, 0)", .0, new DoubleProvider());
  testIcebergIngestion("decimal(38, 10)", null, new BigDecimalProvider());
 
  testIcebergIngestion(
@@ -372,5 +391,50 @@ public void testDecimalAndQueries() throws Exception {
  Arrays.asList(new BigDecimal("-12.3"), new BigDecimal("-12.3"), null),
  "select COUNT({columnName}) from {tableName} where {columnName} = -12.3",
  Arrays.asList(2L));
+
+ List<Object> bigDecimals_9_4 = randomBigDecimal(200, 9, 4);
+ testIcebergIngestAndQuery(
+ "decimal(9, 4)", bigDecimals_9_4, "select {columnName} from {tableName}", bigDecimals_9_4);
+
+ List<Object> bigDecimals_18_9 = randomBigDecimal(200, 18, 9);
+ testIcebergIngestAndQuery(
+ "decimal(18, 9)",
+ bigDecimals_18_9,
+ "select {columnName} from {tableName}",
+ bigDecimals_18_9);
+
+ List<Object> bigDecimals_21_0 = randomBigDecimal(200, 21, 0);
+ testIcebergIngestAndQuery(
+ "decimal(21, 0)",
+ bigDecimals_21_0,
+ "select {columnName} from {tableName}",
+ bigDecimals_21_0);
+
+ List<Object> bigDecimals_38_10 = randomBigDecimal(200, 38, 10);
+ testIcebergIngestAndQuery(
+ "decimal(38, 10)",
+ bigDecimals_38_10,
+ "select {columnName} from {tableName}",
+ bigDecimals_38_10);
+ }
+
+ /** Generate a list of random BigDecimal(p', s') where p' <= precision and s' <= scale */
+ private static List<Object> randomBigDecimal(int count, int precision, int scale) {
+ List<Object> list = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ int intPart = generator.nextInt(precision - scale + 1);
+ int floatPart = generator.nextInt(scale + 1);
+ if (intPart == 0 && floatPart == 0) {
+ list.add(null);
+ continue;
+ }
+ list.add(
+ new BigDecimal(
+ (generator.nextBoolean() ? "-" : "")
+ + randomStringGenerator.generate(intPart)
+ + "."
+ + randomStringGenerator.generate(floatPart)));
+ }
+ return list;
  }
 }