Skip to content

Commit

Permalink
test updates
Browse files Browse the repository at this point in the history
  • Loading branch information
@vkorukanti
vkorukanti committed Jun 21, 2023
1 parent 20d6d33 commit 84b3128
Show file tree
Hide file tree
Showing 2 changed files with 44 additions and 25 deletions.
1 change: 0 additions & 1 deletion kernel/kernel-default/src/main/java/io/delta/kernel/parquet/ParquetConverters.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -498,7 +498,6 @@ public void resizeIfNeeded()
public static class FloatColumnConverter
extends BasePrimitiveColumnConverter
{

// working state
private float[] values;

Expand Down
68 changes: 44 additions & 24 deletions kernel/kernel-default/src/test/java/io/delta/kernel/parquet/TestParquetBatchReader.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,20 +97,14 @@ public class TestParquetBatchReader
public void readAllTypesOfData()
throws Exception
{
ParquetBatchReader batchReader = new ParquetBatchReader(newConf(Optional.of(90)));
List<ColumnarBatch> batches =
readAsBatches(batchReader, ALL_TYPES_FILE, ALL_TYPES_FILE_SCHEMA);

for (int rowId = 0; rowId < 200; rowId += 2) {
verifyRowFromAllTypesFile(ALL_TYPES_FILE_SCHEMA, batches, rowId);
}
readAndVerify(ALL_TYPES_FILE_SCHEMA, 90 /* readBatchSize */);
}

@Test
public void readSubsetOfColumns()
throws Exception
{
StructType ALL_TYPES_FILE_SCHEMA = new StructType()
StructType readSchema = new StructType()
.add("byteType", ByteType.INSTANCE)
.add("booleanType", BooleanType.INSTANCE)
.add("stringType", StringType.INSTANCE)
Expand All @@ -123,27 +117,21 @@ public void readSubsetOfColumns()
new ArrayType(IntegerType.INSTANCE, true)
);

ParquetBatchReader batchReader = new ParquetBatchReader(newConf(Optional.of(73)));

List<ColumnarBatch> batches =
readAsBatches(batchReader, ALL_TYPES_FILE, ALL_TYPES_FILE_SCHEMA);
readAndVerify(readSchema, 73 /* readBatchSize */);
}

@Test
public void readSubsetOfColumnsWithMissingColumnsInFile()
throws Exception
{
StructType ALL_TYPES_FILE_SCHEMA = new StructType()
StructType readSchema = new StructType()
.add("booleanType", BooleanType.INSTANCE)
.add("integerType", IntegerType.INSTANCE)
.add("missing_column_primitive", DateType.INSTANCE)
.add("missing_column_struct",
new StructType().add("ab", IntegerType.INSTANCE));

ParquetBatchReader batchReader = new ParquetBatchReader(newConf(Optional.of(23)));

List<ColumnarBatch> batches =
readAsBatches(batchReader, ALL_TYPES_FILE, ALL_TYPES_FILE_SCHEMA);
readAndVerify(readSchema, 23 /* readBatchSize */);
}

private static Configuration newConf(Optional<Integer> batchSize)
Expand All @@ -155,6 +143,19 @@ private static Configuration newConf(Optional<Integer> batchSize)
return conf;
}

private static void readAndVerify(StructType readSchema, int readBatchSize)
throws Exception
{
ParquetBatchReader batchReader =
new ParquetBatchReader(newConf(Optional.of(readBatchSize)));
List<ColumnarBatch> batches =
readAsBatches(batchReader, ALL_TYPES_FILE, readSchema);

for (int rowId = 0; rowId < 200; rowId++) {
verifyRowFromAllTypesFile(readSchema, batches, rowId);
}
}

private static List<ColumnarBatch> readAsBatches(
ParquetBatchReader parquetReader,
String path,
Expand Down Expand Up @@ -274,7 +275,7 @@ private static void verifyRowFromAllTypesFile(
break;
}
case "binarytype": {
byte[] expValue = (rowId % 57 != 0) ? Integer.toString(rowId).getBytes() : null;
byte[] expValue = (rowId % 59 != 0) ? Integer.toString(rowId).getBytes() : null;
if (expValue == null) {
assertTrue(vector.isNullAt(batchWithIdx._2));
}
Expand Down Expand Up @@ -344,20 +345,39 @@ private static void verifyRowFromAllTypesFile(
assertEquals(expValue0, actValue0);

// entry 1: key = if (rowId % 27 != 0) rowId + 2 else null
Integer key1 = (rowId % 27 == 0) ? null : rowId + 2;
// TODO: Not sure if this is a bug or expected behavior. In Delta-Spark,
// whenever the map key value is null - it is stored as 0. Not sure
// what happens for non-integer keys.
// Integer key1 = (rowId % 27 == 0) ? null : rowId + 2;
Integer key1 = (rowId % 27 == 0) ? 0 : rowId + 2;
Long actValue1 = actValue.get(key1);
Long expValue1 = rowId + 9L;
assertEquals(expValue1, actValue1);
}
break;
}
case "map_of_complex": {
// Map(i + 1 -> (if (i % 10 == 0) Row((i*20).longValue()) else null))
assertFalse(vector.isNullAt(batchWithIdx._2));
Map<Integer, Row> actValue = vector.getMap(batchWithIdx._2);

// entry 0: key = rowId
Integer key0 = rowId + 1;
boolean expValue0IsNull = rowId % 10 != 0;
Row actValue0 = actValue.get(key0);
if (expValue0IsNull) {
assertNull(actValue0);
}
else {
Long actValue0Member = actValue0.getLong(0);
Long expValue0Member = rowId * 20L;
assertEquals(expValue0Member, actValue0Member);
}
break;
}
case "missing_column_primitive": {
break;
}
case "missing_column_primitive":
case "missing_column_struct": {
assertTrue(vector.isNullAt(batchWithIdx._2));
break;
}
default:
Expand All @@ -367,8 +387,8 @@ private static void verifyRowFromAllTypesFile(
}
}

private static Tuple2<ColumnarBatch, Integer>
getBatchForRowId(List<ColumnarBatch> batches, int rowId)
private static Tuple2<ColumnarBatch, Integer> getBatchForRowId(
List<ColumnarBatch> batches, int rowId)
{
int indexStart = 0;
for (ColumnarBatch batch : batches) {
Expand Down

0 comments on commit 84b3128

Please sign in to comment.