diff --git a/pom.xml b/pom.xml index 1725ac89a..f0023e815 100644 --- a/pom.xml +++ b/pom.xml @@ -415,6 +415,12 @@ + + org.assertj + assertj-core + 3.26.3 + test + org.mockito mockito-core @@ -634,6 +640,11 @@ commons-lang3 test + + org.assertj + assertj-core + test + org.hamcrest hamcrest-core 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 f6c26f9bd..7b0adaabc 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/AbstractRowBuffer.java @@ -641,15 +641,16 @@ public synchronized void close(String name) { * * @param rowCount: count of rows in the given buffer * @param colStats: map of column name to RowBufferStats - * @param setDefaultValues: whether to set default values for null fields the EPs + * @param setAllDefaultValues: whether to set default values for all null fields the EPs + * irrespective of the data type of this column * @return the EPs built from column stats */ static EpInfo buildEpInfoFromStats( - long rowCount, Map colStats, boolean setDefaultValues) { + long rowCount, Map colStats, boolean setAllDefaultValues) { EpInfo epInfo = new EpInfo(rowCount, new HashMap<>()); for (Map.Entry colStat : colStats.entrySet()) { RowBufferStats stat = colStat.getValue(); - FileColumnProperties dto = new FileColumnProperties(stat, setDefaultValues); + FileColumnProperties dto = new FileColumnProperties(stat, setAllDefaultValues); String colName = colStat.getValue().getColumnDisplayName(); epInfo.getColumnEps().put(colName, dto); } diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java b/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java index edc8fd4c9..307093d3e 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/BlobBuilder.java @@ -134,7 +134,7 @@ static Blob constructBlobAndMetadata( AbstractRowBuffer.buildEpInfoFromStats( serializedChunk.rowCount, serializedChunk.columnEpStatsMapCombined, - internalParameterProvider.setDefaultValuesInEp())) + internalParameterProvider.setAllDefaultValuesInEp())) .setFirstInsertTimeInMs(serializedChunk.chunkMinMaxInsertTimeInMs.getFirst()) .setLastInsertTimeInMs(serializedChunk.chunkMinMaxInsertTimeInMs.getSecond()); diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java b/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java index 8d8bff3f5..035a88373 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/DataValidationUtil.java @@ -1150,7 +1150,7 @@ static void checkValueInRange( static void checkFixedLengthByteArray(byte[] bytes, int length, final long insertRowIndex) { if (bytes.length != length) { throw new SFException( - ErrorCode.INVALID_FORMAT_ROW, + ErrorCode.INVALID_VALUE_ROW, String.format( "Binary length mismatch: expected=%d, actual=%d, rowIndex:%d", length, bytes.length, insertRowIndex)); diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/FileColumnProperties.java b/src/main/java/net/snowflake/ingest/streaming/internal/FileColumnProperties.java index 3a8dbc2b6..b3c7aedf5 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/FileColumnProperties.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/FileColumnProperties.java @@ -10,6 +10,9 @@ import com.fasterxml.jackson.annotation.JsonProperty; import java.math.BigInteger; import java.util.Objects; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.apache.parquet.schema.LogicalTypeAnnotation; /** Audit register endpoint/FileColumnPropertyDTO property list. */ class FileColumnProperties { @@ -50,45 +53,107 @@ class FileColumnProperties { // Default value to use for min/max real when all data in the given column is NULL public static final Double DEFAULT_MIN_MAX_REAL_VAL_FOR_EP = 0d; - FileColumnProperties(RowBufferStats stats, boolean setDefaultValues) { + // Default value to use for min/max string when all data in the given Iceberg column is NULL + public static final String DEFAULT_MIN_MAX_STR_VAL_FOR_EP = ""; + + /** + * @param setAllDefaultValues Whether to set defaults for ALL fields, or only some. BDEC sets it + * for all but iceberg does not. + */ + FileColumnProperties(RowBufferStats stats, boolean setAllDefaultValues) { this.setColumnOrdinal(stats.getOrdinal()); this.setFieldId(stats.getFieldId()); this.setCollation(stats.getCollationDefinitionString()); + + if (setAllDefaultValues) { + /* Set every column to default value for FDN columns if the all row values are null */ + setIntValues(stats); + setRealValues(stats); + setStringValues(stats, false /* replaceNullWithEmptyString */); + } else { + /* Only set corresponding min/max stats to default value for Iceberg columns if the all row values are null */ + switch (stats.getPrimitiveType().getPrimitiveTypeName()) { + case BOOLEAN: + case INT32: + case INT64: + setIntValues(stats); + break; + + case FLOAT: + case DOUBLE: + setRealValues(stats); + break; + + case BINARY: + setStringValues(stats, true /* replaceNullWithEmptyString */); + break; + + case FIXED_LEN_BYTE_ARRAY: + if (stats.getPrimitiveType().getLogicalTypeAnnotation() + instanceof LogicalTypeAnnotation.DecimalLogicalTypeAnnotation) { + setIntValues(stats); + } else { + setStringValues(stats, true /* replaceNullWithEmptyString */); + } + break; + + default: + throw new SFException( + ErrorCode.INTERNAL_ERROR, + "Unsupported Iceberg column type: " + + stats.getPrimitiveType().getPrimitiveTypeName()); + } + } + + this.setMaxLength(stats.getCurrentMaxLength()); + this.setMaxStrNonCollated(null); + this.setMinStrNonCollated(null); + this.setNullCount(stats.getCurrentNullCount()); + this.setDistinctValues(stats.getDistinctValues()); + } + + private void setIntValues(RowBufferStats stats) { this.setMaxIntValue( stats.getCurrentMaxIntValue() == null - ? (setDefaultValues ? DEFAULT_MIN_MAX_INT_VAL_FOR_EP : null) + ? DEFAULT_MIN_MAX_INT_VAL_FOR_EP : stats.getCurrentMaxIntValue()); + this.setMinIntValue( stats.getCurrentMinIntValue() == null - ? (setDefaultValues ? DEFAULT_MIN_MAX_INT_VAL_FOR_EP : null) + ? DEFAULT_MIN_MAX_INT_VAL_FOR_EP : stats.getCurrentMinIntValue()); - this.setMinRealValue( - stats.getCurrentMinRealValue() == null - ? (setDefaultValues ? DEFAULT_MIN_MAX_REAL_VAL_FOR_EP : null) - : stats.getCurrentMinRealValue()); + } + + private void setRealValues(RowBufferStats stats) { this.setMaxRealValue( stats.getCurrentMaxRealValue() == null - ? (setDefaultValues ? DEFAULT_MIN_MAX_REAL_VAL_FOR_EP : null) + ? DEFAULT_MIN_MAX_REAL_VAL_FOR_EP : stats.getCurrentMaxRealValue()); - this.setMaxLength(stats.getCurrentMaxLength()); - this.setMaxStrNonCollated(null); - this.setMinStrNonCollated(null); - - // current hex-encoded min value, truncated down to 32 bytes - if (stats.getCurrentMinStrValue() != null) { - String truncatedAsHex = truncateBytesAsHex(stats.getCurrentMinStrValue(), false); - this.setMinStrValue(truncatedAsHex); - } + this.setMinRealValue( + stats.getCurrentMinRealValue() == null + ? DEFAULT_MIN_MAX_REAL_VAL_FOR_EP + : stats.getCurrentMinRealValue()); + } + private void setStringValues(RowBufferStats stats, boolean replaceNullWithEmptyString) { // current hex-encoded max value, truncated up to 32 bytes if (stats.getCurrentMaxStrValue() != null) { - String truncatedAsHex = truncateBytesAsHex(stats.getCurrentMaxStrValue(), true); + String truncatedAsHex = + truncateBytesAsHex(stats.getCurrentMaxStrValue(), true /* truncateUp */); this.setMaxStrValue(truncatedAsHex); + } else if (replaceNullWithEmptyString) { + this.setMaxStrValue(DEFAULT_MIN_MAX_STR_VAL_FOR_EP); } - this.setNullCount(stats.getCurrentNullCount()); - this.setDistinctValues(stats.getDistinctValues()); + // current hex-encoded min value, truncated down to 32 bytes + if (stats.getCurrentMinStrValue() != null) { + String truncatedAsHex = + truncateBytesAsHex(stats.getCurrentMinStrValue(), false /* truncateUp */); + this.setMinStrValue(truncatedAsHex); + } else if (replaceNullWithEmptyString) { + this.setMinStrValue(DEFAULT_MIN_MAX_STR_VAL_FOR_EP); + } } @JsonProperty("columnId") diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/InternalParameterProvider.java b/src/main/java/net/snowflake/ingest/streaming/internal/InternalParameterProvider.java index 11a2858f6..4ab8ecc4b 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/InternalParameterProvider.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/InternalParameterProvider.java @@ -20,9 +20,11 @@ boolean getEnableChunkEncryption() { return !isIcebergMode; } - boolean setDefaultValuesInEp() { - // When in Iceberg mode, we need to populate nulls (instead of zeroes) in the minIntValue / - // maxIntValue / minRealValue / maxRealValue fields of the EP Metadata. + boolean setAllDefaultValuesInEp() { + // When in non-iceberg mode, we want to default the stats for all data types (int/real/string) + // to 0 / to "". + // However when in iceberg mode, we want to default only those stats that are + // relevant to the current datatype. return !isIcebergMode; } 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 ed19971ab..339210f65 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/ParquetRowBuffer.java @@ -27,6 +27,7 @@ import net.snowflake.ingest.utils.SFException; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.Type; /** @@ -88,21 +89,26 @@ public void setupSchema(List columns) { /* Set up fields using top level column information */ validateColumnCollation(column); ParquetTypeInfo typeInfo = ParquetTypeGenerator.generateColumnParquetTypeInfo(column, id); - parquetTypes.add(typeInfo.getParquetType()); + Type parquetType = typeInfo.getParquetType(); + parquetTypes.add(parquetType); this.metadata.putAll(typeInfo.getMetadata()); int columnIndex = parquetTypes.size() - 1; - fieldIndex.put( - column.getInternalName(), - new ParquetColumn(column, columnIndex, typeInfo.getParquetType())); + fieldIndex.put(column.getInternalName(), new ParquetColumn(column, columnIndex, parquetType)); + if (!column.getNullable()) { addNonNullableFieldName(column.getInternalName()); } + if (!clientBufferParameters.getIsIcebergMode()) { /* Streaming to FDN table doesn't support sub-columns, set up the stats here. */ this.statsMap.put( column.getInternalName(), new RowBufferStats( - column.getName(), column.getCollation(), column.getOrdinal(), null /* fieldId */)); + column.getName(), + column.getCollation(), + column.getOrdinal(), + null /* fieldId */, + parquetType.isPrimitive() ? parquetType.asPrimitiveType() : null)); if (onErrorOption == OpenChannelRequest.OnErrorOption.ABORT || onErrorOption == OpenChannelRequest.OnErrorOption.SKIP_BATCH) { @@ -116,7 +122,8 @@ public void setupSchema(List columns) { column.getName(), column.getCollation(), column.getOrdinal(), - null /* fieldId */)); + null /* fieldId */, + parquetType.isPrimitive() ? parquetType.asPrimitiveType() : null)); } } @@ -175,24 +182,27 @@ public void setupSchema(List columns) { if (clientBufferParameters.getIsIcebergMode()) { for (ColumnDescriptor columnDescriptor : schema.getColumns()) { String columnPath = concatDotPath(columnDescriptor.getPath()); + PrimitiveType primitiveType = columnDescriptor.getPrimitiveType(); /* set fieldId to 0 for non-structured columns */ - int fieldId = - columnDescriptor.getPath().length == 1 - ? 0 - : columnDescriptor.getPrimitiveType().getId().intValue(); + int fieldId = columnDescriptor.getPath().length == 1 ? 0 : primitiveType.getId().intValue(); int ordinal = schema.getType(columnDescriptor.getPath()[0]).getId().intValue(); this.statsMap.put( columnPath, - new RowBufferStats(columnPath, null /* collationDefinitionString */, ordinal, fieldId)); + new RowBufferStats( + columnPath, null /* collationDefinitionString */, ordinal, fieldId, primitiveType)); if (onErrorOption == OpenChannelRequest.OnErrorOption.ABORT || onErrorOption == OpenChannelRequest.OnErrorOption.SKIP_BATCH) { this.tempStatsMap.put( columnPath, new RowBufferStats( - columnPath, null /* collationDefinitionString */, ordinal, fieldId)); + columnPath, + null /* collationDefinitionString */, + ordinal, + fieldId, + primitiveType)); } } } diff --git a/src/main/java/net/snowflake/ingest/streaming/internal/RowBufferStats.java b/src/main/java/net/snowflake/ingest/streaming/internal/RowBufferStats.java index 2fae695f0..4d7781c78 100644 --- a/src/main/java/net/snowflake/ingest/streaming/internal/RowBufferStats.java +++ b/src/main/java/net/snowflake/ingest/streaming/internal/RowBufferStats.java @@ -11,6 +11,7 @@ import java.util.Objects; import net.snowflake.ingest.utils.ErrorCode; import net.snowflake.ingest.utils.SFException; +import org.apache.parquet.schema.PrimitiveType; /** Keeps track of the active EP stats, used to generate a file EP info */ class RowBufferStats { @@ -25,6 +26,12 @@ class RowBufferStats { */ private final Integer fieldId; + private final String collationDefinitionString; + /** Display name is required for the registration endpoint */ + private final String columnDisplayName; + /** Primitive type of the column, only used for Iceberg columns */ + private final PrimitiveType primitiveType; + private byte[] currentMinStrValue; private byte[] currentMaxStrValue; private BigInteger currentMinIntValue; @@ -34,22 +41,27 @@ class RowBufferStats { private long currentNullCount; // for binary or string columns private long currentMaxLength; - private final String collationDefinitionString; - /** Display name is required for the registration endpoint */ - private final String columnDisplayName; - /** Creates empty stats */ RowBufferStats( - String columnDisplayName, String collationDefinitionString, int ordinal, Integer fieldId) { + String columnDisplayName, + String collationDefinitionString, + int ordinal, + Integer fieldId, + PrimitiveType primitiveType) { this.columnDisplayName = columnDisplayName; this.collationDefinitionString = collationDefinitionString; this.ordinal = ordinal; this.fieldId = fieldId; + this.primitiveType = primitiveType; reset(); } RowBufferStats(String columnDisplayName) { - this(columnDisplayName, null, -1, null); + this(columnDisplayName, null, -1, null, null); + } + + RowBufferStats(String columnDisplayName, PrimitiveType primitiveType) { + this(columnDisplayName, null, -1, null, primitiveType); } void reset() { @@ -69,7 +81,8 @@ RowBufferStats forkEmpty() { this.getColumnDisplayName(), this.getCollationDefinitionString(), this.getOrdinal(), - this.getFieldId()); + this.getFieldId(), + this.getPrimitiveType()); } // TODO performance test this vs in place update @@ -87,7 +100,8 @@ static RowBufferStats getCombinedStats(RowBufferStats left, RowBufferStats right left.columnDisplayName, left.getCollationDefinitionString(), left.getOrdinal(), - left.getFieldId()); + left.getFieldId(), + left.getPrimitiveType()); if (left.currentMinIntValue != null) { combined.addIntValue(left.currentMinIntValue); @@ -238,6 +252,10 @@ Integer getFieldId() { return fieldId; } + PrimitiveType getPrimitiveType() { + return primitiveType; + } + /** * Compares two byte arrays lexicographically. If the two arrays share a common prefix then the * lexicographic comparison is the result of comparing two elements, as if by Byte.compare(byte, diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/BlobBuilderTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/BlobBuilderTest.java index 185fa5ded..1330152a4 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/BlobBuilderTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/BlobBuilderTest.java @@ -15,7 +15,10 @@ import net.snowflake.ingest.utils.Pair; import net.snowflake.ingest.utils.SFException; import org.apache.parquet.hadoop.BdecParquetWriter; +import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Types; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; @@ -98,7 +101,19 @@ private List> createChannelDataPerTable(int metada channelData .getColumnEps() - .putIfAbsent(columnName, new RowBufferStats(columnName, null, 1, isIceberg ? 0 : null)); + .putIfAbsent( + columnName, + isIceberg + ? new RowBufferStats( + columnName, + null, + 1, + 1, + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .id(1) + .named("test")) + : new RowBufferStats(columnName, null, 1, null, null)); channelData.setChannelContext( new ChannelFlushContext("channel1", "DB", "SCHEMA", "TABLE", 1L, "enc", 1L)); return Collections.singletonList(channelData); diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/FileColumnPropertiesTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/FileColumnPropertiesTest.java index 7b131b310..f4ffa11c4 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/FileColumnPropertiesTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/FileColumnPropertiesTest.java @@ -4,6 +4,9 @@ package net.snowflake.ingest.streaming.internal; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Types; import org.junit.Assert; import org.junit.Test; import org.junit.runners.Parameterized; @@ -19,10 +22,21 @@ public static Object[] isIceberg() { @Test public void testFileColumnPropertiesConstructor() { // Test simple construction - RowBufferStats stats = new RowBufferStats("COL", null, 1, isIceberg ? 1 : null); + RowBufferStats stats = + isIceberg + ? new RowBufferStats( + "COL", + null, + 1, + 1, + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .id(1) + .named("test")) + : new RowBufferStats("COL", null, 1, null, null); stats.addStrValue("bcd"); stats.addStrValue("abcde"); - FileColumnProperties props = new FileColumnProperties(stats, isIceberg); + FileColumnProperties props = new FileColumnProperties(stats, !isIceberg); Assert.assertEquals(1, props.getColumnOrdinal()); Assert.assertEquals(isIceberg ? 1 : null, props.getFieldId()); Assert.assertEquals("6162636465", props.getMinStrValue()); @@ -31,10 +45,21 @@ public void testFileColumnPropertiesConstructor() { Assert.assertNull(props.getMaxStrNonCollated()); // Test that truncation is performed - stats = new RowBufferStats("COL", null, 1, isIceberg ? 0 : null); + stats = + isIceberg + ? new RowBufferStats( + "COL", + null, + 1, + 1, + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .id(1) + .named("test")) + : new RowBufferStats("COL", null, 1, null, null); stats.addStrValue("aßßßßßßßßßßßßßßßß"); Assert.assertEquals(33, stats.getCurrentMinStrValue().length); - props = new FileColumnProperties(stats, isIceberg); + props = new FileColumnProperties(stats, !isIceberg); Assert.assertEquals(1, props.getColumnOrdinal()); Assert.assertNull(props.getMinStrNonCollated()); Assert.assertNull(props.getMaxStrNonCollated()); 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 cd7354f09..ca0fd5295 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/FlushServiceTest.java @@ -52,6 +52,8 @@ import net.snowflake.ingest.utils.ErrorCode; import net.snowflake.ingest.utils.ParameterProvider; import net.snowflake.ingest.utils.SFException; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Types; import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; @@ -877,8 +879,12 @@ public void testBuildAndUpload() throws Exception { Map eps1 = new HashMap<>(); Map eps2 = new HashMap<>(); - RowBufferStats stats1 = new RowBufferStats("COL1"); - RowBufferStats stats2 = new RowBufferStats("COL1"); + RowBufferStats stats1 = + new RowBufferStats( + "COL1", Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("COL1")); + RowBufferStats stats2 = + new RowBufferStats( + "COL1", Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("COL1")); eps1.put("one", stats1); eps2.put("one", stats2); @@ -1115,7 +1121,9 @@ public void testBlobBuilder() throws Exception { Map eps1 = new HashMap<>(); - RowBufferStats stats1 = new RowBufferStats("COL1"); + RowBufferStats stats1 = + new RowBufferStats( + "COL1", Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("COL1")); eps1.put("one", stats1); 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 5e5b96fc3..5d7873b95 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/RowBufferTest.java @@ -29,6 +29,8 @@ import org.apache.commons.codec.binary.Hex; import org.apache.commons.lang3.StringUtils; import org.apache.parquet.hadoop.BdecParquetReader; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Types; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -563,12 +565,18 @@ private void testDoubleQuotesColumnNameHelper(OpenChannelRequest.OnErrorOption o public void testBuildEpInfoFromStats() { Map colStats = new HashMap<>(); - RowBufferStats stats1 = new RowBufferStats("intColumn"); + RowBufferStats stats1 = + new RowBufferStats( + "intColumn", + Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("intColumn")); stats1.addIntValue(BigInteger.valueOf(2)); stats1.addIntValue(BigInteger.valueOf(10)); stats1.addIntValue(BigInteger.valueOf(1)); - RowBufferStats stats2 = new RowBufferStats("strColumn"); + RowBufferStats stats2 = + new RowBufferStats( + "strColumn", + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY).id(2).named("strColumn")); stats2.addStrValue("alice"); stats2.addStrValue("bob"); stats2.incCurrentNullCount(); @@ -603,8 +611,14 @@ public void testBuildEpInfoFromNullColumnStats() { final String realColName = "realCol"; Map colStats = new HashMap<>(); - RowBufferStats stats1 = new RowBufferStats(intColName); - RowBufferStats stats2 = new RowBufferStats(realColName); + RowBufferStats stats1 = + new RowBufferStats( + intColName, + Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named(intColName)); + RowBufferStats stats2 = + new RowBufferStats( + realColName, + Types.optional(PrimitiveType.PrimitiveTypeName.DOUBLE).id(2).named(realColName)); stats1.incCurrentNullCount(); stats2.incCurrentNullCount(); @@ -618,22 +632,18 @@ public void testBuildEpInfoFromNullColumnStats() { FileColumnProperties intColumnResult = columnResults.get(intColName); Assert.assertEquals(-1, intColumnResult.getDistinctValues()); Assert.assertEquals( - isIcebergMode ? null : FileColumnProperties.DEFAULT_MIN_MAX_INT_VAL_FOR_EP, - intColumnResult.getMinIntValue()); + FileColumnProperties.DEFAULT_MIN_MAX_INT_VAL_FOR_EP, intColumnResult.getMinIntValue()); Assert.assertEquals( - isIcebergMode ? null : FileColumnProperties.DEFAULT_MIN_MAX_INT_VAL_FOR_EP, - intColumnResult.getMaxIntValue()); + FileColumnProperties.DEFAULT_MIN_MAX_INT_VAL_FOR_EP, intColumnResult.getMaxIntValue()); Assert.assertEquals(1, intColumnResult.getNullCount()); Assert.assertEquals(0, intColumnResult.getMaxLength()); FileColumnProperties realColumnResult = columnResults.get(realColName); Assert.assertEquals(-1, intColumnResult.getDistinctValues()); Assert.assertEquals( - isIcebergMode ? null : FileColumnProperties.DEFAULT_MIN_MAX_REAL_VAL_FOR_EP, - realColumnResult.getMinRealValue()); + FileColumnProperties.DEFAULT_MIN_MAX_REAL_VAL_FOR_EP, realColumnResult.getMinRealValue()); Assert.assertEquals( - isIcebergMode ? null : FileColumnProperties.DEFAULT_MIN_MAX_REAL_VAL_FOR_EP, - realColumnResult.getMaxRealValue()); + FileColumnProperties.DEFAULT_MIN_MAX_REAL_VAL_FOR_EP, realColumnResult.getMaxRealValue()); Assert.assertEquals(1, realColumnResult.getNullCount()); Assert.assertEquals(0, realColumnResult.getMaxLength()); } @@ -642,12 +652,18 @@ public void testBuildEpInfoFromNullColumnStats() { public void testInvalidEPInfo() { Map colStats = new HashMap<>(); - RowBufferStats stats1 = new RowBufferStats("intColumn"); + RowBufferStats stats1 = + new RowBufferStats( + "intColumn", + Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("intColumn")); stats1.addIntValue(BigInteger.valueOf(2)); stats1.addIntValue(BigInteger.valueOf(10)); stats1.addIntValue(BigInteger.valueOf(1)); - RowBufferStats stats2 = new RowBufferStats("strColumn"); + RowBufferStats stats2 = + new RowBufferStats( + "strColumn", + Types.optional(PrimitiveType.PrimitiveTypeName.BINARY).id(2).named("strColumn")); stats2.addStrValue("alice"); stats2.incCurrentNullCount(); stats2.incCurrentNullCount(); diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientTest.java index 0dbeeebee..14de4342b 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/SnowflakeStreamingIngestClientTest.java @@ -61,6 +61,8 @@ import net.snowflake.ingest.utils.SnowflakeURL; import net.snowflake.ingest.utils.Utils; import org.apache.commons.lang3.tuple.Pair; +import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.Types; import org.bouncycastle.asn1.nist.NISTObjectIdentifiers; import org.bouncycastle.openssl.jcajce.JcaPEMWriter; import org.bouncycastle.operator.OperatorCreationException; @@ -500,8 +502,11 @@ public void testRegisterBlobRequestCreationSuccess() throws Exception { .build(); Map columnEps = new HashMap<>(); - columnEps.put("column", new RowBufferStats("COL1")); - EpInfo epInfo = AbstractRowBuffer.buildEpInfoFromStats(1, columnEps, isIcebergMode); + columnEps.put( + "column", + new RowBufferStats( + "COL1", Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("COL1"))); + EpInfo epInfo = AbstractRowBuffer.buildEpInfoFromStats(1, columnEps, !isIcebergMode); ChunkMetadata chunkMetadata = ChunkMetadata.builder() @@ -549,8 +554,11 @@ public void testRegisterBlobRequestCreationSuccess() throws Exception { private Pair, Set> getRetryBlobMetadata() { Map columnEps = new HashMap<>(); - columnEps.put("column", new RowBufferStats("COL1")); - EpInfo epInfo = AbstractRowBuffer.buildEpInfoFromStats(1, columnEps, isIcebergMode); + columnEps.put( + "column", + new RowBufferStats( + "COL1", Types.optional(PrimitiveType.PrimitiveTypeName.INT32).id(1).named("COL1"))); + EpInfo epInfo = AbstractRowBuffer.buildEpInfoFromStats(1, columnEps, !isIcebergMode); ChannelMetadata channelMetadata1 = ChannelMetadata.builder() diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/AbstractDataTypeTest.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/AbstractDataTypeTest.java index c39ffe967..3c58fb000 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/AbstractDataTypeTest.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/AbstractDataTypeTest.java @@ -21,12 +21,15 @@ import net.snowflake.ingest.streaming.OpenChannelRequest; import net.snowflake.ingest.streaming.SnowflakeStreamingIngestChannel; import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClient; -import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClientFactory; +import net.snowflake.ingest.streaming.internal.SnowflakeStreamingIngestClientInternal; import net.snowflake.ingest.utils.Constants; +import net.snowflake.ingest.utils.ParameterProvider; import net.snowflake.ingest.utils.SFException; +import net.snowflake.ingest.utils.SnowflakeURL; +import net.snowflake.ingest.utils.Utils; +import org.assertj.core.api.Assertions; import org.junit.After; import org.junit.Assert; -import org.junit.Before; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import org.junit.runners.Parameterized.Parameter; @@ -60,7 +63,7 @@ public abstract class AbstractDataTypeTest { private String schemaName = "PUBLIC"; private SnowflakeStreamingIngestClient client; - private static final ObjectMapper objectMapper = new ObjectMapper(); + protected static final ObjectMapper objectMapper = new ObjectMapper(); @Parameters(name = "{index}: {0}") public static Object[] compressionAlgorithms() { @@ -69,8 +72,7 @@ public static Object[] compressionAlgorithms() { @Parameter public String compressionAlgorithm; - @Before - public void before() throws Exception { + public void before(boolean isIceberg) throws Exception { databaseName = String.format("SDK_DATATYPE_COMPATIBILITY_IT_%s", getRandomIdentifier()); conn = TestUtils.getConnection(true); conn.createStatement().execute(String.format("create or replace database %s;", databaseName)); @@ -84,7 +86,16 @@ public void before() throws Exception { props.setProperty(ROLE, "ACCOUNTADMIN"); } props.setProperty(BDEC_PARQUET_COMPRESSION_ALGORITHM, compressionAlgorithm); - client = SnowflakeStreamingIngestClientFactory.builder("client1").setProperties(props).build(); + + // Override Iceberg mode client lag to 1 second for faster test execution + Map parameterMap = new HashMap<>(); + parameterMap.put(ParameterProvider.MAX_CLIENT_LAG, 1000L); + + Properties prop = Utils.createProperties(props); + SnowflakeURL accountURL = new SnowflakeURL(prop.getProperty(Constants.ACCOUNT_URL)); + client = + new SnowflakeStreamingIngestClientInternal<>( + "client1", accountURL, prop, parameterMap, isIceberg, false); } @After @@ -112,6 +123,23 @@ protected String createTable(String dataType) throws SQLException { String.format( "create or replace table %s (%s string, %s %s)", tableName, SOURCE_COLUMN_NAME, VALUE_COLUMN_NAME, dataType)); + + return tableName; + } + + protected String createIcebergTable(String dataType) throws SQLException { + String tableName = getRandomIdentifier(); + String baseLocation = + String.format("%s/%s/%s", databaseName, dataType.replace(" ", "_"), tableName); + conn.createStatement() + .execute( + String.format( + "create or replace iceberg table %s (%s string, %s %s) " + + "catalog = 'SNOWFLAKE' " + + "external_volume = 'streaming_ingest' " + + "base_location = '%s';", + tableName, SOURCE_COLUMN_NAME, VALUE_COLUMN_NAME, dataType, baseLocation)); + return tableName; } @@ -196,7 +224,14 @@ protected void expectNotSupported(String dataType, T value) throws Exception */ void testIngestion(String dataType, VALUE expectedValue, Provider selectProvider) throws Exception { - ingestAndAssert(dataType, expectedValue, null, expectedValue, null, selectProvider); + ingestAndAssert( + dataType, expectedValue, null, expectedValue, null, selectProvider, false /* isIceberg */); + } + + void testIcebergIngestion( + String dataType, VALUE expectedValue, Provider selectProvider) throws Exception { + ingestAndAssert( + dataType, expectedValue, null, expectedValue, null, selectProvider, true /* isIceberg */); } /** @@ -209,7 +244,30 @@ void testIngestion( JDBC_READ expectedValue, Provider selectProvider) throws Exception { - ingestAndAssert(dataType, streamingIngestWriteValue, null, expectedValue, null, selectProvider); + ingestAndAssert( + dataType, + streamingIngestWriteValue, + null, + expectedValue, + null, + selectProvider, + false /* isIceberg */); + } + + void testIcebergIngestion( + String dataType, + STREAMING_INGEST_WRITE streamingIngestWriteValue, + JDBC_READ expectedValue, + Provider selectProvider) + throws Exception { + ingestAndAssert( + dataType, + streamingIngestWriteValue, + null, + expectedValue, + null, + selectProvider, + true /* isIceberg */); } /** @@ -218,7 +276,7 @@ void testIngestion( */ void testJdbcTypeCompatibility(String typeName, T value, Provider provider) throws Exception { - ingestAndAssert(typeName, value, value, value, provider, provider); + ingestAndAssert(typeName, value, value, value, provider, provider, false /* isIceberg */); } /** Simplified version where write value for streaming ingest and JDBC are the same */ @@ -230,22 +288,29 @@ void testJdbcTypeCompatibility( Provider selectProvider) throws Exception { ingestAndAssert( - typeName, writeValue, writeValue, expectedValue, insertProvider, selectProvider); + typeName, + writeValue, + writeValue, + expectedValue, + insertProvider, + selectProvider, + false /* isIceberg */); } /** * Ingests values with streaming ingest and JDBC driver, SELECTs them back with WHERE condition * and asserts they exist. * + * @param Type ingested by streaming ingest + * @param Type written by JDBC driver + * @param Type read by JDBC driver * @param dataType Snowflake data type * @param streamingIngestWriteValue Value ingested by streaming ingest * @param jdbcWriteValue Value written by JDBC driver * @param expectedValue Expected value received from JDBC driver SELECT * @param insertProvider JDBC parameter provider for INSERT * @param selectProvider JDBC parameter provider for SELECT ... WHERE - * @param Type ingested by streaming ingest - * @param Type written by JDBC driver - * @param Type read by JDBC driver + * @param isIceberg whether the table is an iceberg table */ void ingestAndAssert( String dataType, @@ -253,13 +318,14 @@ void ingestAndAssert( JDBC_WRITE jdbcWriteValue, JDBC_READ expectedValue, Provider insertProvider, - Provider selectProvider) + Provider selectProvider, + boolean isIceberg) throws Exception { if (jdbcWriteValue == null ^ insertProvider == null) throw new IllegalArgumentException( "jdbcWriteValue and provider must be both null or not null"); boolean insertAlsoWithJdbc = jdbcWriteValue != null; - String tableName = createTable(dataType); + String tableName = isIceberg ? createIcebergTable(dataType) : createTable(dataType); String offsetToken = UUID.randomUUID().toString(); // Insert using JDBC @@ -287,13 +353,13 @@ void ingestAndAssert( if (expectedValue == null) { selectQuery = String.format("select count(*) from %s where %s is NULL", tableName, VALUE_COLUMN_NAME); - } else if (dataType.startsWith("TIMESTAMP_")) { + } else if (dataType.toUpperCase().startsWith("TIMESTAMP")) { selectQuery = String.format( "select count(*) from %s where to_varchar(%s, 'YYYY-MM-DD HH24:MI:SS.FF TZHTZM') =" + " ?;", tableName, VALUE_COLUMN_NAME); - } else if (dataType.startsWith("TIME")) { + } else if (dataType.toUpperCase().startsWith("TIME")) { selectQuery = String.format( "select count(*) from %s where to_varchar(%s, 'HH24:MI:SS.FF TZHTZM') = ?;", @@ -310,7 +376,9 @@ void ingestAndAssert( Assert.assertTrue(resultSet.next()); int count = resultSet.getInt(1); Assert.assertEquals(insertAlsoWithJdbc ? 2 : 1, count); - migrateTable(tableName); // migration should always succeed + if (!isIceberg) { + migrateTable(tableName); // migration should always succeed + } } void assertVariant( @@ -427,4 +495,38 @@ protected final void ingestManyAndMigrate( TestUtils.waitForOffset(channel, offsetToken); migrateTable(tableName); // migration should always succeed } + + protected void testIcebergIngestAndQuery( + String dataType, + Iterable values, + String queryTemplate, + Iterable expectedValues) + throws Exception { + String tableName = createIcebergTable(dataType); + SnowflakeStreamingIngestChannel channel = openChannel(tableName); + String offsetToken = null; + for (Object value : values) { + offsetToken = UUID.randomUUID().toString(); + channel.insertRow(createStreamingIngestRow(value), offsetToken); + } + TestUtils.waitForOffset(channel, offsetToken); + + String verificationQuery = + queryTemplate.replace("{tableName}", tableName).replace("{columnName}", VALUE_COLUMN_NAME); + ResultSet resultSet = conn.createStatement().executeQuery(verificationQuery); + + for (Object expectedValue : expectedValues) { + Assertions.assertThat(resultSet.next()).isTrue(); + Object res = resultSet.getObject(1); + if (expectedValue instanceof BigDecimal) { + Assertions.assertThat(res) + .usingComparatorForType(BigDecimal::compareTo, BigDecimal.class) + .usingRecursiveComparison() + .isEqualTo(expectedValue); + } else { + Assertions.assertThat(res).isEqualTo(expectedValue); + } + } + Assertions.assertThat(resultSet.next()).isFalse(); + } } diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/BinaryIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/BinaryIT.java index 2a5023a3a..a379f27ad 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/BinaryIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/BinaryIT.java @@ -1,9 +1,14 @@ package net.snowflake.ingest.streaming.internal.datatypes; import org.bouncycastle.util.encoders.Hex; +import org.junit.Before; import org.junit.Test; public class BinaryIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(false); + } @Test public void testBinary() throws Exception { diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/DateTimeIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/DateTimeIT.java index 62829656c..7695a5a12 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/DateTimeIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/DateTimeIT.java @@ -21,6 +21,7 @@ public class DateTimeIT extends AbstractDataTypeTest { @Before public void setup() throws Exception { + super.before(false); // Set to a random time zone not to interfere with any of the tests conn.createStatement().execute("alter session set timezone = 'America/New_York';"); } diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergDateTimeIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergDateTimeIT.java new file mode 100644 index 000000000..3fe53aed5 --- /dev/null +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergDateTimeIT.java @@ -0,0 +1,383 @@ +package net.snowflake.ingest.streaming.internal.datatypes; + +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.time.OffsetTime; +import java.time.ZoneOffset; +import java.util.Arrays; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("This test can be enabled after server side Iceberg EP support is released") +public class IcebergDateTimeIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(true); + } + + @Test + public void testDate() throws Exception { + testIcebergIngestion("date", "9999-12-31", new StringProvider()); + testIcebergIngestion("date", "1582-10-05", new StringProvider()); + testIcebergIngestion("date", LocalDate.parse("1998-09-08"), "1998-09-08", new StringProvider()); + testIcebergIngestion( + "date", LocalDateTime.parse("1998-09-08T02:00:00.123"), "1998-09-08", new StringProvider()); + testIcebergIngestion("date", null, new StringProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("date", "2000-01-32", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("date", new Object(), "2000-01-01", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("date not null", null, new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testDateAndQueries() throws Exception { + testIcebergIngestAndQuery( + "date", + Arrays.asList( + "1998-09-08", + LocalDate.parse("1998-09-11"), + LocalDateTime.parse("1998-09-05T02:00:00.123")), + "select {columnName} from {tableName}", + Arrays.asList( + Date.valueOf("1998-09-08"), Date.valueOf("1998-09-11"), Date.valueOf("1998-09-05"))); + testIcebergIngestAndQuery( + "date", + Arrays.asList(null, null, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "date", + Arrays.asList("9999-12-31", "2024-10-08", "0000-01-01"), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Date.valueOf("9999-12-31"))); + testIcebergIngestAndQuery( + "date", + Arrays.asList(null, "2001-01-01", "2000-01-01", null, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "date", + Arrays.asList("2001-01-01", "2001-01-01", "2001-01-01", "2001-01-01", null), + "select COUNT(*) from {tableName} where {columnName} = '2001-01-01'", + Arrays.asList(4L)); + } + + @Test + public void testTime() throws Exception { + testIcebergIngestion("time", "00:00:00", "00:00:00.000000 Z", new StringProvider()); + testIcebergIngestion("time", "23:59:59", "23:59:59.000000 Z", new StringProvider()); + testIcebergIngestion("time", "12:00:00", "12:00:00.000000 Z", new StringProvider()); + testIcebergIngestion("time", "12:00:00.123", "12:00:00.123000 Z", new StringProvider()); + testIcebergIngestion("time", "12:00:00.123456", "12:00:00.123456 Z", new StringProvider()); + testIcebergIngestion("time", "12:00:00.123456789", "12:00:00.123456 Z", new StringProvider()); + testIcebergIngestion( + "time", LocalTime.of(23, 59, 59), "23:59:59.000000 Z", new StringProvider()); + testIcebergIngestion( + "time", + OffsetTime.of(12, 0, 0, 123000, ZoneOffset.ofHoursMinutes(0, 0)), + "12:00:00.000123 Z", + new StringProvider()); + testIcebergIngestion("time", null, new StringProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("time", "12:00:00.123456789012", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "time", + LocalDateTime.parse("1998-09-08T02:00:00.123"), + "02:00:00", + new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("time not null", null, new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testTimeAndQueries() throws Exception { + testIcebergIngestAndQuery( + "time", + Arrays.asList( + "00:00:00", + LocalTime.of(23, 59, 59), + OffsetTime.of(12, 0, 0, 123000, ZoneOffset.ofHoursMinutes(0, 0))), + "select {columnName} from {tableName}", + Arrays.asList( + Time.valueOf("00:00:00"), Time.valueOf("23:59:59"), Time.valueOf("12:00:00"))); + testIcebergIngestAndQuery( + "time", + Arrays.asList(null, null, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "time", + Arrays.asList("23:59:59", "12:00:00", "00:00:00"), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Time.valueOf("23:59:59"))); + testIcebergIngestAndQuery( + "time", + Arrays.asList(null, "12:00:00", "12:00:00", null, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "time", + Arrays.asList("12:00:00", "12:00:00", "12:00:00", "12:00:00", null), + "select COUNT(*) from {tableName} where {columnName} = '12:00:00'", + Arrays.asList(4L)); + } + + @Test + public void testTimestamp() throws Exception { + testIcebergIngestion( + "timestamp_ntz(6)", + "2000-12-31T23:59:59", + "2000-12-31 23:59:59.000000 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ntz(6)", + "2000-12-31T23:59:59.123456", + "2000-12-31 23:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ntz(6)", + "2000-12-31T23:59:59.123456789+08:00", + "2000-12-31 23:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ntz(6)", + LocalDate.parse("2000-12-31"), + "2000-12-31 00:00:00.000000 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ntz(6)", + LocalDateTime.parse("2000-12-31T23:59:59.123456789"), + "2000-12-31 23:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ntz(6)", + OffsetDateTime.parse("2000-12-31T23:59:59.123456789Z"), + "2000-12-31 23:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion("timestamp_ntz(6)", null, new StringProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "timestamp", "2000-12-31T23:59:59.123456789012", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "timestamp", + new Object(), + "2000-12-31 00:00:00.000000 Z", + new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("timestamp_ntz(6) not null", null, new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testTimestampAndQueries() throws Exception { + testIcebergIngestAndQuery( + "timestamp_ntz(6)", + Arrays.asList( + "2000-12-31T23:59:59", + LocalDateTime.parse("2000-12-31T23:59:59.123456789"), + OffsetDateTime.parse("2000-12-31T23:59:59.123456789Z")), + "select {columnName} from {tableName}", + Arrays.asList( + Timestamp.valueOf("2000-12-31 23:59:59"), + Timestamp.valueOf("2000-12-31 23:59:59.123456"), + Timestamp.valueOf("2000-12-31 23:59:59.123456"))); + testIcebergIngestAndQuery( + "timestamp_ntz(6)", + Arrays.asList(null, null, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "timestamp_ntz(6)", + Arrays.asList( + "2000-12-31T23:59:59", + "2000-12-31T23:59:59.123456789", + "2000-12-31T23:59:59.123456789"), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Timestamp.valueOf("2000-12-31 23:59:59.123456"))); + testIcebergIngestAndQuery( + "timestamp_ntz(6)", + Arrays.asList(null, "2000-12-31T23:59:59", "2000-12-31T23:59:59", null, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "timestamp_ntz(6)", + Arrays.asList( + "2000-12-31T23:59:59", + "2000-12-31T23:59:59", + "2000-12-31T23:59:59", + "2000-12-31T23:59:59", + null), + "select COUNT(*) from {tableName} where {columnName} = '2000-12-31T23:59:59'", + Arrays.asList(4L)); + } + + @Test + public void testTimestampTZ() throws Exception { + conn.createStatement().execute("alter session set timezone = 'UTC';"); + testIcebergIngestion( + "timestamp_ltz(6)", + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31 15:59:59.000000 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ltz(6)", + "2000-12-31T23:59:59.123456789+00:00", + "2000-12-31 23:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion( + "timestamp_ltz(6)", + "2000-12-31T23:59:59.123456-08:00", + "2001-01-01 07:59:59.123456 Z", + new StringProvider()); + testIcebergIngestion("timestamp_ltz(6)", null, new StringProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "timestamp", "2000-12-31T23:59:59.123456789012+08:00", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "timestamp", + new Object(), + "2000-12-31 00:00:00.000000 Z", + new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("timestamp_ltz(6) not null", null, new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testTimestampTZAndQueries() throws Exception { + conn.createStatement().execute("alter session set timezone = 'UTC';"); + testIcebergIngestAndQuery( + "timestamp_ltz(6)", + Arrays.asList( + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.123456789+00:00", + "2000-12-31T23:59:59.123456-08:00"), + "select {columnName} from {tableName}", + Arrays.asList( + Timestamp.valueOf("2000-12-31 15:59:59"), + Timestamp.valueOf("2000-12-31 23:59:59.123456"), + Timestamp.valueOf("2001-01-01 07:59:59.123456"))); + testIcebergIngestAndQuery( + "timestamp_ltz(6)", + Arrays.asList(null, null, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "timestamp_ltz(6)", + Arrays.asList( + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.123456789+00:00", + "2000-12-31T23:59:59.123456-08:00"), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Timestamp.valueOf("2001-01-01 07:59:59.123456"))); + testIcebergIngestAndQuery( + "timestamp_ltz(6)", + Arrays.asList( + null, + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.000000+08:00", + null, + null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "timestamp_ltz(6)", + Arrays.asList( + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.000000+08:00", + "2000-12-31T23:59:59.000000+08:00", + null), + "select COUNT(*) from {tableName} where {columnName} = '2000-12-31T23:59:59.000000+08:00'", + Arrays.asList(4L)); + } +} diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergLogicalTypesIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergLogicalTypesIT.java new file mode 100644 index 000000000..33e062d72 --- /dev/null +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergLogicalTypesIT.java @@ -0,0 +1,250 @@ +package net.snowflake.ingest.streaming.internal.datatypes; + +import java.util.Arrays; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("This test can be enabled after server side Iceberg EP support is released") +public class IcebergLogicalTypesIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(true); + } + + @Test + public void testBoolean() throws Exception { + testIcebergIngestion("boolean", true, new BooleanProvider()); + testIcebergIngestion("boolean", false, new BooleanProvider()); + testIcebergIngestion("boolean", 1, true, new BooleanProvider()); + testIcebergIngestion("boolean", 0, false, new BooleanProvider()); + testIcebergIngestion("boolean", "1", true, new BooleanProvider()); + testIcebergIngestion("boolean", "0", false, new BooleanProvider()); + testIcebergIngestion("boolean", "true", true, new BooleanProvider()); + testIcebergIngestion("boolean", "false", false, new BooleanProvider()); + testIcebergIngestion("boolean", null, new BooleanProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("boolean", new Object(), true, new BooleanProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("boolean not null", null, new BooleanProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testBooleanQueries() throws Exception { + testIcebergIngestAndQuery( + "boolean", + Arrays.asList(true, false, false, true, false), + "select {columnName} from {tableName}", + Arrays.asList(true, false, false, true, false)); + testIcebergIngestAndQuery( + "boolean", + Arrays.asList(1, "0", 1, false, 1), + "select {columnName} from {tableName}", + Arrays.asList(true, false, true, false, true)); + testIcebergIngestAndQuery( + "boolean", + Arrays.asList("true", 1, null, "1", true), + "select MIN({columnName}) from {tableName}", + Arrays.asList(true)); + testIcebergIngestAndQuery( + "boolean", + Arrays.asList(null, null, null, "true", false), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "boolean", + Arrays.asList(null, null, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((Object) null)); + } + + @Test + public void testBinary() throws Exception { + testIcebergIngestion("binary", new byte[] {1, 2, 3}, new ByteArrayProvider()); + testIcebergIngestion("binary", "313233", new byte[] {49, 50, 51}, new ByteArrayProvider()); + testIcebergIngestion("binary", new byte[8388608], new ByteArrayProvider()); + testIcebergIngestion("binary", null, new ByteArrayProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("binary", new byte[8388608 + 1], new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "binary", new Object(), new byte[] {1, 2, 3, 4, 5}, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("binary not null", null, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testBinaryAndQueries() throws Exception { + testIcebergIngestAndQuery( + "binary", + Arrays.asList(new byte[] {1, 2, 3}, new byte[] {4, 5, 6}, new byte[] {7, 8, 9}), + "select {columnName} from {tableName}", + Arrays.asList(new byte[] {1, 2, 3}, new byte[] {4, 5, 6}, new byte[] {7, 8, 9})); + testIcebergIngestAndQuery( + "binary", + Arrays.asList("313233", new byte[] {4, 5, 6}, "373839"), + "select {columnName} from {tableName}", + Arrays.asList(new byte[] {49, 50, 51}, new byte[] {4, 5, 6}, new byte[] {55, 56, 57})); + testIcebergIngestAndQuery( + "binary", + Arrays.asList(new byte[8388608], new byte[8388608], new byte[8388608]), + "select {columnName} from {tableName}", + Arrays.asList(new byte[8388608], new byte[8388608], new byte[8388608])); + testIcebergIngestAndQuery( + "binary", + Arrays.asList(null, null, null), + "select {columnName} from {tableName}", + Arrays.asList(null, null, null)); + testIcebergIngestAndQuery( + "binary", + Arrays.asList(null, new byte[] {1, 2, 3}, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(2L)); + byte[] max = new byte[8388608]; + Arrays.fill(max, (byte) 0xFF); + testIcebergIngestAndQuery( + "binary", + Arrays.asList(new byte[8388608], max), + "select MAX({columnName}) from {tableName}", + Arrays.asList(max)); + testIcebergIngestAndQuery( + "binary", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + } + + @Test + public void testFixedLenByteArray() throws Exception { + testIcebergIngestion("fixed(3)", new byte[] {1, 2, 3}, new ByteArrayProvider()); + testIcebergIngestion("fixed(3)", "313233", new byte[] {49, 50, 51}, new ByteArrayProvider()); + testIcebergIngestion("fixed(8388608)", new byte[8388608], new ByteArrayProvider()); + testIcebergIngestion("fixed(3)", null, new ByteArrayProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "fixed(10)", + "313233", + new byte[] {49, 50, 51, 0, 0, 0, 0, 0, 0, 0}, + new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "fixed(3)", new byte[] {49, 50, 51, 52}, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "fixed(3)", "313", new byte[] {49, 50}, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "fixed(3)", new Object(), new byte[] {1, 2, 3, 4, 5}, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("fixed(3) not null", null, new ByteArrayProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testFixedLenByteArrayAndQueries() throws Exception { + testIcebergIngestAndQuery( + "fixed(3)", + Arrays.asList(new byte[] {1, 2, 3}, new byte[] {4, 5, 6}, new byte[] {7, 8, 9}), + "select {columnName} from {tableName}", + Arrays.asList(new byte[] {1, 2, 3}, new byte[] {4, 5, 6}, new byte[] {7, 8, 9})); + testIcebergIngestAndQuery( + "fixed(3)", + Arrays.asList("313233", new byte[] {4, 5, 6}, "373839"), + "select {columnName} from {tableName}", + Arrays.asList(new byte[] {49, 50, 51}, new byte[] {4, 5, 6}, new byte[] {55, 56, 57})); + testIcebergIngestAndQuery( + "fixed(8388608)", + Arrays.asList(new byte[8388608], new byte[8388608], new byte[8388608]), + "select {columnName} from {tableName}", + Arrays.asList(new byte[8388608], new byte[8388608], new byte[8388608])); + testIcebergIngestAndQuery( + "fixed(3)", + Arrays.asList(null, null, null), + "select {columnName} from {tableName}", + Arrays.asList(null, null, null)); + testIcebergIngestAndQuery( + "fixed(3)", + Arrays.asList(null, new byte[] {1, 2, 3}, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(2L)); + byte[] max = new byte[8388608]; + Arrays.fill(max, (byte) 0xFF); + testIcebergIngestAndQuery( + "fixed(8388608)", + Arrays.asList(new byte[8388608], max), + "select MAX({columnName}) from {tableName}", + Arrays.asList(max)); + testIcebergIngestAndQuery( + "fixed(3)", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + } +} 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 new file mode 100644 index 000000000..b8df92228 --- /dev/null +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergNumericTypesIT.java @@ -0,0 +1,357 @@ +package net.snowflake.ingest.streaming.internal.datatypes; + +import java.math.BigDecimal; +import java.util.Arrays; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("This test can be enabled after server side Iceberg EP support is released") +public class IcebergNumericTypesIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(true); + } + + @Test + public void testInt() throws Exception { + testIcebergIngestion("int", 1, new IntProvider()); + testIcebergIngestion("int", -.0f, 0, new IntProvider()); + testIcebergIngestion("int", 0.5f, 1, new IntProvider()); + testIcebergIngestion("int", "100.4", 100, new IntProvider()); + testIcebergIngestion("int", new BigDecimal("1000000.09"), 1000000, new IntProvider()); + testIcebergIngestion("int", Integer.MAX_VALUE, new IntProvider()); + testIcebergIngestion("int", Integer.MIN_VALUE, new IntProvider()); + testIcebergIngestion("int", null, new IntProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("int", 1L + Integer.MAX_VALUE, new LongProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, () -> testIcebergIngestion("int", true, 0, new IntProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, () -> testIcebergIngestion("int not null", null, new IntProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testIntAndQueries() throws Exception { + testIcebergIngestAndQuery( + "int", + Arrays.asList(1, -0, Integer.MAX_VALUE, Integer.MIN_VALUE, null), + "select {columnName} from {tableName}", + Arrays.asList(1L, 0L, (long) Integer.MAX_VALUE, (long) Integer.MIN_VALUE, null)); + testIcebergIngestAndQuery( + "int", + Arrays.asList(null, null, null, Integer.MIN_VALUE), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "int", + Arrays.asList(1, -0, Integer.MAX_VALUE, Integer.MIN_VALUE), + "select MAX({columnName}) from {tableName}", + Arrays.asList((long) Integer.MAX_VALUE)); + testIcebergIngestAndQuery( + "int", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "int", + Arrays.asList( + Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE, Integer.MAX_VALUE, null), + "select COUNT({columnName}) from {tableName} where {columnName} = 2147483647", + Arrays.asList(4L)); + } + + @Test + public void testLong() throws Exception { + testIcebergIngestion("long", 1L, new LongProvider()); + testIcebergIngestion("long", -.0f, 0L, new LongProvider()); + testIcebergIngestion("long", 0.5f, 1L, new LongProvider()); + testIcebergIngestion("long", "100.4", 100L, new LongProvider()); + testIcebergIngestion("long", new BigDecimal("1000000.09"), 1000000L, new LongProvider()); + testIcebergIngestion("long", Long.MAX_VALUE, new LongProvider()); + testIcebergIngestion("long", Long.MIN_VALUE, new LongProvider()); + testIcebergIngestion("long", null, new LongProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("long", Double.MAX_VALUE, new DoubleProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("long", Double.NaN, new DoubleProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, () -> testIcebergIngestion("long", false, 0L, new LongProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("long not null", null, new LongProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testLongAndQueries() throws Exception { + testIcebergIngestAndQuery( + "long", + Arrays.asList(1L, -0L, Long.MAX_VALUE, Long.MIN_VALUE, null), + "select {columnName} from {tableName}", + Arrays.asList(1L, 0L, Long.MAX_VALUE, Long.MIN_VALUE, null)); + testIcebergIngestAndQuery( + "long", + Arrays.asList(null, null, null, Long.MIN_VALUE), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "long", + Arrays.asList(1L, -0L, Long.MAX_VALUE, Long.MIN_VALUE), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Long.MAX_VALUE)); + testIcebergIngestAndQuery( + "long", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "long", + Arrays.asList(Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE, Long.MAX_VALUE, null), + "select COUNT({columnName}) from {tableName} where {columnName} = 9223372036854775807", + Arrays.asList(4L)); + } + + @Test + public void testFloat() throws Exception { + testIcebergIngestion("float", 1.0f, new FloatProvider()); + testIcebergIngestion("float", -.0f, .0f, new FloatProvider()); + testIcebergIngestion("float", Float.POSITIVE_INFINITY, new FloatProvider()); + testIcebergIngestion("float", "NaN", Float.NaN, new FloatProvider()); + testIcebergIngestion("float", new BigDecimal("1000.0"), 1000f, new FloatProvider()); + testIcebergIngestion("float", Double.MAX_VALUE, Float.POSITIVE_INFINITY, new FloatProvider()); + testIcebergIngestion("float", null, new FloatProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("float", new Object(), 1f, new FloatProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("float not null", null, new LongProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testFloatAndQueries() throws Exception { + testIcebergIngestAndQuery( + "float", + Arrays.asList(1.0f, -0.0f, Float.POSITIVE_INFINITY, Float.NaN, null), + "select {columnName} from {tableName}", + Arrays.asList(1.0, -0.0, (double) Float.POSITIVE_INFINITY, (double) Float.NaN, null)); + testIcebergIngestAndQuery( + "float", + Arrays.asList(null, null, null, Float.NaN), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "float", + Arrays.asList(1.0f, -0.0f, Float.POSITIVE_INFINITY, Float.NaN), + "select MAX({columnName}) from {tableName}", + Arrays.asList((double) Float.NaN)); + testIcebergIngestAndQuery( + "float", + Arrays.asList(1.0f, -0.0f, Float.POSITIVE_INFINITY, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList((double) Float.POSITIVE_INFINITY)); + testIcebergIngestAndQuery( + "float", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "float", + Arrays.asList( + Float.POSITIVE_INFINITY, + Float.POSITIVE_INFINITY, + Float.POSITIVE_INFINITY, + Float.POSITIVE_INFINITY, + null), + "select COUNT({columnName}) from {tableName} where {columnName} = 'Infinity'", + Arrays.asList(4L)); + } + + @Test + public void testDouble() throws Exception { + testIcebergIngestion("double", 1.0, new DoubleProvider()); + testIcebergIngestion("double", -.0, .0, new DoubleProvider()); + testIcebergIngestion("double", Double.POSITIVE_INFINITY, new DoubleProvider()); + testIcebergIngestion("double", "NaN", Double.NaN, new DoubleProvider()); + testIcebergIngestion("double", new BigDecimal("1000.0"), 1000.0, new DoubleProvider()); + testIcebergIngestion("double", Double.MAX_VALUE, Double.MAX_VALUE, new DoubleProvider()); + testIcebergIngestion("double", null, new DoubleProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("double", new Object(), 1.0, new DoubleProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("double not null", null, new LongProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testDoubleAndQueries() throws Exception { + testIcebergIngestAndQuery( + "double", + Arrays.asList(1.0, -0.0, Double.POSITIVE_INFINITY, Double.NaN, null), + "select {columnName} from {tableName}", + Arrays.asList(1.0, -0.0, Double.POSITIVE_INFINITY, Double.NaN, null)); + testIcebergIngestAndQuery( + "double", + Arrays.asList(null, null, null, Double.NaN), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "double", + Arrays.asList(1.0, -0.0, Double.POSITIVE_INFINITY, Double.NaN), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Double.NaN)); + testIcebergIngestAndQuery( + "double", + Arrays.asList(1.0, -0.0, Double.POSITIVE_INFINITY, null), + "select MAX({columnName}) from {tableName}", + Arrays.asList(Double.POSITIVE_INFINITY)); + testIcebergIngestAndQuery( + "double", + Arrays.asList(null, null, null), + "select MIN({columnName}) from {tableName}", + Arrays.asList((Object) null)); + testIcebergIngestAndQuery( + "double", + Arrays.asList( + Double.POSITIVE_INFINITY, + Double.POSITIVE_INFINITY, + Double.POSITIVE_INFINITY, + Double.POSITIVE_INFINITY, + null), + "select COUNT({columnName}) from {tableName} where {columnName} = 'Infinity'", + Arrays.asList(4L)); + } + + @Test + public void testDecimal() throws Exception { + testIcebergIngestion("decimal(3, 1)", new BigDecimal("-12.3"), new BigDecimalProvider()); + testIcebergIngestion("decimal(1, 0)", new BigDecimal("-0.0"), new BigDecimalProvider()); + 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(38, 10)", null, new BigDecimalProvider()); + + testIcebergIngestion( + "decimal(38, 10)", + "1234567890123456789012345678.1234567890", + new BigDecimal("1234567890123456789012345678.1234567890"), + new BigDecimalProvider()); + + testIcebergIngestion( + "decimal(3, 1)", "12.21999", new BigDecimal("12.2"), new BigDecimalProvider()); + testIcebergIngestion( + "decimal(5, 0)", "12345.52199", new BigDecimal("12346"), new BigDecimalProvider()); + testIcebergIngestion( + "decimal(5, 2)", "12345e-2", new BigDecimal("123.45"), new BigDecimalProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "decimal(3, 1)", new BigDecimal("123.23"), new BigDecimalProvider())); + + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("decimal(38, 10) not null", null, new BigDecimalProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testDecimalAndQueries() throws Exception { + testIcebergIngestAndQuery( + "decimal(3, 1)", + Arrays.asList(new BigDecimal("-12.3"), new BigDecimal("-0.0"), null), + "select {columnName} from {tableName}", + Arrays.asList(new BigDecimal("-12.3"), new BigDecimal("-0.0"), null)); + testIcebergIngestAndQuery( + "decimal(38, 10)", + Arrays.asList(null, null, null), + "select COUNT(*) from {tableName} where {columnName} is null", + Arrays.asList(3L)); + testIcebergIngestAndQuery( + "decimal(32, 10)", + Arrays.asList(new BigDecimal("-233333.3"), new BigDecimal("-23.03"), null), + "select MAX({columnName}) from {tableName}", + Arrays.asList(new BigDecimal("-23.03"))); + testIcebergIngestAndQuery( + "decimal(11, 1)", + Arrays.asList(new BigDecimal("-1222222222.3"), new BigDecimal("-0.0"), null), + "select MIN({columnName}) from {tableName}", + Arrays.asList(new BigDecimal("-1222222222.3"))); + testIcebergIngestAndQuery( + "decimal(3, 1)", + Arrays.asList(new BigDecimal("-12.3"), new BigDecimal("-12.3"), null), + "select COUNT({columnName}) from {tableName} where {columnName} = -12.3", + Arrays.asList(2L)); + } +} diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStringIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStringIT.java new file mode 100644 index 000000000..9ee7788a1 --- /dev/null +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStringIT.java @@ -0,0 +1,80 @@ +package net.snowflake.ingest.streaming.internal.datatypes; + +import java.math.BigDecimal; +import java.util.Arrays; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.apache.commons.lang3.StringUtils; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("This test can be enabled after server side Iceberg EP support is released") +public class IcebergStringIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(true); + } + + @Test + public void testString() throws Exception { + testIcebergIngestion("string", "test", new StringProvider()); + testIcebergIngestion("string", 123, "123", new StringProvider()); + testIcebergIngestion("string", 123.45, "123.45", new StringProvider()); + testIcebergIngestion("string", true, "true", new StringProvider()); + testIcebergIngestion( + "string", new BigDecimal("123456.789"), "123456.789", new StringProvider()); + testIcebergIngestion("string", StringUtils.repeat("a", 16 * 1024 * 1024), new StringProvider()); + testIcebergIngestion("string", "❄️", new StringProvider()); + testIcebergIngestion("string", null, new StringProvider()); + + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("string", new Object(), "test", new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + testIcebergIngestion( + "string", StringUtils.repeat("a", 16 * 1024 * 1024 + 1), new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_VALUE_ROW.getMessageCode()); + + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> testIcebergIngestion("string not null", null, new StringProvider())); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + } + + @Test + public void testStringAndQueries() throws Exception { + testIcebergIngestAndQuery( + "string", + Arrays.asList("test", "test2", "test3", null, "❄️"), + "select {columnName} from {tableName}", + Arrays.asList("test", "test2", "test3", null, "❄️")); + testIcebergIngestAndQuery( + "string", Arrays.asList(null, null, null, null, "aaa"), + "select COUNT(*) from {tableName} where {columnName} is null", Arrays.asList(4L)); + testIcebergIngestAndQuery( + "string", + Arrays.asList(StringUtils.repeat("a", 16 * 1024 * 1024), null, null, null, "aaa"), + "select MAX({columnName}) from {tableName}", + Arrays.asList(StringUtils.repeat("a", 16 * 1024 * 1024))); + testIcebergIngestAndQuery( + "string", + Arrays.asList(StringUtils.repeat("a", 33), StringUtils.repeat("*", 3), null, ""), + "select MAX(LENGTH({columnName})) from {tableName}", + Arrays.asList(33L)); + } +} diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStructuredIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStructuredIT.java new file mode 100644 index 000000000..6bcc34635 --- /dev/null +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/IcebergStructuredIT.java @@ -0,0 +1,168 @@ +package net.snowflake.ingest.streaming.internal.datatypes; + +import com.fasterxml.jackson.databind.JsonNode; +import java.sql.ResultSet; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import net.snowflake.ingest.TestUtils; +import net.snowflake.ingest.streaming.SnowflakeStreamingIngestChannel; +import net.snowflake.ingest.utils.ErrorCode; +import net.snowflake.ingest.utils.SFException; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +@Ignore("This test can be enabled after server side Iceberg EP support is released") +public class IcebergStructuredIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(true); + } + + @Test + public void testStructuredDataType() throws Exception { + assertStructuredDataType( + "object(a int, b string, c boolean)", "{\"a\": 1, \"b\": \"test\", \"c\": true}"); + assertStructuredDataType("map(string, int)", "{\"key1\": 1}"); + assertStructuredDataType("array(int)", "[1, 2, 3]"); + assertMap( + "map(string, int)", + new HashMap() { + { + put("key", 1); + } + }); + assertStructuredDataType("array(string)", null); + + /* Map with null key */ + SFException ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + assertMap( + "map(string, int)", + new HashMap() { + { + put(null, 1); + } + })); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + /* Unknown field */ + ex = + Assertions.catchThrowableOfType( + SFException.class, + () -> + assertStructuredDataType("object(a int, b string)", "{\"a\": 1, \"c\": \"test\"}")); + Assertions.assertThat(ex) + .extracting(SFException::getVendorCode) + .isEqualTo(ErrorCode.INVALID_FORMAT_ROW.getMessageCode()); + + /* Null struct, map list. TODO: SNOW-1727532 Should be fixed with null values EP calculation. */ + Assertions.assertThatThrownBy( + () -> assertStructuredDataType("object(a int, b string, c boolean) not null", null)) + .isInstanceOf(NullPointerException.class); + Assertions.assertThatThrownBy(() -> assertStructuredDataType("map(string, int) not null", null)) + .isInstanceOf(NullPointerException.class); + Assertions.assertThatThrownBy(() -> assertStructuredDataType("array(int) not null", null)) + .isInstanceOf(NullPointerException.class); + + /* Nested data types. Should be fixed. Fixed in server side. */ + Assertions.assertThatThrownBy( + () -> assertStructuredDataType("array(array(int))", "[[1, 2], [3, 4]]")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "array(map(string, int))", "[{\"key1\": 1}, {\"key2\": 2}]")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "array(object(a int, b string, c boolean))", + "[{\"a\": 1, \"b\": \"test\", \"c\": true}]")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "map(string, object(a int, b string, c boolean))", + "{\"key1\": {\"a\": 1, \"b\": \"test\", \"c\": true}}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> assertStructuredDataType("map(string, array(int))", "{\"key1\": [1, 2, 3]}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "map(string, map(string, int))", "{\"key1\": {\"key2\": 2}}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "map(string, array(array(int)))", "{\"key1\": [[1, 2], [3, 4]]}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "map(string, array(map(string, int)))", + "{\"key1\": [{\"key2\": 2}, {\"key3\": 3}]}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "map(string, array(object(a int, b string, c boolean)))", + "{\"key1\": [{\"a\": 1, \"b\": \"test\", \"c\": true}]}")) + .isInstanceOf(SFException.class); + Assertions.assertThatThrownBy( + () -> + assertStructuredDataType( + "object(a int, b array(int), c map(string, int))", + "{\"a\": 1, \"b\": [1, 2, 3], \"c\": {\"key1\": 1}}")) + .isInstanceOf(SFException.class); + } + + private void assertStructuredDataType(String dataType, String value) throws Exception { + String tableName = createIcebergTable(dataType); + String offsetToken = UUID.randomUUID().toString(); + + /* Ingest using streaming ingest */ + SnowflakeStreamingIngestChannel channel = openChannel(tableName); + channel.insertRow( + createStreamingIngestRow( + value == null ? null : objectMapper.readValue(value, Object.class)), + offsetToken); + TestUtils.waitForOffset(channel, offsetToken); + + /* Verify the data */ + ResultSet res = + conn.createStatement().executeQuery(String.format("select * from %s", tableName)); + res.next(); + String tmp = res.getString(2); + JsonNode actualNode = tmp == null ? null : objectMapper.readTree(tmp); + JsonNode expectedNode = value == null ? null : objectMapper.readTree(value); + Assertions.assertThat(actualNode).isEqualTo(expectedNode); + } + + private void assertMap(String dataType, Map value) throws Exception { + String tableName = createIcebergTable(dataType); + String offsetToken = UUID.randomUUID().toString(); + + /* Ingest using streaming ingest */ + SnowflakeStreamingIngestChannel channel = openChannel(tableName); + channel.insertRow(createStreamingIngestRow(value), offsetToken); + TestUtils.waitForOffset(channel, offsetToken); + + /* Verify the data */ + ResultSet res = + conn.createStatement().executeQuery(String.format("select * from %s", tableName)); + res.next(); + String tmp = res.getString(2); + JsonNode actualNode = tmp == null ? null : objectMapper.readTree(tmp); + JsonNode expectedNode = value == null ? null : objectMapper.valueToTree(value); + Assertions.assertThat(actualNode).isEqualTo(expectedNode); + } +} diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/LogicalTypesIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/LogicalTypesIT.java index 6bc769a94..526d4e17d 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/LogicalTypesIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/LogicalTypesIT.java @@ -2,9 +2,14 @@ import java.math.BigDecimal; import java.math.BigInteger; +import org.junit.Before; import org.junit.Test; public class LogicalTypesIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(false); + } @Test public void testLogicalTypes() throws Exception { diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NullIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NullIT.java index fa01eebf0..23e933c14 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NullIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NullIT.java @@ -1,9 +1,14 @@ package net.snowflake.ingest.streaming.internal.datatypes; import java.util.Arrays; +import org.junit.Before; import org.junit.Test; public class NullIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(false); + } @Test public void testNullIngestion() throws Exception { diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NumericTypesIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NumericTypesIT.java index 4e6ade712..ebdb37015 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NumericTypesIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/NumericTypesIT.java @@ -2,9 +2,14 @@ import java.math.BigDecimal; import java.math.BigInteger; +import org.junit.Before; import org.junit.Test; public class NumericTypesIT extends AbstractDataTypeTest { + @Before + public void before() throws Exception { + super.before(false); + } @Test public void testIntegers() throws Exception { diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/SemiStructuredIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/SemiStructuredIT.java index 432312357..b2eaa8d48 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/SemiStructuredIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/SemiStructuredIT.java @@ -16,10 +16,14 @@ import java.util.HashMap; import java.util.Map; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; public class SemiStructuredIT extends AbstractDataTypeTest { - + @Before + public void before() throws Exception { + super.before(false); + } // TODO SNOW-664249: There is a few-byte mismatch between the value sent by the user and its // server-side representation. Validation leaves a small buffer for this difference. private static final int MAX_ALLOWED_LENGTH = 16 * 1024 * 1024 - 64; diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/StringsIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/StringsIT.java index 63dc515f5..acd9b2283 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/StringsIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/datatypes/StringsIT.java @@ -9,6 +9,7 @@ import net.snowflake.ingest.utils.ErrorCode; import net.snowflake.ingest.utils.SFException; import org.junit.Assert; +import org.junit.Before; import org.junit.Ignore; import org.junit.Test; @@ -16,6 +17,11 @@ public class StringsIT extends AbstractDataTypeTest { private static final int MB_16 = 16 * 1024 * 1024; + @Before + public void before() throws Exception { + super.before(false); + } + @Test public void testStrings() throws Exception { testJdbcTypeCompatibility("VARCHAR", "", new StringProvider()); diff --git a/src/test/java/net/snowflake/ingest/streaming/internal/it/ColumnNamesIT.java b/src/test/java/net/snowflake/ingest/streaming/internal/it/ColumnNamesIT.java index 5f2c13f49..d8dbcc72f 100644 --- a/src/test/java/net/snowflake/ingest/streaming/internal/it/ColumnNamesIT.java +++ b/src/test/java/net/snowflake/ingest/streaming/internal/it/ColumnNamesIT.java @@ -15,11 +15,17 @@ import net.snowflake.ingest.streaming.internal.datatypes.AbstractDataTypeTest; import net.snowflake.ingest.utils.SFException; import org.junit.Assert; +import org.junit.Before; import org.junit.Test; public class ColumnNamesIT extends AbstractDataTypeTest { private static final int INGEST_VALUE = 1; + @Before + public void before() throws Exception { + super.before(false); + } + @Test public void testColumnNamesSupport() throws Exception { // Test simple case